# -*- coding: utf-8 -*- """沙箱端到端验证:站在小程序 / admin-web 端视角, 切到 sandbox 后调用后端关键 service / fdw_queries 函数, 断言返回的最大日期 / 业务月份不超过 sandbox_date。 运行方式(在仓库根 / 已有 .env):: python tools/db/verify_sandbox_end_to_end.py [--sandbox-date 2025-09-01] [--site-id ] [--keep-sandbox] 输出 markdown 格式报告 + 控制台 PASS/FAIL 摘要。 重要:本脚本会**临时**修改 ``biz.site_runtime_context``, 脚本结束时自动还原(除非 --keep-sandbox)。 请只在 ``test_zqyy_app`` / ``test_etl_feiqiu`` 这种测试库使用。 """ from __future__ import annotations import argparse import os import sys from datetime import date, datetime, timedelta from pathlib import Path from typing import Any # 让 backend 包可被 import ROOT = Path(__file__).resolve().parents[2] BACKEND = ROOT / "apps" / "backend" sys.path.insert(0, str(BACKEND)) import psycopg2 # noqa: E402 from dotenv import load_dotenv # noqa: E402 load_dotenv(ROOT / ".env", override=False) # 强制走测试库 os.environ["APP_DB_DSN"] = os.environ.get("TEST_APP_DB_DSN") or os.environ["APP_DB_DSN"] os.environ["PG_DSN"] = os.environ.get("TEST_PG_DSN") or os.environ["PG_DSN"] # backend.config 从 ETL_DB_HOST/PORT/... 读取,缺省走 DB_HOST=localhost; # 这里从 PG_DSN(ETL)和 APP_DB_DSN(业务)解析出真实地址,确保 backend 内部连接也走测试库 import psycopg2.extensions # noqa: E402 _etl = psycopg2.extensions.parse_dsn(os.environ["PG_DSN"]) _app = psycopg2.extensions.parse_dsn(os.environ["APP_DB_DSN"]) os.environ.setdefault("DB_HOST", _app.get("host", "localhost")) os.environ.setdefault("DB_PORT", str(_app.get("port", "5432"))) os.environ.setdefault("DB_USER", _app.get("user", "")) os.environ.setdefault("DB_PASSWORD", _app.get("password", "")) os.environ.setdefault("ETL_DB_HOST", _etl.get("host", "localhost")) os.environ.setdefault("ETL_DB_PORT", str(_etl.get("port", "5432"))) os.environ.setdefault("ETL_DB_USER", _etl.get("user", "")) os.environ.setdefault("ETL_DB_PASSWORD", _etl.get("password", "")) os.environ.setdefault("ETL_DB_NAME", _etl.get("dbname", "test_etl_feiqiu")) def _date_of(value: Any) -> date | None: if value is None: return None if isinstance(value, datetime): return value.date() if isinstance(value, date): return value try: return datetime.fromisoformat(str(value)).date() except Exception: return None def _le(actual: date | None, bound: date) -> str: if actual is None: return "OK (None)" return "PASS" if actual <= bound else f"FAIL (>{bound})" # --------------------------------------------------------------------------- # RuntimeContext 切换 # --------------------------------------------------------------------------- def switch_runtime(site_id: int, mode: str, sandbox_date: date | None = None) -> None: """切换业务库 runtime_context(test_zqyy_app)。""" from app.database import get_connection from app.services.runtime_context import new_sandbox_instance_id # live 模式下 site_runtime_context 强制 sandbox_instance_id IS NULL(CHECK 约束); # 'live' 字符串只是后端 task_runtime_filter 对未配置时的兜底,不能直接写表 instance_id = None if mode == "live" else new_sandbox_instance_id() sandbox_date = None if mode == "live" else sandbox_date conn = get_connection() try: with conn.cursor() as cur: cur.execute( """ INSERT INTO biz.site_runtime_context (site_id, mode, sandbox_date, sandbox_instance_id, ai_mode, status, updated_at, updated_by) VALUES (%s, %s, %s, %s, 'live', 'active', NOW(), NULL) ON CONFLICT (site_id) DO UPDATE SET mode = EXCLUDED.mode, sandbox_date = EXCLUDED.sandbox_date, sandbox_instance_id = EXCLUDED.sandbox_instance_id, updated_at = NOW(), updated_by = NULL """, (site_id, mode, sandbox_date, instance_id), ) conn.commit() finally: conn.close() def pick_test_site(min_settle_count: int = 100) -> int: """找一个 etl_feiqiu 数据量较多的测试 site_id。""" from app.database import get_etl_readonly_connection # etl_feiqiu 直查 candidate_sites: list[tuple[int, int]] = [] pg_dsn = os.environ["PG_DSN"] conn = psycopg2.connect(pg_dsn) try: with conn.cursor() as cur: cur.execute( """ SELECT site_id, COUNT(*) AS cnt FROM dws.dws_finance_area_daily GROUP BY site_id ORDER BY cnt DESC LIMIT 5 """ ) candidate_sites = list(cur.fetchall()) finally: conn.close() if not candidate_sites: raise SystemExit("etl_feiqiu 测试库无 dws_finance_area_daily 数据") site_id = candidate_sites[0][0] # 确保该 site 在 zqyy_app 也有 biz.sites 行 from app.database import get_connection conn = get_connection() try: with conn.cursor() as cur: cur.execute("SELECT 1 FROM biz.sites WHERE site_id = %s", (site_id,)) if not cur.fetchone(): # 兜底插入一个 biz.sites 行(site_runtime_context 主键) cur.execute( "INSERT INTO biz.sites (site_id, site_name, is_active) VALUES (%s, %s, true) ON CONFLICT DO NOTHING", (site_id, f"测试门店#{site_id}"), ) conn.commit() finally: conn.close() return int(site_id) # --------------------------------------------------------------------------- # 验证项 # --------------------------------------------------------------------------- def run_view_layer_checks(site_id: int, bd: date, conn_etl) -> list[tuple[str, date | None, str]]: """直查 RLS 视图层(C 方案最底层)。""" rows: list[tuple[str, date | None, str]] = [] targets = [ # (label, sql) ("v_dws_finance_area_daily.stat_date", "SELECT MAX(stat_date) FROM app.v_dws_finance_area_daily"), ("v_dws_finance_daily_summary.stat_date", "SELECT MAX(stat_date) FROM app.v_dws_finance_daily_summary"), ("v_dws_member_visit_detail.visit_date", "SELECT MAX(visit_date) FROM app.v_dws_member_visit_detail"), ("v_dws_member_consumption_summary.stat_date", "SELECT MAX(stat_date) FROM app.v_dws_member_consumption_summary"), ("v_dws_assistant_daily_detail.stat_date", "SELECT MAX(stat_date) FROM app.v_dws_assistant_daily_detail"), ("v_dws_assistant_monthly_summary.stat_month", "SELECT MAX(stat_month) FROM app.v_dws_assistant_monthly_summary"), ("v_dws_assistant_salary_calc.salary_month", "SELECT MAX(salary_month) FROM app.v_dws_assistant_salary_calc"), ("v_dwd_settlement_head.create_time", "SELECT MAX(create_time)::date FROM app.v_dwd_settlement_head"), ("v_dwd_assistant_service_log.create_time", "SELECT MAX(create_time)::date FROM app.v_dwd_assistant_service_log"), ("v_dwd_recharge_order.pay_time", "SELECT MAX(pay_time)::date FROM app.v_dwd_recharge_order"), ("v_dws_member_winback_index.last_visit_time", "SELECT MAX(last_visit_time)::date FROM app.v_dws_member_winback_index"), # 新增 18 个视图样本 ("v_dws_assistant_customer_stats.stat_date", "SELECT MAX(stat_date) FROM app.v_dws_assistant_customer_stats"), ("v_dws_member_assistant_intimacy.calc_time", "SELECT MAX(calc_time)::date FROM app.v_dws_member_assistant_intimacy"), ("v_dws_member_newconv_index.stat_date", "SELECT MAX(stat_date) FROM app.v_dws_member_newconv_index"), ("v_dws_finance_board_cache.computed_at", "SELECT MAX(computed_at)::date FROM app.v_dws_finance_board_cache"), ("v_finance_daily.stat_date", "SELECT MAX(stat_date) FROM app.v_finance_daily"), ] with conn_etl.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.isoformat(),)) for label, sql in targets: try: cur.execute(sql) value = cur.fetchone()[0] d = _date_of(value) rows.append((label, d, _le(d, bd))) except Exception as exc: rows.append((label, None, f"ERROR: {type(exc).__name__}")) conn_etl.rollback() 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.isoformat(),)) conn_etl.commit() return rows def run_service_layer_checks(site_id: int, bd: date) -> list[tuple[str, Any, str]]: """通过后端 service / fdw_queries 函数验证。""" from app.database import get_connection from app.services import board_service, fdw_queries from app.services.runtime_context import ( as_runtime_business_now_str, as_runtime_today_param, as_runtime_year_month_param, get_runtime_context, ) rows: list[tuple[str, Any, str]] = [] conn = get_connection() try: # === 1. RuntimeContext helpers === ctx = get_runtime_context(site_id, conn=conn) rows.append(("ctx.is_sandbox", ctx.is_sandbox, "PASS" if ctx.is_sandbox else "FAIL")) rows.append(("ctx.business_date", ctx.business_date, _le(ctx.business_date, bd) + (" / =bd" if ctx.business_date == bd else ""))) rows.append(("ctx.business_now", ctx.business_now.date(), _le(ctx.business_now.date(), bd))) rows.append(("as_runtime_today_param", as_runtime_today_param(site_id, conn=conn), "PASS" if as_runtime_today_param(site_id, conn=conn) == bd else "FAIL")) rows.append(("as_runtime_year_month_param", as_runtime_year_month_param(site_id, conn=conn), "PASS" if as_runtime_year_month_param(site_id, conn=conn) == f"{bd.year:04d}-{bd.month:02d}" else "FAIL")) rows.append(("as_runtime_business_now_str", as_runtime_business_now_str(site_id, conn=conn), "OK")) # === 2. board_service.calc_date_range(财务看板时间区间) === s, e = board_service._calc_date_range("month", ref_date=ctx.business_date) rows.append(("board.month range end", e, _le(e, bd) + (" / =bd" if e == bd else ""))) s2, e2 = board_service._calc_date_range("quarter", ref_date=ctx.business_date) rows.append(("board.quarter range end", e2, _le(e2, bd))) s3, e3 = board_service._calc_date_range("week", ref_date=ctx.business_date) rows.append(("board.week range end", e3, _le(e3, bd))) # === 3. fdw_queries.get_last_visit_days(客户看板"最近到店"修复点) === # 任取若干 member_id(用 PG_DSN 直连 ETL 测试库) member_ids: list[int] = [] try: etl_conn = psycopg2.connect(os.environ["PG_DSN"]) 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.isoformat(),)) cur.execute( "SELECT member_id FROM app.v_dws_member_visit_detail " "WHERE visit_date <= %s ORDER BY visit_date DESC LIMIT 5", (bd,), ) member_ids = [r[0] for r in cur.fetchall()] etl_conn.commit() etl_conn.close() rows.append(("pick member_ids", member_ids, "OK")) except Exception as exc: rows.append(("pick member_ids", None, f"ERROR: {type(exc).__name__}: {str(exc)[:80]}")) if member_ids: try: last_visit_map = fdw_queries.get_last_visit_days(conn, site_id, member_ids) if last_visit_map: days_list = [v for v in last_visit_map.values() if v is not None] if days_list: min_days = min(days_list) # min_days 至少 0;最近到店日期 <= bd ⇒ days >= 0 rows.append(( "fdw.get_last_visit_days(min)", f"min={min_days} 天前", "PASS" if min_days >= 0 else f"FAIL ({min_days})", )) except Exception as exc: rows.append(("fdw.get_last_visit_days", None, f"ERROR: {type(exc).__name__}: {str(exc)[:80]}")) # === 4. fdw_queries.get_customer_board_recent / recharge / freq60 / recall === for fn_name in [ "get_customer_board_recent", "get_customer_board_recharge", "get_customer_board_freq60", "get_customer_board_recall", ]: try: fn = getattr(fdw_queries, fn_name) resp = fn(conn, site_id, project="ALL", page=1, page_size=10) items = resp.get("items", []) # 找最大日期字段 max_d = None for it in items: for key in ("last_visit_date", "last_recharge", "last_visit"): v = it.get(key) if key in it else None d2 = _date_of(v) if v else None if d2 and (max_d is None or d2 > max_d): max_d = d2 rows.append(( f"fdw.{fn_name}(items={len(items)})", max_d, _le(max_d, bd) if max_d else "OK (no date in items)", )) except Exception as exc: rows.append((f"fdw.{fn_name}", None, f"ERROR: {type(exc).__name__}: {str(exc)[:80]}")) # === 5. fdw_queries.get_coach_60d_stats(任意一对) === if member_ids: try: etl_conn = psycopg2.connect(os.environ["PG_DSN"]) 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.isoformat(),)) cur.execute( "SELECT site_assistant_id FROM app.v_dwd_assistant_service_log " "WHERE tenant_member_id = %s LIMIT 1", (member_ids[0],), ) row = cur.fetchone() etl_conn.commit() etl_conn.close() if row: aid = row[0] stats = fdw_queries.get_coach_60d_stats(conn, site_id, aid, member_ids[0]) rows.append(( f"fdw.get_coach_60d_stats(aid={aid},mid={member_ids[0]})", stats, "OK" if isinstance(stats, dict) else "FAIL", )) except Exception as exc: rows.append(("fdw.get_coach_60d_stats", None, f"ERROR: {type(exc).__name__}: {str(exc)[:80]}")) # === 6. AI 提示词时间锚 === try: now_str = as_runtime_business_now_str(site_id, conn=conn, fmt="%Y-%m-%d %H:%M") d_ai = datetime.strptime(now_str, "%Y-%m-%d %H:%M").date() rows.append(("AI prompt current_time(date)", d_ai, _le(d_ai, bd))) except Exception as exc: rows.append(("AI prompt current_time", None, f"ERROR: {type(exc).__name__}")) finally: conn.close() return rows # --------------------------------------------------------------------------- # 报告 # --------------------------------------------------------------------------- def render_report( site_id: int, bd: date, sandbox_view: list[tuple[str, date | None, str]], sandbox_svc: list[tuple[str, Any, str]], live_view: list[tuple[str, date | None, str]], ) -> str: out: list[str] = [] out.append(f"# 沙箱端到端验证报告\n") out.append(f"- site_id: `{site_id}`") out.append(f"- sandbox_date: `{bd.isoformat()}`") out.append(f"- 生成时间: `{datetime.now().isoformat(timespec='seconds')}`\n") out.append(f"## 1. 视图层(C 方案)\n") out.append("sandbox 模式下,max(各日期列) 必须 <= sandbox_date。\n") out.append("| 视图.列 | 取值 | 结果 |") out.append("|---|---|---|") for label, val, status in sandbox_view: out.append(f"| `app.{label}` | {val} | {status} |") out.append(f"\n### live 模式 baseline(同样 site_id,无 GUC)\n") out.append("| 视图.列 | 取值 | 备注 |") out.append("|---|---|---|") for label, val, _ in live_view: out.append(f"| `app.{label}` | {val} | live (CURRENT_DATE 行为) |") out.append(f"\n## 2. 应用层(B 方案 / RuntimeContext / fdw_queries / AI prompt)\n") out.append("| 调用 | 取值 | 结果 |") out.append("|---|---|---|") for label, val, status in sandbox_svc: out.append(f"| `{label}` | {val} | {status} |") # 汇总 all_rows = sandbox_view + sandbox_svc pass_n = sum(1 for _, _, s in all_rows if s.startswith("PASS") or s == "OK" or s.startswith("OK ")) fail_n = sum(1 for _, _, s in all_rows if s.startswith("FAIL") or s.startswith("ERROR")) other_n = len(all_rows) - pass_n - fail_n out.append(f"\n## 3. 汇总\n") out.append(f"- 总检查项: {len(all_rows)}") out.append(f"- PASS / OK: {pass_n}") out.append(f"- FAIL / ERROR: {fail_n}") out.append(f"- 其他: {other_n}\n") if fail_n == 0: out.append("**结论:PASS — sandbox 模式下,所有关键读取路径都被截到 sandbox_date 之前。**") else: out.append("**结论:FAIL — 仍有项目读到 sandbox_date 之后的数据,详见上表。**") return "\n".join(out) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--sandbox-date", default="2025-09-01") parser.add_argument("--site-id", type=int, default=None) parser.add_argument("--keep-sandbox", action="store_true", help="不还原为 live") parser.add_argument("--out", default="docs/database/changes/2026-05-02__sandbox_e2e_verify_report.md") args = parser.parse_args() bd = datetime.strptime(args.sandbox_date, "%Y-%m-%d").date() site_id = args.site_id or pick_test_site() print(f"[verify] site_id={site_id}, sandbox_date={bd}") # === live baseline === pg_dsn = os.environ["PG_DSN"] etl_conn = psycopg2.connect(pg_dsn) try: live_view = run_view_layer_checks(site_id, date.today() + timedelta(days=10), etl_conn) finally: etl_conn.close() # === sandbox === print(f"[verify] 切换 site_id={site_id} 到 sandbox({bd})") switch_runtime(site_id, "sandbox", bd) try: etl_conn = psycopg2.connect(pg_dsn) try: sandbox_view = run_view_layer_checks(site_id, bd, etl_conn) finally: etl_conn.close() sandbox_svc = run_service_layer_checks(site_id, bd) finally: if not args.keep_sandbox: print(f"[verify] 还原 site_id={site_id} 到 live") switch_runtime(site_id, "live") md = render_report(site_id, bd, sandbox_view, sandbox_svc, live_view) out_path = ROOT / args.out out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(md, encoding="utf-8") print(f"[verify] 报告写入 {out_path.relative_to(ROOT)}") print() print(md.split("## 3. 汇总")[1] if "## 3. 汇总" in md else md[-500:]) if __name__ == "__main__": main()