# AI_CHANGELOG
# - 2026-03-20 | Prompt: R3 项目类型筛选接口重建 | get_skill_types() 从虚构的 v_cfg_skill_type
#   改为查询 app.v_cfg_area_category（真实 RLS 视图），头部插入"不限"选项；
#   get_all_assistants() 移除 _skill_to_category/_category_to_skill 映射字典，
#   改为 _valid_categories set 直接比较；_project_filter_clause() 移除 _project_to_category
#   映射字典，直接用 category_code。
# - 2026-03-20 | Prompt: RNS1.3 FDW 列名修正 | 修正 17 处列名映射（design.md 理想名 → 实际视图列名），
#   gift_rows 每个 cell 改为 GiftCell dict 避免 Pydantic 校验失败，
#   v_dws_member_spending_power_index 降级为空列表，skill_filter 暂不生效
# - 2026-03-24 | Prompt: 修复小程序前端没有档位进度 | batch_query_for_task_list 增加第 8 步
#   查询 app.v_cfg_performance_tier 获取当前有效档位配置，返回 performance_tiers 字段。
# - 2026-03-24 | Prompt: bonus_money 需要 bonus_deduction_ratio | batch_query_for_task_list 第 8 步
#   和 get_performance_tiers 均增加 bonus_deduction_ratio 列（打赏课抽成比例）。
# - 2026-03-25 | Prompt: 保底 relationship_building 任务 | 新增 get_all_service_pairs()，
#   查询 v_dws_member_assistant_relation_index WHERE session_count > 0 的全量关系对，
#   用于保底任务生成（不限 os_label）。
# - 2026-03-25 | Prompt: 任务详情服务记录6项改进 | get_service_records_for_task() 改造：
#   (1) LEFT JOIN v_dim_table 获取台桌名称（同 get_service_records）；
#   (2) 收入改用 settlement_head.assistant_pd_money + assistant_cx_money（助教到手）；
#   (3) LATERAL 子查询 v_dwd_store_goods_sale 聚合商品明细（按 ledger_name GROUP BY SUM）；
#   (4) 返回 drinks 字段（"商品名×数量" 格式）。
# - 2026-03-27 | Prompt: board-finance-integration T2.1-T2.3 |
#   T2.1: get_finance_recharge() 卡余额快照值从 SUM 改为取最后一天（3 查询拆分），
#         新增 consumed 字段从 v_dws_finance_daily_summary.card_consume_total SUM 聚合。
#   T2.2: get_finance_cashflow() 移除"储值卡消费"项，新增"纸币现金"/"扫码收款"拆分
#         （依赖 T1.1 新增字段 cash_paper_amount/scan_pay_amount）。
#   T2.3: get_finance_coach_analysis() 修复小时均价分母（base_hours/bonus_hours），
#         pay 改为对客收费（hours×course_price），share 改为球房分成（hours×deduction）。

"""
ETL RLS 视图查询封装服务

直连 ETL 库（test_etl_feiqiu）查询 app.v_* RLS 视图，实现门店隔离。

⚠️ 架构说明：不使用 zqyy_app 的 fdw_etl.* foreign table，而是直连 ETL 库。
原因：postgres_fdw 不传递自定义 GUC 参数到远端连接，导致 RLS 视图的
current_setting('app.current_site_id') 在远端未设置而报错。
直连 ETL 库后，SET LOCAL app.current_site_id 在同一连接上生效。

⚠️ DWD-DOC 强制规则在此模块统一实施：
- 规则 1: 收入使用 ledger_amount（对应 items_sum 口径）
- 规则 2: 助教费用使用 base_income + bonus_income（对应 pd/cx 拆分）
- DQ-6: 会员信息通过 tenant_member_id JOIN v_dim_member (scd2_is_current=1)
- DQ-7: 会员卡通过 tenant_member_id JOIN v_dim_member_card_account (scd2_is_current=1)
- 废单排除: WHERE is_delete = 0（RLS 视图基于 dwd_assistant_service_log 基表，
  使用 is_delete 而非 _ex 表的 is_trash）

列名映射说明（design.md 理想名 → 实际视图列名）：
  app.v_dwd_assistant_service_log:
    assistant_id → site_assistant_id | member_id → tenant_member_id
    is_trash → is_delete (int, 0=正常) | settle_time → create_time
    service_hours → income_seconds/3600 | items_sum → ledger_amount
    course_type → skill_name | table_name → site_table_id (仅 ID)
  app.v_dws_assistant_salary_calc:
    calc_month → salary_month (date) | coach_level → assistant_level_name
    tier_index → tier_id | basic_hours → base_hours
    total_hours → effective_hours | total_income → gross_salary
    basic_rate → base_course_price | incentive_rate → bonus_course_price
"""

from __future__ import annotations

import logging
from contextlib import contextmanager
from decimal import Decimal
from typing import Any

from app.trace.decorators import trace_service

logger = logging.getLogger(__name__)


def _get_etl_connection(site_id: int):
    """延迟导入 get_etl_readonly_connection，避免模块级导入失败。"""
    from app.database import get_etl_readonly_connection

    return get_etl_readonly_connection(site_id)


@contextmanager
def _fdw_context(conn: Any, site_id: int, *, etl_conn: Any = None):
    """
    上下文管理器：直连 ETL 库 + SET LOCAL app.current_site_id + app.current_business_date。

    ⚠️ 不使用 zqyy_app 的 fdw_etl.* foreign table，而是直连 ETL 库
    查询 app.v_* RLS 视图。原因：postgres_fdw 不传递自定义 GUC 参数
    到远端连接，导致 RLS 视图的 current_setting('app.current_site_id')
    在远端未设置而报错。

    conn 参数保留但不用于 FDW 查询（调用方可能还需要它查 biz.* 表）。

    CHANGE 2026-03-26 | ETL 连接复用：传入 etl_conn 时复用已有连接（不关闭），
    不传时新建连接并在 yield 后自动关闭。避免同一请求内多次新建连接（每次 ~2.6s）。
    CHANGE 2026-05-02 | 同时设置 app.current_business_date / app.current_runtime_mode，
    供 RLS 视图层（C 方案）做日期上界裁剪。conn=None 时降级 live。
    """
    from app.services.runtime_context import (
        MODE_LIVE,
        MODE_SANDBOX,
        get_runtime_context,
    )

    # 业务日：优先从 zqyy_app 业务库的 RuntimeContext 读取；conn 不可用时降级为系统今天
    bd_str = ""
    rt_mode = MODE_LIVE
    try:
        if conn is not None:
            ctx = get_runtime_context(site_id, conn=conn)
            bd_str = ctx.business_date.isoformat()
            rt_mode = MODE_SANDBOX if ctx.is_sandbox else MODE_LIVE
        else:
            from datetime import date as _date
            bd_str = _date.today().isoformat()
    except Exception:
        from datetime import date as _date
        bd_str = _date.today().isoformat()
        rt_mode = MODE_LIVE

    owned = etl_conn is None
    if owned:
        etl_conn = _get_etl_connection(site_id)
    try:
        with etl_conn.cursor() as cur:
            cur.execute("BEGIN")
            cur.execute("SET LOCAL app.current_site_id = %s", (str(site_id),))
            cur.execute("SET LOCAL app.current_business_date = %s", (bd_str,))
            cur.execute("SET LOCAL app.current_runtime_mode = %s", (rt_mode,))
            yield cur
        etl_conn.commit()
    finally:
        if owned:
            etl_conn.close()


@trace_service(description_zh="获取会员信息", description_en="Get member info")
def get_member_info(
    conn: Any, site_id: int, member_ids: list[int],
    *, etl_conn: Any = None,
) -> dict[int, dict]:
    """
    批量查询会员信息（昵称、手机号）。

    ⚠️ DQ-6: 通过 member_id 查询 app.v_dim_member，取 scd2_is_current=1，
    禁止使用 settlement_head.member_phone/member_name。

    返回 {member_id: {"nickname": str, "mobile": str}} 映射。
    """
    if not member_ids:
        return {}

    result: dict[int, dict] = {}
    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        cur.execute(
            """
            SELECT member_id, nickname, mobile
            FROM app.v_dim_member
            WHERE member_id = ANY(%s) AND scd2_is_current = 1
            """,
            (member_ids,),
        )
        for row in cur.fetchall():
            result[row[0]] = {"nickname": row[1], "mobile": row[2]}

    return result


@trace_service(description_zh="获取会员余额", description_en="Get member balance")
def get_member_balance(
    conn: Any, site_id: int, member_ids: list[int],
    *, etl_conn: Any = None,
) -> dict[int, Decimal]:
    """
    批量查询会员储值卡余额。

    ⚠️ DQ-7: 通过 tenant_member_id 关联 app.v_dim_member_card_account，
    取 scd2_is_current=1，禁止使用 settlement_head.member_card_type_name。

    返回 {member_id: balance} 映射。
    """
    if not member_ids:
        return {}

    result: dict[int, Decimal] = {}
    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        # CHANGE 2026-03-29 | 修复多卡膨胀：同一客户多张卡需 SUM 聚合
        cur.execute(
            """
            SELECT tenant_member_id AS member_id, SUM(balance) AS balance
            FROM app.v_dim_member_card_account
            WHERE tenant_member_id = ANY(%s) AND scd2_is_current = 1
            GROUP BY tenant_member_id
            """,
            (member_ids,),
        )
        for row in cur.fetchall():
            result[row[0]] = Decimal(str(row[1])) if row[1] is not None else Decimal("0")

    return result


@trace_service(description_zh="获取最后到店天数", description_en="Get last visit days")
def get_last_visit_days(
    conn: Any, site_id: int, member_ids: list[int],
    *, etl_conn: Any = None,
) -> dict[int, int | None]:
    """
    批量查询客户距上次到店天数。

    来源: app.v_dws_member_consumption_summary。
    consumption_summary 按 stat_date 有多行快照，取最新一行。

    CHANGE 2026-05-02 | 修复客户看板「最近到店」数据不准的问题：
    - 旧版直接用 days_since_last（ETL 在 stat_date 那天预计算的快照值）。
      若 ETL 没跑、跑得迟、或 sandbox_date 与 stat_date 不一致，结果就会严重失真。
    - 新版改为实时计算：``business_date - last_consume_date``，
      仅取 ``stat_date <= business_date`` 的快照行，沙箱模式下也能拿到一致结果。

    返回 {member_id: days_since_visit} 映射，无记录的会员不在结果中。
    """
    if not member_ids:
        return {}

    from app.services.runtime_context import as_runtime_today_param

    ref_date = as_runtime_today_param(site_id, conn=conn)

    result: dict[int, int | None] = {}
    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        cur.execute(
            """
            SELECT DISTINCT ON (member_id)
                   member_id,
                   last_consume_date,
                   stat_date
            FROM app.v_dws_member_consumption_summary
            WHERE member_id = ANY(%s)
              AND stat_date <= %s
            ORDER BY member_id, stat_date DESC
            """,
            (member_ids, ref_date),
        )
        for row in cur.fetchall():
            mid = row[0]
            last_consume = row[1]
            if last_consume is None:
                result[mid] = None
                continue
            try:
                # last_consume_date 在 DWS 中是 date；少数实现可能给 timestamp，统一裁剪
                if hasattr(last_consume, "date"):
                    last_consume = last_consume.date()
                days = (ref_date - last_consume).days
                result[mid] = max(days, 0)
            except Exception:
                result[mid] = None

    return result


@trace_service(description_zh="获取工资计算数据", description_en="Get salary calculation")
def get_salary_calc(
    conn: Any, site_id: int, assistant_id: int, year: int, month: int
) -> dict | None:
    """
    查询助教绩效/档位/收入数据。

    来源: app.v_dws_assistant_salary_calc。
    列名映射:
      salary_month (date, 存储为 YYYY-MM-01) → calc_month
      assistant_level_name → coach_level
      tier_id → tier_index
      base_hours → basic_hours
      effective_hours → total_hours
      gross_salary → total_income
      base_course_price → basic_rate
      bonus_course_price → incentive_rate
      sprint_bonus → bonus_money
      base_income → assistant_pd_money_total
      bonus_income → assistant_cx_money_total

    不存在的字段使用默认值:
      tier_nodes → []  |  total_customers → 0
      next_tier_* → 0  |  tier_completed → False

    返回包含档位、工时、收入等字段的 dict，无数据时返回 None。
    """
    # salary_month 是 date 类型，存储为每月 1 号
    calc_month = f"{year}-{month:02d}-01"

    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT salary_month,
                   assistant_level_name,
                   tier_id,
                   base_hours,
                   bonus_hours,
                   effective_hours,
                   gross_salary,
                   base_course_price,
                   bonus_course_price,
                   sprint_bonus,
                   base_income,
                   bonus_income,
                   room_hours,
                   room_income,
                   total_course_income,
                   total_bonus,
                   top_rank_bonus,
                   recharge_commission,
                   other_bonus,
                   base_deduction,
                   bonus_deduction_ratio
            FROM app.v_dws_assistant_salary_calc
            WHERE assistant_id = %s AND salary_month = %s::date
            """,
            (assistant_id, calc_month),
        )
        row = cur.fetchone()

    if not row:
        return None

    return {
        "calc_month": str(row[0]) if row[0] else calc_month,
        "coach_level": row[1] or "",
        "tier_index": row[2] or 0,
        # 视图无 tier_nodes，由 coach_service._build_tier_nodes() 从 cfg_performance_tier 读取
        "tier_nodes": [],
        "basic_hours": float(row[3]) if row[3] is not None else 0.0,
        "bonus_hours": float(row[4]) if row[4] is not None else 0.0,
        "total_hours": float(row[5]) if row[5] is not None else 0.0,
        "total_income": float(row[6]) if row[6] is not None else 0.0,
        # 视图无 total_customers，需要从服务记录单独统计
        "total_customers": 0,
        "basic_rate": float(row[7]) if row[7] is not None else 0.0,
        "incentive_rate": float(row[8]) if row[8] is not None else 0.0,
        # 视图无 next_tier 信息，由 coach_service 从 tier_nodes 推算
        "next_tier_basic_rate": 0.0,
        "next_tier_incentive_rate": 0.0,
        "next_tier_hours": 0.0,
        "tier_completed": False,
        "bonus_money": float(row[9]) if row[9] is not None else 0.0,
        "assistant_pd_money_total": float(row[10]) if row[10] is not None else 0.0,
        "assistant_cx_money_total": float(row[11]) if row[11] is not None else 0.0,
        # 额外字段：视图中有但 design.md 未列出，保留供后续使用
        "room_hours": float(row[12]) if row[12] is not None else 0.0,
        "room_income": float(row[13]) if row[13] is not None else 0.0,
        "total_course_income": float(row[14]) if row[14] is not None else 0.0,
        "total_bonus": float(row[15]) if row[15] is not None else 0.0,
        # CHANGE 2026-03-24 | 奖金拆分字段 + 当前档位抽成参数，用于 performance 页收入明细和到手费率计算
        "top_rank_bonus": float(row[16]) if row[16] is not None else 0.0,
        "recharge_commission": float(row[17]) if row[17] is not None else 0.0,
        "other_bonus": float(row[18]) if row[18] is not None else 0.0,
        "base_deduction": float(row[19]) if row[19] is not None else 0.0,
        "bonus_deduction_ratio": float(row[20]) if row[20] is not None else 0.0,
    }


def get_monthly_summary(
    conn: Any, site_id: int, assistant_id: int, year: int, month: int
) -> dict | None:
    """
    查询助教当月实时业绩汇总（每日更新）。

    来源: app.v_dws_assistant_monthly_summary。
    用于 banner 展示当月课时/档位/客户数等实时数据，
    不依赖月初结算的 salary_calc。
    """
    stat_month = f"{year}-{month:02d}-01"

    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT effective_hours,
                   base_hours,
                   bonus_hours,
                   room_hours,
                   tier_id,
                   tier_code,
                   tier_name,
                   unique_customers,
                   assistant_level_name,
                   rank_with_ties
            FROM app.v_dws_assistant_monthly_summary
            WHERE assistant_id = %s AND stat_month = %s::date
            """,
            (assistant_id, stat_month),
        )
        row = cur.fetchone()

    if not row:
        return None

    return {
        "effective_hours": float(row[0]) if row[0] is not None else 0.0,
        "base_hours": float(row[1]) if row[1] is not None else 0.0,
        "bonus_hours": float(row[2]) if row[2] is not None else 0.0,
        "room_hours": float(row[3]) if row[3] is not None else 0.0,
        "tier_id": row[4] or 0,
        "tier_code": row[5] or "",
        "tier_name": row[6] or "",
        "unique_customers": row[7] or 0,
        "coach_level": row[8] or "",
        "rank_with_ties": row[9] or 0,
    }


