197 lines
7.5 KiB
Python
197 lines
7.5 KiB
Python
"""KPI数据治理4条规则：入库必检 + 元数据必填 + 编码规范 + 战略分级
|
|
|
|
规则1: 入库必检 — dimension/target_value/unit 必填（创建/更新强制拦截）
|
|
规则2: 元数据必填 — formula/data_source/data_owner 不能为空或占位符（待补充/待指定/-）
|
|
规则3: 编码规范 — kpi_code 必须以 F_/C_/P_/L_ 前缀开头且与 dimension 一致；禁止跨层同名
|
|
规则4: 战略/运营分级 — kpi_level 必须是 strategic/operational
|
|
|
|
API:
|
|
- POST /api/cma/kpi/validate 校验单个KPI数据 → {valid, errors}
|
|
- GET /api/cma/kpi/governance/audit 全量校验 → 按4条规则分组的不合规清单
|
|
"""
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.auth_middleware import require_auth, require_role
|
|
from app.models import KPIDefinition
|
|
|
|
router = APIRouter(
|
|
prefix="/api/cma/kpi",
|
|
tags=["KPI数据治理"],
|
|
dependencies=[Depends(require_role("ceo", "finance", "business", "it"))],
|
|
)
|
|
|
|
# 维度 → 编码前缀
|
|
DIM_PREFIX = {"finance": "F", "customer": "C", "process": "P", "learning": "L"}
|
|
VALID_LEVELS = ("strategic", "operational")
|
|
# 视为"未完善"的占位符值
|
|
PLACEHOLDERS = ("待补充", "待指定", "待完善", "待定", "暂无", "TBD", "tbd", "-", "--", "N/A", "n/a")
|
|
|
|
# 规则2必填元数据字段
|
|
META_FIELDS = [
|
|
("formula", "计算公式"),
|
|
("data_source", "数据来源"),
|
|
("data_owner", "数据责任人"),
|
|
]
|
|
|
|
|
|
def _clean_str(val) -> str:
|
|
if val is None:
|
|
return ""
|
|
if isinstance(val, str):
|
|
return val.strip()
|
|
return str(val).strip()
|
|
|
|
|
|
def _is_placeholder(val) -> bool:
|
|
"""空值或占位符（待补充/待指定/- 等）视为未完善"""
|
|
s = _clean_str(val)
|
|
if not s:
|
|
return True
|
|
return s in PLACEHOLDERS
|
|
|
|
|
|
def validate_kpi_payload(
|
|
data: dict,
|
|
db: Session = None,
|
|
current_kpi_id: Optional[int] = None,
|
|
is_update: bool = False,
|
|
) -> List[dict]:
|
|
"""校验单个KPI数据（4条规则）。
|
|
|
|
- data: 提交的KPI字段字典（创建或更新的载荷）
|
|
- db: SQLAlchemy Session（用于跨层同名/编码唯一性检查，可为None）
|
|
- current_kpi_id: 更新时传KPI自身id（避免自检误报）
|
|
- is_update: 更新模式 — 仅校验载荷中显式出现的字段
|
|
|
|
返回 [{rule: int, field: str, message: str}, ...]，空列表=合规。
|
|
"""
|
|
issues: List[dict] = []
|
|
|
|
def add(rule: int, field: str, message: str):
|
|
issues.append({"rule": rule, "field": field, "message": message})
|
|
|
|
code = _clean_str(data.get("kpi_code"))
|
|
dimension = _clean_str(data.get("dimension"))
|
|
has_code = bool(code)
|
|
|
|
# ── 规则1: 入库必检 dimension/target_value/unit ──
|
|
if not is_update or "dimension" in data:
|
|
if not dimension:
|
|
add(1, "dimension", "缺少dimension(所属维度: finance/customer/process/learning)")
|
|
if not is_update or "target_value" in data:
|
|
tv = data.get("target_value")
|
|
if tv is None or (isinstance(tv, str) and _clean_str(tv) == ""):
|
|
add(1, "target_value", "缺少target_value(目标值)")
|
|
if not is_update or "unit" in data:
|
|
if _is_placeholder(data.get("unit")):
|
|
add(1, "unit", "缺少unit(单位)")
|
|
|
|
# ── 规则2: 元数据必填（不能为空或占位符）──
|
|
for field, label in META_FIELDS:
|
|
if not is_update or field in data:
|
|
if _is_placeholder(data.get(field)):
|
|
add(2, field, f"元数据未完善: {label}({field})不能为空或占位符(待补充/待指定/-)")
|
|
|
|
# ── 规则3: 编码规范 ──
|
|
if not is_update or "kpi_code" in data:
|
|
if not has_code:
|
|
add(3, "kpi_code", "缺少kpi_code(KPI编码)")
|
|
else:
|
|
prefix = code.split("_")[0] if "_" in code else code
|
|
if prefix not in ("F", "C", "P", "L"):
|
|
add(3, "kpi_code", f"编码前缀不符: {code} 应以F_/C_/P_/L_开头")
|
|
elif dimension and DIM_PREFIX.get(dimension) and prefix != DIM_PREFIX[dimension]:
|
|
expected = DIM_PREFIX[dimension]
|
|
add(3, "kpi_code",
|
|
f"编码前缀与维度不符: {code} 前缀{prefix}_ 与维度{dimension}(应为{expected}_)不一致")
|
|
|
|
# 禁止跨层同名: 同一kpi_code不能用于不同dimension
|
|
if has_code and db is not None:
|
|
dup = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == code).first()
|
|
if dup and dup.id != current_kpi_id:
|
|
if dimension and dup.dimension and dup.dimension != dimension:
|
|
add(3, "kpi_code",
|
|
f"跨层同名: {code} 已用于维度{dup.dimension}，不能用于维度{dimension}")
|
|
elif not dimension:
|
|
add(3, "kpi_code", f"编码已存在: {code} 已注册(维度{dup.dimension})，不能重复使用")
|
|
|
|
# ── 规则4: 战略/运营分级 ──
|
|
if not is_update or "kpi_level" in data:
|
|
lv = data.get("kpi_level")
|
|
if lv is not None and lv not in VALID_LEVELS:
|
|
add(4, "kpi_level", f"kpi_level必须是strategic或operational，当前值: {lv}")
|
|
|
|
return issues
|
|
|
|
|
|
def kpi_issues_message(issues: List[dict]) -> List[str]:
|
|
"""issue dict列表 → 纯文本错误列表"""
|
|
return [i["message"] for i in issues]
|
|
|
|
|
|
@router.post("/validate")
|
|
def validate_kpi(
|
|
kpi_data: dict,
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(require_auth),
|
|
):
|
|
"""校验单个KPI数据（不入库）。
|
|
|
|
请求体: KPI字段字典，可选 kpi_id 标识正在编辑的KPI（避免跨层同名误报）。
|
|
返回: {"valid": bool, "errors": [str], "details": [{rule, field, message}]}
|
|
"""
|
|
kpi_id = kpi_data.get("kpi_id") if isinstance(kpi_data.get("kpi_id"), int) else None
|
|
issues = validate_kpi_payload(kpi_data, db=db, current_kpi_id=kpi_id)
|
|
return {
|
|
"valid": len(issues) == 0,
|
|
"errors": kpi_issues_message(issues),
|
|
"details": issues,
|
|
}
|
|
|
|
|
|
@router.get("/governance/audit")
|
|
def governance_audit(
|
|
db: Session = Depends(get_db),
|
|
current_user=Depends(require_auth),
|
|
):
|
|
"""全量校验所有活跃KPI，输出按4条规则分组的不合规清单。"""
|
|
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
|
|
|
|
non_compliant = []
|
|
by_rule: dict = {1: [], 2: [], 3: [], 4: []}
|
|
for k in kpis:
|
|
payload = {c.name: getattr(k, c.name) for c in k.__table__.columns}
|
|
issues = validate_kpi_payload(payload, db=db, current_kpi_id=k.id)
|
|
if issues:
|
|
entry = {
|
|
"kpi_id": k.id,
|
|
"kpi_code": k.kpi_code,
|
|
"kpi_name": k.kpi_name,
|
|
"dimension": k.dimension,
|
|
"kpi_level": k.kpi_level,
|
|
"issues": issues,
|
|
}
|
|
non_compliant.append(entry)
|
|
for i in issues:
|
|
by_rule.setdefault(i["rule"], []).append({
|
|
"kpi_id": k.id,
|
|
"kpi_code": k.kpi_code,
|
|
"kpi_name": k.kpi_name,
|
|
"field": i["field"],
|
|
"message": i["message"],
|
|
})
|
|
|
|
rule_counts = {str(r): len(items) for r, items in by_rule.items()}
|
|
return {
|
|
"total": len(kpis),
|
|
"compliant": len(kpis) - len(non_compliant),
|
|
"non_compliant_count": len(non_compliant),
|
|
"rule_counts": rule_counts,
|
|
"by_rule": {str(r): items for r, items in by_rule.items()},
|
|
"non_compliant": non_compliant,
|
|
}
|