""" 预警自动生成 — 管理会计OS 比对 KPI 实际值与阈值配置(threshold_green/yellow/red),超出则写入 kpi_alerts 在 daily_sync 之后运行 """ import sys import os import re import json import logging from datetime import datetime 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 logger = logging.getLogger("cma.alert_gen") def _parse_threshold(expr: str) -> tuple: """解析阈值表达式,返回 (operator, value) 示例: >=32000000 -> ('>=', 32000000) <20 -> ('<', 20) <=55 -> ('<=', 55) >60 -> ('>', 60) """ m = re.match(r"(>=|<=|>|<|=|!=)\s*([\d.]+)", str(expr).strip()) if m: return m.group(1), float(m.group(2)) return None, None def _check_threshold(actual: float, threshold_expr: str, level: str) -> tuple: """检查实际值是否触发阈值,返回 (触发, 消息)""" if not threshold_expr or actual is None: return False, "" op, val = _parse_threshold(threshold_expr) if op is None: return False, "" triggered = False if op == ">=" and actual >= val: triggered = True elif op == "<=" and actual <= val: triggered = True elif op == ">" and actual > val: triggered = True elif op == "<" and actual < val: triggered = True elif op == "=" and actual == val: triggered = True if triggered: level_names = {"green": "正常", "yellow": "预警", "red": "紧急"} msg = (f"KPI当前值 {actual:.2f},触发{level_names.get(level, level)}阈值 " f"({threshold_expr})") return True, msg return False, "" def run_alert_check(db_session, period: str = None) -> int: """检查所有KPI的实际值是否触发预警,返回新生成的预警数""" if period is None: period = datetime.now().strftime("%Y-%m") 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 # 获取该KPI当前期间的最新值 latest = db_session.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period, ).order_by(KPIValue.calculated_at.desc()).first() if not latest or latest.actual_value is None: continue actual = latest.actual_value # 从绿到红检查(绿灯最高优先级——满足即止) for level in ["green", "yellow", "red"]: expr = thr[level] if not expr: continue triggered, msg = _check_threshold(actual, expr, level) if triggered: # 检查是否已有该期间该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: # 已有预警,跳过 break # 创建新预警 alert = KPIAlert( kpi_id=kpi.id, kpi_value_id=latest.id, alert_level=level, alert_message=f"{kpi.kpi_name}[{period}] {msg}", status="pending", ) db_session.add(alert) new_count += 1 logger.info(f" 新增预警 [{level}] {kpi.kpi_name}: {msg}") break # 只取最高级别 elif level == "red" and not triggered: # 红没触发,如果已有红色预警但当前不满足,自动降级或关闭 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: existing_red.status = "resolved" existing_red.resolution = "自动解除: 当前值不再触发红色阈值" existing_red.resolved_at = datetime.now() logger.info(f" 自动解除预警 [red] {kpi.kpi_name}") db_session.commit() return new_count def generate_and_push(db_session) -> dict: """生成预警并推送,返回统计""" period = datetime.now().strftime("%Y-%m") # 1. 生成预警 logger.info(f"开始预警检查 ({period})...") new_count = run_alert_check(db_session, period) logger.info(f"预警检查完成: 新增 {new_count} 条") # 2. 推送 pushed = 0 if new_count > 0: 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": period, "new_alerts": new_count, "pushed": pushed} if __name__ == "__main__": logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) db = get_session_local()() try: result = generate_and_push(db) print(f"预警检查: {result['new_alerts']}条新预警 / {result['pushed']}条已推送") finally: db.close()