# -*- coding: utf-8 -*-
"""
DWS 助教项目标签任务

按时间窗口计算每位助教在四大项目（BILLIARD/SNOOKER/MAHJONG/KTV）的
工作时长占比，占比≥25% 则分配标签。

数据链路：
    dwd_assistant_service_log (income_seconds)
    → JOIN dim_table (site_table_id → table_id, scd2_is_current=1)
    → get_area_category(area_name, table_name)
    → 按 category_code 汇总 → 计算占比 → 写入 dws_assistant_project_tag

目标表：
    dws.dws_assistant_project_tag

更新策略：
    全量删除重建（按 site_id 删除后重新插入所有时间窗口）
"""
from __future__ import annotations

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

from tasks.dws.base_dws_task import BaseDwsTask, TimeWindow
from neozqyy_shared.datetime_utils import biz_date_sql_expr

# 只计算四大项目，排除 SPECIAL/OTHER
VALID_CATEGORIES = {"BILLIARD", "SNOOKER", "MAHJONG", "KTV"}

# 助教看板的 6 个时间窗口
ASSISTANT_WINDOWS = [
    TimeWindow.THIS_MONTH,
    TimeWindow.THIS_QUARTER,
    TimeWindow.LAST_MONTH,
    TimeWindow.LAST_3_MONTHS_EXCL_CURRENT,
    TimeWindow.LAST_QUARTER,
    TimeWindow.LAST_6_MONTHS,
]

TAG_THRESHOLD = Decimal("0.25")


