# -*- coding: utf-8 -*-
"""
财务看板缓存层任务

功能说明：
    基于 dws_finance_area_daily 日粒度数据，为已完成周期（lastMonth/lastWeek/
    lastQuarter/quarter3/half6）× 9 个区域 = 45 组合维护缓存。
    通过数据指纹（MD5）检测源数据变化，仅对指纹不一致的组合重算并 upsert。

数据来源：
    - dws.dws_finance_area_daily：日粒度原子层
    - dws.dws_finance_board_cache：缓存层（读取已有指纹做对比）

目标表：
    dws.dws_finance_board_cache

更新策略：
    - 更新频率：每天一次（营业日切点后）
    - 幂等方式：ON CONFLICT (site_id, time_range, area_code) DO UPDATE
    - 当期周期（month/week/quarter）不写入缓存

业务规则：
    - 指纹 = MD5(sorted [(stat_date, gross_amount, discount_total), ...])
    - 指纹一致 → 跳过；指纹不一致 → 从日粒度表 SUM 重算后 upsert
    - overview 8 项：occurrence/discount/discount_rate/confirmed_revenue/
      cash_in/cash_out/cash_balance/balance_rate

Requirements: 5.1, 5.2, 5.3, 5.4, 5.5, 5.6, 5.7
"""

from __future__ import annotations

import hashlib
import json
import logging
from datetime import date, timedelta
from decimal import Decimal
from typing import Any, Dict, List, Optional, Tuple

from neozqyy_shared.area_mapping import ALL_AREA_CODES

from .base_dws_task import BaseDwsTask, TaskContext

logger = logging.getLogger(__name__)

# 已完成周期（当期 month/week/quarter 不缓存）
COMPLETED_PERIODS: List[str] = [
    "lastMonth",
    "lastWeek",
    "lastQuarter",
    "quarter3",
    "half6",
]

# 当期周期（不写入缓存）
CURRENT_PERIODS: set[str] = {"month", "week", "quarter"}

_ZERO = Decimal("0")


# ======================================================================
# 纯函数：指纹计算（导出供属性测试直接调用）
# ======================================================================


def compute_fingerprint(rows: list[dict]) -> str:
    """对 (stat_date, gross_amount, discount_total) 排序后计算 MD5 指纹。

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

    Args:
        rows: 日粒度行列表，每行至少包含 stat_date/gross_amount/discount_total

    Returns:
        MD5 十六进制字符串
    """
    sorted_rows = sorted(rows, key=lambda r: str(r["stat_date"]))
    payload = json.dumps(
        [
            (str(r["stat_date"]), str(r["gross_amount"]), str(r["discount_total"]))
            for r in sorted_rows
        ]
    )
    return hashlib.md5(payload.encode()).hexdigest()


def is_current_period(time_range: str) -> bool:
    """判断 time_range 是否为当期周期（不应写入缓存）。

    纯函数，导出供属性测试直接调用。
    """
    return time_range in CURRENT_PERIODS


# ======================================================================
# 日期范围计算（复用 board_service 的逻辑）
# ======================================================================


def _calc_period_date_range(
    time_range: str, ref_date: date | None = None
) -> Tuple[date, date]:
    """计算已完成周期的日期范围。

    与 board_service._calc_date_range 保持一致。
    """
    today = ref_date or date.today()

    if time_range == "lastMonth":
        first_of_this_month = today.replace(day=1)
        last_month_end = first_of_this_month - timedelta(days=1)
        last_month_start = last_month_end.replace(day=1)
        return last_month_start, last_month_end

    if time_range == "lastWeek":
        this_monday = today - timedelta(days=today.weekday())
        last_sunday = this_monday - timedelta(days=1)
        last_monday = last_sunday - timedelta(days=6)
        return last_monday, last_sunday

    if time_range == "lastQuarter":
        q_start_month = (today.month - 1) // 3 * 3 + 1
        this_q_start = date(today.year, q_start_month, 1)
        prev_q_end = this_q_start - timedelta(days=1)
        prev_q_start_month = (prev_q_end.month - 1) // 3 * 3 + 1
        prev_q_start = date(prev_q_end.year, prev_q_start_month, 1)
        return prev_q_start, prev_q_end

    if time_range == "quarter3":
        first_of_this_month = today.replace(day=1)
        end = first_of_this_month - timedelta(days=1)
        start = _month_offset(first_of_this_month, -3)
        return start, end

    if time_range == "half6":
        first_of_this_month = today.replace(day=1)
        end = first_of_this_month - timedelta(days=1)
        start = _month_offset(first_of_this_month, -6)
        return start, end

    raise ValueError(f"不支持的已完成周期: {time_range}")


def _month_offset(base_date: date, months: int) -> date:
    """按月偏移日期（保持日不越界）。"""
    import calendar

    total_months = base_date.year * 12 + (base_date.month - 1) + months
    year = total_months // 12
    month = total_months % 12 + 1
    last_day = calendar.monthrange(year, month)[1]
    day = min(base_date.day, last_day)
    return date(year, month, day)


