# -*- coding: utf-8 -*-
"""
助教服务客户统计任务

功能说明：
    以"助教+客户"为粒度，统计服务关系和滚动窗口指标
    
数据来源：
    - dwd_assistant_service_log: 助教服务流水
    - dim_member: 会员维度

目标表：
    dws.dws_assistant_customer_stats

更新策略：
    - 更新频率：每日更新
    - 幂等方式：delete-before-insert（按统计日期）
    
业务规则：
    - 散客处理：member_id=0 不进入此表统计
    - 滚动窗口：7/10/15/30/60/90天
    - 活跃度：近7天/30天是否有服务

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

from __future__ import annotations

from datetime import date, datetime, timedelta
from decimal import Decimal
from typing import Any, Dict, List, Optional, Set, Tuple

from .base_dws_task import BaseDwsTask, TaskContext
from .dws_helpers import mask_mobile, calc_days_since

from neozqyy_shared.datetime_utils import biz_date_sql_expr


class AssistantCustomerTask(BaseDwsTask):
    """
    助教服务客户统计任务
    
    统计每个助教与每个客户的服务关系：
    - 首次/最近服务日期
    - 累计服务统计
    - 滚动窗口统计（7/10/15/30/60/90天）
    - 活跃度指标
    """

    # CHANGE 2025-07-15 | task 1.3: 声明日期列，供基类默认 load() 使用
    DATE_COL = "stat_date"
    
    def get_task_code(self) -> str:
        return "DWS_ASSISTANT_CUSTOMER"
    
    def get_target_table(self) -> str:
        return "dws_assistant_customer_stats"
    
    def get_primary_keys(self) -> List[str]:
        return ["site_id", "assistant_id", "member_id", "stat_date"]
    
    # ==========================================================================
    # ETL主流程
    # ==========================================================================
    
    def extract(self, context: TaskContext) -> Dict[str, Any]:
        """
        提取数据
        """
        stat_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",
            self.get_task_code(), stat_date
        )
        
        # 计算最大回溯日期（90天窗口）
        lookback_start = stat_date - timedelta(days=90)
        
        # 1. 获取助教-客户服务记录（包含历史全量用于累计统计）
        service_pairs = self._extract_service_pairs(site_id, stat_date)
        
        # 2. 获取会员信息
        member_info = self._extract_member_info(site_id)
        
        # 3. 获取助教信息
        assistant_info = self._extract_assistant_info(site_id)
        
        return {
            'service_pairs': service_pairs,
            'member_info': member_info,
            'assistant_info': assistant_info,
            'stat_date': stat_date,
            'site_id': site_id
        }
    
    def transform(self, extracted: Dict[str, Any], context: TaskContext) -> List[Dict[str, Any]]:
        """
        转换数据：计算各窗口统计
        """
        service_pairs = extracted['service_pairs']
        member_info = extracted['member_info']
        assistant_info = extracted['assistant_info']
        stat_date = extracted['stat_date']
        site_id = extracted['site_id']
        
        self.logger.info(
            "%s: 转换数据，%d 条服务关系记录",
            self.get_task_code(), len(service_pairs)
        )
        
        # 构建统计记录
        results = []
        
        for pair in service_pairs:
            assistant_id = pair.get('assistant_id')
            member_id = pair.get('member_id')
            
            # 跳过散客
            if self.is_guest(member_id):
                continue
            
            asst_info = assistant_info.get(assistant_id, {})
            memb_info = member_info.get(member_id, {})
            
            # 构建记录
            record = {
                'site_id': site_id,
                'tenant_id': self.config.get("app.tenant_id", site_id),
                'assistant_id': assistant_id,
                'assistant_nickname': asst_info.get('nickname', pair.get('assistant_nickname')),
                'member_id': member_id,
                'member_nickname': memb_info.get('nickname'),
                'member_mobile': self._mask_mobile(memb_info.get('mobile')),
                'stat_date': stat_date,
                # 全量累计统计
                'first_service_date': pair.get('first_service_date'),
                'last_service_date': pair.get('last_service_date'),
                'total_service_count': self.safe_int(pair.get('total_service_count', 0)),
                'total_service_hours': self.safe_decimal(pair.get('total_service_hours', 0)),
                'total_service_amount': self.safe_decimal(pair.get('total_service_amount', 0)),
                # 滚动窗口统计
                'service_count_7d': self.safe_int(pair.get('service_count_7d', 0)),
                'service_count_10d': self.safe_int(pair.get('service_count_10d', 0)),
                'service_count_15d': self.safe_int(pair.get('service_count_15d', 0)),
                'service_count_30d': self.safe_int(pair.get('service_count_30d', 0)),
                'service_count_60d': self.safe_int(pair.get('service_count_60d', 0)),
                'service_count_90d': self.safe_int(pair.get('service_count_90d', 0)),
                'service_hours_7d': self.safe_decimal(pair.get('service_hours_7d', 0)),
                'service_hours_10d': self.safe_decimal(pair.get('service_hours_10d', 0)),
                'service_hours_15d': self.safe_decimal(pair.get('service_hours_15d', 0)),
                'service_hours_30d': self.safe_decimal(pair.get('service_hours_30d', 0)),
                'service_hours_60d': self.safe_decimal(pair.get('service_hours_60d', 0)),
                'service_hours_90d': self.safe_decimal(pair.get('service_hours_90d', 0)),
                'service_amount_7d': self.safe_decimal(pair.get('service_amount_7d', 0)),
                'service_amount_10d': self.safe_decimal(pair.get('service_amount_10d', 0)),
                'service_amount_15d': self.safe_decimal(pair.get('service_amount_15d', 0)),
                'service_amount_30d': self.safe_decimal(pair.get('service_amount_30d', 0)),
                'service_amount_60d': self.safe_decimal(pair.get('service_amount_60d', 0)),
                'service_amount_90d': self.safe_decimal(pair.get('service_amount_90d', 0)),
                # 活跃度指标
                'days_since_last': self._calc_days_since(stat_date, pair.get('last_service_date')),
                'is_active_7d': self.safe_int(pair.get('service_count_7d', 0)) > 0,
                'is_active_30d': self.safe_int(pair.get('service_count_30d', 0)) > 0,
            }
            results.append(record)
        
        return results
    
    # load() 已移除——使用 BaseDwsTask 默认实现（DATE_COL="stat_date"）

    # ==========================================================================
    # 数据提取方法
    # ==========================================================================
    
    def _extract_service_pairs(
        self,
        site_id: int,
        stat_date: date
    ) -> List[Dict[str, Any]]:
        """
        提取助教-客户服务统计（含滚动窗口）
        """
        # CHANGE 2026-03-01 | business-day-cutoff 6.3: DATE(start_use_time) → 营业日归属表达式
        cutoff = self.config.get("app.business_day_start_hour", 8)
        biz_expr = biz_date_sql_expr("start_use_time", cutoff)
        sql = f"""
            WITH service_base AS (
                SELECT 
                    site_assistant_id AS assistant_id,
                    nickname AS assistant_nickname,
                    tenant_member_id AS member_id,
                    {biz_expr} AS service_date,
                    income_seconds,
                    ledger_amount
                FROM dwd.dwd_assistant_service_log
                WHERE site_id = %s
                  AND tenant_member_id IS NOT NULL
                  AND tenant_member_id != 0
                  AND is_delete = 0
            )
            SELECT 
                assistant_id,
                (ARRAY_AGG(assistant_nickname ORDER BY service_date DESC))[1] AS assistant_nickname,
                member_id,
                MIN(service_date) AS first_service_date,
                MAX(service_date) AS last_service_date,
                -- 全量累计
                COUNT(*) AS total_service_count,
                SUM(income_seconds) / 3600.0 AS total_service_hours,
                SUM(ledger_amount) AS total_service_amount,
                -- 7天窗口
                COUNT(CASE WHEN service_date >= %s - INTERVAL '6 days' THEN 1 END) AS service_count_7d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '6 days' THEN income_seconds ELSE 0 END) / 3600.0 AS service_hours_7d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '6 days' THEN ledger_amount ELSE 0 END) AS service_amount_7d,
                -- 10天窗口
                COUNT(CASE WHEN service_date >= %s - INTERVAL '9 days' THEN 1 END) AS service_count_10d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '9 days' THEN income_seconds ELSE 0 END) / 3600.0 AS service_hours_10d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '9 days' THEN ledger_amount ELSE 0 END) AS service_amount_10d,
                -- 15天窗口
                COUNT(CASE WHEN service_date >= %s - INTERVAL '14 days' THEN 1 END) AS service_count_15d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '14 days' THEN income_seconds ELSE 0 END) / 3600.0 AS service_hours_15d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '14 days' THEN ledger_amount ELSE 0 END) AS service_amount_15d,
                -- 30天窗口
                COUNT(CASE WHEN service_date >= %s - INTERVAL '29 days' THEN 1 END) AS service_count_30d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '29 days' THEN income_seconds ELSE 0 END) / 3600.0 AS service_hours_30d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '29 days' THEN ledger_amount ELSE 0 END) AS service_amount_30d,
                -- 60天窗口
                COUNT(CASE WHEN service_date >= %s - INTERVAL '59 days' THEN 1 END) AS service_count_60d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '59 days' THEN income_seconds ELSE 0 END) / 3600.0 AS service_hours_60d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '59 days' THEN ledger_amount ELSE 0 END) AS service_amount_60d,
                -- 90天窗口
                COUNT(CASE WHEN service_date >= %s - INTERVAL '89 days' THEN 1 END) AS service_count_90d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '89 days' THEN income_seconds ELSE 0 END) / 3600.0 AS service_hours_90d,
                SUM(CASE WHEN service_date >= %s - INTERVAL '89 days' THEN ledger_amount ELSE 0 END) AS service_amount_90d
            FROM service_base
            GROUP BY assistant_id, member_id
            HAVING MAX(service_date) >= %s - INTERVAL '90 days'
        """
        # 构建参数（每个窗口需要3个日期参数）
        params = [site_id]
        for _ in range(6):  # 6个窗口，每个3个参数
            params.extend([stat_date, stat_date, stat_date])
        params.append(stat_date)  # HAVING条件
        
        rows = self.db.query(sql, tuple(params))
        return [dict(row) for row in rows] if rows else []
    
    def _extract_member_info(self, site_id: int) -> Dict[int, Dict[str, Any]]:
        """
        提取会员信息
        """
        # CHANGE 2026-02-21 | dim_member 无 site_id 字段，改用 register_site_id；
        # 加 scd2_is_current=1 只取当前有效版本
        # CHANGE 2026-02-22 | 需求 B：通过事实表反查，支持跨店消费会员
        sql = """
            SELECT 
                member_id,
                nickname,
                mobile
            FROM dwd.dim_member
            WHERE member_id IN (
                SELECT DISTINCT tenant_member_id
                FROM dwd.dwd_assistant_service_log
                WHERE site_id = %s
                  AND tenant_member_id IS NOT NULL
                  AND tenant_member_id != 0
            ) AND scd2_is_current = 1
        """
        rows = self.db.query(sql, (site_id,))

        result = {}
        for row in (rows or []):
            row_dict = dict(row)
            result[row_dict['member_id']] = row_dict
        return result

    
    def _extract_assistant_info(self, site_id: int) -> Dict[int, Dict[str, Any]]:
        """
        提取助教信息
        """
        sql = """
            SELECT 
                assistant_id,
                nickname
            FROM dwd.dim_assistant
            WHERE site_id = %s
              AND scd2_is_current = 1
        """
        rows = self.db.query(sql, (site_id,))
        
        result = {}
        for row in (rows or []):
            row_dict = dict(row)
            result[row_dict['assistant_id']] = row_dict
        return result
    
    # ==========================================================================
    # 工具方法
    # ==========================================================================
    
    def _mask_mobile(self, mobile: Optional[str]) -> Optional[str]:
        return mask_mobile(mobile)

    def _calc_days_since(self, stat_date: date, last_date: Optional[date]) -> Optional[int]:
        return calc_days_since(stat_date, last_date)


# 便于外部导入
__all__ = ['AssistantCustomerTask']
