#!/usr/bin/env python3
"""cash_plan「到期未收款」预警引擎修复 — 生产级验收电池
任务：dama-cashplan-alert-engine-fix-20260913（提出者 wecom-dama / 审计 fin-alert-flood-e2）

覆盖验收项：
  F1 新建预警 entity_id 恒等于入参 entity（+ 其他创建点复查）
  F2 KPI 选择实体感知：博海应收=BH_AR_BALANCE(433) 现金=BH_CASH_BALANCE(434)；
     兜底查询显式 ORDER BY id（不再命中 BH_ADM_EXPENSE 437）
  F3 去重命中时刷新 alert_message（金额/逾期天数）与 suggestion.days_late
  F4 余额快照批次开关（默认 skip 不逐户；aggregate 聚合 1 条；per_customer 旧行为）
  F5 POST /api/cma/admin/alerts/check 无鉴权=401；带 entity_id 只处理该实体

安全设计（零生产写入）：
  * F1-F4 全部在「生产库副本 cma_cashvfy」上执行（mysqldump 一次性复制）
  * F3 用副本内合成计划（source='vfy_cashplan_fix'，仅存在于副本）构造"非快照 + 存量陈旧预警"场景
  * F5 走 in-process TestClient，app.database 的 engine/SessionLocal 重定向到副本
  * 唯一对生产库的访问是只读 COUNT（前后对比，证明未被写入）
  * 副本用完即 DROP；不进常驻 cron

用法：cd /root/cma-management/backend && venv/bin/python3 scripts/verify_cashplan_alert_engine_fix.py
"""
import json
import os
import re
import subprocess
import sys
from datetime import datetime

BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, BASE)

import pymysql
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

PROD_DB = "cma"
SCRATCH_DB = "cma_cashvfy"
SOCKET = "/var/run/mysqld/mysqld.sock"
SCRATCH_URL = f"mysql+pymysql://root@localhost/{SCRATCH_DB}?unix_socket={SOCKET}&charset=utf8mb4"
VFY_SOURCE = "vfy_cashplan_fix"

RESULTS = []
_ENGINES = []
_SESSIONS = []


def check(name, cond, detail=""):
    RESULTS.append((bool(cond), name, detail))
    print(f"{'PASS' if cond else 'FAIL'} | {name}" + (f" | {detail}" if detail else ""))
    return bool(cond)


def raw(sql, db=None, fetch=True, args=None):
    """root socket 直连（副本运维 + 生产只读计数）"""
    conn = pymysql.connect(unix_socket=SOCKET, user="root", database=db, charset="utf8mb4")
    try:
        cur = conn.cursor()
        cur.execute(sql, args) if args else cur.execute(sql)
        rows = cur.fetchall() if fetch else None
        conn.commit()
        return rows
    finally:
        conn.close()


def new_session():
    eng = create_engine(SCRATCH_URL)
    _ENGINES.append(eng)
    sess = sessionmaker(bind=eng)()
    _SESSIONS.append(sess)
    return sess


def prod_counts():
    q = lambda s: raw(s, db=PROD_DB)[0][0]
    return {
        "total": q("SELECT COUNT(*) FROM kpi_alerts"),
        "e1": q("SELECT COUNT(*) FROM kpi_alerts WHERE entity_id=1"),
        "e2_pending": q("SELECT COUNT(*) FROM kpi_alerts WHERE entity_id=2 AND status='pending'"),
        "kpi437_pending": q("SELECT COUNT(*) FROM kpi_alerts WHERE kpi_id=437 AND status='pending'"),
        "plans_e2_pending": q("SELECT COUNT(*) FROM cash_plans WHERE entity_id=2 AND status='pending'"),
    }


def amount_of(msg):
    m = re.search(r"([\d.]+)万", msg or "")
    return float(m.group(1)) if m else None


def days_of(msg):
    m = re.search(r"已逾期(\d+)天", msg or "")
    return int(m.group(1)) if m else None