# ======================================================================
# ETL 任务类
# ======================================================================


class FinanceBoardCacheTask(BaseDwsTask):
    """
    财务看板缓存层任务

    遍历 5 个已完成周期 × 9 个区域 = 45 组合，
    通过数据指纹检测变化，仅对需要重算的组合 upsert 缓存。
    """

    DATE_COL = None  # 缓存表无 stat_date 列

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

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

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

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

    def extract(self, context: TaskContext) -> Dict[str, Any]:
        """遍历 45 组合，从日粒度表读取源数据行 + 从缓存表读取已有指纹。"""
        site_id = context.store_id

        self.logger.info(
            "%s: 开始提取，站点 %s",
            self.get_task_code(),
            site_id,
        )

        combinations: List[Dict[str, Any]] = []

        for time_range in COMPLETED_PERIODS:
            start_date, end_date = _calc_period_date_range(time_range)

            for area_code in ALL_AREA_CODES:
                # 从日粒度表读取源数据行
                daily_rows = self._read_daily_rows(
                    site_id, start_date, end_date, area_code
                )

                # 从缓存表读取已有指纹
                existing_fp = self._read_existing_fingerprint(
                    site_id, time_range, area_code
                )

                combinations.append(
                    {
                        "time_range": time_range,
                        "area_code": area_code,
                        "start_date": start_date,
                        "end_date": end_date,
                        "daily_rows": daily_rows,
                        "existing_fingerprint": existing_fp,
                    }
                )

        return {
            "combinations": combinations,
            "site_id": site_id,
        }

    def _read_daily_rows(
        self,
        site_id: int,
        start_date: date,
        end_date: date,
        area_code: str,
    ) -> List[Dict[str, Any]]:
        """从 dws_finance_area_daily 读取日粒度行。"""
        sql = """
            SELECT stat_date, gross_amount, discount_total,
                   confirmed_income,
                   cash_inflow_total, cash_outflow_total, cash_balance_change
            FROM dws.dws_finance_area_daily
            WHERE site_id = %s
              AND stat_date >= %s
              AND stat_date <= %s
              AND area_code = %s
        """
        rows = self.db.query(sql, (site_id, start_date, end_date, area_code))
        return [dict(row) for row in rows] if rows else []

    def _read_existing_fingerprint(
        self,
        site_id: int,
        time_range: str,
        area_code: str,
    ) -> Optional[str]:
        """从缓存表读取已有指纹。"""
        sql = """
            SELECT data_fingerprint
            FROM dws.dws_finance_board_cache
            WHERE site_id = %s
              AND time_range = %s
              AND area_code = %s
        """
        rows = self.db.query(sql, (site_id, time_range, area_code))
        if rows:
            return dict(rows[0]).get("data_fingerprint")
        return None

    # ======================================================================
    # Transform（纯函数逻辑，标记需重算的组合）
    # ======================================================================

    def transform(
        self, extracted: Dict[str, Any], context: TaskContext
    ) -> List[Dict[str, Any]]:
        """计算指纹，与已有指纹对比，标记需重算的组合。"""
        site_id = extracted["site_id"]
        combinations = extracted["combinations"]

        return transform_cache_combinations(
            combinations=combinations,
            site_id=site_id,
            logger=self.logger,
        )

    # ======================================================================
    # Load（upsert 需重算的组合）
    # ======================================================================

    def load(self, transformed: List[Dict[str, Any]], context: TaskContext) -> dict:
        """对需重算的组合 upsert 到缓存表。"""
        if not transformed:
            return {
                "counts": {
                    "fetched": 0,
                    "inserted": 0,
                    "updated": 0,
                    "skipped": 0,
                    "errors": 0,
                }
            }

        site_id = context.store_id
        upserted = 0
        skipped = 0

        for combo in transformed:
            if not combo.get("needs_recompute"):
                skipped += 1
                continue

            # 从日粒度表 SUM 计算 overview 指标
            overview = self._sum_overview(
                site_id,
                combo["start_date"],
                combo["end_date"],
                combo["area_code"],
            )

            self._upsert_cache(
                site_id=site_id,
                time_range=combo["time_range"],
                area_code=combo["area_code"],
                start_date=combo["start_date"],
                end_date=combo["end_date"],
                overview=overview,
                fingerprint=combo["new_fingerprint"],
            )
            upserted += 1

        self.logger.info(
            "%s: upsert %d 组合，跳过 %d 组合",
            self.get_task_code(),
            upserted,
            skipped,
        )

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

    def _sum_overview(
        self,
        site_id: int,
        start_date: date,
        end_date: date,
        area_code: str,
    ) -> Dict[str, Any]:
        """从日粒度表 SUM 计算 overview 8 项指标。"""
        sql = """
            SELECT
                COALESCE(SUM(gross_amount), 0)          AS occurrence,
                COALESCE(SUM(discount_total), 0)        AS discount,
                COALESCE(SUM(confirmed_income), 0)      AS confirmed_revenue,
                COALESCE(SUM(cash_inflow_total), 0)     AS cash_in,
                COALESCE(SUM(cash_outflow_total), 0)    AS cash_out,
                COALESCE(SUM(cash_balance_change), 0)   AS cash_balance
            FROM dws.dws_finance_area_daily
            WHERE site_id = %s
              AND stat_date >= %s
              AND stat_date <= %s
              AND area_code = %s
        """
        rows = self.db.query(sql, (site_id, start_date, end_date, area_code))
        if rows:
            row = dict(rows[0])
            occurrence = Decimal(str(row.get("occurrence", 0)))
            discount = Decimal(str(row.get("discount", 0)))
            confirmed_revenue = Decimal(str(row.get("confirmed_revenue", 0)))
            cash_in = Decimal(str(row.get("cash_in", 0)))
            cash_out = Decimal(str(row.get("cash_out", 0)))
            cash_balance = Decimal(str(row.get("cash_balance", 0)))

            # discount_rate = discount / occurrence（避免除零）
            discount_rate = (
                (discount / occurrence) if occurrence != 0 else _ZERO
            )
            # balance_rate = cash_balance / cash_in（避免除零）
            balance_rate = (
                (cash_balance / cash_in) if cash_in != 0 else _ZERO
            )

            return {
                "occurrence": occurrence,
                "discount": discount,
                "discount_rate": discount_rate,
                "confirmed_revenue": confirmed_revenue,
                "cash_in": cash_in,
                "cash_out": cash_out,
                "cash_balance": cash_balance,
                "balance_rate": balance_rate,
            }

        return {k: _ZERO for k in [
            "occurrence", "discount", "discount_rate", "confirmed_revenue",
            "cash_in", "cash_out", "cash_balance", "balance_rate",
        ]}

    def _upsert_cache(
        self,
        site_id: int,
        time_range: str,
        area_code: str,
        start_date: date,
        end_date: date,
        overview: Dict[str, Any],
        fingerprint: str,
    ) -> None:
        """ON CONFLICT DO UPDATE 写入缓存表。"""
        sql = """
            INSERT INTO dws.dws_finance_board_cache (
                site_id, time_range, area_code,
                start_date, end_date,
                occurrence, discount, discount_rate, confirmed_revenue,
                cash_in, cash_out, cash_balance, balance_rate,
                data_fingerprint, computed_at, updated_at
            ) VALUES (
                %s, %s, %s,
                %s, %s,
                %s, %s, %s, %s,
                %s, %s, %s, %s,
                %s, NOW(), NOW()
            )
            ON CONFLICT (site_id, time_range, area_code) DO UPDATE SET
                start_date = EXCLUDED.start_date,
                end_date = EXCLUDED.end_date,
                occurrence = EXCLUDED.occurrence,
                discount = EXCLUDED.discount,
                discount_rate = EXCLUDED.discount_rate,
                confirmed_revenue = EXCLUDED.confirmed_revenue,
                cash_in = EXCLUDED.cash_in,
                cash_out = EXCLUDED.cash_out,
                cash_balance = EXCLUDED.cash_balance,
                balance_rate = EXCLUDED.balance_rate,
                data_fingerprint = EXCLUDED.data_fingerprint,
                computed_at = NOW(),
                updated_at = NOW()
        """
        with self.db.conn.cursor() as cur:
            cur.execute(
                sql,
                (
                    site_id,
                    time_range,
                    area_code,
                    start_date,
                    end_date,
                    overview["occurrence"],
                    overview["discount"],
                    overview["discount_rate"],
                    overview["confirmed_revenue"],
                    overview["cash_in"],
                    overview["cash_out"],
                    overview["cash_balance"],
                    overview["balance_rate"],
                    fingerprint,
                ),
            )


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


