Hermes CI Fix
c36ed35348
后端 backend/app/api/data_quality.py: - 新增 GET /api/cma/data-quality/check-governance?entity_id= 端点 - 7条规则: unit_check单位校验/dup_alert重复预警/orphan_check孤儿预警/virtual_pollution虚拟污染/entity_check实体归属/kpi_completeness KPI完整性/reconciliation勾稽验证 - 质量评分: error扣10/warning扣5, 每条规则最多扣一次, 最低0分 - 兼容实体筛选(0全部/1酣客/2博海), 向后兼容既有/check端点 前端 frontend/src/api/index.ts + views/DataQuality.vue: - dataQualityApi.checkGovernance() 新增 - 页面新增治理规则区: 质量评分仪表盘 + 7规则红黄绿状态表 + 异常明细展开 + 重新检查按钮 + 实体筛选
519 lines
22 KiB
Python
519 lines
22 KiB
Python
"""自动数据质量监控 — 任务3
|
|
定期检查KPI值异常、连续持平、数据缺失等
|
|
"""
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func, and_, text
|
|
from typing import Optional
|
|
from datetime import datetime, timedelta
|
|
import json
|
|
import logging
|
|
|
|
from app.database import get_db
|
|
from app.auth_middleware import require_auth, require_role
|
|
from app.models import KPIDefinition, KPIValue, KpiDataQualityLog, OperationLog
|
|
from app.api.kpis import kpi_to_dict
|
|
|
|
logger = logging.getLogger("data-quality")
|
|
|
|
router = APIRouter(prefix="/api/cma/data-quality", tags=["数据质量"],
|
|
dependencies=[Depends(require_role("ceo", "finance", "business", "it"))],
|
|
)
|
|
WRITE_ROLES = Depends(require_role("ceo", "finance", "it"))
|
|
|
|
|
|
def _log_to_dict(log):
|
|
d = {c.name: getattr(log, c.name) for c in log.__table__.columns}
|
|
if hasattr(log, 'kpi') and log.kpi:
|
|
d["kpi_code"] = log.kpi.kpi_code
|
|
d["kpi_name"] = log.kpi.kpi_name
|
|
return d
|
|
|
|
|
|
# ============================================================
|
|
# 质量检查
|
|
# ============================================================
|
|
|
|
@router.get("/check")
|
|
def run_quality_check(db: Session = Depends(get_db)):
|
|
"""扫描全部KPI，生成数据质量报告"""
|
|
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
|
|
issues = []
|
|
current_period = datetime.now().strftime("%Y-%m")
|
|
|
|
for kpi in kpis:
|
|
# 获取最近12个月的值
|
|
values = db.query(KPIValue).filter(
|
|
KPIValue.kpi_id == kpi.id,
|
|
KPIValue.actual_value.isnot(None),
|
|
).order_by(KPIValue.period.desc()).limit(12).all()
|
|
|
|
# 1. 检查数据缺失
|
|
if not values:
|
|
issues.append({
|
|
"kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name,
|
|
"check_type": "missing_data",
|
|
"severity": "critical",
|
|
"detail": {"missing_months": 12, "latest_period": None, "total_values": 0},
|
|
"suggestion": "请初始化KPI数据，建议导入至少3个月历史数据",
|
|
})
|
|
continue
|
|
|
|
latest_val = values[0]
|
|
latest_period = latest_val.period
|
|
|
|
# 计算缺失月数
|
|
if latest_period:
|
|
try:
|
|
lp_parts = latest_period.split("-")
|
|
lp_date = datetime(int(lp_parts[0]), int(lp_parts[1]), 1)
|
|
now_date = datetime.now().replace(day=1)
|
|
missing_months = max(0, (now_date.year - lp_date.year) * 12 + (now_date.month - lp_date.month) - 1)
|
|
if missing_months > 1:
|
|
issues.append({
|
|
"kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name,
|
|
"check_type": "missing_data",
|
|
"severity": "warning" if missing_months <= 3 else "critical",
|
|
"detail": {"missing_months": missing_months, "latest_period": latest_period, "total_values": len(values)},
|
|
"suggestion": f"数据缺失{missing_months}个月，建议从ERP系统同步或手动补录",
|
|
})
|
|
except Exception:
|
|
pass
|
|
|
|
# 2. 检查环比骤变（需要至少2个月的值）
|
|
if len(values) >= 2 and latest_val.actual_value:
|
|
prev_val = values[1].actual_value
|
|
if prev_val and prev_val != 0:
|
|
change_pct = abs((latest_val.actual_value - prev_val) / prev_val * 100)
|
|
if change_pct > 50:
|
|
issues.append({
|
|
"kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name,
|
|
"check_type": "abnormal_change",
|
|
"severity": "warning" if change_pct <= 100 else "critical",
|
|
"detail": {
|
|
"change_pct": round(change_pct, 1),
|
|
"current_value": latest_val.actual_value,
|
|
"previous_value": prev_val,
|
|
"current_period": latest_val.period,
|
|
"previous_period": values[1].period,
|
|
},
|
|
"suggestion": f"环比变化{round(change_pct,1)}%，建议核实数据是否录入错误",
|
|
})
|
|
|
|
# 3. 检查连续3期持平
|
|
if len(values) >= 3:
|
|
last_3 = [v.actual_value for v in values[:3] if v.actual_value is not None]
|
|
if len(last_3) >= 3 and len(set(last_3)) == 1:
|
|
issues.append({
|
|
"kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name,
|
|
"check_type": "flat_data",
|
|
"severity": "warning",
|
|
"detail": {"flat_value": last_3[0], "periods": [v.period for v in values[:3]]},
|
|
"suggestion": "连续3期数据完全相同，请确认数据源是否正常更新",
|
|
})
|
|
|
|
# 4. 检查值异常（偏离历史均值超过3倍标准差）
|
|
if len(values) >= 4 and latest_val.actual_value:
|
|
hist_vals = [v.actual_value for v in values[1:] if v.actual_value is not None]
|
|
if len(hist_vals) >= 3:
|
|
mean_val = sum(hist_vals) / len(hist_vals)
|
|
variance = sum((v - mean_val) ** 2 for v in hist_vals) / len(hist_vals)
|
|
stddev = variance ** 0.5 if variance > 0 else mean_val * 0.1
|
|
if stddev > 0 and abs(latest_val.actual_value - mean_val) > 3 * stddev:
|
|
issues.append({
|
|
"kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name,
|
|
"check_type": "value_outlier",
|
|
"severity": "warning",
|
|
"detail": {
|
|
"current_value": latest_val.actual_value,
|
|
"mean": round(mean_val, 2),
|
|
"stddev": round(stddev, 2),
|
|
"z_score": round(abs(latest_val.actual_value - mean_val) / stddev, 2),
|
|
},
|
|
"suggestion": "当前值偏离历史均值超过3倍标准差，建议核实",
|
|
})
|
|
|
|
# 写入质量日志
|
|
created_count = 0
|
|
for issue in issues:
|
|
existing = db.query(KpiDataQualityLog).filter(
|
|
KpiDataQualityLog.kpi_id == issue["kpi_id"],
|
|
KpiDataQualityLog.check_type == issue["check_type"],
|
|
KpiDataQualityLog.status == "open",
|
|
).first()
|
|
if not existing:
|
|
log = KpiDataQualityLog(
|
|
kpi_id=issue["kpi_id"],
|
|
check_type=issue["check_type"],
|
|
severity=issue["severity"],
|
|
detail=issue["detail"],
|
|
suggestion=issue["suggestion"],
|
|
status="open",
|
|
)
|
|
db.add(log)
|
|
created_count += 1
|
|
|
|
db.commit()
|
|
return {
|
|
"total_kpis": len(kpis),
|
|
"issues_found": len(issues),
|
|
"new_logs": created_count,
|
|
"issues": issues,
|
|
}
|
|
|
|
|
|
# ============================================================
|
|
# 质量日志CRUD
|
|
# ============================================================
|
|
|
|
@router.get("/logs")
|
|
def list_quality_logs(
|
|
kpi_id: Optional[int] = None,
|
|
severity: Optional[str] = None,
|
|
check_type: Optional[str] = None,
|
|
status: Optional[str] = None,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""获取数据质量日志"""
|
|
query = db.query(KpiDataQualityLog)
|
|
if kpi_id:
|
|
query = query.filter(KpiDataQualityLog.kpi_id == kpi_id)
|
|
if severity:
|
|
query = query.filter(KpiDataQualityLog.severity == severity)
|
|
if check_type:
|
|
query = query.filter(KpiDataQualityLog.check_type == check_type)
|
|
if status:
|
|
query = query.filter(KpiDataQualityLog.status == status)
|
|
|
|
logs = query.order_by(KpiDataQualityLog.created_at.desc()).limit(100).all()
|
|
result = []
|
|
for log in logs:
|
|
d = _log_to_dict(log)
|
|
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == log.kpi_id).first()
|
|
if kpi:
|
|
d["kpi_code"] = kpi.kpi_code
|
|
d["kpi_name"] = kpi.kpi_name
|
|
result.append(d)
|
|
return {"data": result, "total": len(result)}
|
|
|
|
|
|
@router.put("/logs/{log_id}")
|
|
def update_quality_log(log_id: int, data: dict, db: Session = Depends(get_db), user=WRITE_ROLES):
|
|
"""更新质量日志（解决/忽略）"""
|
|
log = db.query(KpiDataQualityLog).filter(KpiDataQualityLog.id == log_id).first()
|
|
if not log:
|
|
raise HTTPException(404, "日志不存在")
|
|
if "status" in data:
|
|
log.status = data["status"]
|
|
if data["status"] == "resolved":
|
|
log.resolved_at = datetime.now()
|
|
if "suggestion" in data:
|
|
log.suggestion = data["suggestion"]
|
|
db.commit()
|
|
return _log_to_dict(log)
|
|
|
|
|
|
@router.delete("/logs/{log_id}")
|
|
def delete_quality_log(log_id: int, db: Session = Depends(get_db), user=WRITE_ROLES):
|
|
log = db.query(KpiDataQualityLog).filter(KpiDataQualityLog.id == log_id).first()
|
|
if log:
|
|
db.delete(log)
|
|
db.commit()
|
|
return {"message": "已删除"}
|
|
|
|
|
|
# ============================================================
|
|
# 数据质量看板统计
|
|
# ============================================================
|
|
|
|
@router.get("/stats")
|
|
def quality_stats(db: Session = Depends(get_db)):
|
|
"""数据质量统计"""
|
|
total_kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").count()
|
|
total_logs = db.query(KpiDataQualityLog).count()
|
|
open_logs = db.query(KpiDataQualityLog).filter(KpiDataQualityLog.status == "open").count()
|
|
|
|
# 按严重程度统计
|
|
severity_counts = {}
|
|
for s in ("info", "warning", "critical"):
|
|
cnt = db.query(KpiDataQualityLog).filter(
|
|
KpiDataQualityLog.severity == s,
|
|
KpiDataQualityLog.status == "open",
|
|
).count()
|
|
if cnt:
|
|
severity_counts[s] = cnt
|
|
|
|
# 按检查类型统计
|
|
type_counts = {}
|
|
for t in ("abnormal_change", "flat_data", "missing_data", "value_outlier"):
|
|
cnt = db.query(KpiDataQualityLog).filter(
|
|
KpiDataQualityLog.check_type == t,
|
|
KpiDataQualityLog.status == "open",
|
|
).count()
|
|
if cnt:
|
|
type_counts[t] = cnt
|
|
|
|
# ── 数据审计看板统计 ──
|
|
# KPI完整度评分
|
|
all_kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
|
|
total = len(all_kpis)
|
|
complete_kpis = 0
|
|
missing_metadata_count = 0
|
|
missing_data_count = 0
|
|
stale_data_count = 0
|
|
|
|
from datetime import datetime, timedelta
|
|
six_months_ago = datetime.now() - timedelta(days=180)
|
|
|
|
for kpi in all_kpis:
|
|
# 元数据完整度检查
|
|
has_meta = all([
|
|
kpi.formula and kpi.formula.strip(),
|
|
kpi.data_source and kpi.data_source.strip(),
|
|
kpi.data_owner and kpi.data_owner.strip(),
|
|
kpi.unit and kpi.unit.strip(),
|
|
kpi.target_value is not None,
|
|
])
|
|
if has_meta:
|
|
complete_kpis += 1
|
|
else:
|
|
missing_metadata_count += 1
|
|
|
|
# 数据缺失检查（是否有实际值）
|
|
val = db.query(KPIValue).filter(
|
|
KPIValue.kpi_id == kpi.id,
|
|
KPIValue.actual_value.isnot(None),
|
|
).first()
|
|
if not val:
|
|
missing_data_count += 1
|
|
|
|
# 超30天未更新预警
|
|
latest_val = db.query(KPIValue).filter(
|
|
KPIValue.kpi_id == kpi.id,
|
|
KPIValue.actual_value.isnot(None),
|
|
).order_by(KPIValue.period.desc()).first()
|
|
if latest_val and latest_val.calculated_at:
|
|
if latest_val.calculated_at < six_months_ago:
|
|
stale_data_count += 1
|
|
|
|
completeness_score = round(complete_kpis / total * 100, 1) if total > 0 else 0
|
|
missing_rate = round(missing_data_count / total * 100, 1) if total > 0 else 0
|
|
|
|
return {
|
|
"total_kpis": total_kpis,
|
|
"total_logs": total_logs,
|
|
"open_logs": open_logs,
|
|
"severity_counts": severity_counts,
|
|
"type_counts": type_counts,
|
|
# 数据审计看板
|
|
"completeness": {
|
|
"score": completeness_score,
|
|
"complete": complete_kpis,
|
|
"total": total,
|
|
"missing_metadata": missing_metadata_count,
|
|
},
|
|
"data_missing": {
|
|
"count": missing_data_count,
|
|
"rate": missing_rate,
|
|
"total": total,
|
|
},
|
|
"stale_data": {
|
|
"count": stale_data_count,
|
|
"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,
|
|
}
|