"""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"} # 历史遗留兼容前缀: EXT_ = 科目余额表导入的财务科目KPI（仅限 finance 维度） LEGACY_PREFIX_DIM = {"EXT": "finance", "FB": "process"} # 品牌级业务KPI前缀（跨维度合法，2026-08-21 治理审计适配: BH_ = 博海业务KPI系列） BRAND_PREFIXES = ("BH",) VALID_LEVELS = ("strategic", "operational") # 视为"未完善"的占位符值 PLACEHOLDERS = ("待补充", "待指定", "待完善", "待定", "暂无", "TBD", "tbd", "-", "--", "N/A", "n/a") # 规则2必填元数据字段 META_FIELDS = [ ("formula", "计算公式"), ("data_source", "数据来源"), ("data_owner", "数据责任人"), ("kpi_name", "KPI名称"), ] 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"): # 兼容历史遗留 EXT_ 前缀（科目余额表导入的财务科目KPI，仅限finance维度） # 兼容 FB_ 前缀（财务Bot KPI，仅限process维度）与 BH_ 品牌前缀（跨维度合法） if not (LEGACY_PREFIX_DIM.get(prefix) and dimension == LEGACY_PREFIX_DIM[prefix]) \ and prefix not in BRAND_PREFIXES: 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, }