# -*- coding: utf-8 -*-
"""
财务区域日粒度汇总任务

功能说明：
    以 (site_id, stat_date, area_code) 为粒度，按桌台区域聚合当日财务数据。
    输出 9 行：7 个具体区域 (hallA~ktv) + hall（历史兼容）+ all。

数据来源：
    - dwd_settlement_head + dim_table (scd2_is_current=1)：结算单 + 桌台区域
    - dws_finance_daily_summary：全局现金流/充值/卡消费（仅 all 行使用）

目标表：
    dws.dws_finance_area_daily

更新策略：
    - 更新频率：每小时更新当日数据
    - 幂等方式：delete-before-insert（按 site_id + stat_date 删除后插入 9 行）

业务规则：
    - 收入恒等式：gross_amount = table_fee + goods + assistant_pd + assistant_cx
    - 优惠恒等式：discount_total = groupbuy + vip + manual + gift_card + rounding + other
    - 确认收入：confirmed_income = gross_amount - discount_total
    - 非 all 行现金流/卡消费/充值字段 = 0
    - hall 行 = 各具体区域之和（历史兼容）
    - all 行 = 各具体区域之和（收入/优惠），现金流/充值/卡消费来自 dws_finance_daily_summary
      无台桌订单和补时长等 all-only 台区只合入 all，不合入具体区域
    - settle_type IN (1, 3) 过滤
    - discount_gift_card 使用赠送卡消费金额口径

Requirements: 3.1, 3.2, 3.3, 3.4, 3.5, 3.6, 3.7, 3.8, 3.9, 8.1, 8.2
"""

from __future__ import annotations

import logging
from collections import defaultdict
from datetime import date
from decimal import Decimal
from typing import Any, Dict, List

from neozqyy_shared.area_mapping import (
    ALL_AREA_CODES,
    SPECIFIC_AREA_CODES,
    resolve_area_code,
)
from neozqyy_shared.datetime_utils import biz_date_sql_expr

from .base_dws_task import TaskContext
from .finance_base_task import FinanceBaseTask

logger = logging.getLogger(__name__)

# 收入字段（按区域聚合）
_REVENUE_FIELDS = [
    "table_fee_amount",
    "goods_amount",
    "assistant_pd_amount",
    "assistant_cx_amount",
    "gross_amount",
]

# 优惠字段（按区域聚合）
_DISCOUNT_FIELDS = [
    "discount_groupbuy",
    "discount_vip",
    "discount_manual",
    "discount_gift_card",
    "discount_rounding",
    "discount_other",
    "discount_total",
]

# 现金流字段（仅 all 行有值）
_CASHFLOW_FIELDS = [
    "cash_pay_amount",
    "cash_paper_amount",
    "scan_pay_amount",
    "groupbuy_pay_amount",
    "recharge_cash_inflow",
    "cash_inflow_total",
    "cash_outflow_total",
    "cash_balance_change",
]

# 卡消费字段（仅 all 行有值）
_CARD_FIELDS = [
    "card_consume_total",
    "recharge_card_consume",
    "gift_card_consume",
]

# 充值字段（仅 all 行有值）
_RECHARGE_FIELDS = [
    "recharge_cash",
    "first_recharge_cash",
    "renewal_cash",
]

# 所有仅 all 行有值的字段
_ALL_ONLY_FIELDS = _CASHFLOW_FIELDS + _CARD_FIELDS + _RECHARGE_FIELDS

# 按区域聚合的字段（收入 + 优惠 + order_count + member_order_count）
_AREA_AGG_FIELDS = _REVENUE_FIELDS + _DISCOUNT_FIELDS + ["confirmed_income", "order_count", "member_order_count"]

# 计数字段（int 类型，非 Decimal）— 用于 _build_area_row / _build_sum_row 区分类型转换
_COUNT_FIELDS = {"order_count", "member_order_count"}

_ZERO = Decimal("0")