def parse_sug(sug):
    try:
        d = json.loads(sug) if sug else {}
        return d if isinstance(d, dict) else {}
    except Exception:
        return {}


def setup_scratch():
    raw(f"DROP DATABASE IF EXISTS {SCRATCH_DB}", fetch=False)
    raw(f"CREATE DATABASE {SCRATCH_DB} CHARACTER SET utf8mb4", fetch=False)
    cmd = f"mysqldump --single-transaction --routines --triggers {PROD_DB} | mysql {SCRATCH_DB}"
    p = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    if p.returncode != 0:
        raise RuntimeError(f"副本复制失败: {p.stderr[:500]}")
    n = raw("SELECT COUNT(*) FROM kpi_alerts", db=SCRATCH_DB)[0][0]
    print(f"[setup] cma -> {SCRATCH_DB} 完成, kpi_alerts={n}")
    # 还原 8/29 洪水现场：副本内把实体的 cash_plan 预警状态恢复成 pending（验证 F3 刷新所需）
    raw("UPDATE kpi_alerts SET status='pending', resolved_at=NULL, resolution=NULL "
        "WHERE entity_id=2 AND alert_type='cash_plan'", db=SCRATCH_DB, fetch=False)
    return n


def seed_synthetic_plans():
    """副本内合成 3 条非快照逾期应收计划（F1 新建 / F3 刷新场景）"""
    import datetime as _dt
    today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
    pdate = today.replace(day=1) - _dt.timedelta(days=1)  # 上月最后一天
    rows = [("VFY-测试客户A", 12.3), ("VFY-测试客户B", 0.4), ("VFY-测试客户C", 4567.89)]
    for cp, amt in rows:
        raw("INSERT INTO cash_plans (entity_id, plan_type, amount, plan_date, counterparty, description, "
            "status, source) VALUES (2, 'receive', %s, %s, %s, '系统验证临时数据', 'pending', %s)",
            db=SCRATCH_DB, fetch=False, args=(amt, pdate.strftime("%Y-%m-%d"), cp, VFY_SOURCE))
    ids = [r[0] for r in raw("SELECT id FROM cash_plans WHERE source=%s ORDER BY id",
                             db=SCRATCH_DB, args=(VFY_SOURCE,))]
    # 计划A：植入一条「陈旧 + 错挂」预警（entity=1 / kpi=437 / 金额虚增 / 逾期天数过期）
    stale_msg = f"【到期未收款】应收款VFY-测试客户A 1230000.0万 原计划{pdate:%Y-%m-%d}到期，已逾期10天"
    raw("INSERT INTO kpi_alerts (entity_id, kpi_id, alert_level, alert_message, alert_type, suggestion, status) "
        "VALUES (1, 437, 'red', %s, 'cash_plan', %s, 'pending')",
        db=SCRATCH_DB, fetch=False,
        args=(stale_msg, json.dumps({"actions": ["联系客户催收"], "plan_id": ids[0], "days_late": 10},
                                    ensure_ascii=False)))
    return ids, pdate


