# -*- coding: utf-8 -*-
"""
助教订单流水四项统计任务

功能说明：
    以"助教+日期"为粒度，计算每名助教每日的订单流水贡献：
    - order_gross_revenue: 订单总流水（台费 + 酒水食品 + 所有助教服务费）
    - order_net_revenue: 订单净流水（订单总流水 - 所有助教服务分成）
    - time_weighted_revenue: 时效贡献流水（按服务时长折算的个人贡献）
    - time_weighted_net_revenue: 时效净贡献（时效贡献流水 - 个人服务分成）

数据来源：
    - dwd_settlement_head: 结算主表
    - dwd_table_fee_log: 台费明细
    - dwd_assistant_service_log: 助教服务记录

目标表：
    dws.dws_assistant_order_contribution

更新策略：
    - 幂等方式：delete-before-insert（按日期窗口）

核心算法：
    时效贡献流水按以下步骤计算：
    1. 每张台桌的有效计费时长 = MAX(助教总服务时长, 台桌使用时长)
    2. 台费分摊 = table_fee × (个人服务时长 / 有效计费时长)
    3. 个人服务费直接计入
    4. 酒水食品按助教总时长比例均分

    超休/打赏课（course_type=BONUS）不参与订单级分摊，
    四项统计均设为该助教个人的服务流水和分成。

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

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Any, Dict, List

from .base_dws_task import BaseDwsTask, TaskContext

from neozqyy_shared.datetime_utils import biz_date_sql_expr


# =============================================================================
# 数据结构
# =============================================================================

@dataclass
class TableUsage:
    """台桌使用信息"""
    table_id: int
    table_area: str           # 区域名称（A/B/C/S/TV/M1-M7 等）
    usage_seconds: int        # 台桌使用时长（秒）
    table_fee: Decimal        # 台费/房费


@dataclass
class AssistantService:
    """助教服务记录"""
    assistant_id: int
    table_id: int
    service_seconds: int      # 服务时长（秒）
    ledger_amount: Decimal    # 服务流水（助教收费）
    commission: Decimal       # 助教分成
    skill_id: int
    course_type: str          # BASE / BONUS / ROOM
    nickname: str = ""        # 助教昵称（用于输出）


@dataclass
class OrderData:
    """订单聚合数据（一个结算单的完整信息）"""
    order_settle_id: int
    site_id: int
    total_table_fee: Decimal       # 台费总额
    total_goods_amount: Decimal    # 酒水食品总额
    tables: List[TableUsage] = field(default_factory=list)
    assistants: List[AssistantService] = field(default_factory=list)
    stat_date: date | None = None  # 订单日期（pay_time 的日期部分）


# =============================================================================
# 助教订单流水统计任务
# =============================================================================

class AssistantOrderContributionTask(BaseDwsTask):
    """
    助教订单流水四项统计任务

    粒度：(site_id, assistant_id, stat_date)
    策略：delete-before-insert 幂等更新
    """

    DATE_COL = "stat_date"

    def get_task_code(self) -> str:
        return "DWS_ASSISTANT_ORDER_CONTRIBUTION"

    def get_target_table(self) -> str:
        return "dws_assistant_order_contribution"

    def get_primary_keys(self) -> List[str]:
        return ["site_id", "assistant_id", "stat_date"]

    # =========================================================================
    # ETL 主流程（骨架，后续任务实现）
    # =========================================================================

    def extract(self, context: TaskContext) -> Dict[str, Any]:
        """提取数据：从 DWD 层读取结算、台费和助教服务数据，按订单聚合为 OrderData"""
        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

        self.logger.info(
            "%s: 提取数据，日期范围 %s ~ %s",
            self.get_task_code(), start_date, end_date
        )

        # 1. 提取台桌结账订单的结算主表（settle_type=1 为台桌结账）
        settlements = self._extract_settlements(site_id, start_date, end_date)

        # 2. 提取台费明细
        table_fees = self._extract_table_fees(site_id, start_date, end_date)

        # 3. 提取助教服务记录（含课程类型映射）
        service_logs = self._extract_service_logs(site_id, start_date, end_date)

        # 4. 按 order_settle_id 聚合为 OrderData 列表
        orders = self._aggregate_to_orders(settlements, table_fees, service_logs)

        self.logger.info(
            "%s: 提取完成，结算单 %d 条，聚合订单 %d 条",
            self.get_task_code(), len(settlements), len(orders)
        )

        return {
            'orders': orders,
            'start_date': start_date,
            'end_date': end_date,
            'site_id': site_id,
        }

    def transform(self, extracted: Dict[str, Any], context: TaskContext) -> List[Dict[str, Any]]:
        """转换数据：调用四项统计计算，按 (assistant_id, stat_date) 聚合日度统计"""
        orders: List[OrderData] = extracted['orders']
        site_id = extracted['site_id']

        self.logger.info(
            "%s: 转换数据，订单 %d 条",
            self.get_task_code(), len(orders)
        )

        # 按 (assistant_id, stat_date) 聚合
        agg: Dict[tuple, Dict[str, Any]] = {}

        for order in orders:
            # 跳过无助教服务的订单
            if not order.assistants:
                continue

            # 获取订单日期（从结算主表的 pay_time 推导，存储在 order 中）
            stat_date = getattr(order, 'stat_date', None)
            if stat_date is None:
                continue

            # 收集该订单所有参与助教（去重）
            assistant_ids = set(a.assistant_id for a in order.assistants)

            for aid in assistant_ids:
                contribution = self.compute_assistant_contribution(order, aid)
                key = (aid, stat_date)

                if key not in agg:
                    # 获取助教昵称（取第一条服务记录的昵称）
                    nickname = next(
                        (a.nickname for a in order.assistants if a.assistant_id == aid),
                        None
                    )
                    agg[key] = {
                        'site_id': site_id,
                        'tenant_id': order.site_id,  # tenant_id 与 site_id 相同
                        'assistant_id': aid,
                        'assistant_nickname': nickname,
                        'stat_date': stat_date,
                        'order_gross_revenue': Decimal('0'),
                        'order_net_revenue': Decimal('0'),
                        'time_weighted_revenue': Decimal('0'),
                        'time_weighted_net_revenue': Decimal('0'),
                        'order_count': 0,
                        'total_service_seconds': 0,
                    }

                rec = agg[key]
                rec['order_gross_revenue'] += contribution['order_gross_revenue']
                rec['order_net_revenue'] += contribution['order_net_revenue']
                rec['time_weighted_revenue'] += contribution['time_weighted_revenue']
                rec['time_weighted_net_revenue'] += contribution['time_weighted_net_revenue']
                rec['order_count'] += 1
                # 累加该助教在该订单中的总服务时长
                rec['total_service_seconds'] += sum(
                    a.service_seconds for a in order.assistants if a.assistant_id == aid
                )

        result = list(agg.values())
        self.logger.info(
            "%s: 转换完成，输出 %d 条助教日度统计",
            self.get_task_code(), len(result)
        )
        return result

    # load() 使用 BaseDwsTask 默认实现（DATE_COL="stat_date"）

    # =========================================================================
    # 数据提取方法
    # =========================================================================

    def _extract_settlements(
        self, site_id: int, start_date: date, end_date: date
    ) -> List[Dict[str, Any]]:
        """提取消费结算订单的结算主表

        settle_type IN (1, 3)：
        - 1 = 台桌结账（含台费、商品、陪打、超休）
        - 3 = 商城订单（含纯超休/激励课 477 笔、商品、超休+商品混合）
        settle_type=3 中陪打(pd)为 0、台费为 0，但超休(cx)占全口径 85%+。
        """
        # CHANGE 2026-03-01 | business-day-cutoff 6.2: DATE(pay_time) → 营业日归属表达式
        cutoff = self.config.get("app.business_day_start_hour", 8)
        biz_expr = biz_date_sql_expr("pay_time", cutoff)
        sql = f"""
            SELECT
                order_settle_id,
                site_id,
                tenant_id,
                table_charge_money,
                goods_money,
                {biz_expr} AS stat_date
            FROM dwd.dwd_settlement_head
            WHERE site_id = %s
              AND settle_type IN (1, 3)
              AND {biz_expr} >= %s
              AND {biz_expr} <= %s
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []

    def _extract_table_fees(
        self, site_id: int, start_date: date, end_date: date
    ) -> List[Dict[str, Any]]:
        """提取台费明细

        每条记录对应一张台桌在一个订单中的台费信息。
        real_table_use_seconds 为台桌实际使用时长。
        """
        # CHANGE 2026-03-01 | business-day-cutoff 6.2: DATE(start_use_time) → 营业日归属表达式
        cutoff = self.config.get("app.business_day_start_hour", 8)
        biz_expr = biz_date_sql_expr("tfl.start_use_time", cutoff)
        sql = f"""
            SELECT
                tfl.order_settle_id,
                tfl.site_table_id AS table_id,
                COALESCE(tfl.site_table_area_name, '') AS table_area,
                COALESCE(tfl.real_table_use_seconds, 0) AS usage_seconds,
                COALESCE(tfl.ledger_amount, 0) AS table_fee
            FROM dwd.dwd_table_fee_log tfl
            WHERE tfl.site_id = %s
              AND {biz_expr} >= %s
              AND {biz_expr} <= %s
              AND COALESCE(tfl.is_delete, 0) = 0
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []

    def _extract_service_logs(
        self, site_id: int, start_date: date, end_date: date
    ) -> List[Dict[str, Any]]:
        """提取助教服务记录（含课程类型映射）

        通过 LEFT JOIN cfg_skill_type 获取 course_type_code，
        real_service_money 为助教分成。
        """
        # CHANGE 2026-03-01 | business-day-cutoff 6.2: DATE(start_use_time) → 营业日归属表达式
        cutoff = self.config.get("app.business_day_start_hour", 8)
        biz_expr = biz_date_sql_expr("asl.start_use_time", cutoff)
        sql = f"""
            SELECT
                asl.order_settle_id,
                asl.site_assistant_id AS assistant_id,
                asl.nickname,
                asl.site_table_id AS table_id,
                COALESCE(asl.income_seconds, 0) AS service_seconds,
                COALESCE(asl.ledger_amount, 0) AS ledger_amount,
                COALESCE(asl.real_service_money, 0) AS commission,
                COALESCE(asl.skill_id, 0) AS skill_id,
                COALESCE(cst.course_type_code, 'BASE') AS course_type
            FROM dwd.dwd_assistant_service_log asl
            LEFT JOIN dws.cfg_skill_type cst
                ON asl.skill_id = cst.skill_id
                AND cst.is_active = TRUE
            WHERE asl.site_id = %s
              AND {biz_expr} >= %s
              AND {biz_expr} <= %s
              AND COALESCE(asl.is_delete, 0) = 0
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []

    def _aggregate_to_orders(
        self,
        settlements: List[Dict[str, Any]],
        table_fees: List[Dict[str, Any]],
        service_logs: List[Dict[str, Any]],
    ) -> List[OrderData]:
        """按 order_settle_id 聚合为 OrderData 列表

        只保留有助教服务记录的订单（无助教的订单在 transform 中也会跳过）。
        """
        from collections import defaultdict

        # 按 order_settle_id 索引台费和服务记录
        table_fee_map: Dict[int, List[Dict]] = defaultdict(list)
        for tf in table_fees:
            table_fee_map[tf['order_settle_id']].append(tf)

        service_map: Dict[int, List[Dict]] = defaultdict(list)
        for sl in service_logs:
            service_map[sl['order_settle_id']].append(sl)

        orders: List[OrderData] = []
        for settle in settlements:
            oid = settle['order_settle_id']
            svc_list = service_map.get(oid)
            # 跳过无助教服务的订单
            if not svc_list:
                continue

            tables = [
                TableUsage(
                    table_id=int(tf['table_id']),
                    table_area=tf['table_area'],
                    usage_seconds=int(tf['usage_seconds']),
                    table_fee=Decimal(str(tf['table_fee'])),
                )
                for tf in table_fee_map.get(oid, [])
            ]

            assistants = [
                AssistantService(
                    assistant_id=int(sl['assistant_id']),
                    table_id=int(sl['table_id']),
                    service_seconds=int(sl['service_seconds']),
                    ledger_amount=Decimal(str(sl['ledger_amount'])),
                    commission=Decimal(str(sl['commission'])),
                    skill_id=int(sl['skill_id']),
                    course_type=sl['course_type'],
                    nickname=sl.get('nickname', ''),
                )
                for sl in svc_list
            ]

            orders.append(OrderData(
                order_settle_id=int(oid),
                site_id=int(settle['site_id']),
                total_table_fee=Decimal(str(settle.get('table_charge_money') or 0)),
                total_goods_amount=Decimal(str(settle.get('goods_money') or 0)),
                tables=tables,
                assistants=assistants,
                stat_date=settle.get('stat_date'),
            ))

        return orders

    # =========================================================================
    # 核心计算（纯函数，不依赖数据库，便于属性测试）
    # =========================================================================

    @staticmethod
    def compute_order_gross_revenue(order: OrderData) -> Decimal:
        """订单总流水 = 台费 + 酒水食品 + 所有助教服务费

        每个参与助教获得相同的 order_gross_revenue 值。
        """
        total_service_amount = sum(
            (a.ledger_amount for a in order.assistants), Decimal('0')
        )
        return order.total_table_fee + order.total_goods_amount + total_service_amount

    @staticmethod
    def compute_order_net_revenue(order: OrderData) -> Decimal:
        """订单净流水 = 订单总流水 - 所有助教服务分成

        每个参与助教获得相同的 order_net_revenue 值。
        """
        gross = AssistantOrderContributionTask.compute_order_gross_revenue(order)
        total_commission = sum(
            (a.commission for a in order.assistants), Decimal('0')
        )
        return gross - total_commission

    @staticmethod
    def compute_time_weighted_revenue(
        order: OrderData, assistant_id: int
    ) -> Decimal:
        """时效贡献流水 = 台费按时长分摊 + 个人服务费 + 酒水食品按时长比例

        算法步骤：
        1. 每张台桌：billable_seconds = MAX(助教总服务时长, 台桌使用时长)
           台费分摊 = table_fee × (个人服务时长 / billable_seconds)
        2. 个人服务费（ledger_amount）直接计入
        3. 酒水食品按个人总服务时长占所有助教总服务时长的比例均分

        超休/打赏课（BONUS）：四项统计均设为个人服务流水和分成，
        不参与订单级分摊。此逻辑在调用方处理，本方法仅处理常规情况。

        边界情况：
        - 台桌使用时长为 0 且助教总服务时长也为 0：台费分摊 = 0
        - 助教总服务时长为 0：酒水食品分摊 = 0
        """
        # --- 筛选该助教的服务记录（排除 BONUS 类型） ---
        my_services = [
            a for a in order.assistants
            if a.assistant_id == assistant_id and a.course_type != "BONUS"
        ]
        all_non_bonus = [a for a in order.assistants if a.course_type != "BONUS"]

        # 如果该助教无非 BONUS 服务记录，返回 0
        if not my_services:
            return Decimal('0')

        # --- 步骤 1：台费按时长分摊 ---
        table_fee_share = Decimal('0')
        for table in order.tables:
            # 该台桌上所有助教的服务时长之和
            table_total_svc = sum(
                a.service_seconds for a in all_non_bonus
                if a.table_id == table.table_id
            )
            # 该助教在该台桌的服务时长
            my_table_svc = sum(
                a.service_seconds for a in my_services
                if a.table_id == table.table_id
            )
            if my_table_svc == 0:
                continue

            # 有效计费时长 = MAX(助教总服务时长, 台桌使用时长)
            billable_seconds = max(table_total_svc, table.usage_seconds)
            if billable_seconds <= 0:
                continue

            table_fee_share += table.table_fee * Decimal(my_table_svc) / Decimal(billable_seconds)

        # --- 步骤 2：个人服务费直接计入 ---
        personal_service = sum(
            (a.ledger_amount for a in my_services), Decimal('0')
        )

        # --- 步骤 3：酒水食品按总时长比例均分 ---
        my_total_seconds = sum(a.service_seconds for a in my_services)
        all_total_seconds = sum(a.service_seconds for a in all_non_bonus)

        if all_total_seconds > 0 and my_total_seconds > 0:
            goods_share = order.total_goods_amount * Decimal(my_total_seconds) / Decimal(all_total_seconds)
        else:
            goods_share = Decimal('0')

        return table_fee_share + personal_service + goods_share

    @staticmethod
    def compute_time_weighted_net_revenue(
        time_weighted_revenue: Decimal, assistant_commission: Decimal
    ) -> Decimal:
        """时效净贡献 = 时效贡献流水 - 个人服务分成"""
        return time_weighted_revenue - assistant_commission

    @staticmethod
    def compute_assistant_contribution(
        order: OrderData, assistant_id: int
    ) -> Dict[str, Decimal]:
        """计算单个助教在单个订单中的四项统计（含 BONUS 特殊处理）

        返回字典包含：
        - order_gross_revenue
        - order_net_revenue
        - time_weighted_revenue
        - time_weighted_net_revenue
        - total_commission（该助教个人分成，辅助字段）

        超休/打赏课（BONUS）：四项统计均设为个人服务流水和分成，
        不参与订单级分摊。
        """
        cls = AssistantOrderContributionTask

        # 该助教的所有服务记录
        my_services = [a for a in order.assistants if a.assistant_id == assistant_id]
        if not my_services:
            return {
                'order_gross_revenue': Decimal('0'),
                'order_net_revenue': Decimal('0'),
                'time_weighted_revenue': Decimal('0'),
                'time_weighted_net_revenue': Decimal('0'),
                'total_commission': Decimal('0'),
            }

        # 分离 BONUS 和非 BONUS 服务
        bonus_services = [a for a in my_services if a.course_type == "BONUS"]
        normal_services = [a for a in my_services if a.course_type != "BONUS"]

        # BONUS 部分：直接用个人流水
        bonus_revenue = sum((a.ledger_amount for a in bonus_services), Decimal('0'))
        bonus_commission = sum((a.commission for a in bonus_services), Decimal('0'))

        if normal_services:
            # 有常规服务：按正常逻辑计算
            normal_commission = sum((a.commission for a in normal_services), Decimal('0'))
            total_commission = normal_commission + bonus_commission
            gross = cls.compute_order_gross_revenue(order)
            net = cls.compute_order_net_revenue(order)
            twr = cls.compute_time_weighted_revenue(order, assistant_id)

            # 合成最终值（先算 time_weighted_revenue 再减 total_commission，
            # 保证 twnr == twr_final - total_commission 精度一致）
            twr_final = twr + bonus_revenue
            twnr_final = twr_final - total_commission

            return {
                'order_gross_revenue': gross + bonus_revenue,
                'order_net_revenue': net + (bonus_revenue - bonus_commission),
                'time_weighted_revenue': twr_final,
                'time_weighted_net_revenue': twnr_final,
                'total_commission': total_commission,
            }
        else:
            # 纯 BONUS 助教：四项统计均为个人流水
            return {
                'order_gross_revenue': bonus_revenue,
                'order_net_revenue': bonus_revenue - bonus_commission,
                'time_weighted_revenue': bonus_revenue,
                'time_weighted_net_revenue': bonus_revenue - bonus_commission,
                'total_commission': bonus_commission,
            }


# 便于外部导入
__all__ = [
    'TableUsage',
    'AssistantService',
    'OrderData',
    'AssistantOrderContributionTask',
]