# 已知不属于看板 7 个具体区域、但应合入 all 的物理台区。
_ALL_ONLY_AREA_NAMES = {"补时长", "虚拟台"}


class FinanceAreaDailyTask(FinanceBaseTask):
    """
    财务区域日粒度汇总任务

    按 (site_id, stat_date, area_code) 粒度预计算收入、优惠、现金流等数据。
    每次写入 9 行（hallA~ktv + hall + all）。
    """

    DATE_COL = "stat_date"

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

    def get_target_table(self) -> str:
        return "dws.dws_finance_area_daily"

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

    # ======================================================================
    # Extract
    # ======================================================================

    def extract(self, context: TaskContext) -> Dict[str, Any]:
        """提取结算单（含区域）和全局现金流/充值/卡消费数据。"""
        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. 结算单 + dim_table 区域映射
        settlement_rows = self._extract_settlement_with_area(
            site_id, start_date, end_date
        )

        # 2. 全局现金流/充值/卡消费（从现有 dws_finance_daily_summary）
        global_summary = self._extract_global_summary(site_id, start_date, end_date)

        return {
            "settlement_rows": settlement_rows,
            "global_summary": global_summary,
            "start_date": start_date,
            "end_date": end_date,
            "site_id": site_id,
        }

    def _extract_settlement_with_area(
        self, site_id: int, start_date: date, end_date: date
    ) -> List[Dict[str, Any]]:
        """从 dwd_settlement_head + dim_table 提取结算单（含区域名称）。"""
        cutoff = self.config.get("app.business_day_start_hour", 8)
        biz_expr = biz_date_sql_expr("sh.pay_time", cutoff)
        sql = f"""
            SELECT
                {biz_expr}                          AS stat_date,
                sh.table_id                         AS table_id,
                dt.site_table_area_name             AS area_name,
                sh.settle_type,
                -- 收入
                sh.table_charge_money               AS table_fee_amount,
                sh.goods_money                      AS goods_amount,
                sh.assistant_pd_money               AS assistant_pd_amount,
                sh.assistant_cx_money               AS assistant_cx_amount,
                (sh.table_charge_money + sh.goods_money
                 + sh.assistant_pd_money + sh.assistant_cx_money)
                                                    AS gross_amount,
                -- 优惠原始字段
                sh.coupon_amount,
                sh.pl_coupon_sale_amount,
                sh.adjust_amount,
                sh.member_discount_amount,
                sh.rounding_amount,
                sh.gift_card_amount,
                -- 会员标识（散客 member_id ≤ 0，会员 member_id > 0，详见 CLAUDE.md DWS 规范）
                sh.member_id
            FROM dwd.dwd_settlement_head sh
            LEFT JOIN dwd.dim_table dt
                ON dt.table_id = sh.table_id
               AND dt.site_id  = sh.site_id
               AND dt.scd2_is_current = 1
            WHERE sh.site_id = %s
              AND {biz_expr} >= %s
              AND {biz_expr} <= %s
              AND sh.settle_type IN (1, 3)
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []

    def _extract_global_summary(
        self, site_id: int, start_date: date, end_date: date
    ) -> List[Dict[str, Any]]:
        """从 dws_finance_daily_summary 提取全局现金流/充值/卡消费字段。"""
        sql = """
            SELECT
                stat_date,
                cash_pay_amount,
                cash_paper_amount,
                scan_pay_amount,
                groupbuy_pay_amount,
                recharge_cash_inflow,
                cash_inflow_total,
                cash_outflow_total,
                cash_balance_change,
                card_consume_total,
                recharge_card_consume,
                gift_card_consume,
                recharge_cash,
                first_recharge_amount AS first_recharge_cash,
                renewal_amount        AS renewal_cash
            FROM dws.dws_finance_daily_summary
            WHERE site_id = %s
              AND stat_date >= %s
              AND stat_date <= %s
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []

    # ======================================================================
    # Transform（纯函数，不依赖数据库）
    # ======================================================================

    def transform(
        self, extracted: Dict[str, Any], context: TaskContext
    ) -> List[Dict[str, Any]]:
        """按区域聚合收入和优惠，构建 9 行输出。"""
        settlement_rows = extracted["settlement_rows"]
        global_summary = extracted["global_summary"]
        site_id = extracted["site_id"]

        return transform_area_daily(
            settlement_rows=settlement_rows,
            global_summary=global_summary,
            site_id=site_id,
            tenant_id=self.config.get("app.tenant_id", site_id),
            safe_decimal_fn=self.safe_decimal,
            logger=self.logger,
        )

    # ======================================================================
    # Load（覆盖基类：按 site_id + stat_date 删除，而非仅按 DATE_COL）
    # ======================================================================

    def load(self, transformed: List[Dict[str, Any]], context: TaskContext) -> dict:
        """delete-before-insert：按 site_id + stat_date 删除后插入 9 行。"""
        if not transformed:
            return {
                "counts": {
                    "fetched": 0,
                    "inserted": 0,
                    "updated": 0,
                    "skipped": 0,
                    "errors": 0,
                }
            }

        # 目标表已含 schema 前缀
        target_table = self.get_target_table()

        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 = f"""
            DELETE FROM {target_table}
            WHERE site_id = %s
              AND stat_date >= %s
              AND stat_date <= %s
        """

        columns = list(transformed[0].keys())
        cols_str = ", ".join(columns)
        placeholders = ", ".join(["%s"] * len(columns))
        insert_sql = f"INSERT INTO {target_table} ({cols_str}) VALUES ({placeholders})"

        with self.db.conn.cursor() as cur:
            cur.execute(delete_sql, (context.store_id, start_date, end_date))
            deleted = cur.rowcount

            inserted = 0
            for row in transformed:
                values = [row.get(col) for col in columns]
                cur.execute(insert_sql, values)
                inserted += cur.rowcount

        self.logger.info(
            "%s: 删除 %d 行，插入 %d 行",
            self.get_task_code(),
            deleted,
            inserted,
        )

        return {
            "counts": {
                "fetched": len(transformed),
                "inserted": inserted,
                "updated": 0,
                "skipped": 0,
                "errors": 0,
            },
            "extra": {"deleted": deleted},
        }