@trace_service(description_zh="批量查询任务列表ETL数据", description_en="Batch query for task list")
def batch_query_for_task_list(
    conn: Any,
    site_id: int,
    assistant_id: int,
    member_ids: list[int],
    year: int,
    month: int,
) -> dict:
    """
    单连接批量查询任务列表所需的所有 ETL 数据。

    CHANGE 2026-03-23: 合并 7 次独立 ETL 连接为 1 次，消除连接开销瓶颈。
    原来每个查询各自 _fdw_context → 新建连接 → BEGIN → SET LOCAL → 查询 → 关闭，
    7 次连接开销约 350-1400ms。合并后仅 1 次连接，节省 ~85% 连接时间。

    返回 {member_info, balance, last_visit, rs, monthly_summary, salary_cur, salary_prev}。
    """
    member_info_map: dict[int, dict] = {}
    balance_map: dict[int, Decimal] = {}
    last_visit_map: dict[int, int | None] = {}
    rs_map: dict[int, Decimal] = {}
    wbi_map: dict[int, dict] = {}
    recent60d_map: dict[int, dict] = {}
    monthly_summary: dict | None = None
    salary_cur: dict | None = None
    salary_prev: dict | None = None

    stat_month = f"{year}-{month:02d}-01"
    prev_year, prev_month = (year, month - 1) if month > 1 else (year - 1, 12)
    prev_stat_month = f"{prev_year}-{prev_month:02d}-01"

    with _fdw_context(conn, site_id) as cur:
        # 1. 会员信息
        if member_ids:
            cur.execute(
                """
                SELECT member_id, nickname, mobile
                FROM app.v_dim_member
                WHERE member_id = ANY(%s) AND scd2_is_current = 1
                """,
                (member_ids,),
            )
            for row in cur.fetchall():
                member_info_map[row[0]] = {"nickname": row[1], "mobile": row[2]}

            # 2. 余额
            cur.execute(
                """
                SELECT tenant_member_id AS member_id, balance
                FROM app.v_dim_member_card_account
                WHERE tenant_member_id = ANY(%s) AND scd2_is_current = 1
                """,
                (member_ids,),
            )
            for row in cur.fetchall():
                balance_map[row[0]] = Decimal(str(row[1])) if row[1] is not None else Decimal("0")

            # 3. 最后到店天数（基于消费汇总表，口径=结算单）
            # FIX: 原查 v_dwd_assistant_service_log 导致无助教服务的客户缺失到店记录
            # CHANGE 2026-05-02 | 实时按 business_date - last_consume_date 计算，
            #                     不再依赖 ETL 预计算的 days_since_last（解决看板显示偏差）。
            from app.services.runtime_context import as_runtime_today_param
            _ref_date = as_runtime_today_param(site_id, conn=conn)
            cur.execute(
                """
                SELECT DISTINCT ON (member_id)
                       member_id, last_consume_date, stat_date
                FROM app.v_dws_member_consumption_summary
                WHERE member_id = ANY(%s)
                  AND stat_date <= %s
                ORDER BY member_id, stat_date DESC
                """,
                (member_ids, _ref_date),
            )
            for row in cur.fetchall():
                mid = row[0]
                last_consume = row[1]
                if last_consume is None:
                    last_visit_map[mid] = None
                    continue
                try:
                    if hasattr(last_consume, "date"):
                        last_consume = last_consume.date()
                    last_visit_map[mid] = max((_ref_date - last_consume).days, 0)
                except Exception:
                    last_visit_map[mid] = None

            # 4. RS 指数
            cur.execute(
                """
                SELECT member_id, COALESCE(rs_display, 0)
                FROM app.v_dws_member_assistant_relation_index
                WHERE assistant_id = %s AND member_id = ANY(%s)
                """,
                (assistant_id, member_ids),
            )
            for row in cur.fetchall():
                rs_map[row[0]] = Decimal(str(row[1]))

            # 4b. WBI 期望到店间隔 + 期望下次到店日期
            # CHANGE 2026-03-24 | 新增：用于任务卡片"预期X天"标签
            cur.execute(
                """
                SELECT member_id, ideal_interval_days, ideal_next_visit_date
                FROM app.v_dws_member_winback_index
                WHERE member_id = ANY(%s)
                """,
                (member_ids,),
            )
            for row in cur.fetchall():
                wbi_map[row[0]] = {
                    "ideal_interval_days": float(row[1]) if row[1] is not None else None,
                    "ideal_next_visit_date": row[2].isoformat() if row[2] is not None else None,
                }

            # 4c. 近60天服务汇总（口径同 task-detail serviceSummary）
            # CHANGE 2026-03-27 | 新增：按 assistant_id + member_id 聚合近60天总时长和到手收入
            # 到手 = hours × net_rate（基础课: base_price - deduction; 激励课: bonus_price × (1 - ratio)）
            cur.execute(
                """
                SELECT sl.tenant_member_id,
                       SUM(sl.income_seconds / 3600.0) AS total_hours,
                       SUM(
                           CASE
                               WHEN sl.skill_name ILIKE '%%激励%%' OR sl.skill_name ILIKE '%%超休%%'
                               THEN (sl.income_seconds / 3600.0) * COALESCE(sc.bonus_course_price * (1 - sc.bonus_deduction_ratio), 0)
                               ELSE (sl.income_seconds / 3600.0) * COALESCE(sc.base_course_price - sc.base_deduction, 0)
                           END
                       ) AS total_income
                FROM app.v_dwd_assistant_service_log sl
                LEFT JOIN app.v_dws_assistant_salary_calc sc
                    ON sl.site_assistant_id = sc.assistant_id
                    AND date_trunc('month', sl.create_time)::date = sc.salary_month
                WHERE sl.site_assistant_id = %s
                  AND sl.tenant_member_id = ANY(%s)
                  AND sl.is_delete = 0
                  AND sl.create_time >= (%s::date - INTERVAL '60 days')::timestamptz
                  AND sl.create_time <  (%s::date + INTERVAL '1 day')::timestamptz
                GROUP BY sl.tenant_member_id
                """,
                (assistant_id, member_ids, _ref_date, _ref_date),
            )
            for row in cur.fetchall():
                recent60d_map[row[0]] = {
                    "hours": round(float(row[1]), 2) if row[1] is not None else 0.0,
                    "income": round(float(row[2]), 2) if row[2] is not None else 0.0,
                }

        # 5. 当月 monthly_summary
        cur.execute(
            """
            SELECT effective_hours, base_hours, bonus_hours, room_hours,
                   tier_id, tier_code, tier_name, unique_customers,
                   assistant_level_name, rank_with_ties
            FROM app.v_dws_assistant_monthly_summary
            WHERE assistant_id = %s AND stat_month = %s::date
            """,
            (assistant_id, stat_month),
        )
        row = cur.fetchone()
        if row:
            monthly_summary = {
                "effective_hours": float(row[0]) if row[0] is not None else 0.0,
                "base_hours": float(row[1]) if row[1] is not None else 0.0,
                "bonus_hours": float(row[2]) if row[2] is not None else 0.0,
                "room_hours": float(row[3]) if row[3] is not None else 0.0,
                "tier_id": row[4] or 0,
                "tier_code": row[5] or "",
                "tier_name": row[6] or "",
                "unique_customers": row[7] or 0,
                "coach_level": row[8] or "",
                "rank_with_ties": row[9] or 0,
            }

        # 6. 当月 salary_calc
        cur.execute(
            """
            SELECT salary_month, assistant_level_name, tier_id,
                   base_hours, bonus_hours, effective_hours, gross_salary,
                   base_course_price, bonus_course_price, sprint_bonus,
                   base_income, bonus_income, room_hours, room_income,
                   total_course_income, total_bonus
            FROM app.v_dws_assistant_salary_calc
            WHERE assistant_id = %s AND salary_month = %s::date
            """,
            (assistant_id, stat_month),
        )
        row = cur.fetchone()
        if row:
            salary_cur = _parse_salary_row(row, stat_month)

        # 7. 上月 salary_calc
        cur.execute(
            """
            SELECT salary_month, assistant_level_name, tier_id,
                   base_hours, bonus_hours, effective_hours, gross_salary,
                   base_course_price, bonus_course_price, sprint_bonus,
                   base_income, bonus_income, room_hours, room_income,
                   total_course_income, total_bonus
            FROM app.v_dws_assistant_salary_calc
            WHERE assistant_id = %s AND salary_month = %s::date
            """,
            (assistant_id, prev_stat_month),
        )
        row = cur.fetchone()
        if row:
            salary_prev = _parse_salary_row(row, prev_stat_month)

        # 8. 绩效档位配置（用于构建 tier_nodes + bonus_money 计算）
        # CHANGE 2026-03-24 | 增加 bonus_deduction_ratio 用于打赏课抽成差额计算
        # CHANGE 2026-05-02 | 用 business_date 替代 CURRENT_DATE，沙箱按当时生效档位
        from app.services.runtime_context import as_runtime_today_param as _rt_today
        _ref_date = _rt_today(site_id, conn=conn)
        cur.execute(
            """
            SELECT tier_id, tier_code, tier_name, tier_level,
                   min_hours, max_hours, base_deduction, bonus_deduction_ratio
            FROM app.v_cfg_performance_tier
            WHERE effective_from <= %s::date
              AND effective_to >= %s::date
            ORDER BY tier_level
            """,
            (_ref_date, _ref_date),
        )
        tier_rows = cur.fetchall()
        performance_tiers = [
            {
                "tier_id": r[0],
                "tier_code": r[1],
                "tier_name": r[2],
                "tier_level": r[3],
                "min_hours": float(r[4]) if r[4] is not None else 0.0,
                "max_hours": float(r[5]) if r[5] is not None else None,
                "base_deduction": float(r[6]) if r[6] is not None else 0.0,
                "bonus_deduction_ratio": float(r[7]) if r[7] is not None else 0.0,
            }
            for r in tier_rows
        ]

    return {
        "member_info": member_info_map,
        "balance": balance_map,
        "last_visit": last_visit_map,
        "rs": rs_map,
        "wbi": wbi_map,
        "recent60d": recent60d_map,
        "monthly_summary": monthly_summary,
        "salary_cur": salary_cur,
        "salary_prev": salary_prev,
        "performance_tiers": performance_tiers,
    }


def _parse_salary_row(row: tuple, fallback_month: str) -> dict:
    """解析 salary_calc 查询结果行为 dict。"""
    return {
        "calc_month": str(row[0]) if row[0] else fallback_month,
        "coach_level": row[1] or "",
        "tier_index": row[2] or 0,
        "tier_nodes": [],
        "basic_hours": float(row[3]) if row[3] is not None else 0.0,
        "bonus_hours": float(row[4]) if row[4] is not None else 0.0,
        "total_hours": float(row[5]) if row[5] is not None else 0.0,
        "total_income": float(row[6]) if row[6] is not None else 0.0,
        "total_customers": 0,
        "basic_rate": float(row[7]) if row[7] is not None else 0.0,
        "incentive_rate": float(row[8]) if row[8] is not None else 0.0,
        "next_tier_basic_rate": 0.0,
        "next_tier_incentive_rate": 0.0,
        "next_tier_hours": 0.0,
        "tier_completed": False,
        "bonus_money": float(row[9]) if row[9] is not None else 0.0,
        "assistant_pd_money_total": float(row[10]) if row[10] is not None else 0.0,
        "assistant_cx_money_total": float(row[11]) if row[11] is not None else 0.0,
        "room_hours": float(row[12]) if row[12] is not None else 0.0,
        "room_income": float(row[13]) if row[13] is not None else 0.0,
        "total_course_income": float(row[14]) if row[14] is not None else 0.0,
        "total_bonus": float(row[15]) if row[15] is not None else 0.0,
    }


@trace_service(description_zh="获取绩效档位配置", description_en="Get performance tiers")
def get_performance_tiers(
    conn: Any, site_id: int
) -> list[dict]:
    """
    查询当前有效的绩效档位配置。

    来源: app.v_cfg_performance_tier（RLS 视图）。
    按 tier_level 升序返回，仅包含当前日期有效的档位。

    ⚠️ feiqiu-data-rules 规则 6: 绩效档位必须从配置表读取，禁止硬编码。

    返回 [{tier_id, tier_code, tier_name, tier_level, min_hours, max_hours,
           base_deduction, bonus_deduction_ratio}, ...]。
    CHANGE 2026-05-02 | 用 business_date 替代 CURRENT_DATE，沙箱按当时生效档位
    """
    from app.services.runtime_context import as_runtime_today_param as _rt_today
    ref_date = _rt_today(site_id, conn=conn)
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT tier_id, tier_code, tier_name, tier_level,
                   min_hours, max_hours, base_deduction, bonus_deduction_ratio
            FROM app.v_cfg_performance_tier
            WHERE effective_from <= %s::date
              AND effective_to >= %s::date
            ORDER BY tier_level
            """,
            (ref_date, ref_date),
        )
        rows = cur.fetchall()

    return [
        {
            "tier_id": r[0],
            "tier_code": r[1],
            "tier_name": r[2],
            "tier_level": r[3],
            "min_hours": float(r[4]) if r[4] is not None else 0.0,
            "max_hours": float(r[5]) if r[5] is not None else None,
            "base_deduction": float(r[6]) if r[6] is not None else 0.0,
            "bonus_deduction_ratio": float(r[7]) if r[7] is not None else 0.0,
        }
        for r in rows
    ]


@trace_service(description_zh="获取等级映射", description_en="Get level map")
def get_level_map(conn: Any, site_id: int) -> dict[int, str]:
    """
    从 cfg_assistant_level_price 动态读取 level_code → level_name 映射。

    ⚠️ feiqiu-data-rules 规则 6: 等级名称必须从配置表读取，禁止硬编码。

    返回 {8: "助教管理", 10: "初级", 20: "中级", 30: "高级", 40: "星级"}。
    查询失败时返回空 dict（调用方应优雅降级）。
    """
    try:
        from app.services.runtime_context import as_runtime_today_param as _rt_today
        ref_date = _rt_today(site_id, conn=conn)
        with _fdw_context(conn, site_id) as cur:
            cur.execute(
                """
                SELECT DISTINCT level_code, level_name
                FROM app.v_cfg_assistant_level_price
                WHERE effective_from <= %s::date
                  AND effective_to >= %s::date
                ORDER BY level_code
                """,
                (ref_date, ref_date),
            )
            return {row[0]: row[1] for row in cur.fetchall()}
    except Exception:
        return {}


@trace_service(description_zh="获取服务记录汇总", description_en="Get service records summary")
def get_service_records_summary(
    conn: Any,
    site_id: int,
    assistant_id: int,
    year: int,
    month: int,
) -> dict:
    """
    单条 SQL 直接聚合月度汇总：count / sum(hours) / sum(income)。

    用途：替代"先拉全量再 Python 算 summary"的高耗模式（PERF-2）。
    口径与 get_service_records 完全一致（同表/同 JOIN/同费率公式）。
    返回 { total_count, total_hours, total_hours_raw, total_income }；
    total_hours_raw 暂沿用 0.0（DWD 层服务记录无折前时长字段）。
    """
    start_date = f"{year}-{month:02d}-01"
    if month == 12:
        end_date = f"{year + 1}-01-01"
    else:
        end_date = f"{year}-{month + 1:02d}-01"

    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT COUNT(*),
                   COALESCE(SUM(sl.income_seconds / 3600.0), 0),
                   COALESCE(SUM(
                       CASE
                           WHEN sl.skill_name ILIKE '%%激励%%' OR sl.skill_name ILIKE '%%超休%%'
                           THEN (sl.income_seconds / 3600.0) * COALESCE(sc.bonus_course_price * (1 - sc.bonus_deduction_ratio), 0)
                           ELSE (sl.income_seconds / 3600.0) * COALESCE(sc.base_course_price - sc.base_deduction, 0)
                       END
                   ), 0)
            FROM app.v_dwd_assistant_service_log sl
            LEFT JOIN app.v_dws_assistant_salary_calc sc
                ON sl.site_assistant_id = sc.assistant_id
                AND date_trunc('month', sl.create_time)::date = sc.salary_month
            WHERE sl.site_assistant_id = %s AND sl.is_delete = 0
              AND sl.create_time >= %s::timestamptz
              AND sl.create_time < %s::timestamptz
            """,
            (assistant_id, start_date, end_date),
        )
        row = cur.fetchone()

    if not row:
        return {"total_count": 0, "total_hours": 0.0, "total_hours_raw": 0.0, "total_income": 0.0}

    return {
        "total_count": int(row[0] or 0),
        "total_hours": round(float(row[1] or 0), 2),
        "total_hours_raw": 0.0,  # DWD 层无折前时长字段；与原 compute_summary 行为一致
        "total_income": round(float(row[2] or 0), 2),
    }


@trace_service(description_zh="获取服务记录", description_en="Get service records")
def get_service_records(
    conn: Any,
    site_id: int,
    assistant_id: int,
    year: int,
    month: int,
    limit: int,
    offset: int,
) -> list[dict]:
    """
    查询助教服务记录明细。

    来源: app.v_dwd_assistant_service_log。
    列名映射:
      assistant_service_id → id
      site_assistant_id → assistant_id（WHERE 过滤用）
      tenant_member_id → member_id
      is_delete = 0 → 废单排除（RLS 视图用 is_delete 而非 is_trash）
      create_time → settle_time
      start_use_time → start_time
      last_use_time → end_time
      income_seconds / 3600.0 → service_hours（业绩时长）
      ledger_amount → income（items_sum 口径）
      skill_name → course_type（数据库原始值，不做二次映射）
      dim_table.table_name → table_name（桌号名称，JOIN dim_table 获取）

    CHANGE 2026-03-26 | 收入口径: JOIN salary_calc 取到手费率，hours × net_rate 算真正到手

    ⚠️ DQ-6: 客户姓名通过 tenant_member_id LEFT JOIN v_dim_member (scd2_is_current=1)。

    返回按 create_time DESC 排序的记录列表。
    """
    start_date = f"{year}-{month:02d}-01"
    if month == 12:
        end_date = f"{year + 1}-01-01"
    else:
        end_date = f"{year}-{month + 1:02d}-01"

    records: list[dict] = []
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT sl.assistant_service_id,
                   dm.nickname AS customer_name,
                   sl.tenant_member_id,
                   sl.create_time,
                   sl.start_use_time,
                   sl.last_use_time,
                   sl.income_seconds / 3600.0 AS service_hours,
                   sl.skill_name,
                   COALESCE(dt.table_name, '') AS table_name,
                   -- CHANGE 2026-03-26 | 到手 = hours × net_rate
                   CASE
                       WHEN sl.skill_name ILIKE '%%激励%%' OR sl.skill_name ILIKE '%%超休%%'
                       THEN (sl.income_seconds / 3600.0) * COALESCE(sc.bonus_course_price * (1 - sc.bonus_deduction_ratio), 0)
                       ELSE (sl.income_seconds / 3600.0) * COALESCE(sc.base_course_price - sc.base_deduction, 0)
                   END AS income
            FROM app.v_dwd_assistant_service_log sl
            LEFT JOIN app.v_dim_member dm
                ON sl.tenant_member_id = dm.member_id
                AND dm.scd2_is_current = 1
            LEFT JOIN app.v_dim_table dt
                ON sl.site_table_id = dt.table_id
                AND dt.scd2_is_current = 1
            LEFT JOIN app.v_dws_assistant_salary_calc sc
                ON sl.site_assistant_id = sc.assistant_id
                AND date_trunc('month', sl.create_time)::date = sc.salary_month
            WHERE sl.site_assistant_id = %s AND sl.is_delete = 0
              AND sl.create_time >= %s::timestamptz
              AND sl.create_time < %s::timestamptz
            ORDER BY sl.create_time DESC
            LIMIT %s OFFSET %s
            """,
            (assistant_id, start_date, end_date, limit, offset),
        )
        for row in cur.fetchall():
            records.append({
                "id": row[0],
                "customer_name": row[1],
                "member_id": row[2],
                "settle_time": row[3],
                "start_time": row[4],
                "end_time": row[5],
                "service_hours": float(row[6]) if row[6] is not None else 0.0,
                "course_type": row[7],
                "table_name": row[8] or "",
                "income": float(row[9]) if row[9] is not None else 0.0,
            })

    return records


@trace_service(description_zh="获取近90天服务记录", description_en="Get service records for 90 days")
def get_service_records_90days(
    conn: Any,
    site_id: int,
    assistant_id: int,
    ref_date: str,
) -> list[dict]:
    """
    查询助教近90天的服务记录（常客统计用）。

    ref_date: 参考日期（如 "2026-03-01"），向前推90天。
    CHANGE 2026-03-26 | 收入口径: JOIN salary_calc 取到手费率
    返回 member_id 级聚合：{ member_id, customer_name, count, total_hours, total_income }。
    """
    records: list[dict] = []
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT sl.tenant_member_id,
                   dm.nickname AS customer_name,
                   COUNT(*) AS cnt,
                   SUM(sl.income_seconds / 3600.0) AS total_hours,
                   SUM(
                       CASE
                           WHEN sl.skill_name ILIKE '%%激励%%' OR sl.skill_name ILIKE '%%超休%%'
                           THEN (sl.income_seconds / 3600.0) * COALESCE(sc.bonus_course_price * (1 - sc.bonus_deduction_ratio), 0)
                           ELSE (sl.income_seconds / 3600.0) * COALESCE(sc.base_course_price - sc.base_deduction, 0)
                       END
                   ) AS total_income
            FROM app.v_dwd_assistant_service_log sl
            LEFT JOIN app.v_dim_member dm
                ON sl.tenant_member_id = dm.member_id
                AND dm.scd2_is_current = 1
            LEFT JOIN app.v_dws_assistant_salary_calc sc
                ON sl.site_assistant_id = sc.assistant_id
                AND date_trunc('month', sl.create_time)::date = sc.salary_month
            WHERE sl.site_assistant_id = %s AND sl.is_delete = 0
              AND sl.create_time >= (%s::date - INTERVAL '90 days')::timestamptz
              AND sl.create_time < %s::timestamptz
            GROUP BY sl.tenant_member_id, dm.nickname
            """,
            (assistant_id, ref_date, ref_date),
        )
        for row in cur.fetchall():
            records.append({
                "member_id": row[0],
                "customer_name": row[1] or "未知客户",
                "count": row[2],
                "total_hours": float(row[3]) if row[3] is not None else 0.0,
                "total_income": float(row[4]) if row[4] is not None else 0.0,
            })
    return records


