Neo-ZQYY/apps/backend/app/ai/data_fetchers/member_data.py

"""客户消费数据获取模块（应用 3/6/7 共用）。

从 ETL 库 app.v_* RLS 视图获取客户近 N 个月消费数据，从业务库获取备注。
金额口径统一使用拆分字段（table_charge_money + assistant_pd/cx_money + goods_money），禁止 consume_money。
会员信息通过 member_id JOIN v_dim_member (scd2_is_current=1) 获取。
"""
# CHANGE 2026-03-23 | Prompt: FDW 迁移——fdw_etl.* → app.* 直连 ETL 库
# intent: 将所有 fdw_etl.* 外部表引用改为 app.v_* RLS 视图（直连 ETL 库），列名同步修正
# 连接方式不变（get_etl_readonly_connection），仅改 SQL 表名和列名

from __future__ import annotations

import asyncio
import logging
from datetime import date, datetime
from decimal import Decimal
from functools import partial
from typing import Any

from app.database import get_connection, get_etl_readonly_connection

logger = logging.getLogger(__name__)

# 消费记录最大返回数
MAX_CONSUMPTION_RECORDS = 100
# 备注最大返回数
MAX_NOTES = 50
# 备注单条最大字符数
MAX_NOTE_LENGTH = 500
# FDW 查询超时（秒）
FDW_QUERY_TIMEOUT_SEC = 5


async def fetch_member_consumption_data(
    site_id: int,
    member_id: int,
    months: int = 3,
) -> dict[str, Any]:
    """获取客户近 N 个月消费数据。

    返回结构对应 NS2 设计文档中 main_data：
    - consumption_records: 消费记录列表（最多 100 条，settle_date DESC）
    - member_cards: 会员卡明细列表
    - card_balance_total: 储值卡余额合计
    - stored_value_balance_total: 储值余额合计
    - expected_visit_date: 预计到店日期
    - days_since_last_visit: 距上次到店天数
    - member_nickname: 会员昵称

    Raises:
        TimeoutError: FDW 查询超时（>5s）
        ConnectionError: FDW 连接失败
    """
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        None,
        partial(_fetch_member_consumption_data_sync, site_id, member_id, months),
    )


def _fetch_member_consumption_data_sync(
    site_id: int,
    member_id: int,
    months: int,
) -> dict[str, Any]:
    """同步实现：在单个 FDW 连接上串行执行多个查询。

    CHANGE 2026-05-02 | 所有窗口查询都按业务日上界裁剪，
    sandbox 模式下不再读取 sandbox_date 之后的真实消费 / 到店。
    """
    from app.services.runtime_context import as_runtime_today_param

    conn = None
    try:
        conn = get_etl_readonly_connection(site_id)
        ref_date = as_runtime_today_param(site_id)
        # RLS 隔离 + 语句超时（get_etl_readonly_connection 的 SET LOCAL 在 commit 后失效，
        # 需在查询事务中重新设置）
        # CHANGE 2026-05-02 | 同时下发 app.current_business_date，供 RLS 视图业务日上界裁剪
        with conn.cursor() as cur:
            cur.execute(
                "SET LOCAL app.current_site_id = %s", (str(site_id),)
            )
            cur.execute(
                "SET LOCAL app.current_business_date = %s", (ref_date.isoformat(),)
            )
            cur.execute(
                "SET LOCAL statement_timeout = %s",
                (f"{FDW_QUERY_TIMEOUT_SEC * 1000}",),  # 毫秒
            )

        # 1. 会员昵称
        nickname = _query_member_nickname(conn, member_id)

        # 2. 消费记录（台桌结账 + 商城订单）
        records, total_count = _query_consumption_records(conn, member_id, months, ref_date)

        # 3. 会员卡明细
        cards = _query_member_cards(conn, member_id)

        # 4. 余额汇总
        balance_info = _query_balance_summary(conn, member_id)

        # 5. 到店数据
        visit_info = _query_visit_info(conn, member_id, ref_date)

        result: dict[str, Any] = {
            "member_nickname": nickname,
            "consumption_records": records,
            "member_cards": cards,
            "card_balance_total": balance_info.get("card_balance_total", Decimal("0")),
            "stored_value_balance_total": balance_info.get(
                "stored_value_balance_total", Decimal("0")
            ),
            "expected_visit_date": visit_info.get("expected_visit_date"),
            "days_since_last_visit": visit_info.get("days_since_last_visit"),
        }
        if total_count > MAX_CONSUMPTION_RECORDS:
            result["truncated"] = True
            result["total_count"] = total_count

        conn.commit()
        return result

    except Exception as e:
        # psycopg2 超时异常包含 "statement timeout"
        err_msg = str(e).lower()
        if "statement timeout" in err_msg or "timeout" in err_msg:
            raise TimeoutError(
                f"FDW 查询超时（>{FDW_QUERY_TIMEOUT_SEC}s）: member_id={member_id}"
            ) from e
        if "connection" in err_msg or "connect" in err_msg:
            raise ConnectionError(
                f"FDW 连接失败: member_id={member_id}, error={e}"
            ) from e
        raise
    finally:
        if conn:
            conn.close()