# ======================================================================
# 纯函数：transform 核心逻辑（方便属性测试直接调用）
# ======================================================================


def transform_area_daily(
    settlement_rows: List[Dict[str, Any]],
    global_summary: List[Dict[str, Any]],
    site_id: int,
    tenant_id: int,
    safe_decimal_fn=None,
    logger=None,
) -> List[Dict[str, Any]]:
    """按区域聚合结算单，构建 9 行输出。

    这是一个纯函数（不依赖数据库），方便属性测试直接调用。

    Args:
        settlement_rows: 结算单行（含 area_name、settle_type 等）
        global_summary: 全局现金流/充值/卡消费（按 stat_date 索引）
        site_id: 门店 ID
        tenant_id: 租户 ID
        safe_decimal_fn: Decimal 安全转换函数（默认使用内置实现）
        logger: 日志记录器（可选）

    Returns:
        9 × N 行字典列表（N = 涉及的日期数）
    """
    if safe_decimal_fn is None:
        safe_decimal_fn = _safe_decimal
    if logger is None:
        logger = logging.getLogger(__name__)

    # 全局汇总按日期索引
    global_index: Dict[date, Dict[str, Any]] = {}
    for row in global_summary:
        sd = row.get("stat_date")
        if sd is not None:
            global_index[sd] = row

    # ── 第一步：按 (stat_date, area_code) 聚合结算单 ──────────────
    # area_agg[stat_date][area_code] = {field: Decimal, ...}
    area_agg: Dict[date, Dict[str, Dict[str, Decimal]]] = defaultdict(
        lambda: defaultdict(lambda: defaultdict(lambda: _ZERO))
    )
    # 收集所有涉及的日期
    all_dates: set[date] = set()
    # 未知/无具体区域计数（汇总后一次性输出，避免逐行日志噪音）
    _unknown_area_counts: Dict[str, int] = defaultdict(int)
    _all_only_area_counts: Dict[str, int] = defaultdict(int)

    for row in settlement_rows:
        sd = row.get("stat_date")
        if sd is None:
            continue
        all_dates.add(sd)

        # settle_type 过滤（extract 已过滤，transform 层再次防御）
        settle_type = row.get("settle_type")
        if settle_type not in (1, 3):
            continue

        area_name = row.get("area_name")
        table_id = row.get("table_id")
        area_code = resolve_area_code(area_name)

        if area_code is None:
            unmatched_label = _format_unmatched_area_label(area_name, table_id)
            if _is_all_only_area(area_name, table_id):
                _all_only_area_counts[unmatched_label] += 1
            else:
                _unknown_area_counts[unmatched_label] += 1

        # 提取金额
        table_fee = safe_decimal_fn(row.get("table_fee_amount", 0))
        goods = safe_decimal_fn(row.get("goods_amount", 0))
        pd_amount = safe_decimal_fn(row.get("assistant_pd_amount", 0))
        cx_amount = safe_decimal_fn(row.get("assistant_cx_amount", 0))
        gross = table_fee + goods + pd_amount + cx_amount

        coupon = safe_decimal_fn(row.get("coupon_amount", 0))
        pl_coupon_sale = safe_decimal_fn(row.get("pl_coupon_sale_amount", 0))
        adjust = safe_decimal_fn(row.get("adjust_amount", 0))
        member_discount = safe_decimal_fn(row.get("member_discount_amount", 0))
        rounding = safe_decimal_fn(row.get("rounding_amount", 0))
        gift_card_amount = safe_decimal_fn(row.get("gift_card_amount", 0))

        # 团购优惠 = coupon_amount - 团购支付金额
        groupbuy_pay = pl_coupon_sale if pl_coupon_sale > 0 else _ZERO
        discount_groupbuy = max(coupon - groupbuy_pay, _ZERO)

        # 赠送卡消费金额口径（与现有 ETL 一致）
        discount_gift_card = gift_card_amount

        # 手动调整 = adjust_amount（简化：不拆分大客户，区域级无法区分）
        discount_manual = adjust

        discount_total = (
            discount_groupbuy
            + member_discount
            + discount_manual
            + discount_gift_card
            + rounding
        )
        # discount_other 在区域级暂为 0（adjust 全部归入 manual）
        discount_other = _ZERO

        # 重新计算 discount_total 确保恒等式
        discount_total = (
            discount_groupbuy
            + member_discount
            + discount_manual
            + discount_gift_card
            + rounding
            + discount_other
        )

        confirmed_income = gross - discount_total

        # 会员订单判定（散客 member_id ≤ 0 不计入，与全店 finance_base_task.py 逻辑一致）
        member_id_raw = row.get("member_id")
        try:
            is_member_order = 1 if (member_id_raw is not None and int(member_id_raw) > 0) else 0
        except (TypeError, ValueError):
            is_member_order = 0

        fields = {
            "table_fee_amount": table_fee,
            "goods_amount": goods,
            "assistant_pd_amount": pd_amount,
            "assistant_cx_amount": cx_amount,
            "gross_amount": gross,
            "discount_groupbuy": discount_groupbuy,
            "discount_vip": member_discount,
            "discount_manual": discount_manual,
            "discount_gift_card": discount_gift_card,
            "discount_rounding": rounding,
            "discount_other": discount_other,
            "discount_total": discount_total,
            "confirmed_income": confirmed_income,
            "order_count": 1,
            "member_order_count": is_member_order,
        }

        # 累加到具体区域（未匹配区域的订单累加到 _unknown 桶，供 all 行汇总使用）
        # CHANGE 2026-04-23 | 修复 pre-existing bug：area_code 为 None 的订单之前既不计入具体区域也不计入 all，
        #   导致全店 order_count/member_order_count > 各区域之和。现在 _unknown 桶会合入 all 汇总。
        if area_code is not None:
            bucket = area_agg[sd][area_code]
        else:
            bucket = area_agg[sd]["_unknown"]
        for k, v in fields.items():
            bucket[k] = bucket[k] + v

    # 汇总输出 all-only 区域（无台桌订单、补时长等），这些记录合入 all 属正常口径。
    if _all_only_area_counts:
        summary = ", ".join(f"'{k}': {v}次" for k, v in _all_only_area_counts.items())
        logger.info(
            "DWS_FINANCE_AREA_DAILY: 共 %d 条结算单无具体区域（已计入 all，不计入任何具体区域）: %s",
            sum(_all_only_area_counts.values()),
            summary,
        )

    # 汇总输出真正未知区域名称（避免逐行 warning 刷屏）
    if _unknown_area_counts:
        summary = ", ".join(f"'{k}': {v}次" for k, v in _unknown_area_counts.items())
        logger.warning(
            "DWS_FINANCE_AREA_DAILY: 共 %d 条结算单区域未匹配（已计入 all 但不计入任何具体区域，请检查 dim_table/AREA_LABEL_MAP）: %s",
            sum(_unknown_area_counts.values()),
            summary,
        )

    # 也收集 global_summary 中的日期
    for sd in global_index:
        all_dates.add(sd)

    # ── 第二步：构建 9 行输出 ─────────────────────────────────────
    results: List[Dict[str, Any]] = []

    for sd in sorted(all_dates):
        day_agg = area_agg.get(sd, {})

        # 计算各具体区域行
        specific_rows: Dict[str, Dict[str, Any]] = {}
        for ac in SPECIFIC_AREA_CODES:
            bucket = day_agg.get(ac, {})
            row_data = _build_area_row(
                site_id=site_id,
                tenant_id=tenant_id,
                stat_date=sd,
                area_code=ac,
                agg=bucket,
            )
            specific_rows[ac] = row_data

        # CHANGE 2026-03-29 | hall = 台球大厅（hallA+hallB+hallC），不含 vip/snooker/mahjong/ktv
        _HALL_CODES = ("hallA", "hallB", "hallC")
        hall_source = [specific_rows[ac] for ac in _HALL_CODES if ac in specific_rows]
        hall_row = _build_sum_row(
            site_id=site_id,
            tenant_id=tenant_id,
            stat_date=sd,
            area_code="hall",
            source_rows=hall_source,
        )

        # all 行 = 各具体区域之和 + 未匹配区域订单（_unknown 桶）+ 全局现金流/充值/卡消费
        # CHANGE 2026-04-23 | 修复全店 vs 区域和差异：未匹配区域订单通过 _unknown 桶合入 all
        all_sources = list(specific_rows.values())
        unknown_bucket = day_agg.get("_unknown", {})
        if unknown_bucket:
            all_sources.append(_build_area_row(
                site_id=site_id,
                tenant_id=tenant_id,
                stat_date=sd,
                area_code="_unknown",
                agg=unknown_bucket,
            ))
        all_row = _build_sum_row(
            site_id=site_id,
            tenant_id=tenant_id,
            stat_date=sd,
            area_code="all",
            source_rows=all_sources,
        )
        # 填充全局现金流/充值/卡消费
        gs = global_index.get(sd, {})
        for field in _ALL_ONLY_FIELDS:
            all_row[field] = safe_decimal_fn(gs.get(field, 0))

        # 输出顺序：all, hall, hallA, hallB, hallC, vip, snooker, mahjong, ktv
        results.append(all_row)
        results.append(hall_row)
        for ac in SPECIFIC_AREA_CODES:
            results.append(specific_rows[ac])

    return results


