# -*- coding: utf-8 -*-
"""
库存日度汇总任务

功能说明：
    以"门店+日期+商品"为粒度，汇总每日库存数据

数据来源：
    - dwd.dwd_goods_stock_summary：库存汇总明细（按 fetched_at 日期聚合）

目标表：
    dws.dws_goods_stock_daily_summary

更新策略：
    - 更新频率：每日更新
    - 幂等方式：upsert（ON CONFLICT DO UPDATE）

业务规则：
    - 按 fetched_at 的日期部分分组，同一天同一商品可能有多条 DWD 记录
    - 数值指标取 SUM 聚合（入库/出库/销售等为累计量）
    - current_stock 取当日最后一条记录的值（期末快照）
    - range_start_stock 取当日第一条记录的值（期初快照）
    - range_end_stock 取当日最后一条记录的值（期末快照）
"""

from __future__ import annotations

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

from neozqyy_shared.datetime_utils import biz_date_sql_expr

from .base_dws_task import BaseDwsTask, TaskContext


class GoodsStockDailyTask(BaseDwsTask):
    """
    库存日度汇总任务

    从 dwd.dwd_goods_stock_summary 提取数据，按日粒度汇总后
    upsert 写入 dws.dws_goods_stock_daily_summary。
    """

    DATE_COL = "stat_date"

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

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

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

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

    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 ~ %s",
            self.get_task_code(), site_id, start_date, end_date,
        )

        cutoff = self.config.get("app.business_day_start_hour", 8)
        biz_expr = biz_date_sql_expr("fetched_at", cutoff)
        sql = f"""
            SELECT
                site_goods_id,
                goods_name,
                goods_unit,
                goods_category_id,
                goods_category_second_id,
                category_name,
                range_start_stock,
                range_end_stock,
                range_in,
                range_out,
                range_sale,
                range_sale_money,
                range_inventory,
                current_stock,
                site_id,
                tenant_id,
                fetched_at,
                {biz_expr} AS biz_date
            FROM dwd.dwd_goods_stock_summary
            WHERE site_id = %s
              AND {biz_expr} >= %s
              AND {biz_expr} <= %s
            ORDER BY fetched_at
        """
        rows = self.query_dwd(sql, (site_id, start_date, end_date))

        self.logger.info(
            "%s: 提取到 %d 条 DWD 记录", self.get_task_code(), len(rows),
        )

        return {
            "rows": rows,
            "start_date": start_date,
            "end_date": end_date,
            "site_id": site_id,
        }

    # ======================================================================
    # Transform
    # ======================================================================

    def transform(
        self, extracted: Dict[str, Any], context: TaskContext
    ) -> List[Dict[str, Any]]:
        """按日粒度汇总：同一天同一商品聚合为一条记录"""
        rows = extracted.get("rows", [])
        site_id = extracted["site_id"]

        if not rows:
            self.logger.info("%s: 无数据需要汇总", self.get_task_code())
            return []

        # 按 (stat_date, site_goods_id) 分组聚合
        # key: (date, site_goods_id) -> 聚合数据
        agg: Dict[tuple, Dict[str, Any]] = {}

        for row in rows:
            fetched_at = row.get("fetched_at")
            if fetched_at is None:
                continue
            # 使用 SQL 层计算的营业日归属日期
            stat_date = row.get("biz_date")
            if stat_date is None:
                stat_date = (
                    fetched_at.date()
                    if hasattr(fetched_at, "date")
                    else fetched_at
                )
            site_goods_id = row.get("site_goods_id")
            if site_goods_id is None:
                continue

            key = (stat_date, site_goods_id)

            if key not in agg:
                # 首条记录：初始化，期初取第一条
                agg[key] = {
                    "site_id": site_id,
                    "tenant_id": row.get("tenant_id"),
                    "stat_date": stat_date,
                    "site_goods_id": site_goods_id,
                    "goods_name": row.get("goods_name"),
                    "goods_unit": row.get("goods_unit"),
                    "goods_category_id": row.get("goods_category_id"),
                    "goods_category_second_id": row.get("goods_category_second_id"),
                    "category_name": row.get("category_name"),
                    "range_start_stock": self.safe_decimal(row.get("range_start_stock")),
                    "range_end_stock": self.safe_decimal(row.get("range_end_stock")),
                    "range_in": self.safe_decimal(row.get("range_in")),
                    "range_out": self.safe_decimal(row.get("range_out")),
                    "range_sale": self.safe_decimal(row.get("range_sale")),
                    "range_sale_money": self.safe_decimal(row.get("range_sale_money")),
                    "range_inventory": self.safe_decimal(row.get("range_inventory")),
                    "current_stock": self.safe_decimal(row.get("current_stock")),
                    "stat_period": "daily",
                }
            else:
                # 后续记录：累加数值指标，更新期末快照
                rec = agg[key]
                rec["range_in"] += self.safe_decimal(row.get("range_in"))
                rec["range_out"] += self.safe_decimal(row.get("range_out"))
                rec["range_sale"] += self.safe_decimal(row.get("range_sale"))
                rec["range_sale_money"] += self.safe_decimal(row.get("range_sale_money"))
                rec["range_inventory"] += self.safe_decimal(row.get("range_inventory"))
                # 期末/当前库存取最后一条（rows 已按 fetched_at 排序）
                rec["range_end_stock"] = self.safe_decimal(row.get("range_end_stock"))
                rec["current_stock"] = self.safe_decimal(row.get("current_stock"))

        result = list(agg.values())
        self.logger.info(
            "%s: 汇总完成，生成 %d 条日度记录",
            self.get_task_code(), len(result),
        )
        return result

    # ======================================================================
    # Load
    # ======================================================================

    def load(
        self, transformed: List[Dict[str, Any]], context: TaskContext
    ) -> Dict[str, Any]:
        """upsert 写入 DWS 目标表"""
        if not transformed:
            return {
                "counts": {
                    "fetched": 0,
                    "inserted": 0,
                    "updated": 0,
                    "skipped": 0,
                    "errors": 0,
                }
            }

        columns = [
            "site_id", "tenant_id", "stat_date", "site_goods_id",
            "goods_name", "goods_unit", "goods_category_id",
            "goods_category_second_id", "category_name",
            "range_start_stock", "range_end_stock",
            "range_in", "range_out", "range_sale",
            "range_sale_money", "range_inventory", "current_stock",
            "stat_period",
        ]

        inserted, updated = self.upsert(transformed, columns=columns)

        self.logger.info(
            "%s: 写入完成，inserted=%d",
            self.get_task_code(), inserted,
        )

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


__all__ = ["GoodsStockDailyTask"]
