# -*- coding: utf-8 -*-
# AI_CHANGELOG [2025-07-16] task 2.3: 从 FinanceDailyTask 提取共享提取方法，创建财务任务基类
"""
财务任务共享基类

提供财务类 DWS 任务的公共数据提取方法，避免 SQL 重复。
子类：FinanceDailyTask、FinanceRechargeTask、FinanceIncomeStructureTask、FinanceDiscountDetailTask

共享方法：
    - _extract_settlement_summary: 结账单日汇总
    - _extract_recharge_summary: 充值日汇总
    - _extract_groupbuy_summary: 团购核销日汇总
    - _extract_platform_summary: 平台回款/服务费汇总
    - _extract_big_customer_discounts: 大客户优惠明细
    - _extract_gift_card_consume_summary: 赠送卡消费汇总
"""

from __future__ import annotations

import calendar
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
from .dws_helpers import parse_id_list


class FinanceBaseTask(BaseDwsTask):
    """财务任务共享基类，提供公共数据提取方法。"""

    # ------------------------------------------------------------------
    # 结算汇总
    # ------------------------------------------------------------------
    def _extract_settlement_summary(
        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("pay_time", cutoff)
        sql = f"""
            SELECT 
                {biz_expr} AS stat_date,
                COUNT(*) AS order_count,
                COUNT(CASE WHEN member_id != 0 AND member_id IS NOT NULL THEN 1 END) AS member_order_count,
                COUNT(CASE WHEN member_id = 0 OR member_id IS NULL THEN 1 END) AS guest_order_count,
                -- 发生额（正价）
                SUM(table_charge_money) AS table_fee_amount,
                SUM(goods_money) AS goods_amount,
                SUM(assistant_pd_money) AS assistant_pd_amount,
                SUM(assistant_cx_money) AS assistant_cx_amount,
                SUM(table_charge_money + goods_money + assistant_pd_money + assistant_cx_money) AS gross_amount,
                -- 支付
                SUM(pay_amount) AS cash_pay_amount,
                SUM(recharge_card_amount) AS card_pay_amount,
                SUM(balance_amount) AS balance_pay_amount,
                -- 优惠
                SUM(coupon_amount) AS coupon_amount,
                SUM(adjust_amount) AS adjust_amount,
                SUM(member_discount_amount) AS member_discount_amount,
                SUM(rounding_amount) AS rounding_amount,
                SUM(pl_coupon_sale_amount) AS pl_coupon_sale_amount,
                -- CHANGE 2026-03-07 | consume_money → items_sum 口径校准
                -- consume_money 存在三种历史口径混合，DWS 层统一使用 items_sum
                SUM(table_charge_money + goods_money + assistant_pd_money
                    + assistant_cx_money + electricity_money) AS items_sum
            FROM dwd.dwd_settlement_head
            WHERE site_id = %s
              AND {biz_expr} >= %s
              AND {biz_expr} <= %s
              -- CHANGE 2026-03-07 | 排除退货(6)/退款(7)，仅保留台桌结账(1)+商城订单(3)
              AND settle_type IN (1, 3)
            GROUP BY {biz_expr}
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []

    # ------------------------------------------------------------------
    # 充值汇总
    # ------------------------------------------------------------------
    def _extract_recharge_summary(
        self,
        site_id: int,
        start_date: date,
        end_date: date,
    ) -> List[Dict[str, Any]]:
        """充值日汇总（充值订单按日聚合）"""
        # CHANGE 2026-02-21 | BUG 8: dwd_recharge_order 无 pay_money/gift_money，实际字段为 pay_amount/point_amount
        cutoff = self.config.get("app.business_day_start_hour", 8)
        biz_expr = biz_date_sql_expr("pay_time", cutoff)
        sql = f"""
            SELECT 
                {biz_expr} AS stat_date,
                COUNT(*) AS recharge_count,
                SUM(pay_amount + point_amount) AS recharge_total,
                SUM(pay_amount) AS recharge_cash,
                SUM(point_amount) AS recharge_gift,
                COUNT(CASE WHEN is_first = 1 THEN 1 END) AS first_recharge_count,
                SUM(CASE WHEN is_first = 1 THEN pay_amount + point_amount ELSE 0 END) AS first_recharge_total,
                SUM(CASE WHEN is_first = 1 THEN pay_amount ELSE 0 END) AS first_recharge_cash,
                SUM(CASE WHEN is_first = 1 THEN point_amount ELSE 0 END) AS first_recharge_gift,
                COUNT(CASE WHEN is_first = 2 THEN 1 END) AS renewal_count,
                SUM(CASE WHEN is_first = 2 THEN pay_amount + point_amount ELSE 0 END) AS renewal_total,
                SUM(CASE WHEN is_first = 2 THEN pay_amount ELSE 0 END) AS renewal_cash,
                SUM(CASE WHEN is_first = 2 THEN point_amount ELSE 0 END) AS renewal_gift,
                COUNT(DISTINCT member_id) AS recharge_member_count
            FROM dwd.dwd_recharge_order
            WHERE site_id = %s
              AND {biz_expr} >= %s
              AND {biz_expr} <= %s
            GROUP BY {biz_expr}
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []

    # ------------------------------------------------------------------
    # 支付方式拆分
    # ------------------------------------------------------------------
    # CHANGE 2026-03-27 | board-finance-integration T1.1 | 新增支付方式拆分
    def _extract_payment_split(
        self,
        site_id: int,
        start_date: date,
        end_date: date,
    ) -> List[Dict[str, Any]]:
        """支付方式拆分日汇总（纸币现金 vs 扫码/离线支付）"""
        cutoff = self.config.get("app.business_day_start_hour", 8)
        biz_expr = biz_date_sql_expr("p.pay_time", cutoff)
        sql = f"""
            SELECT 
                {biz_expr} AS stat_date,
                SUM(CASE WHEN p.payment_method = 2 THEN p.pay_amount ELSE 0 END) AS cash_paper_amount,
                SUM(CASE WHEN p.payment_method = 4 THEN p.pay_amount ELSE 0 END) AS scan_pay_amount
            FROM dwd.dwd_payment p
            WHERE p.site_id = %s
              AND p.pay_status = 2
              AND p.relate_type = 2
              AND {biz_expr} >= %s
              AND {biz_expr} <= %s
            GROUP BY {biz_expr}
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []

    # ------------------------------------------------------------------
    # 团购核销汇总
    # ------------------------------------------------------------------
    def _extract_groupbuy_summary(
        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("sh.pay_time", cutoff)
        sql = f"""
            SELECT 
                {biz_expr} AS stat_date,
                COUNT(CASE WHEN sh.coupon_amount > 0 THEN 1 END) AS groupbuy_count,
                SUM(
                    CASE 
                        WHEN sh.coupon_amount > 0 THEN
                            CASE 
                                WHEN sh.pl_coupon_sale_amount > 0 THEN sh.pl_coupon_sale_amount
                                ELSE COALESCE(gr.ledger_unit_price, 0)
                            END
                        ELSE 0
                    END
                ) AS groupbuy_pay_total
            FROM dwd.dwd_settlement_head sh
            LEFT JOIN dwd.dwd_groupbuy_redemption gr
                ON gr.order_settle_id = sh.order_settle_id
               AND COALESCE(gr.is_delete, 0) = 0
            WHERE sh.site_id = %s
              AND {biz_expr} >= %s
              AND {biz_expr} <= %s
            GROUP BY {biz_expr}
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []

    # ------------------------------------------------------------------
    # 平台回款/服务费汇总
    # ------------------------------------------------------------------
    def _extract_platform_summary(
        self,
        site_id: int,
        start_date: date,
        end_date: date,
    ) -> List[Dict[str, Any]]:
        """平台回款/服务费汇总（来自导入表）"""
        sql = """
            SELECT 
                settlement_date AS stat_date,
                SUM(settlement_amount) AS settlement_amount,
                SUM(commission_amount) AS commission_amount,
                SUM(service_fee) AS service_fee
            FROM dws.dws_platform_settlement
            WHERE site_id = %s
              AND settlement_date >= %s
              AND settlement_date <= %s
            GROUP BY settlement_date
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []

    # ------------------------------------------------------------------
    # 大客户优惠
    # ------------------------------------------------------------------
    def _extract_big_customer_discounts(
        self,
        site_id: int,
        start_date: date,
        end_date: date,
    ) -> List[Dict[str, Any]]:
        """大客户优惠明细（基于手动调整中标记的大客户订单）"""
        member_ids = self._parse_id_list(
            self.config.get("dws.discount.big_customer_member_ids")
        )
        order_ids = self._parse_id_list(
            self.config.get("dws.discount.big_customer_order_ids")
        )
        if not member_ids and not order_ids:
            return []

        cutoff = self.config.get("app.business_day_start_hour", 8)
        biz_expr = biz_date_sql_expr("pay_time", cutoff)
        sql = f"""
            SELECT 
                {biz_expr} AS stat_date,
                order_settle_id,
                member_id,
                adjust_amount
            FROM dwd.dwd_settlement_head
            WHERE site_id = %s
              AND {biz_expr} >= %s
              AND {biz_expr} <= %s
              AND adjust_amount != 0
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        if not rows:
            return []

        result: Dict[date, Dict[str, Any]] = {}
        for row in rows:
            row_dict = dict(row)
            stat_date = row_dict.get("stat_date")
            if not stat_date:
                continue
            order_id = row_dict.get("order_settle_id")
            member_id = row_dict.get("member_id")
            if order_id not in order_ids and member_id not in member_ids:
                continue
            amount = abs(self.safe_decimal(row_dict.get("adjust_amount", 0)))
            entry = result.setdefault(
                stat_date,
                {"stat_date": stat_date, "big_customer_amount": Decimal("0"), "big_customer_count": 0},
            )
            entry["big_customer_amount"] += amount
            entry["big_customer_count"] += 1

        return list(result.values())

    # ------------------------------------------------------------------
    # 赠送卡消费汇总
    # ------------------------------------------------------------------

    # 赠送卡类型 ID 常量，供子类复用
    GIFT_CARD_TYPE_IDS = (
        2791990152417157,  # 台费卡
        2794699703437125,  # 酒水卡
        2793266846533445,  # 活动抵用券
    )

    def _extract_gift_card_consume_summary(
        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_change = biz_date_sql_expr("change_time", cutoff)
        id_list = ", ".join(str(card_id) for card_id in self.GIFT_CARD_TYPE_IDS)
        sql = f"""
            SELECT 
                {biz_expr_change} AS stat_date,
                SUM(ABS(change_amount)) AS gift_card_consume
            FROM dwd.dwd_member_balance_change
            WHERE site_id = %s
              AND {biz_expr_change} >= %s
              AND {biz_expr_change} <= %s
              AND from_type = 1
              AND change_amount < 0
              AND COALESCE(is_delete, 0) = 0
              AND card_type_id IN ({id_list})
            GROUP BY {biz_expr_change}
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))
        return [dict(row) for row in rows] if rows else []

    # ------------------------------------------------------------------
    # 支出汇总（按月分摊到日）
    # ------------------------------------------------------------------
    def _extract_expense_summary(
        self,
        site_id: int,
        start_date: date,
        end_date: date,
    ) -> List[Dict[str, Any]]:
        """支出汇总（来自导入表，按月分摊到日）"""
        if start_date > end_date:
            return []

        start_month = start_date.replace(day=1)
        end_month = end_date.replace(day=1)

        sql = """
            SELECT 
                expense_month,
                SUM(expense_amount) AS expense_amount
            FROM dws.dws_finance_expense_summary
            WHERE site_id = %s
              AND expense_month >= %s
              AND expense_month <= %s
            GROUP BY expense_month
        """
        rows = self.db.query(sql, (site_id, start_month, end_month))
        if not rows:
            return []

        daily_totals: Dict[date, Decimal] = {}
        for row in rows:
            row_dict = dict(row)
            month_date = row_dict.get("expense_month")
            if not month_date:
                continue
            amount = self.safe_decimal(row_dict.get("expense_amount", 0))
            days_in_month = calendar.monthrange(month_date.year, month_date.month)[1]
            daily_amount = amount / Decimal(str(days_in_month)) if days_in_month > 0 else Decimal("0")

            for day in range(1, days_in_month + 1):
                stat_date = date(month_date.year, month_date.month, day)
                if stat_date < start_date or stat_date > end_date:
                    continue
                daily_totals[stat_date] = daily_totals.get(stat_date, Decimal("0")) + daily_amount

        return [
            {"stat_date": sd, "expense_amount": amt}
            for sd, amt in sorted(daily_totals.items())
        ]

    # ------------------------------------------------------------------
    # 辅助
    # ------------------------------------------------------------------
    def _parse_id_list(self, value: Any) -> set:
        """解析逗号分隔的 ID 列表（委托 dws_helpers）"""
        return parse_id_list(value)


__all__ = ["FinanceBaseTask"]
