Files
cma-management/backend/app/api/data_quality.py
T
Hermes CI Fix 242f3cf5ec fix(fetch-caliber): 取数层排除半年/预测伪期间 + subjects 响应补 entity_id
R3 期间口径(仅 YYYY-MM)延伸到取数层:新增 alert_entity_scope.month_period_clause()(跨库可移植 SQL 等价式),应用到 15 个文件 26 处「按 period 倒序取最新值/最近N期」位点;dashboard 原 MySQL-only REGEXP 正则收口到同单一事实源。

GET /api/cma/subjects 响应增补 entity_id(NULL=全局标准准则科目 / N=账套私有明细),既有 10 字段语义与名称不变。

证据:pending-tasks/_reports/yanxue-20260924-cma-fetch-caliber-fixes-fullstack.md(改前/改后 A/B:F_NET_PROFIT 2026-H2/-55.0 → 2026-08/-10.46)。新增 7 例回归测试;全量 1 failed(预存在) 1473 passed,基线同项。
2026-09-26 02:42:04 +08:00

938 lines
43 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""自动数据质量监控 — 任务3
定期检查KPI值异常、连续持平、数据缺失等
"""
from fastapi import APIRouter, Depends, HTTPException, Query, Request
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.deps import assert_entity_scope, get_entity_id
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
from app.utils.alert_entity_scope import month_period_clause
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
# ── 陈旧数据阈值(单一来源 2026-09-24)──
# 阈值取值由任总 2026-09-24 00:35 拍板(见任务件勘误块);本常量为唯一取值来源,
# 接口与前端文案一律引用本值,勿在注释/文案里另写天数(防口径漂移)。
STALE_THRESHOLD_DAYS = 30
# 质量日志分页契约(默认 50/页、上限 200 —— 与 /rule-detail 同规格)
LOG_PAGE_SIZE_DEFAULT = 50
LOG_PAGE_SIZE_MAX = 200
def _scoped_quality_log_query(db: Session, entity_id: int):
"""质量日志「按会话账套收口」的唯一口径(/logs 与 /stats 共用,防两处漂移)
kpi_data_quality_log 表**无 entity_id 列**(本单禁止加列)⇒ 经 kpi_id
join kpi_definitions,取 kpi_definitions.entity_id 判定归属。
内连接语义:kpi_id 无对应 KPI 行的孤儿日志无法归属账套 ⇒ 一律不计(不猜账套)。
"""
return (
db.query(KpiDataQualityLog)
.join(KPIDefinition, KPIDefinition.id == KpiDataQualityLog.kpi_id)
.filter(KPIDefinition.entity_id == entity_id)
)
# ============================================================
# 质量检查
# ============================================================
@router.get("/check")
def run_quality_check(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""扫描本账套KPI,生成数据质量报告(账套隔离 2026-09-21)"""
kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id,
).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),
month_period_clause(KPIValue.period),
).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,
page: int = Query(1, ge=1, description="页码,从 1 开始(下钻与服务端分页)"),
page_size: int = Query(LOG_PAGE_SIZE_DEFAULT, ge=1, le=LOG_PAGE_SIZE_MAX,
description="每页条数,默认 50,上限 200"),
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""获取数据质量日志(账套隔离 2026-09-21;服务端分页 + 真实 total 2026-09-24)
kpi_data_quality_log 表无 entity_id 列(见本文件 :235 说明),按本单选项(1)
经 kpi_id join kpi_definitions 以 kpi_definitions.entity_id 收口(零迁移)。
join 为内连接:kpi_id 无对应 KPI 行的孤儿日志无法归属账套,一律不返回。
`total` = 同筛选条件下的**真实总数**(COUNT(*),不受分页影响)——卡片数字
(/stats)与列表 total 共用 `_scoped_quality_log_query` 同一口径,故二者必然自洽。
"""
query = _scoped_quality_log_query(db, entity_id)
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)
total = query.count()
logs = (
query.order_by(KpiDataQualityLog.created_at.desc(), KpiDataQualityLog.id.desc())
.limit(page_size).offset((page - 1) * page_size).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": total, "page": page, "page_size": page_size}
def _scoped_quality_log(db: Session, log_id: int, entity_id: int):
"""按会话账套取质量日志行(表无 entity_id 列 → 经 kpi_id join kpi_definitions 归属)"""
return (
db.query(KpiDataQualityLog)
.join(KPIDefinition, KPIDefinition.id == KpiDataQualityLog.kpi_id)
.filter(KpiDataQualityLog.id == log_id,
KPIDefinition.entity_id == entity_id)
.first()
)
@router.put("/logs/{log_id}")
def update_quality_log(log_id: int, data: dict, db: Session = Depends(get_db), user=WRITE_ROLES,
entity_id: int = Depends(get_entity_id)):
"""更新质量日志(解决/忽略)(账套隔离 2026-09-21:跨账套 → 404)"""
log = _scoped_quality_log(db, log_id, entity_id)
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,
entity_id: int = Depends(get_entity_id)):
"""删除质量日志(账套隔离 2026-09-21:跨账套 → 404;不存在 → 404,对齐 PUT 语义)"""
log = _scoped_quality_log(db, log_id, entity_id)
if not log:
raise HTTPException(404, "日志不存在")
db.delete(log)
db.commit()
return {"message": "已删除"}
# ============================================================
# 数据质量看板统计
# ============================================================
@router.get("/stats")
def quality_stats(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""数据质量统计(账套隔离 2026-09-21;日志四项计数收口 2026-09-24)
口径(与 /logs 共用 `_scoped_quality_log_query`,故卡片数字与列表 total 必然自洽):
* total_logs / open_logs / severity_counts / type_counts:质量日志**本账套**计数
(kpi_data_quality_log 无 entity_id 列 ⇒ 经 kpi_id join kpi_definitions 归属)
* 层级:异常类四项为**记录数**(条),total_kpis 与完整性/缺失率为**实体数**(个)
"""
total_kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id,
).count()
# 账套收口:本账套(内连接;孤儿日志不计)
total_logs = _scoped_quality_log_query(db, entity_id).count()
open_logs = _scoped_quality_log_query(db, entity_id).filter(
KpiDataQualityLog.status == "open").count()
# 按严重程度统计(本账套)
severity_counts = {}
for s in ("info", "warning", "critical"):
cnt = _scoped_quality_log_query(db, entity_id).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 = _scoped_quality_log_query(db, entity_id).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",
KPIDefinition.entity_id == entity_id, # 账套隔离 2026-09-21
).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
stale_threshold = datetime.now() - timedelta(days=STALE_THRESHOLD_DAYS)
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
# 超 STALE_THRESHOLD_DAYS 天未更新预警(阈值单一来源,勿在注释/文案里另写天数)
latest_val = db.query(KPIValue).filter(
KPIValue.kpi_id == kpi.id,
KPIValue.actual_value.isnot(None),
month_period_clause(KPIValue.period),
).order_by(KPIValue.period.desc()).first()
if latest_val and latest_val.calculated_at:
if latest_val.calculated_at < stale_threshold:
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": STALE_THRESHOLD_DAYS,
},
}
# 数据缺失(data_missing)口径的 KPI 明细 —— 下钻用(只读,2026-09-24)
# 口径与上面 quality_stats() 的 data_missing 计数逐字一致:
# active KPI 且 kpi_values 中不存在 actual_value 非空的行
MISSING_DATA_CALIBER = ("active KPI 且无任何实际值"
"(kpi_values.actual_value 非空的行为 0)")
@router.get("/missing-data-kpis")
def missing_data_kpis(
page: int = Query(1, ge=1, description="页码,从 1 开始"),
page_size: int = Query(LOG_PAGE_SIZE_DEFAULT, ge=1, le=LOG_PAGE_SIZE_MAX,
description="每页条数,默认 50,上限 200"),
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""「缺失率统计」卡片下钻(只读):无任何实际值的 active KPI 清单
- 账套由会话 token 解析(`Depends(get_entity_id)`),**不新增账套入参**;
- `total` = 同口径 COUNT(*)(不受分页影响),与 `/stats.data_missing.count` 必然相等;
- 列表字段:KPI 编码 / 名称 / 负责人 / 数据源 / 单位(不含值明细,概览用)。
"""
sql = ("SELECT d.id, d.kpi_code, d.kpi_name, d.data_owner, d.data_source, d.unit "
"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 AND v.actual_value IS NOT NULL)")
total = int(db.execute(text("SELECT COUNT(*) FROM (" + sql + ") t"),
{"eid": entity_id}).scalar() or 0)
rows = db.execute(
text(sql + " ORDER BY d.id LIMIT :lim OFFSET :off"),
{"eid": entity_id, "lim": page_size, "off": (page - 1) * page_size},
).fetchall()
items = [{
"kpi_id": r.id, "kpi_code": r.kpi_code, "kpi_name": r.kpi_name,
"data_owner": r.data_owner, "data_source": r.data_source, "unit": r.unit,
} for r in rows]
return {"caliber": MISSING_DATA_CALIBER, "total": total, "page": page,
"page_size": page_size, "items": items}
# ============================================================
# 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最多列出的条数(避免响应过大)
# ============================================================
# rule-detail(异常数可点击 → 全量结构化明细)— 2026-09-23 P1
# 口径:total 一律取 COUNT(*)(真实总数,不受明细 LIMIT 影响);
# items 结构化 + 服务端分页(LIMIT/OFFSET,单次响应不塞全量)。
# 判据与 _run_governance_checks 逐字一致 —— 本单只改聚合口径,
# 不改阈值/级别/扣分公式(配置化是第2步,不在本单)。
# ============================================================
RULE_SOURCES = {
"unit_check": "cash_plans",
"dup_alert": "kpi_alerts",
"orphan_check": "kpi_alerts",
"virtual_pollution": "cash_plans",
"entity_check": "kpi_values",
"kpi_completeness": "kpi_definitions",
"reconciliation": "kpi_definitions",
}
RULE_ACTIONS = {
"unit_check": "核对金额单位:确为元则改按万元口径录入,或修正计划金额(阈值 >10000 不变)",
"dup_alert": "同一 plan_id 只保留一条 pending 预警,其余重复项关闭(去重键错误)",
"orphan_check": "确认预警指向的 cash_plans 是否已删除:补建计划或关闭该预警",
"virtual_pollution": "测试/虚拟来源计划不得混入真实数据:归档或清除后重跑本规则",
"entity_check": "修正 kpi_values.entity_id,使其与 kpi_definitions.entity_id 一致",
"kpi_completeness": "为无实际值的 active KPI 补录数据,或按实际情况停用该 KPI",
"reconciliation": "核对预算月度合计与年度目标口径,修正差异 >20% 的预算行",
}
def _plan_id_set(db: Session, eid: int):
"""已存在的 cash_plans id 集合(规则3 判孤儿用;eid 非 0 时只看该实体)"""
sql = "SELECT id FROM cash_plans" + (" WHERE entity_id = :eid" if eid else "")
return {row[0] for row in db.execute(text(sql), {"eid": eid}).fetchall()}
def _pending_alert_plan_links(db: Session, eid: int):
"""规则2/3 共用数据:pending cash_plan 预警 → [(alert_id, kpi_id, plan_id, kpi_entity_id)]
返回 (parsed, raw_rows);Python 侧解析 suggestion JSON(不依赖 MySQL JSON 函数)。
"""
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"
rows = db.execute(text(alert_sql), {"eid": eid}).fetchall()
parsed = []
for r in rows:
pid = _extract_plan_id(r.suggestion)
if pid is not None:
parsed.append((r.id, r.kpi_id, pid, r.kpi_entity_id))
return parsed, rows
def _dup_groups(parsed):
"""规则2 判据:同 plan_id 多条 pending 预警 → [(plan_id, [(alert_id, kpi_id, ent_id), ...])],按条数降序"""
group_map = {}
for alert_id, kpi_id, pid, ent_id in parsed:
group_map.setdefault(pid, []).append((alert_id, kpi_id, ent_id))
groups = [(pid, ids) for pid, ids in group_map.items() if len(ids) > 1]
groups.sort(key=lambda x: -len(x[1]))
return groups
def _detail_count(db: Session, rule: str, eid: int, manual: bool = False) -> int:
"""本规则真实总数:一律 COUNT(*)(dup/orphan 由预警分组得出,同样不受 LIMIT 影响)"""
if rule == "dup_alert":
parsed, _ = _pending_alert_plan_links(db, eid)
return len(_dup_groups(parsed))
if rule == "orphan_check":
parsed, _ = _pending_alert_plan_links(db, eid)
pset = _plan_id_set(db, eid)
return len([1 for _a, _k, pid, _e in parsed if pid not in pset])
ent = " AND cp.entity_id = :eid" if eid else ""
if rule == "unit_check":
sql = "SELECT COUNT(*) FROM cash_plans cp WHERE cp.amount > 10000" + ent
elif rule == "virtual_pollution":
if manual:
sql = "SELECT COUNT(*) FROM cash_plans cp WHERE cp.source = 'manual'" + ent
else:
sql = ("SELECT COUNT(*) FROM cash_plans cp WHERE (cp.source LIKE '%test%' "
"OR cp.source LIKE '%sync%' OR cp.source LIKE '%虚拟%' OR cp.source LIKE '%demo%')" + ent)
elif rule == "entity_check":
sql = ("SELECT COUNT(*) 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:
sql += " AND v.entity_id = :eid"
elif rule == "kpi_completeness":
sql = ("SELECT COUNT(*) 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:
sql += " AND d.entity_id = :eid"
elif rule == "reconciliation":
sql = ("SELECT COUNT(*) FROM (SELECT d.id, "
"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:
sql += " AND d.entity_id = :eid"
sql += " GROUP BY d.id HAVING ABS(diff_pct) > 20) t"
else:
return 0
return int(db.execute(text(sql), {"eid": eid}).scalar() or 0)
def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, manual: bool = False):
"""结构化明细条目(服务端 LIMIT/OFFSET 分页;字段:record_id/entity_id/source/key_fields/summary/suggested_action)"""
if rule == "dup_alert":
parsed, _ = _pending_alert_plan_links(db, eid)
items = []
for pid, alerts in _dup_groups(parsed):
items.append({
"record_id": pid,
"entity_id": alerts[0][2],
"source": RULE_SOURCES[rule],
"key_fields": {"plan_id": pid, "pending_alert_count": len(alerts),
"alert_ids": [a[0] for a in alerts]},
"summary": "计划#%s 存在 %d 条 pending 应收预警(去重键错误)" % (pid, len(alerts)),
"suggested_action": RULE_ACTIONS[rule],
})
return items[offset:offset + limit]
if rule == "orphan_check":
parsed, _ = _pending_alert_plan_links(db, eid)
pset = _plan_id_set(db, eid)
items = []
for alert_id, kpi_id, pid, ent_id in parsed:
if pid in pset:
continue
items.append({
"record_id": alert_id,
"entity_id": ent_id,
"source": RULE_SOURCES[rule],
"key_fields": {"alert_id": alert_id, "kpi_id": kpi_id, "plan_id": pid},
"summary": "预警#%d(kpi#%d) 指向的 plan#%s 不存在" % (alert_id, kpi_id, pid),
"suggested_action": RULE_ACTIONS[rule],
})
return items[offset:offset + limit]
params = {"eid": eid, "lim": limit, "off": offset}
ent = " AND cp.entity_id = :eid" if eid else ""
if rule == "unit_check":
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 :lim OFFSET :off"
), params).fetchall()
return [{
"record_id": r.id,
"entity_id": r.entity_id,
"source": RULE_SOURCES[rule],
"key_fields": {"amount": float(r.amount) if r.amount is not None else None,
"source": r.source, "description": r.description,
"threshold": ">10000(万元口径可疑)"},
"summary": "计划#%d 金额 %s(超 10000 阈值,疑似单位错乱)" % (r.id, r.amount),
"suggested_action": RULE_ACTIONS[rule],
} for r in rows]
if rule == "virtual_pollution":
if manual:
where = "cp.source = 'manual'"
else:
where = ("(cp.source LIKE '%test%' OR cp.source LIKE '%sync%' "
"OR cp.source LIKE '%虚拟%' OR cp.source LIKE '%demo%')")
rows = db.execute(text(
"SELECT cp.id, cp.entity_id, cp.source, cp.description FROM cash_plans cp "
"WHERE " + where + ent + " ORDER BY cp.id LIMIT :lim OFFSET :off"
), params).fetchall()
return [{
"record_id": r.id,
"entity_id": r.entity_id,
"source": RULE_SOURCES[rule],
"key_fields": {"source": r.source, "description": r.description,
"scope": "manual(待人工确认)" if manual else "污染源(test/sync/虚拟/demo)"},
"summary": "计划#%d source=%s%s" % (r.id, r.source, "" if not manual else "(待人工确认)"),
"suggested_action": RULE_ACTIONS[rule],
} for r in rows]
if rule == "entity_check":
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:
sql += " AND v.entity_id = :eid"
sql += " ORDER BY v.id LIMIT :lim OFFSET :off"
rows = db.execute(text(sql), params).fetchall()
return [{
"record_id": r.id,
"entity_id": r.v_eid,
"source": RULE_SOURCES[rule],
"key_fields": {"kpi_id": r.kpi_id, "kpi_code": r.kpi_code,
"value_entity_id": r.v_eid, "definition_entity_id": r.d_eid},
"summary": "值#%d %s:实体%s ≠ 定义实体%s" % (r.id, r.kpi_code, r.v_eid, r.d_eid),
"suggested_action": RULE_ACTIONS[rule],
} for r in rows]
if rule == "kpi_completeness":
sql = ("SELECT d.id, d.entity_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:
sql += " AND d.entity_id = :eid"
sql += " ORDER BY d.id LIMIT :lim OFFSET :off"
rows = db.execute(text(sql), params).fetchall()
return [{
"record_id": r.id,
"entity_id": r.entity_id,
"source": RULE_SOURCES[rule],
"key_fields": {"kpi_code": r.kpi_code, "kpi_name": r.kpi_name, "status": "active"},
"summary": "%s %s(active 但无任何实际值)" % (r.kpi_code, r.kpi_name),
"suggested_action": RULE_ACTIONS[rule],
} for r in rows]
if rule == "reconciliation":
sql = ("SELECT d.id, d.entity_id, 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:
sql += " AND d.entity_id = :eid"
sql += " GROUP BY d.id HAVING ABS(diff_pct) > 20 ORDER BY ABS(diff_pct) DESC LIMIT :lim OFFSET :off"
rows = db.execute(text(sql), params).fetchall()
return [{
"record_id": r.id,
"entity_id": r.entity_id,
"source": RULE_SOURCES[rule],
"key_fields": {"kpi_code": r.kpi_code, "monthly_sum": float(r.monthly_sum or 0),
"target_yearly": float(r.target_yearly or 0), "diff_pct": float(r.diff_pct or 0)},
"summary": "%s 预算合计%s vs 年度目标%s 差异%s%%" % (
r.kpi_code, round(r.monthly_sum or 0, 1), r.target_yearly, r.diff_pct),
"suggested_action": RULE_ACTIONS[rule],
} for r in rows]
return []
# ============================================================
# 财务七规则检查(governance-check)— 2026-08-30 P1 最终方案
# 端点收敛(cma-residual-cross-entity-cleanup-20260914):
# /check-governance 已下线 —— 全仓无调用方(前端仅残留 api 定义、无 .vue 调用);
# 其 entity_id 参数无会话鉴权(0=全部/1/2 任取)→ 跨租户;与 governance-check 功能重复。
# 本端点评分口径:
# * 评分规则: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()
unit_total = _detail_count(db, "unit_check", eid) # 真实总数(不再受明细 LIMIT 200 截断)
issues.append({
"rule": "unit_check", "level": "error",
"count": unit_total,
"detail": [f"plan#{r.id} 金额{r.amount}(疑似元)" for r in rows[:DETAIL_LIMIT]]
+ (["…等%d条" % unit_total] if unit_total > DETAIL_LIMIT else []),
})
# ── 规则2/3 共用数据:pending cash_plan 预警(实体筛选经 kpi_definitions)──
parsed, alert_rows = _pending_alert_plan_links(db, eid)
# 已存在的 cash_plans id 集合(规则3 判断孤儿用;实体筛选时仅看该实体下 plan)
plan_id_set = _plan_id_set(db, eid)
# ── 规则2 重复预警:同 plan_id 多条 pending 预警 ──
dup_groups = _dup_groups(parsed)
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, _ent 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()
poll_total = _detail_count(db, "virtual_pollution", eid) # 真实总数(不再受 LIMIT 200 截断)
manual_total = _detail_count(db, "virtual_pollution", eid, manual=True)
issues.append({
"rule": "virtual_pollution", "level": "error",
"count": poll_total,
"detail": [f"plan#{r.id} source={r.source} desc={r.description or ''}" for r in rows[:DETAIL_LIMIT]]
+ (["…等%d条" % poll_total] if poll_total > DETAIL_LIMIT else []),
"manual_count": manual_total,
"manual_detail": [f"plan#{r.id} source=manual(待人工确认)" for r in manual_rows[:DETAIL_LIMIT]]
+ (["…等%d条" % manual_total] if manual_total > 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()
ent_total = _detail_count(db, "entity_check", eid) # 真实总数
issues.append({
"rule": "entity_check", "level": "error",
"count": ent_total,
"detail": [f"值#{r.id} {r.kpi_code} 实体{r.v_eid}≠定义实体{r.d_eid}" for r in ent_rows[:DETAIL_LIMIT]]
+ (["…等%d条" % ent_total] if ent_total > 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()
comp_total = _detail_count(db, "kpi_completeness", eid) # 真实总数(不再受 LIMIT 300 截断)
issues.append({
"rule": "kpi_completeness", "level": "warning",
"count": comp_total,
"detail": [f"{r.kpi_code} {r.kpi_name}(无值)" for r in comp_rows[:DETAIL_LIMIT]]
+ (["…等%d个KPI" % comp_total] if comp_total > 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()
recon_total = _detail_count(db, "reconciliation", eid) # 真实总数
issues.append({
"rule": "reconciliation", "level": "warning",
"count": recon_total,
"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" % recon_total] if recon_total > 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(
request: Request,
entity_id: Optional[int] = Query(None, ge=0, description="实体ID: 0=全部(仅Bot通道), 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=全部,仅无会话token的Bot通道)。
多租户(cma-cross-entity-endpoints-auth-20260914):
登录会话按 token 绑定账套取数,显式 entity_id 与会话账套不一致 → 403(同 growth-quality 文案);
未传 entity_id 时按会话账套(不再默认全部实体)。
"""
session_entity = assert_entity_scope(request, entity_id, db)
if session_entity is not None:
scope = session_entity
else:
# Bot 通道(无 token):保持原口径(缺省 0=全部)
scope = 0 if entity_id is None else entity_id
return _run_governance_checks(db, scope)
@router.get("/rule-detail")
def rule_detail(
rule: Optional[str] = Query(None, description="七规则 key 之一;缺省/空 → 空集(不 500)"),
page: int = Query(1, ge=1, description="页码,从 1 开始"),
page_size: int = Query(50, ge=1, le=200, description="每页条数,默认 50,上限 200"),
scope: Optional[str] = Query(None, description="virtual_pollution 专用:manual=source='manual' 待人工确认组"),
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""异常明细下钻(只读,2026-09-23 P1「异常数可点击看全量」):
- `total` 一律来自 COUNT(*)(真实总数,不受明细查询 LIMIT 影响);
- `items` 结构化字段:record_id / entity_id / source / key_fields / summary / suggested_action;
- 服务端分页(LIMIT/OFFSET),单次响应不塞全量;
- 账套由会话 token 解析(Depends(get_entity_id)),**不新增账套入参**;
- 兼容:rule 缺省/空 → 空集(200,不 500);未知 rule → 400 + 明确文案。
"""
key = (rule or "").strip()
if not key:
return {"rule": "", "level": "", "name": "", "desc": "",
"scope": None, "total": 0, "page": page, "page_size": page_size, "items": []}
if key not in RULES_META:
raise HTTPException(400, f"未知规则 rule={key};可选: {', '.join(RULES_META.keys())}")
manual = key == "virtual_pollution" and (scope or "").strip().lower() == "manual"
eid = entity_id or 0
total = _detail_count(db, key, eid, manual=manual)
offset = (page - 1) * page_size
items = _detail_items(db, key, eid, page_size, offset, manual=manual)
return {
"rule": key,
"level": RULES_META[key]["level"],
"name": RULES_META[key]["name"],
"desc": RULES_META[key]["desc"],
"scope": "manual" if manual else None,
"total": total,
"page": page,
"page_size": page_size,
"items": items,
}