diff --git a/backend/app/api/bot_bridge.py b/backend/app/api/bot_bridge.py index dc4a39d9..e0625c7c 100644 --- a/backend/app/api/bot_bridge.py +++ b/backend/app/api/bot_bridge.py @@ -17,6 +17,8 @@ from app.models import ( ) from app.models.budget_plan import BudgetPlan from app.models.cost_model import StandardCost, ActualCost, AbcActivity, AbcAllocation +from app.models import KPICausality +import json logger = logging.getLogger("cma.bot_bridge") @@ -601,3 +603,134 @@ def bot_nlp( } resolved = m.get(intent, intent) return bot_query(q=resolved, bot=bot, db=db) + + +# ════════════════════════════════════════════════════════════ +# 聚合接口（Agent化生产链路 · 行动1, 2026-08-25） +# 建议A: KPI值更新联动预警检查 | 建议B: KPI创建联动关联 +# ════════════════════════════════════════════════════════════ + +@router.post("/kpi-value-with-check") +def bot_kpi_value_with_check(data: dict, db: Session = Depends(get_db), bot: dict = Depends(verify_bot_key)): + """聚合A: 写KPI值 + 自动跑该KPI预警检查（Agent一次调用，免自拼check-all） + body: {kpi_id, actual_value, period?, entity_id?, run_check?}""" + kpi_id = data.get("kpi_id") + actual_value = data.get("actual_value") + period = data.get("period") + entity_id = int(data.get("entity_id") or 1) + run_check = bool(data.get("run_check", True)) + if not kpi_id or actual_value is None: + raise HTTPException(400, "kpi_id 和 actual_value 必填") + # 校验 KPI 归属（多租户） + kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() + if not kpi or kpi.entity_id != entity_id: + raise HTTPException(404, "KPI不存在") + period = period or datetime.now().strftime("%Y-%m") + + # ① 写值（upsert） + existing = db.query(KPIValue).filter( + KPIValue.kpi_id == kpi_id, KPIValue.period == period).first() + if existing: + existing.actual_value = float(actual_value) + existing.source_type = "bot" + kv = existing + else: + kv = KPIValue(kpi_id=kpi_id, entity_id=entity_id, period=period, + actual_value=float(actual_value), source_type="bot", data_status="verified") + db.add(kv) + db.commit() + db.refresh(kv) + + # ② 跑该KPI关联的预警规则（复用 alert_rules 检查函数，非全量） + alerts = [] + if run_check: + from app.api.alert_rules import AlertRule, _check_static, _check_dynamic, _check_trend + rules = db.query(AlertRule).filter( + AlertRule.kpi_id == kpi_id, AlertRule.entity_id == entity_id, + AlertRule.enabled == 1).all() + for rule in rules: + try: + params = json.loads(rule.params) if isinstance(rule.params, str) else (rule.params or {}) + value = float(actual_value) + if rule.rule_type == "static": + level, msg = _check_static(value, params, kpi) + elif rule.rule_type == "dynamic": + level, msg = _check_dynamic(kpi_id, value, params, db) + elif rule.rule_type == "trend_up": + level, msg = _check_trend(kpi_id, value, "up", params, db) + elif rule.rule_type == "trend_down": + level, msg = _check_trend(kpi_id, value, "down", params, db) + else: + continue + if level and level != "green": + dup = db.query(KPIAlert).filter( + KPIAlert.kpi_id == kpi_id, KPIAlert.kpi_value_id == kv.id, + KPIAlert.alert_level == level, KPIAlert.status == "pending").first() + if not dup: + db.add(KPIAlert(kpi_id=kpi_id, kpi_value_id=kv.id, alert_level=level, + alert_message=msg, status="pending", alert_type="bot")) + alerts.append({"rule_id": rule.id, "rule_type": rule.rule_type, + "level": level, "message": msg}) + except Exception as e: + logger.warning(f"聚合检查失败 rule={rule.id}: {e}") + db.commit() + return {"kpi_id": kpi_id, "kpi_code": kpi.kpi_code, "value": float(actual_value), + "period": period, "alerts": alerts, "status": "ok"} + + +@router.post("/kpis/create-with-links") +def bot_kpi_create_with_links(data: dict, db: Session = Depends(get_db), bot: dict = Depends(verify_bot_key)): + """聚合B: 创建KPI + 关联战略地图 + 批量因果链（Agent建KPI标准动作） + body: {kpi_code, kpi_name, dimension, entity_id?, target_value?, unit?, link_map_id?, link_causality?}""" + entity_id = int(data.get("entity_id") or 1) + from app.api.kpis import _validate_kpi_data, apply_calc_type_inference + kpi_data = {k: v for k, v in data.items() if k not in ("entity_id", "link_map_id", "link_causality")} + + # ① 创建KPI（编码唯一 + 治理校验 + 强制企业） + code = kpi_data.get("kpi_code", "") + if not code: + raise HTTPException(400, "kpi_code 必填") + if db.query(KPIDefinition).filter( + KPIDefinition.kpi_code == code, KPIDefinition.entity_id == entity_id).first(): + raise HTTPException(400, f"KPI编码 {code} 已存在") + errs = _validate_kpi_data(kpi_data, db=db, is_update=False) + if errs: + raise HTTPException(422, detail={"message": "数据校验不通过", "errors": errs}) + kpi_data["entity_id"] = entity_id + kpi_data = apply_calc_type_inference(kpi_data) + kpi = KPIDefinition(**kpi_data) + db.add(kpi) + db.commit() + db.refresh(kpi) + + # ② 关联战略地图 + link_map_id = data.get("link_map_id") + if link_map_id: + m = db.query(StrategicMap).filter( + StrategicMap.id == link_map_id, StrategicMap.entity_id == entity_id).first() + if m: + kpi.map_id = link_map_id + db.commit() + + # ③ 批量因果链（源=新KPI → 目标列表） + links = [] + for c in data.get("link_causality") or []: + tgt = c.get("target_kpi_id") + if not tgt or int(tgt) == kpi.id: + continue + tgt_kpi = db.query(KPIDefinition).filter( + KPIDefinition.id == int(tgt), KPIDefinition.entity_id == entity_id).first() + if not tgt_kpi: + continue + if db.query(KPICausality).filter( + KPICausality.source_kpi_id == kpi.id, + KPICausality.target_kpi_id == int(tgt)).first(): + continue + db.add(KPICausality(source_kpi_id=kpi.id, target_kpi_id=int(tgt), + strength=c.get("strength", 0.5), lag_months=c.get("lag_months", 1), + direction=c.get("direction", "positive"))) + links.append({"source": kpi.kpi_code, "target": tgt_kpi.kpi_code, + "strength": c.get("strength", 0.5)}) + db.commit() + return {"kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "map_id": link_map_id, + "causality_links": links, "status": "ok"}