diff --git a/backend/app/api/data_quality.py b/backend/app/api/data_quality.py index 642f6cbe..7f2acc7e 100644 --- a/backend/app/api/data_quality.py +++ b/backend/app/api/data_quality.py @@ -3,7 +3,7 @@ """ from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session -from sqlalchemy import func, and_ +from sqlalchemy import func, and_, text from typing import Optional from datetime import datetime, timedelta import json @@ -322,3 +322,197 @@ def quality_stats(db: Session = Depends(get_db)): "threshold_days": 180, }, } + + +# ============================================================ +# DAMA 数据治理规则检查（财务七规则）— 2026-08-30 P1 +# ============================================================ + +RULES_META = { + "unit_check": {"name": "单位校验", "level": "error", "desc": "cash_plans.amount > 10000（万元口径可疑，疑似单位错乱）"}, + "dup_alert": {"name": "重复预警", "level": "error", "desc": "同一plan_id存在多条pending应收预警（去重键错误）"}, + "orphan_check": {"name": "孤儿预警", "level": "error", "desc": "预警suggestion.plan_id指向不存在的cash_plans记录"}, + "virtual_pollution": {"name": "虚拟污染", "level": "error", "desc": "cash_plans.source含test/虚拟等测试标识混入真实数据"}, + "entity_check": {"name": "实体归属", "level": "error", "desc": "kpi_values.entity_id与kpi_definitions.entity_id不一致"}, + "kpi_completeness": {"name": "KPI完整性", "level": "warning", "desc": "active状态KPI无任何实际值的数量"}, + "reconciliation": {"name": "勾稽验证", "level": "warning", "desc": "预算月度合计 vs 年度目标差异>20%"}, +} + +DETAIL_LIMIT = 10 # 每条规则detail最多列出的条数（避免响应过大） + + +def _run_rule_checks(db: Session, entity_id: int = 0): + """执行7条DAMA治理规则，返回 issues 列表。entity_id=0 表示全部实体。""" + entity_filter = " AND cp.entity_id = :eid" if entity_id else "" + + issues = [] + + # ── 规则1 单位校验 ── + rows = db.execute(text( + "SELECT cp.id, cp.entity_id, cp.amount, cp.source, cp.description " + "FROM cash_plans cp WHERE cp.amount > 10000" + entity_filter + " ORDER BY cp.amount DESC" + ), {"eid": entity_id}).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 重复预警（同plan_id多条pending应收预警）── + if entity_id: + dup_sql = text( + "SELECT JSON_EXTRACT(a.suggestion, '$.plan_id') AS pid, COUNT(*) c, MAX(p.entity_id) eid " + "FROM kpi_alerts a JOIN cash_plans p ON p.id = JSON_EXTRACT(a.suggestion, '$.plan_id') " + "WHERE a.alert_type='cash_plan' AND a.status='pending' AND JSON_VALID(a.suggestion) " + "AND a.suggestion LIKE '%plan_id%' AND p.entity_id = :eid " + "GROUP BY pid HAVING c > 1 ORDER BY c DESC" + ) + else: + dup_sql = text( + "SELECT JSON_EXTRACT(a.suggestion, '$.plan_id') AS pid, COUNT(*) c, MAX(p.entity_id) eid " + "FROM kpi_alerts a JOIN cash_plans p ON p.id = JSON_EXTRACT(a.suggestion, '$.plan_id') " + "WHERE a.alert_type='cash_plan' AND a.status='pending' AND JSON_VALID(a.suggestion) " + "AND a.suggestion LIKE '%plan_id%' " + "GROUP BY pid HAVING c > 1 ORDER BY c DESC" + ) + dup_rows = db.execute(dup_sql, {"eid": entity_id}).fetchall() + issues.append({ + "rule": "dup_alert", "level": "error", + "count": len(dup_rows), + "detail": [f"plan#{r.pid} 重复预警×{r.c}" for r in dup_rows[:DETAIL_LIMIT]] + + (["…等%d个plan" % len(dup_rows)] if len(dup_rows) > DETAIL_LIMIT else []), + }) + + # ── 规则3 孤儿预警（plan_id指向不存在的cash_plans）── + orphan_sql = text( + "SELECT a.id, a.kpi_id, JSON_EXTRACT(a.suggestion, '$.plan_id') AS pid " + "FROM kpi_alerts a " + "WHERE a.alert_type='cash_plan' AND a.status='pending' AND JSON_VALID(a.suggestion) " + "AND a.suggestion LIKE '%plan_id%' " + "AND NOT EXISTS (SELECT 1 FROM cash_plans p WHERE p.id = JSON_EXTRACT(a.suggestion, '$.plan_id')) " + "ORDER BY a.id LIMIT 200" + ) + orphan_rows = db.execute(orphan_sql).fetchall() + issues.append({ + "rule": "orphan_check", "level": "error", + "count": len(orphan_rows), + "detail": [f"预警#{r.id}(kpi#{r.kpi_id}) → plan#{r.pid} 不存在" for r in orphan_rows[:DETAIL_LIMIT]] + + (["…等%d条" % len(orphan_rows)] if len(orphan_rows) > DETAIL_LIMIT else []), + }) + + # ── 规则4 虚拟污染（source含test/虚拟标识）── + 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 '%虚拟%' OR cp.source LIKE '%demo%'" + + entity_filter + " ORDER BY cp.id LIMIT 200" + ), {"eid": entity_id}).fetchall() + issues.append({ + "rule": "virtual_pollution", "level": "error", + "count": len(rows), + "detail": [f"plan#{r.id} source={r.source}" for r in rows[:DETAIL_LIMIT]] + + (["…等%d条" % len(rows)] if len(rows) > DETAIL_LIMIT else []), + }) + + # ── 规则5 实体归属（kpi_values.entity_id != kpi_definitions.entity_id）── + if entity_id: + ent_sql = text( + "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 != d.entity_id AND v.entity_id = :eid ORDER BY v.id LIMIT 200" + ) + else: + ent_sql = text( + "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 != d.entity_id ORDER BY v.id LIMIT 200" + ) + ent_rows = db.execute(ent_sql, {"eid": entity_id}).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无任何值）── + if entity_id: + comp_sql = text( + "SELECT d.id, d.kpi_code, d.kpi_name FROM kpi_definitions d " + "WHERE d.status='active' AND d.entity_id = :eid " + "AND NOT EXISTS (SELECT 1 FROM kpi_values v WHERE v.kpi_id = d.id) ORDER BY d.id LIMIT 300" + ) + else: + comp_sql = text( + "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) ORDER BY d.id LIMIT 300" + ) + comp_rows = db.execute(comp_sql, {"eid": entity_id}).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 勾稽验证（预算月度合计 vs 年度目标差异>20%）── + if entity_id: + recon_sql = text( + "SELECT d.kpi_code, d.kpi_name, d.target_yearly, " + "SUM(b.budget_value) AS monthly_sum, " + "ROUND((SUM(b.budget_value) - 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 d.entity_id = :eid " + "GROUP BY d.id HAVING ABS(diff_pct) > 20 ORDER BY ABS(diff_pct) DESC LIMIT 200" + ) + else: + recon_sql = text( + "SELECT d.kpi_code, d.kpi_name, d.target_yearly, " + "SUM(b.budget_value) AS monthly_sum, " + "ROUND((SUM(b.budget_value) - 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 " + "GROUP BY d.id HAVING ABS(diff_pct) > 20 ORDER BY ABS(diff_pct) DESC LIMIT 200" + ) + recon_rows = db.execute(recon_sql, {"eid": entity_id}).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 []), + }) + + return issues + + +@router.get("/check-governance") +def check_governance( + entity_id: Optional[int] = Query(0, description="实体ID: 0=全部, 1=酣客, 2=博海"), + db: Session = Depends(get_db), +): + """DAMA数据治理规则检查（财务七规则）— 返回质量评分+异常清单。 + + 评分规则: 满分100，error级规则每条扣10分，warning级规则每条扣5分， + 每条规则最多扣一次分（按规则是否命中，不按count累扣），最低0分。 + """ + issues = _run_rule_checks(db, entity_id or 0) + + # 计算评分 + score = 100 + for item in issues: + if item["count"] > 0: + score -= 10 if item["level"] == "error" else 5 + score = max(0, score) + + passed = [item["rule"] for item in issues if item["count"] == 0] + + return { + "checked_at": datetime.now().isoformat(timespec="seconds"), + "entity_id": entity_id or 0, + "score": score, + "total_rules": len(issues), + "issues": issues, + "passed": passed, + "rules_meta": RULES_META, + } diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index 7fc1e49c..1586ce52 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -344,6 +344,8 @@ export const dataQualityApi = { logs: (params?: any) => api.get('/data-quality/logs', { params }), updateLog: (id: number, data: any) => api.put(`/data-quality/logs/${id}`, data), deleteLog: (id: number) => api.delete(`/data-quality/logs/${id}`), + // DAMA数据治理规则检查（财务七规则） + checkGovernance: (params?: any) => api.get('/data-quality/check-governance', { params }), } export const biReportApi = { diff --git a/frontend/src/views/DataQuality.vue b/frontend/src/views/DataQuality.vue index c56fae92..6aac344f 100644 --- a/frontend/src/views/DataQuality.vue +++ b/frontend/src/views/DataQuality.vue @@ -121,6 +121,80 @@ + + + + + + +
+ + + +
+ 检查时间: {{ govCheckedAt || '-' }}
+ 通过 {{ govPassedCount }} / {{ govIssues.length }} 条规则 +
+
+ + + + + + + + + + + + + + + + + + + + + + + + + +