Neo-ZQYY/scripts/ops/gen_rls_business_date_migration.py

# -*- coding: utf-8 -*-
"""
从测试库 / 真实库实时读取目标视图的定义（pg_get_viewdef），
为每个视图生成 ``CREATE OR REPLACE VIEW`` 块，在 WHERE 末尾追加业务日上界。

数据库实际列签名可能比 ``schemas/app.sql`` 文件中的快照新（增列），
因此本脚本走 ``pg_get_viewdef`` 兜底，确保 ``CREATE OR REPLACE`` 不会
因为列签名漂移而失败。

输出 SQL 文件：db/etl_feiqiu/migrations/20260502__rls_views_business_date_upper_bound.sql

环境变量：``TEST_PG_DSN`` 或 ``PG_DSN``。
"""

from __future__ import annotations

import os
import re
from pathlib import Path

import psycopg2
from dotenv import load_dotenv

ROOT = Path(__file__).resolve().parents[2]
OUT = ROOT / "db" / "etl_feiqiu" / "migrations" / "20260502__rls_views_business_date_upper_bound.sql"

# 需要改造的视图 → 业务日上界条件（None 表示不强制裁剪，仅占位）
VIEWS_WITH_BD: dict[str, str] = {
    # ── 财务事实 ─────────────────────────────────────────────
    "app.v_dws_finance_area_daily": "stat_date <= app.business_date_now()",
    "app.v_dws_finance_daily_summary": "stat_date <= app.business_date_now()",
    "app.v_dws_finance_discount_detail": "stat_date <= app.business_date_now()",
    "app.v_dws_finance_expense_summary": "expense_month <= date_trunc('month', app.business_date_now())::date",
    "app.v_dws_finance_income_structure": "stat_date <= app.business_date_now()",
    "app.v_dws_finance_recharge_summary": "stat_date <= app.business_date_now()",
    # ── 助教 / 任务事实 ─────────────────────────────────────
    "app.v_assistant_daily": "stat_date <= app.business_date_now()",
    "app.v_dws_assistant_daily_detail": "stat_date <= app.business_date_now()",
    "app.v_dws_assistant_finance_analysis": "stat_date <= app.business_date_now()",
    "app.v_dws_assistant_monthly_summary": "stat_month <= date_trunc('month', app.business_date_now())::date",
    "app.v_dws_assistant_salary_calc": "salary_month <= date_trunc('month', app.business_date_now())::date",
    # ── 客户事实 ────────────────────────────────────────────
    "app.v_dws_member_consumption_summary": "stat_date <= app.business_date_now()",
    "app.v_dws_member_visit_detail": "visit_date <= app.business_date_now()",
    "app.v_dws_member_winback_index": "COALESCE(last_visit_time::date, '0001-01-01'::date) <= app.business_date_now()",
    # ── DWD 事实 ────────────────────────────────────────────
    "app.v_dwd_settlement_head": "COALESCE(create_time::date, '0001-01-01'::date) <= app.business_date_now()",
    "app.v_dwd_assistant_service_log": "COALESCE(create_time::date, '0001-01-01'::date) <= app.business_date_now()",
    "app.v_dwd_recharge_order": "COALESCE(pay_time::date, '0001-01-01'::date) <= app.business_date_now()",
    "app.v_dwd_store_goods_sale": "COALESCE(create_time::date, '0001-01-01'::date) <= app.business_date_now()",
    "app.v_dwd_table_fee_log": "COALESCE(create_time::date, '0001-01-01'::date) <= app.business_date_now()",
    # ── DIM SCD2 / 配置维度 ────────────────────────────────
    "app.v_cfg_assistant_level_price": "effective_from <= app.business_date_now() AND effective_to >= app.business_date_now()",
    "app.v_cfg_performance_tier": "effective_from <= app.business_date_now() AND effective_to >= app.business_date_now()",
    "app.v_cfg_bonus_rules": "effective_from <= app.business_date_now() AND effective_to >= app.business_date_now()",
    "app.v_cfg_index_parameters": "effective_from <= app.business_date_now() AND effective_to >= app.business_date_now()",
    # ── DWS 业务事实 / 汇总 (补 18 个) ───────────────────
    "app.v_dws_assistant_customer_stats": "stat_date <= app.business_date_now()",
    "app.v_dws_assistant_order_contribution": "stat_date <= app.business_date_now()",
    "app.v_dws_assistant_project_tag": "computed_at::date <= app.business_date_now()",
    "app.v_dws_assistant_recharge_commission": "commission_month <= date_trunc('month', app.business_date_now())::date",
    "app.v_dws_coach_area_hours": "stat_month <= date_trunc('month', app.business_date_now())::date",
    "app.v_dws_finance_board_cache": "computed_at::date <= app.business_date_now()",
    "app.v_dws_member_assistant_intimacy": "calc_time::date <= app.business_date_now()",
    "app.v_dws_member_assistant_relation_index": "COALESCE(stat_date, calc_time::date) <= app.business_date_now()",
    "app.v_dws_member_newconv_index": "stat_date <= app.business_date_now()",
    "app.v_dws_member_project_tag": "computed_at::date <= app.business_date_now()",
    "app.v_dws_member_spending_power_index": "calc_time::date <= app.business_date_now()",
    "app.v_dws_order_summary": "order_date <= app.business_date_now()",
    "app.v_dws_platform_settlement": "settlement_date <= app.business_date_now()",
    "app.v_finance_daily": "stat_date <= app.business_date_now()",
    "app.v_member_consumption": "stat_date <= app.business_date_now()",
    "app.v_order_summary": "order_date <= app.business_date_now()",
}