@trace_service(description_zh="获取任务关联服务记录", description_en="Get service records for task")
def get_service_records_for_task(
    conn: Any,
    site_id: int,
    assistant_id: int,
    member_id: int,
    limit: int,
) -> list[dict]:
    """
    查询特定客户的服务记录（TASK-2 用）。

    类似 get_service_records，但按 tenant_member_id 过滤，不限月份范围。
    ⚠️ 废单排除: WHERE is_delete = 0。
    ⚠️ DQ-6: 客户姓名通过 tenant_member_id LEFT JOIN v_dim_member (scd2_is_current=1)。

    CHANGE 2026-03-25 | 台桌名称: LEFT JOIN v_dim_table 获取 table_name（同 get_service_records）
    CHANGE 2026-03-26 | 收入口径: JOIN salary_calc 取到手费率，hours × net_rate 算真正到手
    CHANGE 2026-03-25 | 酒水商品: 子查询 v_dwd_store_goods_sale 聚合 "商品名×数量" 明细

    返回按 create_time DESC 排序的记录列表，最多 limit 条。
    """
    records: list[dict] = []
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT sl.assistant_service_id,
                   dm.nickname AS customer_name,
                   sl.tenant_member_id,
                   sl.create_time,
                   sl.start_use_time,
                   sl.last_use_time,
                   sl.income_seconds / 3600.0 AS service_hours,
                   -- CHANGE 2026-03-27 | real_use_seconds 不用于业绩，惩罚在 DWS 层计算
                   -- DWD 层服务记录卡片不显示折前时长，统一返回 NULL
                   NULL AS service_hours_raw,
                   sl.skill_name,
                   COALESCE(dt.table_name, '') AS table_name,
                   -- CHANGE 2026-03-26 | 到手 = hours × net_rate（基础课: base_price - deduction; 激励课: bonus_price × (1 - ratio)）
                   CASE
                       WHEN sl.skill_name ILIKE '%%激励%%' OR sl.skill_name ILIKE '%%超休%%'
                       THEN (sl.income_seconds / 3600.0) * COALESCE(sc.bonus_course_price * (1 - sc.bonus_deduction_ratio), 0)
                       ELSE (sl.income_seconds / 3600.0) * COALESCE(sc.base_course_price - sc.base_deduction, 0)
                   END AS income,
                   gs_agg.drinks
            FROM app.v_dwd_assistant_service_log sl
            LEFT JOIN app.v_dim_member dm
                ON sl.tenant_member_id = dm.member_id
                AND dm.scd2_is_current = 1
            LEFT JOIN app.v_dim_table dt
                ON sl.site_table_id = dt.table_id
                AND dt.scd2_is_current = 1
            LEFT JOIN app.v_dws_assistant_salary_calc sc
                ON sl.site_assistant_id = sc.assistant_id
                AND date_trunc('month', sl.create_time)::date = sc.salary_month
            LEFT JOIN LATERAL (
                SELECT string_agg(gs.ledger_name || '×' || gs.total_count, '、') AS drinks
                FROM (
                    SELECT ledger_name, SUM(ledger_count) AS total_count
                    FROM app.v_dwd_store_goods_sale
                    WHERE order_settle_id = sl.order_settle_id
                      AND is_delete = 0
                    GROUP BY ledger_name
                ) gs
            ) gs_agg ON true
            WHERE sl.site_assistant_id = %s
              AND sl.tenant_member_id = %s
              AND sl.is_delete = 0
            ORDER BY sl.create_time DESC
            LIMIT %s
            """,
            (assistant_id, member_id, limit),
        )
        for row in cur.fetchall():
            svc_hours = float(row[6]) if row[6] is not None else 0.0
            # CHANGE 2026-03-27 | 折前时长：SQL 层已判断，无惩罚时 row[7] 为 NULL
            svc_hours_raw = float(row[7]) if row[7] is not None else None
            records.append({
                "id": row[0],
                "customer_name": row[1],
                "member_id": row[2],
                "settle_time": row[3],
                "start_time": row[4],
                "end_time": row[5],
                "service_hours": round(svc_hours, 2),
                "service_hours_raw": round(svc_hours_raw, 2) if svc_hours_raw is not None else None,
                "course_type": row[8],
                "table_name": row[9] or "",
                "income": float(row[10]) if row[10] is not None else 0.0,
                "is_estimate": False,
                "drinks": row[11],
            })

    return records


@trace_service(description_zh="获取近60天消费金额", description_en="Get 60-day consumption")
def get_consumption_60d(
    conn: Any, site_id: int, member_id: int,
    *, etl_conn: Any = None,
) -> Decimal | None:
    """
    查询客户近 60 天消费金额。

    来源: app.v_dws_member_consumption_summary（DWS 预聚合表）。
    与 board-customer spend60 维度统一口径：items_sum，60天窗口，日粒度。
    取最新 stat_date 的快照行。
    """
    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        cur.execute(
            """
            SELECT consume_amount_60d
            FROM app.v_dws_member_consumption_summary
            WHERE member_id = %s
            ORDER BY stat_date DESC
            LIMIT 1
            """,
            (member_id,),
        )
        row = cur.fetchone()
    return Decimal(str(row[0])) if row and row[0] is not None else None


@trace_service(description_zh="获取关系指数", description_en="Get relation index")
def get_relation_index(
    conn: Any, site_id: int, member_id: int,
    *, etl_conn: Any = None,
) -> list[dict]:
    """
    查询客户与助教的关系指数列表。

    来源: app.v_dws_member_assistant_relation_index。
    返回按 relation_index 降序排列的列表。
    """
    records: list[dict] = []
    # CHANGE 2026-03-19 | 修正列名：实际视图列为 assistant_id/rs_display/session_count/total_duration_minutes
    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        cur.execute(
            """
            SELECT assistant_id,
                   rs_display AS relation_index,
                   session_count AS service_count,
                   total_duration_minutes / 60.0 AS total_hours
            FROM app.v_dws_member_assistant_relation_index
            WHERE member_id = %s
            ORDER BY rs_display DESC
            """,
            (member_id,),
        )
        for row in cur.fetchall():
            records.append({
                "assistant_id": row[0],
                "relation_index": float(row[1]) if row[1] is not None else 0.0,
                "service_count": row[2] or 0,
                "total_hours": float(row[3]) if row[3] is not None else 0.0,
                "total_income": 0.0,
            })
    return records


# AI_CHANGELOG
# - 2026-03-20: R1 修复 — 5 列(table_charge_money, goods_money, assistant_pd_money,
#   assistant_cx_money, settle_type)从 sl(service_log) 改为 sh(settlement_head)，
#   添加 LEFT JOIN v_dwd_settlement_head，WHERE settle_type 引用也改为 sh。
#   原因：这些字段属于结算单头表，不在助教服务日志视图中。
#   验证：MCP 端到端查询通过。
@trace_service(description_zh="获取消费记录", description_en="Get consumption records")
def get_consumption_records(
    conn: Any, site_id: int, member_id: int, limit: int, offset: int,
    *, etl_conn: Any = None,
    start_date: str | None = None, end_date: str | None = None,
) -> list[dict]:
    """
    查询客户消费记录（CUST-1 consumptionRecords 用）。

    按结算单（order_settle_id）粒度返回，同一结算单下的多个助教聚合到 coaches 数组。
    来源: v_dwd_settlement_head + v_dwd_assistant_service_log + v_dim_assistant。
    ⚠️ DWD-DOC 规则 1: totalAmount 使用 SUM(ledger_amount)（items_sum 口径）。
    ⚠️ 废单排除: is_delete = 0。
    ⚠️ 正向交易: settle_type IN (1, 3)。
    """
    records: list[dict] = []
    date_clause = ""
    date_params: list = []
    if start_date:
        date_clause += " AND sh.create_time >= %s::timestamptz"
        date_params.append(start_date)
    if end_date:
        date_clause += " AND sh.create_time < %s::timestamptz"
        date_params.append(end_date)

    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        cur.execute(
            f"""
            SELECT sh.order_settle_id AS id,
                   sh.create_time AS settle_time,
                   MIN(sl.start_use_time) AS start_time,
                   MAX(sl.last_use_time) AS end_time,
                   SUM(sl.income_seconds) / 3600.0 AS service_hours,
                   SUM(sl.ledger_amount) AS total_amount,
                   MIN(sl.site_table_id) AS table_id,
                   sh.table_charge_money,
                   sh.goods_money,
                   sh.consume_money,
                   sh.adjust_amount,
                   gs_agg.drinks,
                   json_agg(json_build_object(
                       'assistant_id', sl.site_assistant_id,
                       'assistant_name', COALESCE(da.nickname, da.real_name, ''),
                       'assistant_level', da.level,
                       'service_hours', sl.income_seconds / 3600.0,
                       'ledger_amount', sl.ledger_amount,
                       'course_type', sl.skill_name
                   ) ORDER BY sl.ledger_amount DESC NULLS LAST) AS coaches_json
            FROM app.v_dwd_settlement_head sh
            INNER JOIN app.v_dwd_assistant_service_log sl
                ON sh.order_settle_id = sl.order_settle_id
               AND sl.tenant_member_id = %s
               AND sl.is_delete = 0
            LEFT JOIN app.v_dim_assistant da
                ON sl.site_assistant_id = da.assistant_id
               AND da.scd2_is_current = 1
            LEFT JOIN LATERAL (
                SELECT string_agg(gs.ledger_name || '*' || gs.total_count, ' | ' ORDER BY gs.subtotal DESC) AS drinks
                FROM (
                    SELECT ledger_name,
                           SUM(ledger_count) AS total_count,
                           SUM(ledger_amount) AS subtotal
                    FROM app.v_dwd_store_goods_sale
                    WHERE order_settle_id = sh.order_settle_id
                      AND is_delete = 0
                    GROUP BY ledger_name
                ) gs
            ) gs_agg ON true
            WHERE sh.settle_type IN (1, 3)
              {date_clause}
            GROUP BY sh.order_settle_id, sh.create_time,
                     sh.table_charge_money, sh.goods_money,
                     sh.consume_money, sh.adjust_amount,
                     gs_agg.drinks
            ORDER BY sh.create_time DESC
            LIMIT %s OFFSET %s
            """,
            (member_id,) + tuple(date_params) + (limit, offset),
        )
        for row in cur.fetchall():
            records.append({
                "id": str(row[0]) if row[0] else "",
                "settle_time": row[1],
                "start_time": row[2],
                "end_time": row[3],
                "service_hours": float(row[4]) if row[4] is not None else 0.0,
                "total_amount": float(row[5]) if row[5] is not None else 0.0,
                "table_id": row[6],
                "table_charge_money": float(row[7]) if row[7] is not None else 0.0,
                "goods_money": float(row[8]) if row[8] is not None else 0.0,
                "consume_money": float(row[9]) if row[9] is not None else 0.0,
                "adjust_amount": float(row[10]) if row[10] is not None else 0.0,
                "drinks": row[11],
                "coaches_json": row[12] or [],
            })
    return records


@trace_service(description_zh="获取累计服务次数", description_en="Get total service count")
def get_total_service_count(
    conn: Any, site_id: int, member_id: int,
    *, etl_conn: Any = None,
    assistant_id: int | None = None,
) -> int:
    """
    查询客户累计服务总次数（跨所有月份）。

    来源: app.v_dwd_assistant_service_log。
    ⚠️ 废单排除: is_delete = 0。
    CHANGE 2026-03-27 | 新增 assistant_id 过滤，只统计指定助教的服务次数。
    """
    where = "tenant_member_id = %s AND is_delete = 0"
    params: list = [member_id]
    if assistant_id:
        where += " AND site_assistant_id = %s"
        params.append(assistant_id)
    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        cur.execute(
            f"SELECT COUNT(*) FROM app.v_dwd_assistant_service_log WHERE {where}",
            params,
        )
        row = cur.fetchone()
    return row[0] if row else 0


@trace_service(description_zh="获取助教近60天统计", description_en="Get coach 60-day stats")
def get_coach_60d_stats(
    conn: Any, site_id: int, assistant_id: int, member_id: int
) -> dict:
    """
    查询特定助教对特定客户的近 60 天统计。

    来源: app.v_dwd_assistant_service_log。
    ⚠️ 废单排除: is_delete = 0。
    CHANGE 2026-05-02 | 用 business_date 替代 CURRENT_DATE，沙箱不读「未来」60 天。
    返回 {service_count, total_hours, avg_hours}。
    """
    from app.services.runtime_context import as_runtime_today_param
    ref_date = as_runtime_today_param(site_id, conn=conn)
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT COUNT(*) AS service_count,
                   COALESCE(SUM(income_seconds / 3600.0), 0) AS total_hours,
                   CASE WHEN COUNT(*) > 0
                        THEN SUM(income_seconds / 3600.0) / COUNT(*)
                        ELSE 0 END AS avg_hours
            FROM app.v_dwd_assistant_service_log
            WHERE site_assistant_id = %s
              AND tenant_member_id = %s
              AND is_delete = 0
              AND create_time >= (%s::date - INTERVAL '60 days')::timestamptz
              AND create_time <  (%s::date + INTERVAL '1 day')::timestamptz
            """,
            (assistant_id, member_id, ref_date, ref_date),
        )
        row = cur.fetchone()
    if not row:
        return {"service_count": 0, "total_hours": 0.0, "avg_hours": 0.0}
    return {
        "service_count": row[0] or 0,
        "total_hours": float(row[1]) if row[1] is not None else 0.0,
        "avg_hours": float(row[2]) if row[2] is not None else 0.0,
    }


@trace_service(description_zh="获取客户服务记录", description_en="Get customer service records")
def get_customer_service_records(
    conn: Any,
    site_id: int,
    member_id: int,
    year: int,
    month: int,
    table: str | None,
    limit: int,
    offset: int,
    *,
    etl_conn: Any = None,
    assistant_id: int | None = None,
) -> tuple[list[dict], int]:
    """
    查询客户按月服务记录（CUST-2 用）。

    来源: app.v_dwd_assistant_service_log + v_dim_table + v_dws_assistant_salary_calc + v_dwd_store_goods_sale。
    ⚠️ DQ-6: 助教姓名通过 v_dim_assistant 获取。
    ⚠️ 废单排除: is_delete = 0。
    CHANGE 2026-03-27 | 新增 assistant_id 过滤，只返回指定助教的服务记录。
    CHANGE 2026-03-27 | 统一卡片数据：复用 get_service_records_for_task 的 SQL 模式
      — 台桌名称: LEFT JOIN v_dim_table
      — 到手金额: LEFT JOIN salary_calc，hours × net_rate
      — 酒水商品: LEFT JOIN LATERAL v_dwd_store_goods_sale 聚合
    返回 (records, total_count)。
    """
    start_date = f"{year}-{month:02d}-01"
    if month == 12:
        end_date = f"{year + 1}-01-01"
    else:
        end_date = f"{year}-{month + 1:02d}-01"

    base_where = """
        sl.tenant_member_id = %s
        AND sl.is_delete = 0
        AND sl.create_time >= %s::timestamptz
        AND sl.create_time < %s::timestamptz
    """
    params: list = [member_id, start_date, end_date]
    if assistant_id:
        base_where += " AND sl.site_assistant_id = %s"
        params.append(assistant_id)
    if table:
        base_where += " AND sl.site_table_id::text = %s"
        params.append(table)

    records: list[dict] = []
    total_count = 0
    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        # 总数查询
        cur.execute(
            f"SELECT COUNT(*) FROM app.v_dwd_assistant_service_log sl WHERE {base_where}",
            params,
        )
        row = cur.fetchone()
        total_count = row[0] if row else 0

        # CHANGE 2026-03-27 | 统一卡片数据：与 get_service_records_for_task 对齐
        # — 台桌名称、到手金额（salary_calc）、酒水商品（LATERAL）
        cur.execute(
            f"""
            SELECT sl.assistant_service_id AS id,
                   sl.create_time,
                   sl.start_use_time,
                   sl.last_use_time,
                   sl.income_seconds / 3600.0 AS service_hours,
                   -- CHANGE 2026-03-27 | real_use_seconds 不用于业绩，惩罚在 DWS 层计算
                   -- DWD 层服务记录卡片不显示折前时长，统一返回 NULL
                   NULL AS service_hours_raw,
                   sl.skill_name AS course_type,
                   COALESCE(dt.table_name, '') AS table_name,
                   COALESCE(da.nickname, da.real_name, '') AS assistant_name,
                   -- 到手 = hours × net_rate（与 get_service_records_for_task 同口径）
                   CASE
                       WHEN sl.skill_name ILIKE '%%激励%%' OR sl.skill_name ILIKE '%%超休%%'
                       THEN (sl.income_seconds / 3600.0) * COALESCE(sc.bonus_course_price * (1 - sc.bonus_deduction_ratio), 0)
                       ELSE (sl.income_seconds / 3600.0) * COALESCE(sc.base_course_price - sc.base_deduction, 0)
                   END AS income,
                   gs_agg.drinks
            FROM app.v_dwd_assistant_service_log sl
            LEFT JOIN app.v_dim_assistant da
                ON sl.site_assistant_id = da.assistant_id
                AND da.scd2_is_current = 1
            LEFT JOIN app.v_dim_table dt
                ON sl.site_table_id = dt.table_id
                AND dt.scd2_is_current = 1
            LEFT JOIN app.v_dws_assistant_salary_calc sc
                ON sl.site_assistant_id = sc.assistant_id
                AND date_trunc('month', sl.create_time)::date = sc.salary_month
            LEFT JOIN LATERAL (
                SELECT string_agg(gs.ledger_name || '×' || gs.total_count, '、') AS drinks
                FROM (
                    SELECT ledger_name, SUM(ledger_count) AS total_count
                    FROM app.v_dwd_store_goods_sale
                    WHERE order_settle_id = sl.order_settle_id
                      AND is_delete = 0
                    GROUP BY ledger_name
                ) gs
            ) gs_agg ON true
            WHERE {base_where}
            ORDER BY sl.create_time DESC
            LIMIT %s OFFSET %s
            """,
            params + [limit, offset],
        )
        for row in cur.fetchall():
            svc_hours = float(row[4]) if row[4] is not None else 0.0
            # CHANGE 2026-03-27 | 折前时长：SQL 层已判断，无惩罚时 row[5] 为 NULL
            svc_hours_raw = float(row[5]) if row[5] is not None else None
            records.append({
                "id": str(row[0]) if row[0] else "",
                "create_time": row[1],
                "start_time": row[2],
                "end_time": row[3],
                "service_hours": round(svc_hours, 2),
                "service_hours_raw": round(svc_hours_raw, 2) if svc_hours_raw is not None else None,
                "course_type": row[6] or "",
                "table_name": row[7] or "",
                "assistant_name": row[8] or "",
                "income": float(row[9]) if row[9] is not None else 0.0,
                "drinks": row[10],
            })
    return records, total_count


@trace_service(description_zh="获取助教信息", description_en="Get assistant info")
def get_assistant_info(
    conn: Any, site_id: int, assistant_id: int
) -> dict | None:
    """
    查询助教基本信息。

    来源: app.v_dim_assistant。
    返回 {name, avatar, level, skills, hire_date} 或 None。
    """
    # CHANGE 2026-03-19 | 修正列名：v_dim_assistant 实际列为 real_name/nickname/level(int)/entry_time
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT assistant_id,
                   COALESCE(nickname, real_name, '') AS name,
                   level,
                   entry_time AS hire_date
            FROM app.v_dim_assistant
            WHERE assistant_id = %s AND scd2_is_current = 1
            """,
            (assistant_id,),
        )
        row = cur.fetchone()
    if not row:
        return None
    # CHANGE 2026-03-19 | feiqiu-data-rules 规则 6: 等级名称从配置表动态读取
    level_map = get_level_map(conn, site_id)
    return {
        "id": row[0],
        "name": row[1] or "",
        "level": level_map.get(row[2], "") if row[2] else "",
        "hire_date": row[3].strftime("%Y-%m-%d") if row[3] else None,
        # 视图中无 avatar/skills/work_years，使用默认值
        "avatar": "",
        "skills": [],
        "work_years": 0.0,
    }


@trace_service(description_zh="批量获取多月工资数据", description_en="Get salary calculation for multiple months")
def get_salary_calc_multi_months(
    conn: Any, site_id: int, assistant_id: int, months: list[str]
) -> dict[str, dict]:
    """
    批量查询多个月份的绩效数据。

    来源: app.v_dws_assistant_salary_calc。
    months 格式: ["2026-03-01", "2026-02-01", ...]
    返回 {month_str: {effective_hours, gross_salary, base_income, bonus_income}}。
    """
    if not months:
        return {}

    result: dict[str, dict] = {}
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT salary_month,
                   effective_hours,
                   gross_salary,
                   base_income,
                   bonus_income
            FROM app.v_dws_assistant_salary_calc
            WHERE assistant_id = %s
              AND salary_month = ANY(%s::date[])
            ORDER BY salary_month DESC
            """,
            (assistant_id, months),
        )
        for row in cur.fetchall():
            month_key = str(row[0])
            result[month_key] = {
                "effective_hours": float(row[1]) if row[1] is not None else 0.0,
                "gross_salary": float(row[2]) if row[2] is not None else 0.0,
                "base_income": float(row[3]) if row[3] is not None else 0.0,
                "bonus_income": float(row[4]) if row[4] is not None else 0.0,
            }
    return result