def _build_area_row(
    site_id: int,
    tenant_id: int,
    stat_date: date,
    area_code: str,
    agg: Dict[str, Decimal],
) -> Dict[str, Any]:
    """构建单个区域行。"""
    row: Dict[str, Any] = {
        "site_id": site_id,
        "tenant_id": tenant_id,
        "stat_date": stat_date,
        "area_code": area_code,
    }
    # 收入 + 优惠 + confirmed_income + order_count + member_order_count
    for field in _AREA_AGG_FIELDS:
        if field in _COUNT_FIELDS:
            row[field] = int(agg.get(field, 0))
        else:
            row[field] = agg.get(field, _ZERO)
    # 非 all 行：现金流/卡消费/充值 = 0
    for field in _ALL_ONLY_FIELDS:
        row[field] = _ZERO
    return row


def _build_sum_row(
    site_id: int,
    tenant_id: int,
    stat_date: date,
    area_code: str,
    source_rows: List[Dict[str, Any]],
) -> Dict[str, Any]:
    """构建 hall/all 汇总行（收入/优惠 = 各具体区域之和）。"""
    row: Dict[str, Any] = {
        "site_id": site_id,
        "tenant_id": tenant_id,
        "stat_date": stat_date,
        "area_code": area_code,
    }
    for field in _AREA_AGG_FIELDS:
        if field in _COUNT_FIELDS:
            row[field] = sum(int(r.get(field, 0)) for r in source_rows)
        else:
            row[field] = sum(
                (r.get(field, _ZERO) for r in source_rows), _ZERO
            )
    # 非 all 行：现金流/卡消费/充值 = 0（all 行由调用方覆盖）
    for field in _ALL_ONLY_FIELDS:
        row[field] = _ZERO
    return row