def transform_cache_combinations(
    combinations: List[Dict[str, Any]],
    site_id: int,
    logger: Any = None,
) -> List[Dict[str, Any]]:
    """对每个组合计算指纹并与已有指纹对比，标记需重算的组合。

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

    Args:
        combinations: extract 输出的组合列表
        site_id: 门店 ID
        logger: 日志记录器（可选）

    Returns:
        带 needs_recompute/new_fingerprint 标记的组合列表
    """
    if logger is None:
        logger = logging.getLogger(__name__)

    results: List[Dict[str, Any]] = []

    for combo in combinations:
        daily_rows = combo.get("daily_rows", [])
        existing_fp = combo.get("existing_fingerprint")

        new_fp = compute_fingerprint(daily_rows)

        needs_recompute = new_fp != existing_fp

        if needs_recompute:
            logger.info(
                "DWS_FINANCE_BOARD_CACHE: %s/%s 指纹变化，需重算",
                combo["time_range"],
                combo["area_code"],
            )

        results.append(
            {
                **combo,
                "site_id": site_id,
                "new_fingerprint": new_fp,
                "needs_recompute": needs_recompute,
            }
        )

    return results


__all__ = [
    "FinanceBoardCacheTask",
    "compute_fingerprint",
    "is_current_period",
    "transform_cache_combinations",
    "COMPLETED_PERIODS",
    "CURRENT_PERIODS",
]