# 跳过原因记录（用于审计文档）
VIEWS_SKIPPED: dict[str, str] = {
    "app.v_assistant": "无日期列；纯 dim 当前快照",
    "app.v_cfg_area_category": "无日期列；纯静态配置",
    "app.v_member": "无日期列；纯当前会员快照",
    "app.v_site": "无日期列；纯门店元数据",
    # SCD2 维度：保留 scd2_is_current=1 语义不动；
    # 想要"sandbox 当时的维度状态"需把过滤改为 scd2_start_time <= bd AND (scd2_end_time > bd OR is null)，
    # 但这会让一行"当前生效"变成多行（其中一行是当时生效），影响 JOIN 与上层调用方。先保留现状，后续视需求评估。
    "app.v_dim_assistant": "SCD2 dim：scd2_is_current=1 当前快照（保留）",
    "app.v_dim_member": "SCD2 dim：scd2_is_current=1 当前快照（保留）",
    "app.v_dim_member_card_account": "SCD2 dim：scd2_is_current=1 当前快照（保留）",
    "app.v_dim_staff": "SCD2 dim：scd2_is_current=1 当前快照（保留）",
    "app.v_dim_staff_ex": "SCD2 dim：scd2_is_current=1 当前快照（保留）",
    "app.v_dim_table": "SCD2 dim：scd2_is_current=1 当前快照（保留）",
}

def fetch_view_def(cur, schema: str, view: str) -> str:
    """返回 ``pg_get_viewdef`` 的视图体（去掉末尾分号），格式化为多行。"""
    cur.execute(
        "SELECT pg_get_viewdef(%s::regclass, true)", (f"{schema}.{view}",)
    )
    row = cur.fetchone()
    if not row or not row[0]:
        raise RuntimeError(f"未取到视图定义: {schema}.{view}")
    body = row[0].rstrip().rstrip(";").rstrip()
    return body


def add_business_date_clause(view_body: str, predicate: str) -> str:
    """在 WHERE 末尾追加 AND <predicate>。

    若视图体含 ``ORDER BY``，将谓词插入到 ORDER BY 前；否则末尾追加。
    """
    body = view_body.rstrip()
    upper = body.upper()
    order_by_idx = upper.rfind("ORDER BY")
    if order_by_idx != -1:
        head = body[:order_by_idx].rstrip()
        tail = body[order_by_idx:]
        if "WHERE" in head.upper():
            return f"{head}\n    AND {predicate}\n  {tail}"
        return f"{body}\n  WHERE {predicate}"
    if "WHERE" in body.upper():
        return f"{body}\n    AND {predicate}"
    return f"{body}\n  WHERE {predicate}"


def main() -> None:
    load_dotenv()
    dsn = os.environ.get("TEST_PG_DSN") or os.environ.get("PG_DSN")
    if not dsn:
        raise SystemExit("请配置 TEST_PG_DSN 或 PG_DSN")

    out_lines: list[str] = []
    out_lines.append(
        "-- =============================================================================\n"
        "-- ETL 库（etl_feiqiu）/ app schema —— RLS 视图业务日上界裁剪\n"
        "-- 由 scripts/ops/gen_rls_business_date_migration.py 自动生成。\n"
        "-- 沙箱模式下，业务读取层只看到 sandbox_date 及之前的数据。\n"
        "-- =============================================================================\n"
    )
    out_lines.append("BEGIN;\n")

    out_lines.append(
        "-- helper：业务日 GUC 读取，缺省回退当前真实日期\n"
        "CREATE OR REPLACE FUNCTION app.business_date_now()\n"
        "RETURNS date\n"
        "LANGUAGE sql\n"
        "STABLE\n"
        "AS $$\n"
        "  SELECT COALESCE(\n"
        "    NULLIF(current_setting('app.current_business_date', true), '')::date,\n"
        "    CURRENT_DATE\n"
        "  );\n"
        "$$;\n"
        "COMMENT ON FUNCTION app.business_date_now() IS\n"
        "'返回当前业务日（GUC app.current_business_date），未设置时回退 CURRENT_DATE。';\n"
    )

    conn = psycopg2.connect(dsn)
    try:
        with conn.cursor() as cur:
            for view_name, predicate in VIEWS_WITH_BD.items():
                schema, view = view_name.split(".", 1)
                body = fetch_view_def(cur, schema, view)
                body_with_bd = add_business_date_clause(body, predicate)
                out_lines.append(f"\n-- {view_name}：加业务日上界 → {predicate}")
                out_lines.append(f"CREATE OR REPLACE VIEW {view_name} AS")
                out_lines.append(body_with_bd + ";\n")
    finally:
        conn.close()

    out_lines.append("\nCOMMIT;\n")
    out_lines.append(
        "\n-- 回滚：DROP FUNCTION app.business_date_now() CASCADE;\n"
        "-- 然后重新执行 db/etl_feiqiu/schemas/app.sql 即可恢复 live 行为\n"
    )

    OUT.write_text("\n".join(out_lines), encoding="utf-8")
    print(f"OK: 写入 {OUT.relative_to(ROOT)}, 共 {len(VIEWS_WITH_BD)} 个视图")


if __name__ == "__main__":
    main()