class AssistantProjectTagTask(BaseDwsTask):
    """助教项目标签 ETL 任务"""

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

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

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

    def extract(self, context) -> Dict[str, Any]:
        site_id = context.store_id
        self.logger.info("%s: 提取助教服务数据", self.get_task_code())

        # 加载配置（cfg_area_category 等）
        self.load_config_cache()

        # 提取台桌信息（用于 get_area_category 的 table_name 参数）
        table_info = self._extract_table_info(site_id)

        # 按时间窗口提取助教服务时长
        window_data: Dict[str, List[Dict]] = {}
        for window in ASSISTANT_WINDOWS:
            time_range = self.get_time_window_range(window)
            rows = self._extract_assistant_durations(
                site_id, time_range.start, time_range.end
            )
            window_data[window.value] = rows

        return {
            "window_data": window_data,
            "table_info": table_info,
            "site_id": site_id,
        }

    def _extract_table_info(self, site_id: int) -> Dict[int, Dict[str, Any]]:
        """提取台桌维度信息"""
        sql = """
            SELECT table_id, table_name, site_table_area_name AS area_name
            FROM dwd.dim_table
            WHERE site_id = %s AND scd2_is_current = 1
        """
        rows = self.db.query(sql, (site_id,))
        return {r["table_id"]: dict(r) for r in (rows or [])}

    def _extract_assistant_durations(
        self, site_id: int, start_date: date, end_date: date
    ) -> List[Dict[str, Any]]:
        """提取助教服务时长明细（按助教+台桌聚合）"""
        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.site_assistant_id AS assistant_id,
                asl.site_table_id AS table_id,
                COALESCE(SUM(asl.income_seconds), 0) AS duration_seconds
            FROM dwd.dwd_assistant_service_log asl
            WHERE asl.site_id = %(site_id)s
              AND {biz_expr} >= %(start_date)s
              AND {biz_expr} <= %(end_date)s
              AND asl.is_delete = 0
            GROUP BY asl.site_assistant_id, asl.site_table_id
        """
        rows = self.db.query(sql, {
            "site_id": site_id,
            "start_date": start_date,
            "end_date": end_date,
        })
        return [dict(r) for r in rows] if rows else []

    def transform(self, extracted: Dict[str, Any], context) -> List[Dict[str, Any]]:
        table_info = extracted["table_info"]
        site_id = extracted["site_id"]
        tenant_id = getattr(context, "tenant_id", 0) or 0
        results: List[Dict[str, Any]] = []

        for window_value, rows in extracted["window_data"].items():
            # 按助教汇总各项目时长
            # assistant_id → category_code → seconds
            assistant_cats: Dict[int, Dict[str, int]] = {}

            for row in rows:
                aid = row["assistant_id"]
                tid = row["table_id"]
                secs = self.safe_int(row["duration_seconds"])
                if secs <= 0:
                    continue

                # 通过 dim_table 获取区域和台桌名
                tinfo = table_info.get(tid, {})
                area_name = tinfo.get("area_name")
                table_name = tinfo.get("table_name")
                cat = self.get_area_category(area_name, table_name)
                code = cat.get("category_code", "OTHER")

                # 只计算四大项目
                if code not in VALID_CATEGORIES:
                    continue

                if aid not in assistant_cats:
                    assistant_cats[aid] = {}
                assistant_cats[aid][code] = assistant_cats[aid].get(code, 0) + secs

            # 计算占比并生成记录
            for aid, cats in assistant_cats.items():
                total = sum(cats.values())
                if total <= 0:
                    continue

                for code, secs in cats.items():
                    pct = Decimal(str(secs)) / Decimal(str(total))
                    pct = pct.quantize(Decimal("0.0001"))
                    cat_info = self._get_category_display(code)

                    results.append({
                        "site_id": site_id,
                        "tenant_id": tenant_id,
                        "assistant_id": aid,
                        "time_window": window_value,
                        "category_code": code,
                        "category_name": cat_info["category_name"],
                        "short_name": cat_info["short_name"],
                        "duration_seconds": secs,
                        "total_seconds": total,
                        "percentage": float(pct),
                        "is_tagged": pct >= TAG_THRESHOLD,
                    })

        self.logger.info(
            "%s: 生成 %d 条标签记录（其中 %d 条达标）",
            self.get_task_code(),
            len(results),
            sum(1 for r in results if r["is_tagged"]),
        )
        return results

    def _get_category_display(self, code: str) -> Dict[str, str]:
        """从配置缓存获取分类的显示名和简写"""
        cache = self.load_config_cache()
        for key, cat in cache.area_categories.items():
            if cat.get("category_code") == code:
                return {
                    "category_name": cat.get("display_name") or cat.get("category_name", code),
                    "short_name": cat.get("short_name", code[:1]),
                }
        # 兜底
        fallback = {
            "BILLIARD": ("🎱 中式/追分", "🎱"),
            "SNOOKER": ("斯诺克", "斯"),
            "MAHJONG": ("🀄 麻将/棋牌", "🀄"),
            "KTV": ("🎤 团建/K歌", "🎤"),
        }
        name, short = fallback.get(code, (code, code[:1]))
        return {"category_name": name, "short_name": short}

    def load(self, transformed, context) -> dict:
        if not transformed:
            return {"status": "SUCCESS", "counts": {"inserted": 0, "deleted": 0}}

        site_id = transformed[0]["site_id"]

        # 全量删除该门店的标签数据后重建
        delete_sql = "DELETE FROM dws.dws_assistant_project_tag WHERE site_id = %s"
        deleted = self.db.execute(delete_sql, (site_id,))

        insert_sql = """
            INSERT INTO dws.dws_assistant_project_tag (
                site_id, tenant_id, assistant_id, time_window,
                category_code, category_name, short_name,
                duration_seconds, total_seconds, percentage, is_tagged,
                computed_at, created_at, updated_at
            ) VALUES (
                %(site_id)s, %(tenant_id)s, %(assistant_id)s, %(time_window)s,
                %(category_code)s, %(category_name)s, %(short_name)s,
                %(duration_seconds)s, %(total_seconds)s, %(percentage)s, %(is_tagged)s,
                NOW(), NOW(), NOW()
            )
        """
        for row in transformed:
            self.db.execute(insert_sql, row)

        self.logger.info(
            "%s: 删除 %d 条，插入 %d 条",
            self.get_task_code(), deleted, len(transformed),
        )
        return {
            "status": "SUCCESS",
            "counts": {"inserted": len(transformed), "deleted": deleted},
        }
