Hermes CI Fix
2c2dbc2b2d
ci/woodpecker/push/main Pipeline failed
任务：0-P1-project-20260922-actionplan-entity-write-fullstack
父单：project-20260921-cma-entity-isolation-batch5（§B 治本 0c2e57a）
同族已修（勿重做）：alerts.py / deviation_push.py / ai_suggestions.py（d03e322）
缺陷（唯一活写入点）：
scripts/alert_generator.py::run_alert_check 为 red/yellow 预警自动建 ActionPlan 时
不传 entity_id（模型 default=1 + 生产 DDL `DEFAULT 1` 双默认静默兜底）⇒ 非 e1 账套的
KPI 触发阈值时「同一次事务里预警归属正确、计划归属落 1」；读侧 dashboard.py 工作台按
ActionPlan.entity_id 取出计划、KPI 名不做账套过滤渲染 ⇒ 酣客(e1) 工作台显示博海(e2)/
演示(e6) 的 KPI 名（实测 19/83 行归属不符，09-17/18/22 07:00 各新增 6/7/6 行，每日复发）。
改动：
1. 写入点补账套：ActionPlan(entity_id=kpi.entity_id, …) —— 归属唯一事实源 = 来源 KPI 账套；
2. 写前同源校验 action_plan_entity_guard()：来源账套(kpi.entity_id) 与目标账套
(alert.entity_id，即本次遍历账套) 不一致或任一为空 → 不建计划 + OperationLog
(action="skip_plan", reason=plan_entity_attribution_mismatch) 留痕 + 计入统计输出，
不 raise、不中断流水线（口径照抄 bot_bridge_v2 R1 账套守卫 / B3(b) skip_alert 留痕姿势）；
3. 统计/汇总：stats 增 skipped_plan_entity_mismatch / skipped_plan_kpis，并在
generate_and_push 返回值与 coverage_note 中显式输出（不得静默丢弃）；
4. 新增回归 tests/test_alert_generator_entity_attribution_20260922.py（T1/T2/T3 改前 RED、
改后 GREEN；T4 正对照改前即绿，不构成本单判据）。
边界（零越界）：零 DDL（未动列默认、无 alembic/migration/SQL 脚本）、未改 models、
未动同族三处已修文件、未写生产库、未调写端点、未改 cron/前端、未 push、未部署、未重启。
回归：backend `pytest -q` 1374 passed / 0 failed（基线 1370 + 新增 4）
575 lines
30 KiB
Python
575 lines
30 KiB
Python
"""
|
||
预警自动生成 — 管理会计OS
|
||
比对 KPI 实际值与阈值配置（threshold_green/yellow/red），超出则写入 kpi_alerts
|
||
在 daily_sync 之后运行
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import re
|
||
import json
|
||
import logging
|
||
from typing import Optional
|
||
from datetime import datetime, timedelta
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from app.database import get_session_local
|
||
from app.models import KPIDefinition, KPIValue, KPIAlert, ActionPlan, OperationLog
|
||
from app.utils.threshold_parser import eval_threshold, LEVEL_PRIORITY
|
||
# 裁定2a（2026-09-17 财务口径）：比率类指标统一「百分比(%)」口径（取数×100），
|
||
# 未核对/疑似单位错配的项降级待核（不落预警、不推送）
|
||
from app.utils.caliber_units import resolve_caliber
|
||
# 裁定 R1/R2/R3（2026-09-19 财务口径）：账套范围（demo 排除）/ 期间窗口 / 规整月期间
|
||
from app.utils.alert_entity_scope import (
|
||
ALERT_WINDOW_MONTHS, excluded_entity_ids, is_month_period, kpi_excluded,
|
||
period_in_window, window_lower_period,
|
||
)
|
||
|
||
logger = logging.getLogger("cma.alert_gen")
|
||
|
||
#: 月期间正则 —— 只认 YYYY-MM（排除 2026-H1/H2、2025-FY 等伪期间；裁定2b）
|
||
PERIOD_REGEX = r"^[0-9]{4}-[0-9]{2}$"
|
||
_PERIOD_RE = re.compile(PERIOD_REGEX)
|
||
|
||
|
||
def resolve_latest_period(db_session, kpi_id=None):
|
||
"""「最近可用期间」（裁定2b）：kpi_values 内最新 YYYY-MM，排除 H1/H2 伪期间。
|
||
|
||
给定 kpi_id → 返回**该 KPI 自己**的最近可用期间：期间口径不硬编码当月，
|
||
也不被别人（他账套/演示账套）更新的新期间带偏——实证：全局 MAX(period)=2026-09
|
||
但那是演示账套 e6 的 8 条数据，酣客(e1) 的最新可用期间是 2026-08。
|
||
|
||
实现说明：伪期间过滤在 **Python 侧**做（`SELECT DISTINCT period` 只有十几行），
|
||
不依赖 MySQL 的 `REGEXP` —— SQLite 测试库不注册 REGEXP，用 SQL 侧正则会让
|
||
单测与生产走两套语义。
|
||
"""
|
||
q = db_session.query(KPIValue.period).filter(KPIValue.actual_value.isnot(None))
|
||
if kpi_id is not None:
|
||
q = q.filter(KPIValue.kpi_id == kpi_id)
|
||
for (p,) in q.distinct().order_by(KPIValue.period.desc()).all():
|
||
if _PERIOD_RE.fullmatch(str(p or "")):
|
||
return p
|
||
return None
|
||
|
||
|
||
def _parse_threshold(expr: str) -> tuple:
|
||
"""[兼容保留] 统一委托 app.utils.threshold_parser.parse_threshold"""
|
||
from app.utils.threshold_parser import parse_threshold as _p
|
||
return _p(expr)
|
||
|
||
|
||
def cleanup_green_alerts(db_session) -> int:
|
||
"""Bug1修复: 清理历史遗留的green级别pending预警（正常值不应出现在待办）"""
|
||
greens = db_session.query(KPIAlert).filter(
|
||
KPIAlert.alert_level == "green",
|
||
KPIAlert.status.in_(["pending", "processing"]),
|
||
).all()
|
||
count = 0
|
||
for g in greens:
|
||
g.status = "resolved"
|
||
g.resolution = "自动清理: green(正常)级别不生成待处理预警"
|
||
g.resolved_at = datetime.now()
|
||
count += 1
|
||
if count:
|
||
db_session.commit()
|
||
logger.info(f" 清理历史green预警 {count} 条 → resolved")
|
||
return count
|
||
|
||
|
||
def _check_threshold(actual: float, threshold_expr: str, level: str) -> tuple:
|
||
"""检查实际值是否触发阈值，返回 (触发, 消息)（统一走 app.utils.threshold_parser）"""
|
||
if not threshold_expr or actual is None:
|
||
return False, ""
|
||
|
||
if not eval_threshold(actual, threshold_expr):
|
||
return False, ""
|
||
|
||
msg = (f"KPI当前值 {actual:.2f}，触发{ {'green': '正常', 'yellow': '预警', 'red': '紧急'}.get(level, level) }阈值 "
|
||
f"({threshold_expr})")
|
||
return True, msg
|
||
|
||
|
||
def action_plan_entity_guard(db_session, *, kpi, alert, stats: Optional[dict] = None) -> bool:
|
||
"""写前同源校验（2026-09-22 派生单 B4）—— 建行动计划的账套归属门。
|
||
|
||
口径照抄同族守卫姿势（`app/api/bot_bridge_v2.py` 的 R1 账套守卫 / B3(b) `skip_alert` 留痕）：
|
||
· 来源账套 = KPI 定义归属（`kpi.entity_id`）；
|
||
· 目标账套 = **同一次事务里刚建的预警归属**（`alert.entity_id`，即本次遍历账套）；
|
||
· 两者都存在且相等 → 返回 True（调用方建计划并**显式写** `entity_id=kpi.entity_id`）；
|
||
· 任一为空 / 不相等 → 返回 False：**不建计划**、落 `OperationLog(action="skip_plan")` 留痕、
|
||
计入统计输出；**不 raise、不中断流水线**（预警与后续 KPI 照常处理）。
|
||
|
||
为什么必须设门而不是「照传 None」：`action_plans.entity_id` 为 NOT NULL DEFAULT 1
|
||
（模型 `default=1` 与生产 DDL 双默认）——传空值会被**静默兜底到账套 1**，
|
||
正是「酣客(e1) 工作台渲染博海(e2)/演示(e6) KPI 名」的成因（同事务内预警归属正确、
|
||
计划归属落 1，只有 KPI 不属于 e1 时暴露）。
|
||
"""
|
||
kpi_entity_id = getattr(kpi, "entity_id", None)
|
||
target_entity_id = getattr(alert, "entity_id", None)
|
||
if (kpi_entity_id is not None and target_entity_id is not None
|
||
and int(kpi_entity_id) == int(target_entity_id)):
|
||
return True
|
||
|
||
st = stats if stats is not None else {}
|
||
st["skipped_plan_entity_mismatch"] = st.get("skipped_plan_entity_mismatch", 0) + 1
|
||
st.setdefault("skipped_plan_kpis", []).append(
|
||
"%s[kpi_e=%s→plan_e=%s]" % (getattr(kpi, "kpi_code", None), kpi_entity_id, target_entity_id))
|
||
db_session.add(OperationLog(
|
||
user_id=None,
|
||
action="skip_plan",
|
||
target_type="action_plan",
|
||
target_id=getattr(alert, "id", None),
|
||
detail={
|
||
"reason": "plan_entity_attribution_mismatch",
|
||
"kpi_id": getattr(kpi, "id", None),
|
||
"kpi_code": getattr(kpi, "kpi_code", None),
|
||
"kpi_entity_id": kpi_entity_id,
|
||
"entity_id": target_entity_id,
|
||
"alert_id": getattr(alert, "id", None),
|
||
"source": "scripts.alert_generator.run_alert_check",
|
||
},
|
||
))
|
||
logger.warning(
|
||
"B4 账套守卫：跳过建计划（KPI 归属=%s / 目标账套=%s 不一致或为空；"
|
||
"不再静默落默认账套 1）kpi_code=%s alert_id=%s",
|
||
kpi_entity_id, target_entity_id, getattr(kpi, "kpi_code", None), getattr(alert, "id", None))
|
||
return False
|
||
|
||
|
||
def run_alert_check(db_session, period: Optional[str] = None, dry_run: bool = False,
|
||
stats: Optional[dict] = None, now: Optional[datetime] = None) -> int:
|
||
"""检查所有KPI的实际值是否触发预警，返回（将）新生成的预警数
|
||
|
||
2026-09-17 财务口径裁定 2a/2b：
|
||
· **期间（2b）**：period=None → 每个 KPI 回落「最近可用期间」（自己的最新 YYYY-MM，
|
||
排除 H1/H2 伪期间），不再硬编码当月（当月无值 → 永久 0 新预警）；
|
||
· **单位（2a）**：判定前经 resolve_caliber() 归一到「百分比(%)」口径（取数×100，
|
||
与阈值 150/80 同量纲）；**未核对/疑似单位错配**的项降级待核 → 不落预警、不推送；
|
||
· dry_run=True → 只计算不写库（抽查复核与放行前证据用）。
|
||
|
||
2026-09-19 财务口径裁定 R1/R2/R3（`alert_entity_scope` 单一事实源）：
|
||
· **R1 账套**：`status='demo'` 的账套（演示账套）不参与遍历 → 跳过并计
|
||
`skipped_demo_kpis`；
|
||
· **R2 期间窗口**：仅生成「当月往前 6 个自然月（含当月）」内的预警；更早期间跳过
|
||
（`out_of_window`），与 2b 叠加 = max(各 KPI 最近可用期间, 当月-6月)；
|
||
· **R3 伪期间**：仅规整月期间 YYYY-MM 参与，H1/H2/FY 跳过（`skipped_pseudo_period`）。
|
||
· now 参数仅用于口径复算/测试（缺省=系统当前时间）。
|
||
|
||
2026-09-22（本单·第 4 处写入点）：red/yellow 自动建计划前经 `action_plan_entity_guard`
|
||
写前同源校验，并**显式写** `ActionPlan.entity_id = kpi.entity_id` —— 修复前不传归属、
|
||
由模型 default=1 / 生产 DDL DEFAULT 1 静默兜底，导致非 e1 账套的 KPI 触发时落跨账套计划。
|
||
"""
|
||
st = stats if stats is not None else {}
|
||
window_lower = window_lower_period(now)
|
||
excluded_eids = excluded_entity_ids(db_session)
|
||
for k, v in (("period_source", "fixed" if period else "per_kpi_latest"), ("period_fixed", period),
|
||
("periods_used", {}), ("normalized", 0), ("normalized_kpis", []),
|
||
("dimension_suspect", 0), ("suspect_kpis", []), ("no_period", 0),
|
||
("no_value", 0), ("planned", []), ("skipped_existing", 0),
|
||
# R1/R2/R3 口径证据
|
||
("excluded_demo_entities", sorted(excluded_eids)),
|
||
("window_lower_period", window_lower),
|
||
("window_months", ALERT_WINDOW_MONTHS),
|
||
("skipped_demo_kpis", 0), ("demo_kpis", []),
|
||
("skipped_pseudo_period", 0), ("pseudo_period_kpis", []),
|
||
("out_of_window", 0), ("out_of_window_kpis", []),
|
||
# 2026-09-22 计划归属守卫（B4）证据
|
||
("skipped_plan_entity_mismatch", 0), ("skipped_plan_kpis", [])):
|
||
st.setdefault(k, v)
|
||
|
||
kpis = db_session.query(KPIDefinition).filter(
|
||
KPIDefinition.status == "active"
|
||
).all()
|
||
|
||
new_count = 0
|
||
for kpi in kpis:
|
||
# R1（裁定 2026-09-19）：demo 演示账套不参与预警遍历
|
||
if kpi_excluded(db_session, kpi, excluded_eids):
|
||
st["skipped_demo_kpis"] += 1
|
||
st["demo_kpis"].append("%s/e%s" % (kpi.kpi_code, kpi.entity_id or 1))
|
||
continue
|
||
|
||
# 跳过无阈值的KPI
|
||
thr = {
|
||
"red": kpi.threshold_red,
|
||
"yellow": kpi.threshold_yellow,
|
||
"green": kpi.threshold_green,
|
||
}
|
||
if not any(thr.values()):
|
||
continue
|
||
|
||
# 期间口径（裁定2b）：显式指定 → 固定期间；缺省 → 该 KPI 的最近可用期间
|
||
p = period or resolve_latest_period(db_session, kpi.id)
|
||
if not p:
|
||
st["no_period"] += 1
|
||
logger.warning(f" 跳过（无可用期间）: {kpi.kpi_name} 无任何 YYYY-MM 实测值")
|
||
continue
|
||
|
||
# R3（裁定 2026-09-19）：仅规整月期间 YYYY-MM 参与，伪期间一律不参与
|
||
if not is_month_period(p):
|
||
st["skipped_pseudo_period"] += 1
|
||
st["pseudo_period_kpis"].append("%s[%s]" % (kpi.kpi_code, p))
|
||
logger.warning(" 跳过（伪期间 R3）: %s 期间=%s 非 YYYY-MM" % (kpi.kpi_name, p))
|
||
continue
|
||
|
||
# R2（裁定 2026-09-19）：期间窗口 = 当月往前 6 个自然月（含当月），更早不新生成
|
||
if str(p) < window_lower:
|
||
st["out_of_window"] += 1
|
||
st["out_of_window_kpis"].append("%s[%s]" % (kpi.kpi_code, p))
|
||
logger.info(" 跳过（期间超出窗口 R2，窗口下限 %s）: %s[%s]" % (
|
||
window_lower, kpi.kpi_name, p))
|
||
continue
|
||
|
||
# 获取该KPI该期间的最新值
|
||
latest = db_session.query(KPIValue).filter(
|
||
KPIValue.kpi_id == kpi.id,
|
||
KPIValue.period == p,
|
||
).order_by(KPIValue.calculated_at.desc()).first()
|
||
|
||
if not latest or latest.actual_value is None:
|
||
st["no_value"] += 1
|
||
continue
|
||
|
||
# 单位口径（裁定2a）：取数归一 + 疑似错配降级门
|
||
cal = resolve_caliber(latest.actual_value, kpi)
|
||
if cal["status"] == "dimension_suspect":
|
||
st["dimension_suspect"] += 1
|
||
st["suspect_kpis"].append("%s[%s] actual=%s unit=%s" % (
|
||
kpi.kpi_code, p, latest.actual_value, kpi.unit))
|
||
logger.warning(" [降级待核·疑似单位错配→不落预警不推送] %s[%s] 实测=%s 单位=%s ← %s" % (
|
||
kpi.kpi_name, p, latest.actual_value, kpi.unit, cal["reason"]))
|
||
continue
|
||
|
||
actual = cal["value"]
|
||
if cal["status"] == "normalized":
|
||
st["normalized"] += 1
|
||
st["normalized_kpis"].append("%s[%s] %s → %s（×%s）" % (
|
||
kpi.kpi_code, p, latest.actual_value, actual, cal["factor"]))
|
||
st["periods_used"][p] = st["periods_used"].get(p, 0) + 1
|
||
|
||
# 判定顺序：红灯优先 → 绿灯 → 黄灯（最严重者优先，消除宽黄灯遮蔽红灯）
|
||
# 2026-09-11 修复「黄灯遮蔽红灯」：原顺序 绿→黄→红命中即止，宽黄灯(如 Y='>76')
|
||
# 会让红灯(如 R='>91')永远不可达（id=6 实测值 95 被判黄）；
|
||
# green 先于 yellow：green 是主条件，yellow 是更宽的兜底带（id=1 G='>=800' Y='>=700'）。
|
||
# Bug1修复: green(正常)不生成pending预警，只对yellow/red落库
|
||
for level in LEVEL_PRIORITY:
|
||
expr = thr[level]
|
||
if not expr:
|
||
continue
|
||
triggered, msg = _check_threshold(actual, expr, level)
|
||
if triggered:
|
||
if level == "green":
|
||
# 正常状态：不生成待处理预警（如有旧的red/yellow预警自动解除）
|
||
break
|
||
if cal.get("note"):
|
||
msg = "%s｜%s" % (msg, cal["note"])
|
||
# 检查是否已有该期间该KPI同级别的预警
|
||
existing = db_session.query(KPIAlert).filter(
|
||
KPIAlert.kpi_id == kpi.id,
|
||
KPIAlert.kpi_value_id == latest.id,
|
||
KPIAlert.alert_level == level,
|
||
).first()
|
||
|
||
if existing:
|
||
# 已有预警，跳过
|
||
st["skipped_existing"] += 1
|
||
break
|
||
|
||
st["planned"].append({
|
||
"kpi": "%s/%s" % (kpi.kpi_code, kpi.kpi_name), "period": p,
|
||
"actual": actual, "raw_actual": latest.actual_value, "factor": cal["factor"],
|
||
"level": level, "message": msg,
|
||
})
|
||
if dry_run:
|
||
new_count += 1
|
||
break
|
||
|
||
# 创建新预警
|
||
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,
|
||
alert_message=f"{kpi.kpi_name}[{p}] {msg}",
|
||
status="pending",
|
||
)
|
||
db_session.add(alert)
|
||
db_session.flush() # 获取alert.id
|
||
new_count += 1
|
||
logger.info(f" 新增预警 [{level}] {kpi.kpi_name}: {msg}")
|
||
|
||
# Bug2修复: red/yellow预警自动创建行动计划（预警→行动闭环）
|
||
if level in ("red", "yellow"):
|
||
plan_title = f"改善: {kpi.kpi_name} {msg}"
|
||
if kpi.kpi_name and "现金" in kpi.kpi_name or kpi.kpi_code == "F_OP_CFLOW":
|
||
plan_title = f"资金缺口应对: {msg} —— 催收应收+安排融资"
|
||
elif kpi.kpi_code == "F_AR_DAYS":
|
||
plan_title = f"应收催收: {msg} —— 逾期客户专项跟进"
|
||
elif kpi.kpi_code == "C_REBATE_RATE":
|
||
plan_title = f"渠补谈判: {msg} —— 下游渠道返利谈判"
|
||
elif kpi.kpi_code == "F_COST_RATIO":
|
||
plan_title = f"费用压缩: {msg} —— 管理费削减计划"
|
||
# 2026-09-22（本单主修·第 4 处写入点）：写前同源校验 + 显式写归属。
|
||
# 修复前构造 ActionPlan 不传 entity_id → 模型 default=1 / DDL DEFAULT 1
|
||
# 静默兜底，非 e1 账套的 KPI 触发即产生「预警归属正确、计划归属落 1」的
|
||
# 跨账套计划（酣客工作台渲染博海/demo KPI 名的唯一活写入点）。
|
||
if action_plan_entity_guard(db_session, kpi=kpi, alert=alert, stats=st):
|
||
plan = ActionPlan(
|
||
entity_id=kpi.entity_id, # 归属唯一事实源：来源 KPI 的账套
|
||
alert_id=alert.id,
|
||
kpi_id=kpi.id,
|
||
title=plan_title,
|
||
assignee="财务部",
|
||
priority="high" if level == "red" else "medium",
|
||
due_date=datetime.now().date() + timedelta(days=3 if level == "red" else 7),
|
||
created_by="alert_generator",
|
||
)
|
||
db_session.add(plan)
|
||
db_session.flush()
|
||
alert.action_plan_linked_id = plan.id
|
||
logger.info(f" ↳ 自动创建行动计划 #{plan.id}: {plan_title}")
|
||
break # 只取最高级别
|
||
elif level == "red" and not triggered:
|
||
# 红没触发（判定顺序中 red 最先）→ 若该期间该KPI已有 pending/processing 红灯，
|
||
# 说明当前值已不再触发红色阈值，自动解除（保留原有自动降级行为）
|
||
existing_red = db_session.query(KPIAlert).filter(
|
||
KPIAlert.kpi_id == kpi.id,
|
||
KPIAlert.kpi_value_id == latest.id,
|
||
KPIAlert.alert_level == "red",
|
||
KPIAlert.status.in_(["pending", "processing"]),
|
||
).first()
|
||
if existing_red and not dry_run:
|
||
existing_red.status = "resolved"
|
||
existing_red.resolution = "自动解除: 当前值不再触发红色阈值"
|
||
existing_red.resolved_at = datetime.now()
|
||
logger.info(f" 自动解除预警 [red] {kpi.kpi_name}")
|
||
|
||
if dry_run:
|
||
db_session.rollback() # 抽查复核/放行前证据：确保零写入
|
||
return new_count
|
||
db_session.commit()
|
||
return new_count
|
||
|
||
|
||
def run_rule_engine_checks(db_session) -> dict:
|
||
"""预警规则引擎（rule_type 驱动）— 预测偏差 + 客户集中度（P2 2026-09-10）
|
||
|
||
复用 alert_rules 的检查端点实现（FastAPI 路由函数=带 Depends 默认值的普通函数，
|
||
显式传参即可复用，零逻辑复制），按存在规则的账套逐个执行。
|
||
任何单点失败只记录日志，不影响静态阈值主流程。
|
||
"""
|
||
from app.api.alert_rules import (
|
||
AlertRule, run_forecast_deviation_check, run_customer_concentration_check,
|
||
)
|
||
|
||
stats = {"forecast_deviation": 0, "customer_concentration": 0,
|
||
"skipped_demo_entities": 0, "excluded_entities": sorted(excluded_entity_ids(db_session))}
|
||
try:
|
||
rows = db_session.query(AlertRule.entity_id).filter(
|
||
AlertRule.enabled == 1,
|
||
AlertRule.rule_type.in_(("forecast_deviation", "customer_concentration")),
|
||
).distinct().all()
|
||
except Exception as e:
|
||
logger.error(f"规则引擎检查跳过（规则查询失败）: {e}")
|
||
return stats
|
||
|
||
for (eid,) in rows:
|
||
# R1（裁定 2026-09-19）：demo 演示账套不参与预警遍历
|
||
if int(eid or 1) in set(stats["excluded_entities"]):
|
||
stats["skipped_demo_entities"] += 1
|
||
logger.info(f" 规则引擎跳过（demo 账套 R1）: entity={eid}")
|
||
continue
|
||
for key, fn in (("forecast_deviation", run_forecast_deviation_check),
|
||
("customer_concentration", run_customer_concentration_check)):
|
||
try:
|
||
r = fn(db=db_session, entity_id=eid)
|
||
n = int(r.get("generated", 0) or 0)
|
||
stats[key] += n
|
||
if n:
|
||
logger.info(f" 规则预警 entity={eid} {key}: {n}条")
|
||
except Exception as e:
|
||
logger.error(f"规则检查失败 entity={eid} {key}: {e}")
|
||
return stats
|
||
|
||
|
||
def _enabled_channel_count(db_session) -> int:
|
||
"""R4（裁定 2026-09-19）：推送出口计数 —— 生成/推送解耦，空通道只记录不报错。"""
|
||
from app.models import NotificationChannel
|
||
try:
|
||
return int(db_session.query(NotificationChannel).filter(
|
||
NotificationChannel.enabled == True # noqa: E712 - 与其他调用点一致
|
||
).count())
|
||
except Exception as e: # pragma: no cover - 防御分支
|
||
logger.warning(f"读取通知渠道失败（按无渠道处理）: {e}")
|
||
return 0
|
||
|
||
|
||
def generate_and_push(db_session, period: Optional[str] = None, dry_run: bool = False,
|
||
now: Optional[datetime] = None) -> dict:
|
||
"""生成预警并推送，返回统计
|
||
|
||
2026-09-17 财务口径裁定 2a/2b：
|
||
· **period=None** → 判定期间 = 各 KPI「最近可用期间」（裁定2b，不再硬编码当月）；
|
||
显式给出 period → 固定该期间（抽查复核用）；
|
||
· **dry_run=True** → 零写入、零推送：返回将生成的预警清单与口径证据（放行前抽查复核）。
|
||
|
||
2026-09-19 裁定 R1–R4：
|
||
· R1/R2/R3 在 run_alert_check 内落地（demo 账套排除 / 期间窗口 / 规整月期间）；
|
||
· **R4 生成与推送解耦**：`notification_channels` 为空时仅入库、不报错、不阻断；
|
||
"""
|
||
cur_month = (now or datetime.now()).strftime("%Y-%m")
|
||
st = {}
|
||
logger.info("开始预警检查（当月=%s；判定期间=%s；窗口下限=%s；排除账套=%s）..." % (
|
||
cur_month, ("固定 %s" % period) if period else "各 KPI 最近可用期间（裁定2b）",
|
||
window_lower_period(now), sorted(excluded_entity_ids(db_session))))
|
||
|
||
# 0. Bug1修复: 先清理历史green预警（dry-run 不写）
|
||
cleaned = 0 if dry_run else cleanup_green_alerts(db_session)
|
||
|
||
# 1. 生成预警（期间口径 + 单位口径均在此落地）
|
||
new_count = run_alert_check(db_session, period=period, dry_run=dry_run, stats=st, now=now)
|
||
logger.info(f"预警检查完成: {'将新增' if dry_run else '新增'} {new_count} 条")
|
||
|
||
# 1a. 覆盖度自检（2026-09-16 项3；2026-09-17 裁定2b 过渡期保留并扩展「期间回落」明细）：
|
||
# 若**当月**无任何 KPIValue，「0 条新预警」并不代表"确无阈值触发"，而是"当月无数据"；
|
||
# 裁定2b 起改为按各 KPI 最近可用期间判定，故此处显式记录回落后的期间分布。
|
||
coverage = {"current_month": cur_month, "periods_used": st.get("periods_used", {})}
|
||
try:
|
||
kpis_with_values = 0
|
||
active_kpis = db_session.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
|
||
for k in active_kpis:
|
||
if not any((k.threshold_red, k.threshold_yellow, k.threshold_green)):
|
||
continue
|
||
hit = db_session.query(KPIValue).filter(
|
||
KPIValue.kpi_id == k.id, KPIValue.period == cur_month,
|
||
KPIValue.actual_value.isnot(None),
|
||
).first()
|
||
if hit:
|
||
kpis_with_values += 1
|
||
coverage["kpis_with_threshold_and_value"] = kpis_with_values
|
||
coverage["period"] = cur_month
|
||
if kpis_with_values == 0:
|
||
coverage["latest_period_in_db"] = resolve_latest_period(db_session)
|
||
logger.warning(
|
||
"覆盖度告警: 当月(%s)无任何 KPI 实际值 → 本次按「最近可用期间」判定（期间分布 %s）；"
|
||
"「0 条新预警」= 当月无数据可判，非「确无阈值触发」。库内最新月期间=%s" % (
|
||
cur_month, st.get("periods_used"), coverage["latest_period_in_db"]))
|
||
except Exception as e:
|
||
logger.error(f"覆盖度自检失败: {e}")
|
||
coverage = {"error": str(e)}
|
||
|
||
# 1b. 规则引擎检查（预测偏差 / 客户集中度，P2 2026-09-10；dry-run 不写）
|
||
rule_stats = {"forecast_deviation": 0, "customer_concentration": 0} if dry_run \
|
||
else run_rule_engine_checks(db_session)
|
||
|
||
# 2. 推送（dry-run 零推送）
|
||
# R4（裁定 2026-09-19）：生成与推送解耦 —— notification_channels 为空时
|
||
# **仅入库、不报错、不阻断**（推送出口缺失只是"未触达"，不等于"未生成"）。
|
||
channels_enabled = _enabled_channel_count(db_session)
|
||
push_skipped_no_channel = channels_enabled == 0
|
||
pushed = 0
|
||
if new_count > 0 and not dry_run:
|
||
if push_skipped_no_channel:
|
||
logger.info("推送出口缺失（notification_channels 启用数=0）→ 仅入库、不推送、不阻断（裁定 R4）")
|
||
else:
|
||
try:
|
||
from app.utils.notifier import push_pending_alerts
|
||
pushed = push_pending_alerts(db_session)
|
||
logger.info(f"推送完成: {pushed} 条")
|
||
except Exception as e:
|
||
logger.error(f"推送失败（不影响已入库预警，生成/推送解耦 R4）: {e}")
|
||
|
||
return {"period": cur_month, "dry_run": dry_run,
|
||
"period_source": st.get("period_source"), "period_fixed": st.get("period_fixed"),
|
||
"periods_used": st.get("periods_used", {}),
|
||
"new_alerts": new_count, "pushed": pushed,
|
||
# R4：推送出口（生成/推送解耦证据）
|
||
"channels_enabled": channels_enabled,
|
||
"push_skipped_no_channel": push_skipped_no_channel,
|
||
"cleaned_green": cleaned, "rule_engine": rule_stats,
|
||
"coverage": coverage,
|
||
# 裁定2a 口径证据
|
||
"normalized": st.get("normalized", 0), "normalized_kpis": st.get("normalized_kpis", []),
|
||
"dimension_suspect": st.get("dimension_suspect", 0), "suspect_kpis": st.get("suspect_kpis", []),
|
||
# 期间/取值覆盖明细
|
||
"no_period": st.get("no_period", 0), "no_value": st.get("no_value", 0),
|
||
"skipped_existing": st.get("skipped_existing", 0), "planned": st.get("planned", []),
|
||
# 裁定 R1/R2/R3 口径证据
|
||
"excluded_demo_entities": st.get("excluded_demo_entities", []),
|
||
"skipped_demo_kpis": st.get("skipped_demo_kpis", 0),
|
||
"window_lower_period": st.get("window_lower_period"), "window_months": st.get("window_months"),
|
||
"out_of_window": st.get("out_of_window", 0), "out_of_window_kpis": st.get("out_of_window_kpis", []),
|
||
"skipped_pseudo_period": st.get("skipped_pseudo_period", 0),
|
||
"pseudo_period_kpis": st.get("pseudo_period_kpis", []),
|
||
# 2026-09-22 计划归属守卫（B4）：归属不一致/为空 → 跳过建计划的留痕计数
|
||
"skipped_plan_entity_mismatch": st.get("skipped_plan_entity_mismatch", 0),
|
||
"skipped_plan_kpis": st.get("skipped_plan_kpis", [])}
|
||
|
||
|
||
def coverage_note(result: dict) -> str:
|
||
"""汇总行追加提示：期间口径回落后显式说明「0 条 = 无数据可判」而非「确无触发」
|
||
|
||
2026-09-16 新增（项3）；2026-09-17 裁定2b 起追加「期间回落」明细。
|
||
"""
|
||
cov = result.get("coverage") or {}
|
||
parts = []
|
||
used = result.get("periods_used") or cov.get("periods_used") or {}
|
||
if used and cov.get("current_month") not in used:
|
||
parts.append(" | 期间回落(裁定2b): 当月(%s)无 KPI 值 → 按各 KPI 最近可用期间判定 %s"
|
||
% (cov.get("current_month"), used))
|
||
if cov.get("kpis_with_threshold_and_value") == 0:
|
||
parts.append(" | ⚠️ 当月(%s)无KPI值: 本次检查按最近可用期间判定（库内最新月期间 %s）→ 0 条不等于确无触发"
|
||
% (cov.get("current_month"), cov.get("latest_period_in_db")))
|
||
if result.get("dimension_suspect"):
|
||
parts.append(" | ⛔ 降级待核(疑似单位错配，不推送) %s 项: %s"
|
||
% (result["dimension_suspect"], (result.get("suspect_kpis") or [])[:3]))
|
||
if result.get("normalized"):
|
||
parts.append(" | 口径归一 ×100 %s 项" % result["normalized"])
|
||
# 裁定 R1/R2/R3/R4（2026-09-19）证据
|
||
if result.get("skipped_demo_kpis"):
|
||
parts.append(" | R1 排除 demo 账套 %s（跳过 KPI %s 项）"
|
||
% (result.get("excluded_demo_entities"), result["skipped_demo_kpis"]))
|
||
if result.get("out_of_window"):
|
||
parts.append(" | R2 窗口(下限 %s)外不新生成 %s 项"
|
||
% (result.get("window_lower_period"), result["out_of_window"]))
|
||
if result.get("skipped_pseudo_period"):
|
||
parts.append(" | R3 伪期间跳过 %s 项" % result["skipped_pseudo_period"])
|
||
if result.get("push_skipped_no_channel"):
|
||
parts.append(" | R4 推送出口缺失(启用渠道 %s) → 仅入库、不阻断"
|
||
% result.get("channels_enabled"))
|
||
# 2026-09-22 计划归属守卫（B4）：跨账套计划不再静默落默认账套 1
|
||
if result.get("skipped_plan_entity_mismatch"):
|
||
parts.append(" | B4 计划归属守卫：跳过建计划 %s 项 %s（KPI 归属与目标账套不一致/为空）"
|
||
% (result["skipped_plan_entity_mismatch"],
|
||
(result.get("skipped_plan_kpis") or [])[:3]))
|
||
return "".join(parts)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
|
||
ap = argparse.ArgumentParser(description="预警生成与推送（2026-09-17 口径裁定 2a/2b）")
|
||
ap.add_argument("--dry-run", action="store_true", help="只计算不写库不推送（抽查复核/放行前证据）")
|
||
ap.add_argument("--period", default=None, help="固定判定期间 YYYY-MM（缺省=各 KPI 最近可用期间，裁定2b）")
|
||
ap.add_argument("--json", action="store_true", help="JSON 输出完整证据（含将生成的预警清单）")
|
||
args = ap.parse_args()
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
)
|
||
|
||
db = get_session_local()()
|
||
try:
|
||
result = generate_and_push(db, period=args.period, dry_run=args.dry_run)
|
||
if args.json:
|
||
print(json.dumps(result, ensure_ascii=False, indent=1, default=str))
|
||
else:
|
||
print(f"预警检查{'（dry-run 零写入零推送）' if args.dry_run else ''}: "
|
||
f"{result['new_alerts']}条新预警 / {result['pushed']}条已推送"
|
||
f" / 规则引擎{result['rule_engine']}{coverage_note(result)}")
|
||
finally:
|
||
db.close()
|
||
|