From 8c5853f64b76797b1ab90a449d0ff9a068d9b9ba Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Sat, 19 Sep 2026 11:12:31 +0800 Subject: [PATCH] fix(alerts): enforce caliber ruling R5-supp (period linkage) deviation_engine: link kpi_value_id on deviation/trend alert rows; skip when period not attributable or kpi_values row missing; dedupe key now (entity_id,kpi_id,kpi_value_id,alert_type) deviation_push: same-source R1/R5 guards, no-period request writes nothing tests: add test_alert_caliber_r5_residual_20260919 (13 cases incl. negative controls); adapt r1r4/r5 fixtures to the supplemental ruling --- backend/app/api/deviation_push.py | 65 +++- backend/app/utils/deviation_engine.py | 143 +++++--- .../tests/test_alert_caliber_r1r4_20260919.py | 11 +- .../tests/test_alert_caliber_r5_20260919.py | 52 ++- ...test_alert_caliber_r5_residual_20260919.py | 317 ++++++++++++++++++ 5 files changed, 497 insertions(+), 91 deletions(-) create mode 100644 backend/tests/test_alert_caliber_r5_residual_20260919.py diff --git a/backend/app/api/deviation_push.py b/backend/app/api/deviation_push.py index b77da43b..d437f1b2 100644 --- a/backend/app/api/deviation_push.py +++ b/backend/app/api/deviation_push.py @@ -8,9 +8,10 @@ from app.database import get_db from app.auth_middleware import require_auth, require_role from app.deps import get_entity_id from app.models import StrategicMap, KPIDefinition, KPIValue, KPIAlert, OperationLog, ActionPlan, BudgetPlan +# 裁定 R5-补1（2026-09-19 财务口径）：差异反打路径同源接入 R1/R5 守卫 +from app.utils.alert_entity_scope import entity_alert_guard, guard_alert_period import json import logging -from datetime import datetime logger = logging.getLogger("cma.deviation_push") @@ -49,6 +50,22 @@ def push_deviation_to_map( if new_value is None: raise HTTPException(400, "缺少 newValue") + # R5-补1（财务裁定 2026-09-19）：「请求体无 period 一律不落库」 + # 旧码在缺 period 时用当月冒充期间再落库（预警 + kpi_values 双写）—— + # 该兜底正好绕过 R5「可归期」判据，故此处直接整请求零业务写入返回。 + if not period: + logger.info( + "R5-补1 期间守卫：deviation_push 请求无 period，整请求零业务写入" + " entity_id=%s map_id=%s node_id=%s", entity_id, map_id, node_id) + return { + "success": False, + "message": "缺少 period：按 R5-补1（期间不可归期）未落库任何数据", + "alert_created": False, + "alert_skipped_reason": "no_period", + "zero_write": True, + "review_topic_created": False, + } + # 1. 查找战略地图（2026-09-18 派生单: 地图须归属会话账套，NULL 归属按 e1 兼容） m = db.query(StrategicMap).filter(StrategicMap.id == map_id).first() if not m or (m.entity_id or 1) != entity_id: @@ -95,9 +112,8 @@ def push_deviation_to_map( raise HTTPException(404, f"KPI {kpi_code} 不存在") # 3. 更新实际值到 KPIValue 表 - if not period: - period = datetime.now().strftime("%Y-%m") - + #（R5-补1：period 已在上方强制校验，原「缺 period 用当月兜底」分支已删除—— + # 该兜底会伪造期间，正是 R5「不可归期」的成因之一） existing_value = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period, @@ -106,20 +122,22 @@ def push_deviation_to_map( if existing_value: existing_value.actual_value = new_value existing_value.source_type = "manual" + kv_row = existing_value else: - kv = KPIValue( + kv_row = KPIValue( kpi_id=kpi.id, period=period, actual_value=new_value, source_type="manual", ) - db.add(kv) + db.add(kv_row) db.flush() # 4. 检查是否触发预警 alert_created = False alert_id = None + alert_skipped_reason = None if kpi.target_value and kpi.target_value > 0: ratio = new_value / kpi.target_value if ratio < 0.7: @@ -132,17 +150,28 @@ def push_deviation_to_map( alert_level = None 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, - status="pending", - ) - db.add(alert) - db.flush() - alert_created = True - alert_id = alert.id + # R1/R5 双守卫（裁定 R5-补1）：demo 账套 / 非规整月期间 → 该 kpi_alerts 行不入库， + # 返回体登记跳过原因（KPI 值照写，与 bot_bridge 同口径：裁定只约束预警行） + if not entity_alert_guard(db, entity_id, source="deviation_push.push_deviation_to_map", + detail="kpi=%s" % kpi_code): + alert_skipped_reason = "R1_demo_entity_excluded" + elif not guard_alert_period(period, source="deviation_push.push_deviation_to_map", + detail="kpi=%s" % kpi_code): + alert_skipped_reason = "R5_non_attributable_period" + + if alert_skipped_reason is None: + alert = KPIAlert( + entity_id=kpi.entity_id or 1, # 2026-09-13 F1: 显式写归属 + kpi_id=kpi.id, + kpi_value_id=kv_row.id, # R5-补1: 期间关联（同 kpi 同期间行 → 可派生） + alert_level=alert_level, + alert_message=alert_msg, + status="pending", + ) + db.add(alert) + db.flush() + alert_created = True + alert_id = alert.id # 5. 生成战略回顾会议题 review_topic_created = False @@ -180,6 +209,7 @@ def push_deviation_to_map( "new_value": new_value, "period": period, "alert_created": alert_created, + "alert_skipped_reason": alert_skipped_reason, "review_topic_created": review_topic_created, }, ensure_ascii=False), ) @@ -194,6 +224,7 @@ def push_deviation_to_map( "new_value": new_value, "alert_created": alert_created, "alert_id": alert_id, + "alert_skipped_reason": alert_skipped_reason, "review_topic_created": review_topic_created, } diff --git a/backend/app/utils/deviation_engine.py b/backend/app/utils/deviation_engine.py index e1f2617d..39481b0a 100644 --- a/backend/app/utils/deviation_engine.py +++ b/backend/app/utils/deviation_engine.py @@ -17,6 +17,7 @@ from app.models import KPIDefinition, KPIValue, KPIAlert, BudgetPlan, SystemConf # 裁定 R1/R2/R3（2026-09-19 财务口径）：demo 账套排除 / 期间窗口 / 规整月期间 from app.utils.alert_entity_scope import ( excluded_entity_ids, kpi_excluded, period_in_window, window_lower_period, + guard_alert_period, # R5-补1（2026-09-19）：期间可归期 + 关联 kpi_values ) logger = logging.getLogger("cma.deviation") @@ -80,13 +81,46 @@ def get_budget_for_kpi(db, kpi_id: int, period: str, version: str = None) -> Opt return plan.budget_value if plan else None -def get_actual_for_kpi(db, kpi_id: int, period: str) -> Optional[float]: - """获取指定KPI在指定期间的实际值""" - val = db.query(KPIValue).filter( +def get_kpi_value_row(db, kpi_id: int, period: str): + """取 (kpi_id, period) 的 KPIValue 行（取行口径唯一来源）。 + + 与 get_actual_for_kpi 同源：预警行关联的 kpi_values 行 = 数值来源那一行。 + """ + return db.query(KPIValue).filter( KPIValue.kpi_id == kpi_id, KPIValue.period == period, ).order_by(KPIValue.calculated_at.desc()).first() - return val.actual_value if val else None + + +def get_actual_for_kpi(db, kpi_id: int, period: str) -> Optional[float]: + """获取指定KPI在指定期间的实际值""" + row = get_kpi_value_row(db, kpi_id, period) + return row.actual_value if row else None + + +def resolve_alert_kpi_value(db, kpi, period: str): + """R5-补1（财务裁定 2026-09-19）：解析差异/趋势预警行必须关联的 `kpi_values` 行。 + + 裁定原文：「此类预警行必须在落库时写入 `kpi_value_id`，指向 **同 `entity_id` + + 同 `kpi_id` + 文案所示规整月期间** 的 `kpi_values` 行。文案期间不可解析为 `YYYY-MM`， + 或该 (kpi_id, period) 无 `kpi_values` 行 → 按 R5 `guard_alert_period` 跳过不入库 + （不得以 `kpi_value_id=NULL` 落行——『文案里有期间』不构成可归期）。」 + + - 账套约束由 `kpi_id` 传递：KPIDefinition 归属单一账套，故 kpi_id 已唯一确定 entity_id； + 取行口径与 `get_kpi_value_row`（= `get_actual_for_kpi` 的来源行）完全一致， + 保证「预警关联的那一行」就是该预警数值的来源行，不会关联到别的账套/别的期间。 + - 返回 `(kpi_value_row | None, skip_reason | None)`；跳过原因供调用点计数与日志。 + """ + if not guard_alert_period(period, source="deviation_engine.resolve_alert_kpi_value", + detail="kpi=%s" % getattr(kpi, "kpi_code", "")): + return None, "R5_non_attributable_period" + row = get_kpi_value_row(db, kpi.id, period) + if row is None: + logger.info( + "R5-补1 期间关联：跳过不入库（该 (kpi, period) 无 kpi_values 行，期间不可派生）" + " kpi=%s period=%s", getattr(kpi, "kpi_code", ""), period) + return None, "no_kpi_value_row" + return row, None def calc_period_deviation(db, kpi_id: int, period: str) -> dict: @@ -320,6 +354,10 @@ def run_deviation_check(db_session, period: str = None) -> int: 2026-09-19 财务口径裁定 R1/R2/R3（`alert_entity_scope`）： · R1 —— `status='demo'` 的账套不参与遍历（demo 仅演示，不产生业务预警）； · R2/R3 —— 期间须为规整月 `YYYY-MM` 且落在窗口（当月往前 6 个自然月，含当月）内。 + · R5-补1 —— 差异/趋势分支**必须落 `kpi_value_id` 关联**（同 kpi 同期间的 `kpi_values` 行）； + 文案期间不可解析为 `YYYY-MM` 或该 (kpi, period) 无 `kpi_values` 行 → 跳过不入库； + 趋势类以「计算基准期（趋势末端当期）」为关联期间；差异类去重键 + = `(entity_id, kpi_id, kpi_value_id, alert_type)`。 """ if period is None: period = datetime.now().strftime("%Y-%m") @@ -335,6 +373,8 @@ def run_deviation_check(db_session, period: str = None) -> int: new_count = 0 skipped_demo = 0 + skipped_period_linkage = 0 + skipped_linkage_kpis = [] for kpi in kpis: # R1（裁定 2026-09-19）：demo 演示账套不参与预警遍历 if kpi_excluded(db_session, kpi, excluded): @@ -349,53 +389,76 @@ def run_deviation_check(db_session, period: str = None) -> int: f"vs 预算{deviation['budget_value']}，" f"差异率{deviation['deviation_rate']}%" ) + alert_type = result["alert_type"] or "actual" - # 检查是否已有同KPI同期间的差异预警 - existing = db_session.query(KPIAlert).filter( - KPIAlert.kpi_id == kpi.id, - KPIAlert.alert_message.contains("[差异预警]"), - KPIAlert.alert_message.contains(period), - KPIAlert.status.in_(["pending", "processing"]), - ).first() + # R5-补1（裁定 2026-09-19）：期间可归期 + 必须落 kpi_value_id 关联 + #（文案里的期间不构成可归期；无 kpi_values 行 → 跳过不入库） + linked_kv, _skip_reason = resolve_alert_kpi_value(db_session, kpi, period) + if linked_kv is None: + skipped_period_linkage += 1 + skipped_linkage_kpis.append("%s[%s]" % (kpi.kpi_code, period)) + else: + # 去重键（R5-补1 变更）：(entity_id, kpi_id, kpi_value_id, alert_type) + #（原按 `alert_message contains period` 判断，改关联后按可归期目标去重） + existing = db_session.query(KPIAlert).filter( + KPIAlert.entity_id == (kpi.entity_id or 1), + KPIAlert.kpi_id == kpi.id, + KPIAlert.kpi_value_id == linked_kv.id, + KPIAlert.alert_type == alert_type, + KPIAlert.status.in_(["pending", "processing"]), + ).first() - 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}", - alert_type=result["alert_type"] or "actual", - suggestion=result["suggestion"], - status="pending", - ) - db_session.add(alert) - new_count += 1 - logger.info(f" 新增差异预警 [{result['kpi_alert_level']}] {kpi.kpi_name}: 差异率{deviation['deviation_rate']}%") + if not existing: + alert = KPIAlert( + entity_id=kpi.entity_id or 1, # 2026-09-13 F1: 显式写归属（原缺省=1 串账套） + kpi_id=kpi.id, + kpi_value_id=linked_kv.id, # R5-补1: 期间经该行可派生（可归期） + alert_level=result["kpi_alert_level"], + alert_message=f"[差异预警] {alert_msg}", + alert_type=alert_type, + suggestion=result["suggestion"], + status="pending", + ) + db_session.add(alert) + new_count += 1 + logger.info(f" 新增差异预警 [{result['kpi_alert_level']}] {kpi.kpi_name}: 差异率{deviation['deviation_rate']}%") # 2. 趋势异常检测（每期检查连续3期） trend = check_trend_anomaly(db_session, kpi.id, period, consecutive=3) if trend.get("anomaly") and trend.get("level") in ("yellow", "red"): trend_alert_msg = trend["message"] - existing_trend = db_session.query(KPIAlert).filter( - KPIAlert.kpi_id == kpi.id, - KPIAlert.alert_message.contains("[趋势预警]"), - KPIAlert.status.in_(["pending", "processing"]), - ).first() + # R5-补1：趋势类以「计算基准期（趋势末端当期）」为关联期间；该期无行 → 跳过 + trend_periods = trend.get("periods") or [] + base_period = trend_periods[-1] if trend_periods else period + linked_kv, _skip_reason = resolve_alert_kpi_value(db_session, kpi, base_period) + if linked_kv is None: + skipped_period_linkage += 1 + skipped_linkage_kpis.append("%s[%s]" % (kpi.kpi_code, base_period)) + else: + existing_trend = db_session.query(KPIAlert).filter( + KPIAlert.kpi_id == kpi.id, + KPIAlert.alert_message.contains("[趋势预警]"), + KPIAlert.status.in_(["pending", "processing"]), + ).first() - 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}", - status="pending", - ) - db_session.add(alert) - new_count += 1 - logger.info(f" 新增趋势预警 [{trend['level']}] {trend_alert_msg}") + if not existing_trend: + alert = KPIAlert( + entity_id=kpi.entity_id or 1, # 2026-09-13 F1: 显式写归属 + kpi_id=kpi.id, + kpi_value_id=linked_kv.id, # R5-补1: 趋势末端当期关联 + alert_level=trend["level"], + alert_message=f"[趋势预警] {trend_alert_msg}", + status="pending", + ) + db_session.add(alert) + new_count += 1 + logger.info(f" 新增趋势预警 [{trend['level']}] {trend_alert_msg}") if skipped_demo: logger.info(" 差异预警跳过 demo 账套 KPI %s 项（R1，裁定 2026-09-19）", skipped_demo) + if skipped_period_linkage: + logger.info(" 差异/趋势预警跳过（R5-补1 期间不可归期或无 kpi_values 行）KPI %s 项: %s", + skipped_period_linkage, skipped_linkage_kpis) db_session.commit() return new_count diff --git a/backend/tests/test_alert_caliber_r1r4_20260919.py b/backend/tests/test_alert_caliber_r1r4_20260919.py index c2d06037..c41a1c22 100644 --- a/backend/tests/test_alert_caliber_r1r4_20260919.py +++ b/backend/tests/test_alert_caliber_r1r4_20260919.py @@ -158,10 +158,13 @@ class TestR1DemoEntityExcluded: import app.utils.deviation_engine as de _demo_and_active(db) - create_test_kpi(db, kpi_code="DEMO_REVENUE", kpi_name="演示收入", - entity_id=6, unit="%", threshold_red=RED_EXPR) - create_test_kpi(db, kpi_code="F_REVENUE", kpi_name="营业收入", - entity_id=1, unit="%", threshold_red=RED_EXPR) + k_demo = create_test_kpi(db, kpi_code="DEMO_REVENUE", kpi_name="演示收入", + entity_id=6, unit="%", threshold_red=RED_EXPR) + k_real = create_test_kpi(db, kpi_code="F_REVENUE", kpi_name="营业收入", + entity_id=1, unit="%", threshold_red=RED_EXPR) + # R5-补1（2026-09-19）：差异预警须落 kpi_value_id 关联，故夹具补该期间的 kpi_values 行 + #（否则按补1 跳过不入库，本用例将观察不到「遍历范围」这一被测口径） + _kpi_value(db, k_real, "2026-08", 14.45, entity_id=1) # 让所有 KPI 都「差异触发」，只观察遍历范围（不依赖预算数据形态） monkeypatch.setattr(de, "build_deviation_alert", lambda db_s, kpi, period, *a, **kw: { diff --git a/backend/tests/test_alert_caliber_r5_20260919.py b/backend/tests/test_alert_caliber_r5_20260919.py index 660d59c0..4a936608 100644 --- a/backend/tests/test_alert_caliber_r5_20260919.py +++ b/backend/tests/test_alert_caliber_r5_20260919.py @@ -15,10 +15,15 @@ 类A（可落地守卫，本套件 RED→GREEN 对象）：期间已知、可能为伪期间的生成路径 —— 规则引擎（`latest_value.period` 按字符串 desc 取，`2026-H1` 会排在 `2026-09` 之前）、 Bot 写通道（`bot_bridge` / `bot_bridge_v2` 按显式 period 写）。 -类B（现状登记，见 `TestR5ClassBStatusQuo`）：设计上不携带月度期间的族（差异/趋势/集中度/ - 到期未收款/资金缺口预测），其期间**无从从库派生**（`kpi_value_id IS NULL`）； - 裁定边界原文「若日后确有『无期间合法预警』的业务需求，由产品侧另提，本裁定不预设」 - → 本单登记为待产品侧决策项，不做静默屏蔽（会灭掉既有族并触发既有套件失败）。 +类B（现状登记，见 `TestR5ClassBStatusQuo`）：设计上不携带月度期间的族（集中度/到期未收款/ + 资金缺口预测），其期间**无从从库派生**（`kpi_value_id IS NULL`）。 + +2026-09-19 补充裁定 R5-补（财务，收口两个残口）后本套件的口径变化： +- **补1**：差异/趋势/差异反打路径改为**必须落 `kpi_value_id`**（不可归期即跳过）， + 差异类去重键改 `(entity_id, kpi_id, kpi_value_id, alert_type)` → **差异/趋势已由类B 转入类A**， + 其守卫与负对照用例落在 `tests/test_alert_caliber_r5_residual_20260919.py`； +- **补2**：类B「无月度期间族」**不屏蔽（现状保留）**，登记为已知例外，不再作为待决策项挂账 + → `TestR5ClassBStatusQuo` 由「待产品侧决策」的登记，改为「按补2 保留」的登记。 """ from datetime import datetime @@ -217,31 +222,18 @@ class TestR5BotWritePaths: class TestR5ClassBStatusQuo: """类B 现状登记（非守卫断言）——1b 实测证据：这些生成路径产出的行**期间无从从库派生**。 - 裁定边界原文：「若日后确有『无期间合法预警』的业务需求，由产品侧另提，本裁定不预设」。 - 本单结论：按边界**不做静默屏蔽**（屏蔽会灭掉 423 行既有族 + 触发既有套件失败）， - 登记为待产品侧决策项。若产品侧裁定需屏蔽，本用例应随之更新。 + 裁定演进（2026-09-19 §补充裁定 R5-补2）：类B「无月度期间族」**不屏蔽（现状保留）**， + 理由 = 驱动量是日期/占比/趋势，可行动性不以月度期间为前提（属 R5 边界条款所指 + 「无期间合法预警」业务需求）；屏蔽会一次性灭掉「到期未收款」等既有能力（存量 423 行族） + 并触发既有套件失败 → **登记为已知例外，不再作为待决策项挂账**。 + + ⚠ 差异路径原属本类，**已由补1 转入类A**（必须落 kpi_value_id，不可归期即跳过）—— + 其守卫与负对照用例见 `tests/test_alert_caliber_r5_residual_20260919.py`， + 本类不再保留其「无期间关联」的现状断言（该断言与补1 冲突，已随裁定删除）。 + 类B 余族的现状登记同样落在该新套件（`TestR5ResidualClassBStatusQuo`）。 """ - def test_deviation_path_row_has_no_derivable_period(self, db: Session, monkeypatch): - """差异预警路径（cron daily_sync 第 4 步）：期间为规整月，但行内无 kpi_value 关联 - ⇒ 库内派生期间 = NULL（id 17 同类形态，可复现）""" - import app.utils.deviation_engine as de - - _ent(db, 1, "active") - create_test_kpi(db, kpi_code="C_NEW_CLIENTS", kpi_name="新客户数", entity_id=1, unit="家") - monkeypatch.setattr(de, "build_deviation_alert", - lambda db_s, kpi, period, *a, **kw: { - "triggered": True, "kpi_alert_level": "red", "alert_type": "budget", - "suggestion": None, - "deviation": {"actual_value": 88.8, "budget_value": 10.0, - "deviation_rate": 788.0}}) - monkeypatch.setattr(de, "check_trend_anomaly", lambda *a, **kw: {"anomaly": False}) - - n = de.run_deviation_check(db, period="2026-07") - db.expire_all() - - rows = db.query(KPIAlert).all() - assert n == 1 and len(rows) == 1 - assert rows[0].kpi_value_id is None, "差异预警不落 kpi_value 关联" - assert "[差异预警]" in rows[0].alert_message and "2026-07" in rows[0].alert_message, \ - "期间存在于文案中，但不可由库派生（类B：待产品侧决定是否补关联）" + # 原 `test_deviation_path_row_has_no_derivable_period`（差异路径「无期间关联」现状断言） + # 已随 §补充裁定 R5-补1 删除：差异路径转为类A，必须落 kpi_value_id，不可归期即跳过。 + # 替代用例（含负对照：旧逻辑取 NULL / 新逻辑取该期间行）见 + # tests/test_alert_caliber_r5_residual_20260919.py::TestR5ResidualDeviationLinkage（6 例）。 diff --git a/backend/tests/test_alert_caliber_r5_residual_20260919.py b/backend/tests/test_alert_caliber_r5_residual_20260919.py new file mode 100644 index 00000000..8bb6c75a --- /dev/null +++ b/backend/tests/test_alert_caliber_r5_residual_20260919.py @@ -0,0 +1,317 @@ +# -*- coding: utf-8 -*- +"""预警「期间可归期」残口收敛测试（2026-09-19，任务 fullstack-20260919-alert-period-linkage-r5-residual） + +依据：`wecom-finance/reports/CMA_预警口径裁定_20260919.md` + +- **§R5**（2026-09-19）：预警生效判据 = 期间为规整月 `YYYY-MM`；不可归期者不参与预警。 +- **§补充裁定 R5-补1**（同日，收口残口一）：差异/趋势/差异反打路径**必须落 `kpi_value_id`** + （指向同账套 + 同 KPI + 文案所示规整月期间的 `kpi_values` 行）；文案期间不可解析为 `YYYY-MM` + 或该 (kpi_id, period) 无 `kpi_values` 行 → 跳过不入库；趋势类以**趋势末端当期**为关联期间； + 差异类去重键改为 `(entity_id, kpi_id, kpi_value_id, alert_type)`； + `deviation_push` 同源接入 `guard_alert_period` + `entity_alert_guard`，请求体无 `period` 一律不落库。 +- **§补充裁定 R5-补2**（同日，收口残口二）：类B「无月度期间族」**不屏蔽（现状保留）**， + 登记为已知例外，不与类A 混算。 + +本套件每个「期间关联」用例均带**负对照**：夹具同时给出「旧逻辑会取到的目标（未关联 / 别期行）」 +与「新逻辑应取到的目标」，断言新逻辑取新目标、且**不等于**旧目标——否则判据未生效也能通过。 +""" +from datetime import datetime + +import pytest +from sqlalchemy.orm import Session + +from tests.conftest import create_test_kpi, create_test_user +from app.models import (Entity, KPIValue, KPIAlert, KPIDefinition, StrategicMap, + ActionPlan, OperationLog) + +NOW = datetime(2026, 9, 19, 10, 0) + + +def _ent(db, eid, status, name=None): + db.add(Entity(id=eid, name=name or ("账套%d" % eid), short_name="e%d" % eid, status=status)) + db.commit() + + +def _kpi_value(db, kpi, period, value, entity_id=None): + row = KPIValue(entity_id=entity_id, kpi_id=kpi.id, period=period, actual_value=value) + db.add(row) + db.commit() + return row + + +def _trigger_deviation(db, monkeypatch): + """隔离趋势分支：只观察差异分支的落库形态""" + import app.utils.deviation_engine as de + + monkeypatch.setattr(de, "build_deviation_alert", + lambda db_s, kpi, period, *a, **kw: { + "triggered": True, "kpi_alert_level": "red", "alert_type": "budget", + "suggestion": "探针建议", + "deviation": {"actual_value": 88.8, "budget_value": 10.0, + "deviation_rate": 788.0}}) + monkeypatch.setattr(de, "check_trend_anomaly", lambda *a, **kw: {"anomaly": False}) + return de + + +# ─────────────────── 残口一：差异/趋势路径期间关联（补1） ─────────────────── + +class TestR5ResidualDeviationLinkage: + def test_diff_alert_links_current_period_row_not_other_period(self, db: Session, monkeypatch): + """差异预警必须关联「文案所示期间」那一行（负对照：不得关联别期行，也不得留空） + + 夹具两个 kpi_values 行：2026-06（旧逻辑在别处的可能目标 / 干扰项）与 2026-07（文案期间）。 + """ + de = _trigger_deviation(db, monkeypatch) + _ent(db, 1, "active") + k = create_test_kpi(db, kpi_code="C_NEW_CLIENTS", kpi_name="新客户数", entity_id=1, unit="家") + kv_prev = _kpi_value(db, k, "2026-06", 60.0) + kv_cur = _kpi_value(db, k, "2026-07", 88.8) + + n = de.run_deviation_check(db, period="2026-07") + db.expire_all() + + rows = db.query(KPIAlert).all() + assert n == 1 and len(rows) == 1 + assert rows[0].kpi_value_id == kv_cur.id, "新逻辑取新目标（文案期间 2026-07 的 kpi_values 行）" + assert rows[0].kpi_value_id != kv_prev.id, "负对照：不得关联别期行" + assert rows[0].kpi_value_id is not None, "负对照：旧逻辑取旧目标（NULL，期间不可派生）" + # 派生期间可复算 = 可归期 + derived = db.query(KPIValue).filter(KPIValue.id == rows[0].kpi_value_id).first() + assert derived.period == "2026-07" + + def test_diff_alert_skipped_when_no_kpi_value_row(self, db: Session, monkeypatch): + """该 (kpi, period) 无 kpi_values 行 → 跳过不入库（不得以 NULL 关联落行）""" + de = _trigger_deviation(db, monkeypatch) + _ent(db, 1, "active") + create_test_kpi(db, kpi_code="C_NEW_CLIENTS", kpi_name="新客户数", entity_id=1, unit="家") + + n = de.run_deviation_check(db, period="2026-07") + db.expire_all() + + assert n == 0 + assert db.query(KPIAlert).count() == 0 + + def test_trend_alert_links_trend_end_period(self, db: Session, monkeypatch): + """趋势类以「计算基准期（趋势末端当期）」为关联期间（负对照：不得关联趋势起点期）""" + import app.utils.deviation_engine as de + + monkeypatch.setattr(de, "build_deviation_alert", lambda *a, **kw: {"triggered": False}) + _ent(db, 1, "active") + k = create_test_kpi(db, kpi_code="SALES_TOTAL", kpi_name="销售额", entity_id=1, unit="元") + kv_start = _kpi_value(db, k, "2026-05", 30.0) + _kpi_value(db, k, "2026-06", 20.0) + kv_end = _kpi_value(db, k, "2026-07", 10.0) # 连续 3 期下滑 → 趋势末端 + + n = de.run_deviation_check(db, period="2026-07") + db.expire_all() + + rows = db.query(KPIAlert).all() + assert n == 1 and len(rows) == 1 + assert "[趋势预警]" in rows[0].alert_message + assert rows[0].kpi_value_id == kv_end.id, "新逻辑：关联趋势末端当期" + assert rows[0].kpi_value_id != kv_start.id, "负对照：不得关联趋势起点期" + assert rows[0].kpi_value_id is not None, "负对照：旧逻辑取旧目标（NULL）" + + def test_trend_alert_skipped_when_base_period_not_attributable(self, db: Session, monkeypatch): + """趋势末端期间不可归期（非 YYYY-MM）→ 跳过不入库（补1：文案期间不可解析即跳过）""" + import app.utils.deviation_engine as de + + monkeypatch.setattr(de, "build_deviation_alert", lambda *a, **kw: {"triggered": False}) + monkeypatch.setattr(de, "check_trend_anomaly", + lambda *a, **kw: {"anomaly": True, "level": "yellow", + "message": "连续3期下滑", "periods": ["2026-H1"]}) + _ent(db, 1, "active") + create_test_kpi(db, kpi_code="SALES_TOTAL", kpi_name="销售额", entity_id=1, unit="元") + + n = de.run_deviation_check(db, period="2026-07") + db.expire_all() + + assert n == 0 and db.query(KPIAlert).count() == 0 + + def test_diff_dedupe_key_is_linkage_tuple(self, db: Session, monkeypatch): + """去重键改为 (entity_id, kpi_id, kpi_value_id, alert_type) + + 负对照：预置一条「文案含同期间、但 kpi_value_id 为空」的历史行（= 旧去重键会命中的形态）， + 新逻辑按可归期目标去重 → 仍应补一条带关联的预警（而非被文案去重吃掉）。 + """ + de = _trigger_deviation(db, monkeypatch) + _ent(db, 1, "active") + k = create_test_kpi(db, kpi_code="C_NEW_CLIENTS", kpi_name="新客户数", entity_id=1, unit="家") + _kpi_value(db, k, "2026-07", 88.8) + # 旧形态遗留行：文案带期间、无期间关联 + db.add(KPIAlert(entity_id=1, kpi_id=k.id, kpi_value_id=None, alert_level="red", + alert_type="budget", status="pending", + alert_message="[差异预警] 新客户数[2026-07] 差异预警: 实际88.8 vs 预算10.0，差异率788.0%")) + db.commit() + + n = de.run_deviation_check(db, period="2026-07") + db.expire_all() + + rows = db.query(KPIAlert).order_by(KPIAlert.id).all() + assert n == 1 and len(rows) == 2, "新去重键按可归期目标判定 → 补一条带关联的行" + linked = [r for r in rows if r.kpi_value_id is not None] + assert len(linked) == 1 and linked[0].alert_type == "budget" + + def test_diff_dedupe_does_not_duplicate_same_target(self, db: Session, monkeypatch): + """正对照：同一可归期目标重复触发不重复入库（幂等）""" + de = _trigger_deviation(db, monkeypatch) + _ent(db, 1, "active") + k = create_test_kpi(db, kpi_code="C_NEW_CLIENTS", kpi_name="新客户数", entity_id=1, unit="家") + _kpi_value(db, k, "2026-07", 88.8) + + assert de.run_deviation_check(db, period="2026-07") == 1 + assert de.run_deviation_check(db, period="2026-07") == 0 + assert db.query(KPIAlert).count() == 1 + + +# ─────────────────── 残口三：deviation_push 的 R1/R5 缺口（补1） ─────────────────── + +class TestR5ResidualDeviationPush: + def _fixture(self, db, entity_id=1, code="TEST_001"): + k = create_test_kpi(db, kpi_code=code, kpi_name="探针KPI", entity_id=entity_id, + unit="%", target_value=100.0) + db.add(StrategicMap(entity_id=entity_id, title="探针地图", status="published", + dimensions=[{"key": "finance", "name": "财务维度", + "objectives": [{"name": "探针目标", "kpis": [code]}]}])) + db.commit() + m = db.query(StrategicMap).filter(StrategicMap.entity_id == entity_id).first() + return k, m + + def _call(self, db, m, data, entity_id=1): + from app.api.deviation_push import push_deviation_to_map + from app.models import User + + uname = "r5res%d" % entity_id + user = db.query(User).filter(User.username == uname).first() + if user is None: + user = create_test_user(db, username=uname, role="ceo") + return push_deviation_to_map(data=data, db=db, entity_id=entity_id, current_user=user) + + def test_demo_entity_alert_skipped_value_still_written(self, db: Session): + """R1：demo 会话账套不产生业务预警（KPI 值照写，与 bot_bridge 同口径）""" + _ent(db, 6, "demo", "演示企业（虚构数据）") + k, m = self._fixture(db, entity_id=6, code="DEMO_PUSH") + + r = self._call(db, m, {"mapId": m.id, "nodeId": "finance-0", "newValue": 10.0, + "period": "2026-09", "createReviewTopic": False}, entity_id=6) + db.expire_all() + + assert r["alert_created"] is False + assert r["alert_skipped_reason"] == "R1_demo_entity_excluded" + assert db.query(KPIAlert).count() == 0 + assert db.query(KPIValue).filter(KPIValue.kpi_id == k.id).count() == 1, "值照写" + + def test_pseudo_period_alert_skipped(self, db: Session): + """R5：伪期间请求不产生预警行（返回体登记跳过原因）""" + _ent(db, 1, "active") + k, m = self._fixture(db, code="PSEUDO_PUSH") + + r = self._call(db, m, {"mapId": m.id, "nodeId": "finance-0", "newValue": 10.0, + "period": "2026-H1", "createReviewTopic": False}) + db.expire_all() + + assert r["alert_created"] is False + assert r["alert_skipped_reason"] == "R5_non_attributable_period" + assert db.query(KPIAlert).count() == 0 + + def test_missing_period_zero_write(self, db: Session): + """补1：「请求体无 period 一律不落库」——整请求零业务写入""" + _ent(db, 1, "active") + self._fixture(db, code="NOPERIOD_PUSH") + + r = self._call(db, {"id": 1}, {"mapId": 1, "nodeId": "finance-0", "newValue": 10.0}) + db.expire_all() + + assert r["alert_created"] is False and r["zero_write"] is True + assert r["alert_skipped_reason"] == "no_period" + assert db.query(KPIValue).count() == 0, "不得用当月兜底冒充期间写值" + assert db.query(KPIAlert).count() == 0 + assert db.query(ActionPlan).count() == 0 + assert db.query(OperationLog).count() == 0 + + def test_valid_request_alert_links_request_period_row(self, db: Session): + """正对照 + 期间关联：规整月 + active 账套 → 预警落库且关联本次请求期间的行""" + _ent(db, 1, "active") + k, m = self._fixture(db, code="VALID_PUSH") + kv_other = _kpi_value(db, k, "2026-08", 50.0) # 干扰项：别期行 + + r = self._call(db, m, {"mapId": m.id, "nodeId": "finance-0", "newValue": 10.0, + "period": "2026-09", "createReviewTopic": False}) + db.expire_all() + + assert r["alert_created"] is True and r["alert_skipped_reason"] is None + alert = db.query(KPIAlert).filter(KPIAlert.id == r["alert_id"]).first() + linked = db.query(KPIValue).filter(KPIValue.id == alert.kpi_value_id).first() + assert linked is not None and linked.period == "2026-09" + assert alert.kpi_value_id != kv_other.id, "负对照：不得关联别期行" + + def test_repeat_push_reuses_value_row_linkage(self, db: Session): + """同一 (kpi, period) 二次反打 → kpi_values 行复用（值就地更新，不新增行） + + 现状登记：本端点**无预警去重**（第二次反打仍落一条 pending 预警，关联同一 `kpi_values` 行）—— + 裁定 R5-补1 未规定该端点的去重语义（只规定了差异/趋势类的去重键），故按现状登记， + 见报告「观察项」；本用例只锁「期间关联一致 + kpi_values 不重复」这两条已定口径。 + """ + _ent(db, 1, "active") + k, m = self._fixture(db, code="REPEAT_PUSH") + d = {"mapId": 1, "nodeId": "finance-0", "newValue": 10.0, + "period": "2026-09", "createReviewTopic": False} + + r1 = self._call(db, m, dict(d)) + r2 = self._call(db, m, dict(d)) + db.expire_all() + + assert db.query(KPIValue).filter(KPIValue.kpi_id == k.id).count() == 1, "值行复用" + rows = db.query(KPIAlert).order_by(KPIAlert.id).all() + assert len(rows) == 2, "现状登记：该端点无预警去重" + assert rows[0].kpi_value_id == rows[1].kpi_value_id, "两次反打关联同一期间行" + assert r1["alert_id"] != r2["alert_id"] + + +# ─────────────────── 残口二：类B 不屏蔽（补2，现状登记） ─────────────────── + +class TestR5ResidualClassBStatusQuo: + """补2 裁定：类B「无月度期间族」**不屏蔽（现状保留）**，登记为已知例外。 + + 本类为**现状登记**（非守卫断言）：这些族的驱动量是日期/占比/趋势，其可行动性不以月度期间 + 为前提；屏蔽会灭掉「到期未收款」等既有能力（存量 423 行族）并触发既有套件失败。 + 若产品侧日后改判屏蔽，本类应随之更新（与 `test_alert_caliber_r5_20260919.py` 同口径）。 + """ + + def test_customer_concentration_site_still_writes_without_linkage(self, db: Session, monkeypatch): + """集中度（类B，alert_rules:940）落库仍**不经** R5 期间守卫 → 行内无 kpi_value 关联 + + 夹具以 monkeypatch 替代占比计算，观察点 = 「落库是否被期间守卫拦截」。 + """ + from app.api import alert_rules as ar + + ar.AlertRule.__table__.create(bind=db.get_bind(), checkfirst=True) + _ent(db, 1, "active") + k = create_test_kpi(db, kpi_code="CONC_KPI", kpi_name="客户集中度", entity_id=1, unit="%") + db.add(ar.AlertRule(entity_id=1, kpi_id=k.id, rule_type="customer_concentration", enabled=1, + params={"threshold": 30})) + db.commit() + monkeypatch.setattr(ar, "_check_customer_concentration", + lambda *a, **kw: ("red", "[客户集中度] 探针 占比 88%")) + + r = ar.run_customer_concentration_check(db=db, entity_id=1) + db.expire_all() + + rows = db.query(KPIAlert).filter(KPIAlert.alert_type == "concentration").all() + assert r["generated"] == 1 and len(rows) == 1, "类B 现状保留：集中度预警照常入库" + assert rows[0].kpi_value_id is None, "类B 行不携带月度期间（登记形态）" + + def test_class_b_creation_sites_have_no_r5_period_guard(self): + """登记：类B 其余创建点（集中度/预测值/资金缺口/到期未收款）**未接入** R5 期间守卫 + + 判据 = 源码级登记（这些文件在本单中零改动），与 `git diff --stat` 互为佐证。 + """ + import inspect + from app.api import alert_rules as ar + from app.utils import cash_forecast_engine as cfe + + for fn in (ar.run_customer_concentration_check, + ar.run_forecast_alert_check, + cfe.check_cash_alerts): + src = inspect.getsource(fn) + assert "guard_alert_period" not in src, fn.__name__