# AI_CHANGELOG # - 2026-03-20 | Prompt: RNS1.3 FDW 列名修正 | 修正 17 处列名映射（design.md 理想名 → 实际视图列名）， # gift_rows 每个 cell 改为 GiftCell dict 避免 Pydantic 校验失败， # v_dws_member_spending_power_index 降级为空列表，skill_filter 暂不生效 """ 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 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 库 + SET LOCAL app.current_site_id。 ⚠️ 不使用 zqyy_app 的 fdw_etl.* foreign table，而是直连 ETL 库 查询 app.v_* RLS 视图。原因：postgres_fdw 不传递自定义 GUC 参数 到远端连接，导致 RLS 视图的 current_setting('app.current_site_id') 在远端未设置而报错。 conn 参数保留但不用于 FDW 查询（调用方可能还需要它查 biz.* 表）。 ETL 连接在 yield 后自动关闭。 """ 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),)) yield cur etl_conn.commit() finally: etl_conn.close() def get_member_info( conn: Any, site_id: int, member_ids: list[int] ) -> 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) 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 def get_member_balance( conn: Any, site_id: int, member_ids: list[int] ) -> 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) as cur: 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(): result[row[0]] = Decimal(str(row[1])) if row[1] is not None else Decimal("0") return result def get_last_visit_days( conn: Any, site_id: int, member_ids: list[int] ) -> dict[int, int | None]: """ 批量查询客户距上次到店天数。 来源: app.v_dwd_assistant_service_log。 废单排除: is_delete = 0（RLS 视图使用 is_delete 而非 is_trash）。 时间字段: create_time（对应 design.md 中的 settle_time）。 会员字段: tenant_member_id（对应 design.md 中的 member_id）。 返回 {member_id: days_since_visit} 映射，无记录的会员不在结果中。 """ if not member_ids: return {} result: dict[int, int | None] = {} with _fdw_context(conn, site_id) as cur: cur.execute( """ SELECT tenant_member_id, CURRENT_DATE - MAX(create_time::date) AS days_since_visit FROM app.v_dwd_assistant_service_log WHERE tenant_member_id = ANY(%s) AND is_delete = 0 GROUP BY tenant_member_id """, (member_ids,), ) for row in cur.fetchall(): result[row[0]] = row[1] return result 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 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, } 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}, ...]。 """ with _fdw_context(conn, site_id) as cur: cur.execute( """ SELECT tier_id, tier_code, tier_name, tier_level, min_hours, max_hours FROM app.v_cfg_performance_tier WHERE effective_from <= CURRENT_DATE AND effective_to >= CURRENT_DATE ORDER BY tier_level """ ) 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, } for r in rows ] 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: 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 <= CURRENT_DATE AND effective_to >= CURRENT_DATE ORDER BY level_code """ ) return {row[0]: row[1] for row in cur.fetchall()} except Exception: return {} 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 real_use_seconds / 3600.0 → service_hours_raw ledger_amount → income（items_sum 口径） skill_name → course_type ⚠️ 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.real_use_seconds / 3600.0 AS service_hours_raw, sl.skill_name, sl.site_table_id, sl.ledger_amount 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 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, "service_hours_raw": float(row[7]) if row[7] is not None else 0.0, "course_type": row[8], "table_name": str(row[9]) if row[9] is not None else None, "income": float(row[10]) if row[10] is not None else 0.0, # is_estimate 不存在于视图中，默认 False "is_estimate": False, }) return records 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)。 返回按 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, sl.real_use_seconds / 3600.0 AS service_hours_raw, sl.skill_name, sl.site_table_id, sl.ledger_amount 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 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(): 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, "service_hours_raw": float(row[7]) if row[7] is not None else 0.0, "course_type": row[8], "table_name": str(row[9]) if row[9] is not None else None, "income": float(row[10]) if row[10] is not None else 0.0, "is_estimate": False, }) return records def get_consumption_60d( conn: Any, site_id: int, member_id: int ) -> Decimal | None: """ 查询客户近 60 天消费金额。 来源: app.v_dwd_assistant_service_log。 ⚠️ DWD-DOC 规则 1: 使用 ledger_amount（items_sum 口径）。 ⚠️ 废单排除: is_delete = 0。 """ with _fdw_context(conn, site_id) as cur: cur.execute( """ SELECT COALESCE(SUM(ledger_amount), 0) FROM app.v_dwd_assistant_service_log WHERE tenant_member_id = %s AND is_delete = 0 AND create_time >= (CURRENT_DATE - INTERVAL '60 days')::timestamptz """, (member_id,), ) row = cur.fetchone() return Decimal(str(row[0])) if row and row[0] is not None else None def get_relation_index( conn: Any, site_id: int, member_id: int ) -> 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) 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 def get_consumption_records( conn: Any, site_id: int, member_id: int, limit: int, offset: int ) -> list[dict]: """ 查询客户消费记录（CUST-1 consumptionRecords 用）。 来源: app.v_dwd_assistant_service_log + v_dim_assistant。 ⚠️ DWD-DOC 规则 1: totalAmount 使用 ledger_amount。 ⚠️ DWD-DOC 规则 2: coaches fee 使用 assistant_pd_money / assistant_cx_money。 ⚠️ 废单排除: is_delete = 0。 ⚠️ 正向交易: settle_type IN (1, 3)。 ⚠️ DQ-6: 助教姓名通过 v_dim_assistant 获取。 """ records: list[dict] = [] with _fdw_context(conn, site_id) as cur: cur.execute( """ SELECT sl.assistant_service_id AS id, sl.create_time AS settle_time, sl.start_use_time AS start_time, sl.last_use_time AS end_time, sl.income_seconds / 3600.0 AS service_hours, sl.ledger_amount AS total_amount, sl.skill_name AS course_type, sl.site_table_id AS table_id, sl.site_assistant_id AS assistant_id, COALESCE(da.real_name, da.nickname, '') AS assistant_name, da.level AS assistant_level, sl.table_charge_money, sl.goods_money, sl.assistant_pd_money, sl.assistant_cx_money, sl.settle_type 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 WHERE sl.tenant_member_id = %s AND sl.is_delete = 0 AND sl.settle_type IN (1, 3) ORDER BY sl.create_time DESC LIMIT %s OFFSET %s """, (member_id, 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, "course_type": row[6] or "", "table_id": row[7], "assistant_id": row[8], "assistant_name": row[9] or "", "level": row[10] or "", "table_charge_money": float(row[11]) if row[11] is not None else 0.0, "goods_money": float(row[12]) if row[12] is not None else 0.0, "assistant_pd_money": float(row[13]) if row[13] is not None else 0.0, "assistant_cx_money": float(row[14]) if row[14] is not None else 0.0, "settle_type": row[15], }) return records def get_total_service_count( conn: Any, site_id: int, member_id: int ) -> int: """ 查询客户累计服务总次数（跨所有月份）。 来源: app.v_dwd_assistant_service_log。 ⚠️ 废单排除: is_delete = 0。 """ with _fdw_context(conn, site_id) as cur: cur.execute( """ SELECT COUNT(*) FROM app.v_dwd_assistant_service_log WHERE tenant_member_id = %s AND is_delete = 0 """, (member_id,), ) row = cur.fetchone() return row[0] if row else 0 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。 返回 {service_count, total_hours, avg_hours}。 """ 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 >= (CURRENT_DATE - INTERVAL '60 days')::timestamptz """, (assistant_id, member_id), ) 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, } def get_customer_service_records( conn: Any, site_id: int, member_id: int, year: int, month: int, table: str | None, limit: int, offset: int, ) -> tuple[list[dict], int]: """ 查询客户按月服务记录（CUST-2 用）。 来源: app.v_dwd_assistant_service_log + v_dim_assistant。 ⚠️ DQ-6: 助教姓名通过 v_dim_assistant 获取。 ⚠️ DWD-DOC 规则 1: income 使用 ledger_amount。 ⚠️ 废单排除: is_delete = 0。 返回 (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 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) 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 # 分页记录查询 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, sl.real_use_seconds / 3600.0 AS service_hours_raw, sl.ledger_amount AS income, sl.skill_name AS course_type, sl.site_table_id, COALESCE(da.real_name, da.nickname, '') AS assistant_name 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 WHERE {base_where} ORDER BY sl.create_time DESC LIMIT %s OFFSET %s """, params + [limit, offset], ) for row in cur.fetchall(): records.append({ "id": str(row[0]) if row[0] else "", "create_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, "service_hours_raw": float(row[5]) if row[5] is not None else 0.0, "income": float(row[6]) if row[6] is not None else 0.0, "course_type": row[7] or "", "table_id": row[8], "assistant_name": row[9] or "", }) return records, total_count 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(real_name, nickname, '') 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": str(row[3]) if row[3] else None, # 视图中无 avatar/skills/work_years，使用默认值 "avatar": "", "skills": [], "work_years": 0.0, } 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 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 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( """ 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, mca.balance 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 app.v_dim_member_card_account mca ON sl.tenant_member_id = mca.tenant_member_id AND mca.scd2_is_current = 1 WHERE sl.site_assistant_id = %s AND sl.is_delete = 0 GROUP BY sl.tenant_member_id, dm.nickname, mca.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 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。 ⚠️ DWD-DOC 规则 1: income 使用 ledger_amount。 ⚠️ 废单排除: 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, sl.ledger_amount AS income, sl.skill_name AS course_type, sl.site_table_id AS table_id 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 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], }) return records # --------------------------------------------------------------------------- # BOARD-1 助教看板 FDW 查询 # --------------------------------------------------------------------------- def get_all_assistants( conn: Any, site_id: int, skill_filter: str = "all" ) -> list[dict]: """ 查询门店全部助教列表（BOARD-1 用）。 CHANGE 2026-03-19 | P1 修复：通过 LEFT JOIN v_dws_assistant_project_tag 获取技能标签， 支持 skill_filter 筛选（chinese/snooker/mahjong/karaoke/all）。 category_code 映射：BILLIARD→chinese, SNOOKER→snooker, MAHJONG→mahjong, KTV→karaoke。 """ # CHANGE 2026-03-19 | feiqiu-data-rules 规则 6: 等级名称从配置表动态读取 _skill_to_category = { "chinese": "BILLIARD", "snooker": "SNOOKER", "mahjong": "MAHJONG", "karaoke": "KTV", } _category_to_skill = {v: k for k, v in _skill_to_category.items()} 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 _skill_to_category: 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_to_category[skill_filter],) cur.execute( f""" SELECT da.assistant_id, COALESCE(da.real_name, da.nickname, '') 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 {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 [] skill_labels = [_category_to_skill.get(c, c) for c in skill_codes if c] records.append({ "assistant_id": row[0], "name": row[1] or "", "skill": ",".join(skill_labels) if skill_labels else "", "level": level_map.get(row[2], "") if row[2] else "", }) return records 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 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 "", } return result 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 "💙" 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-19 | P0 修复：原查询使用 v_dws_assistant_monthly_summary 的 total_ledger_amount（总消费额）冒充储值金额，口径完全错误。 改为使用 v_dws_assistant_recharge_commission（助教储值提成明细表）， 正确取 recharge_amount（储值金额）和 commission_amount（提成金额）。 返回 {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: cur.execute( """ SELECT assistant_id, COALESCE(SUM(recharge_amount), 0) AS sv_amount, COUNT(DISTINCT recharge_order_id) AS sv_customer_count, COALESCE(SUM(commission_amount), 0) AS sv_consume FROM app.v_dws_assistant_recharge_commission WHERE assistant_id = ANY(%s) AND commission_month >= %s::date AND commission_month <= %s::date GROUP BY 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) -> tuple[str, tuple]: """ 生成项目筛选 SQL 片段（用于 BOARD-2 会员维度查询）。 CHANGE 2026-03-19 | P1 修复：通过 v_dws_member_project_tag 子查询实现项目筛选。 project 参数映射：chinese→BILLIARD, snooker→SNOOKER, mahjong→MAHJONG, karaoke→KTV。 返回 (sql_fragment, params)，sql_fragment 以 AND 开头，可直接拼入 WHERE 子句。 """ _project_to_category = { "chinese": "BILLIARD", "snooker": "SNOOKER", "mahjong": "MAHJONG", "karaoke": "KTV", } if project == "all" or project not in _project_to_category: return "", () category_code = _project_to_category[project] clause = """ AND vd.member_id 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, (category_code,) 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 (计算列), visits_30d 不存在（有 visits_14d/visits_60d），用 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) 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] # 分页数据 offset = (page - 1) * page_size cur.execute( f""" SELECT wi.member_id, dm.nickname, wi.ideal_interval_days, CURRENT_DATE - wi.last_visit_time::date AS elapsed_days, (CURRENT_DATE - wi.last_visit_time::date) - COALESCE(wi.ideal_interval_days, 0) AS overdue_days, wi.visits_14d, 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 app.v_dim_member_card_account ca ON wi.member_id = ca.tenant_member_id AND ca.scd2_is_current = 1 WHERE 1=1 {proj_clause} ORDER BY wi.display_score DESC 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 "", "ideal_days": row[2] or 0, "elapsed_days": row[3] or 0, "overdue_days": row[4] or 0, "visits_30d": row[5] or 0, # 实际为 visits_14d 近似 "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[str]: """从三维分数派生潜力标签（display_score 均为 0-100 区间）。""" tags = [] threshold = 60.0 if level_score is not None and float(level_score) >= threshold: tags.append("high_level") if speed_score is not None and float(speed_score) >= threshold: tags.append("fast_growth") if stability_score is not None and float(stability_score) >= threshold: tags.append("stable") return tags 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 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 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 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]), }) return {"items": items, "total": total, "page": page, "page_size": page_size} 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) with _fdw_context(conn, site_id) as cur: cur.execute( f""" SELECT COUNT(*) FROM app.v_dim_member_card_account ca WHERE ca.scd2_is_current = 1 {proj_clause} """, proj_params, ) total = cur.fetchone()[0] offset = (page - 1) * page_size cur.execute( f""" SELECT ca.tenant_member_id AS member_id, dm.nickname, ca.balance, vd.days_since_last, vd.consume_amount_60d FROM app.v_dim_member_card_account ca LEFT JOIN app.v_dim_member dm ON ca.tenant_member_id = dm.member_id AND dm.scd2_is_current = 1 LEFT JOIN app.v_dws_member_consumption_summary vd ON ca.tenant_member_id = vd.member_id WHERE ca.scd2_is_current = 1 {proj_clause} ORDER BY ca.balance DESC 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, "last_visit_date": row[3], # days_since_last (int) "monthly_consume": float(row[4]) if row[4] is not None else 0.0, }) return {"items": items, "total": total, "page": page, "page_size": page_size} 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) 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] 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 >= CURRENT_DATE - INTERVAL '60 days' ) AS recharges_60d, COALESCE(ca.balance, 0) AS current_balance 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 app.v_dim_member_card_account ca ON ro.member_id = ca.tenant_member_id AND ca.scd2_is_current = 1 WHERE 1=1 {proj_clause} GROUP BY ro.member_id, dm.nickname, ca.balance ORDER BY MAX(ro.pay_time::date) DESC 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 "", "last_recharge_date": row[2], "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, }) return {"items": items, "total": total, "page": page, "page_size": page_size} 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 降序。 """ proj_clause, proj_params = _project_filter_clause(project) 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 1=1 {proj_clause} """, 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 >= CURRENT_DATE - INTERVAL '30 days') AS visits_30d, AVG(vd.total_consume) AS avg_spend FROM app.v_dws_member_visit_detail vd WHERE 1=1 {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.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 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 "", "last_visit_date": row[2], "visit_freq": float(row[3]) if row[3] is not None else 0.0, "ideal_days": float(row[4]) if row[4] is not None else 0.0, "visits_30d": row[5] or 0, "avg_spend": float(row[6]) if row[6] is not None else 0.0, }) return {"items": items, "total": total, "page": page, "page_size": page_size} 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 降序。 """ 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_consumption_summary cs WHERE 1=1 {proj_clause} """, proj_params, ) total = cur.fetchone()[0] offset = (page - 1) * page_size cur.execute( f""" SELECT cs.member_id, dm.nickname, cs.consume_amount_60d, cs.visit_count_60d, cs.avg_ticket_amount FROM app.v_dws_member_consumption_summary cs LEFT JOIN app.v_dim_member dm ON cs.member_id = dm.member_id AND dm.scd2_is_current = 1 WHERE 1=1 {proj_clause} ORDER BY cs.consume_amount_60d DESC 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} 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 降序。 """ 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_consumption_summary cs WHERE 1=1 {proj_clause} """, proj_params, ) total = cur.fetchone()[0] offset = (page - 1) * page_size cur.execute( f""" SELECT cs.member_id, dm.nickname, cs.visit_count_60d, cs.consume_amount_60d FROM app.v_dws_member_consumption_summary cs LEFT JOIN app.v_dim_member dm ON cs.member_id = dm.member_id AND dm.scd2_is_current = 1 WHERE 1=1 {proj_clause} ORDER BY cs.visit_count_60d DESC LIMIT %s OFFSET %s """, (*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_days": 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) 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]) -> dict[int, list[dict]]: """ 批量查询客户最近 8 周的到店次数（用于 freq60 维度柱状图）。 来源: app.v_dwd_assistant_service_log，按 ISO 周分组。 返回 {member_id: [{val: int, pct: int}, ...]}，固定 8 个元素。 """ cur.execute( """ WITH weekly AS ( SELECT tenant_member_id AS member_id, DATE_TRUNC('week', create_time::date) AS week_start, COUNT(*) AS cnt FROM app.v_dwd_assistant_service_log WHERE tenant_member_id = ANY(%s) AND is_delete = 0 AND create_time >= CURRENT_DATE - INTERVAL '56 days' GROUP BY tenant_member_id, DATE_TRUNC('week', create_time::date) ) SELECT member_id, week_start, cnt FROM weekly ORDER BY member_id, week_start """, (member_ids,), ) from collections import defaultdict raw: dict[int, dict[str, int]] = defaultdict(dict) for row in cur.fetchall(): raw[row[0]][str(row[1])] = row[2] # 生成最近 8 周的周一日期 from datetime import date, timedelta today = date.today() this_monday = today - timedelta(days=today.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 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) 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 LIMIT %s OFFSET %s ) SELECT mt.member_id, dm.nickname, mt.max_rs, mt.top_assistant_id, mt.top_rs, COALESCE(da.real_name, da.nickname, '') 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 """, (*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_heart": float(row[4]) if row[4] is not None else 0.0, }) return {"items": items, "total": total, "page": page, "page_size": page_size} 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-20 | v_dim_assistant 无 assistant_name/skill_name， # 改用 COALESCE(real_name, nickname)，cls 暂返回空 cur.execute( """ SELECT ri.member_id, COALESCE(da.real_name, da.nickname, '') 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) ORDER BY ri.member_id, ri.rs_display DESC """, (member_ids,), ) for row in cur.fetchall(): mid = row[0] if mid in result: result[mid].append({ "name": row[1] or "", "cls": "", # v_dim_assistant 无 skill_name 列 "heart_score": float(row[2]) if row[2] is not None else 0.0, }) return result # --------------------------------------------------------------------------- # BOARD-3 财务看板 FDW 查询（6 板块） # --------------------------------------------------------------------------- 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, } 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。 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 """ with _fdw_context(conn, site_id) as cur: cur.execute( """ SELECT SUM(recharge_cash) AS actual_income, SUM(first_recharge_cash) AS first_charge, SUM(renewal_cash) AS renew_charge, SUM(cash_card_balance) AS card_balance, SUM(total_card_balance) AS all_card_balance, SUM(gift_card_balance) AS gift_balance_total, SUM(gift_liquor_balance) AS gift_liquor_balance, SUM(gift_table_fee_balance) AS gift_table_fee_balance, SUM(gift_voucher_balance) AS gift_voucher_balance, 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), ) row = cur.fetchone() def _f(v): return float(v) if v is not None else 0.0 if not row or row[0] is None: return _empty_recharge_data() gift_balance = _f(row[5]) # CHANGE 2026-07-22 | Prompt: gift-card-breakdown Task 7.2 | 直接原因: gift_rows 填充真实细分数据 gift_liquor_balance = _f(row[6]) gift_table_fee_balance = _f(row[7]) gift_voucher_balance = _f(row[8]) gift_liquor_recharge = _f(row[9]) gift_table_fee_recharge = _f(row[10]) gift_voucher_recharge = _f(row[11]) # CHANGE 2026-03-20 | gift_rows 每个 cell 必须是 GiftCell dict（{"value": float}）， # 不能是裸 float，否则 Pydantic ResponseValidationError _gc = lambda v: {"value": v} return { "actual_income": _f(row[0]), "first_charge": _f(row[1]), "renew_charge": _f(row[2]), "consumed": 0.0, # 视图无消耗列，暂返回 0 "card_balance": _f(row[3]), "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": _f(row[4]), } 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, } def get_finance_revenue( conn: Any, site_id: int, start_date: str, end_date: str, area: str = "all" ) -> dict: """ BOARD-3 应计收入：收入结构 + 价格构成 + 优惠明细 + 渠道分布。 CHANGE 2026-03-19 | P0 修复： 1. structure_type='AREA' → is_sub=True（区域子行），'INCOME_TYPE' → is_sub=False（主行） 2. price_items 从 INCOME_TYPE 行提取 3. channel_items 从 v_dws_finance_daily_summary 聚合渠道支付数据 4. discount 按 discount_detail 总额均摊到 confirmed_total ⚠️ DWD-DOC 规则 2: 助教行使用 assistant_pd_money（陪打）+ assistant_cx_money（超休）。 """ with _fdw_context(conn, site_id) as cur: # 收入结构主表（按 structure_type + category_code 分类） cur.execute( """ SELECT structure_type, category_code, category_name, SUM(income_amount) AS income_amount, SUM(order_count) AS order_count, SUM(duration_minutes) AS duration_minutes FROM app.v_dws_finance_income_structure WHERE stat_date >= %s::date AND stat_date <= %s::date GROUP BY structure_type, category_code, category_name ORDER BY structure_type, category_code """, (start_date, end_date), ) structure_rows = [] price_items = [] total_income = 0.0 for row in cur.fetchall(): s_type = row[0] or "" amt = float(row[3]) if row[3] is not None else 0.0 is_sub = (s_type == "AREA") # 只有主行（INCOME_TYPE）计入总收入，避免 AREA 子行重复计算 if not is_sub: total_income += amt price_items.append({ "label": row[2] or row[1] or "", "amount": amt, }) structure_rows.append({ "id": row[1] or "", "name": row[2] or "", "desc": None, "is_sub": is_sub, "amount": amt, "discount": 0.0, "booked": amt, }) # 优惠明细 cur.execute( """ SELECT discount_type_code, discount_type_name, SUM(discount_amount) AS amount FROM app.v_dws_finance_discount_detail WHERE stat_date >= %s::date AND stat_date <= %s::date GROUP BY discount_type_code, discount_type_name ORDER BY discount_type_code """, (start_date, end_date), ) discount_items = [] total_discount = 0.0 for row in cur.fetchall(): amt = float(row[2]) if row[2] is not None else 0.0 total_discount += amt discount_items.append({ "label": row[1] or row[0] or "", "amount": amt, }) # 渠道分布（从 v_dws_finance_daily_summary 聚合） 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, COALESCE(SUM(platform_settlement_amount), 0) AS platform_pay FROM app.v_dws_finance_daily_summary WHERE stat_date >= %s::date AND stat_date <= %s::date """, (start_date, end_date), ) ch = cur.fetchone() channel_items = [] if ch: for label, val in [ ("现金/扫码", ch[0]), ("团购核销", ch[1]), ("储值卡消费", ch[2]), ("赠送卡消费", ch[3]), ("平台结算", ch[4]), ]: channel_items.append({ "label": label, "amount": float(val) if val is not None else 0.0, }) confirmed_total = total_income - abs(total_discount) return { "structure_rows": structure_rows, "price_items": price_items, "total_occurrence": total_income, "discount_items": discount_items, "confirmed_total": confirmed_total, "channel_items": channel_items, } 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 (充值现金流入) ⚠️ DWD-DOC 规则 7: platform_settlement_amount 和 groupbuy_pay_amount 互斥。 """ with _fdw_context(conn, site_id) as cur: cur.execute( """ SELECT SUM(cash_pay_amount) AS consume_cash, SUM(groupbuy_pay_amount) AS consume_groupbuy, SUM(card_consume_total) AS consume_card, 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[0] is None: return { "consume_items": [ {"label": "现金收款", "amount": 0.0}, {"label": "团购核销", "amount": 0.0}, {"label": "储值卡消费", "amount": 0.0}, ], "recharge_items": [{"label": "充值收款", "amount": 0.0}], "total": 0.0, } return { "consume_items": [ {"label": "现金收款", "amount": _f(row[0])}, {"label": "团购核销", "amount": _f(row[1])}, {"label": "储值卡消费", "amount": _f(row[2])}, ], "recharge_items": [{"label": "充值收款", "amount": _f(row[3])}], "total": _f(row[4]), } 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) ⚠️ 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 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 分组聚合。 来源: app.v_dws_assistant_salary_calc。 返回 basic（基础课/陪打）+ incentive（激励课/超休）两个子表。 """ with _fdw_context(conn, site_id) as cur: cur.execute( """ SELECT assistant_level_name AS level, SUM(base_income) AS pd_pay, SUM(bonus_income) AS cx_pay, SUM(effective_hours) AS total_hours, COUNT(DISTINCT assistant_id) AS coach_count FROM app.v_dws_assistant_salary_calc WHERE salary_month >= %s::date AND salary_month <= %s::date GROUP BY assistant_level_name ORDER BY assistant_level_name """, (start_date, end_date), ) rows = cur.fetchall() basic_rows = [] incentive_rows = [] total_pd = 0.0 total_cx = 0.0 total_hours = 0.0 for row in rows: level = row[0] or "" pd_pay = float(row[1]) if row[1] is not None else 0.0 cx_pay = float(row[2]) if row[2] is not None else 0.0 hours = float(row[3]) if row[3] is not None else 0.0 count = row[4] or 0 total_pd += pd_pay total_cx += cx_pay total_hours += hours # 基础课（陪打） pd_hourly = pd_pay / hours if hours > 0 else 0.0 basic_rows.append({ "level": level, "pay": pd_pay, "share": 0.0, # 服务层计算占比 "hourly": round(pd_hourly, 2), }) # 激励课（超休） cx_hourly = cx_pay / hours if hours > 0 else 0.0 incentive_rows.append({ "level": level, "pay": cx_pay, "share": 0.0, "hourly": round(cx_hourly, 2), }) # 计算占比 for r in basic_rows: r["share"] = round(r["pay"] / total_pd * 100, 1) if total_pd > 0 else 0.0 for r in incentive_rows: r["share"] = round(r["pay"] / total_cx * 100, 1) if total_cx > 0 else 0.0 avg_pd_hourly = round(total_pd / total_hours, 2) if total_hours > 0 else 0.0 avg_cx_hourly = round(total_cx / total_hours, 2) if total_hours > 0 else 0.0 return { "basic": { "total_pay": total_pd, "total_share": 100.0 if total_pd > 0 else 0.0, "avg_hourly": avg_pd_hourly, "rows": basic_rows, }, "incentive": { "total_pay": total_cx, "total_share": 100.0 if total_cx > 0 else 0.0, "avg_hourly": avg_cx_hourly, "rows": incentive_rows, }, } def get_skill_types(conn: Any, site_id: int) -> list[dict]: """ CONFIG-1: 查询技能类型配置。 来源: ETL cfg 表（app.v_cfg_skill_type 或类似配置视图）。 查询失败时由调用方降级返回空数组。 """ with _fdw_context(conn, site_id) as cur: cur.execute( """ SELECT skill_key, skill_label, emoji, css_cls FROM app.v_cfg_skill_type ORDER BY sort_order """ ) items = [] for row in cur.fetchall(): items.append({ "key": row[0] or "", "label": row[1] or "", "emoji": row[2] or "", "cls": row[3] or "", }) return items