# -*- coding: utf-8 -*-
"""
收入结构分析任务

功能说明：
    以"日期+区域/类型"为粒度，分析收入结构
    
数据来源：
    - dwd_settlement_head: 结账单头表（台费、商品、助教正价）
    - dwd_table_fee_log: 台费流水（区域关联）
    - dwd_assistant_service_log: 助教服务流水（区域关联）
    - cfg_area_category: 区域分类映射

目标表：
    billiards_dws.dws_finance_income_structure

更新策略：
    - 更新频率：每日更新
    - 幂等方式：delete-before-insert（按日期+类型）
    
业务规则：
    - 结构类型1（INCOME_TYPE）：按收入类型分析（台费/商品/助教基础课/助教附加课）
    - 结构类型2（AREA）：按区域分析（普通台球区/VIP包厢/斯诺克/麻将/KTV等）
    - 区域映射使用cfg_area_category配置

作者：ETL团队
创建日期：2026-02-01
"""

from __future__ import annotations

from datetime import date, datetime, timedelta
from decimal import Decimal
from typing import Any, Dict, List, Optional, Tuple

from .base_dws_task import BaseDwsTask, TaskContext


class FinanceIncomeStructureTask(BaseDwsTask):
    """
    收入结构分析任务
    
    分析收入的两种维度：
    1. INCOME_TYPE: 按收入类型（台费/商品/助教基础课/助教附加课）
    2. AREA: 按区域（使用cfg_area_category映射）
    """
    
    def get_task_code(self) -> str:
        return "DWS_FINANCE_INCOME_STRUCTURE"
    
    def get_target_table(self) -> str:
        return "dws_finance_income_structure"
    
    def get_primary_keys(self) -> List[str]:
        return ["site_id", "stat_date", "structure_type", "category_code"]
    
    def extract(self, context: TaskContext) -> Dict[str, Any]:
        """
        抽取数据
        
        分两条路径抽取：
        1. 按收入类型汇总（来自settlement_head）
        2. 按区域汇总（来自table_fee_log和assistant_service_log）
        """
        start_date = context.window_start.date() if hasattr(context.window_start, 'date') else context.window_start
        end_date = context.window_end.date() if hasattr(context.window_end, 'date') else context.window_end
        site_id = context.store_id
        
        # 按收入类型汇总
        income_by_type = self._extract_income_by_type(site_id, start_date, end_date)
        
        # 按区域汇总
        income_by_area = self._extract_income_by_area(site_id, start_date, end_date)
        
        return {
            'income_by_type': income_by_type,
            'income_by_area': income_by_area,
        }
    
    def _extract_income_by_type(
        self, 
        site_id: int, 
        start_date: date, 
        end_date: date
    ) -> List[Dict[str, Any]]:
        """
        按收入类型汇总
        
        收入类型分类：
        - TABLE_FEE: 台费收入 (table_charge_money)
        - GOODS: 商品收入 (goods_money)
        - ASSISTANT_BASE: 助教基础课 (assistant_pd_money)
        - ASSISTANT_BONUS: 助教附加课 (assistant_cx_money)
        """
        sql = """
            SELECT 
                pay_time::DATE AS stat_date,
                -- 台费收入
                COALESCE(SUM(table_charge_money), 0) AS table_fee_income,
                COUNT(CASE WHEN table_charge_money > 0 THEN 1 END) AS table_fee_orders,
                -- 商品收入
                COALESCE(SUM(goods_money), 0) AS goods_income,
                COUNT(CASE WHEN goods_money > 0 THEN 1 END) AS goods_orders,
                -- 助教基础课收入（PD=陪打）
                COALESCE(SUM(assistant_pd_money), 0) AS assistant_base_income,
                COUNT(CASE WHEN assistant_pd_money > 0 THEN 1 END) AS assistant_base_orders,
                -- 助教附加课收入（CX=超休/促销）
                COALESCE(SUM(assistant_cx_money), 0) AS assistant_bonus_income,
                COUNT(CASE WHEN assistant_cx_money > 0 THEN 1 END) AS assistant_bonus_orders,
                -- 总订单数
                COUNT(*) AS total_orders
            FROM billiards_dwd.dwd_settlement_head
            WHERE site_id = %(site_id)s
              AND pay_time >= %(start_date)s
              AND pay_time < %(end_date)s + INTERVAL '1 day'
              AND settle_status = 1  -- 已结账
            GROUP BY pay_time::DATE
            ORDER BY stat_date
        """
        rows = self.db.query(sql, {
            'site_id': site_id,
            'start_date': start_date,
            'end_date': end_date,
        })
        return [dict(row) for row in rows] if rows else []
    
    def _extract_income_by_area(
        self, 
        site_id: int, 
        start_date: date, 
        end_date: date
    ) -> List[Dict[str, Any]]:
        """
        按区域汇总收入
        
        关联dim_table获取区域名称，再映射到cfg_area_category
        """
        sql = """
            WITH area_orders AS (
                SELECT 
                    tfl.pay_time::DATE AS stat_date,
                    dt.site_table_area_name AS area_name,
                    tfl.order_settle_id,
                    COALESCE(tfl.ledger_amount, 0) AS income_amount,
                    COALESCE(tfl.ledger_time_seconds, 0) AS duration_seconds
                FROM billiards_dwd.dwd_table_fee_log tfl
                LEFT JOIN billiards_dwd.dim_table dt 
                    ON dt.site_table_id = tfl.site_table_id
                WHERE tfl.site_id = %(site_id)s
                  AND tfl.pay_time >= %(start_date)s
                  AND tfl.pay_time < %(end_date)s + INTERVAL '1 day'
                  AND COALESCE(tfl.is_delete, 0) = 0
                
                UNION ALL
                
                SELECT 
                    asl.start_use_time::DATE AS stat_date,
                    dt.site_table_area_name AS area_name,
                    asl.order_settle_id,
                    COALESCE(asl.ledger_amount, 0) AS income_amount,
                    COALESCE(asl.income_seconds, 0) AS duration_seconds
                FROM billiards_dwd.dwd_assistant_service_log asl
                LEFT JOIN billiards_dwd.dim_table dt 
                    ON dt.site_table_id = asl.site_table_id
                WHERE asl.site_id = %(site_id)s
                  AND asl.start_use_time >= %(start_date)s
                  AND asl.start_use_time < %(end_date)s + INTERVAL '1 day'
                  AND asl.is_delete = 0
            )
            SELECT 
                stat_date,
                area_name,
                COALESCE(SUM(income_amount), 0) AS income_amount,
                COALESCE(SUM(duration_seconds), 0) AS duration_seconds,
                COUNT(DISTINCT order_settle_id) AS order_count
            FROM area_orders
            GROUP BY stat_date, area_name
            ORDER BY stat_date, area_name
        """
        rows = self.db.query(sql, {
            'site_id': site_id,
            'start_date': start_date,
            'end_date': end_date,
        })
        return [dict(row) for row in rows] if rows else []
    
    def transform(self, data: Dict[str, Any], context: TaskContext) -> List[Dict[str, Any]]:
        """
        转换数据
        
        将抽取的数据转换为目标表格式：
        1. 按收入类型展开（每种类型一条记录）
        2. 按区域展开（每个区域一条记录）
        3. 计算占比
        """
        site_id = context.store_id
        tenant_id = self.config.get("app.tenant_id", site_id)
        
        records = []
        
        # 处理按收入类型的数据
        income_type_records = self._transform_income_by_type(
            data.get('income_by_type', []),
            site_id,
            tenant_id
        )
        records.extend(income_type_records)
        
        # 处理按区域的数据
        area_records = self._transform_income_by_area(
            data.get('income_by_area', []),
            site_id,
            tenant_id
        )
        records.extend(area_records)
        
        return records
    
    def _transform_income_by_type(
        self,
        income_data: List[Dict[str, Any]],
        site_id: int,
        tenant_id: int
    ) -> List[Dict[str, Any]]:
        """
        转换按收入类型的数据
        
        将每日汇总数据展开为4条记录（台费/商品/基础课/附加课）
        """
        # 收入类型定义
        income_types = [
            ('TABLE_FEE', '台费收入', 'table_fee_income', 'table_fee_orders'),
            ('GOODS', '商品收入', 'goods_income', 'goods_orders'),
            ('ASSISTANT_BASE', '助教基础课', 'assistant_base_income', 'assistant_base_orders'),
            ('ASSISTANT_BONUS', '助教附加课', 'assistant_bonus_income', 'assistant_bonus_orders'),
        ]
        
        records = []
        
        for daily_data in income_data:
            stat_date = daily_data.get('stat_date')
            
            # 计算当日总收入（用于计算占比）
            total_income = sum(
                self.safe_decimal(daily_data.get(field, 0))
                for _, _, field, _ in income_types
            )
            
            # 为每种收入类型生成一条记录
            for type_code, type_name, income_field, order_field in income_types:
                income_amount = self.safe_decimal(daily_data.get(income_field, 0))
                order_count = daily_data.get(order_field, 0) or 0
                
                # 计算占比（避免除零）
                income_ratio = (income_amount / total_income) if total_income > 0 else Decimal('0')
                
                records.append({
                    'site_id': site_id,
                    'tenant_id': tenant_id,
                    'stat_date': stat_date,
                    'structure_type': 'INCOME_TYPE',
                    'category_code': type_code,
                    'category_name': type_name,
                    'income_amount': income_amount,
                    'income_ratio': round(income_ratio, 4),
                    'order_count': order_count,
                    'duration_minutes': 0,  # 收入类型维度不统计时长
                })
        
        return records
    
    def _transform_income_by_area(
        self,
        area_data: List[Dict[str, Any]],
        site_id: int,
        tenant_id: int
    ) -> List[Dict[str, Any]]:
        """
        转换按区域的数据
        
        将区域名称映射到cfg_area_category的category_code
        """
        records = []
        
        # 加载区域分类配置
        self.load_config_cache()
        
        # 按日期分组计算总收入（用于计算占比）
        daily_totals = {}
        for row in area_data:
            stat_date = row.get('stat_date')
            income = self.safe_decimal(row.get('income_amount', 0))
            daily_totals[stat_date] = daily_totals.get(stat_date, Decimal('0')) + income
        
        # 按日期+区域聚合（相同category_code需要合并）
        aggregated = {}
        
        for row in area_data:
            stat_date = row.get('stat_date')
            area_name = row.get('area_name') or '未知区域'
            income_amount = self.safe_decimal(row.get('income_amount', 0))
            duration_seconds = row.get('duration_seconds', 0) or 0
            order_count = row.get('order_count', 0) or 0
            
            # 映射区域名称到分类代码
            category = self.get_area_category(area_name)
            category_code = category.get('category_code', 'OTHER')
            category_name = category.get('category_name', '其他区域')
            
            # 聚合键
            key = (stat_date, category_code)
            
            if key not in aggregated:
                aggregated[key] = {
                    'stat_date': stat_date,
                    'category_code': category_code,
                    'category_name': category_name,
                    'income_amount': Decimal('0'),
                    'duration_seconds': 0,
                    'order_count': 0,
                }
            
            aggregated[key]['income_amount'] += income_amount
            aggregated[key]['duration_seconds'] += duration_seconds
            aggregated[key]['order_count'] += order_count
        
        # 生成记录
        for key, agg_data in aggregated.items():
            stat_date = agg_data['stat_date']
            total_income = daily_totals.get(stat_date, Decimal('1'))
            income_amount = agg_data['income_amount']
            
            # 计算占比
            income_ratio = (income_amount / total_income) if total_income > 0 else Decimal('0')
            
            records.append({
                'site_id': site_id,
                'tenant_id': tenant_id,
                'stat_date': stat_date,
                'structure_type': 'AREA',
                'category_code': agg_data['category_code'],
                'category_name': agg_data['category_name'],
                'income_amount': income_amount,
                'income_ratio': round(income_ratio, 4),
                'order_count': agg_data['order_count'],
                'duration_minutes': agg_data['duration_seconds'] // 60,
            })
        
        return records
    
    def _map_area_to_category(
        self,
        area_name: str,
        area_categories: Dict[str, Dict[str, Any]]
    ) -> Dict[str, Any]:
        """
        兼容旧逻辑的映射方法（当前使用 get_area_category）
        """
        return self.get_area_category(area_name)
    
    def load(self, records: List[Dict[str, Any]], context: TaskContext) -> Dict[str, Any]:
        """
        加载数据到目标表
        
        使用幂等方式：delete-before-insert（按日期范围）
        """
        if not records:
            return {'inserted': 0, 'deleted': 0}
        
        site_id = context.store_id
        start_date = context.window_start.date() if hasattr(context.window_start, 'date') else context.window_start
        end_date = context.window_end.date() if hasattr(context.window_end, 'date') else context.window_end
        
        # 删除窗口内的旧数据
        delete_sql = """
            DELETE FROM billiards_dws.dws_finance_income_structure
            WHERE site_id = %(site_id)s
              AND stat_date >= %(start_date)s
              AND stat_date <= %(end_date)s
        """
        deleted = self.db.execute(delete_sql, {
            'site_id': site_id,
            'start_date': start_date,
            'end_date': end_date,
        })
        
        # 批量插入新数据
        insert_sql = """
            INSERT INTO billiards_dws.dws_finance_income_structure (
                site_id, tenant_id, stat_date,
                structure_type, category_code, category_name,
                income_amount, income_ratio,
                order_count, duration_minutes,
                created_at, updated_at
            ) VALUES (
                %(site_id)s, %(tenant_id)s, %(stat_date)s,
                %(structure_type)s, %(category_code)s, %(category_name)s,
                %(income_amount)s, %(income_ratio)s,
                %(order_count)s, %(duration_minutes)s,
                NOW(), NOW()
            )
        """
        
        inserted = 0
        for record in records:
            self.db.execute(insert_sql, record)
            inserted += 1
        
        return {
            'deleted': deleted or 0,
            'inserted': inserted,
        }