# -*- coding: utf-8 -*-
"""
充值统计任务

功能说明：
    以"日期"为粒度，统计充值数据
    
数据来源：
    - dwd_recharge_order: 充值订单
    - dim_member_card_account: 会员卡账户（余额快照）

目标表：
    dws.dws_finance_recharge_summary

更新策略：
    - 更新频率：每日更新
    - 幂等方式：delete-before-insert（按日期）
    
业务规则：
    - 首充/续充：通过 is_first 字段区分
    - 现金/赠送：通过 pay_amount/point_amount 区分
    - 卡余额：区分储值卡和赠送卡

作者：ETL团队
创建日期：2026-02-01
"""

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 TaskContext
from .finance_base_task import FinanceBaseTask


# CHANGE 2025-07-16 | task 2.3: 继承 FinanceBaseTask，复用 _extract_recharge_summary
class FinanceRechargeTask(FinanceBaseTask):
    """
    充值统计任务
    """

    # CHANGE 2025-07-15 | task 1.3: 声明日期列，供基类默认 load() 使用
    DATE_COL = "stat_date"
    
    def get_task_code(self) -> str:
        return "DWS_FINANCE_RECHARGE"
    
    def get_target_table(self) -> str:
        return "dws_finance_recharge_summary"
    
    def get_primary_keys(self) -> List[str]:
        return ["site_id", "stat_date"]
    
    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
        
        recharge_summary = self._extract_recharge_summary(site_id, start_date, end_date)
        card_balances = self._extract_card_balances(site_id, end_date)
        # CHANGE 2026-07-18 | task 4.1: 调用赠送卡新增充值拆分方法
        gift_recharge_breakdown = self._extract_gift_recharge_breakdown(site_id, start_date, end_date)
        
        return {
            'recharge_summary': recharge_summary,
            'card_balances': card_balances,
            'gift_recharge_breakdown': gift_recharge_breakdown,
            'start_date': start_date,
            'end_date': end_date,
            'site_id': site_id
        }
    
    def transform(self, extracted: Dict[str, Any], context: TaskContext) -> List[Dict[str, Any]]:
        recharge_summary = extracted['recharge_summary']
        card_balances = extracted['card_balances']
        # CHANGE 2026-07-18 | task 4.2: 提取赠送卡新增充值拆分数据
        gift_recharge_breakdown = extracted.get('gift_recharge_breakdown', {})
        site_id = extracted['site_id']
        
        results = []
        for recharge in recharge_summary:
            stat_date = recharge.get('stat_date')
            
            # 仅有当前快照时，统一写入（避免窗口内其他日期为0）
            balance = card_balances
            
            record = {
                'site_id': site_id,
                'tenant_id': self.config.get("app.tenant_id", site_id),
                'stat_date': stat_date,
                'recharge_count': self.safe_int(recharge.get('recharge_count', 0)),
                'recharge_total': self.safe_decimal(recharge.get('recharge_total', 0)),
                'recharge_cash': self.safe_decimal(recharge.get('recharge_cash', 0)),
                'recharge_gift': self.safe_decimal(recharge.get('recharge_gift', 0)),
                'first_recharge_count': self.safe_int(recharge.get('first_recharge_count', 0)),
                'first_recharge_cash': self.safe_decimal(recharge.get('first_recharge_cash', 0)),
                'first_recharge_gift': self.safe_decimal(recharge.get('first_recharge_gift', 0)),
                'first_recharge_total': self.safe_decimal(recharge.get('first_recharge_total', 0)),
                'renewal_count': self.safe_int(recharge.get('renewal_count', 0)),
                'renewal_cash': self.safe_decimal(recharge.get('renewal_cash', 0)),
                'renewal_gift': self.safe_decimal(recharge.get('renewal_gift', 0)),
                'renewal_total': self.safe_decimal(recharge.get('renewal_total', 0)),
                'recharge_member_count': self.safe_int(recharge.get('recharge_member_count', 0)),
                'new_member_count': self.safe_int(recharge.get('new_member_count', 0)),
                'total_card_balance': self.safe_decimal(balance.get('total_balance', 0)),
                'cash_card_balance': self.safe_decimal(balance.get('cash_balance', 0)),
                'gift_card_balance': self.safe_decimal(balance.get('gift_balance', 0)),
                # CHANGE 2026-07-18 | task 4.2: 赠送卡细分余额（来自 card_balances）
                'gift_liquor_balance': self.safe_decimal(balance.get('gift_liquor_balance', 0)),
                'gift_table_fee_balance': self.safe_decimal(balance.get('gift_table_fee_balance', 0)),
                'gift_voucher_balance': self.safe_decimal(balance.get('gift_voucher_balance', 0)),
                # CHANGE 2026-07-18 | task 4.2: 赠送卡细分新增充值（来自 gift_recharge_breakdown）
                'gift_liquor_recharge': self.safe_decimal(gift_recharge_breakdown.get('gift_liquor_recharge', 0)),
                'gift_table_fee_recharge': self.safe_decimal(gift_recharge_breakdown.get('gift_table_fee_recharge', 0)),
                'gift_voucher_recharge': self.safe_decimal(gift_recharge_breakdown.get('gift_voucher_recharge', 0)),
            }
            results.append(record)
        
        return results
    
    # load() 已移除——使用 BaseDwsTask 默认实现（DATE_COL="stat_date"）

    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 ELSE 0 END) AS first_recharge_cash,
                SUM(CASE WHEN is_first = 1 THEN point_amount ELSE 0 END) AS first_recharge_gift,
                SUM(CASE WHEN is_first = 1 THEN pay_amount + point_amount ELSE 0 END) AS first_recharge_total,
                COUNT(CASE WHEN is_first = 2 THEN 1 END) AS renewal_count,
                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,
                SUM(CASE WHEN is_first = 2 THEN pay_amount + point_amount ELSE 0 END) AS renewal_total,
                COUNT(DISTINCT member_id) AS recharge_member_count,
                COUNT(DISTINCT CASE WHEN is_first = 1 THEN member_id END) AS new_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-07-17 | task 2.1: card_type_id → 细分余额字段名映射
    GIFT_TYPE_FIELD_MAP = {
        2794699703437125: 'gift_liquor_balance',     # 酒水卡
        2791990152417157: 'gift_table_fee_balance',   # 台费卡
        2793266846533445: 'gift_voucher_balance',     # 抵用券
    }

    # CHANGE 2026-07-18 | task 3.1: card_type_id → 细分充值字段名映射
    GIFT_RECHARGE_FIELD_MAP = {
        2794699703437125: 'gift_liquor_recharge',     # 酒水卡
        2791990152417157: 'gift_table_fee_recharge',   # 台费卡
        2793266846533445: 'gift_voucher_recharge',     # 抵用券
    }

    # CHANGE 2026-07-18 | task 3.1: 按 card_type_id 拆分赠送卡新增充值
    def _extract_gift_recharge_breakdown(self, site_id: int, start_date: date, end_date: date) -> Dict[str, Decimal]:
        """按卡类型拆分赠送卡新增充值（JOIN 充值订单与会员卡账户维度表）"""
        cutoff = self.config.get("app.business_day_start_hour", 8)
        biz_expr = biz_date_sql_expr("ro.pay_time", cutoff)
        sql = f"""
            SELECT dca.card_type_id, SUM(ro.point_amount) AS gift_recharge
            FROM dwd.dwd_recharge_order ro
            JOIN dwd.dim_member_card_account dca
                ON ro.tenant_member_card_id = dca.tenant_member_id
            WHERE ro.site_id = %s
              AND {biz_expr} >= %s AND {biz_expr} <= %s
              AND dca.card_type_id IN (2794699703437125, 2791990152417157, 2793266846533445)
              AND dca.scd2_is_current = 1
              AND COALESCE(dca.is_delete, 0) = 0
            GROUP BY dca.card_type_id
        """
        rows = self.db.query(sql, (site_id, start_date, end_date))

        result: Dict[str, Decimal] = {
            field: Decimal('0') for field in self.GIFT_RECHARGE_FIELD_MAP.values()
        }
        for row in (rows or []):
            field_name = self.GIFT_RECHARGE_FIELD_MAP.get(row['card_type_id'])
            if field_name:
                result[field_name] = self.safe_decimal(row['gift_recharge'])
        return result

    def _extract_card_balances(self, site_id: int, stat_date: date) -> Dict[str, Decimal]:
        CASH_CARD_TYPE_ID = 2793249295533893
        GIFT_CARD_TYPE_IDS = [2791990152417157, 2793266846533445, 2794699703437125]
        
        # CHANGE 2026-04-08 | 修复卡余额快照失真：改为门店全量会员
        # 原逻辑通过 dwd_recharge_order 反查，遗漏"有卡无充值"会员（赠卡、活动等）
        # 余额代表门店预收负债，应覆盖所有持卡会员
        sql = """
            SELECT dca.card_type_id, SUM(dca.balance) AS total_balance
            FROM dwd.dim_member_card_account dca
            JOIN dwd.dim_member m
                ON dca.tenant_member_id = m.member_id
                AND m.scd2_is_current = 1
            WHERE m.register_site_id = %s
              AND dca.scd2_is_current = 1
              AND COALESCE(dca.is_delete, 0) = 0
            GROUP BY dca.card_type_id
        """
        rows = self.db.query(sql, (site_id,))
        
        cash_balance = Decimal('0')
        gift_balance = Decimal('0')
        # CHANGE 2026-07-17 | task 2.1: 按 card_type_id 拆分赠送卡余额
        gift_breakdown: Dict[str, Decimal] = {
            field: Decimal('0') for field in self.GIFT_TYPE_FIELD_MAP.values()
        }
        
        for row in (rows or []):
            card_type_id = row['card_type_id']
            balance = self.safe_decimal(row['total_balance'])
            if card_type_id == CASH_CARD_TYPE_ID:
                cash_balance += balance
            elif card_type_id in GIFT_CARD_TYPE_IDS:
                gift_balance += balance
                # 写入细分字段（未知 card_type_id 不会命中 GIFT_TYPE_FIELD_MAP）
                field_name = self.GIFT_TYPE_FIELD_MAP.get(card_type_id)
                if field_name:
                    gift_breakdown[field_name] += balance
        
        return {
            'cash_balance': cash_balance,
            'gift_balance': gift_balance,
            'total_balance': cash_balance + gift_balance,
            **gift_breakdown,
        }


__all__ = ['FinanceRechargeTask']
