# -*- coding: utf-8 -*-
"""
助教日度业绩明细任务

功能说明：
    以"助教+日期"为粒度，汇总每日业绩明细
    
数据来源：
    - dwd_assistant_service_log: 助教服务流水
    - dwd_assistant_service_log_ex: 扩展表（提供 is_trash 废除标记）
    - dim_assistant: 助教维度（SCD2，获取当日等级）
    - cfg_skill_type: 技能→课程类型映射

目标表：
    dws.dws_assistant_daily_detail

更新策略：
    - 更新频率：每小时增量更新
    - 幂等方式：delete-before-insert（按日期窗口）
    
业务规则：
    - 有效业绩：通过 dwd_assistant_service_log_ex.is_trash 字段判断是否被废除
    - 助教等级：使用SCD2 as-of取值，获取统计日当日生效的等级
    - 课程类型：通过skill_id映射，分为基础课和附加课

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

from __future__ import annotations

from collections import defaultdict
from datetime import date, datetime, time, timedelta
from decimal import Decimal, ROUND_HALF_UP
from typing import Any, Dict, List, Optional, Set, Tuple

from neozqyy_shared.datetime_utils import biz_date_sql_expr

from .base_dws_task import BaseDwsTask, CourseType, TaskContext

# 惩罚区域集合：大厅 A/B/C/S/TV + 麻将房 M1–M7
PENALTY_AREAS: Set[str] = {
    "A", "B", "C", "S", "TV",
    "M1", "M2", "M3", "M4", "M5", "M6", "M7",
}


class AssistantDailyTask(BaseDwsTask):
    """
    助教日度业绩明细任务
    
    汇总每个助教每天的：
    - 服务次数（总/基础课/附加课）
    - 计费时长（秒/小时）
    - 计费金额
    - 服务客户数（去重）
    - 服务台桌数（去重）
    - 被废除的记录统计
    """

    # CHANGE 2025-07-15 | task 1.3: 声明日期列，供基类默认 load() 使用
    DATE_COL = "stat_date"
    
    def get_task_code(self) -> str:
        return "DWS_ASSISTANT_DAILY"
    
    def get_target_table(self) -> str:
        return "dws_assistant_daily_detail"
    
    def get_primary_keys(self) -> List[str]:
        return ["site_id", "assistant_id", "stat_date"]
    
    # ==========================================================================
    # ETL主流程
    # ==========================================================================
    
    def extract(self, context: TaskContext) -> Dict[str, Any]:
        """
        提取数据：从DWD层读取助教服务记录
        """
        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. 获取助教服务记录（含 is_trash 标记，来自 _ex 表 JOIN）
        service_records = self._extract_service_records(site_id, start_date, end_date)
        
        # 2. 加载配置缓存
        self.load_config_cache()
        
        return {
            'service_records': service_records,
            '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]]:
        """
        转换数据：按助教+日期聚合，并执行定档折算惩罚检测
        """
        service_records = extracted['service_records']
        site_id = extracted['site_id']
        
        self.logger.info(
            "%s: 转换数据，服务记录 %d 条",
            self.get_task_code(), len(service_records)
        )
        
        # 按助教+日期聚合
        aggregated = self._aggregate_by_assistant_date(
            service_records, 
            site_id
        )

        # ── 定档折算惩罚检测 ──
        # 构造重叠检测所需的记录格式
        overlap_records = []
        for r in service_records:
            start_t = r.get("start_use_time")
            end_t = r.get("last_use_time")
            if start_t is None or end_t is None:
                continue
            overlap_records.append({
                "assistant_id": r.get("assistant_id"),
                "table_id": r.get("table_id"),
                "table_area": r.get("table_area_name", ""),
                "start_time": start_t,
                "end_time": end_t,
                "service_date": r.get("service_date"),
            })

        violations = self.detect_overlap_violations(overlap_records, PENALTY_AREAS)

        # 将惩罚信息填充到聚合结果
        for agg in aggregated:
            aid = agg["assistant_id"]
            stat_date = agg["stat_date"]
            key = (aid, stat_date)

            if agg.get("is_exempt"):
                # 豁免：不计算惩罚
                agg["penalty_minutes"] = Decimal("0")
                agg["penalty_reason"] = None
                agg["is_exempt"] = True
                agg["per_hour_contribution"] = None
            elif key in violations:
                # 有违规：计算惩罚
                # 取第一条违规信息（同一天可能有多条，取最严重的）
                v_list = violations[key]
                overlap_count = max(v["overlap_count"] for v in v_list)
                # per_hour_contribution 需要从台费数据计算
                # 此处使用聚合后的 base_ledger_amount 和 base_hours 近似
                base_hours = agg.get("base_hours", Decimal("0"))
                base_amount = agg.get("base_ledger_amount", Decimal("0"))
                if base_hours > 0:
                    per_hour = base_amount / base_hours / Decimal(str(overlap_count))
                else:
                    per_hour = Decimal("0")

                actual_minutes = agg.get("base_hours", Decimal("0")) * Decimal("60")
                penalty = self.compute_penalty_minutes(actual_minutes, per_hour)

                agg["penalty_minutes"] = penalty
                agg["penalty_reason"] = (
                    f"规则2违规：同台桌{overlap_count}名助教重叠挂台，"
                    f"单人每小时贡献={per_hour:.2f}元"
                )
                agg["is_exempt"] = False
                agg["per_hour_contribution"] = per_hour
            else:
                # 无违规
                agg["penalty_minutes"] = Decimal("0")
                agg["penalty_reason"] = None
                agg["is_exempt"] = False
                agg["per_hour_contribution"] = None
        
        return aggregated
    
    # load() 已移除——使用 BaseDwsTask 默认实现（DATE_COL="stat_date"）

    # ==========================================================================
    # 数据提取方法
    # ==========================================================================
    
    def _extract_service_records(
        self,
        site_id: int,
        start_date: date,
        end_date: date
    ) -> List[Dict[str, Any]]:
        """
        提取助教服务记录

        JOIN _ex 表取 is_trash 字段，用于直接判断服务是否被废除。
        """
        # CHANGE 2026-02-26: dwd_assistant_service_log 无 table_area_name 列，
        # 改为 JOIN dim_table 取 site_table_area_name
        # CHANGE 2026-03-01 | business-day-cutoff 6.1: DATE() → 营业日归属表达式
        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.assistant_service_id,
                asl.order_settle_id,
                asl.site_assistant_id AS assistant_id,
                asl.nickname AS assistant_nickname,
                asl.assistant_level,
                asl.skill_id,
                asl.skill_name,
                asl.tenant_member_id AS member_id,
                asl.site_table_id AS table_id,
                asl.income_seconds,
                asl.real_use_seconds,
                asl.ledger_amount,
                asl.ledger_unit_price,
                asl.start_use_time,
                asl.last_use_time,
                COALESCE(dt.site_table_area_name, '') AS table_area_name,
                {biz_expr} AS service_date,
                COALESCE(ex.is_trash, 0) AS is_trash
            FROM dwd.dwd_assistant_service_log asl
            LEFT JOIN dwd.dwd_assistant_service_log_ex ex
                ON asl.assistant_service_id = ex.assistant_service_id
            LEFT JOIN dwd.dim_table dt
                ON asl.site_table_id = dt.table_id
                AND dt.scd2_is_current = 1
            WHERE asl.site_id = %s
              AND {biz_expr} >= %s
              AND {biz_expr} <= %s
              AND asl.is_delete = 0
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []
    
    
    # ==========================================================================
    # 数据转换方法
    # ==========================================================================
    
    def _aggregate_by_assistant_date(
        self,
        service_records: List[Dict[str, Any]],
        site_id: int
    ) -> List[Dict[str, Any]]:
        """
        按助教+日期聚合服务记录
        """
        # 聚合字典：(assistant_id, service_date) -> aggregated_data
        agg_dict: Dict[Tuple[int, date], Dict[str, Any]] = {}
        
        for record in service_records:
            assistant_id = record.get('assistant_id')
            service_date = record.get('service_date')
            
            if not assistant_id or not service_date:
                continue
            
            key = (assistant_id, service_date)
            
            # 初始化聚合数据
            if key not in agg_dict:
                # 获取助教当日等级（SCD2 as-of）
                level_info = self.get_assistant_level_asof(assistant_id, service_date)
                
                # CHANGE 2026-02-27 | level_name 始终由 code 静态映射得出
                # SCD2 仅用于取历史 level_code（等级可能变过），
                # name 不再依赖 SCD2 返回值，避免 SCD2 缺失时 NULL
                level_code = level_info.get('level_code') if level_info else record.get('assistant_level')
                level_name = self.level_code_to_name(level_code)

                agg_dict[key] = {
                    'site_id': site_id,
                    'tenant_id': self.config.get("app.tenant_id", site_id),
                    'assistant_id': assistant_id,
                    'assistant_nickname': record.get('assistant_nickname'),
                    'stat_date': service_date,
                    'assistant_level_code': level_code,
                    'assistant_level_name': level_name,
                    'total_service_count': 0,
                    'base_service_count': 0,
                    'bonus_service_count': 0,
                    'room_service_count': 0,
                    'total_seconds': 0,
                    'base_seconds': 0,
                    'bonus_seconds': 0,
                    'room_seconds': 0,
                    'total_hours': Decimal('0'),
                    'base_hours': Decimal('0'),
                    'bonus_hours': Decimal('0'),
                    'room_hours': Decimal('0'),
                    'total_ledger_amount': Decimal('0'),
                    'base_ledger_amount': Decimal('0'),
                    'bonus_ledger_amount': Decimal('0'),
                    'room_ledger_amount': Decimal('0'),
                    'unique_customers': set(),
                    'unique_tables': set(),
                    'trashed_seconds': 0,
                    'trashed_count': 0,
                }
            
            agg = agg_dict[key]
            
            # 获取服务信息
            service_id = record.get('assistant_service_id')
            income_seconds = self.safe_int(record.get('income_seconds', 0))
            ledger_amount = self.safe_decimal(record.get('ledger_amount', 0))
            skill_id = record.get('skill_id')
            member_id = record.get('member_id')
            table_id = record.get('table_id')
            
            # 判断课程类型
            course_type = self.get_course_type(skill_id) if skill_id else CourseType.BASE
            is_base = course_type == CourseType.BASE
            is_bonus = course_type == CourseType.BONUS
            is_room = course_type == CourseType.ROOM
            
            # 检查是否被废除（使用 _ex 表的 is_trash 标记）
            is_trashed = bool(record.get('is_trash', 0))
            
            if is_trashed:
                # 废除记录：直接用服务记录的 income_seconds 作为废除时长
                agg['trashed_seconds'] += income_seconds
                agg['trashed_count'] += 1
            else:
                # 正常记录累加
                agg['total_service_count'] += 1
                agg['total_seconds'] += income_seconds
                agg['total_ledger_amount'] += ledger_amount
                
                if is_base:
                    agg['base_service_count'] += 1
                    agg['base_seconds'] += income_seconds
                    agg['base_ledger_amount'] += ledger_amount
                elif is_bonus:
                    agg['bonus_service_count'] += 1
                    agg['bonus_seconds'] += income_seconds
                    agg['bonus_ledger_amount'] += ledger_amount
                elif is_room:
                    agg['room_service_count'] += 1
                    agg['room_seconds'] += income_seconds
                    agg['room_ledger_amount'] += ledger_amount
            
            # 客户和台桌去重统计（不论是否废除）
            if member_id and not self.is_guest(member_id):
                agg['unique_customers'].add(member_id)
            if table_id:
                agg['unique_tables'].add(table_id)
        
        # 转换为列表并计算派生字段
        result = []
        for key, agg in agg_dict.items():
            # 计算小时数
            agg['total_hours'] = self.seconds_to_hours(agg['total_seconds'])
            agg['base_hours'] = self.seconds_to_hours(agg['base_seconds'])
            agg['bonus_hours'] = self.seconds_to_hours(agg['bonus_seconds'])
            agg['room_hours'] = self.seconds_to_hours(agg['room_seconds'])
            
            # 转换set为count
            agg['unique_customers'] = len(agg['unique_customers'])
            agg['unique_tables'] = len(agg['unique_tables'])
            
            result.append(agg)
        
        return result

    # ==========================================================================
    # 定档折算惩罚 — 纯函数（静态方法，不依赖数据库）
    # ==========================================================================

    @staticmethod
    def detect_overlap_violations(
        service_records: List[Dict[str, Any]],
        penalty_areas: Set[str],
    ) -> Dict[Tuple[int, date], List[Dict[str, Any]]]:
        """
        检测同一台桌同一时间段超过 2 名助教挂台的违规。

        输入:
            service_records: 服务记录列表，每条需包含
                assistant_id, table_id, table_area, start_time, end_time, service_date
            penalty_areas: 需要检测的区域集合（如 PENALTY_AREAS）

        输出:
            {(assistant_id, service_date): [violation_info, ...]}
            violation_info 包含 table_id, overlap_count, assistant_ids 等

        算法：
            1. 过滤出属于惩罚区域的记录
            2. 按 (table_id, service_date) 分组
            3. 对每组用扫描线算法检测最大同时在线助教数
            4. 若峰值 > 2，标记所有参与助教为违规
        """
        # 过滤：仅保留惩罚区域内的记录，且时间信息完整
        filtered = []
        for r in service_records:
            area = r.get("table_area", "")
            if area not in penalty_areas:
                continue
            if r.get("start_time") is None or r.get("end_time") is None:
                continue
            filtered.append(r)

        # 按 (table_id, service_date) 分组
        groups: Dict[Tuple[int, date], List[Dict[str, Any]]] = defaultdict(list)
        for r in filtered:
            key = (r["table_id"], r["service_date"])
            groups[key].append(r)

        violations: Dict[Tuple[int, date], List[Dict[str, Any]]] = defaultdict(list)

        for (table_id, svc_date), records in groups.items():
            if len(records) <= 2:
                # 不可能超过 2 名助教
                continue

            # 扫描线：收集所有事件点，检测峰值
            events: List[Tuple[Any, int, int]] = []  # (time, +1/-1, assistant_id)
            for r in records:
                aid = r["assistant_id"]
                events.append((r["start_time"], 1, aid))
                events.append((r["end_time"], -1, aid))

            # 按时间排序；同一时刻先处理 +1（开始）再处理 -1（结束），
            # 这样"恰好交接"也算重叠
            events.sort(key=lambda e: (e[0], -e[1]))

            # 扫描：追踪当前在线助教集合
            active: Dict[int, int] = defaultdict(int)  # assistant_id -> 计数
            max_overlap = 0
            max_overlap_aids: Set[int] = set()

            for t, delta, aid in events:
                active[aid] += delta
                if active[aid] <= 0:
                    del active[aid]

                current_count = len(active)
                if current_count > max_overlap:
                    max_overlap = current_count
                    max_overlap_aids = set(active.keys())
                elif current_count == max_overlap and current_count > 2:
                    max_overlap_aids |= set(active.keys())

            if max_overlap > 2:
                violation_info = {
                    "table_id": table_id,
                    "service_date": svc_date,
                    "overlap_count": max_overlap,
                    "assistant_ids": max_overlap_aids,
                }
                # 为每个涉及的助教记录违规
                for aid in max_overlap_aids:
                    violations[(aid, svc_date)].append(violation_info)

        return dict(violations)

    @staticmethod
    def compute_penalty_minutes(
        actual_minutes: Decimal,
        per_hour_contribution: Decimal,
        threshold: Decimal = Decimal("24"),
    ) -> Decimal:
        """
        计算惩罚分钟数（纯函数）。

        规则：
            - per_hour_contribution >= threshold → 0（满额计入）
            - per_hour_contribution < threshold →
                actual_minutes × (1 - per_hour_contribution / threshold)
            - per_hour_contribution < 0 → 视为 0（防御性编程）

        结果范围：[0, actual_minutes]
        """
        if actual_minutes <= 0:
            return Decimal("0")

        # 防御性：负值视为 0
        phc = max(per_hour_contribution, Decimal("0"))

        if phc >= threshold:
            return Decimal("0")

        # penalty = actual_minutes × (1 - phc / threshold)
        ratio = Decimal("1") - phc / threshold
        penalty = actual_minutes * ratio

        # 确保结果在 [0, actual_minutes] 范围内
        penalty = max(Decimal("0"), min(penalty, actual_minutes))
        return penalty


# 便于外部导入
__all__ = ['AssistantDailyTask', 'PENALTY_AREAS']
