""" 预警自动生成 — 管理会计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 # 裁定 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 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 参数仅用于口径复算/测试(缺省=系统当前时间)。 """ 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", [])): 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} —— 管理费削减计划" 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, "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", [])} 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")) 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()