@trace_service(description_zh="获取月度客户数", description_en="Get monthly customer count")
def get_monthly_customer_count(
    conn: Any, site_id: int, assistant_id: int, months: list[str]
) -> dict[str, int]:
    """
    批量查询各月不重复客户数。

    来源: app.v_dwd_assistant_service_log。
    COUNT(DISTINCT tenant_member_id)，过滤 is_delete = 0。
    months 格式: ["2026-03-01", "2026-02-01", ...]
    返回 {month_str: customer_count}。
    """
    if not months:
        return {}

    # 计算整体时间范围
    sorted_months = sorted(months)
    start_date = sorted_months[0]
    # 最后一个月的下个月
    last = sorted_months[-1]
    last_parts = last.split("-")
    y, m = int(last_parts[0]), int(last_parts[1])
    if m == 12:
        end_date = f"{y + 1}-01-01"
    else:
        end_date = f"{y}-{m + 1:02d}-01"

    result: dict[str, int] = {}
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT DATE_TRUNC('month', create_time)::date AS month,
                   COUNT(DISTINCT tenant_member_id) AS customer_count
            FROM app.v_dwd_assistant_service_log
            WHERE site_assistant_id = %s
              AND is_delete = 0
              AND create_time >= %s::timestamptz
              AND create_time < %s::timestamptz
            GROUP BY DATE_TRUNC('month', create_time)::date
            """,
            (assistant_id, start_date, end_date),
        )
        for row in cur.fetchall():
            result[str(row[0])] = row[1]
    return result


@trace_service(description_zh="获取助教TOP客户", description_en="Get coach top customers")
def get_coach_top_customers(
    conn: Any, site_id: int, assistant_id: int, limit: int = 20
) -> list[dict]:
    """
    查询助教 TOP 客户（按服务次数降序）。

    来源: app.v_dwd_assistant_service_log + v_dim_member + v_dim_member_card_account。
    ⚠️ DQ-6: 客户姓名通过 tenant_member_id JOIN v_dim_member。
    ⚠️ DQ-7: 余额通过 tenant_member_id JOIN v_dim_member_card_account。
    ⚠️ DWD-DOC 规则 1: consume 使用 ledger_amount（items_sum 口径）。
    ⚠️ 废单排除: is_delete = 0。
    """
    records: list[dict] = []
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            WITH card_balance AS (
                -- CHANGE 2026-03-29 | 多卡膨胀修复：先按 tenant_member_id 聚合卡余额
                SELECT tenant_member_id, SUM(COALESCE(balance, 0)) AS total_balance
                FROM app.v_dim_member_card_account
                WHERE scd2_is_current = 1
                GROUP BY tenant_member_id
            )
            SELECT sl.tenant_member_id AS member_id,
                   dm.nickname AS customer_name,
                   COUNT(*) AS service_count,
                   COALESCE(SUM(sl.ledger_amount), 0) AS total_consume,
                   COALESCE(cb.total_balance, 0) AS customer_balance
            FROM app.v_dwd_assistant_service_log sl
            LEFT JOIN app.v_dim_member dm
                ON sl.tenant_member_id = dm.member_id AND dm.scd2_is_current = 1
            LEFT JOIN card_balance cb
                ON sl.tenant_member_id = cb.tenant_member_id
            WHERE sl.site_assistant_id = %s
              AND sl.is_delete = 0
              AND sl.tenant_member_id > 0
            GROUP BY sl.tenant_member_id, dm.nickname, cb.total_balance
            ORDER BY service_count DESC
            LIMIT %s
            """,
            (assistant_id, limit),
        )
        for row in cur.fetchall():
            records.append({
                "member_id": row[0],
                "customer_name": row[1] or "",
                "service_count": row[2] or 0,
                "total_consume": float(row[3]) if row[3] is not None else 0.0,
                "customer_balance": float(row[4]) if row[4] is not None else 0.0,
            })
    return records


@trace_service(description_zh="获取助教服务记录", description_en="Get coach service records")
def get_coach_service_records(
    conn: Any,
    site_id: int,
    assistant_id: int,
    limit: int = 20,
    offset: int = 0,
) -> list[dict]:
    """
    查询助教近期服务记录（COACH-1 serviceRecords 用）。

    来源: app.v_dwd_assistant_service_log + v_dim_member。
    ⚠️ DQ-6: 客户姓名通过 tenant_member_id JOIN v_dim_member。
    CHANGE 2026-03-26 | 收入口径: JOIN salary_calc 取到手费率，hours × net_rate 算真正到手
    ⚠️ 废单排除: is_delete = 0。
    """
    records: list[dict] = []
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT sl.assistant_service_id AS id,
                   sl.tenant_member_id AS member_id,
                   dm.nickname AS customer_name,
                   sl.create_time,
                   sl.income_seconds / 3600.0 AS service_hours,
                   -- CHANGE 2026-03-26 | 到手 = hours × net_rate
                   CASE
                       WHEN sl.skill_name ILIKE '%%激励%%' OR sl.skill_name ILIKE '%%超休%%'
                       THEN (sl.income_seconds / 3600.0) * COALESCE(sc.bonus_course_price * (1 - sc.bonus_deduction_ratio), 0)
                       ELSE (sl.income_seconds / 3600.0) * COALESCE(sc.base_course_price - sc.base_deduction, 0)
                   END AS income,
                   sl.skill_name AS course_type,
                   sl.site_table_id AS table_id,
                   dt.table_name AS table_name
            FROM app.v_dwd_assistant_service_log sl
            LEFT JOIN app.v_dim_member dm
                ON sl.tenant_member_id = dm.member_id
                AND dm.scd2_is_current = 1
            LEFT JOIN app.v_dws_assistant_salary_calc sc
                ON sl.site_assistant_id = sc.assistant_id
                AND date_trunc('month', sl.create_time)::date = sc.salary_month
            LEFT JOIN app.v_dim_table dt
                ON sl.site_table_id = dt.table_id
                AND dt.scd2_is_current = 1
            WHERE sl.site_assistant_id = %s
              AND sl.is_delete = 0
            ORDER BY sl.create_time DESC
            LIMIT %s OFFSET %s
            """,
            (assistant_id, limit, offset),
        )
        for row in cur.fetchall():
            records.append({
                "id": row[0],
                "member_id": row[1],
                "customer_name": row[2] or "",
                "create_time": row[3],
                "service_hours": float(row[4]) if row[4] is not None else 0.0,
                "income": float(row[5]) if row[5] is not None else 0.0,
                "course_type": row[6] or "",
                "table_id": row[7],
                "table_name": row[8] or "",
            })
    return records

# ---------------------------------------------------------------------------
# BOARD-1 助教看板 FDW 查询
# ---------------------------------------------------------------------------


@trace_service(description_zh="获取所有助教列表", description_en="Get all assistants")
def get_all_assistants(
    conn: Any, site_id: int, skill_filter: str = "ALL"
) -> list[dict]:
    """
    查询门店全部助教列表（BOARD-1 用）。

    CHANGE 2026-03-20 | R3 修复：skill_filter 直接接收 category_code
    （BILLIARD/SNOOKER/MAHJONG/KTV/ALL），去掉 chinese→BILLIARD 映射层。
    """
    # CHANGE 2026-03-19 | feiqiu-data-rules 规则 6: 等级名称从配置表动态读取
    # CHANGE 2026-03-20 | R3 修复：去掉 _skill_to_category 映射，直接用 category_code
    _valid_categories = {"BILLIARD", "SNOOKER", "MAHJONG", "KTV"}

    level_map = get_level_map(conn, site_id)
    records: list[dict] = []
    with _fdw_context(conn, site_id) as cur:
        # 筛选条件：如果指定了技能，只返回被标记的助教
        filter_clause = ""
        params: tuple = ()
        if skill_filter != "ALL" and skill_filter in _valid_categories:
            filter_clause = """
                AND da.assistant_id IN (
                    SELECT apt.assistant_id
                    FROM app.v_dws_assistant_project_tag apt
                    WHERE apt.category_code = %s AND apt.is_tagged = true
                )
            """
            params = (skill_filter,)

        cur.execute(
            f"""
            SELECT da.assistant_id,
                   COALESCE(da.nickname, da.real_name, '') AS name,
                   da.level,
                   array_agg(DISTINCT apt.category_code) FILTER (WHERE apt.is_tagged = true) AS skills
            FROM app.v_dim_assistant da
            LEFT JOIN app.v_dws_assistant_project_tag apt
                ON da.assistant_id = apt.assistant_id
            WHERE da.scd2_is_current = 1
              AND da.leave_status = 0
            {filter_clause}
            GROUP BY da.assistant_id, da.real_name, da.nickname, da.level
            ORDER BY da.assistant_id
            """,
            params,
        )
        for row in cur.fetchall():
            skill_codes = row[3] if row[3] else []
            # CHANGE 2026-03-20 | R3 修复：直接返回 category_code，不再反向映射为旧值
            records.append({
                "assistant_id": row[0],
                "name": row[1] or "",
                "skill": ",".join(c for c in skill_codes if c) if skill_codes else "",
                "level": level_map.get(row[2], "") if row[2] else "",
            })
    return records


@trace_service(description_zh="批量获取工资数据", description_en="Get salary calculation batch")
def get_salary_calc_batch(
    conn: Any,
    site_id: int,
    assistant_ids: list[int],
    start_date: str,
    end_date: str,
) -> dict[int, dict]:
    """
    批量查询助教绩效数据（BOARD-1 perf/salary 维度用）。

    来源: app.v_dws_assistant_salary_calc。
    按 assistant_id 分组，对日期范围内的月份数据聚合。

    ⚠️ DWD-DOC 规则 1: 收入使用 items_sum 口径（gross_salary 对应 items_sum）
    ⚠️ DWD-DOC 规则 2: 费用使用 base_income (assistant_pd_money) + bonus_income (assistant_cx_money)

    返回 {assistant_id: {effective_hours, gross_salary, base_income, bonus_income}}。
    """
    if not assistant_ids:
        return {}

    result: dict[int, dict] = {}
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT assistant_id,
                   SUM(effective_hours) AS effective_hours,
                   SUM(gross_salary) AS gross_salary,
                   SUM(base_income) AS base_income,
                   SUM(bonus_income) AS bonus_income,
                   MAX(assistant_level_name) AS level_name,
                   SUM(base_hours + bonus_hours + room_hours) AS raw_hours,
                   BOOL_OR(is_new_hire) AS is_new_hire
            FROM app.v_dws_assistant_salary_calc
            WHERE assistant_id = ANY(%s)
              AND salary_month >= %s::date
              AND salary_month <= %s::date
            GROUP BY assistant_id
            """,
            (assistant_ids, start_date, end_date),
        )
        for row in cur.fetchall():
            result[row[0]] = {
                "effective_hours": float(row[1]) if row[1] is not None else 0.0,
                "gross_salary": float(row[2]) if row[2] is not None else 0.0,
                "base_income": float(row[3]) if row[3] is not None else 0.0,
                "bonus_income": float(row[4]) if row[4] is not None else 0.0,
                "level_name": row[5] or "",
                "raw_hours": float(row[6]) if row[6] is not None else 0.0,
                "is_new_hire": bool(row[7]) if row[7] is not None else False,
            }
    return result


@trace_service(description_zh="批量获取助教TOP客户", description_en="Get top customers for coaches")
def get_top_customers_for_coaches(
    conn: Any, site_id: int, assistant_ids: list[int]
) -> dict[int, list[str]]:
    """
    批量查询助教 Top 3 客户（按亲密度降序，BOARD-1 用）。

    来源: app.v_dws_member_assistant_relation_index + app.v_dim_member。
    ⚠️ DQ-6: 客户姓名通过 member_id JOIN v_dim_member，取 scd2_is_current=1。
    ⚠️ 亲密度 emoji 四级映射: > 8.5 → 💖, > 7 → 🧡, > 5 → 💛, ≤ 5 → 💙。

    返回 {assistant_id: ["💖 王先生", "💛 李女士", ...]}，每个助教最多 3 个。
    """
    if not assistant_ids:
        return {}

    result: dict[int, list[str]] = {aid: [] for aid in assistant_ids}
    with _fdw_context(conn, site_id) as cur:
        # 使用窗口函数取每个助教的 Top 3
        cur.execute(
            """
            WITH ranked AS (
                SELECT ri.assistant_id,
                       ri.rs_display,
                       dm.nickname,
                       ROW_NUMBER() OVER (
                           PARTITION BY ri.assistant_id
                           ORDER BY ri.rs_display DESC
                       ) AS rn
                FROM app.v_dws_member_assistant_relation_index ri
                LEFT JOIN app.v_dim_member dm
                    ON ri.member_id = dm.member_id
                    AND dm.scd2_is_current = 1
                WHERE ri.assistant_id = ANY(%s)
            )
            SELECT assistant_id, rs_display, nickname
            FROM ranked
            WHERE rn <= 3
            ORDER BY assistant_id, rn
            """,
            (assistant_ids,),
        )
        for row in cur.fetchall():
            aid = row[0]
            rs = float(row[1]) if row[1] is not None else 0.0
            name = row[2] or "未知"
            emoji = _rs_emoji(rs)
            result[aid].append(f"{emoji} {name}")
    return result


def _rs_emoji(rs_display: float) -> str:
    """亲密度 emoji 四级映射。"""
    if rs_display > 8.5:
        return "💖"
    if rs_display > 7:
        return "🧡"
    if rs_display > 5:
        return "💛"
    return "💙"


@trace_service(description_zh="获取助教储值数据", description_en="Get coach stored value data")
def get_coach_sv_data(
    conn: Any,
    site_id: int,
    assistant_ids: list[int],
    start_date: str,
    end_date: str,
) -> dict[int, dict]:
    """
    批量查询助教储值维度数据（BOARD-1 sv 维度用）。

    CHANGE 2026-03-29 | 修正数据源：客源储值 = 该助教关联客户的卡余额合计。
    从 relation_index 获取助教-客户关联，JOIN card_account 获取余额。
    sv_amount = 关联客户余额合计
    sv_customer_count = 关联客户数（有余额的）
    sv_consume = 关联客户 60 天消费合计

    返回 {assistant_id: {sv_amount, sv_customer_count, sv_consume}}。
    """
    if not assistant_ids:
        return {}

    result: dict[int, dict] = {}
    with _fdw_context(conn, site_id) as cur:
        # CHANGE 2026-04-07 | Fix-6：sv_consume 改为从结算表按 start_date/end_date 过滤，
        # 使其随时间筛选联动，而非固定 60 天窗口。
        cur.execute(
            """
            WITH coach_members AS (
                SELECT ri.assistant_id,
                       ri.member_id
                FROM app.v_dws_member_assistant_relation_index ri
                WHERE ri.assistant_id = ANY(%s)
                  AND ri.session_count > 0
            ),
            period_consume AS (
                -- DWD-DOC 规则 1: items_sum 需拆分计算，settlement_head 无此字段
                SELECT sh.member_id,
                       COALESCE(SUM(
                           sh.table_charge_money + sh.goods_money
                           + sh.assistant_pd_money + sh.assistant_cx_money
                           + sh.electricity_money
                       ), 0) AS consume_amount
                FROM app.v_dwd_settlement_head sh
                WHERE sh.member_id = ANY(SELECT member_id FROM coach_members)
                  AND sh.settle_type IN (1, 3)
                  AND sh.pay_time::date >= %s::date
                  AND sh.pay_time::date <= %s::date
                GROUP BY sh.member_id
            )
            SELECT cm.assistant_id,
                   COALESCE(SUM(ca_agg.balance), 0) AS sv_amount,
                   COUNT(DISTINCT CASE WHEN ca_agg.balance > 0 THEN cm.member_id END) AS sv_customer_count,
                   COALESCE(SUM(pc.consume_amount), 0) AS sv_consume
            FROM coach_members cm
            LEFT JOIN (
                SELECT tenant_member_id, SUM(balance) AS balance
                FROM app.v_dim_member_card_account
                WHERE scd2_is_current = 1
                GROUP BY tenant_member_id
            ) ca_agg ON cm.member_id = ca_agg.tenant_member_id
            LEFT JOIN period_consume pc
                ON cm.member_id = pc.member_id
            GROUP BY cm.assistant_id
            """,
            (assistant_ids, start_date, end_date),
        )
        for row in cur.fetchall():
            result[row[0]] = {
                "sv_amount": float(row[1]),
                "sv_customer_count": int(row[2]),
                "sv_consume": float(row[3]),
            }
    return result


# ---------------------------------------------------------------------------
# BOARD-2 客户看板 FDW 查询（8 维度 + 批量助教查询）
# ---------------------------------------------------------------------------


def _project_filter_clause(project: str, member_col: str = "member_id") -> tuple[str, tuple]:
    """
    生成项目筛选 SQL 片段（用于 BOARD-2 会员维度查询）。

    CHANGE 2026-03-20 | R3 修复：project 参数直接接收 category_code
    （BILLIARD/SNOOKER/MAHJONG/KTV/ALL），去掉 chinese→BILLIARD 映射层。
    CHANGE 2026-04-07 | Fix-1：member_col 参数化，修复 6 个维度别名不匹配导致 SQL 500。
    返回 (sql_fragment, params)，sql_fragment 以 AND 开头，可直接拼入 WHERE 子句。
    """
    _valid_categories = {"BILLIARD", "SNOOKER", "MAHJONG", "KTV"}
    if project == "ALL" or project not in _valid_categories:
        return "", ()
    clause = f"""
        AND {member_col} IN (
            SELECT mpt.member_id
            FROM app.v_dws_member_project_tag mpt
            WHERE mpt.category_code = %s AND mpt.is_tagged = true
        )
    """
    return clause, (project,)