def _query_member_nickname(conn: Any, member_id: int) -> str:
    """从 app.v_dim_member 获取会员昵称（scd2_is_current=1）。"""
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT nickname
            FROM app.v_dim_member
            WHERE member_id = %s AND scd2_is_current = 1
            LIMIT 1
            """,
            (member_id,),
        )
        row = cur.fetchone()
        return row[0] if row and row[0] else ""


def _query_consumption_records(
    conn: Any, member_id: int, months: int, ref_date: date
) -> tuple[list[dict], int]:
    """从 app.v_dwd_settlement_head + app.v_dwd_table_fee_log 获取消费记录。

    仅包含正向交易（settle_type IN (1, 3)）。
    ⚠️ 费用拆分字段（table_charge_money, assistant_pd/cx_money）在 settlement_head 上。
    ⚠️ table_fee_log 提供台桌时长（real_table_use_seconds）和桌台ID（site_table_id）。
    ⚠️ 列名映射: settle_date→create_time, settle_id→order_settle_id, sale_amount→ledger_amount。
    CHANGE 2026-05-02 | 用 ref_date（业务日）替代 CURRENT_DATE，沙箱不读「未来」消费。
    返回 (records, total_count)。
    """
    with conn.cursor() as cur:
        # 先查总数
        cur.execute(
            """
            SELECT COUNT(*)
            FROM app.v_dwd_settlement_head sh
            WHERE sh.member_id = %s
              AND sh.settle_type IN (1, 3)
              AND sh.create_time >= (%s::date - (INTERVAL '1 month' * %s))
              AND sh.create_time <  (%s::date + INTERVAL '1 day')
            """,
            (member_id, ref_date, months, ref_date),
        )
        total_count = cur.fetchone()[0]

        # 查询消费记录（限制 100 条）
        # table_charge_money/assistant_pd_money/assistant_cx_money 直接从 settlement_head 取
        # 台桌信息从 table_fee_log 取（site_table_id, real_table_use_seconds）
        # 商品金额从 store_goods_sale 聚合
        # 助教姓名从 service_log JOIN dim_assistant 获取
        cur.execute(
            """
            SELECT
                sh.create_time AS settle_date,
                sh.settle_type,
                sh.table_charge_money + sh.assistant_pd_money + sh.assistant_cx_money
                    + COALESCE(sg.goods_money, 0) AS items_sum,
                COALESCE(sh.table_charge_money, 0) AS table_charge_money,
                COALESCE(sh.assistant_pd_money, 0) AS assistant_pd_money,
                COALESCE(sh.assistant_cx_money, 0) AS assistant_cx_money,
                COALESCE(sg.goods_money, 0) AS goods_money,
                tfl.site_table_id AS room_name,
                COALESCE(tfl.real_table_use_seconds / 60, 0) AS duration_minutes,
                coaches.assistant_names
            FROM app.v_dwd_settlement_head sh
            LEFT JOIN app.v_dwd_table_fee_log tfl
                ON sh.order_settle_id = tfl.order_settle_id
            LEFT JOIN LATERAL (
                SELECT SUM(sgs.ledger_amount) AS goods_money
                FROM app.v_dwd_store_goods_sale sgs
                WHERE sgs.order_settle_id = sh.order_settle_id
            ) sg ON true
            LEFT JOIN LATERAL (
                SELECT string_agg(DISTINCT COALESCE(da.nickname, da.real_name, ''), ', ')
                    AS assistant_names
                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.order_settle_id = sh.order_settle_id
                  AND sl.is_delete = 0
            ) coaches ON true
            WHERE sh.member_id = %s
              AND sh.settle_type IN (1, 3)
              AND sh.create_time >= (%s::date - (INTERVAL '1 month' * %s))
              AND sh.create_time <  (%s::date + INTERVAL '1 day')
            ORDER BY sh.create_time DESC
            LIMIT %s
            """,
            (member_id, ref_date, months, ref_date, MAX_CONSUMPTION_RECORDS),
        )
        columns = [desc[0] for desc in cur.description]
        rows = cur.fetchall()

    records = []
    for row in rows:
        record = {}
        for col, val in zip(columns, row):
            if isinstance(val, (date, datetime)):
                record[col] = val.isoformat()
            elif isinstance(val, Decimal):
                record[col] = float(val)
            else:
                record[col] = val
        # assistant_names: 确保是列表
        names = record.get("assistant_names")
        if names and isinstance(names, str):
            record["assistant_names"] = [n.strip() for n in names.split(",") if n.strip()]
        elif not names:
            record["assistant_names"] = []
        records.append(record)

    return records, total_count


def _query_member_cards(conn: Any, member_id: int) -> list[dict]:
    """从 app.v_dim_member_card_account 获取会员卡明细。
    ⚠️ 列名映射: member_id→tenant_member_id, gift_balance 不存在（用 balance - principal_balance 近似）。
    """
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT member_card_type_name AS card_type,
                   COALESCE(balance, 0) AS balance,
                   COALESCE(balance, 0) - COALESCE(principal_balance, 0) AS gift_balance
            FROM app.v_dim_member_card_account
            WHERE tenant_member_id = %s AND scd2_is_current = 1
            """,
            (member_id,),
        )
        rows = cur.fetchall()

    return [
        {
            "card_type": row[0] or "",
            "balance": float(row[1]) if row[1] else 0.0,
            "gift_balance": float(row[2]) if row[2] else 0.0,
        }
        for row in rows
    ]


