diff --git a/backend/app/api/data_quality.py b/backend/app/api/data_quality.py index ee1f91aa..cddf5167 100644 --- a/backend/app/api/data_quality.py +++ b/backend/app/api/data_quality.py @@ -4,7 +4,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request from sqlalchemy.orm import Session from sqlalchemy import func, and_, text -from typing import Optional +from typing import Optional, Any, Tuple from datetime import datetime, timedelta import json import logging @@ -12,7 +12,11 @@ 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.models import (KPIDefinition, KPIValue, KpiDataQualityLog, OperationLog, + QualityRuleConfig, User) +from app.utils.oplog import log_op +from app import dq_rule_config as dqcfg +from pydantic import BaseModel, Field from app.api.kpis import kpi_to_dict from app.utils.alert_entity_scope import month_period_clause @@ -485,6 +489,100 @@ RULE_ACTIONS = { } +# ============================================================ +# 规则配置取值口（ASK-20260928-05）—— 引擎侧**唯一**配置读取入口 +# 取值经 `app/dq_rule_config.py` 三层解析（账套行 → 模板行(entity_id=0) → 内置默认； +# 禁止跨账套回退）。本文件**不得**再出现可配置项的裸露常量（阈值 / 关键词 / 条数下限 / +# 差异阈值与方向 / 扣分上限与系数 / 明细条数）—— 否则「改了配置不生效」。 +# 锁死面（级别、判定表达式本体、评分函数、作用域、规则集…）仍在本文件硬编码，见清单 §4。 +# ============================================================ +RECON_DIFF_EXPR = ("ROUND((COALESCE(SUM(b.budget_value),0) - d.target_yearly) " + "/ d.target_yearly * 100, 1)") +RECON_BASE_SQL = ("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'") + + +def _load_cfg(db: Session, eid: int) -> dict: + """完整解析结果（含每项来源与 config_version）；任何异常 → 内置默认形态（读路径绝不 500）。""" + try: + return dqcfg.resolve(db, eid) + except Exception: # noqa: BLE001 + logger.warning("DQ_RULE_CONFIG_FALLBACK_BUILTIN entity=%s", eid, exc_info=True) + return dqcfg.builtin_resolution(eid) + + +def _cfg_values(db: Session, eid: int) -> dict: + """引擎取值口：{rule_key: {param: value}, ..., 'global': {...}}""" + return _load_cfg(db, eid)["values"] + + +def _rule_cfg(cfg: dict, rule: str) -> dict: + return (cfg or {}).get(rule) or {} + + +def _detail_limit(cfg: dict, rule: str) -> int: + return int(_rule_cfg(cfg, rule).get("detail_limit") or dqcfg.DEFAULT_DETAIL_LIMIT) + + +def _unit_threshold(cfg: dict) -> int: + return int(_rule_cfg(cfg, "unit_check").get("amount_threshold", 10000)) + + +def _dup_min_count(cfg: dict) -> int: + return int(_rule_cfg(cfg, "dup_alert").get("min_count", 2)) + + +def _manual_source(cfg: dict) -> str: + return str(_rule_cfg(cfg, "virtual_pollution").get("manual_source_value") or "manual") + + +def _recon_pct(cfg: dict) -> float: + return float(_rule_cfg(cfg, "reconciliation").get("diff_threshold_pct", 20)) + + +def _recon_having(cfg: dict) -> str: + """规则7 比较方向：仅三种枚举（abs/upper_only/lower_only），不接受自由表达式。""" + mode = str(_rule_cfg(cfg, "reconciliation").get("compare_mode") or "abs") + if mode == "upper_only": + return "diff_pct > :pct" + if mode == "lower_only": + return "diff_pct < (0 - :pct)" + return "ABS(diff_pct) > :pct" + + +def _entity_clause(eid: int, col: str) -> Tuple[str, dict]: + """账套过滤子句 + 绑定参数（eid=0 → 全账套，无子句）。""" + if eid: + return f" AND {col} = :eid", {"eid": eid} + return "", {} + + +def _vp_predicate(cfg: dict) -> Tuple[str, dict]: + """规则4 污染主体谓词：嫌疑关键词（大小写不敏感包含式）**减去**来源白名单（豁免优先）。 + + 关键词/白名单均来自配置；匹配方式固定为包含式（禁正则与 SQL 片段 —— 锁死面第 6 条）； + 关键词为空 ⇒ 恒假谓词（不误报任何行）。绑定参数化，无字面量拼接（注入面 = 0）。 + """ + vp = _rule_cfg(cfg, "virtual_pollution") + keywords = [str(k) for k in (vp.get("suspect_keywords") or []) if str(k).strip()] + whitelist = [str(w) for w in (vp.get("source_whitelist") or []) if str(w).strip()] + params: dict = {} + parts = [] + for i, kw in enumerate(keywords): + key = f"vpkw{i}" + parts.append(f"LOWER(cp.source) LIKE :{key}") + params[key] = f"%{kw.lower()}%" + pred = "(" + " OR ".join(parts) + ")" if parts else "(1 = 0)" + if whitelist: + names = [] + for i, w in enumerate(whitelist): + key = f"vpwl{i}" + names.append(f":{key}") + params[key] = w.lower() + pred += f" AND LOWER(cp.source) NOT IN ({', '.join(names)})" + return pred, params + + 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 "") @@ -512,64 +610,79 @@ def _pending_alert_plan_links(db: Session, eid: int): return parsed, rows -def _dup_groups(parsed): - """规则2 判据：同 plan_id 多条 pending 预警 → [(plan_id, [(alert_id, kpi_id, ent_id), ...])]，按条数降序""" +def _dup_groups(parsed, min_count: int = 2): + """规则2 判据：同 plan_id 的 pending 预警条数 **>= min_count**（默认 2 == 原 `len(ids) > 1`） + + min_count 来自配置（`dup_alert.min_count`，清单 §3.2，边界 [2,20] 越界拒绝）。 + """ + threshold = max(2, int(min_count or 2)) 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 = [(pid, ids) for pid, ids in group_map.items() if len(ids) >= threshold] groups.sort(key=lambda x: -len(x[1])) return groups -def _detail_count(db: Session, rule: str, eid: int, manual: bool = False) -> int: +def _detail_count(db: Session, rule: str, eid: int, manual: bool = False, + cfg: Optional[dict] = None) -> int: """本规则真实总数：一律 COUNT(*)（dup/orphan 由预警分组得出，同样不受 LIMIT 影响）""" + cfg = cfg or _cfg_values(db, eid) + if rule == "dup_alert": parsed, _ = _pending_alert_plan_links(db, eid) - return len(_dup_groups(parsed)) + return len(_dup_groups(parsed, _dup_min_count(cfg))) 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 "" + ent, params = _entity_clause(eid, "cp.entity_id") if rule == "unit_check": - sql = "SELECT COUNT(*) FROM cash_plans cp WHERE cp.amount > 10000" + ent + sql = "SELECT COUNT(*) FROM cash_plans cp WHERE cp.amount > :amt" + ent + params["amt"] = _unit_threshold(cfg) elif rule == "virtual_pollution": if manual: - sql = "SELECT COUNT(*) FROM cash_plans cp WHERE cp.source = 'manual'" + ent + sql = "SELECT COUNT(*) FROM cash_plans cp WHERE cp.source = :msrc" + ent + params["msrc"] = _manual_source(cfg) 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) + pred, vp_params = _vp_predicate(cfg) + params.update(vp_params) + sql = "SELECT COUNT(*) FROM cash_plans cp WHERE " + pred + 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" + params["eid"] = 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" + params["eid"] = 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'") + sql = ("SELECT COUNT(*) FROM (SELECT d.id, " + RECON_DIFF_EXPR + " AS diff_pct " + + RECON_BASE_SQL) if eid: sql += " AND d.entity_id = :eid" - sql += " GROUP BY d.id HAVING ABS(diff_pct) > 20) t" + params["eid"] = eid + sql += " GROUP BY d.id HAVING " + _recon_having(cfg) + ") t" + params["pct"] = _recon_pct(cfg) else: return 0 - return int(db.execute(text(sql), {"eid": eid}).scalar() or 0) + return int(db.execute(text(sql), params).scalar() or 0) -def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, manual: bool = False): +def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, + manual: bool = False, cfg: Optional[dict] = None): """结构化明细条目（服务端 LIMIT/OFFSET 分页；字段：record_id/entity_id/source/key_fields/summary/suggested_action）""" + cfg = cfg or _cfg_values(db, eid) + if rule == "dup_alert": parsed, _ = _pending_alert_plan_links(db, eid) items = [] - for pid, alerts in _dup_groups(parsed): + for pid, alerts in _dup_groups(parsed, _dup_min_count(cfg)): items.append({ "record_id": pid, "entity_id": alerts[0][2], @@ -598,41 +711,43 @@ def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, man }) return items[offset:offset + limit] - params = {"eid": eid, "lim": limit, "off": offset} - ent = " AND cp.entity_id = :eid" if eid else "" + ent, params = _entity_clause(eid, "cp.entity_id") + params.update({"lim": limit, "off": offset}) if rule == "unit_check": + threshold = _unit_threshold(cfg) 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() + "WHERE cp.amount > :amt" + ent + " ORDER BY cp.amount DESC LIMIT :lim OFFSET :off" + ), {**params, "amt": threshold}).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), + "threshold": ">%s（万元口径可疑）" % threshold}, + "summary": "计划#%d 金额 %s（超 %s 阈值，疑似单位错乱）" % (r.id, r.amount, threshold), "suggested_action": RULE_ACTIONS[rule], } for r in rows] if rule == "virtual_pollution": if manual: - where = "cp.source = 'manual'" + where, kw_params = "cp.source = :msrc", {"msrc": _manual_source(cfg)} + scope_text = "%s（待人工确认）" % _manual_source(cfg) else: - where = ("(cp.source LIKE '%test%' OR cp.source LIKE '%sync%' " - "OR cp.source LIKE '%虚拟%' OR cp.source LIKE '%demo%')") + where, kw_params = _vp_predicate(cfg) + scope_text = "污染源（%s）" % "/".join(cfg["virtual_pollution"]["suspect_keywords"]) 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() + ), {**params, **kw_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）"}, + "scope": scope_text}, "summary": "计划#%d source=%s%s" % (r.id, r.source, "" if not manual else "（待人工确认）"), "suggested_action": RULE_ACTIONS[rule], } for r in rows] @@ -644,6 +759,8 @@ def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, man if eid: sql += " AND v.entity_id = :eid" sql += " ORDER BY v.id LIMIT :lim OFFSET :off" + if not eid: + params.pop("eid", None) rows = db.execute(text(sql), params).fetchall() return [{ "record_id": r.id, @@ -661,6 +778,8 @@ def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, man if eid: sql += " AND d.entity_id = :eid" sql += " ORDER BY d.id LIMIT :lim OFFSET :off" + if not eid: + params.pop("eid", None) rows = db.execute(text(sql), params).fetchall() return [{ "record_id": r.id, @@ -674,13 +793,15 @@ def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, man 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'") + + RECON_DIFF_EXPR + " AS diff_pct " + RECON_BASE_SQL) 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() + sql += (" GROUP BY d.id HAVING " + _recon_having(cfg) + + " ORDER BY ABS(diff_pct) DESC LIMIT :lim OFFSET :off") + recon_params = {**params, "pct": _recon_pct(cfg)} + if not eid: + recon_params.pop("eid", None) + rows = db.execute(text(sql), recon_params).fetchall() return [{ "record_id": r.id, "entity_id": r.entity_id, @@ -695,16 +816,16 @@ def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, man 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) +# * 评分规则：error 扣 min(error 上限, count×系数)，warning 扣 min(warning 上限, count×系数) +# —— 上限/系数自 2026-09-28（ASK-20260928-05）起取自 `dq_rule_config`（默认 = 原 15×3 / 10×1） # * 规则2/3 用 Python 解析 suggestion JSON（不依赖 MySQL JSON 函数） -# * 规则4 区分 error(test/sync/虚拟) 与 manual(待人工确认 warning) +# * 规则4 区分 污染关键词（error）与 manual（待人工确认 warning） # * 规则7 按 status='active' 口径（与 budget.py 一致，避免多版本叠加失真） # 只读幂等：不写库、不创建 KpiDataQualityLog # ============================================================ @@ -721,142 +842,204 @@ def _extract_plan_id(suggestion: str): 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 = [] +def _disabled_issue(rule: str, cfg: dict) -> dict: + """被配置关闭的规则：零扣分、零明细，但**仍占位**（规则集锁死 7 条，响应形状稳定）。""" + return {"rule": rule, "level": RULES_META[rule]["level"], "enabled": False, + "count": 0, "detail": [], "deducted": 0} - # ── 规则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 []), - }) + +def _run_governance_checks(db: Session, entity_id: int = 0): + """执行财务七规则，返回 issues 列表（含 deducted 扣分）。entity_id=0 表示全部实体。 + + 配置化（ASK-20260928-05）：阈值/关键词/条数下限/差异阈值与方向/扣分上限与系数/明细条数 + 一律经 `_cfg_values()`（三层解析）取值；级别与判定表达式本体**锁死**（见 dq_rule_config）。 + """ + eid = entity_id or 0 + cfg = _cfg_values(db, eid) + resolution = _load_cfg(db, eid) + issues = [] + disabled = [] + + # ── 规则1 单位校验：amount > 阈值（默认 10000 万元口径可疑）── + if cfg["unit_check"].get("enabled", True): + threshold = _unit_threshold(cfg) + limit = _detail_limit(cfg, "unit_check") + ent, params = _entity_clause(eid, "cp.entity_id") + rows = db.execute(text( + "SELECT cp.id, cp.entity_id, cp.amount, cp.source, cp.description " + "FROM cash_plans cp WHERE cp.amount > :amt" + ent + " ORDER BY cp.amount DESC LIMIT 200" + ), {**params, "amt": threshold}).fetchall() + unit_total = _detail_count(db, "unit_check", eid, cfg=cfg) # 真实总数（不受明细 LIMIT 截断） + issues.append({ + "rule": "unit_check", "level": RULES_META["unit_check"]["level"], + "count": unit_total, + "detail": [f"plan#{r.id} 金额{r.amount}(疑似元)" for r in rows[:limit]] + + (["…等%d条" % unit_total] if unit_total > limit else []), + }) + else: + disabled.append("unit_check") + issues.append(_disabled_issue("unit_check", cfg)) # ── 规则2/3 共用数据：pending cash_plan 预警（实体筛选经 kpi_definitions）── - parsed, alert_rows = _pending_alert_plan_links(db, eid) + 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 []), - }) + # ── 规则2 重复预警：同 plan_id 的 pending 预警 ≥ min_count（默认 2）── + if cfg["dup_alert"].get("enabled", True): + limit = _detail_limit(cfg, "dup_alert") + dup_groups = _dup_groups(parsed, _dup_min_count(cfg)) + issues.append({ + "rule": "dup_alert", "level": RULES_META["dup_alert"]["level"], + "count": len(dup_groups), + "detail": [f"plan#{pid} 重复预警×{len(ids)}" for pid, ids in dup_groups[:limit]] + + (["…等%d个plan" % len(dup_groups)] if len(dup_groups) > limit else []), + }) + else: + disabled.append("dup_alert") + issues.append(_disabled_issue("dup_alert", cfg)) # ── 规则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 []), - }) + if cfg["orphan_check"].get("enabled", True): + limit = _detail_limit(cfg, "orphan_check") + 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": RULES_META["orphan_check"]["level"], + "count": len(orphan_rows), + "detail": [f"预警#{aid}(kpi#{kid}) → plan#{pid} 不存在" for aid, kid, pid in orphan_rows[:limit]] + + (["…等%d条" % len(orphan_rows)] if len(orphan_rows) > limit else []), + }) + else: + disabled.append("orphan_check") + issues.append(_disabled_issue("orphan_check", cfg)) - # ── 规则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 []), - }) + # ── 规则4 虚拟污染：source 命中嫌疑关键词（去白名单）→ error；manual 值 → 待人工确认 warning ── + if cfg["virtual_pollution"].get("enabled", True): + limit = _detail_limit(cfg, "virtual_pollution") + ent, params = _entity_clause(eid, "cp.entity_id") + pred, vp_params = _vp_predicate(cfg) + rows = db.execute(text( + "SELECT cp.id, cp.entity_id, cp.source, cp.description FROM cash_plans cp " + "WHERE " + pred + ent + " ORDER BY cp.id LIMIT 200" + ), {**params, **vp_params}).fetchall() + manual_rows = db.execute(text( + "SELECT cp.id, cp.entity_id, cp.source, cp.description FROM cash_plans cp " + "WHERE cp.source = :msrc" + ent + " ORDER BY cp.id LIMIT 200" + ), {**params, "msrc": _manual_source(cfg)}).fetchall() + poll_total = _detail_count(db, "virtual_pollution", eid, cfg=cfg) + manual_total = _detail_count(db, "virtual_pollution", eid, manual=True, cfg=cfg) + issues.append({ + "rule": "virtual_pollution", "level": RULES_META["virtual_pollution"]["level"], + "count": poll_total, + "detail": [f"plan#{r.id} source={r.source} desc={r.description or ''}" for r in rows[:limit]] + + (["…等%d条" % poll_total] if poll_total > limit else []), + "manual_count": manual_total, + "manual_detail": [f"plan#{r.id} source=manual(待人工确认)" for r in manual_rows[:limit]] + + (["…等%d条" % manual_total] if manual_total > limit else []), + }) + else: + disabled.append("virtual_pollution") + issues.append(_disabled_issue("virtual_pollution", cfg)) # ── 规则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 []), - }) + if cfg["entity_check"].get("enabled", True): + limit = _detail_limit(cfg, "entity_check") + 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, cfg=cfg) # 真实总数 + issues.append({ + "rule": "entity_check", "level": RULES_META["entity_check"]["level"], + "count": ent_total, + "detail": [f"值#{r.id} {r.kpi_code} 实体{r.v_eid}≠定义实体{r.d_eid}" for r in ent_rows[:limit]] + + (["…等%d条" % ent_total] if ent_total > limit else []), + }) + else: + disabled.append("entity_check") + issues.append(_disabled_issue("entity_check", cfg)) # ── 规则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 []), - }) + if cfg["kpi_completeness"].get("enabled", True): + limit = _detail_limit(cfg, "kpi_completeness") + 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, cfg=cfg) + issues.append({ + "rule": "kpi_completeness", "level": RULES_META["kpi_completeness"]["level"], + "count": comp_total, + "detail": [f"{r.kpi_code} {r.kpi_name}(无值)" for r in comp_rows[:limit]] + + (["…等%d个KPI" % comp_total] if comp_total > limit else []), + }) + else: + disabled.append("kpi_completeness") + issues.append(_disabled_issue("kpi_completeness", cfg)) - # ── 规则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 []), - }) + # ── 规则7 勾稽验证：预算(active口径)合计 vs 年度目标，差异按配置阈值/方向判定 ── + if cfg["reconciliation"].get("enabled", True): + limit = _detail_limit(cfg, "reconciliation") + recon_sql = ( + "SELECT d.kpi_code, d.kpi_name, d.target_yearly, " + "COALESCE(SUM(b.budget_value),0) AS monthly_sum, " + + RECON_DIFF_EXPR + " AS diff_pct " + RECON_BASE_SQL + ) + if eid: + recon_sql += " AND d.entity_id = :eid" + recon_sql += (" GROUP BY d.id HAVING " + _recon_having(cfg) + + " ORDER BY ABS(diff_pct) DESC LIMIT 200") + recon_rows = db.execute(text(recon_sql), + {"eid": eid, "pct": _recon_pct(cfg)}).fetchall() + recon_total = _detail_count(db, "reconciliation", eid, cfg=cfg) + issues.append({ + "rule": "reconciliation", "level": RULES_META["reconciliation"]["level"], + "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[:limit]] + + (["…等%d个KPI" % recon_total] if recon_total > limit else []), + }) + else: + disabled.append("reconciliation") + issues.append(_disabled_issue("reconciliation", cfg)) - # ── 评分：error 扣 min(15, count*3)，warning 扣 min(10, count*1)；规则4 manual 按 warning 附加扣 ── + # ── 评分：各规则的「上限 × 系数」取自配置（默认 error 15×3 / warning 10×1；规则4 manual 附加）── total_deduct = 0 for item in issues: + rule = item["rule"] + rule_cfg = cfg.get(rule) or {} + if not rule_cfg.get("enabled", True): + item["deducted"] = 0 + continue ded = 0 if item["count"] > 0: - ded += min(15, item["count"] * 3) if item["level"] == "error" else min(10, item["count"] * 1) + cap = int(rule_cfg.get("deduct_cap", + 15 if item["level"] == "error" else 10)) + factor = int(rule_cfg.get("deduct_factor", + 3 if item["level"] == "error" else 1)) + ded += min(cap, item["count"] * factor) # 规则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) + if rule == "virtual_pollution" and manual_cnt > 0: + ded += min(int(rule_cfg.get("manual_cap", 10)), + manual_cnt * int(rule_cfg.get("manual_factor", 1))) item["deducted"] = ded total_deduct += ded score = max(0, 100 - total_deduct) - passed = [item["rule"] for item in issues if item["count"] == 0] + passed = [item["rule"] for item in issues if item["count"] == 0 and item["rule"] not in disabled] + excluded = list(resolution.get("scope", {}).get("excluded_entities") or []) return { "checked_at": datetime.now().isoformat(timespec="seconds"), "entity_id": eid, @@ -865,7 +1048,15 @@ def _run_governance_checks(db: Session, entity_id: int = 0): "total_rules": len(issues), "issues": issues, "passed": passed, + "disabled": disabled, "rules_meta": RULES_META, + # 配置化（ASK-20260928-05）：口径快照指纹 —— 事后可解释「当时的分数是哪一版配置算出来的」 + "config_version": resolution.get("config_version"), + "config_source": {"table_ready": resolution.get("table_ready"), + "entity_config_enabled": resolution.get("entity_config_enabled")}, + # 演示/测试账套排除（F6）：默认空集 ⇒ 不改变评分语义，只做标注 + "scope_excluded_entities": excluded, + "excluded_from_business_caliber": bool(eid and eid in excluded), } @@ -878,7 +1069,8 @@ def governance_check( """财务七规则检查（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-总扣分) + 评分: 满分100，各规则扣分 = min(配置上限, 命中数×配置系数)（默认 error 15×3、warning 10×1）， + score = max(0, 100 − 总扣分)；阈值/关键词/权重自 2026-09-28 起由 quality_rule_config 配置（ASK-20260928-05）。 实体筛选: entity_id 参数（0=全部，仅无会话token的Bot通道）。 多租户（cma-cross-entity-endpoints-auth-20260914）： @@ -899,7 +1091,7 @@ 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' 待人工确认组"), + scope: Optional[str] = Query(None, description="virtual_pollution 专用：manual=待人工确认组"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): @@ -909,7 +1101,8 @@ def rule_detail( - `items` 结构化字段：record_id / entity_id / source / key_fields / summary / suggested_action； - 服务端分页（LIMIT/OFFSET），单次响应不塞全量； - 账套由会话 token 解析（Depends(get_entity_id)），**不新增账套入参**； - - 兼容：rule 缺省/空 → 空集（200，不 500）；未知 rule → 400 + 明确文案。 + - 兼容：rule 缺省/空 → 空集（200，不 500）；未知 rule → 400 + 明确文案； + - 判据与 `governance-check` 共用同一配置解析（2026-09-28 起），两处口径不会漂移。 """ key = (rule or "").strip() if not key: @@ -920,18 +1113,177 @@ def rule_detail( manual = key == "virtual_pollution" and (scope or "").strip().lower() == "manual" eid = entity_id or 0 - total = _detail_count(db, key, eid, manual=manual) + cfg = _cfg_values(db, eid) + total = _detail_count(db, key, eid, manual=manual, cfg=cfg) offset = (page - 1) * page_size - items = _detail_items(db, key, eid, page_size, offset, manual=manual) + items = _detail_items(db, key, eid, page_size, offset, manual=manual, cfg=cfg) return { "rule": key, "level": RULES_META[key]["level"], "name": RULES_META[key]["name"], "desc": RULES_META[key]["desc"], "scope": "manual" if manual else None, + "enabled": bool((cfg.get(key) or {}).get("enabled", True)), "total": total, "page": page, "page_size": page_size, "items": items, } + +# ============================================================ +# 规则配置端点（ASK-20260928-05 最小实现：新表 + 端点 + 只读 UI） +# 读：GET /rule-config（四角色，账套由 token 绑定，不新增账套入参） +# 写：PUT /rule-config（ceo/it；平台级参数仅 it；Bot 通道禁写；变更与审计同事务） +# 史：GET /rule-config/history（ceo/it） +# 边界（清单 §4/§5）：级别、判定表达式、评分函数、作用域等 10 类锁死 —— 不在 SPEC 即拒绝。 +# ============================================================ + + +class DqRuleConfigUpdate(BaseModel): + """单参数配置变更（一次只改一档：清单 §5.1「一次只改一档，改完复跑七规则对账」）。""" + rule_key: str = Field(..., description="七规则 key 之一，或 global（全局项）") + param_key: str = Field(..., description="参数名（见 GET /rule-config 的 params）") + value: Any = Field(None, description="新值（类型见参数规格：bool/int/float/str/数组/枚举）") + change_reason: str = Field(..., description=f"变更理由（≥{dqcfg.MIN_REASON_LEN} 字，落审计）") + + +@router.get("/rule-config") +def get_rule_config( + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """财务七规则配置（**只读**）：三层解析后的生效值 + 每项来源 + 锁死项清单。 + + 来源语义：`entity`=本账套覆盖行 / `template`=全账套模板行(entity_id=0) / `builtin`=代码内置默认。 + 账套由会话 token 解析（`Depends(get_entity_id)`），**不新增账套入参**；演示/测试账套只读不可写。 + """ + res = _load_cfg(db, entity_id) + return { + "entity_id": entity_id, + "config_entity_id": res["config_entity_id"], + "entity_config_enabled": res["entity_config_enabled"], + "config_entities": list(dqcfg.CONFIG_ENTITIES), + "table_ready": res["table_ready"], + "config_version": res["config_version"], + "rule_param_count": dqcfg.count_rule_params(), + "rules": { + key: { + "name": RULES_META[key]["name"], + "level": RULES_META[key]["level"], + "desc": RULES_META[key]["desc"], + "params": res["rules"][key], + } for key in dqcfg.RULE_KEYS + }, + "global_params": res["global"], + "scope": res["scope"], + "scope_note": "演示/测试账套排除（F6）当前只做标注、不改评分语义；默认空集 = 不排除", + "locked": [dict(it) for it in dqcfg.LOCKED_ITEMS], + "readonly": True, + "generated_at": datetime.now().isoformat(timespec="seconds"), + } + + +@router.get("/rule-config/history") +def get_rule_config_history( + limit: int = Query(20, ge=1, le=200, description="返回条数上限"), + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + _user: User = Depends(require_role("ceo", "it")), +): + """本账套配置变更历史（ceo/it）——读 `operation_logs`（action=update_dq_rule_config）。 + + `operation_logs` 无 entity_id 列 ⇒ 按 `detail.entity_id` 过滤（清单 §5.2 既定口径）。 + """ + return {"entity_id": entity_id, "items": dqcfg.history(db, entity_id, limit)} + + +@router.put("/rule-config") +def update_rule_config( + payload: DqRuleConfigUpdate, + request: Request, + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), + current_user: User = Depends(require_role("ceo", "it")), +): + """修改单个配置项（ceo/it；平台级参数仅 it）。 + + 校验顺序：账套可写 → Bot 通道禁写 → 参数在开放面（锁死/未知一律 400） + → 平台级仅 it → change_reason → 取值边界（拒绝或钳制）→ 写行 + 审计同事务。 + """ + if entity_id not in dqcfg.CONFIG_ENTITIES: + raise HTTPException( + 400, f"账套 entity_id={entity_id} 不开放规则配置（仅 {list(dqcfg.CONFIG_ENTITIES)}；" + f"演示/测试账套只读，ASK-20260928-05 口径②）") + if str(request.headers.get("X-BOT-KEY") or "").strip(): + # 锁死面第 9 条：Bot 通道无 users 行 → 无操作人可留痕 = 审计断链 + raise HTTPException(403, "Bot 通道禁止写规则配置（无操作人可留痕，审计会断链）") + + rule_key = (payload.rule_key or "").strip() + param_key = (payload.param_key or "").strip() + if dqcfg.param_spec(rule_key, param_key) is None: + # 复用参数校验器的文案：锁死项 / 未知 key / 该规则无开放参数 + probe = dqcfg.validate_param(rule_key, param_key, payload.value) + raise HTTPException(400, probe["error"]) + + level = dqcfg.open_level_of(rule_key, param_key) + if level == dqcfg.OPEN_LEVEL_PLATFORM and current_user.role != "it": + raise HTTPException(403, f"参数 {rule_key}.{param_key} 为平台级配置（仅 it 可写）") + + reason = (payload.change_reason or "").strip() + if len(reason) < dqcfg.MIN_REASON_LEN: + raise HTTPException(400, f"change_reason 必填且不少于 {dqcfg.MIN_REASON_LEN} 字" + f"（防无痕调参刷分；清单 §5.1）") + + checked = dqcfg.validate_param(rule_key, param_key, payload.value) + if not checked["ok"]: + raise HTTPException(400, checked["error"]) + + rule_name = RULES_META.get(rule_key, {}).get("name") or rule_key + try: + result = dqcfg.upsert_param(db, entity_id, rule_key, param_key, checked["value"], + username=getattr(current_user, "username", None)) + log_op( + db, dqcfg.ACTION_UPDATE, target_type=dqcfg.TARGET_TYPE, target_id=result["id"], + user_id=current_user.id, entity_id=entity_id, + detail={ + "entity_id": entity_id, + "rule_key": rule_key, + "rule_name": rule_name, + "param_key": param_key, + "before": result["before"], + "after": result["after"], + "reason": reason, + "open_level": level, + "client_ip": (request.client.host if request.client else None), + "operator": {"user_id": current_user.id, + "username": getattr(current_user, "username", None), + "role": current_user.role}, + }, + ) + # 变更与审计同一事务提交：审计写不进 ⇒ 配置变更一并回滚 + # （绝不留「改了但没审计」的窗口；形态照抄 tenant.py:148 先例） + db.commit() + except Exception: + db.rollback() + logger.error("DQ_RULE_CONFIG_AUDIT_WRITE_FAILED entity=%s %s.%s user=%s → 变更已回滚", + entity_id, rule_key, param_key, getattr(current_user, "id", None), + exc_info=True) + raise HTTPException(500, "审计写入失败，配置变更已回滚") + + after = _load_cfg(db, entity_id) + return { + "success": True, + "entity_id": entity_id, + "rule_key": rule_key, + "param_key": param_key, + "before": result["before"], + "before_source": result["before_source"], + "after": checked["value"], + "warning": checked["warning"], + "config_version": after["config_version"], + "changed_by": getattr(current_user, "username", None), + "audit": {"action": dqcfg.ACTION_UPDATE, "target_type": dqcfg.TARGET_TYPE, + "target_id": result["id"]}, + "note": "配置即生效（下次 governance-check 按新口径计算）；历史评分不重算", + } diff --git a/backend/app/dq_rule_config.py b/backend/app/dq_rule_config.py new file mode 100644 index 00000000..107df1ed --- /dev/null +++ b/backend/app/dq_rule_config.py @@ -0,0 +1,618 @@ +# -*- coding: utf-8 -*- +"""DAMA 数据质量治理规则可配置项（财务七规则）—— 单一事实源 + 三层解析 + +任务：ASK-20260928-05（任总 2026-09-28 13:38「放行 4 和 5」，本项目 Bot 转呈） +来源清单：wecom-dama《财务七规则可配置化 — 可配置项清单 + 建议默认值 + 权限/留痕建议》 + （`wecom-dama/pending-tasks/_reports/project-20260923-quality-rules-config-inventory-dama.md`） + +本模块承担三件事（唯一来源，勿在别处复制常量） +---------------------------------------------- +1. **开放面**：37 个规则级参数（`SPEC`）+ 2 个全局项（`GLOBAL_SPEC`）。清单 §3 逐条落成代码； + 测试断言项数 = 37 / 2，防后续会话漏项或私自扩面。 +2. **锁死面**：清单 §4 的 10 类（判定表达式本体 / 级别归属 / 评分函数 / 实体作用域 / + 规则集与数量 / 匹配方式自由化 / kpi_completeness 口径源 / `entity_id IS NULL` 处置 / + Bot 通道写配置 / 越界权重）——不在 SPEC 里的键一律 400（写端点）， + `LOCKED_PARAM_KEYS` 仅用于把「锁死」讲清楚（可读文案），不是白名单。 +3. **三层解析**：账套行（`entity_id`）→ 模板行（`entity_id=0`）→ 代码内置默认； + **禁止跨账套回退**（历史事故：`cash.current_balance` 全局键被全账套共用）。 + +口径（任总 2026-09-28 放行原文：「口径用清单默认值」） +---------------------------------------------------- +* 默认值 = 清单 §7「建议默认值」总表，即「现行值，上线零行为差异」， + **唯一例外**：`virtual_pollution.suspect_keywords` 去掉 `sync` + （`zxt-sync` 是酣客生产同步来源，现行 134 行 100% 假阳性）——本单唯一的行为变更； +* 演示/测试账套（3/4/5/6）**不开配置**（`CONFIG_ENTITIES = (1, 2)`，清单 §6.4 + ASK-05 口径②）； + 读路径仍按三层解析返回生效值（模板/内置默认），只是写端点拒绝。 + +配置值一律以 JSON 字符串落库（`param_value`），`value_type` 记录类型以便前端/审计呈现。 +""" +from __future__ import annotations + +import hashlib +import json +import logging +from typing import Any, Dict, List, Optional, Tuple + +from sqlalchemy import inspect +from sqlalchemy.orm import Session + +logger = logging.getLogger("cma.dq.rule_config") + +#: 表名（DDL 走部署链特权迁移步 `backend/scripts/db_migrate.py --apply`，属生产写审批范围） +TABLE_NAME = "quality_rule_config" +#: 全账套模板行的 entity_id（0 = 模板，不参与具体账套取数） +TEMPLATE_ENTITY_ID = 0 +#: 开放配置写入的真实账套（口径②：演示/测试账套不开配置） +CONFIG_ENTITIES: Tuple[int, ...] = (1, 2) +#: 内置默认明细展示条数（清单 §3.9 全局项 `detail_limit` 的出厂值） +DEFAULT_DETAIL_LIMIT = 10 +#: 写配置必须带变更理由，最短长度（清单 §5.1「反刷分约束」第 1 条） +MIN_REASON_LEN = 10 + +RULE_KEYS: Tuple[str, ...] = ( + "unit_check", "dup_alert", "orphan_check", "virtual_pollution", + "entity_check", "kpi_completeness", "reconciliation", +) + +# 全局项（清单 §3.9）：platform 级，仅 it 可写 +GLOBAL_RULE_KEY = "global" + +# 规则级参数统一三件套（每条规则都有；`detail_limit` 为 platform 级） +_CAPS = { + "deduct_cap": dict(type="int", default=15, open_level="entity", + bounds=(0, 50), overflow="clamp", + desc="本规则扣分上限（0 = 只报不计分）"), + "deduct_factor": dict(type="int", default=3, open_level="entity", + bounds=(0, 10), overflow="clamp", + desc="本规则每命中一条的扣分系数"), +} +_WARN_CAPS = { + "deduct_cap": dict(type="int", default=10, open_level="entity", + bounds=(0, 50), overflow="clamp", + desc="本规则扣分上限（0 = 只报不计分）"), + "deduct_factor": dict(type="int", default=1, open_level="entity", + bounds=(0, 10), overflow="clamp", + desc="本规则每命中一条的扣分系数"), +} +_DETAIL_LIMIT = dict(type="int", default=DEFAULT_DETAIL_LIMIT, open_level="platform", + bounds=(1, 100), overflow="clamp", + desc="本规则明细展示条数（平台级，仅 it 可写）") +_ENABLED = dict(type="bool", default=True, open_level="entity", + desc="启用开关（关闭后该规则不参与评分，明细不再展示）") + +#: 7 条规则的开放参数（37 项；清单 §3.1–§3.7 逐条落成） +SPEC: Dict[str, Dict[str, dict]] = { + "unit_check": { + "enabled": dict(_ENABLED), + "amount_threshold": dict(type="int", default=10000, open_level="entity", + bounds=(100, 1000000), overflow="reject", + desc="金额阈值（万元）：cash_plans.amount > 阈值 即命中（越界拒绝）"), + **_CAPS, + "detail_limit": dict(_DETAIL_LIMIT), + }, + "dup_alert": { + "enabled": dict(_ENABLED), + "min_count": dict(type="int", default=2, open_level="entity", + bounds=(2, 20), overflow="reject", + desc="同一 plan_id 的 pending 预警条数下限（≥ 该值即重复；越界拒绝）"), + **_CAPS, + "detail_limit": dict(_DETAIL_LIMIT), + }, + "orphan_check": { + "enabled": dict(_ENABLED), + **_CAPS, + "detail_limit": dict(_DETAIL_LIMIT), + }, + "virtual_pollution": { + "enabled": dict(_ENABLED), + "suspect_keywords": dict( + type="json", default=["test", "虚拟", "demo"], open_level="entity", + max_items=20, item_max_len=32, forbidden_chars="%_", + desc="嫌疑关键词表（大小写不敏感包含式匹配；不含 sync —— 本单唯一行为变更）"), + "source_whitelist": dict( + type="json", + default=["zxt-sync", "bohai_ar", "bohai_ar_20260630", "bohai_prepay_20260630", + "bohai_ap_20260630", "receivables_migrate"], + open_level="entity", max_items=20, item_max_len=32, forbidden_chars="%_", + desc="来源白名单（豁免优先于关键词：命中即不算污染）"), + "manual_source_value": dict(type="str", default="manual", open_level="entity", + max_len=50, + desc="「待人工确认」的来源值（非污染，按 warning 附加扣分）"), + **_CAPS, + "manual_cap": dict(type="int", default=10, open_level="entity", + bounds=(0, 30), overflow="clamp", + desc="待人工确认（manual）附加上限"), + "manual_factor": dict(type="int", default=1, open_level="entity", + bounds=(0, 5), overflow="clamp", + desc="待人工确认（manual）附加系数"), + "detail_limit": dict(_DETAIL_LIMIT), + }, + "entity_check": { + "enabled": dict(_ENABLED), + **_CAPS, + "detail_limit": dict(_DETAIL_LIMIT), + }, + "kpi_completeness": { + "enabled": dict(_ENABLED), + **_WARN_CAPS, + "detail_limit": dict(_DETAIL_LIMIT), + }, + "reconciliation": { + "enabled": dict(_ENABLED), + "diff_threshold_pct": dict(type="float", default=20, open_level="entity", + bounds=(1, 200), overflow="reject", + desc="预算月度合计 vs 年度目标差异阈值（%）；越界拒绝"), + "compare_mode": dict(type="enum", default="abs", open_level="entity", + enum=("abs", "upper_only", "lower_only"), + desc="比较方向：abs=双侧 / upper_only=仅超目标 / lower_only=仅低于目标"), + **_WARN_CAPS, + "detail_limit": dict(_DETAIL_LIMIT), + }, +} + +#: 全局项（2 项；清单 §3.9） +GLOBAL_SPEC: Dict[str, dict] = { + "detail_limit": dict(_DETAIL_LIMIT), + "scope.excluded_entities": dict( + type="json", default=[], open_level="platform", max_items=20, item_max_len=6, + desc="排除出「经营口径」的账套（演示/测试；默认空集 = 不排除）。" + "⚠️ 本项当前**只做标注不改评分语义**（F6 属业务口径，待任总/财务裁定后接线）"), +} + +# ── 锁死面（清单 §4；仅用于给出可读文案，非白名单）───────────────────────── +LOCKED_ITEMS: Tuple[dict, ...] = ( + dict(key="expression", item="判定表达式本体（字段/算子/SQL 片段/正则）", + reason="表达式可配 = 治理规则可被一次配置改成永久绿灯"), + dict(key="level", item="级别归属（error/warning）", + reason="级别直接决定扣分上限，降级是最廉价的绕分路径；确有差异请用已开放的扣分权重表达"), + dict(key="score_fn", item="评分函数（max(0,100−Σ)、7 条固定、各规则独立封顶后累加、无总封顶）", + reason="评分是跨规则/跨期可比的口径基线"), + dict(key="entity_scope", item="实体作用域语义（规则 1 用 cash_plans.entity_id；2/3/6/7 用 kpi_definitions.entity_id；5 用 kpi_values.entity_id）", + reason="作用域 = 安全边界，开放会引入跨账套借用通路"), + dict(key="rule_set", item="规则 key 集合与规则数量（固定 7 条）", + reason="规则集是治理骨架，新增规则必须走代码评审"), + dict(key="match_mode", item="匹配方式自由化（关键词升级为正则/SQL 片段）", + reason="注入面 + 不可验证；匹配方式固定为大小写不敏感包含式"), + dict(key="completeness_caliber", item="kpi_completeness 口径源 kpi_definitions.status='active'（含豁免名单 exempt_kpi_codes）", + reason="KPI 存废开关已是 status，再加豁免名单 = 双开关口径分裂"), + dict(key="null_entity_policy", item="规则5 kpi_values.entity_id IS NULL 行的处置", + reason="NULL 语义未裁定前保持 ignore"), + dict(key="bot_write", item="Bot 通道（X-BOT-KEY）写配置", + reason="Bot 通道无 users 行 → 无操作人可留痕 = 审计断链"), + dict(key="weight_range", item="负系数 / 超大上限 / 非整数权重", + reason="负系数会变成加分（可刷分）；一律钳制到开放区间"), +) +#: 键名 → 锁死文案（写端点遇到这些 param_key 时给「锁死」提示而非笼统的未知参数） +LOCKED_PARAM_KEYS = {it["key"]: it for it in LOCKED_ITEMS} + +# 常见别名（历史/前端习惯叫法）→ 锁死项 key +_LOCKED_ALIASES = { + "expr": "expression", "predicate": "expression", "sql": "expression", + "rule_level": "level", "severity": "level", + "score": "score_fn", "score_func": "score_fn", "total_deduct_cap": "score_fn", + "exempt_kpi_codes": "completeness_caliber", "status_source": "completeness_caliber", + "entity_id_source": "entity_scope", "scope_field": "entity_scope", + "rules": "rule_set", "rule_keys": "rule_set", "rule_count": "rule_set", + "match_regex": "match_mode", "regex": "match_mode", "pattern": "match_mode", + "null_policy": "null_entity_policy", +} + +OPEN_LEVEL_ENTITY = "entity" +OPEN_LEVEL_PLATFORM = "platform" + + +# ══════════════════════════════════════════════════════════════════ +# 参数查找 / 校验 +# ══════════════════════════════════════════════════════════════════ +def is_global_param(rule_key: str) -> bool: + return rule_key == GLOBAL_RULE_KEY + + +def spec_of(rule_key: str) -> Dict[str, dict]: + """返回某 rule_key 的参数规格表（全局项用 rule_key='global'）。""" + if is_global_param(rule_key): + return GLOBAL_SPEC + return SPEC.get(rule_key) or {} + + +def param_spec(rule_key: str, param_key: str) -> Optional[dict]: + return spec_of(rule_key).get(param_key) + + +def all_param_pairs() -> List[Tuple[str, str]]: + """(rule_key, param_key) 全量清单——37 规则级 + 2 全局项。""" + pairs = [(r, p) for r in RULE_KEYS for p in SPEC[r]] + pairs += [(GLOBAL_RULE_KEY, p) for p in GLOBAL_SPEC] + return pairs + + +def count_rule_params() -> int: + """规则级开放参数项数（断言用，清单 §10.2 = 37）。""" + return sum(len(SPEC[r]) for r in RULE_KEYS) + + +def open_level_of(rule_key: str, param_key: str) -> str: + spec = param_spec(rule_key, param_key) or {} + return spec.get("open_level") or OPEN_LEVEL_ENTITY + + +def value_type_of(rule_key: str, param_key: str) -> str: + spec = param_spec(rule_key, param_key) or {} + return spec.get("type") or "json" + + +def dumps_value(value: Any) -> str: + """统一序列化（落库格式；bool/int/float/str/数组一律 JSON）。""" + return json.dumps(value, ensure_ascii=False) + + +def loads_value(raw: Any, value_type: str = "") -> Any: + """反序列化；非 JSON 的历史/手写行容错（返回 None 表示不可用 → 上层回落）。""" + if raw is None: + return None + if isinstance(raw, (int, float, bool, list, dict)): + return raw + try: + return json.loads(str(raw)) + except (ValueError, TypeError): + text = str(raw) + if value_type == "int": + try: + return int(float(text)) + except (ValueError, TypeError): + return None + if value_type == "float": + try: + return float(text) + except (ValueError, TypeError): + return None + if value_type == "bool": + if text.strip().lower() in ("1", "true", "yes"): + return True + if text.strip().lower() in ("0", "false", "no"): + return False + return None + if value_type == "str": + return text + return None + + +def _coerce_type(value: Any, spec: dict) -> Tuple[bool, Any]: + """类型强转（拒绝型：类型不符 → 不可用）。bool 与 int 严格区分（Python 里 bool 是 int 子类）。""" + t = spec.get("type") or "json" + if t == "bool": + if isinstance(value, bool): + return True, value + if isinstance(value, str) and value.strip().lower() in ("true", "false"): + return True, value.strip().lower() == "true" + if isinstance(value, int) and value in (0, 1): + return True, bool(value) + return False, None + if t == "int": + if isinstance(value, bool): + return False, None + if isinstance(value, int): + return True, value + if isinstance(value, float) and float(value).is_integer(): + return True, int(value) + if isinstance(value, str): + s = value.strip() + try: + return True, int(s) + except ValueError: + return False, None + return False, None + if t == "float": + if isinstance(value, bool): + return False, None + if isinstance(value, (int, float)): + return True, float(value) + if isinstance(value, str): + try: + return True, float(value.strip()) + except ValueError: + return False, None + return False, None + if t == "str": + if not isinstance(value, str): + return False, None + return True, value + if t == "enum": + if not isinstance(value, str): + return False, None + return True, value.strip() + if t == "json": + if not isinstance(value, list): + return False, None + return True, value + return False, None + + +def validate_param(rule_key: str, param_key: str, value: Any) -> dict: + """参数级校验（清单 §3 的「取值边界 / 越界处理」逐条落地）。 + + 返回 ``{"ok": bool, "value": <规范化值>, "error": str|None, "warning": str|None}``； + 越界处理两类：``reject``（400，附区间文案）/ ``clamp``（钳制到端点 + 响应告警）。 + """ + spec = param_spec(rule_key, param_key) + if spec is None: + locked = LOCKED_PARAM_KEYS.get( + param_key, LOCKED_PARAM_KEYS.get(_LOCKED_ALIASES.get(param_key, "")) + ) + if locked: + return dict(ok=False, value=None, + error=f"参数 {param_key} 属锁死项「{locked['item']}」，不允许配置：{locked['reason']}", + warning=None) + return dict(ok=False, value=None, + error=f"未知参数 rule={rule_key} param={param_key}；" + f"可选: {', '.join(spec_of(rule_key).keys()) or '（该规则无开放参数）'}", + warning=None) + + passed, value = _coerce_type(value, spec) + if not passed: + return dict(ok=False, value=None, + error=f"{rule_key}.{param_key} 类型非法：要求 {spec.get('type')}", warning=None) + + caution = None + if spec.get("type") == "enum": + if value not in tuple(spec.get("enum") or ()): + return dict(ok=False, value=None, + error=f"{rule_key}.{param_key} 取值非法：仅支持 {', '.join(spec['enum'])}", + warning=None) + return dict(ok=True, value=value, error=None, warning=None) + + if spec.get("type") == "json": + max_items = int(spec.get("max_items") or 0) + item_max_len = int(spec.get("item_max_len") or 0) + forbidden = spec.get("forbidden_chars") or "" + if max_items and len(value) > max_items: + return dict(ok=False, value=None, + error=f"{rule_key}.{param_key} 元素过多：上限 {max_items} 项（当前 {len(value)}）", + warning=None) + cleaned: List[str] = [] + for it in value: + if not isinstance(it, str): + if spec.get("item_max_len") == 6 and isinstance(it, int) and not isinstance(it, bool): + it = str(it) # scope.excluded_entities：账套号允许传 int + else: + return dict(ok=False, value=None, + error=f"{rule_key}.{param_key} 仅允许字符串数组元素", warning=None) + s = it.strip() + if not s: + return dict(ok=False, value=None, + error=f"{rule_key}.{param_key} 不允许空串元素", warning=None) + if item_max_len and len(s) > item_max_len: + return dict(ok=False, value=None, + error=f"{rule_key}.{param_key} 元素过长：单元素上限 {item_max_len} 字符（{s}）", + warning=None) + if any(ch in s for ch in forbidden): + return dict(ok=False, value=None, + error=f"{rule_key}.{param_key} 元素含非法字符 {forbidden}（不支持通配符/自由模式）", + warning=None) + cleaned.append(s) + # 去重保序（重复项无意义且会让 LIKE 子句膨胀） + seen, uniq = set(), [] + for s in cleaned: + if s not in seen: + seen.add(s) + uniq.append(s) + if len(uniq) != len(cleaned): + caution = f"{rule_key}.{param_key} 已去重：{len(cleaned)} → {len(uniq)} 项" + return dict(ok=True, value=uniq, error=None, warning=caution) + + if spec.get("type") == "str": + max_len = int(spec.get("max_len") or 0) + s = value + if not s.strip(): + return dict(ok=False, value=None, error=f"{rule_key}.{param_key} 不允许空串", warning=None) + if max_len and len(s) > max_len: + return dict(ok=False, value=None, + error=f"{rule_key}.{param_key} 过长：上限 {max_len} 字符", warning=None) + return dict(ok=True, value=s, error=None, warning=None) + + bounds = spec.get("bounds") + if bounds: + lo, hi = bounds + if value < lo or value > hi: + if spec.get("overflow") == "clamp": + clamped = min(max(value, lo), hi) + logger.warning("DQ_RULE_CONFIG_CLAMPED %s.%s %s → %s（区间 [%s, %s]）", + rule_key, param_key, value, clamped, lo, hi) + return dict(ok=True, value=clamped, error=None, + warning=f"{rule_key}.{param_key} 越界已钳制到区间端点：{value} → {clamped}" + f"（允许区间 [{lo}, {hi}]）") + return dict(ok=False, value=None, + error=f"{rule_key}.{param_key} 越界：允许区间 [{lo}, {hi}]，收到 {value}", + warning=None) + return dict(ok=True, value=value, error=None, warning=None) + + +# ══════════════════════════════════════════════════════════════════ +# 读路径：三层解析 +# ══════════════════════════════════════════════════════════════════ +def table_ready(db: Session) -> bool: + """配置表是否就绪（DDL 未落地时**不得**让读路径 500：回落内置默认）。""" + try: + return bool(inspect(db.get_bind()).has_table(TABLE_NAME)) + except Exception as exc: # noqa: BLE001 + logger.warning("DQ_RULE_CONFIG_TABLE_PROBE_FAILED %s: %s", TABLE_NAME, exc) + return False + + +def _load_rows(db: Session, entity_id: int) -> Dict[Tuple[str, str], dict]: + """读模板行 + 本账套行（一次查询）；返回 {(rule_key, param_key): {entity_id, value, raw, row}}。""" + from app.models import QualityRuleConfig + + eids = [TEMPLATE_ENTITY_ID] + if entity_id != TEMPLATE_ENTITY_ID: + eids.append(entity_id) + rows = db.query(QualityRuleConfig).filter(QualityRuleConfig.entity_id.in_(eids)).all() + out: Dict[Tuple[str, str], dict] = {} + for row in rows: + vt = value_type_of(row.rule_key, row.param_key) + val = loads_value(row.param_value, vt) + if val is None: + logger.warning("DQ_RULE_CONFIG_ROW_UNPARSABLE id=%s %s.%s value=%r → 忽略该行（回落下一层）", + row.id, row.rule_key, row.param_key, row.param_value) + continue + out[(row.rule_key, row.param_key)] = { + "entity_id": row.entity_id, "value": val, "row": row, + } + return out + + +def resolve(db: Session, entity_id: int) -> dict: + """三层解析当前生效配置（账套行 → 模板行 → 内置默认；**不跨账套回退**）。 + + 返回结构（前端/审计/引擎共用）:: + + {"entity_id": 1, "config_version": "…", "table_ready": True, + "rules": {rule_key: {param_key: {"value":…, "source": "entity|template|builtin", + "open_level":…, "desc":…, "updated_by":…, "updated_at":…}}}, + "global": {...同 rules 的形状…}, + "values": {rule_key: {param_key: <裸值>}}, # 引擎侧取值口 + "scope": {"excluded_entities": [...]}} + """ + eid = int(entity_id or 0) + ready = table_ready(db) + rows = _load_rows(db, eid) if ready else {} + return _assemble(eid, ready, rows) + + +def builtin_resolution(entity_id: int = 0) -> dict: + """纯内置默认的解析结果（配置表缺失/读异常时的回落形态，字段与 `resolve()` 逐字一致）。""" + return _assemble(int(entity_id or 0), False, {}) + + +def _assemble(eid: int, ready: bool, rows: Dict[Tuple[str, str], dict]) -> dict: + """把「行 + 内置默认」组装成对外结构（resolve / builtin_resolution 共用）。""" + + def _one(rule_key: str, param_key: str, spec: dict) -> dict: + val, source, row = spec.get("default"), "builtin", None + tmpl = rows.get((rule_key, param_key)) + if tmpl and tmpl["entity_id"] == TEMPLATE_ENTITY_ID: + val, source, row = tmpl["value"], "template", tmpl["row"] + elif eid != TEMPLATE_ENTITY_ID: + own = rows.get((rule_key, param_key)) + if own and own["entity_id"] == eid: + val, source, row = own["value"], "entity", own["row"] + entry = { + "value": val, "source": source, "open_level": spec.get("open_level"), + "desc": spec.get("desc"), "updated_by": None, "updated_at": None, + } + if row is not None: + entry["updated_by"] = getattr(row, "updated_by", None) + ua = getattr(row, "updated_at", None) + entry["updated_at"] = ua.isoformat(timespec="seconds") if ua else None + entry["row_id"] = row.id + return entry + + rules: Dict[str, Dict[str, dict]] = {} + values: Dict[str, Dict[str, Any]] = {} + for rule_key in RULE_KEYS: + rules[rule_key] = {p: _one(rule_key, p, s) for p, s in SPEC[rule_key].items()} + values[rule_key] = {p: e["value"] for p, e in rules[rule_key].items()} + glob: Dict[str, dict] = {p: _one(GLOBAL_RULE_KEY, p, s) for p, s in GLOBAL_SPEC.items()} + values[GLOBAL_RULE_KEY] = {p: e["value"] for p, e in glob.items()} + + # 规则级 detail_limit 未配置时，回落全局 detail_limit（清单 §3.9） + global_limit = values[GLOBAL_RULE_KEY].get("detail_limit") or DEFAULT_DETAIL_LIMIT + for rule_key in RULE_KEYS: + entry = rules[rule_key]["detail_limit"] + if entry["source"] == "builtin": + entry["value"] = global_limit + entry["source"] = "builtin(global)" + values[rule_key]["detail_limit"] = entry["value"] + + version = hashlib.sha1( + json.dumps(values, sort_keys=True, ensure_ascii=False).encode("utf-8") + ).hexdigest()[:12] + return { + "entity_id": eid, + "config_entity_id": eid if eid in CONFIG_ENTITIES else TEMPLATE_ENTITY_ID, + "entity_config_enabled": eid in CONFIG_ENTITIES, + "table_ready": ready, + "config_version": version, + "rules": rules, + "global": glob, + "values": values, + "scope": {"excluded_entities": values[GLOBAL_RULE_KEY].get("scope.excluded_entities") or []}, + } + + +def engine_values(db: Session, entity_id: int) -> dict: + """引擎取值口（data_quality 规则引擎唯一的配置读取入口）。""" + return resolve(db, entity_id)["values"] + + +# ══════════════════════════════════════════════════════════════════ +# 写路径：单参数 upsert（逐项「一次只改一档」） +# ══════════════════════════════════════════════════════════════════ +def upsert_param(db: Session, entity_id: int, rule_key: str, param_key: str, value: Any, + *, username: Optional[str] = None) -> dict: + """写一行配置（调用方负责权限/校验/审计/事务边界）；返回 {id, before, after, source_before}。""" + from app.models import QualityRuleConfig + + before = resolve(db, entity_id) + scope = before.get("global") if is_global_param(rule_key) else before.get("rules") + before_entry = (scope or {}).get(rule_key, {}).get(param_key, {}) or {} + value_type = value_type_of(rule_key, param_key) + payload = dumps_value(value) + + row = (db.query(QualityRuleConfig) + .filter(QualityRuleConfig.entity_id == int(entity_id), + QualityRuleConfig.rule_key == rule_key, + QualityRuleConfig.param_key == param_key) + .first()) + if row is None: + row = QualityRuleConfig( + entity_id=int(entity_id), rule_key=rule_key, param_key=param_key, + param_value=payload, value_type=value_type, + open_level=open_level_of(rule_key, param_key), is_default=False, + description=(param_spec(rule_key, param_key) or {}).get("desc"), + updated_by=username, + ) + db.add(row) + else: + row.param_value = payload + row.value_type = value_type + row.is_default = False + row.updated_by = username + db.flush() # 拿到 id，且撞唯一键时当场暴露 + return { + "id": row.id, + "before": before_entry.get("value"), + "before_source": before_entry.get("source"), + "after": value, + } + + +def history(db: Session, entity_id: int, limit: int = 20) -> List[dict]: + """本账套的配置变更历史（读 `operation_logs`；该表无 entity_id 列 → 按 detail JSON 过滤）。""" + from app.models import OperationLog + + rows = (db.query(OperationLog) + .filter(OperationLog.action == ACTION_UPDATE) + .order_by(OperationLog.id.desc()) + .limit(max(limit * 5, 50)) + .all()) + out: List[dict] = [] + for r in rows: + detail = r.detail if isinstance(r.detail, dict) else {} + if detail.get("entity_id") != int(entity_id): + continue + out.append({ + "id": r.id, "at": r.created_at.isoformat(timespec="seconds") if r.created_at else None, + "user_id": r.user_id, + "rule_key": detail.get("rule_key"), "param_key": detail.get("param_key"), + "before": detail.get("before"), "after": detail.get("after"), + "reason": detail.get("reason"), "target_id": r.target_id, + }) + if len(out) >= limit: + break + return out + + +#: 审计动作名（清单 §5.2 逐字） +ACTION_UPDATE = "update_dq_rule_config" +TARGET_TYPE = "dq_rule_config" diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 837b91b4..89cda4ec 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -181,6 +181,36 @@ class OperationLog(Base): created_at = Column(DateTime, server_default=func.now()) +class QualityRuleConfig(Base): + """DAMA 数据质量治理规则可配置项（财务七规则）— ASK-20260928-05 + + 逐 param 一行；`entity_id=0` = **全账套模板行**（不参与具体账套取数）。 + 读取三层：账套行 → 模板行(0) → 代码内置默认（禁止跨账套回退）； + 规格/默认值/边界/校验的**单一来源** = `app/dq_rule_config.py`（本模型只存行）。 + DDL 走部署链特权迁移步 `backend/scripts/db_migrate.py --apply`（生产写审批范围）。 + """ + __tablename__ = "quality_rule_config" + id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, nullable=False, default=0, index=True, + comment="账套；0=全账套默认模板") + rule_key = Column(String(50), nullable=False, comment="七规则 key 或 global（全局项）") + param_key = Column(String(50), nullable=False, comment="参数名（见 dq_rule_config.SPEC）") + param_value = Column(String(500), nullable=False, comment="值（JSON 序列化字符串）") + value_type = Column(String(20), nullable=False, default="int", + comment="int/float/bool/str/json/enum") + open_level = Column(String(10), nullable=False, default="entity", + comment="entity=账套级(ceo/it) / platform=平台级(仅it)") + is_default = Column(Boolean, nullable=False, default=False, + comment="1=出厂默认行（用于「恢复默认」与审计对照）") + description = Column(String(500), nullable=True, comment="参数说明（面向使用者）") + updated_by = Column(String(50), nullable=True, comment="最后修改人 users.username") + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) + created_at = Column(DateTime, server_default=func.now()) + __table_args__ = ( + UniqueConstraint("entity_id", "rule_key", "param_key", name="uq_rule_param"), + ) + + class NotificationChannel(Base): """通知渠道配置""" __tablename__ = "notification_channels" diff --git a/backend/scripts/seed_dq_rule_config.py b/backend/scripts/seed_dq_rule_config.py new file mode 100644 index 00000000..df6f5d6d --- /dev/null +++ b/backend/scripts/seed_dq_rule_config.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""`quality_rule_config` 出厂默认值落库（全账套模板行 entity_id=0，is_default=1） + +任务：ASK-20260928-05（CMA 财务七规则可配置化 · 最小实现），清单 §8「初始化」条目。 + +**本脚本是可选的**：读路径三层解析的最后一层就是**代码内置默认** +（`app/dq_rule_config.py` 的 SPEC/GLOBAL_SPEC，与清单 §7 建议默认值逐项一致）， +因此**不落库也能上线且行为相同**（零行为差异）。落库的价值只有两个： + + 1. 让「全账套模板」显式可见（UI 的「来源」列由 `内置默认` 变成 `全账套模板`）； + 2. 给「恢复出厂默认 / 当前值 vs 默认值 diff」留一行可审计的锚（`is_default=1`）。 + +幂等契约 +-------- + 首跑 --apply：created = 39（缺失项），skipped = 0 + 二跑 --apply：created = 0，skipped = 39，零副作用（**不覆盖**已存在的行） + 已存在且值与默认**不同** → 只报告 diff（`differs`），**不覆盖**（`--force` 才覆盖） + +边界 +---- +* 生产写：需要在**审批门**放行后执行（`approval-gate`）；不改表结构（DDL 由 + `backend/scripts/db_migrate.py --apply` 的部署链特权步负责）。 +* 连接：与 db_migrate 同口径 —— `--db-url` > `CMA_MIGRATE_DB_USER/PASS`（TCP）> 本机 root socket； + 显式拒绝应用运行账号（`cma_api`）。 +* 回滚：`--rollback latest` 只删本次 --apply 记录在 journal 里的行 id（零 `DELETE` 全表）。 + +用法 +---- + python3 scripts/seed_dq_rule_config.py --check # 只读预检（默认动作） + python3 scripts/seed_dq_rule_config.py --apply # 落默认行 + 写 journal + python3 scripts/seed_dq_rule_config.py --apply --force # 已存在且不同的行也覆盖（慎用） + python3 scripts/seed_dq_rule_config.py --rollback latest +""" +import argparse +import json +import os +import socket +import sys +from datetime import datetime + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from sqlalchemy import create_engine, text # noqa: E402 + +HERE = os.path.dirname(os.path.abspath(__file__)) +DEFAULT_JOURNAL_DIR = os.path.join(HERE, "backups", "dq-rule-config-seed-journal") +DEFAULT_SOCKET = "/var/run/mysqld/mysqld.sock" +DEFAULT_DB = "cma" +FORBIDDEN_RUN_ACCOUNTS = {"cma_api"} +TABLE = "quality_rule_config" + + +def _db_name(args) -> str: + return (args.db_name or os.getenv("CMA_MIGRATE_DB_NAME") + or os.getenv("CMA_DB_NAME") or DEFAULT_DB) + + +def build_engine(args): + db_name = _db_name(args) + if args.db_url: + url = args.db_url if "/" in args.db_url.rsplit("@", 1)[-1] else args.db_url.rstrip("/") + "/" + db_name + return create_engine(url, pool_pre_ping=True), f"db-url({url})" + user = os.getenv("CMA_MIGRATE_DB_USER", "").strip() + if user: + if user in FORBIDDEN_RUN_ACCOUNTS: + raise SystemExit(f"[FAIL] {user} 是应用运行账号，不得用于本脚本；请用专用账号或 root socket。") + pwd = os.getenv("CMA_MIGRATE_DB_PASS", "") + host = os.getenv("CMA_MIGRATE_DB_HOST", "127.0.0.1") + port = os.getenv("CMA_MIGRATE_DB_PORT", "3306") + url = f"mysql+pymysql://{user}:{pwd}@{host}:{port}/{db_name}?charset=utf8mb4" + return create_engine(url, pool_pre_ping=True), f"tcp({user}@{host}:{port}/{db_name})" + sock = args.socket or os.getenv("CMA_MIGRATE_DB_SOCKET") or DEFAULT_SOCKET + if not os.path.exists(sock): + raise SystemExit(f"[FAIL] 未设 CMA_MIGRATE_DB_USER 且 socket 不存在：{sock}") + url = f"mysql+pymysql://root@localhost/{db_name}?unix_socket={sock}&charset=utf8mb4" + return create_engine(url, pool_pre_ping=True), f"socket(root@{sock}/{db_name})" + + +def _defaults(): + """(rule_key, param_key, value, value_type, open_level, desc) 全量默认 —— 单一来源 = app/dq_rule_config。""" + from app import dq_rule_config as dqcfg + + out = [] + for rule in dqcfg.RULE_KEYS: + for param, spec in dqcfg.SPEC[rule].items(): + out.append((rule, param, spec.get("default"), spec.get("type"), + spec.get("open_level"), spec.get("desc"))) + for param, spec in dqcfg.GLOBAL_SPEC.items(): + out.append((dqcfg.GLOBAL_RULE_KEY, param, spec.get("default"), spec.get("type"), + spec.get("open_level"), spec.get("desc"))) + # 规则级 detail_limit 的出厂值 = 全局 detail_limit（内建默认口径） + return out + + +def _journal_dir() -> str: + d = os.getenv("CMA_DQ_SEED_JOURNAL_DIR") or DEFAULT_JOURNAL_DIR + os.makedirs(d, exist_ok=True) + return d + + +def cmd_check(args, engine, conn_desc) -> int: + rows = _defaults() + missing, same, differs = [], [], [] + with engine.connect() as conn: + if not conn.dialect.has_table(conn, TABLE): + print(f"[check] 连接 {conn_desc} → 表 {TABLE} **不存在**：" + f"请先跑 `backend/scripts/db_migrate.py --apply`（D 部署链特权迁移步）。") + return 0 + for rule, param, value, vtype, level, desc in rows: + got = conn.execute(text( + f"SELECT param_value FROM {TABLE} WHERE entity_id=0 AND rule_key=:r AND param_key=:p"), + {"r": rule, "p": param}).scalar() + if got is None: + missing.append((rule, param, value)) + elif json.loads(got) == value: + same.append((rule, param)) + else: + differs.append((rule, param, json.loads(got), value)) + print(f"[check] 连接 {conn_desc}") + print(f"[check] 待插入 {len(missing)} / 已一致 {len(same)} / 值不同 {len(differs)}（共 {len(rows)} 项）") + for rule, param, value in missing[:10]: + print(f" + {rule}.{param} = {value}") + for rule, param, have, want in differs: + print(f" ! {rule}.{param} 库内={have} 默认={want}（不加 --force 不覆盖）") + return 0 + + +def cmd_apply(args, engine, conn_desc) -> int: + rows = _defaults() + created, skipped, differs, failed = [], [], [], [] + with engine.begin() as conn: + if not conn.dialect.has_table(conn, TABLE): + print(f"[FAIL] 表 {TABLE} 不存在 —— 先跑 db_migrate.py --apply") + return 1 + for rule, param, value, vtype, level, desc in rows: + payload = json.dumps(value, ensure_ascii=False) + cur = conn.execute(text( + f"SELECT id, param_value FROM {TABLE} WHERE entity_id=0 AND rule_key=:r AND param_key=:p"), + {"r": rule, "p": param}).fetchone() + if cur is None: + res = conn.execute(text( + f"INSERT INTO {TABLE} (entity_id, rule_key, param_key, param_value, value_type," + f" open_level, is_default, description, updated_by)" + f" VALUES (0, :r, :p, :v, :t, :l, 1, :d, 'seed:ASK-20260928-05')"), + {"r": rule, "p": param, "v": payload, "t": vtype, "l": level, "d": (desc or "")[:500]}) + created.append({"id": int(res.lastrowid), "rule_key": rule, "param_key": param, + "value": value}) + elif json.loads(cur[1]) == value: + skipped.append((rule, param)) + elif args.force: + conn.execute(text(f"UPDATE {TABLE} SET param_value=:v, is_default=1," + f" updated_by='seed:ASK-20260928-05' WHERE id=:i"), + {"v": payload, "i": cur[0]}) + failed.append({"id": int(cur[0]), "rule_key": rule, "param_key": param, + "before": json.loads(cur[1]), "after": value, "forced": True}) + else: + differs.append((rule, param, json.loads(cur[1]), value)) + + payload = { + "action": "seed_dq_rule_config", "status": "applied", + "at": datetime.now().isoformat(timespec="seconds"), + "connection": conn_desc, "db_name": _db_name(args), + "created": created, "created_count": len(created), + "skipped_count": len(skipped), "differs": differs, "forced": failed, + "actor": f"{os.getenv('USER', 'unknown')}@{socket.gethostname()}", + } + # 文件名带微秒：同一秒内连跑两次不会互相覆盖（离线回归实证） + path = os.path.join(_journal_dir(), f"{datetime.now().strftime('%Y%m%dT%H%M%S-%f')}-applied.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh, ensure_ascii=False, indent=2) + print(f"[apply] 连接 {conn_desc}") + print(f"[apply] 新增 {len(created)} / 跳过(已一致) {len(skipped)} / 值不同未覆盖 {len(differs)}" + f" / 强制覆盖 {len(failed)}") + for rule, param, have, want in differs: + print(f" ! {rule}.{param} 库内={have} ≠ 默认={want}（未覆盖）") + print(f"[apply] journal: {path}") + return 0 + + +def cmd_rollback(args, engine, conn_desc) -> int: + d = _journal_dir() + ref = args.rollback + if ref in ("latest", ""): + cands = sorted(f for f in os.listdir(d) if f.endswith("-applied.json")) + if not cands: + raise SystemExit("[FAIL] 没有可回滚的 journal") + path = os.path.join(d, cands[-1]) + else: + path = ref if os.path.exists(ref) else os.path.join(d, os.path.basename(ref)) + payload = json.load(open(path, encoding="utf-8")) + ids = [c["id"] for c in payload.get("created", [])] + print(f"[rollback] journal={os.path.basename(path)} 将删除 {len(ids)} 行（仅本次新增的 id）") + if not ids: + print("[rollback] 无新增行，零动作") + return 0 + with engine.begin() as conn: + res = conn.execute(text(f"DELETE FROM {TABLE} WHERE id IN ({','.join(str(int(i)) for i in ids)})")) + print(f"[rollback] deleted={res.rowcount}（连接 {conn_desc}）") + return 0 + + +def main() -> int: + ap = argparse.ArgumentParser(description="quality_rule_config 出厂默认值落库（ASK-20260928-05）") + g = ap.add_mutually_exclusive_group() + g.add_argument("--check", action="store_true", help="只读预检（默认动作）") + g.add_argument("--apply", action="store_true", help="落默认行 + journal") + g.add_argument("--rollback", metavar="REF", nargs="?", const="latest", + help="回滚最近一次 --apply 新增的行（REF=latest 或 journal 文件名）") + ap.add_argument("--force", action="store_true", help="已存在且值不同时也覆盖（慎用）") + ap.add_argument("--json", action="store_true") + ap.add_argument("--socket", default=None) + ap.add_argument("--db-url", default=None) + ap.add_argument("--db-name", default=None) + args = ap.parse_args() + + engine, desc = build_engine(args) + if args.apply: + return cmd_apply(args, engine, desc) + if args.rollback is not None: + return cmd_rollback(args, engine, desc) + return cmd_check(args, engine, desc) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/tests/test_dq_rule_config_20260928.py b/backend/tests/test_dq_rule_config_20260928.py new file mode 100644 index 00000000..017bc13c --- /dev/null +++ b/backend/tests/test_dq_rule_config_20260928.py @@ -0,0 +1,529 @@ +# -*- coding: utf-8 -*- +"""CMA 财务七规则可配置化（ASK-20260928-05）回归套件 —— 规格 / 三层解析 / 引擎接线 / 端点。 + +任务：任总 2026-09-28 13:38「放行 4 和 5」→ 最小实现 = 新表 `quality_rule_config` + 端点 + 只读 UI， +外加本单唯一的行为变更（去掉 `sync` 污染判定）与「口径用清单默认值」。 + +判据分四组： + A 规格面：37 规则级参数 + 2 全局项；默认值 == 清单 §7；边界「拒绝 / 钳制」两类；锁死项 400。 + B 引擎接线（**真函数** `_run_governance_checks` + 夹具库，非重写逻辑）： + · 默认配置 = 原硬编码口径（零行为差异，除 sync）； + · **sync A/B 三态**：默认(0) / 恢复旧关键词表(命中) / 白名单豁免(0) —— 证明配置真的接线； + · 阈值、条数下限、开关、扣分上限系数、比较方向、明细条数逐项可配； + · `/rule-detail` total 与新口径一致性（两处同源，不漂移）。 + C 端点面：GET 只读四角色可见；PUT 仅 ceo/it（平台级仅 it）；Bot 通道禁写；演示账套拒写； + change_reason 必填；锁死/未知参数 400；越界拒绝或钳制；写行与 `operation_logs` 同事务。 + D 韧性：配置表缺失（DDL 未落地）时读路径回落内置默认、不 500（部署顺序解耦）。 +""" +import json +from datetime import datetime + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app import dq_rule_config as dqcfg +from app.api.data_quality import _run_governance_checks +from app.models import ( + KPIDefinition, KPIValue, KPIAlert, CashPlan, BudgetPlan, QualityRuleConfig, + OperationLog, Entity, +) +from tests.conftest import create_test_user, get_token_for_user, auth_header + +REASON = "回归用例：调整治理规则阈值以验证配置化接线（≥10 字）" + + +# ══════════════════════════════════════════════════════════════════ +# 夹具 +# ══════════════════════════════════════════════════════════════════ +def _mk_kpi(db: Session, **kw) -> KPIDefinition: + defaults = { + "entity_id": 1, "kpi_code": "KPI_X", "kpi_name": "夹具KPI", "status": "active", + "target_value": 100.0, "target_yearly": 100.0, "formula": "x", + "data_source": "test", "data_owner": "财务部", "unit": "元", + } + defaults.update(kw) + k = KPIDefinition(**defaults) + db.add(k) + db.commit() + db.refresh(k) + return k + + +def _mk_plan(db: Session, **kw) -> CashPlan: + defaults = { + "entity_id": 1, "plan_type": "receive", "amount": 10.0, "source": "manual", + "status": "pending", "plan_date": datetime(2026, 8, 1), + } + defaults.update(kw) + p = CashPlan(**defaults) + db.add(p) + db.commit() + db.refresh(p) + return p + + +def _mk_row(db: Session, rule_key: str, param_key: str, value, **kw) -> QualityRuleConfig: + """直接落/更新一行配置（模拟「已存在配置」；端点写路径另有用例）。""" + entity_id = kw.pop("entity_id", 1) + exist = (db.query(QualityRuleConfig) + .filter_by(entity_id=entity_id, rule_key=rule_key, param_key=param_key).first()) + if exist is not None: + exist.param_value = json.dumps(value, ensure_ascii=False) + exist.value_type = dqcfg.value_type_of(rule_key, param_key) + db.commit() + db.refresh(exist) + return exist + row = QualityRuleConfig( + entity_id=entity_id, rule_key=rule_key, param_key=param_key, + param_value=json.dumps(value, ensure_ascii=False), + value_type=dqcfg.value_type_of(rule_key, param_key), + open_level=dqcfg.open_level_of(rule_key, param_key), + is_default=kw.pop("is_default", False), updated_by=kw.pop("updated_by", "fixture"), + **kw, + ) + db.add(row) + db.commit() + db.refresh(row) + return row + + +def _counts(db: Session, eid: int = 0): + """跑一次真函数，返回 (按规则聚合的 issues, 完整结果)。""" + r = _run_governance_checks(db, eid) + return {i["rule"]: i for i in r["issues"]}, r + + +def _vp_fixture(db: Session) -> None: + """规则4 夹具：5 条正常同步来源（酣客形态）+ 2 条测试来源 + 1 条 manual + 1 条博海来源。""" + for i in range(5): + _mk_plan(db, id=100 + i, source="zxt-sync", amount=10.0 + i) + for i in range(2): + _mk_plan(db, id=200 + i, source="test_import", amount=20.0 + i) + _mk_plan(db, id=300, source="manual", amount=30.0) + _mk_plan(db, id=301, source="bohai_ar_20260630", amount=40.0) + + +# ══════════════════════════════════════════════════════════════════ +# A 规格面 +# ══════════════════════════════════════════════════════════════════ +class TestSpecSurface: + def test_param_count_is_37_rule_level_plus_2_global(self): + """清单 §10.2：37 个规则级参数 + 2 个全局项；且 (rule, param) 无重复。""" + assert dqcfg.count_rule_params() == 37 + assert len(dqcfg.GLOBAL_SPEC) == 2 + pairs = dqcfg.all_param_pairs() + assert len(pairs) == 39 + assert len(set(pairs)) == 39 + per_rule = {r: len(dqcfg.SPEC[r]) for r in dqcfg.RULE_KEYS} + assert per_rule == {"unit_check": 5, "dup_alert": 5, "orphan_check": 4, + "virtual_pollution": 9, "entity_check": 4, + "kpi_completeness": 4, "reconciliation": 6} + + def test_defaults_match_inventory_table(self): + """清单 §7 建议默认值逐项（口径「用清单默认值」）；唯一行为变更 = 关键词表无 sync。""" + vp = dqcfg.SPEC["virtual_pollution"] + assert vp["suspect_keywords"]["default"] == ["test", "虚拟", "demo"] + assert "sync" not in vp["suspect_keywords"]["default"] + assert vp["source_whitelist"]["default"] == [ + "zxt-sync", "bohai_ar", "bohai_ar_20260630", "bohai_prepay_20260630", + "bohai_ap_20260630", "receivables_migrate"] + assert vp["manual_source_value"]["default"] == "manual" + assert vp["manual_cap"]["default"] == 10 and vp["manual_factor"]["default"] == 1 + assert dqcfg.SPEC["unit_check"]["amount_threshold"]["default"] == 10000 + assert dqcfg.SPEC["dup_alert"]["min_count"]["default"] == 2 + assert dqcfg.SPEC["reconciliation"]["diff_threshold_pct"]["default"] == 20 + assert dqcfg.SPEC["reconciliation"]["compare_mode"]["default"] == "abs" + for rule in dqcfg.RULE_KEYS: + caps = dqcfg.SPEC[rule] + assert caps["enabled"]["default"] is True + assert caps["detail_limit"]["default"] == 10 + assert caps["deduct_factor"]["default"] == (3 if rule in + ("unit_check", "dup_alert", "orphan_check", "virtual_pollution", "entity_check") else 1) + assert dqcfg.GLOBAL_SPEC["detail_limit"]["default"] == 10 + # F6 排除名单默认空集（不排除）→ 不为本单引入第二个行为变更 + assert dqcfg.GLOBAL_SPEC["scope.excluded_entities"]["default"] == [] + + def test_locked_items_cover_inventory_ten(self): + """清单 §4 十类锁死面完整登记（少一类即视为私自扩面）。""" + assert len(dqcfg.LOCKED_ITEMS) == 10 + keys = {it["key"] for it in dqcfg.LOCKED_ITEMS} + assert {"expression", "level", "score_fn", "entity_scope", "rule_set", + "match_mode", "completeness_caliber", "null_entity_policy", + "bot_write", "weight_range"} == keys + + @pytest.mark.parametrize("rule,param,value,expect_ok", [ + ("unit_check", "amount_threshold", 5, False), # 越界 → 拒绝 + ("unit_check", "amount_threshold", 1000, True), + ("dup_alert", "min_count", 1, False), # 1 无意义 → 拒绝 + ("dup_alert", "min_count", 3, True), + ("reconciliation", "diff_threshold_pct", 500, False), # 越界 → 拒绝 + ("reconciliation", "compare_mode", "both", False), # 非枚举 → 拒绝 + ("unit_check", "enabled", "yes-please", False), # 非布尔 → 拒绝 + ("virtual_pollution", "suspect_keywords", ["test", ""], False), # 空元素 → 拒绝 + ("virtual_pollution", "suspect_keywords", ["te%st"], False), # 通配符 → 拒绝 + ("virtual_pollution", "suspect_keywords", ["a" * 33], False), # 元素超长 → 拒绝 + ("virtual_pollution", "manual_source_value", "", False), # 空串 → 拒绝 + ]) + def test_validate_reject_cases(self, rule, param, value, expect_ok): + r = dqcfg.validate_param(rule, param, value) + assert r["ok"] is expect_ok + if not expect_ok: + assert r["error"] + + def test_validate_clamp_cases(self): + """权重/明细条数越界 → 钳制到端点 + 告警（清单 §3 的「钳制 + 告警」）。""" + r = dqcfg.validate_param("kpi_completeness", "deduct_cap", 999) + assert r["ok"] and r["value"] == 50 and r["warning"] + r = dqcfg.validate_param("unit_check", "deduct_factor", -3) + assert r["ok"] and r["value"] == 0 and r["warning"] + r = dqcfg.validate_param("unit_check", "detail_limit", 0) + assert r["ok"] and r["value"] == 1 and r["warning"] + + def test_locked_and_unknown_params_rejected(self): + for param in ("level", "expression", "score_fn", "regex", "exempt_kpi_codes"): + r = dqcfg.validate_param("virtual_pollution", param, "x") + assert not r["ok"] + assert "锁死" in r["error"] + r = dqcfg.validate_param("unit_check", "no_such_param", 1) + assert not r["ok"] and "未知参数" in r["error"] + + +# ══════════════════════════════════════════════════════════════════ +# B 引擎接线（真函数 + 夹具库） +# ══════════════════════════════════════════════════════════════════ +class TestEngineWiring: + def test_default_zero_behavior_change_except_sync(self, db): + """默认配置 = 原硬编码口径：阈值 10000 / 条数 2 / 差异 20%；sync 来源不再算污染。""" + _vp_fixture(db) + _mk_plan(db, id=400, amount=50000.0, source="bohai_ar") # 单位校验命中（>10000） + k = _mk_kpi(db, kpi_code="K1", target_yearly=100.0) + db.add(KPIValue(kpi_id=k.id, entity_id=1, period="2026-06", actual_value=50.0)) + for m in range(1, 13): + db.add(BudgetPlan(kpi_id=k.id, entity_id=1, period=f"2026-{m:02d}", + budget_value=200.0 / 12, budget_year=2026, budget_month=m, + version="v1.0", status="active")) + _mk_kpi(db, kpi_code="K2", target_yearly=10.0) # 无值 → KPI完整性 + db.commit() + + issues, r = _counts(db, 1) + assert issues["unit_check"]["count"] == 1 + assert issues["virtual_pollution"]["count"] == 2 # 只剩 test_import 两条 + assert issues["virtual_pollution"]["manual_count"] == 1 + assert issues["kpi_completeness"]["count"] == 1 + assert issues["reconciliation"]["count"] == 1 + assert r["config_version"] and len(r["config_version"]) == 12 + assert r["disabled"] == [] + assert r["config_source"]["table_ready"] is True + assert r["scope_excluded_entities"] == [] + assert r["excluded_from_business_caliber"] is False + + def test_sync_ab_three_states(self, db): + """A/B：默认(0 命中) → 恢复旧关键词表+空白名单(7 命中) → 只留白名单(2 命中)。 + + 三态各跑一次**同一个** `_run_governance_checks`（不是重写逻辑）， + 证明「配置真的驱动引擎」+「去掉 sync 确实消除了 100% 假阳性」。 + """ + _vp_fixture(db) + + issues, _ = _counts(db, 1) + assert issues["virtual_pollution"]["count"] == 2, "默认：zxt-sync 不再误判" + + legacy = _mk_row(db, "virtual_pollution", "suspect_keywords", + ["test", "sync", "虚拟", "demo"]) # 改前口径 + _mk_row(db, "virtual_pollution", "source_whitelist", []) # 改前：无白名单 + issues, r = _counts(db, 1) + assert issues["virtual_pollution"]["count"] == 7, "恢复旧口径 → zxt-sync 5 条重新被误判" + # 扣分 = min(15, 7×3) + manual 附加 min(10, 1×1) = 16（夹具里有 1 条 manual） + assert issues["virtual_pollution"]["deducted"] == 16 + + db.query(QualityRuleConfig).filter(QualityRuleConfig.rule_key == "virtual_pollution").delete() + _mk_row(db, "virtual_pollution", "source_whitelist", ["zxt-sync"]) + issues, _ = _counts(db, 1) + assert issues["virtual_pollution"]["count"] == 2, "白名单豁免优先于关键词（豁免即不算污染）" + db.delete(legacy) + db.commit() + + def test_threshold_and_switch_and_caps(self, db): + _mk_plan(db, id=500, amount=150.0) + _mk_plan(db, id=501, amount=50000.0) + + issues, _ = _counts(db, 1) + assert issues["unit_check"]["count"] == 1 # 默认 10000 + + _mk_row(db, "unit_check", "amount_threshold", 100) # 下调阈值 → 2 条 + issues, _ = _counts(db, 1) + assert issues["unit_check"]["count"] == 2 + + _mk_row(db, "unit_check", "deduct_cap", 0) # 只报不计分 + issues, r = _counts(db, 1) + assert issues["unit_check"]["count"] == 2 and issues["unit_check"]["deducted"] == 0 + + _mk_row(db, "unit_check", "enabled", False) # 关开关 → 零命中零扣分 + issues, r = _counts(db, 1) + assert issues["unit_check"]["count"] == 0 and issues["unit_check"]["deducted"] == 0 + assert issues["unit_check"]["enabled"] is False + assert "unit_check" in r["disabled"] and "unit_check" not in r["passed"] + assert r["total_rules"] == 7 # 规则集锁死 7 条 + + def test_dup_alert_min_count(self, db): + k = _mk_kpi(db, kpi_code="K9", target_yearly=10.0) + _mk_plan(db, id=600, source="manual") + for _ in range(2): # plan#600 两条 pending + db.add(KPIAlert(kpi_id=k.id, alert_type="cash_plan", status="pending", + alert_message="x", suggestion=json.dumps({"plan_id": 600}))) + db.commit() + issues, _ = _counts(db, 1) + assert issues["dup_alert"]["count"] == 1 # 默认 ≥2 即重复 + + _mk_row(db, "dup_alert", "min_count", 3) + issues, _ = _counts(db, 1) + assert issues["dup_alert"]["count"] == 0 # 阈值抬到 3 → 不再重复 + + def test_reconciliation_threshold_and_direction(self, db): + k = _mk_kpi(db, kpi_code="K7", target_yearly=100.0) + for m in range(1, 13): + db.add(BudgetPlan(kpi_id=k.id, entity_id=1, period=f"2026-{m:02d}", + budget_value=200.0 / 12, budget_year=2026, budget_month=m, + version="v1.0", status="active")) # 合计 200 → +100% + k2 = _mk_kpi(db, kpi_code="K8", target_yearly=100.0) + for m in range(1, 13): + db.add(BudgetPlan(kpi_id=k2.id, entity_id=1, period=f"2026-{m:02d}", + budget_value=50.0 / 12, budget_year=2026, budget_month=m, + version="v1.0", status="active")) # 合计 50 → -50% + db.commit() + + issues, _ = _counts(db, 1) + assert issues["reconciliation"]["count"] == 2 # 默认 abs 双侧 >20% + + _mk_row(db, "reconciliation", "diff_threshold_pct", 150) + issues, _ = _counts(db, 1) + assert issues["reconciliation"]["count"] == 0 # 阈值抬高 + + _mk_row(db, "reconciliation", "diff_threshold_pct", 20) + _mk_row(db, "reconciliation", "compare_mode", "upper_only") + issues, _ = _counts(db, 1) + assert issues["reconciliation"]["count"] == 1 # 只看「超目标」→ 低于目标的 K8 不算 + + _mk_row(db, "reconciliation", "compare_mode", "lower_only") + issues, _ = _counts(db, 1) + assert issues["reconciliation"]["count"] == 1 # 只看「低于目标」→ K8 + + def test_detail_limit_configurable(self, db): + for i in range(4): + _mk_plan(db, id=700 + i, amount=50000.0) + issues, _ = _counts(db, 1) + assert len([d for d in issues["unit_check"]["detail"] if not d.startswith("…")]) == 4 + + _mk_row(db, "unit_check", "detail_limit", 2) + issues, _ = _counts(db, 1) + rows = [d for d in issues["unit_check"]["detail"] if not d.startswith("…")] + assert len(rows) == 2 and any(d.startswith("…") for d in issues["unit_check"]["detail"]) + + def test_entity_config_isolation_no_cross_entity_fallback(self, db): + """账套隔离：账套 2 的行不得被账套 1 读到（禁止跨账套回退）。""" + _mk_plan(db, id=800, entity_id=1, amount=50000.0) + _mk_plan(db, id=801, entity_id=2, amount=50000.0) + _mk_row(db, "unit_check", "amount_threshold", 1000000, entity_id=2) + + issues1, _ = _counts(db, 1) + issues2, _ = _counts(db, 2) + assert issues1["unit_check"]["count"] == 1 # 账套1 用默认 10000 + assert issues2["unit_check"]["count"] == 0 # 账套2 用自己那行 1000000 + + +# ══════════════════════════════════════════════════════════════════ +# C 端点面 +# ══════════════════════════════════════════════════════════════════ +def _ceo_token(db: Session, client: TestClient) -> str: + create_test_user(db, username="cfgceo", role="ceo") + return get_token_for_user(client, "cfgceo") + + +class TestConfigEndpoints: + def test_get_rule_config_readonly_shape(self, db, client): + token = _ceo_token(db, client) + r = client.get("/api/cma/data-quality/rule-config", headers=auth_header(token)) + assert r.status_code == 200, r.text + body = r.json() + assert body["entity_id"] == 1 + assert body["entity_config_enabled"] is True + assert body["config_entities"] == [1, 2] + assert body["rule_param_count"] == 37 + assert set(body["rules"].keys()) == set(dqcfg.RULE_KEYS) + vp = body["rules"]["virtual_pollution"]["params"] + assert vp["suspect_keywords"]["value"] == ["test", "虚拟", "demo"] + assert vp["suspect_keywords"]["source"] == "builtin" + assert vp["suspect_keywords"]["open_level"] == "entity" + assert body["global_params"]["detail_limit"]["value"] == 10 + assert body["scope"]["excluded_entities"] == [] + assert len(body["locked"]) == 10 + assert body["readonly"] is True + + def test_get_rule_config_finance_readonly_allowed(self, db, client): + create_test_user(db, username="cfgfin", role="finance") + token = get_token_for_user(client, "cfgfin") + assert client.get("/api/cma/data-quality/rule-config", + headers=auth_header(token)).status_code == 200 + + def test_put_requires_ceo_or_it(self, db, client): + create_test_user(db, username="cfgfin2", role="finance") + token = get_token_for_user(client, "cfgfin2") + r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token), + json={"rule_key": "unit_check", "param_key": "amount_threshold", + "value": 5000, "change_reason": REASON}) + assert r.status_code == 403, r.text + + def test_put_writes_row_and_audit_same_txn(self, db, client): + token = _ceo_token(db, client) + before_version = client.get("/api/cma/data-quality/rule-config", + headers=auth_header(token)).json()["config_version"] + r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token), + json={"rule_key": "unit_check", "param_key": "amount_threshold", + "value": 5000, "change_reason": REASON}) + assert r.status_code == 200, r.text + body = r.json() + assert body["success"] and body["before"] == 10000 and body["after"] == 5000 + assert body["warning"] is None + + row = db.query(QualityRuleConfig).filter_by( + entity_id=1, rule_key="unit_check", param_key="amount_threshold").one() + assert json.loads(row.param_value) == 5000 and row.updated_by == "cfgceo" + + log = db.query(OperationLog).filter_by(action="update_dq_rule_config").one() + assert log.user_id is not None # 清单 §5.2：user_id 必写 + assert log.target_type == "dq_rule_config" and log.target_id == row.id + assert log.detail["entity_id"] == 1 and log.detail["before"] == 10000 + assert log.detail["after"] == 5000 and log.detail["reason"] == REASON + + # 生效回读：GET 显示来源=账套覆盖；配置指纹随之变化 + g = client.get("/api/cma/data-quality/rule-config", headers=auth_header(token)).json() + entry = g["rules"]["unit_check"]["params"]["amount_threshold"] + assert entry["value"] == 5000 and entry["source"] == "entity" + assert g["config_version"] == body["config_version"] != before_version + + def test_put_locked_unknown_and_reason_rejected(self, db, client): + token = _ceo_token(db, client) + base = {"rule_key": "unit_check", "param_key": "level", "value": "warning", + "change_reason": REASON} + r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token), json=base) + assert r.status_code == 400 and "锁死" in r.json()["detail"] + + base["param_key"] = "nope" + assert client.put("/api/cma/data-quality/rule-config", headers=auth_header(token), + json=base).status_code == 400 + + r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token), + json={"rule_key": "unit_check", "param_key": "amount_threshold", + "value": 5000, "change_reason": "太短"}) + assert r.status_code == 400 and "change_reason" in r.json()["detail"] + + def test_put_boundary_reject_and_clamp(self, db, client): + token = _ceo_token(db, client) + r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token), + json={"rule_key": "unit_check", "param_key": "amount_threshold", + "value": 5, "change_reason": REASON}) + assert r.status_code == 400 and "越界" in r.json()["detail"] + + r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(token), + json={"rule_key": "unit_check", "param_key": "deduct_cap", + "value": 999, "change_reason": REASON}) + assert r.status_code == 200 and r.json()["after"] == 50 and r.json()["warning"] + + def test_platform_param_requires_it(self, db, client): + ceo = _ceo_token(db, client) + payload = {"rule_key": "unit_check", "param_key": "detail_limit", "value": 25, + "change_reason": REASON} + r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(ceo), json=payload) + assert r.status_code == 403, r.text + assert "平台级" in r.json()["detail"] + + create_test_user(db, username="cma", role="it") + it = get_token_for_user(client, "cma") + r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(it), json=payload) + assert r.status_code == 200 and r.json()["after"] == 25 + + def test_bot_channel_write_forbidden(self, db, client): + token = _ceo_token(db, client) + headers = {**auth_header(token), "X-BOT-KEY": "dummy-bot-key"} + r = client.put("/api/cma/data-quality/rule-config", headers=headers, + json={"rule_key": "unit_check", "param_key": "amount_threshold", + "value": 5000, "change_reason": REASON}) + assert r.status_code == 403 and "Bot" in r.json()["detail"] + + def test_demo_entity_write_rejected(self, db, client): + """口径②：演示/测试账套只读、不开配置（写 400，读仍 200）。""" + db.add(Entity(id=6, name="演示企业", short_name="演示", status="demo")) + db.commit() + u = create_test_user(db, username="cfgdemo", role="ceo") + from app.models import UserEntity + db.add(UserEntity(user_id=u.id, entity_id=1, granted_by=1)) + db.add(UserEntity(user_id=u.id, entity_id=6, granted_by=1)) + db.commit() + + # 登录 entity 6（该账号已显式授权 1/6） + resp = client.post("/api/cma/auth/login", json={ + "username": "cfgdemo", "password": "test-fixture-pass", "entity_id": 6}) + assert resp.status_code == 200, resp.text + tok6 = resp.json()["token"] + + assert client.get("/api/cma/data-quality/rule-config", + headers=auth_header(tok6)).status_code == 200 + assert client.get("/api/cma/data-quality/rule-config", + headers=auth_header(tok6)).json()["entity_config_enabled"] is False + r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(tok6), + json={"rule_key": "unit_check", "param_key": "amount_threshold", + "value": 5000, "change_reason": REASON}) + assert r.status_code == 400 and "不开放规则配置" in r.json()["detail"] + + def test_history_endpoint(self, db, client): + token = _ceo_token(db, client) + client.put("/api/cma/data-quality/rule-config", headers=auth_header(token), + json={"rule_key": "dup_alert", "param_key": "min_count", "value": 3, + "change_reason": REASON}) + r = client.get("/api/cma/data-quality/rule-config/history", headers=auth_header(token)) + assert r.status_code == 200 + items = r.json()["items"] + assert items and items[0]["rule_key"] == "dup_alert" + assert items[0]["before"] == 2 and items[0]["after"] == 3 + assert items[0]["reason"] == REASON and items[0]["user_id"] is not None + + def test_rule_detail_total_matches_governance_check(self, db, client): + """两处口径同源：改了阈值后 `/rule-detail` total == `governance-check` count。""" + _mk_plan(db, entity_id=1, id=900, amount=150.0) + token = _ceo_token(db, client) + client.put("/api/cma/data-quality/rule-config", headers=auth_header(token), + json={"rule_key": "unit_check", "param_key": "amount_threshold", + "value": 100, "change_reason": REASON}) + gov = client.get("/api/cma/data-quality/governance-check", + headers=auth_header(token)).json() + count = {i["rule"]: i["count"] for i in gov["issues"]}["unit_check"] + detail = client.get("/api/cma/data-quality/rule-detail?rule=unit_check", + headers=auth_header(token)).json() + assert count == 1 and detail["total"] == count + assert detail["items"][0]["key_fields"]["threshold"] == ">100（万元口径可疑）" + + +# ══════════════════════════════════════════════════════════════════ +# D 韧性：DDL 未落地时不 500 +# ══════════════════════════════════════════════════════════════════ +class TestTableMissingTolerance: + def test_read_falls_back_to_builtin_when_table_absent(self, db, client): + from tests import conftest as cf + + token = _ceo_token(db, client) + _mk_plan(db, entity_id=1, id=950, amount=50000.0) + QualityRuleConfig.__table__.drop(bind=cf.TEST_ENGINE) + try: + assert dqcfg.table_ready(db) is False + r = client.get("/api/cma/data-quality/rule-config", headers=auth_header(token)) + assert r.status_code == 200, r.text + assert r.json()["table_ready"] is False + assert r.json()["rules"]["unit_check"]["params"]["amount_threshold"]["value"] == 10000 + + gov = client.get("/api/cma/data-quality/governance-check", + headers=auth_header(token)) + assert gov.status_code == 200, gov.text + assert {i["rule"]: i["count"] for i in gov.json()["issues"]}["unit_check"] == 1 + finally: + QualityRuleConfig.__table__.create(bind=cf.TEST_ENGINE) diff --git a/frontend/src/views/DataQuality.vue b/frontend/src/views/DataQuality.vue index 8907f80b..37a79482 100644 --- a/frontend/src/views/DataQuality.vue +++ b/frontend/src/views/DataQuality.vue @@ -335,6 +335,104 @@ + + + +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
+ 暂无变更记录 +
+ +
+ @@ -536,6 +634,88 @@ const govDetailPageSize = 50 // 与后端契约一致（默认 50、上限 200 const govDetailLoading = ref(false) const govExporting = ref(false) +// ── 规则配置（只读）— ASK-20260928-05 ── +const cfgLoaded = ref(false) +const cfgLoading = ref(false) +const cfgData = ref({}) +const cfgHistory = ref([]) + +function cfgFmt(v: any): string { + if (v === null || v === undefined) return '-' + if (Array.isArray(v)) return v.length ? v.join(', ') : '（空）' + if (typeof v === 'boolean') return v ? '是' : '否' + return String(v) +} + +const SOURCE_TEXT: Record = { + entity: '账套覆盖', template: '全账套模板', builtin: '内置默认', 'builtin(global)': '内置默认', +} + +const cfgSummary = computed(() => { + const d: any = cfgData.value + if (!d || !d.entity_id) return '' + const parts = [ + `配置表：${d.table_ready ? '已就绪' : '未建（按内置默认运行）'}`, + `配置指纹：${d.config_version || '-'}`, + `开放参数：${d.rule_param_count ?? 0} 项规则级 + ${Object.keys(d.global_params || {}).length} 项全局`, + `本账套可写：${d.entity_config_enabled ? '是' : '否（演示/测试账套只读）'}`, + ] + return parts.join(' · ') +}) + +const cfgRows = computed(() => { + const d: any = cfgData.value + const out: any[] = [] + if (!d || !d.rules) return out + for (const ruleKey of Object.keys(d.rules)) { + const rule = d.rules[ruleKey] + for (const paramKey of Object.keys(rule.params || {})) { + const p = rule.params[paramKey] + out.push({ + ruleKey, ruleName: rule.name, paramKey, + valueText: cfgFmt(p.value), + sourceText: SOURCE_TEXT[p.source] || p.source, + sourceTag: p.source === 'entity' ? 'warning' : (p.source === 'template' ? 'success' : 'info'), + openLevelText: p.open_level === 'platform' ? '平台级(仅it)' : '账套级(ceo/it)', + updatedText: p.updated_at ? `${(p.updated_at || '').replace('T', ' ')}${p.updated_by ? ' · ' + p.updated_by : ''}` : '-', + desc: p.desc || '', + }) + } + } + for (const paramKey of Object.keys(d.global_params || {})) { + const p = d.global_params[paramKey] + out.push({ + ruleKey: 'global', ruleName: '全局项', paramKey, + valueText: cfgFmt(p.value), + sourceText: SOURCE_TEXT[p.source] || p.source, + sourceTag: p.source === 'entity' ? 'warning' : (p.source === 'template' ? 'success' : 'info'), + openLevelText: p.open_level === 'platform' ? '平台级(仅it)' : '账套级(ceo/it)', + updatedText: p.updated_at ? `${(p.updated_at || '').replace('T', ' ')}${p.updated_by ? ' · ' + p.updated_by : ''}` : '-', + desc: p.desc || '', + }) + } + return out +}) + +const cfgLocked = computed(() => cfgData.value?.locked || []) + +async function loadRuleConfig() { + cfgLoading.value = true + try { + cfgData.value = await api.get('/data-quality/rule-config') + // 变更历史仅 ceo/it 可见：无权限时静默留空，不影响只读视图 + try { + const h: any = await api.get('/data-quality/rule-config/history', { params: { limit: 10 } }) + cfgHistory.value = h.items || [] + } catch (e) { + cfgHistory.value = [] + } + } catch (e: any) { + ElMessage.error(e?.response?.data?.detail || '加载规则配置失败') + } + cfgLoading.value = false +} + // 环图颜色一律用**合法 HEX 实值**（ECharts 走 canvas，不解析 CSS 变量，传 CSS 变量会 // 回退成黑色 —— 2026-09-24 修复）。取值与主题色一致：财务 #F56C6C / 学习成长 #E6A23C。 const CHART_COLORS = { @@ -895,6 +1075,11 @@ function handleTabChange(name: string | number) { govLoaded.value = true runGovernanceCheck() } + // Tab3 懒加载：首次切换到"规则配置"（只读）时才请求 + if (name === 'rulecfg' && !cfgLoaded.value) { + cfgLoaded.value = true + loadRuleConfig() + } } async function loadStats() {