174 lines
6.6 KiB
Python
174 lines
6.6 KiB
Python
"""KPI模板库 API — 管理会计OS
|
|
支持系统预置模板 + 用户自定义模板
|
|
从模板实例化创建KPI时，复制模板快照到kpi_definitions"""
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from typing import Optional
|
|
from datetime import datetime
|
|
|
|
from app.database import get_db
|
|
from app.auth_middleware import require_role
|
|
from app.models import KPITemplate, KPIDefinition, OperationLog
|
|
from app.api.kpi_governance import validate_kpi_payload, kpi_issues_message
|
|
|
|
router = APIRouter(prefix="/api/cma/templates", tags=["KPI模板库"],
|
|
dependencies=[Depends(require_role("ceo", "finance", "business", "it"))],
|
|
)
|
|
|
|
WRITE_ROLES = Depends(require_role("ceo", "finance", "it"))
|
|
|
|
|
|
def template_to_dict(t):
|
|
return {c.name: getattr(t, c.name) for c in t.__table__.columns}
|
|
|
|
|
|
@router.get("")
|
|
def list_templates(
|
|
dimension: Optional[str] = None,
|
|
category: Optional[str] = None,
|
|
keyword: Optional[str] = None,
|
|
is_system: Optional[int] = None,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""获取模板列表，支持按维度/类别/关键字筛选"""
|
|
query = db.query(KPITemplate)
|
|
if dimension:
|
|
query = query.filter(KPITemplate.dimension == dimension)
|
|
if category:
|
|
query = query.filter(KPITemplate.category == category)
|
|
if keyword:
|
|
query = query.filter(KPITemplate.kpi_name.contains(keyword))
|
|
if is_system is not None:
|
|
query = query.filter(KPITemplate.is_system == is_system)
|
|
templates = query.order_by(KPITemplate.is_system.desc(), KPITemplate.kpi_code).all()
|
|
return {"total": len(templates), "data": [template_to_dict(t) for t in templates]}
|
|
|
|
|
|
@router.get("/{template_id}")
|
|
def get_template(template_id: int, db: Session = Depends(get_db)):
|
|
t = db.query(KPITemplate).filter(KPITemplate.id == template_id).first()
|
|
if not t:
|
|
raise HTTPException(404, "模板不存在")
|
|
return template_to_dict(t)
|
|
|
|
|
|
@router.post("")
|
|
def create_template(data: dict, db: Session = Depends(get_db), user=WRITE_ROLES):
|
|
"""用户创建自定义模板"""
|
|
existing = db.query(KPITemplate).filter(KPITemplate.kpi_code == data.get("kpi_code", "")).first()
|
|
if existing:
|
|
raise HTTPException(400, f"模板编码 {data['kpi_code']} 已存在")
|
|
t = KPITemplate(
|
|
kpi_code=data.get("kpi_code"),
|
|
kpi_name=data.get("kpi_name"),
|
|
dimension=data.get("dimension"),
|
|
category=data.get("category"),
|
|
formula=data.get("formula"),
|
|
formula_desc=data.get("formula_desc"),
|
|
unit=data.get("unit", "%"),
|
|
target_value=data.get("target_value"),
|
|
description=data.get("description"),
|
|
is_system=0, # 用户创建的永远不是系统模板
|
|
usage_count=0,
|
|
)
|
|
db.add(t)
|
|
db.commit()
|
|
db.refresh(t)
|
|
_log(db, 1, "create", "template", t.id, {"kpi_code": t.kpi_code, "kpi_name": t.kpi_name})
|
|
return template_to_dict(t)
|
|
|
|
|
|
@router.put("/{template_id}")
|
|
def update_template(template_id: int, data: dict, db: Session = Depends(get_db), user=WRITE_ROLES):
|
|
"""修改自定义模板（系统预置不可修改）"""
|
|
t = db.query(KPITemplate).filter(KPITemplate.id == template_id).first()
|
|
if not t:
|
|
raise HTTPException(404, "模板不存在")
|
|
if t.is_system:
|
|
raise HTTPException(403, "系统预置模板不可修改")
|
|
for k, v in data.items():
|
|
if hasattr(t, k) and v is not None:
|
|
setattr(t, k, v)
|
|
db.commit()
|
|
return template_to_dict(t)
|
|
|
|
|
|
@router.delete("/{template_id}")
|
|
def delete_template(template_id: int, db: Session = Depends(get_db), user=WRITE_ROLES):
|
|
"""删除自定义模板（系统预置不可删除）"""
|
|
t = db.query(KPITemplate).filter(KPITemplate.id == template_id).first()
|
|
if not t:
|
|
raise HTTPException(404, "模板不存在")
|
|
if t.is_system:
|
|
raise HTTPException(403, "系统预置模板不可删除")
|
|
db.delete(t)
|
|
db.commit()
|
|
return {"message": "模板已删除"}
|
|
|
|
|
|
@router.post("/{template_id}/instantiate")
|
|
def instantiate_template(template_id: int, data: dict, db: Session = Depends(get_db), user=WRITE_ROLES):
|
|
"""从模板实例化创建KPI，复制模板快照到kpi_definitions"""
|
|
t = db.query(KPITemplate).filter(KPITemplate.id == template_id).first()
|
|
if not t:
|
|
raise HTTPException(404, "模板不存在")
|
|
|
|
kpi_code = data.get("kpi_code", t.kpi_code)
|
|
kpi_name = data.get("kpi_name", t.kpi_name)
|
|
|
|
# 检查编码唯一性
|
|
existing = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == kpi_code).first()
|
|
if existing:
|
|
raise HTTPException(400, f"KPI编码 {kpi_code} 已存在，请修改")
|
|
|
|
# 数据治理：入库必检(规则1) + 编码规范(规则3)
|
|
merged = {
|
|
"kpi_code": kpi_code,
|
|
"kpi_name": kpi_name,
|
|
"dimension": data.get("dimension", t.dimension),
|
|
"category": data.get("category", t.category),
|
|
"formula": data.get("formula", t.formula),
|
|
"unit": data.get("unit", t.unit or "%"),
|
|
"target_value": data.get("target_value", t.target_value),
|
|
}
|
|
gov_issues = [i for i in validate_kpi_payload(merged, db=db) if i["rule"] in (1, 3)]
|
|
if gov_issues:
|
|
raise HTTPException(422, detail={"message": "数据校验不通过", "errors": kpi_issues_message(gov_issues)})
|
|
|
|
kpi = KPIDefinition(
|
|
template_id=t.id,
|
|
is_system=0, # 从模板实例化的KPI不是系统预置
|
|
kpi_code=kpi_code,
|
|
kpi_name=kpi_name,
|
|
dimension=data.get("dimension", t.dimension),
|
|
category=data.get("category", t.category),
|
|
formula=data.get("formula", t.formula),
|
|
formula_desc=data.get("formula_desc", t.formula_desc),
|
|
unit=data.get("unit", t.unit or "%"),
|
|
target_value=data.get("target_value", t.target_value),
|
|
objective=data.get("objective"),
|
|
data_source_type=data.get("data_source_type", "manual"),
|
|
frequency=data.get("frequency", "monthly"),
|
|
responsible_dept=data.get("responsible_dept"),
|
|
responsible_user=data.get("responsible_user"),
|
|
status="active",
|
|
)
|
|
db.add(kpi)
|
|
db.commit()
|
|
db.refresh(kpi)
|
|
|
|
# 更新模板使用计数
|
|
t.usage_count = (t.usage_count or 0) + 1
|
|
db.commit()
|
|
|
|
_log(db, 1, "create", "kpi", kpi.id, {"from_template": template_id, "kpi_code": kpi.kpi_code})
|
|
return {c.name: getattr(kpi, c.name) for c in kpi.__table__.columns}
|
|
|
|
|
|
def _log(db, user_id, action, target_type, target_id, detail):
|
|
import json
|
|
log = OperationLog(user_id=user_id, action=action, target_type=target_type,
|
|
target_id=target_id, detail=json.dumps(detail, ensure_ascii=False) if detail else None)
|
|
db.add(log)
|
|
db.commit()
|