@trace_service(description_zh="获取客户看板-召回维度", description_en="Get customer board recall")
def get_customer_board_recall(
    conn: Any, site_id: int, project: str, page: int, page_size: int
) -> dict:
    """
    BOARD-2 recall 维度：召回指数排行。

    来源: app.v_dws_member_winback_index + app.v_dim_member。
    CHANGE 2026-03-20 | 修正列名映射：
      ideal_days → ideal_interval_days, wbi_score → display_score,
      elapsed_days → CURRENT_DATE - last_visit_time::date (计算列),
      overdue_days → elapsed_days - ideal_interval_days (计算列),
      CHANGE 2026-04-07 | Fix-3：visits_30d 新增字段，替代 visits_14d 近似,
      balance_amount → balance (v_dim_member_card_account)
    ⚠️ DQ-6: 客户姓名通过 member_id JOIN v_dim_member。
    ⚠️ DQ-7: 余额通过 JOIN v_dim_member_card_account。
    按 display_score 降序，LIMIT/OFFSET 分页。
    """
    proj_clause, proj_params = _project_filter_clause(project, "wi.member_id")

    with _fdw_context(conn, site_id) as cur:
        # 总数
        cur.execute(
            f"""
            SELECT COUNT(*)
            FROM app.v_dws_member_winback_index wi
            WHERE 1=1 {proj_clause}

            """,
            proj_params,
        )
        total = cur.fetchone()[0]

        # 分页数据
        # CHANGE 2026-05-02 | elapsed_days/overdue_days 用 business_date 替代 CURRENT_DATE
        from app.services.runtime_context import as_runtime_today_param as _rt_today
        ref_date = _rt_today(site_id, conn=conn)
        offset = (page - 1) * page_size
        cur.execute(
            f"""
            SELECT wi.member_id,
                   dm.nickname,
                   wi.ideal_interval_days,
                   %s::date - wi.last_visit_time::date AS elapsed_days,
                   (%s::date - wi.last_visit_time::date) - COALESCE(wi.ideal_interval_days, 0) AS overdue_days,
                   wi.visits_30d,
                   wi.display_score,
                   COALESCE(ca.balance, 0) AS balance
            FROM app.v_dws_member_winback_index wi
            LEFT JOIN app.v_dim_member dm
                ON wi.member_id = dm.member_id AND dm.scd2_is_current = 1
            LEFT JOIN (
                SELECT tenant_member_id, SUM(balance) AS balance
                FROM app.v_dim_member_card_account
                WHERE scd2_is_current = 1
                GROUP BY tenant_member_id
            ) ca ON wi.member_id = ca.tenant_member_id
            WHERE wi.last_visit_time <= %s::date + INTERVAL '1 day' {proj_clause}
            ORDER BY wi.display_score DESC, wi.member_id
            LIMIT %s OFFSET %s
            """,
            (ref_date, ref_date, ref_date, *proj_params, page_size, offset),
        )
        items = []
        for row in cur.fetchall():
            items.append({
                "member_id": row[0],
                "name": row[1] or "",
                # CHANGE 2026-03-29 | 天数字段转整数，前端显示不带小数
                "ideal_days": int(row[2]) if row[2] is not None else 0,
                "elapsed_days": int(row[3]) if row[3] is not None else 0,
                "overdue_days": int(row[4]) if row[4] is not None else 0,
                "visits_30d": int(row[5]) if row[5] is not None else 0,
                "recall_index": float(row[6]) if row[6] is not None else 0.0,
                "balance": float(row[7]) if row[7] is not None else 0.0,
            })

    return {"items": items, "total": total, "page": page, "page_size": page_size}


def _derive_potential_tags(
    level_score: float | None,
    speed_score: float | None,
    stability_score: float | None,
) -> list[dict]:
    """从三维分数派生潜力标签（display_score 为 0-10 区间）。
    返回 [{text, theme}] 格式，与前端 potentialTags 类型一致。
    """
    tags: list[dict] = []
    threshold = 6.0
    if level_score is not None and float(level_score) >= threshold:
        tags.append({"text": "高消费力", "theme": "success"})
    if speed_score is not None and float(speed_score) >= threshold:
        tags.append({"text": "快增长", "theme": "warning"})
    if stability_score is not None and float(stability_score) >= threshold:
        tags.append({"text": "稳定", "theme": "primary"})
    return tags


@trace_service(description_zh="获取客户看板-潜力维度", description_en="Get customer board potential")
def get_customer_board_potential(
    conn: Any, site_id: int, project: str, page: int, page_size: int
) -> dict:
    """
    BOARD-2 potential 维度：消费潜力指数排行。

    CHANGE 2026-03-19 | P0 修复：v_dws_member_spending_power_index 已通过 FDW 可用，
    替换空列表降级为实际查询。按 display_score 降序排列。
    会员信息通过 member_id JOIN v_dim_member（DQ-6 规则）。
    项目筛选通过 v_dws_member_project_tag 子查询。
    """
    proj_clause, proj_params = _project_filter_clause(project)

    with _fdw_context(conn, site_id) as cur:
        # 计数（带项目筛选）
        cur.execute(
            f"""
            SELECT COUNT(*)
            FROM app.v_dws_member_spending_power_index spi
            WHERE 1=1
            {f"AND spi.member_id IN (SELECT member_id FROM app.v_dws_member_project_tag WHERE category_code = %s AND is_tagged = true)" if proj_params else ""}
            """,
            proj_params,
        )
        total = cur.fetchone()[0]

        offset = (page - 1) * page_size
        cur.execute(
            f"""
            SELECT spi.member_id,
                   dm.nickname,
                   spi.spend_30,
                   spi.visit_days_30,
                   spi.avg_ticket_90,
                   spi.display_score,
                   spi.score_level_display,
                   spi.score_speed_display,
                   spi.score_stability_display,
                   COALESCE(ca_agg.balance, 0) AS balance
            FROM app.v_dws_member_spending_power_index spi
            LEFT JOIN app.v_dim_member dm
                ON spi.member_id = dm.member_id AND dm.scd2_is_current = 1
            LEFT JOIN (
                SELECT tenant_member_id, SUM(balance) AS balance
                FROM app.v_dim_member_card_account
                WHERE scd2_is_current = 1
                GROUP BY tenant_member_id
            ) ca_agg ON spi.member_id = ca_agg.tenant_member_id
            WHERE 1=1
            {f"AND spi.member_id IN (SELECT member_id FROM app.v_dws_member_project_tag WHERE category_code = %s AND is_tagged = true)" if proj_params else ""}
            ORDER BY spi.display_score DESC NULLS LAST, COALESCE(ca_agg.balance, 0) DESC, spi.member_id
            LIMIT %s OFFSET %s
            """,
            (*proj_params, page_size, offset),
        )
        items = []
        for row in cur.fetchall():
            items.append({
                "member_id": row[0],
                "name": row[1] or "",
                "spend_30d": float(row[2]) if row[2] is not None else 0.0,
                "avg_visits": int(row[3]) if row[3] is not None else 0,
                "avg_spend": float(row[4]) if row[4] is not None else 0.0,
                "potential_score": float(row[5]) if row[5] is not None else 0.0,
                "potential_tags": _derive_potential_tags(row[6], row[7], row[8]),
                "balance": float(row[9]) if row[9] is not None else 0.0,
            })

    return {"items": items, "total": total, "page": page, "page_size": page_size}


@trace_service(description_zh="获取客户看板-余额维度", description_en="Get customer board balance")
def get_customer_board_balance(
    conn: Any, site_id: int, project: str, page: int, page_size: int
) -> dict:
    """
    BOARD-2 balance 维度：余额排行。

    来源: app.v_dim_member_card_account + app.v_dim_member。
    CHANGE 2026-03-20 | 修正列名：balance_amount → balance,
    last_visit_date/monthly_consume 从 v_dws_member_consumption_summary 获取
    （实际列为 days_since_last, consume_amount_60d）。
    ⚠️ DQ-7: 余额通过 tenant_member_id JOIN，取 scd2_is_current=1。
    按 balance 降序。
    """
    proj_clause, proj_params = _project_filter_clause(project, "ca.tenant_member_id")

    with _fdw_context(conn, site_id) as cur:
        # CHANGE 2026-03-28 | 修复客户重复：dim_member_card_account 同一 member 有多条记录，
        # 先聚合余额再 JOIN，避免行膨胀。排除散客（member_id <= 0）。
        cur.execute(
            f"""
            SELECT COUNT(*)
            FROM (
                SELECT ca.tenant_member_id
                FROM app.v_dim_member_card_account ca
                WHERE ca.scd2_is_current = 1
                  AND ca.tenant_member_id > 0
                {proj_clause}
                GROUP BY ca.tenant_member_id
            ) sub
            """,
            proj_params,
        )
        total = cur.fetchone()[0]

        offset = (page - 1) * page_size
        cur.execute(
            f"""
            SELECT ca_agg.member_id,
                   dm.nickname,
                   ca_agg.balance,
                   vd.days_since_last,
                   vd.consume_amount_60d
            FROM (
                SELECT ca.tenant_member_id AS member_id,
                       SUM(ca.balance) AS balance
                FROM app.v_dim_member_card_account ca
                WHERE ca.scd2_is_current = 1
                  AND ca.tenant_member_id > 0
                {proj_clause}
                GROUP BY ca.tenant_member_id
            ) ca_agg
            LEFT JOIN app.v_dim_member dm
                ON ca_agg.member_id = dm.member_id AND dm.scd2_is_current = 1
            LEFT JOIN LATERAL (
                SELECT cs.days_since_last, cs.consume_amount_60d
                FROM app.v_dws_member_consumption_summary cs
                WHERE cs.member_id = ca_agg.member_id
                ORDER BY cs.stat_date DESC
                LIMIT 1
            ) vd ON true
            ORDER BY ca_agg.balance DESC, ca_agg.member_id
            LIMIT %s OFFSET %s
            """,
            (*proj_params, page_size, offset),
        )
        items = []
        for row in cur.fetchall():
            items.append({
                "member_id": row[0],
                "name": row[1] or "",
                "balance": float(row[2]) if row[2] is not None else 0.0,
                # CHANGE 2026-03-29 | last_visit 格式化为"X天前"，ideal_days 从 winback_index 获取
                "last_visit": "今天" if row[3] == 0 else f"{row[3]}天前" if row[3] is not None else "--",
                "last_visit_date": row[3],
                "ideal_days": None,  # balance 维度无 ideal_days，由 board_service 补充
                # CHANGE 2026-04-07 | Fix-4：consume_amount_60d 是 60 天总额，月均 = /2
                "monthly_consume": float(row[4]) / 2 if row[4] is not None else 0.0,
                "available_months": (
                    f"{2 * float(row[2]) / float(row[4]):.1f}个月"
                    if row[2] and row[4] and float(row[4]) > 0
                    else "--"
                ),
            })

    return {"items": items, "total": total, "page": page, "page_size": page_size}


@trace_service(description_zh="获取客户看板-充值维度", description_en="Get customer board recharge")
def get_customer_board_recharge(
    conn: Any, site_id: int, project: str, page: int, page_size: int
) -> dict:
    """
    BOARD-2 recharge 维度：充值记录排行。

    来源: app.v_dwd_recharge_order + app.v_dim_member_card_account。
    CHANGE 2026-03-20 | 修正列名：
      recharge_date → pay_time, recharge_amount → pay_amount,
      balance_amount → balance (v_dim_member_card_account)
    按 last_recharge_date (MAX(pay_time::date)) 降序。
    """
    proj_clause, proj_params = _project_filter_clause(project, "ro.member_id")

    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            f"""
            SELECT COUNT(DISTINCT ro.member_id)
            FROM app.v_dwd_recharge_order ro
            WHERE 1=1 {proj_clause}
            """,
            proj_params,
        )
        total = cur.fetchone()[0]

        # CHANGE 2026-05-02 | 60 天充值窗口、stat_date、pay_time 全部按 business_date 截断
        from app.services.runtime_context import as_runtime_today_param
        ref_date = as_runtime_today_param(site_id, conn=conn)

        offset = (page - 1) * page_size
        cur.execute(
            f"""
            SELECT ro.member_id,
                   dm.nickname,
                   MAX(ro.pay_time::date) AS last_recharge_date,
                   SUM(ro.pay_amount) AS recharge_amount,
                   COUNT(*) FILTER (
                       WHERE ro.pay_time >= %s::date - INTERVAL '60 days'
                         AND ro.pay_time <  %s::date + INTERVAL '1 day'
                   ) AS recharges_60d,
                   COALESCE(ca_agg.balance, 0) AS current_balance,
                   cs.days_since_last
            FROM app.v_dwd_recharge_order ro
            LEFT JOIN app.v_dim_member dm
                ON ro.member_id = dm.member_id AND dm.scd2_is_current = 1
            LEFT JOIN (
                SELECT tenant_member_id, SUM(balance) AS balance
                FROM app.v_dim_member_card_account
                WHERE scd2_is_current = 1
                GROUP BY tenant_member_id
            ) ca_agg ON ro.member_id = ca_agg.tenant_member_id
            LEFT JOIN LATERAL (
                SELECT cs2.days_since_last
                FROM app.v_dws_member_consumption_summary cs2
                WHERE cs2.member_id = ro.member_id
                  AND cs2.stat_date <= %s
                ORDER BY cs2.stat_date DESC
                LIMIT 1
            ) cs ON true
            WHERE ro.pay_time <= %s::date + INTERVAL '1 day' {proj_clause}
            GROUP BY ro.member_id, dm.nickname, ca_agg.balance, cs.days_since_last
            ORDER BY MAX(ro.pay_time::date) DESC, ro.member_id
            LIMIT %s OFFSET %s
            """,
            (ref_date, ref_date, ref_date, ref_date, *proj_params, page_size, offset),
        )
        items = []
        for row in cur.fetchall():
            items.append({
                "member_id": row[0],
                "name": row[1] or "",
                "last_recharge": str(row[2]) if row[2] else "",
                "recharge_amount": float(row[3]) if row[3] is not None else 0.0,
                "recharges_60d": row[4] or 0,
                "current_balance": float(row[5]) if row[5] is not None else 0.0,
                # CHANGE 2026-03-29 | 补充 last_visit 和 ideal_days（头部展示用）
                "last_visit": "今天" if row[6] == 0 else f"{row[6]}天前" if row[6] is not None else "--",
                "ideal_days": None,  # 由 board_service 补充
            })

    return {"items": items, "total": total, "page": page, "page_size": page_size}


@trace_service(description_zh="获取客户看板-近期维度", description_en="Get customer board recent")
def get_customer_board_recent(
    conn: Any, site_id: int, project: str, page: int, page_size: int
) -> dict:
    """
    BOARD-2 recent 维度：最近到店排行。

    CHANGE 2026-03-19 | P2 修复：ideal_days 从 v_dws_member_winback_index.ideal_interval_days 获取，
    不再硬编码为 0。来源: v_dws_member_visit_detail + v_dim_member + v_dws_member_winback_index。
    按 last_visit_date 降序。
    """
    # CHANGE 2026-05-02 | 客户看板「最近到店」修复：
    # 1) WHERE/COUNT 中的 30/60 天窗口按 business_date 计算，沙箱不读「未来」到店；
    # 2) days_ago 用 business_date - last_visit_date，与窗口对齐。
    from app.services.runtime_context import as_runtime_today_param

    ref_date = as_runtime_today_param(site_id, conn=conn)

    proj_clause, proj_params = _project_filter_clause(project, "vd.member_id")

    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            f"""
            SELECT COUNT(DISTINCT vd.member_id)
            FROM app.v_dws_member_visit_detail vd
            WHERE vd.visit_date <= %s {proj_clause}
            """,
            (ref_date, *proj_params),
        )
        total = cur.fetchone()[0]

        offset = (page - 1) * page_size
        cur.execute(
            f"""
            WITH member_agg AS (
                SELECT vd.member_id,
                       MAX(vd.visit_date) AS last_visit_date,
                       COUNT(*) AS total_visits,
                       COUNT(*) FILTER (WHERE vd.visit_date >= %s::date - INTERVAL '30 days') AS visits_30d,
                       COUNT(*) FILTER (WHERE vd.visit_date >= %s::date - INTERVAL '60 days') AS visits_60d,
                       AVG(vd.total_consume) AS avg_spend
                FROM app.v_dws_member_visit_detail vd
                WHERE vd.visit_date <= %s {proj_clause}
                GROUP BY vd.member_id
            )
            SELECT ma.member_id,
                   dm.nickname,
                   ma.last_visit_date,
                   ma.total_visits,
                   COALESCE(wi.ideal_interval_days, 0) AS ideal_days,
                   ma.visits_30d,
                   ma.visits_60d,
                   ma.avg_spend
            FROM member_agg ma
            LEFT JOIN app.v_dim_member dm
                ON ma.member_id = dm.member_id AND dm.scd2_is_current = 1
            LEFT JOIN app.v_dws_member_winback_index wi
                ON ma.member_id = wi.member_id
            ORDER BY ma.last_visit_date DESC, ma.member_id
            LIMIT %s OFFSET %s
            """,
            (ref_date, ref_date, ref_date, *proj_params, page_size, offset),
        )
        items = []
        for row in cur.fetchall():
            last_visit = row[2]
            # CHANGE 2026-05-02 | days_ago 按 business_date 计算，沙箱与窗口对齐
            days_ago = (ref_date - last_visit).days if last_visit else None
            items.append({
                "member_id": row[0],
                "name": row[1] or "",
                "last_visit_date": str(last_visit) if last_visit else "",
                "days_ago": days_ago,
                "visit_freq": float(row[3]) if row[3] is not None else 0.0,
                "ideal_days": int(row[4]) if row[4] is not None else 0,
                "visits_30d": row[5] or 0,
                "visits_60d": row[6] or 0,
                "avg_spend": float(row[7]) if row[7] is not None else 0.0,
            })

    return {"items": items, "total": total, "page": page, "page_size": page_size}


@trace_service(description_zh="获取客户看板-60天消费维度", description_en="Get customer board spend 60 days")
def get_customer_board_spend60(
    conn: Any, site_id: int, project: str, page: int, page_size: int
) -> dict:
    """
    BOARD-2 spend60 维度：60 天消费额排行。

    来源: app.v_dws_member_consumption_summary。
    CHANGE 2026-03-20 | 修正列名：items_sum_60d → consume_amount_60d,
    high_spend_tag/avg_spend 不存在，用 avg_ticket_amount 替代 avg_spend，
    high_spend_tag 通过阈值计算。
    按 consume_amount_60d 降序。
    CHANGE 2026-04-08 | Fix：consumption_summary 按 stat_date 有多行快照，
    用 DISTINCT ON 取最新快照避免同一客户出现多次。
    """
    proj_clause, proj_params = _project_filter_clause(project, "cs.member_id")

    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            f"""
            SELECT COUNT(*)
            FROM (
                SELECT DISTINCT ON (cs.member_id) cs.member_id
                FROM app.v_dws_member_consumption_summary cs
                WHERE 1=1 {proj_clause}
                ORDER BY cs.member_id, cs.stat_date DESC
            ) sub
            """,
            proj_params,
        )
        total = cur.fetchone()[0]

        offset = (page - 1) * page_size
        cur.execute(
            f"""
            WITH latest_cs AS (
                SELECT DISTINCT ON (cs.member_id)
                       cs.member_id, cs.consume_amount_60d,
                       cs.visit_count_60d, cs.avg_ticket_amount
                FROM app.v_dws_member_consumption_summary cs
                WHERE 1=1 {proj_clause}
                ORDER BY cs.member_id, cs.stat_date DESC
            )
            SELECT cs.member_id,
                   dm.nickname,
                   cs.consume_amount_60d,
                   cs.visit_count_60d,
                   cs.avg_ticket_amount
            FROM latest_cs cs
            LEFT JOIN app.v_dim_member dm
                ON cs.member_id = dm.member_id AND dm.scd2_is_current = 1
            ORDER BY cs.consume_amount_60d DESC, cs.member_id
            LIMIT %s OFFSET %s
            """,
            (*proj_params, page_size, offset),
        )
        items = []
        for row in cur.fetchall():
            spend = float(row[2]) if row[2] is not None else 0.0
            items.append({
                "member_id": row[0],
                "name": row[1] or "",
                "spend_60d": spend,
                "visits_60d": row[3] or 0,
                "high_spend_tag": spend > 5000,  # 视图无此列，按阈值计算
                "avg_spend": float(row[4]) if row[4] is not None else 0.0,
            })

    return {"items": items, "total": total, "page": page, "page_size": page_size}