def _safe_decimal(value: Any, default: Decimal = _ZERO) -> Decimal:
    """内置的安全 Decimal 转换（供纯函数使用）。"""
    if value is None:
        return default
    try:
        return Decimal(str(value))
    except Exception:
        return default


def _is_all_only_area(area_name: Any, table_id: Any) -> bool:
    """判断结算单是否属于无具体区域但应合入 all 的正常口径。

    CHANGE 2026-05-02 | 扩大豁免规则，避免噪音 WARNING：
    - "补时长" / "虚拟台" 的带数字/空格变体（如 "补时长2"、"虚拟台 1"）也算 all-only。
    - 维表 site_table_area_name 为空（NULL）但有 table_id 的脏数据，归入 all-only INFO，
      因为这通常是 dim_table SCD2 缺区域名而非真正映射缺口；金额仍合入 all 不丢失。
      真正的「未知非空区域名」（如新店自定义命名未在 AREA_LABEL_MAP 中）才进 WARNING。
    """
    if area_name is None:
        # 无 table_id：本来就没台桌，正常 all-only
        # 有 table_id：维表区域名缺失，作为 dim_table 数据问题，仍归 all-only 但保留可观测性（INFO 行会带 'table_id=… None' 标签）
        return True
    if not isinstance(area_name, str):
        return False
    name = area_name.strip()
    if not name:
        return True
    if name in _ALL_ONLY_AREA_NAMES:
        return True
    # 形如 "补时长2"、"补时长 3"、"虚拟台4" 等编号变体
    for prefix in _ALL_ONLY_AREA_NAMES:
        if name.startswith(prefix):
            tail = name[len(prefix):].strip()
            if not tail or tail.isdigit():
                return True
    return False


def _format_unmatched_area_label(area_name: Any, table_id: Any) -> str:
    """格式化未匹配区域日志标签，区分无台桌订单和维表缺口。"""
    if area_name is None and not table_id:
        return "无台桌"
    if area_name is None:
        return f"table_id={table_id}: None"
    return str(area_name)


__all__ = ["FinanceAreaDailyTask", "transform_area_daily"]
