Hermes CI Fix
3bb55189b1
ci/woodpecker/push/main Pipeline failed
依据：wecom-finance/reports/CMA_口径裁定_20260917.md（任总 2026-09-17 23:14「按建议执行」） 裁定1（现金流「投资活动净额=0」参与判定） - classify_enterprise()：0 为合法输入；投资恒为 0 → 经营(±)×筹资(±) 二维映射出 2 个 候选类型（candidate_type_nos/candidates），不再整体判「未判定」；投资栏 investing_note 标注「本期无投资活动」；经营/筹资亦为 0 → 回退「未判定（无投资活动）」，不报错不静默 - /analysis 增 classification_dimension / investing_note / nets.investing.note；前端出 二维横幅 + 候选行高亮 + 投资栏「本期无投资活动」（不给单一类型名） - import_cashflow_statements.py：打印层修 2D（type_no=None）TypeError 崩溃；NA 措辞改为 「按 0 参与判定」（机读短语 NA_PHRASE_TMPL/parse_na_activities 未动） 裁定2a（比率类单位统一「百分比(%)」口径，先做） - 新增 app/utils/caliber_units.py：逐 KPI 对照表 RATIO_CALIBER_TABLE（唯一事实源）+ 判定侧 唯一入口 resolve_caliber()；3 项需归一 ×100（流动/速动/毛利率），12 项取数已百分数不归一； 表外且命中「疑似错配」窄形态（阈值子句量级≥100 且 0<|实测|<10 且单位属比率类）→ dimension_suspect：不落预警、不推送（唯一规则，防 586/587 假红灯复发） - alert_generator：判定前归一，文案写「口径归一 ×100（原值 1.29 → 129%）」留审计线索 - 新增 scripts/kpi_unit_crosscheck.py（只读对照表产出器） 裁定2b（期间由「当月硬编码」→「最近可用期间」，后做） - resolve_latest_period()：取该 KPI 自己的最新 YYYY-MM（禁全局 MAX，防演示账套劫持）； 伪期间（2026-H1/H2、2025-FY）在 Python 侧过滤（保「单测=生产」语义） - 过渡期覆盖度自检扩展：记录回落后的期间分布 + 当月 0 条归因提示 - --dry-run（零写入零推送）作放行前抽查证据；generate_and_push/run_alert_check 支持 dry_run 测试：backend/tests/test_caliber_rulings_20260917.py 17 项；全量 1158 passed/1 xfailed
412 lines
20 KiB
Python
412 lines
20 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
|
||
from app.utils.threshold_parser import eval_threshold, LEVEL_PRIORITY
|
||
# 裁定2a（2026-09-17 财务口径）：比率类指标统一「百分比(%)」口径（取数×100），
|
||
# 未核对/疑似单位错配的项降级待核（不落预警、不推送）
|
||
from app.utils.caliber_units import resolve_caliber
|
||
|
||
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 run_alert_check(db_session, period: Optional[str] = None, dry_run: bool = False,
|
||
stats: Optional[dict] = 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 → 只计算不写库（抽查复核与放行前证据用）。
|
||
"""
|
||
st = stats if stats is not None else {}
|
||
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)):
|
||
st.setdefault(k, v)
|
||
|
||
kpis = db_session.query(KPIDefinition).filter(
|
||
KPIDefinition.status == "active"
|
||
).all()
|
||
|
||
new_count = 0
|
||
for kpi in kpis:
|
||
# 跳过无阈值的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
|
||
|
||
# 获取该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} —— 管理费削减计划"
|
||
plan = ActionPlan(
|
||
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}
|
||
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:
|
||
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 generate_and_push(db_session, period: Optional[str] = None, dry_run: bool = False) -> dict:
|
||
"""生成预警并推送，返回统计
|
||
|
||
2026-09-17 财务口径裁定 2a/2b：
|
||
· **period=None** → 判定期间 = 各 KPI「最近可用期间」（裁定2b，不再硬编码当月）；
|
||
显式给出 period → 固定该期间（抽查复核用）；
|
||
· **dry_run=True** → 零写入、零推送：返回将生成的预警清单与口径证据（放行前抽查复核）。
|
||
"""
|
||
cur_month = datetime.now().strftime("%Y-%m")
|
||
st = {}
|
||
logger.info("开始预警检查（当月=%s；判定期间=%s）..." % (
|
||
cur_month, ("固定 %s" % period) if period else "各 KPI 最近可用期间（裁定2b）"))
|
||
|
||
# 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)
|
||
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 零推送）
|
||
pushed = 0
|
||
if new_count > 0 and not dry_run:
|
||
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"推送失败: {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,
|
||
"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", [])}
|
||
|
||
|
||
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"])
|
||
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()
|
||
|