@trace_service(description_zh="获取客户看板-60天频次维度", description_en="Get customer board frequency 60 days")
def get_customer_board_freq60(
    conn: Any, site_id: int, project: str, page: int, page_size: int
) -> dict:
    """
    BOARD-2 freq60 维度：60 天到店频次排行。

    来源: app.v_dws_member_consumption_summary（汇总）+ app.v_dwd_assistant_service_log（周粒度）。
    CHANGE 2026-03-20 | 修正列名：items_sum_60d → consume_amount_60d,
    avg_interval_days 不存在，用 60/visit_count_60d 近似计算。
    按 visit_count_60d 降序。
    CHANGE 2026-04-08 | Fix：同 spend60，DISTINCT ON 取最新快照。
    """
    # CHANGE 2026-05-02 | freq60 全链路按 business_date 截断（stat_date <= ref_date + 8 周窗口）
    from app.services.runtime_context import as_runtime_today_param
    ref_date = as_runtime_today_param(site_id, conn=conn)

    proj_clause, proj_params = _project_filter_clause(project, "cs.member_id")

    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            f"""
            SELECT COUNT(*)
            FROM (
                SELECT DISTINCT ON (cs.member_id) cs.member_id
                FROM app.v_dws_member_consumption_summary cs
                WHERE cs.stat_date <= %s {proj_clause}
                ORDER BY cs.member_id, cs.stat_date DESC
            ) sub
            """,
            (ref_date, *proj_params),
        )
        total = cur.fetchone()[0]

        offset = (page - 1) * page_size
        cur.execute(
            f"""
            WITH latest_cs AS (
                SELECT DISTINCT ON (cs.member_id)
                       cs.member_id, cs.visit_count_60d, cs.consume_amount_60d
                FROM app.v_dws_member_consumption_summary cs
                WHERE cs.stat_date <= %s {proj_clause}
                ORDER BY cs.member_id, cs.stat_date DESC
            )
            SELECT cs.member_id,
                   dm.nickname,
                   cs.visit_count_60d,
                   cs.consume_amount_60d
            FROM latest_cs cs
            LEFT JOIN app.v_dim_member dm
                ON cs.member_id = dm.member_id AND dm.scd2_is_current = 1
            ORDER BY cs.visit_count_60d DESC, cs.member_id
            LIMIT %s OFFSET %s
            """,
            (ref_date, *proj_params, page_size, offset),
        )
        items = []
        member_ids = []
        for row in cur.fetchall():
            mid = row[0]
            member_ids.append(mid)
            visits = row[2] or 0
            # avg_interval_days 不存在于视图，近似计算
            avg_interval = round(60.0 / visits, 1) if visits > 0 else 0.0
            items.append({
                "member_id": mid,
                "name": row[1] or "",
                "visits_60d": visits,
                "avg_interval": avg_interval,
                "spend_60d": float(row[3]) if row[3] is not None else 0.0,
                "weekly_visits": [],  # 后续填充
            })

        # 批量查询 8 周到店数据
        if member_ids:
            weekly_map = _get_weekly_visits_batch(cur, member_ids, ref_date=ref_date)
            for item in items:
                item["weekly_visits"] = weekly_map.get(item["member_id"], _empty_weekly())

    return {"items": items, "total": total, "page": page, "page_size": page_size}


def _get_weekly_visits_batch(
    cur: Any, member_ids: list[int], *, ref_date: Any = None,
) -> dict[int, list[dict]]:
    """
    批量查询客户最近 8 周的到店次数（用于 freq60 维度柱状图）。

    CHANGE 2026-04-07 | Fix-5：数据源从 v_dwd_assistant_service_log 改为
    v_dwd_settlement_head（settle_type IN (1,3)），与汇总维度口径一致。
    CHANGE 2026-05-02 | 8 周窗口锚定 ref_date（业务日），沙箱不读「未来」。
    返回 {member_id: [{val: int, pct: int}, ...]}，固定 8 个元素。
    """
    from datetime import date as _date, timedelta as _timedelta

    if ref_date is None:
        ref_date = _date.today()
    elif hasattr(ref_date, "date") and not isinstance(ref_date, _date):
        ref_date = ref_date.date()

    cur.execute(
        """
        WITH weekly AS (
            SELECT member_id,
                   DATE_TRUNC('week', pay_time::date) AS week_start,
                   COUNT(DISTINCT pay_time::date) AS cnt
            FROM app.v_dwd_settlement_head
            WHERE member_id = ANY(%s)
              AND settle_type IN (1, 3)
              AND pay_time >= %s::date - INTERVAL '56 days'
              AND pay_time <  %s::date + INTERVAL '1 day'
            GROUP BY member_id, DATE_TRUNC('week', pay_time::date)
        )
        SELECT member_id, week_start, cnt
        FROM weekly
        ORDER BY member_id, week_start
        """,
        (member_ids, ref_date, ref_date),
    )

    from collections import defaultdict
    raw: dict[int, dict[str, int]] = defaultdict(dict)
    for row in cur.fetchall():
        # CHANGE 2026-03-29 | DATE_TRUNC 返回 timestamp，转为 date 字符串匹配
        week_key = row[1].date() if hasattr(row[1], 'date') else row[1]
        raw[row[0]][str(week_key)] = row[2]

    # 生成最近 8 周的周一日期，以业务日为锚
    this_monday = ref_date - _timedelta(days=ref_date.weekday())
    weeks = [this_monday - _timedelta(weeks=i) for i in range(7, -1, -1)]

    result: dict[int, list[dict]] = {}
    for mid in member_ids:
        vals = [raw.get(mid, {}).get(str(w), 0) for w in weeks]
        max_val = max(vals) if vals else 0
        weekly = []
        for v in vals:
            pct = round(v / max_val * 100) if max_val > 0 else 0
            weekly.append({"val": v, "pct": pct})
        result[mid] = weekly

    return result


def _empty_weekly() -> list[dict]:
    """返回 8 个空周数据。"""
    return [{"val": 0, "pct": 0}] * 8


@trace_service(description_zh="获取客户看板-忠诚维度", description_en="Get customer board loyal")
def get_customer_board_loyal(
    conn: Any, site_id: int, project: str, page: int, page_size: int
) -> dict:
    """
    BOARD-2 loyal 维度：忠诚度排行。

    来源: app.v_dws_member_assistant_relation_index。
    按 max_rs（最高亲密度）降序。
    """
    proj_clause, proj_params = _project_filter_clause(project, "ri.member_id")

    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            f"""
            SELECT COUNT(DISTINCT ri.member_id)
            FROM app.v_dws_member_assistant_relation_index ri
            WHERE 1=1 {proj_clause}
            """,
            proj_params,
        )
        total = cur.fetchone()[0]

        offset = (page - 1) * page_size
        cur.execute(
            f"""
            WITH member_top AS (
                SELECT ri.member_id,
                       MAX(ri.rs_display) AS max_rs,
                       (ARRAY_AGG(ri.assistant_id ORDER BY ri.rs_display DESC))[1] AS top_assistant_id,
                       (ARRAY_AGG(ri.rs_display ORDER BY ri.rs_display DESC))[1] AS top_rs
                FROM app.v_dws_member_assistant_relation_index ri
                WHERE 1=1 {proj_clause}
                GROUP BY ri.member_id
                ORDER BY MAX(ri.rs_display) DESC, ri.member_id
                LIMIT %s OFFSET %s
            )
            SELECT mt.member_id,
                   dm.nickname,
                   mt.max_rs,
                   mt.top_assistant_id,
                   mt.top_rs,
                   COALESCE(da.nickname, da.real_name, '') AS top_coach_name
            FROM member_top mt
            LEFT JOIN app.v_dim_member dm
                ON mt.member_id = dm.member_id AND dm.scd2_is_current = 1
            LEFT JOIN app.v_dim_assistant da
                ON mt.top_assistant_id = da.assistant_id AND da.scd2_is_current = 1
            ORDER BY mt.max_rs DESC, mt.member_id
            """,
            (*proj_params, page_size, offset),
        )
        items = []
        for row in cur.fetchall():
            items.append({
                "member_id": row[0],
                "name": row[1] or "",
                "intimacy": float(row[2]) if row[2] is not None else 0.0,
                "top_assistant_id": row[3],
                "top_coach_name": row[5] or "",
                "top_coach_score": f"{float(row[4]):.1f}" if row[4] is not None else "0",
                "top_coach_heart": float(row[4]) if row[4] is not None else 0.0,
            })

    return {"items": items, "total": total, "page": page, "page_size": page_size}


@trace_service(description_zh="获取客户关联助教", description_en="Get customer assistants")
def get_customer_assistants(
    conn: Any, site_id: int, member_ids: list[int]
) -> dict[int, list[dict]]:
    """
    批量查询客户关联助教列表（BOARD-2 所有维度共用）。

    来源: app.v_dws_member_assistant_relation_index + app.v_dim_assistant。
    含亲密度计算，按 rs_display 降序。

    返回 {member_id: [{name, cls, heart_score, badge, badge_cls}, ...]}。
    """
    if not member_ids:
        return {}

    result: dict[int, list[dict]] = {mid: [] for mid in member_ids}
    with _fdw_context(conn, site_id) as cur:
        # CHANGE 2026-03-28 | 助教姓名优先使用昵称/花名，不显示真实姓名
        # 改用 COALESCE(nickname, real_name)
        cur.execute(
            """
            SELECT ri.member_id,
                   COALESCE(da.nickname, da.real_name, '') AS assistant_name,
                   ri.rs_display
            FROM app.v_dws_member_assistant_relation_index ri
            LEFT JOIN app.v_dim_assistant da
                ON ri.assistant_id = da.assistant_id AND da.scd2_is_current = 1
            WHERE ri.member_id = ANY(%s)
              AND (da.leave_status IS NULL OR da.leave_status = 0)
            ORDER BY ri.member_id, ri.rs_display DESC
            """,
            (member_ids,),
        )
        for row in cur.fetchall():
            mid = row[0]
            if mid in result:
                # CHANGE 2026-03-29 | 每个客户最多显示 3 个关系最好的助教
                if len(result[mid]) < 3:
                    result[mid].append({
                        "name": row[1] or "",
                        "cls": "",
                        "heart_score": float(row[2]) if row[2] is not None else 0.0,
                    })

    return result

# ---------------------------------------------------------------------------
# BOARD-3 财务看板 FDW 查询（6 板块）
# ---------------------------------------------------------------------------


