diff --git a/backend/app/api/data_quality.py b/backend/app/api/data_quality.py index cddf5167..1caa80b9 100644 --- a/backend/app/api/data_quality.py +++ b/backend/app/api/data_quality.py @@ -557,6 +557,44 @@ def _entity_clause(eid: int, col: str) -> Tuple[str, dict]: return "", {} +def _excluded_clause(eid: int, col: str, excluded) -> Tuple[str, dict]: + """经营口径排除子句(F6 / ASK-05 ③-2 接线,2026-09-29)。 + + 语义(DAMA 清单 §6.4 + 拍板件 ③-2「排除」): + * **仅全账套口径(`eid=0` = 经营口径评分 / 明细聚合)生效** —— 被排除账套(演示/测试) + 的行不再计入经营口径的评分与播报口径; + * 单账套查询(`eid!=0`)**保持原判据**:演示/测试账套仍可查看自己的检查结果 + (「参与检查展示」),只由响应字段 `excluded_from_business_caliber` 标注不计入。 + * 名单单一来源 = `dq_rule_config` 的 `scope.excluded_entities`(改配置值即调整,零代码回退); + * 列可为空的历史行按「未排除」处理(`COALESCE(..., 1)` → 归默认账套 1), + 避免 SQL 三值逻辑把 NULL 归属行整批吞掉。 + """ + ids = dqcfg.normalize_entity_ids(excluded) + if eid or not ids: + return "", {} + params = {} + keys = [] + for i, v in enumerate(ids): + key = "nxe%d" % i + keys.append(":" + key) + params[key] = int(v) + return " AND COALESCE(%s, 1) NOT IN (%s)" % (col, ", ".join(keys)), params + + +def _business_excluded(db: Session, eid: int) -> list: + """本口径下生效的经营口径排除名单(**仅全账套 eid=0 生效**,其余口径返回空)。 + + 单一来源 = `dq_rule_config`(`scope.excluded_entities`);本模块不得另立一份账套名单。 + """ + if eid: + return [] + try: + return dqcfg.excluded_entity_ids(db, 0) + except Exception: # noqa: BLE001 + logger.warning("DQ_SCOPE_EXCLUDED_RESOLVE_FAILED eid=0 → 按不排除处理", exc_info=True) + return [] + + def _vp_predicate(cfg: dict) -> Tuple[str, dict]: """规则4 污染主体谓词:嫌疑关键词(大小写不敏感包含式)**减去**来源白名单(豁免优先)。 @@ -583,25 +621,42 @@ def _vp_predicate(cfg: dict) -> Tuple[str, dict]: 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 "") - return {row[0] for row in db.execute(text(sql), {"eid": eid}).fetchall()} +def _plan_id_set(db: Session, eid: int, excluded=None): + """已存在的 cash_plans id 集合(规则3 判孤儿用;eid 非 0 时只看该实体) + + 经营口径(eid=0)下叠加 `scope.excluded_entities` 排除子句(F6 接线)。 + """ + sql = "SELECT id FROM cash_plans" + params = {"eid": eid} + if eid: + sql += " WHERE entity_id = :eid" + else: + exc, exc_params = _excluded_clause(eid, "entity_id", excluded) + if exc: + sql += " WHERE 1 = 1" + exc + params.update(exc_params) + return {row[0] for row in db.execute(text(sql), params).fetchall()} -def _pending_alert_plan_links(db: Session, eid: int): +def _pending_alert_plan_links(db: Session, eid: int, excluded=None): """规则2/3 共用数据:pending cash_plan 预警 → [(alert_id, kpi_id, plan_id, kpi_entity_id)] 返回 (parsed, raw_rows);Python 侧解析 suggestion JSON(不依赖 MySQL JSON 函数)。 + 经营口径(eid=0)下叠加 `scope.excluded_entities` 排除子句(F6 接线,按 kpi_definitions.entity_id)。 """ alert_sql = ( "SELECT a.id, a.kpi_id, a.suggestion, d.entity_id AS kpi_entity_id " "FROM kpi_alerts a JOIN kpi_definitions d ON d.id = a.kpi_id " "WHERE a.alert_type = 'cash_plan' AND a.status = 'pending'" ) + params = {"eid": eid} if eid: alert_sql += " AND d.entity_id = :eid" - rows = db.execute(text(alert_sql), {"eid": eid}).fetchall() + else: + exc, exc_params = _excluded_clause(eid, "d.entity_id", excluded) + alert_sql += exc + params.update(exc_params) + rows = db.execute(text(alert_sql), params).fetchall() parsed = [] for r in rows: pid = _extract_plan_id(r.suggestion) @@ -625,48 +680,68 @@ def _dup_groups(parsed, min_count: int = 2): def _detail_count(db: Session, rule: str, eid: int, manual: bool = False, - cfg: Optional[dict] = None) -> int: - """本规则真实总数:一律 COUNT(*)(dup/orphan 由预警分组得出,同样不受 LIMIT 影响)""" + cfg: Optional[dict] = None, excluded=None) -> int: + """本规则真实总数:一律 COUNT(*)(dup/orphan 由预警分组得出,同样不受 LIMIT 影响) + + 经营口径(eid=0)下叠加 `scope.excluded_entities` 排除(F6 接线)—— + 与 `_run_governance_checks` 的聚合口径逐字一致(两处不得漂移)。 + """ cfg = cfg or _cfg_values(db, eid) + if excluded is None: + excluded = _business_excluded(db, eid) if rule == "dup_alert": - parsed, _ = _pending_alert_plan_links(db, eid) + parsed, _ = _pending_alert_plan_links(db, eid, excluded) 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) + parsed, _ = _pending_alert_plan_links(db, eid, excluded) + pset = _plan_id_set(db, eid, excluded) return len([1 for _a, _k, pid, _e in parsed if pid not in pset]) ent, params = _entity_clause(eid, "cp.entity_id") + exc, exc_params = _excluded_clause(eid, "cp.entity_id", excluded) + params.update(exc_params) if rule == "unit_check": - sql = "SELECT COUNT(*) FROM cash_plans cp WHERE cp.amount > :amt" + ent + sql = "SELECT COUNT(*) FROM cash_plans cp WHERE cp.amount > :amt" + ent + exc params["amt"] = _unit_threshold(cfg) elif rule == "virtual_pollution": if manual: - sql = "SELECT COUNT(*) FROM cash_plans cp WHERE cp.source = :msrc" + ent + sql = "SELECT COUNT(*) FROM cash_plans cp WHERE cp.source = :msrc" + ent + exc params["msrc"] = _manual_source(cfg) else: pred, vp_params = _vp_predicate(cfg) params.update(vp_params) - sql = "SELECT COUNT(*) FROM cash_plans cp WHERE " + pred + ent + sql = "SELECT COUNT(*) FROM cash_plans cp WHERE " + pred + ent + exc 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 + else: + exc5, exc5_params = _excluded_clause(eid, "v.entity_id", excluded) + sql += exc5 + params.update(exc5_params) 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 + else: + exc6, exc6_params = _excluded_clause(eid, "d.entity_id", excluded) + sql += exc6 + params.update(exc6_params) elif rule == "reconciliation": sql = ("SELECT COUNT(*) FROM (SELECT d.id, " + RECON_DIFF_EXPR + " AS diff_pct " + RECON_BASE_SQL) if eid: sql += " AND d.entity_id = :eid" params["eid"] = eid + else: + exc7, exc7_params = _excluded_clause(eid, "d.entity_id", excluded) + sql += exc7 + params.update(exc7_params) sql += " GROUP BY d.id HAVING " + _recon_having(cfg) + ") t" params["pct"] = _recon_pct(cfg) else: @@ -675,12 +750,18 @@ def _detail_count(db: Session, rule: str, eid: 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)""" + manual: bool = False, cfg: Optional[dict] = None, excluded=None): + """结构化明细条目(服务端 LIMIT/OFFSET 分页;字段:record_id/entity_id/source/key_fields/summary/suggested_action) + + 经营口径(eid=0)下与 `_detail_count` / `governance-check` 同源叠加 + `scope.excluded_entities` 排除(F6 接线)——三处口径一律不得漂移。 + """ cfg = cfg or _cfg_values(db, eid) + if excluded is None: + excluded = _business_excluded(db, eid) if rule == "dup_alert": - parsed, _ = _pending_alert_plan_links(db, eid) + parsed, _ = _pending_alert_plan_links(db, eid, excluded) items = [] for pid, alerts in _dup_groups(parsed, _dup_min_count(cfg)): items.append({ @@ -695,8 +776,8 @@ def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, return items[offset:offset + limit] if rule == "orphan_check": - parsed, _ = _pending_alert_plan_links(db, eid) - pset = _plan_id_set(db, eid) + parsed, _ = _pending_alert_plan_links(db, eid, excluded) + pset = _plan_id_set(db, eid, excluded) items = [] for alert_id, kpi_id, pid, ent_id in parsed: if pid in pset: @@ -712,13 +793,15 @@ def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, return items[offset:offset + limit] ent, params = _entity_clause(eid, "cp.entity_id") + exc, exc_params = _excluded_clause(eid, "cp.entity_id", excluded) + params.update(exc_params) 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 > :amt" + ent + " ORDER BY cp.amount DESC LIMIT :lim OFFSET :off" + "WHERE cp.amount > :amt" + ent + exc + " ORDER BY cp.amount DESC LIMIT :lim OFFSET :off" ), {**params, "amt": threshold}).fetchall() return [{ "record_id": r.id, @@ -740,7 +823,7 @@ def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, 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" + "WHERE " + where + ent + exc + " ORDER BY cp.id LIMIT :lim OFFSET :off" ), {**params, **kw_params}).fetchall() return [{ "record_id": r.id, @@ -758,6 +841,10 @@ def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, "WHERE v.entity_id IS NOT NULL AND v.entity_id != d.entity_id") if eid: sql += " AND v.entity_id = :eid" + else: + exc5, exc5_params = _excluded_clause(eid, "v.entity_id", excluded) + sql += exc5 + params.update(exc5_params) sql += " ORDER BY v.id LIMIT :lim OFFSET :off" if not eid: params.pop("eid", None) @@ -777,6 +864,10 @@ def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, "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" + else: + exc6, exc6_params = _excluded_clause(eid, "d.entity_id", excluded) + sql += exc6 + params.update(exc6_params) sql += " ORDER BY d.id LIMIT :lim OFFSET :off" if not eid: params.pop("eid", None) @@ -796,6 +887,10 @@ def _detail_items(db: Session, rule: str, eid: int, limit: int, offset: int, + RECON_DIFF_EXPR + " AS diff_pct " + RECON_BASE_SQL) if eid: sql += " AND d.entity_id = :eid" + else: + exc7, exc7_params = _excluded_clause(eid, "d.entity_id", excluded) + sql += exc7 + params.update(exc7_params) 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)} @@ -853,10 +948,17 @@ def _run_governance_checks(db: Session, entity_id: int = 0): 配置化(ASK-20260928-05):阈值/关键词/条数下限/差异阈值与方向/扣分上限与系数/明细条数 一律经 `_cfg_values()`(三层解析)取值;级别与判定表达式本体**锁死**(见 dq_rule_config)。 + + 经营口径排除(F6 / ASK-05 ③-2,2026-09-29 接线):全账套口径(`entity_id=0`)下, + `scope.excluded_entities` 名单内账套(演示/测试)的行**不再计入**评分与明细; + 单账套口径(`entity_id≠0`)保持原判据(可查看,仅标注不计入经营口径)。 """ eid = entity_id or 0 cfg = _cfg_values(db, eid) resolution = _load_cfg(db, eid) + # 经营口径排除名单(F6 / ASK-05 ③-2 接线):仅全账套口径(eid=0)参与过滤 + excluded = dqcfg.excluded_entities_from_resolution(resolution) + exclusion_applied = bool(not eid and excluded) issues = [] disabled = [] @@ -865,11 +967,13 @@ def _run_governance_checks(db: Session, entity_id: int = 0): threshold = _unit_threshold(cfg) limit = _detail_limit(cfg, "unit_check") ent, params = _entity_clause(eid, "cp.entity_id") + exc, exc_params = _excluded_clause(eid, "cp.entity_id", excluded) + params.update(exc_params) 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" + "FROM cash_plans cp WHERE cp.amount > :amt" + ent + exc + " ORDER BY cp.amount DESC LIMIT 200" ), {**params, "amt": threshold}).fetchall() - unit_total = _detail_count(db, "unit_check", eid, cfg=cfg) # 真实总数(不受明细 LIMIT 截断) + unit_total = _detail_count(db, "unit_check", eid, cfg=cfg, excluded=excluded) # 真实总数(不受明细 LIMIT 截断) issues.append({ "rule": "unit_check", "level": RULES_META["unit_check"]["level"], "count": unit_total, @@ -881,10 +985,10 @@ def _run_governance_checks(db: Session, entity_id: int = 0): 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, excluded) # 已存在的 cash_plans id 集合(规则3 判断孤儿用;实体筛选时仅看该实体下 plan) - plan_id_set = _plan_id_set(db, eid) + plan_id_set = _plan_id_set(db, eid, excluded) # ── 规则2 重复预警:同 plan_id 的 pending 预警 ≥ min_count(默认 2)── if cfg["dup_alert"].get("enabled", True): @@ -918,17 +1022,20 @@ def _run_governance_checks(db: Session, entity_id: int = 0): if cfg["virtual_pollution"].get("enabled", True): limit = _detail_limit(cfg, "virtual_pollution") ent, params = _entity_clause(eid, "cp.entity_id") + exc, exc_params = _excluded_clause(eid, "cp.entity_id", excluded) + params.update(exc_params) 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" + "WHERE " + pred + ent + exc + " 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" + "WHERE cp.source = :msrc" + ent + exc + " 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) + poll_total = _detail_count(db, "virtual_pollution", eid, cfg=cfg, excluded=excluded) + manual_total = _detail_count(db, "virtual_pollution", eid, manual=True, cfg=cfg, + excluded=excluded) issues.append({ "rule": "virtual_pollution", "level": RULES_META["virtual_pollution"]["level"], "count": poll_total, @@ -952,9 +1059,14 @@ def _run_governance_checks(db: Session, entity_id: int = 0): ) if eid: ent_sql += " AND v.entity_id = :eid" + ent_params = {"eid": eid} + if not eid: + exc5, exc5_params = _excluded_clause(eid, "v.entity_id", excluded) + ent_sql += exc5 + ent_params.update(exc5_params) 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) # 真实总数 + ent_rows = db.execute(text(ent_sql), ent_params).fetchall() + ent_total = _detail_count(db, "entity_check", eid, cfg=cfg, excluded=excluded) # 真实总数 issues.append({ "rule": "entity_check", "level": RULES_META["entity_check"]["level"], "count": ent_total, @@ -974,9 +1086,14 @@ def _run_governance_checks(db: Session, entity_id: int = 0): ) if eid: comp_sql += " AND d.entity_id = :eid" + comp_params = {"eid": eid} + if not eid: + exc6, exc6_params = _excluded_clause(eid, "d.entity_id", excluded) + comp_sql += exc6 + comp_params.update(exc6_params) 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) + comp_rows = db.execute(text(comp_sql), comp_params).fetchall() + comp_total = _detail_count(db, "kpi_completeness", eid, cfg=cfg, excluded=excluded) issues.append({ "rule": "kpi_completeness", "level": RULES_META["kpi_completeness"]["level"], "count": comp_total, @@ -997,11 +1114,16 @@ def _run_governance_checks(db: Session, entity_id: int = 0): ) if eid: recon_sql += " AND d.entity_id = :eid" + recon_params = {"eid": eid} + if not eid: + exc7, exc7_params = _excluded_clause(eid, "d.entity_id", excluded) + recon_sql += exc7 + recon_params.update(exc7_params) 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) + recon_params["pct"] = _recon_pct(cfg) + recon_rows = db.execute(text(recon_sql), recon_params).fetchall() + recon_total = _detail_count(db, "reconciliation", eid, cfg=cfg, excluded=excluded) issues.append({ "rule": "reconciliation", "level": RULES_META["reconciliation"]["level"], "count": recon_total, @@ -1039,7 +1161,6 @@ def _run_governance_checks(db: Session, entity_id: int = 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, @@ -1054,9 +1175,16 @@ def _run_governance_checks(db: Session, entity_id: int = 0): "config_version": resolution.get("config_version"), "config_source": {"table_ready": resolution.get("table_ready"), "entity_config_enabled": resolution.get("entity_config_enabled")}, - # 演示/测试账套排除(F6):默认空集 ⇒ 不改变评分语义,只做标注 + # 演示/测试账套排除(F6 / ASK-05 ③-2):**已接线** —— 全账套口径(eid=0)的评分与 + # 明细已按本名单过滤(见 `_excluded_clause`);单账套查询仅标注不计入经营口径。 "scope_excluded_entities": excluded, - "excluded_from_business_caliber": bool(eid and eid in excluded), + "excluded_from_business_caliber": dqcfg.is_excluded_from_business_caliber(eid, excluded), + "business_caliber": { + "caliber": "business" if not eid else "entity", + "exclusion_applied": exclusion_applied, + "excluded_entities": excluded, + "included": not dqcfg.is_excluded_from_business_caliber(eid, excluded), + }, } @@ -1177,9 +1305,18 @@ def get_rule_config( }, "global_params": res["global"], "scope": res["scope"], - "scope_note": "演示/测试账套排除(F6)当前只做标注、不改评分语义;默认空集 = 不排除", + "scope_note": "演示/测试账套排除(F6 / ASK-05 ③-2):已接线 —— 全账套口径的评分与明细" + "不再计入名单内账套;单账套查询仍可查看(展示)。改本配置值即可调整(零代码回退)", "locked": [dict(it) for it in dqcfg.LOCKED_ITEMS], "readonly": True, + "write_contract": { + "roles": ["ceo", "it"], # 与 PUT /rule-config 的 require_role 一致 + "platform_roles": ["it"], # open_level=platform 的项仅 it 可写 + "min_reason_len": dqcfg.MIN_REASON_LEN, + "entity_config_entities": list(dqcfg.CONFIG_ENTITIES), + "bot_channel_write": False, + "endpoint": "PUT /api/cma/data-quality/rule-config", + }, "generated_at": datetime.now().isoformat(timespec="seconds"), } @@ -1239,15 +1376,23 @@ def update_rule_config( if not checked["ok"]: raise HTTPException(400, checked["error"]) + # 平台级项(open_level=platform)落到**全账套模板行**(entity_id=0): + # 「平台级」语义 = 一处写、全账套生效。若写在本账套行上,值只对本账套可见, + # 而全账套(经营)口径(entity_id=0)读不到 ⇒ 名不符实,且无法治理 + # `scope.excluded_entities` 这类平台口径项。写权限仍仅 it(上面已校验)。 + target_entity = (dqcfg.TEMPLATE_ENTITY_ID if level == dqcfg.OPEN_LEVEL_PLATFORM + else entity_id) + 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"], + result = dqcfg.upsert_param(db, target_entity, 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, + "target_entity_id": target_entity, "rule_key": rule_key, "rule_name": rule_name, "param_key": param_key, @@ -1275,6 +1420,7 @@ def update_rule_config( return { "success": True, "entity_id": entity_id, + "target_entity_id": target_entity, "rule_key": rule_key, "param_key": param_key, "before": result["before"], @@ -1285,5 +1431,6 @@ def update_rule_config( "changed_by": getattr(current_user, "username", None), "audit": {"action": dqcfg.ACTION_UPDATE, "target_type": dqcfg.TARGET_TYPE, "target_id": result["id"]}, - "note": "配置即生效(下次 governance-check 按新口径计算);历史评分不重算", + "note": "配置即生效(下次 governance-check 按新口径计算);历史评分不重算" + + (";平台级项写入全账套模板行(entity_id=0)" if target_entity == dqcfg.TEMPLATE_ENTITY_ID else ""), } diff --git a/backend/app/dq_rule_config.py b/backend/app/dq_rule_config.py index 107df1ed..87688c83 100644 --- a/backend/app/dq_rule_config.py +++ b/backend/app/dq_rule_config.py @@ -19,10 +19,15 @@ 口径(任总 2026-09-28 放行原文:「口径用清单默认值」) ---------------------------------------------------- * 默认值 = 清单 §7「建议默认值」总表,即「现行值,上线零行为差异」, - **唯一例外**:`virtual_pollution.suspect_keywords` 去掉 `sync` - (`zxt-sync` 是酣客生产同步来源,现行 134 行 100% 假阳性)——本单唯一的行为变更; + **例外一**:`virtual_pollution.suspect_keywords` 去掉 `sync` + (`zxt-sync` 是酣客生产同步来源,现行 134 行 100% 假阳性)——ASK-05 的行为变更; * 演示/测试账套(3/4/5/6)**不开配置**(`CONFIG_ENTITIES = (1, 2)`,清单 §6.4 + ASK-05 口径②); 读路径仍按三层解析返回生效值(模板/内置默认),只是写端点拒绝。 +* **例外二(2026-09-29 尾项,ASK-05 ③-2 拍板「排除」)**:`scope.excluded_entities` + 出厂默认 `[3,4,5,6]`,且**已接线** —— 全账套(经营)口径 `entity_id=0` 的评分与明细 + 不再计入这些账套的行(判据见 `app/api/data_quality.py::_excluded_clause`)。 + 单账套查询仍可查看(DAMA §6.4「参与检查展示」),只计 `excluded_from_business_caliber=True`。 + 名单为**参数化**:改本配置值即可调整(零代码回退)。 配置值一律以 JSON 字符串落库(`param_value`),`value_type` 记录类型以便前端/审计呈现。 """ @@ -150,13 +155,23 @@ SPEC: Dict[str, Dict[str, dict]] = { }, } +#: 全局项 key(经营口径排除名单,F6 / ASK-05 ③-2) +EXCLUDED_ENTITIES_KEY = "scope.excluded_entities" + +#: `resolve()["scope"]` / GET 响应里排除名单的**字段名**(短名)。 +#: 两者不同名是历史形态:配置项 key 带 `scope.` 前缀(落库/写端点用), +#: 响应 `scope` 段是「已按前缀归组」后的字典 ⇒ 取值为短名。 +EXCLUDED_ENTITIES_FIELD = "excluded_entities" + #: 全局项(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 属业务口径,待任总/财务裁定后接线)"), + EXCLUDED_ENTITIES_KEY: dict( + type="json", default=[3, 4, 5, 6], open_level="platform", max_items=20, item_max_len=6, + desc="排除出「经营口径」评分与播报的账套(3=测试企业 / 4=VerifyCorp / 5=VerifyTest / " + "6=演示企业)。已接线:全账套口径(entity_id=0)的评分与明细不再计入这些账套的行;" + "单账套查询仍可查看(展示),但标记「不计入经营口径」。写权限仅 it(平台级);" + "改本配置值即可调整名单(零代码回退)"), } # ── 锁死面(清单 §4;仅用于给出可读文案,非白名单)───────────────────────── @@ -427,6 +442,59 @@ def validate_param(rule_key: str, param_key: str, value: Any) -> dict: return dict(ok=True, value=value, error=None, warning=None) +# ══════════════════════════════════════════════════════════════════ +# 经营口径排除名单(F6 / ASK-05 ③-2「排除演示账套」)—— 单一事实源 +# ══════════════════════════════════════════════════════════════════ +def normalize_entity_ids(raw: Any) -> List[int]: + """把配置里的账套号规范成 int 列表(去重保序;不可解析的元素忽略 = 视为不排除)。""" + out: List[int] = [] + for v in raw or []: + try: + n = int(v) + except (TypeError, ValueError): + continue + if n < 0 or n in out: + continue + out.append(n) + return out + + +def excluded_entities_from_resolution(resolution: Optional[dict]) -> List[int]: + """从 `resolve()` 结果里取经营口径排除名单(唯一取值口)。 + + ⚠ 兼容两种字段形态:`scope` 段由 `_assemble()` 按前缀归组 ⇒ 短名 + (`EXCLUDED_ENTITIES_FIELD`);直接取配置项全名的情况也接受。 + 缺失字段名会让本函数静默返回空集(= 排除名单形同未接线),故两种都查。 + """ + scope = (resolution or {}).get("scope") or {} + raw = scope.get(EXCLUDED_ENTITIES_FIELD) + if raw is None: + raw = scope.get(EXCLUDED_ENTITIES_KEY) + return normalize_entity_ids(raw) + + +def excluded_entity_ids(db: Session, entity_id: int = TEMPLATE_ENTITY_ID) -> List[int]: + """按账套三层解析后取排除名单(供引擎/播报侧单一来源引用)。 + + 读路径绝不 500:解析异常 → 返回内置默认(`resolve()` 内部已做容错,此处再兜一层)。 + """ + try: + return excluded_entities_from_resolution(resolve(db, entity_id)) + except Exception as exc: # noqa: BLE001 + logger.warning("DQ_SCOPE_EXCLUDED_RESOLVE_FAILED entity=%s: %s → 用内置默认", + entity_id, exc) + return normalize_entity_ids(GLOBAL_SPEC[EXCLUDED_ENTITIES_KEY].get("default")) + + +def is_excluded_from_business_caliber(entity_id: Any, excluded: Any) -> bool: + """单个账套是否被排除出经营口径(entity_id 为空/0 → False,不排除默认账套)。""" + try: + eid = int(entity_id or 0) + except (TypeError, ValueError): + return False + return bool(eid) and eid in set(normalize_entity_ids(excluded)) + + # ══════════════════════════════════════════════════════════════════ # 读路径:三层解析 # ══════════════════════════════════════════════════════════════════ @@ -499,6 +567,12 @@ def _assemble(eid: int, ready: bool, rows: Dict[Tuple[str, str], dict]) -> dict: entry = { "value": val, "source": source, "open_level": spec.get("open_level"), "desc": spec.get("desc"), "updated_by": None, "updated_at": None, + # 类型/取值区间对前端呈现(**仅提示**;越界/类型一律由写端点判定, + # 前端不得据本字段自造「校验通过」——清单 §5.1 反刷分约束) + "value_type": spec.get("type") or "json", + "bounds": list(spec["bounds"]) if spec.get("bounds") else None, + "enum": list(spec["enum"]) if spec.get("enum") else None, + "overflow": spec.get("overflow"), } if row is not None: entry["updated_by"] = getattr(row, "updated_by", None) @@ -554,8 +628,11 @@ def upsert_param(db: Session, entity_id: int, rule_key: str, param_key: str, val 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 {} + if is_global_param(rule_key): + # 全局项:`resolve()["global"]` 直接以 param_key 为键(无 rule_key 一层) + before_entry = (before.get("global") or {}).get(param_key, {}) or {} + else: + before_entry = ((before.get("rules") or {}).get(rule_key, {}) or {}).get(param_key, {}) or {} value_type = value_type_of(rule_key, param_key) payload = dumps_value(value) diff --git a/backend/tests/test_dq_rule_config_20260928.py b/backend/tests/test_dq_rule_config_20260928.py index 017bc13c..8adaa044 100644 --- a/backend/tests/test_dq_rule_config_20260928.py +++ b/backend/tests/test_dq_rule_config_20260928.py @@ -141,8 +141,8 @@ class TestSpecSurface: 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"] == [] + # F6 排除名单出厂默认 = [3,4,5,6](2026-09-29 尾项,拍板件 ③-2「排除」;DAMA §6.4) + assert dqcfg.GLOBAL_SPEC["scope.excluded_entities"]["default"] == [3, 4, 5, 6] def test_locked_items_cover_inventory_ten(self): """清单 §4 十类锁死面完整登记(少一类即视为私自扩面)。""" @@ -215,8 +215,9 @@ class TestEngineWiring: 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 + assert r["scope_excluded_entities"] == [3, 4, 5, 6] + assert r["excluded_from_business_caliber"] is False # eid=1 不在名单 + assert r["business_caliber"]["caliber"] == "entity" def test_sync_ab_three_states(self, db): """A/B:默认(0 命中) → 恢复旧关键词表+空白名单(7 命中) → 只留白名单(2 命中)。 @@ -356,9 +357,23 @@ class TestConfigEndpoints: 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 body["scope"]["excluded_entities"] == [3, 4, 5, 6] # F6 出厂默认(尾项 2026-09-29) assert len(body["locked"]) == 10 assert body["readonly"] is True + # 写契约(前端编辑入口的判据来源,与 PUT 的 require_role 同口径) + wc = body["write_contract"] + assert wc["roles"] == ["ceo", "it"] and wc["platform_roles"] == ["it"] + assert wc["min_reason_len"] == dqcfg.MIN_REASON_LEN == 10 + assert wc["bot_channel_write"] is False + assert wc["entity_config_entities"] == [1, 2] + assert wc["endpoint"] == "PUT /api/cma/data-quality/rule-config" + assert "已接线" in body["scope_note"] + # 前端呈现所需的类型/区间提示(仅提示,非校验通过依据) + th = body["rules"]["unit_check"]["params"]["amount_threshold"] + assert th["value_type"] == "int" and th["bounds"] == [100, 1000000] + assert th["overflow"] == "reject" + assert body["global_params"]["scope.excluded_entities"]["value"] == [3, 4, 5, 6] + assert body["global_params"]["scope.excluded_entities"]["open_level"] == "platform" def test_get_rule_config_finance_readonly_allowed(self, db, client): create_test_user(db, username="cfgfin", role="finance") @@ -527,3 +542,146 @@ class TestTableMissingTolerance: assert {i["rule"]: i["count"] for i in gov.json()["issues"]}["unit_check"] == 1 finally: QualityRuleConfig.__table__.create(bind=cf.TEST_ENGINE) + + +# ══════════════════════════════════════════════════════════════════ +# E 经营口径排除(F6 / ASK-05 ③-2 接线,2026-09-29 尾项) +# 口径:名单只对**全账套口径(eid=0)**生效 —— 被排除账套的行不计入评分/明细; +# 单账套查询(eid!=0)保持展示,仅由响应字段标注「不计入经营口径」。 +# ══════════════════════════════════════════════════════════════════ +GLOBAL_KEY = dqcfg.GLOBAL_RULE_KEY # "global" + + +def _gov(db: Session, eid: int): + """跑真函数,返回 (各规则 count 字典, 完整结果)。""" + r = _run_governance_checks(db, eid) + return {i["rule"]: i["count"] for i in r["issues"]}, r + + +def _seed_caliber_rows(db: Session) -> None: + """账套1(计入经营口径)+ 账套6(演示企业,被排除)各一条 unit_check 命中行。""" + db.add(Entity(id=6, name="演示企业", short_name="演示", status="demo")) + db.commit() + _mk_plan(db, entity_id=1, id=8101, amount=50000.0, source="batch_import") + _mk_plan(db, entity_id=6, id=8102, amount=80000.0, source="batch_import") + + +class TestBusinessCaliberExclusion: + def test_default_is_346_and_single_source(self, db): + """出厂默认 [3,4,5,6];编号规范化;判据函数与解析结果同源。""" + assert dqcfg.GLOBAL_SPEC["scope.excluded_entities"]["default"] == [3, 4, 5, 6] + assert dqcfg.excluded_entity_ids(db, 0) == [3, 4, 5, 6] + assert dqcfg.normalize_entity_ids([6, "5", 5, None, "x", -1, 3, 4]) == [6, 5, 3, 4] + assert dqcfg.is_excluded_from_business_caliber(6, [3, 4, 5, 6]) is True + assert dqcfg.is_excluded_from_business_caliber(1, [3, 4, 5, 6]) is False + assert dqcfg.is_excluded_from_business_caliber(0, [3, 4, 5, 6]) is False # 全账套标识本身不排除 + + def test_caliber_surface_full_scope(self, db): + """全账套口径(eid=0):exclusion_applied=True,excluded_from_business_caliber=False。""" + _seed_caliber_rows(db) + counts, r = _gov(db, 0) + assert r["scope_excluded_entities"] == [3, 4, 5, 6] + assert r["business_caliber"] == { + "caliber": "business", "exclusion_applied": True, + "excluded_entities": [3, 4, 5, 6], "included": True, + } + assert r["excluded_from_business_caliber"] is False + assert counts["unit_check"] == 1 # 演示账套那条(80000)已被排除 + + def test_caliber_surface_single_entity_keeps_display(self, db): + """单账套口径(eid=6):仍可查看自己的结果,仅标注不计入经营口径。""" + _seed_caliber_rows(db) + counts, r = _gov(db, 6) + assert counts["unit_check"] == 1 # 展示保留 + assert r["business_caliber"]["caliber"] == "entity" + assert r["business_caliber"]["exclusion_applied"] is False + assert r["business_caliber"]["included"] is False + assert r["excluded_from_business_caliber"] is True # 标注=不计入 + + def test_ab_aggregate_only_excluded_entity_rows_drop(self, db): + """A/B:名单 [] (关)vs [3,4,5,6](开)—— 差异**只能**来自被排除账套的行。""" + _seed_caliber_rows(db) + _mk_row(db, GLOBAL_KEY, dqcfg.EXCLUDED_ENTITIES_KEY, [], entity_id=0) + ca, ra = _gov(db, 0) + assert ra["business_caliber"]["exclusion_applied"] is False + assert ca["unit_check"] == 2 + + _mk_row(db, GLOBAL_KEY, dqcfg.EXCLUDED_ENTITIES_KEY, [3, 4, 5, 6], entity_id=0) + cb, rb = _gov(db, 0) + assert rb["business_caliber"]["exclusion_applied"] is True + assert cb["unit_check"] == 1 + assert ra["score"] < rb["score"] # 排除后扣分减少 ⇒ 分数上升 + for rule in ca: # 其余规则逐项不变(无越界影响) + if rule != "unit_check": + assert ca[rule] == cb[rule], rule + + def test_ab_detail_count_same_source(self, db): + """明细总数与评分同源:`_detail_count` 在同名单下的读数与 issues.count 一致。""" + from app.api.data_quality import _detail_count + _seed_caliber_rows(db) + with_exc = _detail_count(db, "unit_check", 0, excluded=[3, 4, 5, 6]) + without = _detail_count(db, "unit_check", 0, excluded=[]) + assert without == 2 and with_exc == 1 + counts, _r = _gov(db, 0) + assert counts["unit_check"] == with_exc + # 单账套口径不受名单影响 + assert _detail_count(db, "unit_check", 6, excluded=[3, 4, 5, 6]) == 1 + + def test_platform_write_of_excluded_list_changes_caliber(self, db, client): + """平台级项(仅 it):写向全账套模板行;改名单纯读即改变经营口径(零代码回退)。""" + _seed_caliber_rows(db) + create_test_user(db, username="cmait2", role="it") + it = get_token_for_user(client, "cmait2") + r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(it), + json={"rule_key": GLOBAL_KEY, "param_key": dqcfg.EXCLUDED_ENTITIES_KEY, + "value": [], "change_reason": REASON}) + assert r.status_code == 200, r.text + assert r.json()["target_entity_id"] == 0 and r.json()["after"] == [] + row = db.query(QualityRuleConfig).filter_by( + entity_id=0, rule_key=GLOBAL_KEY, param_key=dqcfg.EXCLUDED_ENTITIES_KEY).one() + assert row.updated_by == "cmait2" + counts, res = _gov(db, 0) + assert res["business_caliber"]["exclusion_applied"] is False and counts["unit_check"] == 2 + # 恢复原型(幂等:同键原位更新,不新增行) + back = client.put("/api/cma/data-quality/rule-config", headers=auth_header(it), + json={"rule_key": GLOBAL_KEY, "param_key": dqcfg.EXCLUDED_ENTITIES_KEY, + "value": [3, 4, 5, 6], "change_reason": REASON}) + assert back.status_code == 200 and back.json()["before"] == [] + counts2, res2 = _gov(db, 0) + assert res2["business_caliber"]["exclusion_applied"] is True and counts2["unit_check"] == 1 + assert db.query(QualityRuleConfig).filter_by( + entity_id=0, rule_key=GLOBAL_KEY, param_key=dqcfg.EXCLUDED_ENTITIES_KEY).count() == 1 + + +class TestPlatformWriteTarget: + """平台级项落全账套模板行(entity_id=0)——「一处写、全账套生效」。""" + + def test_it_writes_template_row_and_audit_carries_target(self, db, client): + create_test_user(db, username="cmait3", role="it") + it = get_token_for_user(client, "cmait3") + r = client.put("/api/cma/data-quality/rule-config", headers=auth_header(it), + json={"rule_key": "unit_check", "param_key": "detail_limit", "value": 25, + "change_reason": REASON}) + assert r.status_code == 200, r.text + body = r.json() + assert body["entity_id"] == 1 and body["target_entity_id"] == 0 + assert "模板行" in body["note"] + assert db.query(QualityRuleConfig).filter_by( + entity_id=0, rule_key="unit_check", param_key="detail_limit").count() == 1 + assert db.query(QualityRuleConfig).filter_by( + entity_id=1, rule_key="unit_check", param_key="detail_limit").count() == 0 + log = db.query(OperationLog).filter_by(action="update_dq_rule_config").one() + assert log.detail["entity_id"] == 1 and log.detail["target_entity_id"] == 0 + assert log.detail["open_level"] == "platform" + + def test_entity_level_param_still_writes_own_row(self, db, client): + """账套级项不受模板改写影响:仍写本账套行,target_entity_id == 本账套。""" + token = _ceo_token(db, client) + r = 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}) + assert r.status_code == 200 and r.json()["target_entity_id"] == 1 + assert db.query(QualityRuleConfig).filter_by( + entity_id=1, rule_key="dup_alert", param_key="min_count").count() == 1 + assert db.query(QualityRuleConfig).filter_by( + entity_id=0, rule_key="dup_alert", param_key="min_count").count() == 0 diff --git a/frontend/src/views/DataQuality.vue b/frontend/src/views/DataQuality.vue index 37a79482..49b7804b 100644 --- a/frontend/src/views/DataQuality.vue +++ b/frontend/src/views/DataQuality.vue @@ -261,6 +261,16 @@ {{ govScore }}