fix(alerts): cash_plan 到期未收款预警引擎修复 — entity归属/KPI实体感知/余额快照批次/去重刷新/端点鉴权 (dama-cashplan-alert-engine-fix-20260913)
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 只跑指定账套
This commit is contained in:
Hermes CI Fix
@@ -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"],
|
||||
|
||||
Reference in New Issue
Block a user