@trace_service(description_zh="获取财务概览", description_en="Get finance overview")
def get_finance_overview(
    conn: Any, site_id: int, start_date: str, end_date: str
) -> dict:
    """
    BOARD-3 经营一览：8 项核心指标（从财务日报聚合）。

    来源: app.v_dws_finance_daily_summary。
    CHANGE 2026-03-20 | 修正列名映射：
      occurrence → gross_amount, discount → discount_total,
      confirmed_revenue → confirmed_income,
      cash_in → cash_inflow_total, cash_out → cash_outflow_total,
      cash_balance → cash_balance_change
    """
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT SUM(gross_amount) AS occurrence,
                   SUM(discount_total) AS discount,
                   CASE WHEN SUM(gross_amount) > 0
                        THEN SUM(discount_total) / SUM(gross_amount)
                        ELSE 0 END AS discount_rate,
                   SUM(confirmed_income) AS confirmed_revenue,
                   SUM(cash_inflow_total) AS cash_in,
                   SUM(cash_outflow_total) AS cash_out,
                   SUM(cash_balance_change) AS cash_balance,
                   CASE WHEN SUM(cash_inflow_total) > 0
                        THEN SUM(cash_balance_change) / SUM(cash_inflow_total)
                        ELSE 0 END AS balance_rate
            FROM app.v_dws_finance_daily_summary
            WHERE stat_date >= %s::date AND stat_date <= %s::date
            """,
            (start_date, end_date),
        )
        row = cur.fetchone()

    if not row or row[0] is None:
        return {
            "occurrence": 0.0, "discount": 0.0, "discount_rate": 0.0,
            "confirmed_revenue": 0.0, "cash_in": 0.0, "cash_out": 0.0,
            "cash_balance": 0.0, "balance_rate": 0.0,
        }

    return {
        "occurrence": float(row[0]) if row[0] is not None else 0.0,
        "discount": float(row[1]) if row[1] is not None else 0.0,
        "discount_rate": float(row[2]) if row[2] is not None else 0.0,
        "confirmed_revenue": float(row[3]) if row[3] is not None else 0.0,
        "cash_in": float(row[4]) if row[4] is not None else 0.0,
        "cash_out": float(row[5]) if row[5] is not None else 0.0,
        "cash_balance": float(row[6]) if row[6] is not None else 0.0,
        "balance_rate": float(row[7]) if row[7] is not None else 0.0,
    }


@trace_service(description_zh="获取财务充值数据", description_en="Get finance recharge")
def get_finance_recharge(
    conn: Any, site_id: int, start_date: str, end_date: str
) -> dict:
    """
    BOARD-3 预收资产：储值卡 5 指标 + 赠送卡 3×4 矩阵。

    来源: app.v_dws_finance_recharge_summary + app.v_dws_finance_daily_summary。
    CHANGE 2026-03-20 | 修正列名映射：
      actual_income → recharge_cash, first_charge → first_recharge_cash,
      renew_charge → renewal_cash, consumed → 不存在（暂返回 0）,
      card_balance → cash_card_balance, all_card_balance → total_card_balance。
    CHANGE 2026-07-22 | Prompt: gift-card-breakdown Task 7.1 | 直接原因: SQL 新增 6 个赠送卡细分字段 SUM 聚合
      新增: gift_liquor_balance, gift_table_fee_balance, gift_voucher_balance,
            gift_liquor_recharge, gift_table_fee_recharge, gift_voucher_recharge
    CHANGE 2026-03-27 | board-finance-integration T2.1 | 卡余额快照值从 SUM 改为取最后一天；
      新增 consumed 从 v_dws_finance_daily_summary.card_consume_total SUM 聚合
    """
    def _f(v):
        return float(v) if v is not None else 0.0

    with _fdw_context(conn, site_id) as cur:
        # CHANGE 2026-03-27 | board-finance-integration T2.1 | 查询 1：流量值 SUM 聚合（充值相关）
        cur.execute(
            """
            SELECT SUM(recharge_cash) AS actual_income,
                   SUM(first_recharge_cash) AS first_charge,
                   SUM(renewal_cash) AS renew_charge,
                   SUM(gift_liquor_recharge) AS gift_liquor_recharge,
                   SUM(gift_table_fee_recharge) AS gift_table_fee_recharge,
                   SUM(gift_voucher_recharge) AS gift_voucher_recharge
            FROM app.v_dws_finance_recharge_summary
            WHERE stat_date >= %s::date AND stat_date <= %s::date
            """,
            (start_date, end_date),
        )
        flow_row = cur.fetchone()

        # CHANGE 2026-03-27 | board-finance-integration T2.1 | 查询 2：快照值取时间范围内最后一天
        cur.execute(
            """
            SELECT cash_card_balance,
                   total_card_balance,
                   gift_card_balance,
                   gift_liquor_balance,
                   gift_table_fee_balance,
                   gift_voucher_balance
            FROM app.v_dws_finance_recharge_summary
            WHERE stat_date = (
                SELECT MAX(stat_date)
                FROM app.v_dws_finance_recharge_summary
                WHERE stat_date >= %s::date AND stat_date <= %s::date
            )
            """,
            (start_date, end_date),
        )
        snap_row = cur.fetchone()

        # CHANGE 2026-03-27 | board-finance-integration T2.1 | 查询 3：consumed 从财务日报 SUM
        cur.execute(
            """
            SELECT COALESCE(SUM(card_consume_total), 0) AS consumed
            FROM app.v_dws_finance_daily_summary
            WHERE stat_date >= %s::date AND stat_date <= %s::date
            """,
            (start_date, end_date),
        )
        consume_row = cur.fetchone()

    if not flow_row or flow_row[0] is None:
        return _empty_recharge_data()

    # 流量值
    actual_income = _f(flow_row[0])
    first_charge = _f(flow_row[1])
    renew_charge = _f(flow_row[2])
    gift_liquor_recharge = _f(flow_row[3])
    gift_table_fee_recharge = _f(flow_row[4])
    gift_voucher_recharge = _f(flow_row[5])

    # 快照值（最后一天）
    card_balance = _f(snap_row[0]) if snap_row else 0.0
    all_card_balance = _f(snap_row[1]) if snap_row else 0.0
    gift_balance = _f(snap_row[2]) if snap_row else 0.0
    gift_liquor_balance = _f(snap_row[3]) if snap_row else 0.0
    gift_table_fee_balance = _f(snap_row[4]) if snap_row else 0.0
    gift_voucher_balance = _f(snap_row[5]) if snap_row else 0.0

    # consumed
    consumed = _f(consume_row[0]) if consume_row else 0.0

    # CHANGE 2026-03-20 | gift_rows 每个 cell 必须是 GiftCell dict（{"value": float}），
    # 不能是裸 float，否则 Pydantic ResponseValidationError
    _gc = lambda v: {"value": v}
    return {
        "actual_income": actual_income,
        "first_charge": first_charge,
        "renew_charge": renew_charge,
        "consumed": consumed,
        "card_balance": card_balance,
        "gift_rows": [
            # 新增行：total = 三个细分之和（保证 total = liquor + table_fee + voucher 恒等）
            {"label": "新增",
             "total": _gc(gift_liquor_recharge + gift_table_fee_recharge + gift_voucher_recharge),
             "liquor": _gc(gift_liquor_recharge),
             "table_fee": _gc(gift_table_fee_recharge),
             "voucher": _gc(gift_voucher_recharge)},
            # 消费行：上游 API 仅提供消费总额，无法按卡类型拆分，细分列保持 0
            {"label": "消费",
             "total": _gc(0.0),
             "liquor": _gc(0.0),
             "table_fee": _gc(0.0),
             "voucher": _gc(0.0)},
            # 余额行：填充对应细分余额
            {"label": "余额",
             "total": _gc(gift_balance),
             "liquor": _gc(gift_liquor_balance),
             "table_fee": _gc(gift_table_fee_balance),
             "voucher": _gc(gift_voucher_balance)},
        ],
        "all_card_balance": all_card_balance,
    }


def _empty_recharge_data() -> dict:
    """预收资产空默认值。"""
    _gc = lambda v: {"value": v}
    empty_row = {"label": "", "total": _gc(0.0), "liquor": _gc(0.0), "table_fee": _gc(0.0), "voucher": _gc(0.0)}
    return {
        "actual_income": 0.0, "first_charge": 0.0, "renew_charge": 0.0,
        "consumed": 0.0, "card_balance": 0.0,
        "gift_rows": [
            {**empty_row, "label": "新增"},
            {**empty_row, "label": "消费"},
            {**empty_row, "label": "余额"},
        ],
        "all_card_balance": 0.0,
    }


@trace_service(description_zh="获取财务收入数据", description_en="Get finance revenue")
def get_finance_revenue(
    conn: Any, site_id: int, start_date: str, end_date: str, area: str = "all"
) -> dict:
    """
    BOARD-3 应计收入：收入结构 + 价格构成 + 优惠明细 + 渠道分布。

    CHANGE 2026-03-27 | board-finance-phase2 T1+T2 | 收入结构改为从 dwd_settlement_head
      按物理区域聚合；优惠减扣按 Demo 4 项重组（从 v_dws_finance_daily_summary discount_* 字段）。
      原因：原 v_dws_finance_income_structure 分类与 Demo 不一致，需按物理区域展示子行。
    ⚠️ DWD-DOC 规则 2: 助教行使用 assistant_pd_money（陪打）+ assistant_cx_money（超休）。
    """
    with _fdw_context(conn, site_id) as cur:
        # AreaFilterEnum value -> physical area label mapping
        # CHANGE 2026-03-28 | Bug3 fix: add area filter to revenue SQL
        _AREA_LABEL_MAP = {
            "hall":    ["A区", "B区", "C区", "台球包厢", "斯诺克", "麻将区", "团建区"],
            "hallA":   ["A区"],
            "hallB":   ["B区"],
            "hallC":   ["C区"],
            "vip":     ["台球包厢"],
            "snooker": ["斯诺克"],
            "mahjong": ["麻将区"],
            "ktv":     ["团建区"],
        }
        area_filter_labels = _AREA_LABEL_MAP.get(area, None)  # None means "all"

        # T1: revenue structure from dwd_settlement_head aggregated by physical area
        area_where = ""
        area_params: list = [start_date, end_date]
        if area_filter_labels:
            area_where = "AND area_label = ANY(%s)"
            area_params.append(area_filter_labels)

        cur.execute(
            f"""
            SELECT area_label,
                   SUM(table_charge_money) AS table_charge,
                   SUM(goods_money) AS goods,
                   SUM(assistant_pd_money) AS pd,
                   SUM(assistant_cx_money) AS cx
            FROM (
                SELECT CASE t.site_table_area_name
                    WHEN 'A区' THEN 'A区'
                    WHEN 'B区' THEN 'B区'
                    WHEN 'C区' THEN 'C区' WHEN 'TV台' THEN 'C区' WHEN '美洲豹赛台' THEN 'C区'
                    WHEN 'VIP包厢' THEN '台球包厢'
                    WHEN '斯诺克区' THEN '斯诺克'
                    WHEN '麻将房' THEN '麻将区' WHEN 'M7' THEN '麻将区' WHEN 'M8' THEN '麻将区'
                    WHEN '666' THEN '麻将区' WHEN '发财' THEN '麻将区'
                    WHEN 'K包' THEN '团建区' WHEN 'k包活动区' THEN '团建区' WHEN '幸会158' THEN '团建区'
                    ELSE NULL
                END AS area_label,
                h.table_charge_money, h.goods_money,
                h.assistant_pd_money, h.assistant_cx_money
                FROM app.v_dwd_settlement_head h
                LEFT JOIN app.v_dim_table t ON h.table_id = t.table_id AND t.scd2_is_current = 1
                WHERE h.settle_type IN (1, 3)
                  AND (h.create_time AT TIME ZONE 'Asia/Shanghai')::date >= %s
                  AND (h.create_time AT TIME ZONE 'Asia/Shanghai')::date <= %s
            ) sub
            WHERE area_label IS NOT NULL
            {area_where}
            GROUP BY area_label
            ORDER BY CASE area_label
                WHEN 'A区' THEN 1 WHEN 'B区' THEN 2 WHEN 'C区' THEN 3
                WHEN '台球包厢' THEN 4 WHEN '斯诺克' THEN 5 WHEN '麻将区' THEN 6 WHEN '团建区' THEN 7
                ELSE 8
            END
            """,
            area_params,
        )
        area_rows = cur.fetchall()

        # 聚合各区域数据
        total_table_charge = 0.0
        total_goods = 0.0
        total_pd = 0.0
        total_cx = 0.0
        area_sub_rows = []
        for row in area_rows:
            tc = float(row[1]) if row[1] is not None else 0.0
            gd = float(row[2]) if row[2] is not None else 0.0
            pd = float(row[3]) if row[3] is not None else 0.0
            cx = float(row[4]) if row[4] is not None else 0.0
            total_table_charge += tc
            total_goods += gd
            total_pd += pd
            total_cx += cx
            area_sub_rows.append({"name": row[0], "amount": tc})

        # 构建 structure_rows（与前端 _loadData 映射一致）
        structure_rows = []
        # "开台与包厢" 主行
        structure_rows.append({
            "id": "table_charge", "name": "开台与包厢", "desc": None,
            "is_sub": False, "amount": total_table_charge,
            "discount": 0.0, "booked": total_table_charge,
        })
        # 各区域子行
        for sub in area_sub_rows:
            structure_rows.append({
                "id": f"area_{sub['name']}", "name": sub["name"], "desc": None,
                "is_sub": True, "amount": sub["amount"],
                "discount": 0.0, "booked": sub["amount"],
            })
        # "助教 基础课" 行
        structure_rows.append({
            "id": "assistant_pd", "name": "助教 基础课", "desc": None,
            "is_sub": False, "amount": total_pd,
            "discount": 0.0, "booked": total_pd,
        })
        # "助教 激励课" 行
        structure_rows.append({
            "id": "assistant_cx", "name": "助教 激励课", "desc": None,
            "is_sub": False, "amount": total_cx,
            "discount": 0.0, "booked": total_cx,
        })
        # "食品酒水" 行
        structure_rows.append({
            "id": "goods", "name": "食品酒水", "desc": None,
            "is_sub": False, "amount": total_goods,
            "discount": 0.0, "booked": total_goods,
        })

        # 发生额构成
        price_items = [
            {"label": "开台消费", "amount": total_table_charge},
            {"label": "酒水商品", "amount": total_goods},
            {"label": "助教服务", "amount": total_pd + total_cx},
        ]
        total_income = total_table_charge + total_goods + total_pd + total_cx

        # ── T2: 优惠减扣（从 v_dws_finance_daily_summary discount_* 字段聚合）──
        cur.execute(
            """
            SELECT COALESCE(SUM(discount_groupbuy), 0) AS groupbuy,
                   COALESCE(SUM(discount_vip), 0) AS vip,
                   COALESCE(SUM(discount_manual), 0) + COALESCE(SUM(discount_other), 0) AS manual_adjust,
                   COALESCE(SUM(discount_gift_card), 0) AS gift_card,
                   COALESCE(SUM(discount_rounding), 0) AS rounding
            FROM app.v_dws_finance_daily_summary
            WHERE stat_date >= %s AND stat_date <= %s
            """,
            (start_date, end_date),
        )
        dr = cur.fetchone()
        groupbuy_d = float(dr[0]) if dr and dr[0] is not None else 0.0
        vip_d = float(dr[1]) if dr and dr[1] is not None else 0.0
        manual_d = float(dr[2]) if dr and dr[2] is not None else 0.0
        gift_card_d = float(dr[3]) if dr and dr[3] is not None else 0.0
        rounding_d = float(dr[4]) if dr and dr[4] is not None else 0.0

        discount_items = [
            {"label": "团购优惠", "amount": groupbuy_d},
            {"label": "会员折扣", "amount": vip_d},
            {"label": "手动调整", "amount": manual_d},
            {"label": "赠送卡抵扣", "desc": "台桌卡+酒水卡+抵用券", "amount": gift_card_d},
            {"label": "其他优惠", "desc": "免单+抹零", "amount": rounding_d},
        ]
        total_discount = groupbuy_d + vip_d + manual_d + gift_card_d + rounding_d

        # 回填收入结构表的优惠分摊
        # CHANGE 2026-03-28 | 按区域 table_charge_money 占比分摊 DWS discount_total
        if total_table_charge > 0 and total_discount > 0:
            for row in structure_rows:
                if row["is_sub"]:
                    # 区域子行：按占比分摊
                    ratio = row["amount"] / total_table_charge
                    row["discount"] = round(total_discount * ratio, 2)
                    row["booked"] = row["amount"] - row["discount"]
                elif row["id"] == "table_charge":
                    # "开台与包厢"主行
                    row["discount"] = total_discount
                    row["booked"] = total_table_charge - total_discount

        # ── 渠道分布（从 v_dws_finance_daily_summary 聚合，label 对齐 Demo）──
        cur.execute(
            """
            SELECT COALESCE(SUM(cash_pay_amount), 0) AS cash_pay,
                   COALESCE(SUM(groupbuy_pay_amount), 0) AS groupbuy_pay,
                   COALESCE(SUM(cash_card_consume), 0) AS cash_card,
                   COALESCE(SUM(gift_card_consume), 0) AS gift_card
            FROM app.v_dws_finance_daily_summary
            WHERE stat_date >= %s::date AND stat_date <= %s::date
            """,
            (start_date, end_date),
        )
        ch = cur.fetchone()
        cash_pay = float(ch[0]) if ch and ch[0] is not None else 0.0
        groupbuy_pay = float(ch[1]) if ch and ch[1] is not None else 0.0
        cash_card = float(ch[2]) if ch and ch[2] is not None else 0.0
        gift_card_consume = float(ch[3]) if ch and ch[3] is not None else 0.0

        channel_items = [
            {"label": "储值卡结算冲销", "amount": cash_card + gift_card_consume},
            {"label": "现金/线上支付", "amount": cash_pay},
            {"label": "团购核销确认收入", "desc": "团购成交价", "amount": groupbuy_pay},
        ]

    confirmed_total = total_income - abs(total_discount)
    return {
        "structure_rows": structure_rows,
        "price_items": price_items,
        "total_occurrence": total_income,
        "discount_items": discount_items,
        # CHANGE 2026-03-27 | board-finance-phase2 T3 | 优惠总计供前端展示
        "discount_total": total_discount,
        "confirmed_total": confirmed_total,
        "channel_items": channel_items,
    }


@trace_service(description_zh="获取财务现金流", description_en="Get finance cashflow")
def get_finance_cashflow(
    conn: Any, site_id: int, start_date: str, end_date: str
) -> dict:
    """
    BOARD-3 现金流入：消费收款 + 充值收款。

    来源: app.v_dws_finance_daily_summary（消费收款 + 充值收款字段均在财务日报中）。
    CHANGE 2026-03-20 | 修正列名映射：
      consume_cash_pay → cash_pay_amount,
      consume_online_pay → groupbuy_pay_amount (团购核销),
      consume_balance_pay → card_consume_total (储值卡消费),
      recharge_income → recharge_cash_inflow (充值现金流入)
    CHANGE 2026-03-27 | board-finance-integration T2.2 | 移除"储值卡消费"（非现金流入），
      新增"纸币现金"和"扫码收款"拆分（依赖 T1.1 新增字段 cash_paper_amount/scan_pay_amount）
    ⚠️ DWD-DOC 规则 7: platform_settlement_amount 和 groupbuy_pay_amount 互斥。
    """
    with _fdw_context(conn, site_id) as cur:
        # CHANGE 2026-03-27 | board-finance-integration T2.2 | SQL 拆分现金收款为纸币+扫码
        # CHANGE 2026-07-22 | Prompt: 财务看板多问题修复 | 直接原因: ETL 未重跑时 cash_paper/scan_pay 为 0，需回退用 cash_pay_amount 合并
        cur.execute(
            """
            SELECT SUM(cash_pay_amount) AS cash_pay,
                   COALESCE(SUM(cash_paper_amount), 0) AS cash_paper,
                   COALESCE(SUM(scan_pay_amount), 0) AS scan_pay,
                   SUM(groupbuy_pay_amount) AS consume_groupbuy,
                   SUM(recharge_cash_inflow) AS recharge_income,
                   SUM(cash_inflow_total) AS total
            FROM app.v_dws_finance_daily_summary
            WHERE stat_date >= %s::date AND stat_date <= %s::date
            """,
            (start_date, end_date),
        )
        row = cur.fetchone()

    def _f(v):
        return float(v) if v is not None else 0.0

    if not row or row[5] is None:
        return {
            "consume_items": [
                {"label": "现金/线上收款", "amount": 0.0},
                {"label": "团购平台回款", "amount": 0.0},
            ],
            "recharge_items": [{"label": "会员充值到账", "amount": 0.0}],
            "total": 0.0,
        }

    # CHANGE 2026-07-22 | Prompt: 财务看板多问题修复 | 直接原因: ETL 未重跑时 cash_paper+scan_pay=0，回退为合并项
    cash_pay = _f(row[0])
    cash_paper = _f(row[1])
    scan_pay = _f(row[2])
    groupbuy = _f(row[3])

    # CHANGE 2026-03-27 | board-finance-phase2 T4 | 现金流入项名和描述对齐 Demo
    if cash_paper + scan_pay > 0:
        # ETL 已重跑，拆分为"纸币现金"和"线上收款"
        consume_items = [
            {"label": "纸币现金", "desc": "柜台现金收款", "amount": cash_paper},
            {"label": "线上收款", "desc": "微信/支付宝/刷卡", "amount": scan_pay},
            {"label": "团购平台", "desc": "美团/抖音回款", "amount": groupbuy},
        ]
    else:
        # ETL 未重跑，合并为一项"现金/线上收款"（降级）
        consume_items = [
            {"label": "现金/线上收款", "amount": cash_pay},
            {"label": "团购平台", "desc": "美团/抖音回款", "amount": groupbuy},
        ]

    return {
        "consume_items": consume_items,
        "recharge_items": [{"label": "会员充值到账", "desc": "首充/续费实收", "amount": _f(row[4])}],
        "total": _f(row[5]),
    }


@trace_service(description_zh="获取财务支出数据", description_en="Get finance expense")
def get_finance_expense(
    conn: Any, site_id: int, start_date: str, end_date: str
) -> dict:
    """
    BOARD-3 现金流出：支出明细 4 子分组。

    来源: app.v_dws_finance_expense_summary + app.v_dws_platform_settlement。
    CHANGE 2026-03-20 | 修正列名映射：
      expense_group → expense_category, expense_label → expense_type_name,
      stat_date → expense_month (expense_summary),
      stat_date → settlement_date (platform_settlement)
    CHANGE 2026-03-27 | board-finance-phase2 T5 | 现金流出固定项名对齐 Demo，空数据时显示固定结构
    ⚠️ DWD-DOC 规则 2: coachItems 中基础课使用 assistant_pd_money，激励课使用 assistant_cx_money。
    """
    with _fdw_context(conn, site_id) as cur:
        # 支出汇总（日期列为 expense_month，分组列为 expense_category）
        cur.execute(
            """
            SELECT expense_category, expense_type_name,
                   SUM(expense_amount) AS amount
            FROM app.v_dws_finance_expense_summary
            WHERE expense_month >= %s::date AND expense_month <= %s::date
            GROUP BY expense_category, expense_type_name
            ORDER BY expense_category, expense_type_name
            """,
            (start_date, end_date),
        )
        groups: dict[str, list[dict]] = {
            "operation": [], "fixed": [], "coach": [], "platform": [],
        }
        total = 0.0
        for row in cur.fetchall():
            group = row[0] or "operation"
            label = row[1] or ""
            amt = float(row[2]) if row[2] is not None else 0.0
            if group in groups:
                groups[group].append({"label": label, "amount": amt})
            total += amt

        # 平台服务费（独立视图，日期列为 settlement_date）
        cur.execute(
            """
            SELECT platform_name, SUM(service_fee) AS fee
            FROM app.v_dws_platform_settlement
            WHERE settlement_date >= %s::date AND settlement_date <= %s::date
            GROUP BY platform_name
            ORDER BY platform_name
            """,
            (start_date, end_date),
        )
        for row in cur.fetchall():
            amt = float(row[1]) if row[1] is not None else 0.0
            groups["platform"].append({"label": row[0] or "", "amount": amt})
            total += amt

    # CHANGE 2026-03-27 | board-finance-phase2 T5 | 空数据时填充固定项名对齐 Demo
    if not groups["coach"]:
        groups["coach"] = [
            {"label": "基础课分成", "amount": 0.0},
            {"label": "激励课分成", "amount": 0.0},
            {"label": "充值提成", "amount": 0.0},
            {"label": "额外奖金", "amount": 0.0},
        ]
    if not groups["platform"]:
        groups["platform"] = [
            {"label": "汇来米", "amount": 0.0},
            {"label": "美团", "amount": 0.0},
            {"label": "抖音", "amount": 0.0},
        ]
    if not groups["operation"]:
        groups["operation"] = [
            {"label": "食品饮料", "amount": 0.0},
            {"label": "耗材", "amount": 0.0},
            {"label": "报销", "amount": 0.0},
        ]
    if not groups["fixed"]:
        groups["fixed"] = [
            {"label": "房租", "amount": 0.0},
            {"label": "水电", "amount": 0.0},
            {"label": "物业", "amount": 0.0},
            {"label": "人员工资", "amount": 0.0},
        ]

    return {
        "operation_items": groups["operation"],
        "fixed_items": groups["fixed"],
        "coach_items": groups["coach"],
        "platform_items": groups["platform"],
        "total": total,
    }


def get_finance_coach_analysis(
    conn: Any, site_id: int, start_date: str, end_date: str
) -> dict:
    """
    BOARD-3 助教分析：按 assistant_level_name 分组聚合。

    三列全部从 DWS dws_assistant_salary_calc 计算：
      - pay = SUM(hours × course_price)
      - share = SUM(hours × deduction)
      - hourly = share / hours（加权平均抽成单价）

    CHANGE 2026-03-28 | board-finance-phase2 bugfix | 改回纯 DWS 方案。
      禁止从 DWD ledger_amount JOIN DWS 等级——DWS 同一助教同月可有多等级记录
      （月中升级），level_map JOIN service_log 会导致行膨胀。
    """
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT assistant_level_name AS level,
                   SUM(base_hours * base_course_price) AS base_pay,
                   SUM(base_hours * base_deduction) AS base_share,
                   SUM(base_hours) AS base_hours,
                   SUM(bonus_hours * bonus_course_price) AS bonus_pay,
                   SUM(bonus_hours * bonus_course_price * bonus_deduction_ratio) AS bonus_share,
                   SUM(bonus_hours) AS bonus_hours
            FROM app.v_dws_assistant_salary_calc
            WHERE salary_month >= %s::date AND salary_month <= %s::date
            GROUP BY assistant_level_name
            ORDER BY CASE assistant_level_name
                WHEN '初级' THEN 1 WHEN '中级' THEN 2
                WHEN '高级' THEN 3 WHEN '星级' THEN 4 ELSE 5
            END
            """,
            (start_date, end_date),
        )
        rows = cur.fetchall()

    basic_rows = []
    incentive_rows = []
    total_base_pay = 0.0
    total_base_share = 0.0
    total_base_hours = 0.0
    total_bonus_pay = 0.0
    total_bonus_share = 0.0
    total_bonus_hours = 0.0

    for row in rows:
        level = row[0] or ""
        base_pay = float(row[1]) if row[1] is not None else 0.0
        base_share = float(row[2]) if row[2] is not None else 0.0
        base_h = float(row[3]) if row[3] is not None else 0.0
        bonus_pay = float(row[4]) if row[4] is not None else 0.0
        bonus_share = float(row[5]) if row[5] is not None else 0.0
        bonus_h = float(row[6]) if row[6] is not None else 0.0

        total_base_pay += base_pay
        total_base_share += base_share
        total_base_hours += base_h
        total_bonus_pay += bonus_pay
        total_bonus_share += bonus_share
        total_bonus_hours += bonus_h

        base_hourly = base_share / base_h if base_h > 0 else 0.0
        basic_rows.append({
            "level": level, "pay": base_pay,
            "share": base_share, "hourly": round(base_hourly, 2),
        })
        bonus_hourly = bonus_share / bonus_h if bonus_h > 0 else 0.0
        incentive_rows.append({
            "level": level, "pay": bonus_pay,
            "share": bonus_share, "hourly": round(bonus_hourly, 2),
        })

    avg_base_hourly = round(total_base_share / total_base_hours, 2) if total_base_hours > 0 else 0.0
    avg_bonus_hourly = round(total_bonus_share / total_bonus_hours, 2) if total_bonus_hours > 0 else 0.0

    return {
        "basic": {
            "total_pay": total_base_pay,
            "total_share": total_base_share,
            "avg_hourly": avg_base_hourly,
            "rows": basic_rows,
        },
        "incentive": {
            "total_pay": total_bonus_pay,
            "total_share": total_bonus_share,
            "avg_hourly": avg_bonus_hourly,
            "rows": incentive_rows,
        },
    }



