716 lines
31 KiB
Python
716 lines
31 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,
|
||
}
|
||
|
||
|
||
# ============================================================
|
||
# 财务七规则检查（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)
|