# -*- coding: utf-8 -*-
"""
DWS 客户项目标签任务

按每位客户最近 N 次消费（开台记录）计算四大项目（BILLIARD/SNOOKER/MAHJONG/KTV）
的消费时长占比，占比≥25% 则分配标签。散客（member_id=0）不参与。

设计思路：
    不按固定日期窗口（30天/60天），而按每位客户最近的消费记录数量取数，
    避免长期未到店或来店频率不稳定的客户标签丢失。

数据链路：
    dwd_table_fee_log (ledger_count)
    → ROW_NUMBER() OVER (PARTITION BY member_id ORDER BY ledger_end_time DESC)
    → 取最近 LAST_N_VISITS 条记录
    → JOIN dim_table (site_table_id → table_id, scd2_is_current=1)
    → get_area_category(area_name, table_name)
    → 按 category_code 汇总 → 计算占比 → 写入 dws_member_project_tag

目标表：
    dws.dws_member_project_tag

更新策略：
    全量删除重建（按 site_id 删除后重新插入）
"""
from __future__ import annotations

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

from tasks.dws.base_dws_task import BaseDwsTask

# 只计算四大项目
VALID_CATEGORIES = {"BILLIARD", "SNOOKER", "MAHJONG", "KTV"}

# 取每位客户最近 30 次消费（开台记录）
LAST_N_VISITS = 30

# 写入 time_window 字段的枚举值
TIME_WINDOW_VALUE = "LAST_30_VISITS"

TAG_THRESHOLD = Decimal("0.25")


class MemberProjectTagTask(BaseDwsTask):
    """客户项目标签 ETL 任务"""

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

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

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

    def extract(self, context) -> Dict[str, Any]:
        site_id = context.store_id
        self.logger.info("%s: 提取客户最近 %d 次消费的台费时长数据",
                         self.get_task_code(), LAST_N_VISITS)

        self.load_config_cache()
        table_info = self._extract_table_info(site_id)
        rows = self._extract_member_durations(site_id)

        return {
            "rows": rows,
            "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_member_durations(self, site_id: int) -> List[Dict[str, Any]]:
        """按每位客户最近 N 次消费提取台费时长明细，排除散客。

        使用 ROW_NUMBER() 按 member_id 分区、ledger_end_time 倒序排名，
        取最近 LAST_N_VISITS 条记录后再按 (member_id, table_id) 聚合。
        """
        sql = """
            WITH ranked AS (
                SELECT tfl.member_id,
                       tfl.site_table_id AS table_id,
                       tfl.ledger_count  AS duration_seconds,
                       ROW_NUMBER() OVER (
                           PARTITION BY tfl.member_id
                           ORDER BY tfl.ledger_end_time DESC
                       ) AS rn
                FROM dwd.dwd_table_fee_log tfl
                WHERE tfl.site_id = %(site_id)s
                  AND COALESCE(tfl.is_delete, 0) = 0
                  AND tfl.member_id IS NOT NULL
                  AND tfl.member_id != 0
            )
            SELECT member_id,
                   table_id,
                   COALESCE(SUM(duration_seconds), 0) AS duration_seconds
            FROM ranked
            WHERE rn <= %(last_n)s
            GROUP BY member_id, table_id
        """
        rows = self.db.query(sql, {
            "site_id": site_id,
            "last_n": LAST_N_VISITS,
        })
        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]] = []

        # member_id → category_code → seconds
        member_cats: Dict[int, Dict[str, int]] = {}

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

            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 mid not in member_cats:
                member_cats[mid] = {}
            member_cats[mid][code] = member_cats[mid].get(code, 0) + secs

        for mid, cats in member_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,
                    "member_id": mid,
                    "time_window": TIME_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 条达标），基于每客户最近 %d 次消费",
            self.get_task_code(),
            len(results),
            sum(1 for r in results if r["is_tagged"]),
            LAST_N_VISITS,
        )
        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_member_project_tag WHERE site_id = %s"
        deleted = self.db.execute(delete_sql, (site_id,))

        insert_sql = """
            INSERT INTO dws.dws_member_project_tag (
                site_id, tenant_id, member_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, %(member_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},
        }
