feat: KPI通用化 — 五档评分/三级分解树/因果链追踪
This commit is contained in:
Hermes CI Fix
+285
-1
@@ -8,7 +8,7 @@ import json
|
||||
|
||||
from app.database import get_db
|
||||
from app.auth_middleware import require_auth, require_role, filter_kpis_by_role, kpi_visible_dims
|
||||
from app.models import StrategicMap, MapObjective, KPIDefinition, KPIValue, KPIAlert, OperationLog, Entity
|
||||
from app.models import StrategicMap, MapObjective, KPIDefinition, KPIValue, KPIAlert, OperationLog, Entity, KPICausality, KPIHierarchy
|
||||
|
||||
router = APIRouter(prefix="/api/cma/kpis", tags=["KPI字典"],
|
||||
dependencies=[Depends(require_role("ceo", "finance", "business", "it"))],
|
||||
@@ -97,6 +97,290 @@ def get_kpi_categories(current_user = Depends(require_auth), db: Session = Depen
|
||||
return {"tree": tree, "total": sum(dim_counts.values())}
|
||||
|
||||
|
||||
# ============================================================
|
||||
# KPI-5: 五档评分引擎（静态路由必须在动态/{kpi_id}之前）
|
||||
# ============================================================
|
||||
|
||||
def _calc_five_tier_score(current_value, target_value):
|
||||
"""五档评分：1-5分"""
|
||||
if current_value is None or target_value is None or target_value == 0:
|
||||
return None, "info"
|
||||
ratio = current_value / target_value
|
||||
if ratio >= 1.2:
|
||||
return 5, "success" # 卓越
|
||||
elif ratio >= 1.0:
|
||||
return 4, "success" # 达标
|
||||
elif ratio >= 0.8:
|
||||
return 3, "warning" # 预警
|
||||
elif ratio >= 0.5:
|
||||
return 2, "danger" # 危险
|
||||
else:
|
||||
return 1, "danger" # 失效
|
||||
|
||||
|
||||
@router.get("/score")
|
||||
def get_kpi_score(
|
||||
entity_id: int = Query(1, ge=1),
|
||||
period: Optional[str] = None,
|
||||
db: Session = Depends(get_db),
|
||||
current_user = Depends(require_auth),
|
||||
):
|
||||
"""五档评分引擎 - 返回各KPI评分和BSC四层汇总
|
||||
评分: 5卓越(≥1.2×目标) 4达标(≥目标) 3预警(≥0.8×目标) 2危险(≥0.5×目标) 1失效(<0.5×目标)
|
||||
"""
|
||||
# 获取该企业所有活跃KPI
|
||||
kpis = db.query(KPIDefinition).filter(
|
||||
KPIDefinition.status == "active",
|
||||
KPIDefinition.entity_id == entity_id,
|
||||
).all()
|
||||
|
||||
if not kpis:
|
||||
return {"entity_id": entity_id, "kpis": [], "layers": {}, "overall": None}
|
||||
|
||||
# 获取企业信息
|
||||
ent = db.query(Entity).filter(Entity.id == entity_id).first()
|
||||
entity_info = {"id": ent.id, "name": ent.name, "short_name": ent.short_name} if ent else {"id": entity_id}
|
||||
|
||||
# 单个KPI评分
|
||||
kpi_scores = []
|
||||
for k in kpis:
|
||||
# 取最新实际值
|
||||
val_query = db.query(KPIValue).filter(
|
||||
KPIValue.kpi_id == k.id,
|
||||
KPIValue.actual_value.isnot(None),
|
||||
)
|
||||
if period:
|
||||
val_query = val_query.filter(KPIValue.period == period)
|
||||
latest_val = val_query.order_by(KPIValue.period.desc()).first()
|
||||
|
||||
current_val = latest_val.actual_value if latest_val else None
|
||||
score, status = _calc_five_tier_score(current_val, k.target_value)
|
||||
|
||||
kpi_scores.append({
|
||||
"kpi_id": k.id,
|
||||
"kpi_code": k.kpi_code,
|
||||
"kpi_name": k.kpi_name,
|
||||
"dimension": k.dimension,
|
||||
"target_value": k.target_value,
|
||||
"current_value": current_val,
|
||||
"score": score,
|
||||
"status": status,
|
||||
"unit": k.unit,
|
||||
"weight": 10,
|
||||
"period": latest_val.period if latest_val else None,
|
||||
})
|
||||
|
||||
# BSC四层汇总
|
||||
layer_map = {
|
||||
"finance": {"label": "财务", "order": 0},
|
||||
"customer": {"label": "客户", "order": 1},
|
||||
"process": {"label": "流程", "order": 2},
|
||||
"learning": {"label": "学习成长", "order": 3},
|
||||
}
|
||||
layers = {}
|
||||
total_weighted_score = 0
|
||||
total_weight = 0
|
||||
|
||||
for dim_key, dim_info in layer_map.items():
|
||||
layer_kpis = [s for s in kpi_scores if s["dimension"] == dim_key and s["score"] is not None]
|
||||
if not layer_kpis:
|
||||
layers[dim_key] = {"label": dim_info["label"], "score": None, "status": "info", "kpi_count": 0, "weighted_score": None}
|
||||
continue
|
||||
|
||||
w = sum(k["weight"] for k in layer_kpis)
|
||||
ws = sum(k["score"] * k["weight"] for k in layer_kpis)
|
||||
avg_score = ws / w if w > 0 else None
|
||||
avg_status = "success" if avg_score and avg_score >= 4 else ("warning" if avg_score and avg_score >= 3 else "danger") if avg_score else "info"
|
||||
|
||||
layers[dim_key] = {
|
||||
"label": dim_info["label"],
|
||||
"score": round(avg_score, 2) if avg_score else None,
|
||||
"status": avg_status,
|
||||
"kpi_count": len(layer_kpis),
|
||||
"weighted_score": round(avg_score, 2) if avg_score else None,
|
||||
}
|
||||
|
||||
if avg_score:
|
||||
total_weighted_score += avg_score * len(layer_kpis)
|
||||
total_weight += len(layer_kpis)
|
||||
|
||||
# 综合得分
|
||||
overall_score = round(total_weighted_score / total_weight, 2) if total_weight > 0 else None
|
||||
overall_status = "success" if overall_score and overall_score >= 4 else ("warning" if overall_score and overall_score >= 3 else "danger") if overall_score else "info"
|
||||
|
||||
return {
|
||||
"entity": entity_info,
|
||||
"kpis": kpi_scores,
|
||||
"layers": layers,
|
||||
"overall": {"score": overall_score, "status": overall_status},
|
||||
}
|
||||
|
||||
|
||||
# ============================================================
|
||||
# KPI-6: KPI三级分解树
|
||||
# ============================================================
|
||||
|
||||
@router.get("/hierarchy")
|
||||
def get_kpi_hierarchy(
|
||||
entity_id: int = Query(1, ge=1),
|
||||
kpi_id: Optional[int] = None,
|
||||
db: Session = Depends(get_db),
|
||||
current_user = Depends(require_auth),
|
||||
):
|
||||
"""KPI三级分解树：公司→部门→个人"""
|
||||
query = db.query(KPIHierarchy).filter(KPIHierarchy.entity_id == entity_id)
|
||||
if kpi_id is not None:
|
||||
query = query.filter(
|
||||
(KPIHierarchy.parent_kpi_id == kpi_id) | (KPIHierarchy.child_kpi_id == kpi_id)
|
||||
)
|
||||
relations = query.order_by(KPIHierarchy.level).all()
|
||||
|
||||
if not relations:
|
||||
# 无层级数据，返回公司级KPI作为根节点
|
||||
kpis = db.query(KPIDefinition).filter(
|
||||
KPIDefinition.status == "active",
|
||||
KPIDefinition.entity_id == entity_id,
|
||||
).limit(20).all()
|
||||
return {
|
||||
"entity_id": entity_id,
|
||||
"tree": [{"id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name,
|
||||
"dimension": k.dimension, "level": 1, "children": []} for k in kpis],
|
||||
"total": len(kpis),
|
||||
}
|
||||
|
||||
# 构建树
|
||||
kpi_ids = set()
|
||||
for r in relations:
|
||||
kpi_ids.add(r.parent_kpi_id)
|
||||
kpi_ids.add(r.child_kpi_id)
|
||||
|
||||
kpi_map = {}
|
||||
for kid in kpi_ids:
|
||||
k = db.query(KPIDefinition).filter(KPIDefinition.id == kid).first()
|
||||
if k:
|
||||
kpi_map[kid] = {"id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name,
|
||||
"dimension": k.dimension, "level": None, "children": []}
|
||||
|
||||
# 分配层级
|
||||
for r in relations:
|
||||
if r.parent_kpi_id in kpi_map:
|
||||
kpi_map[r.parent_kpi_id]["level"] = 1 # 公司级
|
||||
if r.child_kpi_id in kpi_map:
|
||||
current_level = kpi_map[r.child_kpi_id].get("level")
|
||||
new_level = r.level or 2
|
||||
if current_level is None or current_level > new_level:
|
||||
kpi_map[r.child_kpi_id]["level"] = new_level
|
||||
|
||||
# 构造父子关系
|
||||
tree = []
|
||||
added = set()
|
||||
for r in relations:
|
||||
parent = kpi_map.get(r.parent_kpi_id)
|
||||
child = kpi_map.get(r.child_kpi_id)
|
||||
if parent and child:
|
||||
child_node = dict(child)
|
||||
child_node["weight"] = r.weight
|
||||
child_node["child_name"] = r.child_name
|
||||
# 避免重复添加
|
||||
child_key = r.child_kpi_id
|
||||
existing_child = next(
|
||||
(c for c in parent["children"] if c["id"] == child_key), None
|
||||
)
|
||||
if not existing_child:
|
||||
parent["children"].append(child_node)
|
||||
|
||||
# 收集顶级节点（有子节点且未被引用的parent）
|
||||
all_child_ids = {r.child_kpi_id for r in relations}
|
||||
for r in relations:
|
||||
pid = r.parent_kpi_id
|
||||
if pid not in all_child_ids or pid == (kpi_id if kpi_id else -1):
|
||||
if pid not in added and pid in kpi_map:
|
||||
tree.append(kpi_map[pid])
|
||||
added.add(pid)
|
||||
|
||||
# 如果kpi_id指定，返回该节点为根的子树
|
||||
if kpi_id is not None and kpi_id in kpi_map:
|
||||
root = kpi_map[kpi_id]
|
||||
return {"entity_id": entity_id, "tree": [root], "total": len(tree)}
|
||||
|
||||
# 否则按level排序
|
||||
tree.sort(key=lambda n: (n.get("level") or 99, n["kpi_code"]))
|
||||
|
||||
return {"entity_id": entity_id, "tree": tree, "total": len(tree)}
|
||||
|
||||
|
||||
# ============================================================
|
||||
# KPI-8: KPI因果链追踪
|
||||
# ============================================================
|
||||
|
||||
@router.get("/{kpi_id}/causality-chain")
|
||||
def get_kpi_causality_chain(
|
||||
kpi_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
current_user = Depends(require_auth),
|
||||
):
|
||||
"""KPI因果链追踪 — 返回单个KPI的上下游因果链"""
|
||||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||||
if not kpi:
|
||||
raise HTTPException(404, "KPI不存在")
|
||||
|
||||
# 上游（驱动当前KPI的因子）
|
||||
upstream = db.query(KPICausality).filter(KPICausality.target_kpi_id == kpi_id).all()
|
||||
upstream_list = []
|
||||
for c in upstream:
|
||||
src = db.query(KPIDefinition).filter(KPIDefinition.id == c.source_kpi_id).first()
|
||||
if src:
|
||||
upstream_list.append({
|
||||
"causality_id": c.id,
|
||||
"kpi_id": src.id,
|
||||
"kpi_code": src.kpi_code,
|
||||
"kpi_name": src.kpi_name,
|
||||
"dimension": src.dimension,
|
||||
"layer": src.dimension,
|
||||
"strength": c.strength,
|
||||
"lag_months": c.lag_months,
|
||||
"direction": c.direction,
|
||||
"formula": c.formula,
|
||||
})
|
||||
|
||||
# 下游（当前KPI影响的指标）
|
||||
downstream = db.query(KPICausality).filter(KPICausality.source_kpi_id == kpi_id).all()
|
||||
downstream_list = []
|
||||
for c in downstream:
|
||||
tgt = db.query(KPIDefinition).filter(KPIDefinition.id == c.target_kpi_id).first()
|
||||
if tgt:
|
||||
downstream_list.append({
|
||||
"causality_id": c.id,
|
||||
"kpi_id": tgt.id,
|
||||
"kpi_code": tgt.kpi_code,
|
||||
"kpi_name": tgt.kpi_name,
|
||||
"dimension": tgt.dimension,
|
||||
"layer": tgt.dimension,
|
||||
"strength": c.strength,
|
||||
"lag_months": c.lag_months,
|
||||
"direction": c.direction,
|
||||
"formula": c.formula,
|
||||
})
|
||||
|
||||
return {
|
||||
"kpi": {
|
||||
"id": kpi.id,
|
||||
"kpi_code": kpi.kpi_code,
|
||||
"kpi_name": kpi.kpi_name,
|
||||
"dimension": kpi.dimension,
|
||||
"layer": kpi.dimension,
|
||||
},
|
||||
"drives": downstream_list,
|
||||
"driven_by": upstream_list,
|
||||
"total_upstream": len(upstream_list),
|
||||
"total_downstream": len(downstream_list),
|
||||
}
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 动态路由（必须在静态路由之后）
|
||||
# ============================================================
|
||||
|
||||
@router.get("/{kpi_id}")
|
||||
def get_kpi(kpi_id: int, db: Session = Depends(get_db)):
|
||||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||||
|
||||
Reference in New Issue
Block a user