Hermes CI Fix
3dddd36866
包含前后端完整代码： - 前端：Vue3+Vite+ElementPlus - 后端：FastAPI+SQLAlchemy - 模块：驾驶舱/KPI/战略地图/预警/预算/成本/预测/改善行动 - 当前版本：v1.0.0
310 lines
12 KiB
Python
310 lines
12 KiB
Python
"""KPI目标对齐管理 API — 管理会计OS
|
||
支持三种对齐模式：纵向分解 / 横向支撑 / BSC瀑布链
|
||
管理员可初始化选择，后续按模式运作"""
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import func
|
||
from typing import Optional, List
|
||
from datetime import datetime
|
||
import json
|
||
|
||
from app.database import get_db
|
||
from app.auth_middleware import require_auth, require_role
|
||
from app.models import KPIDefinition, OperationLog, RolePermission
|
||
|
||
router = APIRouter(prefix="/api/cma/alignment", tags=["KPI目标对齐"],
|
||
# 不设全局权限，每个接口单独控制
|
||
)
|
||
|
||
# 三种对齐模式定义
|
||
ALIGNMENT_MODES = [
|
||
{
|
||
"key": "vertical_decomposition",
|
||
"name": "纵向分解",
|
||
"description": "上级KPI直接拆分为多个下级KPI，目标值汇总等于上级目标。适用于营收、成本等可量化指标。",
|
||
"example": "公司销售总额2000万 → 区域A 800万 + 区域B 700万 + 区域C 500万",
|
||
},
|
||
{
|
||
"key": "horizontal_support",
|
||
"name": "横向支撑",
|
||
"description": "下级KPI是上级KPI的驱动因子，下级目标达成支撑上级结果。适用于复合型指标。",
|
||
"example": "销售毛利率30% ← 销售总额↑ + 成本控制↓ + 高毛利产品占比↑",
|
||
},
|
||
{
|
||
"key": "bsc_chain",
|
||
"name": "BSC瀑布链",
|
||
"description": "按平衡计分卡因果链层层传导：学习成长→内部流程→客户→财务。",
|
||
"example": "培训完成率↑ → 订单交付及时率↑ → 客户满意度↑ → 销售总额↑",
|
||
},
|
||
]
|
||
|
||
|
||
@router.get("/modes")
|
||
def list_modes():
|
||
"""返回三种对齐模式的定义（公开接口）"""
|
||
return {"modes": ALIGNMENT_MODES}
|
||
|
||
|
||
@router.get("/config")
|
||
def get_alignment_config(db: Session = Depends(get_db)):
|
||
"""获取当前系统对齐模式配置（公开接口，无需认证）"""
|
||
perm = db.query(RolePermission).filter(RolePermission.key == "alignment_config").first()
|
||
if not perm:
|
||
return {
|
||
"mode": None,
|
||
"configured": False,
|
||
"modes": ALIGNMENT_MODES,
|
||
}
|
||
return {
|
||
"mode": perm.value,
|
||
"configured": True,
|
||
"modes": ALIGNMENT_MODES,
|
||
}
|
||
|
||
|
||
@router.post("/config")
|
||
def set_alignment_config(
|
||
data: dict,
|
||
db: Session = Depends(get_db),
|
||
user = Depends(require_role("ceo", "it")),
|
||
):
|
||
"""初始化/修改系统对齐模式（CEO/IT权限）"""
|
||
mode_key = data.get("mode")
|
||
if mode_key not in [m["key"] for m in ALIGNMENT_MODES]:
|
||
raise HTTPException(400, f"无效的对齐模式: {mode_key}")
|
||
|
||
perm = db.query(RolePermission).filter(RolePermission.key == "alignment_config").first()
|
||
if perm:
|
||
perm.value = {"mode": mode_key, "set_at": datetime.now().isoformat()}
|
||
else:
|
||
perm = RolePermission(key="alignment_config", value={"mode": mode_key, "set_at": datetime.now().isoformat()})
|
||
db.add(perm)
|
||
db.commit()
|
||
|
||
return {"message": f"对齐模式已设置为: {mode_key}", "mode": mode_key}
|
||
|
||
|
||
@router.get("/tree")
|
||
def get_alignment_tree(
|
||
kpi_id: Optional[int] = Query(None),
|
||
db: Session = Depends(get_db),
|
||
user = Depends(require_auth),
|
||
):
|
||
"""获取KPI对齐关系树
|
||
|
||
根据当前系统配置的对齐模式，返回KPI的父子层级关系。
|
||
如果指定kpi_id，返回该KPI及其下级树；
|
||
如果不指定，返回整个对齐树。
|
||
"""
|
||
# 获取当前模式
|
||
config_perm = db.query(RolePermission).filter(RolePermission.key == "alignment_config").first()
|
||
mode = config_perm.value.get("mode") if config_perm else None
|
||
if not mode:
|
||
raise HTTPException(400, "系统未配置对齐模式，请先在系统设置中初始化")
|
||
|
||
# 获取所有KPI
|
||
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").order_by(KPIDefinition.kpi_code).all()
|
||
kpi_map = {k.id: k for k in kpis}
|
||
|
||
# 构建父子关系
|
||
if mode == "vertical_decomposition":
|
||
# 纵向分解：BSC编码前缀相同=同一系列
|
||
return _build_vertical_tree(kpis, kpi_id)
|
||
elif mode == "horizontal_support":
|
||
# 横向支撑：按BSC维度+类别的因果关系
|
||
return _build_horizontal_tree(kpis, kpi_id)
|
||
elif mode == "bsc_chain":
|
||
# BSC瀑布链：按维度层级传导
|
||
return _build_bsc_chain(kpis, kpi_id)
|
||
else:
|
||
raise HTTPException(400, f"未知的对齐模式: {mode}")
|
||
|
||
|
||
def _build_vertical_tree(kpis, kpi_id=None):
|
||
"""纵向分解树：按编码前缀分组，同一前缀=同一系列"""
|
||
from collections import defaultdict
|
||
|
||
# 提取前缀（如 F_REVENUE_001 → F_REVENUE）
|
||
groups = defaultdict(list)
|
||
for k in kpis:
|
||
parts = k.kpi_code.rsplit("_", 1)
|
||
prefix = parts[0] if len(parts) > 1 else k.kpi_code
|
||
groups[prefix].append(k)
|
||
|
||
def make_node(kpi):
|
||
return {
|
||
"id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name,
|
||
"dimension": kpi.dimension, "category": kpi.category,
|
||
"target_value": kpi.target_value, "unit": kpi.unit,
|
||
"children": [],
|
||
}
|
||
|
||
trees = []
|
||
# 每个前缀组中，按序号升序，第一个为父级
|
||
for prefix, group in sorted(groups.items()):
|
||
sorted_group = sorted(group, key=lambda k: k.kpi_code)
|
||
if len(sorted_group) > 1:
|
||
parent = make_node(sorted_group[0])
|
||
parent["children"] = [make_node(c) for c in sorted_group[1:]]
|
||
for c in parent["children"]:
|
||
c["alignment_type"] = "vertical_split"
|
||
c["parent_code"] = parent["kpi_code"]
|
||
parent["child_count"] = len(parent["children"])
|
||
trees.append(parent)
|
||
else:
|
||
trees.append(make_node(sorted_group[0]))
|
||
|
||
if kpi_id:
|
||
# 只返回指定KPI的子树
|
||
return _filter_tree(trees, kpi_id)
|
||
|
||
return {"mode": "vertical_decomposition", "mode_name": "纵向分解", "tree": trees, "total": len(kpis)}
|
||
|
||
|
||
def _build_horizontal_tree(kpis, kpi_id=None):
|
||
"""横向支撑树：按BSC维度因果关联"""
|
||
# 因果顺序：learning → process → customer → finance
|
||
dim_order = {"learning": 0, "process": 1, "customer": 2, "finance": 3}
|
||
dim_name = {"finance": "财务", "customer": "客户", "process": "内部流程", "learning": "学习成长"}
|
||
|
||
# 按维度分组
|
||
groups = {"finance": [], "customer": [], "process": [], "learning": []}
|
||
for k in kpis:
|
||
if k.dimension in groups:
|
||
groups[k.dimension].append(k)
|
||
|
||
def make_node(kpi):
|
||
return {
|
||
"id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name,
|
||
"dimension": kpi.dimension, "category": kpi.category,
|
||
"target_value": kpi.target_value, "unit": kpi.unit,
|
||
"children": [],
|
||
}
|
||
|
||
# 构建层级：一个维度节点包含该维度所有KPI
|
||
trees = []
|
||
for dim, ks in sorted(groups.items(), key=lambda x: dim_order.get(x[0], 9)):
|
||
if not ks:
|
||
continue
|
||
dim_node = {
|
||
"id": None,
|
||
"dimension": dim,
|
||
"kpi_name": dim_name.get(dim, dim),
|
||
"is_dimension_group": True,
|
||
"children": [make_node(k) for k in sorted(ks, key=lambda x: x.kpi_code)],
|
||
"child_count": len(ks),
|
||
}
|
||
# 建立因果关联说明
|
||
if dim == "learning":
|
||
dim_node["description"] = "驱动因素：人才培养与创新"
|
||
for c in dim_node["children"]:
|
||
c["drives"] = "internal_process"
|
||
elif dim == "process":
|
||
dim_node["description"] = "过程保障：效率与质量提升"
|
||
for c in dim_node["children"]:
|
||
c["drives"] = "customer"
|
||
elif dim == "customer":
|
||
dim_node["description"] = "市场反馈：客户规模与满意度"
|
||
for c in dim_node["children"]:
|
||
c["drives"] = "finance"
|
||
elif dim == "finance":
|
||
dim_node["description"] = "结果指标：收入与盈利"
|
||
for c in dim_node["children"]:
|
||
c["drives"] = None
|
||
trees.append(dim_node)
|
||
|
||
if kpi_id:
|
||
return _filter_tree(trees, kpi_id)
|
||
|
||
return {
|
||
"mode": "horizontal_support",
|
||
"mode_name": "横向支撑",
|
||
"tree": trees,
|
||
"total": len(kpis),
|
||
"causal_chain": [
|
||
{"from": "学习成长", "to": "内部流程", "logic": "培训与创新→流程效率提升"},
|
||
{"from": "内部流程", "to": "客户", "logic": "流程效率→客户满意度提升"},
|
||
{"from": "客户", "to": "财务", "logic": "客户规模→财务结果达成"},
|
||
],
|
||
}
|
||
|
||
|
||
def _build_bsc_chain(kpis, kpi_id=None):
|
||
"""BSC瀑布链：按category类别间的因果传导"""
|
||
from collections import defaultdict
|
||
|
||
# 每个维度的KPI按category分组
|
||
cat_kpis = defaultdict(list)
|
||
for k in kpis:
|
||
if k.category:
|
||
cat_kpis[k.category].append(k)
|
||
|
||
# BSC瀑布链的传导关系
|
||
chain = [
|
||
{"cat": "talent_pipeline", "label": "人才梯队", "dim": "learning", "feeds": ["supply_chain", "delivery_quality"]},
|
||
{"cat": "employee_engagement", "label": "员工敬业", "dim": "learning", "feeds": ["supply_chain"]},
|
||
{"cat": "innovation", "label": "创新改善", "dim": "learning", "feeds": ["delivery_quality"]},
|
||
{"cat": "supply_chain", "label": "供应链效率", "dim": "process", "feeds": ["delivery_quality"]},
|
||
{"cat": "delivery_quality", "label": "交付质量", "dim": "process", "feeds": ["customer_scale", "customer_satisfaction"]},
|
||
{"cat": "customer_scale", "label": "客户规模", "dim": "customer", "feeds": ["revenue_growth"]},
|
||
{"cat": "customer_concentration", "label": "客户集中度", "dim": "customer", "feeds": ["profitability"]},
|
||
{"cat": "customer_satisfaction", "label": "客户满意", "dim": "customer", "feeds": ["revenue_growth", "profitability"]},
|
||
{"cat": "revenue_growth", "label": "收入增长", "dim": "finance", "feeds": ["profitability"]},
|
||
{"cat": "profitability", "label": "盈利水平", "dim": "finance", "feeds": None},
|
||
{"cat": "cost_control", "label": "成本费用", "dim": "finance", "feeds": ["profitability"]},
|
||
{"cat": "asset_efficiency", "label": "资产效率", "dim": "finance", "feeds": ["profitability"]},
|
||
{"cat": "cash_risk", "label": "现金流风控", "dim": "finance", "feeds": ["profitability"]},
|
||
]
|
||
|
||
def make_node(kpi):
|
||
return {
|
||
"id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name,
|
||
"dimension": kpi.dimension, "category": kpi.category,
|
||
"target_value": kpi.target_value, "unit": kpi.unit,
|
||
}
|
||
|
||
# 构建瀑布链
|
||
trees = []
|
||
for link in chain:
|
||
cat = link["cat"]
|
||
if cat not in cat_kpis:
|
||
continue
|
||
cat_node = {
|
||
"id": None,
|
||
"category": cat,
|
||
"category_label": link["label"],
|
||
"dimension": link["dim"],
|
||
"is_category_group": True,
|
||
"feeds": link["feeds"],
|
||
"children": [make_node(k) for k in sorted(cat_kpis[cat], key=lambda x: x.kpi_code)],
|
||
"child_count": len(cat_kpis[cat]),
|
||
}
|
||
trees.append(cat_node)
|
||
|
||
if kpi_id:
|
||
return _filter_tree(trees, kpi_id)
|
||
|
||
return {
|
||
"mode": "bsc_chain",
|
||
"mode_name": "BSC瀑布链",
|
||
"tree": trees,
|
||
"total": len(kpis),
|
||
"chain": chain,
|
||
}
|
||
|
||
|
||
def _filter_tree(nodes, target_id):
|
||
"""在树中查找包含指定KPI的子树"""
|
||
for node in nodes:
|
||
if node.get("id") == target_id:
|
||
return node
|
||
if node.get("children"):
|
||
for child in node["children"]:
|
||
if child.get("id") == target_id:
|
||
return child
|
||
# 递归查找
|
||
found = _filter_tree(node["children"], target_id)
|
||
if found:
|
||
return found
|
||
return None
|