feat(data-quality): 财务七规则 governance-check 端点+前端Tab（最终方案）

backend/app/api/data_quality.py

@@ -516,3 +516,200 @@ def check_governance(
         "passed": passed,
         "rules_meta": RULES_META,
     }
+
+
+# ============================================================
+# 财务七规则检查（governance-check）— 2026-08-30 P1 最终方案
+# 与 /check-governance 的区别：
+#   * 评分规则不同：error 扣 min(15, count*3)，warning 扣 min(10, count*1)
+#   * 规则2/3 用 Python 解析 suggestion JSON（不依赖 MySQL JSON 函数）
+#   * 规则4 区分 error(test/sync/虚拟) 与 manual(待人工确认 warning)
+#   * 规则7 按 status='active' 口径（与 budget.py 一致，避免多版本叠加失真）
+# 只读幂等：不写库、不创建 KpiDataQualityLog
+# ============================================================
+
+
+def _extract_plan_id(suggestion: str):
+    """从 kpi_alerts.suggestion (Text 存 JSON) 解析 plan_id；解析失败返回 None"""
+    if not suggestion:
+        return None
+    try:
+        data = json.loads(suggestion)
+        return data.get("plan_id")
+    except Exception:
+        return None
+
+
+def _run_governance_checks(db: Session, entity_id: int = 0):
+    """执行财务七规则，返回 issues 列表（含 deducted 扣分）。entity_id=0 表示全部实体。"""
+    eid = entity_id or 0
+    ent = " AND cp.entity_id = :eid" if eid else ""
+    issues = []
+
+    # ── 规则1 单位校验：amount > 10000（万元口径可疑）──
+    rows = db.execute(text(
+        "SELECT cp.id, cp.entity_id, cp.amount, cp.source, cp.description "
+        "FROM cash_plans cp WHERE cp.amount > 10000" + ent + " ORDER BY cp.amount DESC LIMIT 200"
+    ), {"eid": eid}).fetchall()
+    issues.append({
+        "rule": "unit_check", "level": "error",
+        "count": len(rows),
+        "detail": [f"plan#{r.id} 金额{r.amount}(疑似元)" for r in rows[:DETAIL_LIMIT]]
+        + (["…等%d条" % len(rows)] if len(rows) > DETAIL_LIMIT else []),
+    })
+
+    # ── 规则2/3 共用数据：pending cash_plan 预警（实体筛选经 kpi_definitions）──
+    alert_sql = (
+        "SELECT a.id, a.kpi_id, a.suggestion, d.entity_id AS kpi_entity_id "
+        "FROM kpi_alerts a JOIN kpi_definitions d ON d.id = a.kpi_id "
+        "WHERE a.alert_type = 'cash_plan' AND a.status = 'pending'"
+    )
+    if eid:
+        alert_sql += " AND d.entity_id = :eid"
+    alert_rows = db.execute(text(alert_sql), {"eid": eid}).fetchall()
+
+    # Python 侧解析 suggestion → plan_id（不依赖 MySQL JSON 函数）
+    parsed = []          # [(alert_id, kpi_id, plan_id)]
+    for r in alert_rows:
+        pid = _extract_plan_id(r.suggestion)
+        if pid is not None:
+            parsed.append((r.id, r.kpi_id, pid))
+
+    # 已存在的 cash_plans id 集合（规则3 判断孤儿用；实体筛选时仅看该实体下 plan）
+    plan_ids_sql = "SELECT id FROM cash_plans" + (" WHERE entity_id = :eid" if eid else "")
+    plan_id_set = {row[0] for row in db.execute(text(plan_ids_sql), {"eid": eid}).fetchall()}
+
+    # ── 规则2 重复预警：同 plan_id 多条 pending 预警 ──
+    group_map = {}
+    for alert_id, kpi_id, pid in parsed:
+        group_map.setdefault(pid, []).append(alert_id)
+    dup_groups = [(pid, ids) for pid, ids in group_map.items() if len(ids) > 1]
+    dup_groups.sort(key=lambda x: -len(x[1]))
+    issues.append({
+        "rule": "dup_alert", "level": "error",
+        "count": len(dup_groups),
+        "detail": [f"plan#{pid} 重复预警×{len(ids)}" for pid, ids in dup_groups[:DETAIL_LIMIT]]
+        + (["…等%d个plan" % len(dup_groups)] if len(dup_groups) > DETAIL_LIMIT else []),
+    })
+
+    # ── 规则3 孤儿预警：plan_id 指向不存在的 cash_plans ──
+    orphan_rows = [(aid, kid, pid) for aid, kid, pid in parsed if pid not in plan_id_set]
+    issues.append({
+        "rule": "orphan_check", "level": "error",
+        "count": len(orphan_rows),
+        "detail": [f"预警#{aid}(kpi#{kid}) → plan#{pid} 不存在" for aid, kid, pid in orphan_rows[:DETAIL_LIMIT]]
+        + (["…等%d条" % len(orphan_rows)] if len(orphan_rows) > DETAIL_LIMIT else []),
+    })
+
+    # ── 规则4 虚拟污染：source 含 test/sync/虚拟 → error；source='manual' → 待人工确认 warning ──
+    rows = db.execute(text(
+        "SELECT cp.id, cp.entity_id, cp.source, cp.description FROM cash_plans cp "
+        "WHERE (cp.source LIKE '%test%' OR cp.source LIKE '%sync%' OR cp.source LIKE '%虚拟%' OR cp.source LIKE '%demo%')"
+        + ent + " ORDER BY cp.id LIMIT 200"
+    ), {"eid": eid}).fetchall()
+    manual_rows = db.execute(text(
+        "SELECT cp.id, cp.entity_id, cp.source, cp.description FROM cash_plans cp "
+        "WHERE cp.source = 'manual'" + ent + " ORDER BY cp.id LIMIT 200"
+    ), {"eid": eid}).fetchall()
+    issues.append({
+        "rule": "virtual_pollution", "level": "error",
+        "count": len(rows),
+        "detail": [f"plan#{r.id} source={r.source} desc={r.description or ''}" for r in rows[:DETAIL_LIMIT]]
+        + (["…等%d条" % len(rows)] if len(rows) > DETAIL_LIMIT else []),
+        "manual_count": len(manual_rows),
+        "manual_detail": [f"plan#{r.id} source=manual(待人工确认)" for r in manual_rows[:DETAIL_LIMIT]]
+        + (["…等%d条" % len(manual_rows)] if len(manual_rows) > DETAIL_LIMIT else []),
+    })
+
+    # ── 规则5 实体归属：kpi_values.entity_id IS NOT NULL 且 != kpi_definitions.entity_id ──
+    ent_sql = (
+        "SELECT v.id, v.kpi_id, d.kpi_code, v.entity_id AS v_eid, d.entity_id AS d_eid "
+        "FROM kpi_values v JOIN kpi_definitions d ON v.kpi_id = d.id "
+        "WHERE v.entity_id IS NOT NULL AND v.entity_id != d.entity_id"
+    )
+    if eid:
+        ent_sql += " AND v.entity_id = :eid"
+    ent_sql += " ORDER BY v.id LIMIT 200"
+    ent_rows = db.execute(text(ent_sql), {"eid": eid}).fetchall()
+    issues.append({
+        "rule": "entity_check", "level": "error",
+        "count": len(ent_rows),
+        "detail": [f"值#{r.id} {r.kpi_code} 实体{r.v_eid}≠定义实体{r.d_eid}" for r in ent_rows[:DETAIL_LIMIT]]
+        + (["…等%d条" % len(ent_rows)] if len(ent_rows) > DETAIL_LIMIT else []),
+    })
+
+    # ── 规则6 KPI完整性：active KPI 无任何实际值 ──
+    comp_sql = (
+        "SELECT d.id, d.kpi_code, d.kpi_name FROM kpi_definitions d "
+        "WHERE d.status='active' AND NOT EXISTS (SELECT 1 FROM kpi_values v WHERE v.kpi_id = d.id)"
+    )
+    if eid:
+        comp_sql += " AND d.entity_id = :eid"
+    comp_sql += " ORDER BY d.id LIMIT 300"
+    comp_rows = db.execute(text(comp_sql), {"eid": eid}).fetchall()
+    issues.append({
+        "rule": "kpi_completeness", "level": "warning",
+        "count": len(comp_rows),
+        "detail": [f"{r.kpi_code} {r.kpi_name}(无值)" for r in comp_rows[:DETAIL_LIMIT]]
+        + (["…等%d个KPI" % len(comp_rows)] if len(comp_rows) > DETAIL_LIMIT else []),
+    })
+
+    # ── 规则7 勾稽验证：预算(active口径)年度合计 vs 年度目标 差异>20% ──
+    recon_sql = (
+        "SELECT d.kpi_code, d.kpi_name, d.target_yearly, "
+        "COALESCE(SUM(b.budget_value),0) AS monthly_sum, "
+        "ROUND((COALESCE(SUM(b.budget_value),0) - d.target_yearly) / d.target_yearly * 100, 1) AS diff_pct "
+        "FROM kpi_definitions d JOIN budget_plans b ON b.kpi_id = d.id "
+        "WHERE d.status='active' AND d.target_yearly > 0 AND b.status='active'"
+    )
+    if eid:
+        recon_sql += " AND d.entity_id = :eid"
+    recon_sql += " GROUP BY d.id HAVING ABS(diff_pct) > 20 ORDER BY ABS(diff_pct) DESC LIMIT 200"
+    recon_rows = db.execute(text(recon_sql), {"eid": eid}).fetchall()
+    issues.append({
+        "rule": "reconciliation", "level": "warning",
+        "count": len(recon_rows),
+        "detail": [f"{r.kpi_code} 预算合计{round(r.monthly_sum, 1)} vs 年度目标{r.target_yearly} 差异{r.diff_pct}%" for r in recon_rows[:DETAIL_LIMIT]]
+        + (["…等%d个KPI" % len(recon_rows)] if len(recon_rows) > DETAIL_LIMIT else []),
+    })
+
+    # ── 评分：error 扣 min(15, count*3)，warning 扣 min(10, count*1)；规则4 manual 按 warning 附加扣 ──
+    total_deduct = 0
+    for item in issues:
+        ded = 0
+        if item["count"] > 0:
+            ded += min(15, item["count"] * 3) if item["level"] == "error" else min(10, item["count"] * 1)
+        # 规则4 附加：manual 待人工确认（warning 性质）
+        manual_cnt = item.get("manual_count") or 0
+        if item["rule"] == "virtual_pollution" and manual_cnt > 0:
+            ded += min(10, manual_cnt * 1)
+        item["deducted"] = ded
+        total_deduct += ded
+    score = max(0, 100 - total_deduct)
+
+    passed = [item["rule"] for item in issues if item["count"] == 0]
+
+    return {
+        "checked_at": datetime.now().isoformat(timespec="seconds"),
+        "entity_id": eid,
+        "score": score,
+        "total_deduct": total_deduct,
+        "total_rules": len(issues),
+        "issues": issues,
+        "passed": passed,
+        "rules_meta": RULES_META,
+    }
+
+
+@router.get("/governance-check")
+def governance_check(
+    entity_id: Optional[int] = Query(0, description="实体ID: 0=全部, 1=酣客, 2=博海"),
+    db: Session = Depends(get_db),
+):
+    """财务七规则检查（DAMA数据治理）— 只读幂等，不写库。
+
+    7条规则: unit_check/dup_alert/orphan_check/virtual_pollution/entity_check/kpi_completeness/reconciliation
+    评分: 满分100，error 扣 min(15, count*3)，warning 扣 min(10, count*1)，score=max(0, 100-总扣分)
+    实体筛选: entity_id 参数（0=全部）。
+    """
+    return _run_governance_checks(db, entity_id or 0)