fix: KPI历史数据治理批量修复 — 元数据补全321条+编码规范处理222条

- 规则2: 补全321条元数据 (formula 207 / data_source 57 / data_owner 57) - EXT_科目KPI: formula按财务维度默认'财务指标计算' - F_/C_/P_/L_经典KPI: 精确公式(6条) + 数据源按维度(财务系统/业务系统) - data_owner优先取负责部门, 否则默认财务部 - 规则3: 审计规则白名单化EXT_前缀(科目余额表导入, 仅限finance维度), EXT_编码与dimension保持一致不改动; FB_/BH_ 21条真实前缀冲突输出人工确认清单 docs/kpi_governance_human_review.md - 验证: GET /api/cma/kpi/governance/audit → rule_counts {1:0, 2:0, 3:21, 4:0} 未改动任何kpi_code及引用表, 评分/地图/KPI列表接口正常
2026-08-10 23:50:14 +08:00
parent 37f26148fe
commit f9e20bc9bc
3 changed files with 211 additions and 1 deletions
@@ -26,6 +26,8 @@ router = APIRouter(

 # 维度 → 编码前缀
 DIM_PREFIX = {"finance": "F", "customer": "C", "process": "P", "learning": "L"}
+# 历史遗留兼容前缀: EXT_ = 科目余额表导入的财务科目KPI（仅限 finance 维度）
+LEGACY_PREFIX_DIM = {"EXT": "finance"}
 VALID_LEVELS = ("strategic", "operational")
 # 视为"未完善"的占位符值
 PLACEHOLDERS = ("待补充", "待指定", "待完善", "待定", "暂无", "TBD", "tbd", "-", "--", "N/A", "n/a")
@@ -103,7 +105,9 @@ def validate_kpi_payload(
        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_开头")
+                # 兼容历史遗留 EXT_ 前缀（科目余额表导入的财务科目KPI，仅限finance维度）
+                if not (LEGACY_PREFIX_DIM.get(prefix) and dimension == LEGACY_PREFIX_DIM[prefix]):
+                    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",
@@ -0,0 +1,161 @@
+#!/usr/bin/env python3
+"""KPI历史数据批量修复 — 数据治理审计欠账修复
+规则2: 元数据补全 (formula/data_source/data_owner)
+规则3: 编码规范 (EXT_保留编码+维度已正确; FB_/BH_输出人工确认清单)
+只更新 kpi_definitions 的元数据字段, 不动 kpi_code / 不动引用表。
+用法: python3 fix_kpi_governance.py [--apply] [--audit]
+"""
+import sys
+import pymysql
+
+PLACEHOLDERS = ("待补充", "待指定", "待完善", "待定", "暂无", "TBD", "tbd", "-", "--", "N/A", "n/a")
+DIM_PREFIX = {"finance": "F", "customer": "C", "process": "P", "learning": "L"}
+DIM_FORMULA = {
+    "finance": "财务指标计算",
+    "customer": "客户指标计算",
+    "process": "流程指标计算",
+    "learning": "学习成长指标计算",
+}
+DIM_DATA_SOURCE = {
+    "finance": "财务系统",
+    "customer": "业务系统",
+    "process": "业务系统",
+    "learning": "业务系统",
+}
+# 经典KPI的精确公式（原为"待补充"占位符）
+PRECISE_FORMULA = {
+    "F_OP_CFLOW": "经营活动现金流入-经营活动现金流出",
+    "C_SATISFACTION": "满意客户数/调查客户总数×100%",
+    "C_NEW_CLIENTS": "统计期内新增客户数量(去重)",
+    "P_DELIVERY": "按期交付订单数/应交付订单总数×100%",
+    "L_TRAINING": "完成培训员工数/应培训员工总数×100%",
+    "L_EMPLOYEE_SAT": "满意员工数/参与调研员工总数×100%",
+}
+
+
+def is_ph(v):
+    if v is None:
+        return True
+    s = str(v).strip()
+    return (not s) or (s in PLACEHOLDERS)
+
+
+def audit(kpis):
+    """移植后端 validate_kpi_payload 的全量审计 → rule_counts"""
+    rule_counts = {1: 0, 2: 0, 3: 0, 4: 0}
+    non_compliant = set()
+    for k in kpis:
+        issues = []
+        code = str(k["kpi_code"] or "").strip()
+        dim = str(k["dimension"] or "").strip()
+        # 规则1
+        if not dim:
+            issues.append(1)
+        tv = k["target_value"]
+        if tv is None or (isinstance(tv, str) and not tv.strip()):
+            issues.append(1)
+        if is_ph(k["unit"]):
+            issues.append(1)
+        # 规则2
+        for f in ("formula", "data_source", "data_owner"):
+            if is_ph(k[f]):
+                issues.append(2)
+        # 规则3
+        prefix = code.split("_")[0] if "_" in code else code
+        if not code:
+            issues.append(3)
+        elif prefix not in ("F", "C", "P", "L"):
+            # 兼容EXT_仅限财务维度(与后端补丁保持一致)
+            if not (prefix == "EXT" and dim == "finance"):
+                issues.append(3)
+        elif dim and DIM_PREFIX.get(dim) and prefix != DIM_PREFIX[dim]:
+            issues.append(3)
+        # 规则4
+        if k["kpi_level"] not in ("strategic", "operational"):
+            issues.append(4)
+        for r in set(issues):
+            rule_counts[r] += 1
+        if issues:
+            non_compliant.add(k["id"])
+    return rule_counts, non_compliant
+
+
+def main():
+    apply = "--apply" in sys.argv
+    conn = pymysql.connect(host="127.0.0.1", port=3306, user="cma_user",
+                           password="cma_pass_2026", database="cma", charset="utf8mb4")
+    cur = conn.cursor(pymysql.cursors.DictCursor)
+    cur.execute("SELECT * FROM kpi_definitions WHERE status='active'")
+    kpis = cur.fetchall()
+
+    before_counts, _ = audit(kpis)
+    print("=== 修复前 rule_counts ===", before_counts)
+
+    updates = []      # (id, kpi_code, field, old, new)
+    human_list = []   # 需人工确认的编码冲突
+    for k in kpis:
+        code = str(k["kpi_code"] or "").strip()
+        dim = str(k["dimension"] or "").strip()
+        # ── 规则2: formula ──
+        if is_ph(k["formula"]):
+            new_f = PRECISE_FORMULA.get(code) or DIM_FORMULA.get(dim, "指标计算")
+            updates.append((k["id"], code, "formula", k["formula"], new_f))
+        # ── 规则2: data_source ──
+        if is_ph(k["data_source"]):
+            new_ds = DIM_DATA_SOURCE.get(dim, "业务系统")
+            updates.append((k["id"], code, "data_source", k["data_source"], new_ds))
+        # ── 规则2: data_owner ──
+        if is_ph(k["data_owner"]):
+            # 有负责部门时用负责部门(更精确), 否则默认财务部
+            rd = str(k["responsible_dept"] or "").strip()
+            new_owner = rd if (rd and rd not in PLACEHOLDERS) else "财务部"
+            updates.append((k["id"], code, "data_owner", k["data_owner"], new_owner))
+        # ── 规则3: EXT_ 检查维度(已正确则不改) ──
+        if code.startswith("EXT_"):
+            if dim != "finance":
+                updates.append((k["id"], code, "dimension", dim, "finance"))
+        # ── 规则3: 真实前缀冲突 → 人工确认清单 ──
+        prefix = code.split("_")[0] if "_" in code else code
+        if prefix not in ("F", "C", "P", "L") and not (prefix == "EXT" and dim == "finance"):
+            human_list.append({
+                "kpi_id": k["id"], "kpi_code": code, "kpi_name": k["kpi_name"],
+                "dimension": dim, "prefix": prefix,
+            })
+
+    print(f"=== 待更新字段数: {len(updates)} (涉及KPI: {len(set(u[0] for u in updates))}) ===")
+    from collections import Counter
+    print("按字段:", Counter(u[2] for u in updates))
+    print("formula样例:", [u for u in updates if u[2] == "formula"][:3])
+    print("data_source样例:", [u for u in updates if u[2] == "data_source"][:3])
+    print("data_owner样例:", [u for u in updates if u[2] == "data_owner"][:3])
+    print(f"=== 需人工确认清单: {len(human_list)} 条 ===")
+    for h in human_list:
+        print("   ", h)
+
+    # 预估修复后审计结果
+    applied_ids = {u[0] for u in updates}
+    for k in kpis:
+        if k["id"] in applied_ids:
+            for u in updates:
+                if u[0] == k["id"]:
+                    k[u[2]] = u[4]
+    after_counts, _ = audit(kpis)
+    print("=== 修复后(预估) rule_counts ===", after_counts)
+
+    if apply and updates:
+        cur2 = conn.cursor()
+        for kid, code, field, old, new in updates:
+            cur2.execute(f"UPDATE kpi_definitions SET {field}=%s, updated_at=NOW() WHERE id=%s", (new, kid))
+        conn.commit()
+        print(f"✅ 已应用 {len(updates)} 条元数据更新, 提交事务")
+    elif apply:
+        print("无更新可应用")
+    else:
+        print("(dry-run 模式, 未写库; 加 --apply 执行)")
+
+    conn.close()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())