Files
cma-management/backend/app/api/bot_bridge.py
T
Hermes CI Fix 61936302b2 feat(KR): KR完整修复 — krs表打通+方向符号operator+权重自由输入
- 后端okr.py: KR读取从ActionPlan改为krs表, 新增KR CRUD API(POST/PUT/DELETE /okr/{objective_id}/krs) + 批量sync
- operator方向符号: krs表加operator/tolerance/weight/sort_order/monthly_milestones列
- progress方向感知计算(>=/>: current/target, <=/<: target/current, =: 容差), 达成→status=achieved
- 关联KPI自动继承方向(threshold_green解析: F_COST_RATIO<=18等)
- maps.py: 保存地图时自动同步objectives+krs表, JSON→krs数据迁移脚本
- 前端: 权重下拉改自由数字输入(可小数33.33) + 方向选择器(≥/≤/>/</=) + 自动平分按钮 + KPI方向继承提示
- pytest: 10个新测试(krs CRUD/方向感知/权重校验/多租户隔离) + 更新旧KR测试
- 迁移: 现有strategic_maps JSON 12条KR已写入krs表
2026-08-27 10:59:56 +08:00

739 lines
27 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
CMA BOT API桥接层 — 供财务BOT/店研学BOT调用
无需用户登录,使用 BOT API Key 认证
"""
import os, json, logging
from fastapi import APIRouter, Depends, HTTPException, Query, Header, UploadFile, File
from sqlalchemy.orm import Session
from sqlalchemy import func, desc
from datetime import datetime
from typing import Optional
from app.database import get_db
from app.models import (
User, StrategicMap, KPIDefinition, KPITemplate, KPIValue,
DataSourceConfig, KPIAlert, OperationLog, NotificationChannel,
NotificationLog, RolePermission, ActionPlan, OrgNode,
StrategicMapVersion, MapObjective, Objective, KR,
)
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")
router = APIRouter(prefix="/api/cma/bot", tags=["BOT桥接"])
# ── BOT API Key 配置 ──
_BOT_API_KEYS = {}
def _load_bot_keys():
global _BOT_API_KEYS
raw = os.getenv("CMA_BOT_API_KEYS", "")
if not raw:
_BOT_API_KEYS = {
"cma-bot-finance-2026": {"role": "finance", "name": "财务BOT"},
"cma-bot-shop-2026": {"role": "business", "name": "店研学BOT"},
"cma-bot-admin-2026": {"role": "ceo", "name": "管理BOT"},
}
else:
try:
_BOT_API_KEYS = json.loads(raw)
except:
_BOT_API_KEYS = {}
_load_bot_keys()
def verify_bot_key(x_bot_key: str = Header(None, alias="X-BOT-KEY")):
if not x_bot_key or x_bot_key not in _BOT_API_KEYS:
raise HTTPException(401, "无效的BOT API Key")
bot_info = _BOT_API_KEYS[x_bot_key]
logger.info(f"BOT访问: {bot_info['name']} ({bot_info['role']})")
return bot_info
# ═══════════════ 通用工具 ═══════════════
def _float(v):
if v is None: return None
try: return float(v)
except: return None
def _safe_iso(dt):
if dt is None: return None
try: return dt.isoformat() if hasattr(dt, 'isoformat') else str(dt)
except: return None
def _model_dict(obj, fields: dict):
"""安全地将模型字段转为dict"""
result = {}
for key, attr in fields.items():
v = getattr(obj, attr, None)
if isinstance(v, float):
result[key] = _float(v)
else:
result[key] = v
return result
# ═══════════════ 端点 ═══════════════
@router.get("/ping")
def ping():
return {"status": "ok", "version": "1.0", "timestamp": datetime.now().isoformat()}
# ── 总览 ──
@router.get("/overview")
def bot_overview(
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
"""系统总览 — BOT首选入口"""
return {
"bot": bot,
"timestamp": datetime.now().isoformat(),
"stats": {
"kpis_total": db.query(func.count(KPIDefinition.id)).filter(KPIDefinition.status == "active").scalar() or 0,
"alerts_open": db.query(func.count(KPIAlert.id)).filter(KPIAlert.status == "pending").scalar() or 0,
"maps_total": db.query(func.count(StrategicMap.id)).scalar() or 0,
"budget_plans": db.query(func.count(BudgetPlan.id)).scalar() or 0,
"action_plans_pending": db.query(func.count(ActionPlan.id)).filter(ActionPlan.status.in_(["pending", "in_progress"])).scalar() or 0,
"data_sources": db.query(func.count(DataSourceConfig.id)).scalar() or 0,
"users": db.query(func.count(User.id)).scalar() or 0,
"org_nodes": db.query(func.count(OrgNode.id)).scalar() or 0,
}
}
# ── KPI ──
@router.get("/kpis")
def bot_kpis(
dimension: Optional[str] = Query(None),
status: str = Query("active"),
limit: int = Query(200, le=1000),
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
query = db.query(KPIDefinition).filter(KPIDefinition.status == status)
if dimension:
query = query.filter(KPIDefinition.dimension == dimension)
kpis = query.order_by(KPIDefinition.dimension, KPIDefinition.kpi_code).limit(limit).all()
results = []
for k in kpis:
latest = db.query(KPIValue).filter(KPIValue.kpi_id == k.id)\
.order_by(KPIValue.period.desc()).first()
results.append({
"id": k.id, "name": k.kpi_name, "code": k.kpi_code,
"dimension": k.dimension, "category": k.category,
"unit": k.unit, "formula": k.formula,
"frequency": k.frequency, "data_source_type": k.data_source_type,
"target_value": _float(k.target_value),
"threshold_green": k.threshold_green,
"threshold_yellow": k.threshold_yellow,
"threshold_red": k.threshold_red,
"responsible_dept": k.responsible_dept, "owner": k.responsible_user,
"objective": k.objective, "formula_desc": k.formula_desc,
"latest_value": _float(latest.actual_value) if latest else None,
"latest_period": latest.period if latest else None,
"status": k.status,
})
return {"total": len(results), "items": results}
@router.get("/kpis/{kpi_id}/history")
def bot_kpi_history(
kpi_id: int, limit: int = Query(12, le=60),
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
if not kpi:
raise HTTPException(404, "KPI不存在")
values = db.query(KPIValue).filter(KPIValue.kpi_id == kpi_id)\
.order_by(KPIValue.period.desc()).limit(limit).all()
return {
"kpi": {"id": kpi.id, "name": kpi.kpi_name, "code": kpi.kpi_code, "unit": kpi.unit},
"values": [
{
"period": v.period,
"actual": _float(v.actual_value),
"source_type": v.source_type,
"data_status": v.data_status,
} for v in values
],
}
# ── 战略地图 ──
@router.get("/strategic-maps")
def bot_maps(
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
maps = db.query(StrategicMap).order_by(StrategicMap.id.desc()).all()
result = []
for m in maps:
objectives = db.query(MapObjective).filter(MapObjective.map_id == m.id).all()
dims = {}
for obj in objectives:
dk = obj.dimension_key
if dk not in dims:
dims[dk] = []
dims[dk].append({"id": obj.id, "name": obj.name, "description": obj.description})
result.append({
"id": m.id, "title": m.title, "version": m.version,
"status": m.status, "dimensions": m.dimensions,
"objectives": dims,
"created_at": _safe_iso(m.created_at),
"updated_at": _safe_iso(m.updated_at),
})
return {"total": len(result), "items": result}
# ── 预警 ──
@router.get("/alerts")
def bot_alerts(
status: str = Query("pending"),
level: Optional[str] = Query(None),
limit: int = Query(50, le=200),
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
query = db.query(KPIAlert)
query = query.filter(KPIAlert.status == status)
if level:
query = query.filter(KPIAlert.alert_level == level)
alerts = query.order_by(KPIAlert.created_at.desc()).limit(limit).all()
return {
"total": len(alerts),
"items": [
{
"id": a.id, "kpi_id": a.kpi_id,
"level": a.alert_level, "message": a.alert_message,
"status": a.status, "assignee": a.assignee,
"resolution": a.resolution,
"created_at": _safe_iso(a.created_at),
"resolved_at": _safe_iso(a.resolved_at),
} for a in alerts
],
}
# ── 预算 ──
@router.get("/budget/plans")
def bot_budget_plans(
year: Optional[int] = Query(None),
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
query = db.query(BudgetPlan)
if year:
query = query.filter(BudgetPlan.budget_year == year)
plans = query.order_by(BudgetPlan.period.desc()).limit(200).all()
return {
"total": len(plans),
"items": [
{
"id": p.id, "kpi_id": p.kpi_id,
"period": p.period,
"budget_value": _float(p.budget_value),
"year": p.budget_year, "month": p.budget_month,
"version": p.version, "status": p.status,
"remark": p.remark,
} for p in plans
],
}
# ── 成本 ──
@router.get("/cost/standard")
def bot_standard_costs(
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
costs = db.query(StandardCost).filter(StandardCost.status == "active").limit(200).all()
return {
"total": len(costs),
"items": [
{
"id": c.id, "product_code": c.product_code,
"product_name": c.product_name, "cost_type": c.cost_type,
"item_name": c.item_name,
"standard_quantity": _float(c.standard_quantity),
"unit": c.unit,
"standard_price": _float(c.standard_price),
"standard_cost": _float(c.standard_cost),
"version": c.version, "remark": c.remark,
} for c in costs
],
}
@router.get("/cost/actual")
def bot_actual_costs(
period: Optional[str] = Query(None),
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
query = db.query(ActualCost)
if period:
query = query.filter(ActualCost.period == period)
costs = query.order_by(ActualCost.period.desc()).limit(200).all()
return {
"total": len(costs),
"items": [
{
"id": c.id, "period": c.period,
"product_code": c.product_code,
"product_name": c.product_name,
"cost_type": c.cost_type, "item_name": c.item_name,
"actual_quantity": _float(c.actual_quantity),
"actual_price": _float(c.actual_price),
"actual_cost": _float(c.actual_cost),
} for c in costs
],
}
# ── 行动方案 ──
@router.get("/actions")
def bot_actions(
status: Optional[str] = Query(None),
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
query = db.query(ActionPlan)
if status:
query = query.filter(ActionPlan.status == status)
plans = query.order_by(ActionPlan.priority, ActionPlan.id.desc()).limit(100).all()
return {
"total": len(plans),
"items": [
{
"id": p.id, "title": p.title,
"description": p.description, "kpi_id": p.kpi_id,
"assignee": p.assignee, "priority": p.priority,
"status": p.status, "progress": p.progress,
"due_date": _safe_iso(p.due_date),
"created_at": _safe_iso(p.created_at),
} for p in plans
],
}
# ── 组织 ──
@router.get("/organization")
def bot_org(
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
nodes = db.query(OrgNode).order_by(OrgNode.level, OrgNode.sort_order).all()
return {
"total": len(nodes),
"items": [
{
"id": n.id, "name": n.name,
"parent_id": n.parent_id, "level": n.level,
"code": n.code, "sort_order": n.sort_order,
"enabled": n.enabled,
} for n in nodes
],
}
# ── 数据源 ──
@router.get("/data-sources")
def bot_data_sources(
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
sources = db.query(DataSourceConfig).all()
return {
"total": len(sources),
"items": [
{
"id": s.id, "name": s.name,
"source_type": s.source_type,
"api_endpoint": s.api_endpoint,
"sync_type": s.sync_type,
"status": s.status,
"last_sync_at": _safe_iso(s.last_sync_at),
} for s in sources
],
}
# ── 用户 ──
@router.get("/users")
def bot_users(
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
users = db.query(User).all()
return {
"total": len(users),
"items": [
{"id": u.id, "username": u.username, "name": u.name,
"role": u.role, "phone": u.phone}
for u in users
],
}
# ── 统一查询(BOT首选) ──
@router.get("/query")
def bot_query(
q: str = Query("overview", description="overview/kpis/alerts/maps/budget/cost/actions/all"),
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
"""统一查询入口 — BOT用这个一次拿完需要的数据"""
result = {"bot": bot["name"], "role": bot["role"], "timestamp": datetime.now().isoformat()}
if q in ("overview", "all"):
result["overview"] = {
"kpis": db.query(func.count(KPIDefinition.id)).filter(KPIDefinition.status == "active").scalar() or 0,
"alerts_open": db.query(func.count(KPIAlert.id)).filter(KPIAlert.status == "pending").scalar() or 0,
"maps": db.query(func.count(StrategicMap.id)).scalar() or 0,
"budget_plans": db.query(func.count(BudgetPlan.id)).scalar() or 0,
}
if q in ("kpis", "all"):
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").limit(100).all()
result["kpis"] = [
{"id": k.id, "name": k.kpi_name, "code": k.kpi_code,
"dimension": k.dimension, "target": _float(k.target_value), "unit": k.unit}
for k in kpis
]
if q in ("alerts", "all"):
alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending")\
.order_by(KPIAlert.created_at.desc()).limit(20).all()
result["alerts"] = [
{"id": a.id, "level": a.alert_level, "message": a.alert_message,
"kpi_id": a.kpi_id, "created_at": _safe_iso(a.created_at)}
for a in alerts
]
if q in ("maps", "all"):
maps = db.query(StrategicMap).limit(10).all()
result["maps"] = [
{"id": m.id, "title": m.title, "status": m.status,
"version": m.version, "created_at": _safe_iso(m.created_at)}
for m in maps
]
if q in ("budget", "all"):
plans = db.query(BudgetPlan).limit(50).all()
result["budget"] = [
{"id": p.id, "period": p.period, "budget_value": _float(p.budget_value),
"year": p.budget_year, "month": p.budget_month, "status": p.status,
"kpi_id": p.kpi_id}
for p in plans
]
if q in ("cost", "all"):
sc = db.query(StandardCost).limit(50).all()
result["costs"] = [
{"id": c.id, "product": c.product_name, "type": c.cost_type,
"standard": _float(c.standard_cost), "unit": c.unit}
for c in sc
]
if q in ("okr", "all"):
objs = db.query(Objective).filter(Objective.status == "active").all()
result["okr"] = []
for o in objs:
# KR完整修复(2026-08-27): 从krs表读取
krs = db.query(KR).filter(KR.objective_id == o.id).all()
result["okr"].append({
"id": o.id, "title": o.title, "quarter": o.quarter,
"dimension": o.dimension, "progress": o.progress,
"confidence": o.confidence,
"key_results": [
{"title": kr.title, "status": kr.status, "progress": kr.progress,
"operator": kr.operator, "target_value": _float(kr.target_value)}
for kr in krs
]
})
if q in ("actions", "all"):
acts = db.query(ActionPlan).limit(30).all()
result["actions"] = [
{"id": a.id, "title": a.title, "status": a.status,
"progress": a.progress, "assignee": a.assignee}
for a in acts
]
return result
@router.post("/import")
def bot_import_excel(
file: UploadFile = File(...),
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
"""Bot上传Excel导入KPI数据到CMA"""
import pandas as pd, io, hashlib
from app.models import KPIValue
try:
content = file.file.read()
df = pd.read_excel(io.BytesIO(content))
except Exception as e:
raise HTTPException(400, f"Excel解析失败: {e}")
finally:
file.file.close()
# 智能识别列名
col_map = {"kpi_code": ["kpi_code", "KPI编码", "指标编码", "code"],
"period": ["period", "期间", "月份", "month", "日期"],
"actual_value": ["actual_value", "实际值", "值", "金额", "value", "amount"]}
mapped = {}
for field, aliases in col_map.items():
for col in df.columns:
if str(col).strip() in aliases or str(col).strip().lower() in aliases:
mapped[field] = str(col).strip()
break
if "actual_value" not in mapped:
raise HTTPException(400, f"无法识别数值列,支持的列名: {col_map['actual_value']}")
if "kpi_code" not in mapped:
raise HTTPException(400, f"无法识别KPI编码列,支持的列名: {col_map['kpi_code']}")
kpi_col = mapped["kpi_code"]
val_col = mapped["actual_value"]
period_col = mapped.get("period")
count = 0
errors = []
for idx, row in df.iterrows():
try:
kpi_code = str(row[kpi_col]).strip()
val = float(row[val_col])
period = str(row[period_col]).strip() if period_col else datetime.now().strftime("%Y-%m")
kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == kpi_code).first()
if not kpi:
errors.append(f"第{idx+2}行: KPI编码 '{kpi_code}' 不存在,跳过")
continue
kv = KPIValue(kpi_id=kpi.id, entity_id=kpi.entity_id, period=period, actual_value=val,
source_batch=hashlib.md5(f"{datetime.now()}".encode()).hexdigest()[:12])
db.add(kv)
count += 1
except Exception as e:
errors.append(f"第{idx+2}行: {e}")
db.commit()
return {"ok": True, "imported": count, "errors": len(errors), "detail": errors[:5]}
# ── 自然语言查询 ──
@router.post("/okr/create")
def bot_okr_create(
title: str = Query(...),
quarter: str = Query(...),
dimension: Optional[str] = Query(None),
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
"""Bot创建OKR目标"""
from app.models import Objective
obj = Objective(title=title, quarter=quarter, dimension=dimension, owner=bot["name"])
db.add(obj)
db.commit()
db.refresh(obj)
return {"ok": True, "id": obj.id, "title": obj.title, "confidence": obj.confidence}
@router.get("/okr/list")
def bot_okr_list(
quarter: Optional[str] = Query(None),
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
"""Bot列出OKR(含KR进度)"""
from app.models import Objective
q = db.query(Objective)
if quarter:
q = q.filter(Objective.quarter == quarter)
objs = q.order_by(Objective.quarter.desc()).all()
return {"total": len(objs), "items": [
{"id": o.id, "title": o.title, "quarter": o.quarter,
"dimension": o.dimension, "progress": o.progress,
"confidence": o.confidence, "status": o.status,
"kr_count": db.query(func.count(ActionPlan.id)).filter(ActionPlan.objective_id == o.id).scalar() or 0}
for o in objs
]}
@router.get("/nlp")
def bot_nlp(
intent: str = Query("overview"),
bot: dict = Depends(verify_bot_key),
db: Session = Depends(get_db),
):
"""
自然语言意图映射:
overview/总览/finance/财务/alerts/预警/budget/预算/cost/成本/maps/战略/actions/行动
"""
m = {
"总览": "overview", "驾驶舱": "overview",
"财务": "finance", "财务状况": "finance",
"预警": "alerts", "风险": "alerts",
"预算": "budget", "预算执行": "budget",
"成本": "cost", "成本分析": "cost",
"战略": "maps", "战略地图": "maps",
"行动": "actions", "改善": "actions",
"okr": "okr", "目标": "okr", "季度目标": "okr",
}
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"}