def _query_balance_summary(conn: Any, member_id: int) -> dict:
    """从 app.v_dws_member_consumption_summary 获取余额汇总。
    ⚠️ 列名映射: recharge_card_amount→cash_card_balance, balance_amount→total_card_balance。
    """
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT
                COALESCE(cash_card_balance, 0) AS card_balance_total,
                COALESCE(total_card_balance, 0) AS stored_value_balance_total
            FROM app.v_dws_member_consumption_summary
            WHERE member_id = %s
            LIMIT 1
            """,
            (member_id,),
        )
        row = cur.fetchone()

    if not row:
        return {
            "card_balance_total": Decimal("0"),
            "stored_value_balance_total": Decimal("0"),
        }
    return {
        "card_balance_total": row[0],
        "stored_value_balance_total": row[1],
    }


def _query_visit_info(conn: Any, member_id: int, ref_date: date) -> dict:
    """从 app.v_dws_member_visit_detail 获取到店数据，推算预计到店日期。
    ⚠️ 列名映射: last_visit_date→MAX(visit_date), avg_visit_interval_days 需从明细计算。
    CHANGE 2026-05-02 | 仅取 ref_date 及之前的到店明细，days_since 按 ref_date 计算。
    """
    with conn.cursor() as cur:
        # 获取最近到店日期和平均到店间隔
        cur.execute(
            """
            WITH visits AS (
                SELECT visit_date,
                       LAG(visit_date) OVER (ORDER BY visit_date) AS prev_visit
                FROM app.v_dws_member_visit_detail
                WHERE member_id = %s
                  AND visit_date <= %s
            )
            SELECT
                MAX(visit_date) AS last_visit_date,
                AVG(visit_date - prev_visit) AS avg_visit_interval_days
            FROM visits
            WHERE prev_visit IS NOT NULL
            """,
            (member_id, ref_date),
        )
        row = cur.fetchone()

    if not row or not row[0]:
        return {"expected_visit_date": None, "days_since_last_visit": None}

    last_visit = row[0]
    avg_interval = row[1]
    days_since = (ref_date - last_visit).days if isinstance(last_visit, date) else None

    expected = None
    if avg_interval and last_visit:
        from datetime import timedelta
        expected_date = last_visit + timedelta(days=int(avg_interval))
        expected = expected_date.isoformat()

    return {
        "expected_visit_date": expected,
        "days_since_last_visit": days_since,
    }


async def fetch_member_notes(
    site_id: int,
    member_id: int,
    limit: int = MAX_NOTES,
) -> list[dict]:
    """获取客户的全部备注（按 created_at DESC，最多 limit 条）。

    从业务库 biz.notes 查询。
    单条备注内容截断 500 字符，超出附加"…（已截断）"。
    """
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        None,
        partial(_fetch_member_notes_sync, site_id, member_id, limit),
    )


def _fetch_member_notes_sync(
    site_id: int,
    member_id: int,
    limit: int,
) -> list[dict]:
    """同步实现：从业务库查询备注。"""
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT
                    n.content,
                    u.nickname AS recorded_by,
                    n.created_at
                FROM biz.notes n
                LEFT JOIN biz.coach_tasks ct ON ct.id = n.task_id
                LEFT JOIN public.users u ON u.id = n.user_id
                WHERE n.target_id = %s AND n.site_id = %s
                ORDER BY n.created_at DESC
                LIMIT %s
                """,
                (member_id, site_id, limit),
            )
            rows = cur.fetchall()

        notes = []
        for row in rows:
            content = row[0] or ""
            recorded_by = row[1] or ""
            created_at = row[2]

            # 截断处理
            if len(content) > MAX_NOTE_LENGTH:
                content = content[:MAX_NOTE_LENGTH] + "…（已截断）"

            notes.append({
                "recorded_by": recorded_by,
                "content": content,
                "created_at": created_at.isoformat() if isinstance(created_at, (date, datetime)) else str(created_at) if created_at else "",
            })

        return notes
    finally:
        conn.close()