def main():
    import app.utils.cash_forecast_engine as eng

    prod_before = prod_counts()
    print(f"[prod before] {prod_before}")

    # ══════════════════════════════════════════════════════════
    # F2 — KPI 实体感知 + 兜底确定性
    # ══════════════════════════════════════════════════════════
    print("\n=== F2 KPI 实体感知 / ORDER BY 确定性 ===")
    s = new_session()
    ar2 = eng.find_kpi(s, 2, eng.KPI_CODE_CANDIDATES["receivables"], eng.KPI_CODE_SUFFIXES["receivables"])
    cash2 = eng.find_kpi(s, 2, eng.KPI_CODE_CANDIDATES["cash_balance"], eng.KPI_CODE_SUFFIXES["cash_balance"])
    ap2 = eng.find_kpi(s, 2, eng.KPI_CODE_CANDIDATES["payables"], eng.KPI_CODE_SUFFIXES["payables"])
    check("F2 博海应收 → BH_AR_BALANCE(433)",
          bool(ar2) and ar2.kpi_code == "BH_AR_BALANCE" and ar2.id == 433,
          f"got={getattr(ar2, 'kpi_code', None)}({getattr(ar2, 'id', None)})")
    check("F2 博海现金 → BH_CASH_BALANCE(434)",
          bool(cash2) and cash2.kpi_code == "BH_CASH_BALANCE" and cash2.id == 434,
          f"got={getattr(cash2, 'kpi_code', None)}({getattr(cash2, 'id', None)})")
    check("F2 博海应付 → BH_AP_BALANCE(441)",
          bool(ap2) and ap2.kpi_code == "BH_AP_BALANCE" and ap2.id == 441,
          f"got={getattr(ap2, 'kpi_code', None)}({getattr(ap2, 'id', None)})")
    fb_ids = {eng.find_fallback_kpi(s, 2).id for _ in range(20)}
    fb = eng.find_fallback_kpi(s, 2)
    check("F2 博海兜底 KPI 不等于 437(BH_ADM_EXPENSE)", fb.id != 437, f"fallback={fb.kpi_code}({fb.id})")
    check("F2 博海兜底 20 次调用恒定（ORDER BY id）", len(fb_ids) == 1, f"ids={fb_ids}")
    plan_dep = raw("SELECT id FROM kpi_definitions WHERE entity_id=2 LIMIT 1", db=SCRATCH_DB)[0][0]
    plan_dep_uk = raw("SELECT id FROM kpi_definitions FORCE INDEX(uk_entity_kpi) "
                      "WHERE entity_id=2 LIMIT 1", db=SCRATCH_DB)[0][0]
    check("F2 复现旧缺陷：同条件两种答案（执行计划相关）",
          plan_dep == 318 and plan_dep_uk == 437, f"idx_entity={plan_dep} / FORCE uk={plan_dep_uk}")
    ar1 = eng.find_kpi(s, 1, eng.KPI_CODE_CANDIDATES["receivables"], eng.KPI_CODE_SUFFIXES["receivables"])
    cash1 = eng.find_kpi(s, 1, eng.KPI_CODE_CANDIDATES["cash_balance"], eng.KPI_CODE_SUFFIXES["cash_balance"]) or \
        eng.find_kpi(s, 1, eng.KPI_CODE_CANDIDATES["operating_cash_flow"], eng.KPI_CODE_SUFFIXES["operating_cash_flow"])
    check("F2 酣客口径零回归（应收=F_AR_DAYS）", bool(ar1) and ar1.kpi_code == "F_AR_DAYS",
          f"got={getattr(ar1, 'kpi_code', None)}({getattr(ar1, 'id', None)})")
    check("F2 酣客口径零回归（现金=F_OP_CFLOW）", bool(cash1) and cash1.kpi_code == "F_OP_CFLOW",
          f"got={getattr(cash1, 'kpi_code', None)}({getattr(cash1, 'id', None)})")
    s.close()

    # ══════════════════════════════════════════════════════════
    # F4-skip（默认）+ 快照批次不逐户
    # ══════════════════════════════════════════════════════════
    os.environ.pop("CMA_CASH_SNAPSHOT_MODE", None)
    os.environ.pop("CMA_CASH_SNAPSHOT_SOURCES", None)
    print("\n=== F4(skip 默认)：余额快照批次不再逐户生成 ===")
    s = new_session()
    snap_before = raw("SELECT id, kpi_id, entity_id, alert_message FROM kpi_alerts "
                      "WHERE entity_id=2 AND alert_type='cash_plan' AND status='pending'", db=SCRATCH_DB)
    before_ids = set(r[0] for r in raw("SELECT id FROM kpi_alerts", db=SCRATCH_DB))
    e1_ids_before = {r[0] for r in raw("SELECT id FROM kpi_alerts WHERE entity_id=1", db=SCRATCH_DB)}
    snap_plans = raw("SELECT COUNT(*) FROM cash_plans WHERE entity_id=2 AND plan_type='receive' "
                     "AND status='pending' AND plan_date < CURDATE()", db=SCRATCH_DB)[0][0]
    res = eng.check_cash_alerts(s, entity_id=2)
    check("F4 默认模式 = skip", res["balance_snapshot"]["mode"] == "skip",
          f"mode={res['balance_snapshot']['mode']}")
    check("F4 快照批次行全部跳过逐户生成", res["balance_snapshot"]["skipped_rows"] == snap_plans,
          f"skipped={res['balance_snapshot']['skipped_rows']} expect={snap_plans}")
    check("F4 默认模式零新增预警（不重建 216 条伪逾期）", res["new_alerts"] == 0,
          f"new_alerts={res['new_alerts']}")
    check("F2 引擎选中应收 KPI = BH_AR_BALANCE(433)", res["kpi_selection"]["ar_kpi_code"] == "BH_AR_BALANCE",
          f"ar={res['kpi_selection']['ar_kpi_code']} cash={res['kpi_selection']['cash_kpi_code']}")
    check("F2 引擎选中现金 KPI = BH_CASH_BALANCE(434)", res["kpi_selection"]["cash_kpi_code"] == "BH_CASH_BALANCE",
          f"cash={res['kpi_selection']['cash_kpi_code']}")
    after_ids = set(r[0] for r in raw("SELECT id FROM kpi_alerts", db=SCRATCH_DB))
    check("F1 未产生任何新预警行", after_ids <= before_ids, f"新增={sorted(after_ids - before_ids)[:5]}")
    snap_after = raw("SELECT id, kpi_id, entity_id, alert_message FROM kpi_alerts "
                     "WHERE entity_id=2 AND alert_type='cash_plan' AND status='pending'", db=SCRATCH_DB)
    check("F4 skip 模式不触碰存量快照预警（交由财务清理）",
          [tuple(r) for r in snap_before] == [tuple(r) for r in snap_after],
          f"before={len(snap_before)} after={len(snap_after)}")
    e1_ids_after = {r[0] for r in raw("SELECT id FROM kpi_alerts WHERE entity_id=1", db=SCRATCH_DB)}
    check("F1 酣客(entity1) 预警零变化", e1_ids_before == e1_ids_after,
          f"{len(e1_ids_before)} -> {len(e1_ids_after)}")
    s.close()

    # ══════════════════════════════════════════════════════════
    # F1 + F3 — 非快照计划的「新建归属」与「去重刷新」（副本合成计划）
    # ══════════════════════════════════════════════════════════
    print("\n=== F1 新建归属 + F3 去重刷新（副本合成非快照计划）===")
    ids, pdate = seed_synthetic_plans()
    exp_days = (datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) - pdate).days
    s = new_session()
    res2 = eng.check_cash_alerts(s, entity_id=2)
    s.expire_all()
    by_plan = {}
    for aid, eid, kid, msg, sug, st in raw(
            "SELECT id, entity_id, kpi_id, alert_message, suggestion, status FROM kpi_alerts "
            "WHERE alert_type='cash_plan'", db=SCRATCH_DB):
        pid = parse_sug(sug).get("plan_id")
        if pid in ids:
            by_plan[pid] = (aid, eid, kid, msg, parse_sug(sug), st)
    check("F3 命中去重时刷新而非跳过（陈旧预警已更新）", res2["refreshed_alerts"] == 1,
          f"refreshed={res2['refreshed_alerts']}")
    check("F1 新建 2 条（无存量预警的计划）", res2["new_alerts"] == 2, f"new={res2['new_alerts']}")
    check("F1/F3 3 条计划预警全部 entity_id=2", all(v[1] == 2 for v in by_plan.values()),
          f"rows={[(k, v[1], v[2]) for k, v in by_plan.items()]}")
    check("F2 3 条计划预警全部 kpi_id=433", all(v[2] == 433 for v in by_plan.values()),
          f"rows={[(k, v[1], v[2]) for k, v in by_plan.items()]}")
    ok_msg, bad = True, []
    for pid in ids:
        v = by_plan.get(pid)
        if not v:
            bad.append((pid, "缺失"))
            continue
        expected_amt = {ids[0]: 12.3, ids[1]: 0.4, ids[2]: 4567.89}[pid]
        if f"{expected_amt:.1f}万" not in v[3]:
            bad.append((pid, f"金额未刷新: {v[3][:50]}"))
        if f"已逾期{exp_days}天" not in v[3]:
            bad.append((pid, f"逾期天数未刷新: {v[3][:50]}"))
        if v[4].get("days_late") != exp_days:
            bad.append((pid, f"days_late={v[4].get('days_late')} expect={exp_days}"))
    check("F3 文本金额/逾期天数与 suggestion.days_late 全部随现值刷新", not bad, f"异常={bad[:3]}")
    check("F3 虚增金额（1230000.0万）已被现值覆盖",
          "1230000.0万" not in "".join(v[3] for v in by_plan.values()), "")
    check("F3 幂等：二次运行 0 新增 0 刷新",
          (lambda r: r["new_alerts"] == 0 and r["refreshed_alerts"] == 0)(eng.check_cash_alerts(s, entity_id=2)),
          "")
    s.close()

    # ══════════════════════════════════════════════════════════
    # F4-aggregate — 方案(b)：按批次聚合 1 条
    # ══════════════════════════════════════════════════════════
    print("\n=== F4(aggregate) 方案(b)：按批次聚合 ===")
    raw("DELETE FROM kpi_alerts WHERE entity_id=2 AND alert_type='cash_plan'", db=SCRATCH_DB, fetch=False)
    os.environ["CMA_CASH_SNAPSHOT_MODE"] = "aggregate"
    s = new_session()
    res3 = eng.check_cash_alerts(s, entity_id=2)
    check("F4 aggregate 模式生效", res3["balance_snapshot"]["mode"] == "aggregate",
          f"mode={res3['balance_snapshot']['mode']}")
    agg = raw("SELECT id, kpi_id, entity_id, alert_message FROM kpi_alerts "
              "WHERE entity_id=2 AND alert_type='cash_plan' AND status='pending' "
              "AND alert_message LIKE '%%批次汇总%%'", db=SCRATCH_DB)
    check("F4 聚合为每批次 1 条（3 个应收批次）", len(agg) == 3, f"count={len(agg)}")
    check("F4 聚合行归属正确（entity=2 / kpi=433）", all(r[1] == 433 and r[2] == 2 for r in agg),
          f"rows={[(r[0], r[1], r[2]) for r in agg]}")
    ar_agg = [r for r in agg if "bohai_ar_20260630" in r[3]]
    check("F4 聚合文本含户数与合计（210户 / 198.27万）",
          len(ar_agg) == 1 and "210户" in ar_agg[0][3] and "198.27万" in ar_agg[0][3],
          f"msg={ar_agg[0][3] if ar_agg else None}")
    res4 = eng.check_cash_alerts(s, entity_id=2)
    check("F4 aggregate 幂等（二次 0 新增 / 0 变更）",
          res4["new_alerts"] == 0 and res4["refreshed_alerts"] == 0,
          f"new={res4['new_alerts']} refreshed={res4['refreshed_alerts']}")
    s.close()

    # ══════════════════════════════════════════════════════════
    # F4-per_customer — 旧行为开关回归（F1 使其也正确）
    # ══════════════════════════════════════════════════════════
    print("\n=== F4(per_customer) 旧行为开关回归 ===")
    raw("DELETE FROM kpi_alerts WHERE entity_id=2 AND alert_type='cash_plan'", db=SCRATCH_DB, fetch=False)
    os.environ["CMA_CASH_SNAPSHOT_MODE"] = "per_customer"
    s = new_session()
    res5 = eng.check_cash_alerts(s, entity_id=2)
    rows5 = raw("SELECT kpi_id, entity_id, alert_message FROM kpi_alerts "
                "WHERE entity_id=2 AND alert_type='cash_plan' AND status='pending'", db=SCRATCH_DB)
    exp_cnt = snap_plans + 3
    plan_amts = [r[0] for r in raw(
        "SELECT amount FROM cash_plans WHERE entity_id=2 AND plan_type='receive' "
        "AND status='pending' AND plan_date < CURDATE()", db=SCRATCH_DB)]
    exp_sum = sum(round(a, 1) for a in plan_amts)
    check("F4 per_customer 复原旧行为（逐户 216+3 条）", len(rows5) == exp_cnt,
          f"count={len(rows5)} expect={exp_cnt}")
    check("F1 旧模式下行归属仍正确（entity=2 / kpi=433）",
          all(r[0] == 433 and r[1] == 2 for r in rows5), "")
    amts5 = [amount_of(r[2]) or 0 for r in rows5]
    check("F1/F3 旧模式金额按现值渲染（无 10000 倍放大）",
          max(amts5) < 5000 and abs(sum(amts5) - exp_sum) < 0.6,
          f"max={max(amts5)} sum={sum(amts5):.2f}万 expect={exp_sum:.2f}万")
    s.close()
    os.environ.pop("CMA_CASH_SNAPSHOT_MODE", None)

    # ══════════════════════════════════════════════════════════
    # F5 — 端点鉴权 + entity 过滤（in-process TestClient，副本重定向）
    # ══════════════════════════════════════════════════════════
    print("\n=== F5 /api/cma/admin/alerts/check 鉴权 + entity 参数 ===")
    raw("DELETE FROM kpi_alerts WHERE entity_id=2 AND alert_type='cash_plan'", db=SCRATCH_DB, fetch=False)
    import app.database as db_module
    db_module._engine = create_engine(SCRATCH_URL)
    _ENGINES.append(db_module._engine)
    db_module._SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=db_module._engine)
    import app.utils.notifier as notifier
    notifier.push_pending_alerts = lambda db_session: 0   # 禁止脚本触碰企微推送
    from app.main import app as fastapi_app
    from fastapi.testclient import TestClient
    client = TestClient(fastapi_app)
    ENDPOINT = "/api/cma/admin/alerts/check"

    r_noauth = client.post(ENDPOINT)
    check("F5 无鉴权调用返回 401", r_noauth.status_code == 401,
          f"status={r_noauth.status_code} body={r_noauth.text[:60]}")
    r_badtoken = client.post(ENDPOINT, headers={"Authorization": "Bearer deadbeef"})
    check("F5 无效 token 返回 401", r_badtoken.status_code == 401, f"status={r_badtoken.status_code}")

    from app.auth_middleware import create_token
    token = create_token(1, 2)          # users.id=1 (admin) 在副本中同样存在
    r_e2 = client.post(ENDPOINT + "?entity_id=2", headers={"Authorization": f"Bearer {token}"})
    body = r_e2.json() if r_e2.status_code == 200 else {}
    cash = body.get("cash") or {}
    details = cash.get("details") or [{}]
    d0 = details[0] if details else {}
    check("F5 带 entity_id=2 请求 200", r_e2.status_code == 200,
          f"status={r_e2.status_code} body={r_e2.text[:100]}")
    check("F5 只处理指定实体（entities=1）", cash.get("entities") == 1, f"entities={cash.get('entities')}")
    check("F5 处理后只产出 entity=2 记录", d0.get("entity_id") == 2, f"entity_id={d0.get('entity_id')}")
    check("F5 修复后快照批次零重建（skipped=216 / 无快照预警）",
          d0.get("balance_snapshot", {}).get("mode") == "skip"
          and d0.get("balance_snapshot", {}).get("skipped_rows") == snap_plans,
          f"snapshot={d0.get('balance_snapshot')}")
    snap_pids = set(r[0] for r in raw(
        "SELECT id FROM cash_plans WHERE entity_id=2 AND plan_type='receive' AND status='pending' "
        "AND plan_date < CURDATE() AND source IN ('bohai_ar_20260630','bohai_ap_20260630',"
        "'bohai_prepay_20260630','receivables_migrate')", db=SCRATCH_DB))
    snap_alerts = 0
    for (sug,) in raw("SELECT suggestion FROM kpi_alerts WHERE entity_id=2 AND alert_type='cash_plan' "
                      "AND status='pending'", db=SCRATCH_DB):
        if parse_sug(sug).get("plan_id") in snap_pids:
            snap_alerts += 1
    check("F5 调用后不存在指向快照计划的预警（原生 SQL 复核）", snap_alerts == 0, f"snap_alerts={snap_alerts}")
    check("F5 引擎对博海不再重建伪预警（new = 仅 3 条合成非快照计划）",
          d0.get("new_alerts") == 3, f"new={d0.get('new_alerts')}")
    check("F5 响应暴露 KPI 归属便于复核（BH_AR_BALANCE）",
          d0.get("kpi_selection", {}).get("ar_kpi_code") == "BH_AR_BALANCE",
          f"kpi_selection={d0.get('kpi_selection')}")
    r_404 = client.post(ENDPOINT + "?entity_id=9999", headers={"Authorization": f"Bearer {token}"})
    check("F5 不存在/未激活 entity → 404", r_404.status_code == 404, f"status={r_404.status_code}")
    leaked = raw("SELECT COUNT(*) FROM kpi_alerts WHERE entity_id<>2 AND status='pending' "
                 "AND alert_type='cash_plan' AND created_at > NOW() - INTERVAL 5 MINUTE",
                 db=SCRATCH_DB)[0][0]
    check("F5 本次调用未向其他实体写入预警", leaked == 0, f"leaked={leaked}")

    # ══════════════════════════════════════════════════════════
    # 生产库零写入核验
    # ══════════════════════════════════════════════════════════
    prod_after = prod_counts()
    check("生产库 kpi_alerts / cash_plans 零变化", prod_before == prod_after,
          f"before={prod_before}\n                                                after ={prod_after}")

    ok = sum(1 for c, _, _ in RESULTS if c)
    print(f"\n{'='*60}\n汇总: {ok}/{len(RESULTS)} PASS")
    for c, name, detail in RESULTS:
        if not c:
            print(f"  FAIL: {name} | {detail}")
    return 0 if ok == len(RESULTS) else 1


def teardown():
    for es in _SESSIONS:
        try:
            es.close()
        except Exception:
            pass
    for en in _ENGINES:
        try:
            en.dispose()
        except Exception:
            pass
    for attempt in range(3):
        try:
            conn = pymysql.connect(unix_socket=SOCKET, user="root", charset="utf8mb4")
            cur = conn.cursor()
            cur.execute("SET SESSION lock_wait_timeout=8")
            cur.execute(f"DROP DATABASE IF EXISTS {SCRATCH_DB}")
            conn.close()
            print(f"[cleanup] {SCRATCH_DB} 已删除")
            return
        except Exception as e:
            print(f"[cleanup] 第{attempt+1}次删除失败: {e}")
    print(f"[cleanup] 警告：{SCRATCH_DB} 未删除，请手工清理")


if __name__ == "__main__":
    code = 1
    try:
        setup_scratch()
        code = main()
    except Exception:
        import traceback
        traceback.print_exc()
    finally:
        if os.environ.get("KEEP_SCRATCH") != "1":
            teardown()
    sys.exit(code)