def get_finance_coach_analysis_area(
    conn: Any, site_id: int, start_date: str, end_date: str, area_code: str
) -> dict:
    """
    BOARD-3 助教分析（区域过滤版）：area≠all 时从 dws_coach_area_hours
    JOIN dws_assistant_salary_calc 按区域聚合。

    返回结构与 get_finance_coach_analysis 完全一致（basic/incentive 两表）。

    CHANGE 2026-03-29 | Prompt: 助教分析按区域细化 | 新增区域过滤查询
    """
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT
                sc.assistant_level_name AS level,
                SUM(cah.base_hours * sc.base_course_price) AS base_pay,
                SUM(cah.base_hours * sc.base_deduction) AS base_share,
                SUM(cah.base_hours) AS base_hours,
                SUM(cah.bonus_hours * sc.bonus_course_price) AS bonus_pay,
                SUM(cah.bonus_hours * sc.bonus_course_price * sc.bonus_deduction_ratio) AS bonus_share,
                SUM(cah.bonus_hours) AS bonus_hours
            FROM app.v_dws_coach_area_hours cah
            JOIN app.v_dws_assistant_salary_calc sc
                ON cah.assistant_id = sc.assistant_id
                AND cah.stat_month = sc.salary_month
            WHERE cah.area_code = %s
              AND cah.stat_month >= %s::date
              AND cah.stat_month <= %s::date
            GROUP BY sc.assistant_level_name
            ORDER BY CASE sc.assistant_level_name
                WHEN '初级' THEN 1 WHEN '中级' THEN 2
                WHEN '高级' THEN 3 WHEN '星级' THEN 4 ELSE 5
            END
            """,
            (area_code, start_date, end_date),
        )
        rows = cur.fetchall()

    # 与 get_finance_coach_analysis 完全相同的后处理逻辑
    basic_rows = []
    incentive_rows = []
    total_base_pay = 0.0
    total_base_share = 0.0
    total_base_hours = 0.0
    total_bonus_pay = 0.0
    total_bonus_share = 0.0
    total_bonus_hours = 0.0

    for row in rows:
        level = row[0] or ""
        base_pay = float(row[1]) if row[1] is not None else 0.0
        base_share = float(row[2]) if row[2] is not None else 0.0
        base_h = float(row[3]) if row[3] is not None else 0.0
        bonus_pay = float(row[4]) if row[4] is not None else 0.0
        bonus_share = float(row[5]) if row[5] is not None else 0.0
        bonus_h = float(row[6]) if row[6] is not None else 0.0

        total_base_pay += base_pay
        total_base_share += base_share
        total_base_hours += base_h
        total_bonus_pay += bonus_pay
        total_bonus_share += bonus_share
        total_bonus_hours += bonus_h

        base_hourly = base_share / base_h if base_h > 0 else 0.0
        basic_rows.append({
            "level": level, "pay": base_pay,
            "share": base_share, "hourly": round(base_hourly, 2),
        })
        bonus_hourly = bonus_share / bonus_h if bonus_h > 0 else 0.0
        incentive_rows.append({
            "level": level, "pay": bonus_pay,
            "share": bonus_share, "hourly": round(bonus_hourly, 2),
        })

    avg_base_hourly = round(total_base_share / total_base_hours, 2) if total_base_hours > 0 else 0.0
    avg_bonus_hourly = round(total_bonus_share / total_bonus_hours, 2) if total_bonus_hours > 0 else 0.0

    return {
        "basic": {
            "total_pay": total_base_pay,
            "total_share": total_base_share,
            "avg_hourly": avg_base_hourly,
            "rows": basic_rows,
        },
        "incentive": {
            "total_pay": total_bonus_pay,
            "total_share": total_bonus_share,
            "avg_hourly": avg_bonus_hourly,
            "rows": incentive_rows,
        },
    }


@trace_service(description_zh="获取项目类型配置", description_en="Get skill types")
def get_skill_types(conn: Any, site_id: int) -> list[dict]:
    """
    CONFIG-1: 查询项目类型筛选器配置。

    来源: app.v_cfg_area_category（基于 dws.cfg_area_category 去重，
    排除 SPECIAL/OTHER）。返回列表头部插入"不限"选项。
    查询失败时由调用方降级返回空数组。
    """
    # CHANGE 2026-03-20 | R3 修复：原查询虚构的 v_cfg_skill_type 视图不存在，
    # 改为查询 v_cfg_area_category（项目类型配置），value 直接用 category_code
    # （BILLIARD/SNOOKER/MAHJONG/KTV），前端枚举同步修改。
    # 假设：cfg_area_category 的 category_code 是稳定的业务标识，不会频繁变动。
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT category_code, display_name, short_name
            FROM app.v_cfg_area_category
            ORDER BY sort_order
            """
        )
        # 头部插入"不限"选项（后端生成，不存数据库）
        items: list[dict] = [{"key": "ALL", "label": "不限", "emoji": "", "cls": ""}]
        for row in cur.fetchall():
            items.append({
                "key": row[0] or "",
                "label": row[1] or "",
                "emoji": row[2] or "",
                "cls": "",
            })
    return items


# ═══════════════════════════════════════════════════════════════════════════════
# P17：助教客户归属与任务生成引擎 — ETL 查询方法
# ═══════════════════════════════════════════════════════════════════════════════


@trace_service(
    description_zh="获取有归属数据的门店列表",
    description_en="Get active site IDs from relation index",
)
def get_active_site_ids(conn: Any) -> list[int]:
    """
    从 ETL 关系指数表获取所有有 session_count > 0 的 distinct site_id。

    CHANGE 2026-03-29 | 替代 auth.user_assistant_binding 作为 task_generator 的门店入口，
    确保未绑定小程序的门店也能生成任务数据。

    直接查询 dws schema（非 RLS 视图），不需要 site_id 参数。
    """
    from app.database import get_etl_readonly_connection

    # site_id=0 仅用于获取连接，不设置 RLS（查询 dws schema 不受 RLS 限制）
    etl_conn = get_etl_readonly_connection(0)
    try:
        with etl_conn.cursor() as cur:
            cur.execute(
                """
                SELECT DISTINCT site_id
                FROM dws.dws_member_assistant_relation_index
                WHERE session_count > 0
                """
            )
            return [row[0] for row in cur.fetchall()]
    finally:
        etl_conn.close()


@trace_service(
    description_zh="获取归属对（OS）",
    description_en="Get ownership pairs",
)
def get_ownership_pairs(
    conn: Any, site_id: int
) -> list[dict]:
    """
    P17 Step 1: 查询 os_label ∈ {MAIN, COMANAGE} 的所有 (assistant_id, member_id) 对。

    来源: app.v_dws_member_assistant_relation_index（RLS 视图，按 site_id 隔离）
    返回: [{"assistant_id", "member_id", "os_label", "rs", "ms", "ml"}]

    ⚠️ 这是 P17 的核心入口查询，取代旧的 auth.user_assistant_binding 入口。
    只返回 MAIN/COMANAGE 对，POOL/UNASSIGNED 不在常规任务生成范围内。
    """
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT assistant_id,
                   member_id,
                   os_label,
                   COALESCE(rs_display, 0) AS rs,
                   COALESCE(ms_display, 0) AS ms,
                   COALESCE(ml_display, 0) AS ml
            FROM app.v_dws_member_assistant_relation_index
            WHERE os_label IN ('MAIN', 'COMANAGE')
            """
        )
        return [
            {
                "assistant_id": row[0],
                "member_id": row[1],
                "os_label": row[2],
                "rs": Decimal(str(row[3])),
                "ms": Decimal(str(row[4])),
                "ml": Decimal(str(row[5])),
            }
            for row in cur.fetchall()
        ]


@trace_service(
    description_zh="获取 POOL 助教候选",
    description_en="Get pool assistants for transfer",
)
def get_pool_assistants(
    conn: Any, site_id: int, member_id: int
) -> list[dict]:
    """
    P17 转移子流程: 查询某客户的 POOL 助教候选池。

    来源: app.v_dws_member_assistant_relation_index
    返回: [{"assistant_id", "rs", "ms", "ml"}]

    ⚠️ 保护 3（服务关系门槛）在此处实施：只返回 os_label='POOL' 的助教，
    UNASSIGNED 永远不参与转移候选。
    """
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT assistant_id,
                   COALESCE(rs_display, 0) AS rs,
                   COALESCE(ms_display, 0) AS ms,
                   COALESCE(ml_display, 0) AS ml
            FROM app.v_dws_member_assistant_relation_index
            WHERE member_id = %s
              AND os_label = 'POOL'
            """,
            (member_id,),
        )
        return [
            {
                "assistant_id": row[0],
                "rs": Decimal(str(row[1])),
                "ms": Decimal(str(row[2])),
                "ml": Decimal(str(row[3])),
            }
            for row in cur.fetchall()
        ]


@trace_service(
    description_zh="批量获取 WBI",
    description_en="Batch fetch WBI scores",
)
def get_wbi_batch(
    conn: Any, site_id: int
) -> dict[int, Decimal]:
    """
    P17 Step 2: 批量读取全店 WBI（流失回赢指数）。

    来源: app.v_dws_member_winback_index（RLS 视图）
    返回: {member_id: display_score}

    ⚠️ 全表扫描无 WHERE（除 RLS 隔离），与旧 _process_assistant() 行为一致。
    """
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT member_id, COALESCE(display_score, 0)
            FROM app.v_dws_member_winback_index
            """
        )
        return {row[0]: Decimal(str(row[1])) for row in cur.fetchall()}


@trace_service(
    description_zh="批量获取 NCI",
    description_en="Batch fetch NCI scores",
)
def get_nci_batch(
    conn: Any, site_id: int
) -> dict[int, Decimal]:
    """
    P17 Step 2: 批量读取全店 NCI（新客转化指数）。

    来源: app.v_dws_member_newconv_index（RLS 视图）
    返回: {member_id: display_score}

    FIX 2026-04-12: 排除已转老客的会员。NCI 表只在 NEW 阶段写入，
    会员转 OLD 后不再更新，导致残留过时高分。用 WBI status='OLD' 过滤。
    """
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT n.member_id, COALESCE(n.display_score, 0)
            FROM app.v_dws_member_newconv_index n
            WHERE NOT EXISTS (
                SELECT 1 FROM app.v_dws_member_winback_index w
                WHERE w.member_id = n.member_id AND w.status = 'OLD'
            )
            """
        )
        return {row[0]: Decimal(str(row[1])) for row in cur.fetchall()}


@trace_service(
    description_zh="批量获取理想到店周期",
    description_en="Batch fetch ideal interval days",
)
def get_ideal_interval_batch(
    conn: Any, site_id: int
) -> dict[int, float]:
    """
    批量读取全店客户的理想到店周期（天）。

    来源: app.v_dws_member_winback_index（RLS 视图）
    返回: {member_id: ideal_interval_days}（仅包含非 NULL 的记录）

    CHANGE 2026-03-29 | OS 分级分配：用于计算升级倍数
    """
    with _fdw_context(conn, site_id) as cur:
        cur.execute(
            """
            SELECT member_id, ideal_interval_days
            FROM app.v_dws_member_winback_index
            WHERE ideal_interval_days IS NOT NULL
              AND ideal_interval_days > 0
            """
        )
        return {row[0]: float(row[1]) for row in cur.fetchall()}


@trace_service(
    description_zh="获取全量服务关系对",
    description_en="Get all service relation pairs",
)
def get_all_service_pairs(
    conn: Any, site_id: int
) -> list[dict]:
    """
    查询 session_count > 0 的全量 (assistant_id, member_id) 对。

    用于 relationship_building 保底任务生成：只要助教与客户确切发生过服务关系，
    就生成一条 relationship_building 任务（不限 os_label）。

    来源: app.v_dws_member_assistant_relation_index（RLS 视图，按 site_id 隔离）
    返回: [{"assistant_id", "member_id", "rs"}]
    """
    with _fdw_context(conn, site_id) as cur:
        # POOL 客户需 session_count >= 3 才纳入保底任务，MAIN/COMANAGE 无限制
        cur.execute(
            """
            SELECT assistant_id,
                   member_id,
                   COALESCE(rs_display, 0) AS rs
            FROM app.v_dws_member_assistant_relation_index
            WHERE session_count > 0
              AND (os_label IN ('MAIN', 'COMANAGE') OR session_count >= 3)
            """
        )
        return [
            {
                "assistant_id": row[0],
                "member_id": row[1],
                "rs": Decimal(str(row[2])),
            }
            for row in cur.fetchall()
        ]


# ---------------------------------------------------------------------------
# 区域财务看板查询（board-finance-dws-area-refactor）
# ---------------------------------------------------------------------------


@trace_service(
    description_zh="获取区域财务概览", description_en="Get finance overview by area"
)
def get_finance_overview_area(
    conn: Any,
    site_id: int,
    start_date: str,
    end_date: str,
    area_code: str = "all",
    *,
    etl_conn: Any = None,
) -> dict:
    """
    从 v_dws_finance_area_daily 按 area_code 聚合 overview 8 项指标。

    与 get_finance_overview 返回结构完全一致，但数据来源改为区域日粒度表，
    支持按 area_code 过滤。

    Requirements: 6.1, 6.2, 6.5
    """
    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        cur.execute(
            """
            SELECT SUM(gross_amount)       AS occurrence,
                   SUM(discount_total)     AS discount,
                   CASE WHEN SUM(gross_amount) > 0
                        THEN SUM(discount_total) / SUM(gross_amount)
                        ELSE 0 END         AS discount_rate,
                   SUM(confirmed_income)   AS confirmed_revenue,
                   SUM(cash_inflow_total)  AS cash_in,
                   SUM(cash_outflow_total) AS cash_out,
                   SUM(cash_balance_change) AS cash_balance,
                   CASE WHEN SUM(cash_inflow_total) > 0
                        THEN SUM(cash_balance_change) / SUM(cash_inflow_total)
                        ELSE 0 END         AS balance_rate
            FROM app.v_dws_finance_area_daily
            WHERE area_code = %s
              AND stat_date >= %s::date
              AND stat_date <= %s::date
            """,
            (area_code, start_date, end_date),
        )
        row = cur.fetchone()

    _zero = {
        "occurrence": 0.0,
        "discount": 0.0,
        "discount_rate": 0.0,
        "confirmed_revenue": 0.0,
        "cash_in": 0.0,
        "cash_out": 0.0,
        "cash_balance": 0.0,
        "balance_rate": 0.0,
    }
    if not row or row[0] is None:
        return _zero

    return {
        "occurrence": float(row[0]) if row[0] is not None else 0.0,
        "discount": float(row[1]) if row[1] is not None else 0.0,
        "discount_rate": float(row[2]) if row[2] is not None else 0.0,
        "confirmed_revenue": float(row[3]) if row[3] is not None else 0.0,
        "cash_in": float(row[4]) if row[4] is not None else 0.0,
        "cash_out": float(row[5]) if row[5] is not None else 0.0,
        "cash_balance": float(row[6]) if row[6] is not None else 0.0,
        "balance_rate": float(row[7]) if row[7] is not None else 0.0,
    }


@trace_service(
    description_zh="获取区域收入数据", description_en="Get finance revenue by area"
)
def get_finance_revenue_area(
    conn: Any,
    site_id: int,
    start_date: str,
    end_date: str,
    area_code: str = "all",
    *,
    etl_conn: Any = None,
) -> dict:
    """
    从 v_dws_finance_area_daily 按 area_code 聚合 revenue 板块数据。

    返回收入结构 4 项、优惠拆分 6 项、confirmed_total、order_count。
    优惠拆分恒等式：discount_total = 6 项之和。

    Requirements: 6.5, 6.6
    """
    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        cur.execute(
            """
            SELECT SUM(gross_amount)           AS total_occurrence,
                   SUM(table_fee_amount)       AS table_fee_amount,
                   SUM(goods_amount)           AS goods_amount,
                   SUM(assistant_pd_amount)    AS assistant_pd_amount,
                   SUM(assistant_cx_amount)    AS assistant_cx_amount,
                   SUM(discount_total)         AS discount_total,
                   SUM(discount_groupbuy)      AS discount_groupbuy,
                   SUM(discount_vip)           AS discount_vip,
                   SUM(discount_manual)        AS discount_manual,
                   SUM(discount_gift_card)     AS discount_gift_card,
                   SUM(discount_rounding)      AS discount_rounding,
                   SUM(discount_other)         AS discount_other,
                   SUM(confirmed_income)       AS confirmed_total,
                   SUM(order_count)            AS order_count
            FROM app.v_dws_finance_area_daily
            WHERE area_code = %s
              AND stat_date >= %s::date
              AND stat_date <= %s::date
            """,
            (area_code, start_date, end_date),
        )
        row = cur.fetchone()

    _zero = {
        "total_occurrence": 0.0,
        "table_fee_amount": 0.0,
        "goods_amount": 0.0,
        "assistant_pd_amount": 0.0,
        "assistant_cx_amount": 0.0,
        "discount_total": 0.0,
        "discount_groupbuy": 0.0,
        "discount_vip": 0.0,
        "discount_manual": 0.0,
        "discount_gift_card": 0.0,
        "discount_rounding": 0.0,
        "discount_other": 0.0,
        "confirmed_total": 0.0,
        "order_count": 0,
    }
    if not row or row[0] is None:
        return _zero

    def _f(v):
        return float(v) if v is not None else 0.0

    return {
        "total_occurrence": _f(row[0]),
        "table_fee_amount": _f(row[1]),
        "goods_amount": _f(row[2]),
        "assistant_pd_amount": _f(row[3]),
        "assistant_cx_amount": _f(row[4]),
        "discount_total": _f(row[5]),
        "discount_groupbuy": _f(row[6]),
        "discount_vip": _f(row[7]),
        "discount_manual": _f(row[8]),
        "discount_gift_card": _f(row[9]),
        "discount_rounding": _f(row[10]),
        "discount_other": _f(row[11]),
        "confirmed_total": _f(row[12]),
        "order_count": int(row[13]) if row[13] is not None else 0,
    }


@trace_service(
    description_zh="查询财务看板缓存", description_en="Get finance board cache"
)
def get_finance_board_cache(
    conn: Any,
    site_id: int,
    time_range: str,
    area_code: str,
    *,
    etl_conn: Any = None,
) -> dict | None:
    """
    从 v_dws_finance_board_cache 查询缓存。

    返回 overview 8 项 + start_date/end_date + data_fingerprint，
    或 None（缓存未命中）。

    Requirements: 6.1, 6.3
    """
    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        cur.execute(
            """
            SELECT occurrence,
                   discount,
                   discount_rate,
                   confirmed_revenue,
                   cash_in,
                   cash_out,
                   cash_balance,
                   balance_rate,
                   start_date,
                   end_date,
                   data_fingerprint
            FROM app.v_dws_finance_board_cache
            WHERE time_range = %s
              AND area_code = %s
            """,
            (time_range, area_code),
        )
        row = cur.fetchone()

    if not row:
        return None

    def _f(v):
        return float(v) if v is not None else 0.0

    return {
        "occurrence": _f(row[0]),
        "discount": _f(row[1]),
        "discount_rate": _f(row[2]),
        "confirmed_revenue": _f(row[3]),
        "cash_in": _f(row[4]),
        "cash_out": _f(row[5]),
        "cash_balance": _f(row[6]),
        "balance_rate": _f(row[7]),
        "start_date": str(row[8]) if row[8] is not None else None,
        "end_date": str(row[9]) if row[9] is not None else None,
        "data_fingerprint": row[10],
    }


@trace_service(
    description_zh="写入财务看板缓存", description_en="Set finance board cache"
)
def set_finance_board_cache(
    conn: Any,
    site_id: int,
    time_range: str,
    area_code: str,
    data: dict,
    *,
    etl_conn: Any = None,
) -> None:
    """
    写入/更新 dws.dws_finance_board_cache（实际表，不用 RLS 视图）。

    使用 ON CONFLICT (site_id, time_range, area_code) DO UPDATE 保证幂等。
    data dict 应包含 overview 8 项 + start_date/end_date + data_fingerprint。

    注意：写入用实际表而非 RLS 视图，因为 RLS 视图可能不支持 INSERT。
    site_id 通过参数显式传入 INSERT 语句。

    Requirements: 6.2
    """
    with _fdw_context(conn, site_id, etl_conn=etl_conn) as cur:
        cur.execute(
            """
            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,
                created_at, updated_at
            ) VALUES (
                %s, %s, %s,
                %s, %s,
                %s, %s, %s,
                %s, %s, %s,
                %s, %s,
                %s, NOW(),
                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()
            """,
            (
                site_id,
                time_range,
                area_code,
                data.get("start_date"),
                data.get("end_date"),
                data.get("occurrence", 0),
                data.get("discount", 0),
                data.get("discount_rate", 0),
                data.get("confirmed_revenue", 0),
                data.get("cash_in", 0),
                data.get("cash_out", 0),
                data.get("cash_balance", 0),
                data.get("balance_rate", 0),
                data.get("data_fingerprint"),
            ),
        )
