935 lines
43 KiB
Python
935 lines
43 KiB
Python
"""自动数据质量监控 — 任务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
|
||
|
||
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),
|
||
).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),
|
||
).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,
|
||
}
|
||
|