From 59e65a68b97234f34f8dee536d5f36cca796262c Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Sun, 13 Sep 2026 00:37:10 +0800 Subject: [PATCH] =?UTF-8?q?fix(alerts):=20cash=5Fplan=20=E5=88=B0=E6=9C=9F?= =?UTF-8?q?=E6=9C=AA=E6=94=B6=E6=AC=BE=E9=A2=84=E8=AD=A6=E5=BC=95=E6=93=8E?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20=E2=80=94=20entity=E5=BD=92=E5=B1=9E/KPI?= =?UTF-8?q?=E5=AE=9E=E4=BD=93=E6=84=9F=E7=9F=A5/=E4=BD=99=E9=A2=9D?= =?UTF-8?q?=E5=BF=AB=E7=85=A7=E6=89=B9=E6=AC=A1/=E5=8E=BB=E9=87=8D?= =?UTF-8?q?=E5=88=B7=E6=96=B0/=E7=AB=AF=E7=82=B9=E9=89=B4=E6=9D=83=20(dama?= =?UTF-8?q?-cashplan-alert-engine-fix-20260913)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F1 两处 KPIAlert 显式写 entity_id（列默认 1 → 博海预警错挂酣客）；顺带修 deviation_engine/ alert_generator/deviation_push/bot_bridge/bot_bridge_v2 同表创建点 F2 KPI 选择实体感知：候选码 ∪ 实体前缀后缀码（_AR_BALANCE/_CASH_BALANCE/_AP_BALANCE） + 兜底 ORDER BY id（此前命中 uk_entity_kpi 必中 BH_ADM_EXPENSE 437 管理费用） F3 去重命中改为刷新 alert_message/等级/suggestion.days_late/归属（此前 continue → 文本冻结首版） F4 余额快照批次（bohai_*_20260630 / receivables_migrate）开关：默认 skip 不逐户，另备 aggregate(按批次聚合1条) / per_customer(旧行为) F5 POST /api/cma/admin/alerts/check 补鉴权（无 token 401）+ 可选 entity_id 只跑指定账套 --- backend/app/api/bot_bridge.py | 2 +- backend/app/api/bot_bridge_v2.py | 1 + backend/app/api/deviation_push.py | 1 + backend/app/main.py | 42 +- backend/app/utils/cash_forecast_engine.py | 250 +++++++++-- backend/app/utils/deviation_engine.py | 2 + backend/scripts/alert_generator.py | 1 + .../verify_cashplan_alert_engine_fix.py | 423 ++++++++++++++++++ docs/manuals/cash-plan-alerts.md | 58 +++ 9 files changed, 741 insertions(+), 39 deletions(-) create mode 100644 backend/scripts/verify_cashplan_alert_engine_fix.py create mode 100644 docs/manuals/cash-plan-alerts.md diff --git a/backend/app/api/bot_bridge.py b/backend/app/api/bot_bridge.py index 36e5b015..6d2a6073 100644 --- a/backend/app/api/bot_bridge.py +++ b/backend/app/api/bot_bridge.py @@ -713,7 +713,7 @@ def bot_kpi_value_with_check(data: dict, db: Session = Depends(get_db), bot: dic KPIAlert.kpi_id == kpi_id, KPIAlert.kpi_value_id == kv.id, KPIAlert.alert_level == level, KPIAlert.status == "pending").first() if not dup: - db.add(KPIAlert(kpi_id=kpi_id, kpi_value_id=kv.id, alert_level=level, + db.add(KPIAlert(entity_id=entity_id, kpi_id=kpi_id, kpi_value_id=kv.id, alert_level=level, alert_message=msg, status="pending", alert_type="bot")) alerts.append({"rule_id": rule.id, "rule_type": rule.rule_type, "level": level, "message": msg}) diff --git a/backend/app/api/bot_bridge_v2.py b/backend/app/api/bot_bridge_v2.py index 8e3fbf20..e7bb8d56 100644 --- a/backend/app/api/bot_bridge_v2.py +++ b/backend/app/api/bot_bridge_v2.py @@ -380,6 +380,7 @@ def receive_mpm_result( # 创建预警 alert_msg = f"⚠️ {kpi_name}异常: {field_value:.1f}% (阈值: {threshold}%) [entity={entity.short_name or entity.name}, period={period}]" alert = KPIAlert( + entity_id=entity_id, # 2026-09-13 F1: 显式写归属 kpi_id=kpi_id or 1, # fallback to first KPI if not found alert_level="red", alert_message=alert_msg, diff --git a/backend/app/api/deviation_push.py b/backend/app/api/deviation_push.py index b74d9774..3fcc11f1 100644 --- a/backend/app/api/deviation_push.py +++ b/backend/app/api/deviation_push.py @@ -131,6 +131,7 @@ def push_deviation_to_map( if alert_level: alert = KPIAlert( + entity_id=kpi.entity_id or 1, # 2026-09-13 F1: 显式写归属 kpi_id=kpi.id, alert_level=alert_level, alert_message=alert_msg, diff --git a/backend/app/main.py b/backend/app/main.py index c8f7c994..1051e085 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,10 +1,12 @@ """管理会计OS — 主入口""" import logging -from fastapi import FastAPI, Request, Depends +from typing import Optional +from fastapi import FastAPI, Request, Depends, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse +from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from dotenv import load_dotenv -from app.database import init_db +from app.database import init_db, get_db from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json from app.utils.cache import clear_all as clear_cache, delete as delete_cache from scripts.erp_sync import run_sync as run_erp_sync @@ -114,6 +116,22 @@ _ERP_CHANNEL_DISABLED_REASON = ("ERP直连同步通道已于2026-08-31停用（e "财务原始数据真实通道为 zxt 智享通注入（source_type='zxt-sync'），本端点不再可用。") +# ── admin 端点鉴权（2026-09-13）────────────────────────────── +# 未带 token → 显式 401（缺少认证），校验逻辑仍完全委托 require_auth（无效 token → 401）。 +# 说明：FastAPI 的 HTTPBearer(auto_error=True) 对"缺 header"返回 403；本端点验收要求 401， +# 故用 auto_error=False 的实例自行判空，不改动全局 require_auth（避免其他端点行为漂移）。 +_optional_bearer = HTTPBearer(auto_error=False) + + +def require_auth_401( + credentials: Optional[HTTPAuthorizationCredentials] = Depends(_optional_bearer), + db=Depends(get_db), +): + if credentials is None or not credentials.credentials: + raise HTTPException(status_code=401, detail="未认证：缺少 Bearer token") + return require_auth(credentials=credentials, db=db) + + @app.post("/api/cma/admin/erp-sync") def admin_erp_sync(user=Depends(require_auth), kpi_codes: str = None): """[已停用] 手动触发ERP数据同步 — ERP 直连通道 2026-08-31 停用，返回 410""" @@ -127,8 +145,12 @@ def admin_erp_sync_dry_run(user=Depends(require_auth), kpi_codes: str = None): @app.post("/api/cma/admin/alerts/check") -def admin_check_alerts(): - """手动触发预警检查（含资金管理预警：缺口前3天 + 到期未收款）""" +def admin_check_alerts(entity_id: Optional[int] = None, user=Depends(require_auth_401)): + """手动触发预警检查（含资金管理预警：缺口前3天 + 到期未收款） + - 2026-09-13 安全修复：补鉴权（与同文件 /admin/erp-sync 等 admin 端点一致）； + 未带 token 显式返回 401（require_auth_401，语义修正：401=未认证 / 403=无权） + - 新增可选 entity_id：只跑指定实体（不传=全部 active 实体，兼容原行为） + """ from app.database import get_session_local from scripts.alert_generator import generate_and_push from app.utils.cash_forecast_engine import check_cash_alerts @@ -136,10 +158,18 @@ def admin_check_alerts(): db = get_session_local()() try: result = generate_and_push(db) - # 资金管理预警 — 对每个激活企业执行 + # 资金管理预警 — 指定实体或全部激活企业 cash_result = None try: - entities = db.query(Entity).filter(Entity.status == "active").all() + if entity_id is not None: + target = db.query(Entity).filter( + Entity.id == entity_id, Entity.status == "active").first() + if not target: + return JSONResponse(status_code=404, + content={"detail": f"企业不存在或未激活: entity_id={entity_id}"}) + entities = [target] + else: + entities = db.query(Entity).filter(Entity.status == "active").all() cash_list = [] for e in entities: cash_list.append(check_cash_alerts(db, entity_id=e.id)) diff --git a/backend/app/utils/cash_forecast_engine.py b/backend/app/utils/cash_forecast_engine.py index b72fbd69..6a2ce4a1 100644 --- a/backend/app/utils/cash_forecast_engine.py +++ b/backend/app/utils/cash_forecast_engine.py @@ -1,5 +1,6 @@ """现金流预测引擎 — 根据历史KPI数据推算未来30天现金流""" import logging +import os from datetime import datetime, timedelta from typing import Optional, TYPE_CHECKING from sqlalchemy.orm import Session @@ -31,20 +32,92 @@ KPI_CODE_CANDIDATES = { "cash_balance": ["CASH_001", "CASH_BALANCE", "F_CASH"], } +# 实体前缀码后缀模式（F2 实体感知 2026-09-13）——标准码全部未命中时， +# 按后缀匹配「该实体自己的编码」（博海口径 BH_AR_BALANCE/BH_CASH_BALANCE/BH_AP_BALANCE）。 +# 修复前：博海一条候选都不命中 → 走无 ORDER BY 的兜底 first() → +# 命中 uk_entity_kpi 索引时取字典序最小的 BH_ADM_EXPENSE(437 管理费用)，即"应收预警挂管理费用"。 +KPI_CODE_SUFFIXES = { + "operating_cash_flow": [], # 无实体前缀变体 + "receivables": ["_AR_BALANCE"], # 博海 BH_AR_BALANCE(433) + "payables": ["_AP_BALANCE"], # 博海 BH_AP_BALANCE(441) + "cash_balance": ["_CASH_BALANCE"], # 博海 BH_CASH_BALANCE(434) +} -def find_kpi(db: Session, entity_id: int, codes: list) -> Optional["KPIDefinition"]: - """按候选编码列表查找KPI，返回第一个命中的""" +# ── 余额快照批次（F4，2026-09-13）──────────────────────────── +# 这些 source 是「期末余额导入」批次：plan_date 是**导入基准日**（如 2026-06-30 期末）， +# 不是真实到期日 → 逐户生成"到期未收款"会永远 overdue，属伪逾期（DAMA 审计 fin-alert-flood-e2）。 +BALANCE_SNAPSHOT_SOURCES = { + "bohai_ar_20260630", + "bohai_ap_20260630", + "bohai_prepay_20260630", + "receivables_migrate", +} + +# 处理开关：per_customer(逐户，旧行为) / skip(不生成逐户，默认) / aggregate(按批次聚合1条) +_SNAPSHOT_MODE_DEFAULT = "skip" +_SNAPSHOT_MODES = ("per_customer", "skip", "aggregate") + + +def get_balance_snapshot_mode() -> str: + """余额快照批次处理模式（环境变量 CMA_CASH_SNAPSHOT_MODE 可覆盖，非法值回落默认） + - per_customer: 旧行为，逐户生成到期未收款预警 + - skip : 默认。批次行不生成逐户预警（口径待任总拍板） + - aggregate : 按批次聚合 1 条预警（如"应收逾期210户 合计198.27万"）挂应收 KPI + """ + mode = (os.environ.get("CMA_CASH_SNAPSHOT_MODE") or "").strip().lower() + return mode if mode in _SNAPSHOT_MODES else _SNAPSHOT_MODE_DEFAULT + + +def get_balance_snapshot_sources() -> set: + """余额快照批次 source 集合（环境变量 CMA_CASH_SNAPSHOT_SOURCES 逗号分隔可追加/覆盖）""" + env = (os.environ.get("CMA_CASH_SNAPSHOT_SOURCES") or "").strip() + if env: + return {s.strip() for s in env.split(",") if s.strip()} + return set(BALANCE_SNAPSHOT_SOURCES) + + +def find_kpi(db: Session, entity_id: int, codes: list, suffixes: Optional[list] = None) -> Optional["KPIDefinition"]: + """按候选编码列表查找KPI，返回第一个命中的（精确码优先 → 实体前缀后缀码 → None） + + F2(2026-09-13): 全部查询显式 ORDER BY id，保证确定性（不再依赖 MySQL 执行计划）。 + """ from app.models import KPIDefinition for code in codes: kpi = db.query(KPIDefinition).filter( KPIDefinition.kpi_code == code, KPIDefinition.entity_id == entity_id, - ).first() + ).order_by(KPIDefinition.id.asc()).first() if kpi: return kpi + if suffixes: + rows = db.query(KPIDefinition).filter( + KPIDefinition.entity_id == entity_id, + ).order_by(KPIDefinition.id.asc()).all() + for suffix in suffixes: + for row in rows: + if (row.kpi_code or "").upper().endswith(suffix): + return row return None +def find_fallback_kpi(db: Session, entity_id: int) -> Optional["KPIDefinition"]: + """兜底 KPI（F2）：优先 finance 维度，其次任意；两种都带 ORDER BY id（确定性） + + 修复前：`db.query(KPIDefinition).filter(entity_id==entity_id).first()` 无排序 → + 同一条件两种答案（idx_entity→318 / uk_entity_kpi→437）。 + """ + from app.models import KPIDefinition + kpi = db.query(KPIDefinition).filter( + KPIDefinition.entity_id == entity_id, + KPIDefinition.dimension == "finance", + ).order_by(KPIDefinition.id.asc()).first() + if kpi: + return kpi + return db.query(KPIDefinition).filter( + KPIDefinition.entity_id == entity_id, + ).order_by(KPIDefinition.id.asc()).first() + + def get_entity_kpi_history(entity_id: int, kpi_code: str, db: Session, limit_months: int = 6) -> list: """获取实体某个KPI的历史值""" from app.models import KPIDefinition, KPIValue @@ -623,25 +696,65 @@ def forecast_cash_flow_with_plans( def check_cash_alerts(db: Session, entity_id: int = 1) -> dict: - """资金预警 — 缺口前3天预警 + 到期未收款提醒，写入预警中心(kpi_alerts)""" + """资金预警 — 缺口前3天预警 + 到期未收款提醒，写入预警中心(kpi_alerts) + + 2026-09-13 修复（任务 dama-cashplan-alert-engine-fix-20260913 / DAMA 审计 fin-alert-flood-e2）： + F1 两处 KPIAlert 显式写 entity_id（此前不传 → 列默认值 1 → 博海预警错挂酣客工作台） + F2 KPI 选择实体感知（候选码 + 实体前缀后缀码）＋ 兜底显式 ORDER BY id + （此前博海走无排序兜底 → 命中 uk_entity_kpi 索引必中 BH_ADM_EXPENSE(437 管理费用)） + F3 去重命中时刷新文本/等级/逾期天数/归属（此前直接 continue → 金额与天数永久冻结在首版） + F4 余额快照批次行按开关处理（默认 skip，不再逐户生成伪"到期未收款"） + """ import json as _json from app.models import KPIAlert, CashPlan, KPIDefinition result = forecast_cash_flow_with_plans(entity_id, db, days=30) new_alerts = [] + refreshed = [] + snapshot_skipped = [] + snapshot_mode = get_balance_snapshot_mode() + snapshot_sources = get_balance_snapshot_sources() - # 兜底KPI：现金KPI → 经营现金流KPI → 该实体任意KPI - kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["cash_balance"]) or \ - find_kpi(db, entity_id, KPI_CODE_CANDIDATES["operating_cash_flow"]) or \ - db.query(KPIDefinition).filter(KPIDefinition.entity_id == entity_id).first() - ar_kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["receivables"]) or kpi + # 兜底KPI（F2）：现金KPI → 经营现金流KPI → 该实体 finance 维度首个（全部 ORDER BY id，确定性） + kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["cash_balance"], KPI_CODE_SUFFIXES["cash_balance"]) or \ + find_kpi(db, entity_id, KPI_CODE_CANDIDATES["operating_cash_flow"], KPI_CODE_SUFFIXES["operating_cash_flow"]) or \ + find_fallback_kpi(db, entity_id) + ar_kpi = find_kpi(db, entity_id, KPI_CODE_CANDIDATES["receivables"], KPI_CODE_SUFFIXES["receivables"]) or kpi def _exists(msg: str) -> bool: return db.query(KPIAlert).filter( + KPIAlert.entity_id == entity_id, KPIAlert.alert_message == msg, KPIAlert.status.in_(["pending", "processing"]), ).first() is not None + def _to_dict(raw) -> dict: + try: + d = _json.loads(raw) if raw else {} + except Exception: + d = {} + return d if isinstance(d, dict) else {} + + def _refresh(existing, kpi_id: int, level: str, msg: str, suggestion: str) -> bool: + """F3: 按现值刷新已存在的预警（归属/文本/等级/建议），返回是否发生变化""" + changed = False + if existing.entity_id != entity_id: + existing.entity_id = entity_id + changed = True + if existing.kpi_id != kpi_id: + existing.kpi_id = kpi_id + changed = True + if existing.alert_level != level: + existing.alert_level = level + changed = True + if existing.alert_message != msg[:500]: + existing.alert_message = msg[:500] + changed = True + if existing.suggestion != suggestion: + existing.suggestion = suggestion + changed = True + return changed + # ── 1. 缺口前3天预警 ── if kpi: for pre in result["pre_alerts"]: @@ -652,6 +765,7 @@ def check_cash_alerts(db: Session, entity_id: int = 1) -> dict: if _exists(msg): continue alert = KPIAlert( + entity_id=entity_id, # F1: 此前未传 → 列默认值 1 kpi_id=kpi.id, alert_level=level, alert_message=msg[:500], @@ -676,40 +790,99 @@ def check_cash_alerts(db: Session, entity_id: int = 1) -> dict: CashPlan.status == "pending", CashPlan.plan_date < today, ).order_by(CashPlan.plan_date.asc()).all() + + # F4: 余额快照批次（plan_date = 导入基准日，非真实到期日）单独处理 + snapshot_rows = [p for p in overdue if (p.source or "") in snapshot_sources] + overdue = [p for p in overdue if (p.source or "") not in snapshot_sources] + + # 去重索引一次建好（修复原逐行全表扫描），去重键 = suggestion.plan_id / suggestion.snapshot_source + existing_alerts = db.query(KPIAlert).filter( + KPIAlert.alert_type == "cash_plan", + KPIAlert.status.in_(["pending", "processing"]), + ).all() + by_plan = {} + by_snapshot = {} + for ea in existing_alerts: + es = _to_dict(ea.suggestion) + if es.get("plan_id") is not None: + by_plan.setdefault(es["plan_id"], ea) + if es.get("snapshot_source"): + by_snapshot.setdefault(es["snapshot_source"], ea) + + if snapshot_mode == "per_customer": + # 旧行为：批次行也逐户生成（仅显式切换时启用） + overdue = overdue + snapshot_rows + elif snapshot_mode == "skip": + if snapshot_rows: + snapshot_skipped = [{"plan_id": p.id, "source": p.source, + "amount": p.amount, "plan_date": str(p.plan_date)} for p in snapshot_rows] + logger.info(f"余额快照批次 {len(snapshot_rows)} 行按 skip 模式不生成逐户预警 " + f"(entity={entity_id}, mode={snapshot_mode})") + elif snapshot_mode == "aggregate": + grouped = {} + for p in snapshot_rows: + grouped.setdefault(p.source or "unknown", []).append(p) + for src, rows in sorted(grouped.items()): + total = sum((p.amount or 0) for p in rows) + oldest = min(p.plan_date for p in rows) + days_late = (today - oldest).days + level = "red" if days_late >= 7 else "yellow" + msg = (f"【到期未收款·批次汇总】{src} 应收逾期{len(rows)}户 合计{total:.2f}万 " + f"最早到期{oldest.strftime('%Y-%m-%d')}（{days_late}天）") + suggestion = _json.dumps({ + "actions": ["核对应收台账", "按客户逐户催收", "确认批次到期口径"], + "snapshot_source": src, + "plan_count": len(rows), + "total_amount": round(total, 2), + "days_late": days_late, + }, ensure_ascii=False) + existing = by_snapshot.get(src) + if existing is not None: + if _refresh(existing, ar_kpi.id, level, msg, suggestion): + refreshed.append({"type": "overdue_snapshot", "snapshot_source": src, + "alert_id": existing.id, "message": msg}) + continue + alert = KPIAlert( + entity_id=entity_id, + kpi_id=ar_kpi.id, + alert_level=level, + alert_message=msg[:500], + alert_type="cash_plan", + status="pending", + suggestion=suggestion, + ) + db.add(alert) + new_alerts.append({"type": "overdue_snapshot", "level": level, "message": msg}) + logger.info(f"到期未收款批次汇总: {msg}") + for p in overdue: days_late = (today - p.plan_date).days + level = "red" if days_late >= 7 else "yellow" msg = (f"【到期未收款】应收款{p.counterparty or '客户'} {p.amount:.1f}万 " - f"原计划{p.plan_date.strftime('%Y-%m-%d')}到期，已逾期{days_late}天未收回") - # 2026-08-30修复: 按plan_id去重（消息含"已逾期X天"动态文本，全文匹配会每天生成重复预警） - existing_plan_alert = db.query(KPIAlert).filter( - KPIAlert.alert_type == "cash_plan", - KPIAlert.status.in_(["pending", "processing"]), - ).all() - dup = False - for ea in existing_plan_alert: - try: - es = _json.loads(ea.suggestion) if ea.suggestion else {} - if isinstance(es, dict) and es.get("plan_id") == p.id: - dup = True - break - except Exception: - continue - if dup: + f"原计划{p.plan_date.strftime('%Y-%m-%d')}到期，已逾期{days_late}天") + suggestion = _json.dumps({ + "actions": ["联系客户催收", "评估坏账风险", "调整信用政策"], + "plan_id": p.id, + "days_late": days_late, + }, ensure_ascii=False) + existing = by_plan.get(p.id) + if existing is not None: + # F3: 命中不 skip —— 按现值刷新金额/逾期天数/等级/归属 + if _refresh(existing, ar_kpi.id, level, msg, suggestion): + refreshed.append({"type": "overdue_receive", "plan_id": p.id, + "alert_id": existing.id, "message": msg}) continue alert = KPIAlert( + entity_id=entity_id, # F1: 此前未传 → 列默认值 1（错挂酣客工作台） kpi_id=ar_kpi.id, - alert_level="red" if days_late >= 7 else "yellow", + alert_level=level, alert_message=msg[:500], alert_type="cash_plan", status="pending", - suggestion=_json.dumps({ - "actions": ["联系客户催收", "评估坏账风险", "调整信用政策"], - "plan_id": p.id, - "days_late": days_late, - }, ensure_ascii=False), + suggestion=suggestion, ) db.add(alert) - new_alerts.append({"type": "overdue_receive", "level": alert.alert_level, "message": msg}) + new_alerts.append({"type": "overdue_receive", "level": level, "message": msg}) logger.info(f"到期未收款提醒: {msg}") db.commit() @@ -717,6 +890,19 @@ def check_cash_alerts(db: Session, entity_id: int = 1) -> dict: "entity_id": entity_id, "new_alerts": len(new_alerts), "alerts": new_alerts, + "refreshed_alerts": len(refreshed), + "refreshed": refreshed, + "balance_snapshot": { + "mode": snapshot_mode, + "skipped_rows": len(snapshot_skipped), + "sources": sorted(snapshot_sources), + }, + "kpi_selection": { + "ar_kpi_id": ar_kpi.id if ar_kpi else None, + "ar_kpi_code": ar_kpi.kpi_code if ar_kpi else None, + "cash_kpi_id": kpi.id if kpi else None, + "cash_kpi_code": kpi.kpi_code if kpi else None, + }, "gap_dates": result["gap_dates"], "pre_alerts": result["pre_alerts"], "summary": result["summary"], diff --git a/backend/app/utils/deviation_engine.py b/backend/app/utils/deviation_engine.py index 3d2af02e..0d91d8dc 100644 --- a/backend/app/utils/deviation_engine.py +++ b/backend/app/utils/deviation_engine.py @@ -341,6 +341,7 @@ def run_deviation_check(db_session, period: str = None) -> int: if not existing: alert = KPIAlert( + entity_id=kpi.entity_id or 1, # 2026-09-13 F1: 显式写归属（原缺省=1 串账套） kpi_id=kpi.id, alert_level=result["kpi_alert_level"], alert_message=f"[差异预警] {alert_msg}", @@ -364,6 +365,7 @@ def run_deviation_check(db_session, period: str = None) -> int: if not existing_trend: alert = KPIAlert( + entity_id=kpi.entity_id or 1, # 2026-09-13 F1: 显式写归属 kpi_id=kpi.id, alert_level=trend["level"], alert_message=f"[趋势预警] {trend_alert_msg}", diff --git a/backend/scripts/alert_generator.py b/backend/scripts/alert_generator.py index adced5a7..6203e0a8 100644 --- a/backend/scripts/alert_generator.py +++ b/backend/scripts/alert_generator.py @@ -114,6 +114,7 @@ def run_alert_check(db_session, period: str = None) -> int: # 创建新预警 alert = KPIAlert( + entity_id=kpi.entity_id or 1, # 2026-09-13 F1: 显式写归属 kpi_id=kpi.id, kpi_value_id=latest.id, alert_level=level, diff --git a/backend/scripts/verify_cashplan_alert_engine_fix.py b/backend/scripts/verify_cashplan_alert_engine_fix.py new file mode 100644 index 00000000..0be6f0fd --- /dev/null +++ b/backend/scripts/verify_cashplan_alert_engine_fix.py @@ -0,0 +1,423 @@ +#!/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) diff --git a/docs/manuals/cash-plan-alerts.md b/docs/manuals/cash-plan-alerts.md new file mode 100644 index 00000000..1a5b522f --- /dev/null +++ b/docs/manuals/cash-plan-alerts.md @@ -0,0 +1,58 @@ +# CMA 模块操作手册：资金收付款计划预警（cash-plan-alerts） + +> 用途：每开发/验证完一个 CMA 模块，产出/更新本手册。防重复修改（先查手册再动手）、防重复验证（照清单执行）。 +> 更新触发：每次该模块 bug 修复/功能变更 → 同步更新本手册 ⑤修改记录。 + +## ① 模块定位 +- **模块名**：资金收付款计划预警（cash_plan 预警引擎） +- **解决什么问题**：把 cash_plans（收付款计划）里「到期未收款」的计划变成预警中心的待办（kpi_alerts.alert_type='cash_plan'），另有「未来30天资金缺口」预测预警（alert_type='forecast'）。 +- **所属流程环节**：资金管理 → 现金预测 / 预警中心 → 处置 +- **主要角色**：finance（财务，处置预警）、ceo（查看） +- **生成入口**：`POST /api/cma/admin/alerts/check`（手动/admin）、cron、`/api/cma/cash/*` 相关端点 + +## ② 操作步骤 +- 财务：预警中心 `/alerts` 查看「到期未收款」pending 预警 → 处理（联系客户/创建改善计划）→ 置 resolved。 +- 运维/Bot：`POST /api/cma/admin/alerts/check`（**需 Bearer token**）；可带 `?entity_id=N` 只跑单个账套。 +- 余额快照批次口径开关（环境变量，改后需重启后端）： + - `CMA_CASH_SNAPSHOT_MODE` = `skip`（默认，批次行不生成逐户预警）| `aggregate`（按批次聚合 1 条）| `per_customer`（旧行为，逐户） + - `CMA_CASH_SNAPSHOT_SOURCES` = 逗号分隔的 source 名单（覆盖内置名单） + +## ③ 验证路径 +- **自动脚本**：`backend/scripts/verify_cashplan_alert_engine_fix.py`（43 项验收电池） + - 在**生产库副本**（mysqldump → `cma_cashvfy`）上跑，零生产写入；唯一生产访问是只读计数；跑完自动删副本。 + - 覆盖：KPI 归属(433/434)/兜底 ORDER BY 确定性/entity_id 归属/去重刷新/快照批次三档开关/端点 401 与 entity 过滤。 +- **期望输出**：`汇总: 43/43 PASS` + `[cleanup] cma_cashvfy 已删除`，退出码 0。 +- **已测基线**：2026-09-13 → 43/43 PASS（副本）；生产端点实测 `?entity_id=2` 200 且 `new_alerts=0`。 +- **手工复核 SQL**（只读）： + ```sql + -- 博海 pending 红色"到期未收款"应为 0 + SELECT COUNT(*) FROM cma.kpi_alerts WHERE entity_id=2 AND alert_type='cash_plan' AND status='pending'; + -- 新预警归属：应收应挂 433(BH_AR_BALANCE)，现金应挂 434(BH_CASH_BALANCE) + SELECT kpi_id, entity_id, COUNT(*) FROM cma.kpi_alerts WHERE alert_type='cash_plan' GROUP BY kpi_id, entity_id; + ``` + +## ④ 已知边界 / 坑 +- **KPI 选择顺序**：候选标准码（酣客口径 AR_001/F_AR_DAYS/C_AR_BALANCE…）→ **实体前缀后缀码**（`_AR_BALANCE` / `_CASH_BALANCE` / `_AP_BALANCE`，命中博海 BH_*）→ 兜底（finance 维度，`ORDER BY id`）。**任何兜底查询都必须带 ORDER BY**，否则同一条件会出现两种答案（实测 `idx_entity`→318 / `uk_entity_kpi`→437）。 +- **entity_id 必须显式写**：`KPIAlert(entity_id=...)`。该列 default=1，漏传会把博海的预警挂到酣客工作台（2026-09-10 实测 589/590）。 +- **去重键 = suggestion.plan_id**（不是 alert_message，后者含动态"已逾期X天"）。命中后必须**刷新**文本/等级/天数/归属，不能 `continue`。 +- **余额快照批次**（source ∈ bohai_ar_20260630 / bohai_ap_20260630 / bohai_prepay_20260630 / receivables_migrate）：plan_date 是**导入基准日**（期末日），不是到期日 → 逐户生成就会永远 overdue（伪逾期）。默认 `skip` 不逐户生成。 +- **skip 模式不接管存量快照预警**：历史遗留的 pending 快照预警需财务手工置 resolved（引擎不会自动改它们的状态）。 +- **⚠️ 易踩点**： + - `LAST_INSERT_ID()` 是**连接级**的——用「新连接」的 raw 查询取不到刚插入的自增 id（验证脚本曾因此把 plan_id 记成 0）。 + - 金额单位是**万元**：cash_plans.amount 注释=万元，导入票据/台账的「元」值必须先 ÷10000，否则预警文本放大 10000 倍。 + - FastAPI `HTTPBearer(auto_error=True)` 对「无 Authorization 头」返回 **403**；本端点用 `require_auth_401` 显式返回 **401**（验收要求）。 +- **历史 bug**：2026-08-29 博海 216 条伪"到期未收款"洪水（元当万元 + 基准日当到期日 + 兜底挂 BH_ADM_EXPENSE 437 + 错挂 entity 1）→ 2026-09-13 修复（本手册⑤）。 + +## ⑤ 修改记录 +| 日期 | 修改人 | 改动 | commit | 影响 | +|------|--------|------|--------|------| +| 2026-09-13 | wecom-fullstack | 手册创建；F1 显式写 entity_id（引擎2处+deviation_engine/alert_generator/deviation_push/bot_bridge/bot_bridge_v2）；F2 候选码加实体前缀后缀 + 兜底 ORDER BY；F3 去重命中改为刷新文本/等级/天数/归属；F4 余额快照批次开关（默认 skip，另备 aggregate/per_customer）；F5 admin 端点补鉴权（无 token 401）+ 可选 entity_id | (待填) | 博海不再重建伪逾期预警；预警 KPI 归属 433/434；不再跨账套错挂 | + +## ⑥ 关联 +- **依赖表**：`cash_plans`、`kpi_alerts`、`kpi_definitions`、`kpi_values`、`system_config` +- **依赖接口**：`POST /api/cma/admin/alerts/check[?entity_id=N]`、`GET /api/cma/cash/*` +- **关联模块**：预警中心（alerts）、预警规则（alert-rules）、资金预测（cash-flow） +- **代码**：`backend/app/utils/cash_forecast_engine.py::check_cash_alerts`、`backend/app/main.py::admin_check_alerts` + +--- +