Commit b95def37 authored by Data Governance Dev's avatar Data Governance Dev

feat(web): 字段匹配规则外置到 YAML + 检查项自定义输入框 + 卡片折叠

匹配规则外置:
- 新增 web/configs/standards_match.yaml(44 个 indicator + 14 条 length 规则)
- 新增 web/core/match_config.py:线程安全加载 + get_indicator_match / get_length_rules / merge_user_override
- step6 / step7 走 YAML + 用户运行时 override 三级 fallback(用户输入追加而非替换)
- ConnectRequest 新增 match_overrides: dict[str, dict[str, list[str]]]
- 新增 GET /api/match-config:返回前端输入框默认值(逗号分隔格式)

前端检查项 2 个输入框:
- 每行 leaf 后挂「字段名 / 字段注释」2 个 el-input
- loadMatchConfig() 启动时拉 YAML 默认值灌进 form.match_overrides
- buildMatchOverridesForSubmit() 提交时把字符串转 list(Pydantic 期望 list[str])
- 跨字段 indicator(IND-301/302/901/902)UI 隐藏输入框(_skip: true)

卡片折叠 + 动画:
- 连接数据库 / 分析配置 / 数据字典浏览器 三张卡都加折叠按钮 + Vue transition
- startJob() 成功后三张卡自动折叠,焦点落到「分析进度」
- el-card__body 折叠时 padding 上下 1px(默认 20px),padding 加 0.28s 过渡
- el-card__header 折叠时 padding 6px,margin-bottom 折叠时 8px
- i 图标移到输入框右侧
- progress-block margin 改为 0

Bug fix:
- testConnection 也走 buildMatchOverridesForSubmit()(否则 names/comments 是 string → 422)
- inline <style> 全局 CSS 不能用 :deep(),全部改为直接选择器

WORKLOG 详细记录所有决策点。
parent a0c39eb1
This diff is collapsed.
......@@ -190,6 +190,89 @@ async def analysis_tree():
return {"groups": out_groups}
# ── 字段匹配规则配置(前端 UI 输入框默认值来源)──
# GET /api/match-config → 返回 YAML 配置(按 step_id 组织)。
# key = step_id(如 "length_check" / "std_ind_001_a")
# value = {
# "names": str, # 字段名(逗号分隔,UI 输入框格式)
# "comments": str, # 字段注释(逗号分隔)
# "skip": bool, # 跨字段/跨表 indicator → UI 隐藏输入框
# }
# 加载失败时返回空 dict,前端按「空 = 不预填」处理。
@router.get("/match-config")
async def get_match_config():
from ..core.match_config import _load
from ..core.orchestrator import get_step_defs
cfg = _load()
step7 = (cfg.get("step7_standards") or {}) if isinstance(cfg, dict) else {}
step6_rules = ((cfg.get("step6_length") or {}).get("rules") or []) if isinstance(cfg, dict) else []
# 把所有 step_id 收集起来:保证 UI 上每个 leaf 都有一份默认配置
# step_id → step 标识
# standard_id → indicator 的原始 ID(用于 YAML 反查;可能为 None)
by_step: dict[str, dict] = {}
# ── Step 7:每个 indicator 的 step_id(如 "std_ind_001_a")────
# YAML 配置里的 key 是 standard_id(如 "IND-001-a"),需要 map 到 step_id。
# 约定:step_id = "std_ind_" + standard_id.lower().replace("-", "_")
for sid, meta in getattr(_list_std_step_ids_safe(), "__iter__", lambda: [])() or []:
# meta: {"id": "IND-001-a", "name": "...", ...}
std_id = meta.get("id", "")
rec = step7.get(std_id) or {}
if rec.get("_skip"):
by_step[sid] = {"names": "", "comments": "", "skip": True}
else:
atf = rec.get("applies_to_fields") or []
cmk = rec.get("comment_keywords") or []
by_step[sid] = {
"names": ", ".join(atf),
"comments": ", ".join(cmk),
"skip": False,
}
# ── Step 6:length_check 聚合所有规则的 names / comments ──────
all_names: list[str] = []
all_comments: list[str] = []
for r in step6_rules:
# 从 regex pattern 抽出可读的 alternation(仅做显示,不强求一致)
# 例:^(id_?card|id_?number)$ → "id_card, id_number"
# 简单实现:拿 | 分隔的 alternation 串
pat = r.get("name_pattern") or ""
body = pat.strip("^$")
for piece in body.split("|"):
p = piece.strip().rstrip("$").lstrip("^")
# 剥掉分组括号((?:...)/(...) 等)→ UI 显示干净
p = p.strip("()")
# 把正则简化为可读示例(id_?card → id_card / id_card;这里只列原文)
if p:
all_names.append(p)
for kw in (r.get("comment_keywords") or []):
if kw not in all_comments:
all_comments.append(kw)
by_step["length_check"] = {
"names": ", ".join(all_names),
"comments": ", ".join(all_comments),
"skip": False,
}
# ── 其余 step(merge / empty / missing_comments 等)无需输入框 ──
# 前端按 step_id 找不到时按 skip=True 处理即可
return {"steps": by_step}
def _list_std_step_ids_safe():
"""延迟加载 standards 包;失败时返回空 list-like。"""
try:
from standards.registry import list_step_ids as _list
return _list()
except Exception as e:
logger.warning(f"加载 standard step_ids 失败: {e}")
return []
# ── 列出已注册国标标准 ──
@router.get("/standards", response_model=StandardsListResponse)
async def list_standards():
......
This diff is collapsed.
......@@ -231,6 +231,7 @@ class JobManager:
tables=job.req.tables,
run_dir=run_dir,
enable_llm=job.req.enable_llm,
match_overrides=getattr(job.req, "match_overrides", None),
on_log=lambda entry: self._sync_log(job, entry),
on_step_start=lambda step_id, title: self._sync_step_start(job, step_id, title),
on_step_done=lambda step_id, title, ok: self._sync_step_done(job, step_id, title, ok),
......
"""字段匹配规则配置加载(2026-08-12 起替代代码硬编码)
职责:
- 从 web/configs/standards_match.yaml 加载匹配规则(启动时缓存一次)
- 提供 step6 / step7 的查询接口(带 fallback 到代码硬编码的兜底语义)
- 提供用户运行时 override 的合并工具
使用:
- web/core/step_impl/step6_length_check.py 启动时按本文件构建 LENGTH_RULES
- web/core/step_impl/step7_standards.py 每次跑 indicator 时调
`get_indicator_match(standard_id)` 拿匹配规则
- web/api/routes.py 在 GET /api/match-config 把当前配置返回给前端做默认值
设计原则:
- 文件缺失 / 解析失败:返回空 dict;调用方回退到代码内置默认值
- 不抛异常(治理任务启动时不应被配置文件卡住)
"""
from __future__ import annotations
import logging
import threading
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
# ── 配置路径 ──────────────────────────────────────────────
_CONFIG_PATH = (
Path(__file__).resolve().parent.parent / "configs" / "standards_match.yaml"
)
# ── 进程级缓存(启动加载一次) ───────────────────────────────
_cache: dict[str, Any] | None = None
_cache_lock = threading.Lock()
def _load() -> dict[str, Any]:
"""读 YAML 并缓存;失败返回空 dict。"""
global _cache
if _cache is not None:
return _cache
with _cache_lock:
if _cache is not None:
return _cache
if not _CONFIG_PATH.exists():
logger.info(
f"match config 缺失({_CONFIG_PATH}),"
"step6/7 将回退到代码内置默认值"
)
_cache = {}
return _cache
try:
import yaml # 延迟导入;缺 PyYAML 时降级
text = _CONFIG_PATH.read_text(encoding="utf-8")
data = yaml.safe_load(text)
if not isinstance(data, dict):
logger.warning(
f"match config 顶层不是 dict({type(data).__name__}),"
"回退到空配置"
)
_cache = {}
return _cache
_cache = data
n_ind = len(data.get("step7_standards") or {})
n_len = len(data.get("step6_length", {}).get("rules") or [])
logger.info(
f"match config 已加载: step7_standards {n_ind} 个 indicator, "
f"step6_length {n_len} 条规则"
)
return _cache
except Exception as e:
logger.warning(f"match config 加载失败: {e},回退到空配置")
_cache = {}
return _cache
def reset_cache_for_tests() -> None:
"""测试用:清掉缓存,下次 load() 重新读盘。"""
global _cache
with _cache_lock:
_cache = None
# ── Step 7 查询接口 ────────────────────────────────────────────
def get_indicator_match(standard_id: str) -> dict[str, list[str]]:
"""获取某个 indicator 的匹配规则。
返回 {"applies_to_fields": [...], "comment_keywords": [...]}
其中任何 key 不存在时为空 list。
配置文件里没写该 indicator(首次跑 / 配置缺失)→ 返回空 dict;
调用方应继续 fallback 到 BaseStandard 子类的类属性。
"""
cfg = _load()
section = cfg.get("step7_standards") or {}
if not isinstance(section, dict):
return {}
rec = section.get(standard_id) or {}
if not isinstance(rec, dict):
return {}
if rec.get("_skip"):
return {"_skip": True} # 跨字段/跨表 indicator
atf = list(rec.get("applies_to_fields") or [])
cmk = list(rec.get("comment_keywords") or [])
return {"applies_to_fields": atf, "comment_keywords": cmk}
# ── Step 6 查询接口 ────────────────────────────────────────────
def get_length_rules() -> list[dict] | None:
"""获取 step6 的 LENGTH_RULES 配置。
返回 list[dict](每条含 name_pattern / comment_keywords / expected_length /
standard / description),配置文件缺失或 step6_length.rules 为空时返回 None
—— 表示使用代码内置 LENGTH_RULES。
"""
cfg = _load()
rules = ((cfg.get("step6_length") or {}).get("rules")) or None
if not rules:
return None
return rules
# ── 用户 override 合并 ────────────────────────────────────────
def merge_user_override(
base_atf: list[str] | None,
base_cmk: list[str] | None,
user_override: dict[str, list[str]] | None,
) -> tuple[list[str], list[str]]:
"""合并用户运行时 override 到基础匹配规则。
Args:
base_atf: 基础 applies_to_fields(来自配置或类属性)
base_cmk: 基础 comment_keywords
user_override: {"names": [...], "comments": [...]}
任一为 None/空都不会改基础集合
Returns:
(merged_atf, merged_cmk) —— 去重,顺序:基础在前 + override 在后
"""
atf = list(base_atf or [])
cmk = list(base_cmk or [])
if not user_override:
return atf, cmk
extra_names = [n for n in (user_override.get("names") or []) if n]
extra_cmts = [c for c in (user_override.get("comments") or []) if c]
if extra_names:
seen = set(atf)
for n in extra_names:
if n not in seen:
atf.append(n)
seen.add(n)
if extra_cmts:
seen = set(cmk)
for c in extra_cmts:
if c not in seen:
cmk.append(c)
seen.add(c)
return atf, cmk
\ No newline at end of file
......@@ -31,6 +31,13 @@ class ConnectRequest(BaseModel):
enable_llm: bool = True
# 可选:报告标题
report_title: Optional[str] = None
# 可选:每个 step 的「字段名 / 字段注释」用户运行时 override
# key = step_id(如 "length_check" / "std_ind_001_a")
# value = {"names": ["id_card", ...], "comments": ["身份证", ...]}
# 语义:追加到 web/configs/standards_match.yaml 的同名 indicator 规则上,
# 用于本次任务(不写回配置文件)。
# 跨字段 / 跨表 indicator(IND-301/302/901/902)即使提交了 override 也会被忽略。
match_overrides: dict[str, dict[str, list[str]]] = Field(default_factory=dict)
class TestConnectionRequest(ConnectRequest):
......
......@@ -95,46 +95,58 @@ def get_step_defs() -> list[StepDef]:
return sorted(_STEPS.values(), key=lambda s: s.order)
# ── Step 函数:参数统一 (cfg, dict_data, llm, log, cancel_event, table_filter) ──
# ── Step 函数:参数统一 (cfg, dict_data, llm, log, cancel_event, table_filter, match_overrides) ──
# 每个 _run_* 接收 orchestrator 提供的统一上下文,做最薄的适配(调真正的 step_impl)
# match_overrides: dict[step_id] -> {"names": [...], "comments": [...]}
# 仅 step6(length_check)+ step7 单值 indicator 会消费;其他 step 忽略。
def _run_merge_redundancy(*, cfg, dict_data, llm, log, cancel_event, table_filter):
def _run_merge_redundancy(*, cfg, dict_data, llm, log, cancel_event, table_filter, match_overrides):
from .step_impl.step2_merge_redundancy import run_step2
data = run_step2(dict_data, llm=llm, log=log, table_filter=table_filter)
return {"section_key": "merge_candidates", "data": data}
def _run_empty_fields(*, cfg, dict_data, llm, log, cancel_event, table_filter):
def _run_empty_fields(*, cfg, dict_data, llm, log, cancel_event, table_filter, match_overrides):
from .step_impl.step4_empty_fields import run_step4
data = run_step4(cfg, log=log, dict_data=dict_data, table_filter=table_filter)
return {"section_key": "empty_fields", "data": data}
def _run_missing_comments(*, cfg, dict_data, llm, log, cancel_event, table_filter): # noqa: ARG001
def _run_missing_comments(*, cfg, dict_data, llm, log, cancel_event, table_filter, match_overrides): # noqa: ARG001
"""缺失注释字段检查(纯规则,不需要 LLM,llm 参数保留仅为签名统一)"""
from .step_impl.step5_missing_comments import run_step5
data = run_step5(dict_data, log=log, table_filter=table_filter)
return {"section_key": "missing_comments", "data": data}
def _run_length_check(*, cfg, dict_data, llm, log, cancel_event, table_filter):
def _run_length_check(*, cfg, dict_data, llm, log, cancel_event, table_filter, match_overrides):
from .step_impl.step6_length_check import run_step6
data = run_step6(dict_data, log=log, table_filter=table_filter)
override = (match_overrides or {}).get("length_check")
data = run_step6(dict_data, log=log, table_filter=table_filter,
match_override=override)
return {"section_key": "length_issues", "data": data}
def _run_standards_one(standard_id: str):
"""返回单个 indicator step 的 runner(捕获 standard_id 闭包)。"""
def _runner(*, cfg, dict_data, llm, log, cancel_event, table_filter):
def _runner(*, cfg, dict_data, llm, log, cancel_event, table_filter, match_overrides):
from .step_impl.step7_standards import run_step7_for_indicator
# step_id 是 std_ind_xxx,但前端 match_overrides 按 standard_id(如 IND-001-a)组织;
# 这里两种 key 都接受,向后兼容。
override = None
if match_overrides:
override = match_overrides.get(standard_id) or match_overrides.get(
f"std_ind_{standard_id.lower().replace('-', '_')}"
)
return run_step7_for_indicator(
standard_id, cfg, dict_data=dict_data, log=log, table_filter=table_filter,
match_override=override,
)
return _runner
def _run_standards(*, cfg, dict_data, llm, log, cancel_event, table_filter): # noqa: ARG001
def _run_standards(*, cfg, dict_data, llm, log, cancel_event, table_filter, match_overrides): # noqa: ARG001
"""旧版「国家标准校验」聚合入口 —— 已被 per-indicator 模式取代,保留以兼容历史调用。
实际 orchestrator 不再注册此 step;用户勾选的是 std_ind_* 单 indicator。
"""
......@@ -575,6 +587,7 @@ async def run_governance_workflow(
tables: list[str],
run_dir: Path,
enable_llm: bool = True,
match_overrides: dict[str, dict[str, list[str]]] | None = None,
on_log: Callable[[dict], None] | None = None,
on_step_start: Callable[[str, str], None] | None = None,
on_step_done: Callable[[str, str, bool], None] | None = None,
......@@ -719,6 +732,7 @@ async def run_governance_workflow(
log=log,
cancel_event=cancel_event,
table_filter=table_filter,
match_overrides=match_overrides,
),
)
section_key = output.get("section_key")
......
......@@ -229,11 +229,17 @@ def _match_rule(col_name: str, col_comment: str, rule: LengthRule) -> tuple[bool
def run_step6(dict_data: dict, log: Callable | None = None,
table_filter: set[str] | None = None) -> dict:
table_filter: set[str] | None = None,
match_override: dict | None = None) -> dict:
"""字段长度检查(纯规则匹配)。
Args:
table_filter: 前端勾选的表名集合(None/空 = 全部)。
match_override: 用户运行时 override
{"names": [...], "comments": [...]}
追加到 YAML 配置 + 内置 LENGTH_RULES 上。
names:每个 name 当作 substring regex(re.escape)追加一条规则;
comments:追加到每条现有规则的 comment_keywords 集合里。
"""
columns = dict_data.get("data_dictionary", [])
if table_filter:
......@@ -243,9 +249,12 @@ def run_step6(dict_data: dict, log: Callable | None = None,
log("WARN", "未获取到任何字段元数据,跳过", step="length_check")
return _wrap({"summary": {"total_issues": 0}, "issues": []})
# ── 构造有效规则集:YAML 配置(若有) + 内置兜底 + 用户 override ──
rules = _build_effective_rules(match_override)
if log:
log("INFO",
f"内置长度规则: {len(LENGTH_RULES)} 条, 待扫描字段: {len(columns)}",
f"长度规则: {len(rules)} 条(YAML/内置 + 用户 override), "
f"待扫描字段: {len(columns)}",
step="length_check")
issues = []
......@@ -260,7 +269,7 @@ def run_step6(dict_data: dict, log: Callable | None = None,
if dt not in ("varchar", "char") or not max_len:
continue
for rule in LENGTH_RULES:
for rule in rules:
hit, basis = _match_rule(col_name, col_comment, rule)
if not hit:
continue
......@@ -298,10 +307,82 @@ def run_step6(dict_data: dict, log: Callable | None = None,
return _wrap({
"summary": {
"total_issues": len(issues),
"rules_applied": len(LENGTH_RULES),
"rules_applied": len(rules),
"matched_by_name": match_by_name,
"matched_by_comment": match_by_comment,
"wasted_bytes_total": sum(i.get("wasted_bytes_per_row") or 0 for i in issues),
},
"issues": issues[:200],
})
def _build_effective_rules(
match_override: dict | None,
) -> list[LengthRule]:
"""构造本次运行的 LENGTH_RULES 列表。
顺序:
1) YAML 配置(web/configs/standards_match.yaml)→ 缺失则用代码内置
2) 用户运行时 override
- names → 每个 name 当 substring regex 追加(默认 expected_length=18)
- comments → 追加到每条现有规则的 comment_keywords
"""
from ..match_config import get_length_rules, merge_user_override
cfg_rules = get_length_rules()
if cfg_rules:
base_rules = [_rule_from_dict(r) for r in cfg_rules]
source = "YAML 配置"
else:
base_rules = list(LENGTH_RULES)
source = "代码内置(YAML 缺失)"
if not match_override:
return base_rules
# 用户 override —— 追加到基础规则集
user_names = [n for n in (match_override.get("names") or []) if n]
user_comments = [c for c in (match_override.get("comments") or []) if c]
# 1) 用户 names:每个 name 追加一条 substring 规则
for name in user_names:
base_rules.append(LengthRule(
name_pattern=re.compile(re.escape(name), re.I),
comment_keywords=tuple(user_comments),
expected_length=18, # 默认身份证口径 —— 用户可改 YAML 自定义
standard="USER_OVERRIDE",
description=f"用户自定义匹配({name})",
))
# 2) 用户 comments:每条现有规则都追加
if user_comments:
for i, rule in enumerate(base_rules):
existing = list(rule.comment_keywords)
new_kw = list(merge_user_override(
base_atf=None, base_cmk=existing,
user_override={"names": None, "comments": user_comments},
)[1])
base_rules[i] = LengthRule(
name_pattern=rule.name_pattern,
comment_keywords=tuple(new_kw),
expected_length=rule.expected_length,
standard=rule.standard,
description=rule.description,
)
logger.info(
f"length_check 规则来源: {source}; "
f"用户 override names={len(user_names)} comments={len(user_comments)}"
)
return base_rules
def _rule_from_dict(d: dict) -> LengthRule:
"""YAML dict → LengthRule(容错:缺字段时退化)。"""
return LengthRule(
name_pattern=re.compile(d["name_pattern"], re.I),
comment_keywords=tuple(d.get("comment_keywords") or []),
expected_length=int(d.get("expected_length") or 0),
standard=d.get("standard") or "",
description=d.get("description") or "",
)
\ No newline at end of file
......@@ -466,6 +466,8 @@ def _empty_indicator_data() -> dict:
def _collect_hit_columns(
std_instance: BaseStandard,
all_columns: list[dict],
applies_to_fields: list[str] | None = None,
comment_keywords: list[str] | None = None,
) -> tuple[list[dict], str]:
"""按 applies_to_fields(字段名) + comment_keywords(字段注释 substring,
不区分大小写)两路并集命中字段;返回 (命中字段列表, 命中原因描述)。
......@@ -473,9 +475,17 @@ def _collect_hit_columns(
2026-08-12 恢复:用户决定走回关键字匹配方案,每个 indicator 用自身
的 applies_to_fields(精确字段名)+ comment_keywords(注释 substring)
自动推断「应该检查哪些字段」。前端不再需要逐字段勾选。
参数优先级:显式传入 > std_instance 类属性(向后兼容)。
"""
if applies_to_fields is None:
applicable = list(getattr(std_instance, "applies_to_fields", []) or [])
else:
applicable = list(applies_to_fields)
if comment_keywords is None:
comment_kws = [k.strip().lower() for k in (getattr(std_instance, "comment_keywords", []) or []) if k]
else:
comment_kws = [k.strip().lower() for k in comment_keywords if k]
applicable_set = set(applicable)
hits: list[dict] = []
......@@ -516,11 +526,45 @@ def _collect_hit_columns(
return hits, " / ".join(reason_bits) or "(无命中条件)"
def _effective_match(
std_instance: BaseStandard,
match_override: dict | None,
) -> tuple[list[str], list[str], bool]:
"""算出一个 indicator 在本次运行中的有效匹配集(合并所有来源)。
合并顺序:
1) YAML 配置(standards_match.yaml)→ 缺失则用类属性兜底
2) 用户运行时 override(UI 输入框)→ 追加
Returns:
(applies_to_fields, comment_keywords, is_cross_field)
is_cross_field=True 表示该 indicator 走跨字段/跨表,UI 不显示输入框。
"""
from ..match_config import get_indicator_match, merge_user_override
std_id = std_instance.standard_id
cfg_match = get_indicator_match(std_id)
if cfg_match.get("_skip"):
return [], [], True # 跨字段 / 跨表 indicator
# 基础 = YAML 配置(若存在)或类属性兜底
base_atf = cfg_match.get("applies_to_fields")
base_cmk = cfg_match.get("comment_keywords")
if base_atf is None and base_cmk is None:
# YAML 没写该 indicator(首次跑 / 新加 indicator)→ 回退到类属性
base_atf = list(getattr(std_instance, "applies_to_fields", []) or [])
base_cmk = list(getattr(std_instance, "comment_keywords", []) or [])
# 用户 override 追加
return merge_user_override(base_atf, base_cmk, match_override) + (False,)
def _run_round1_one(
cfg: DBConfig,
all_columns: list[dict],
std_instance: BaseStandard,
log: Callable | None,
match_override: dict | None = None,
) -> dict:
"""单值校验:跑一个 indicator 实例(适用于 22 个单值类 indicator)。
......@@ -536,8 +580,17 @@ def _run_round1_one(
f" · {std_id} ({getattr(std_instance, 'group', '通用')}) 启动",
step=indicator_step_id(std_id))
# 算有效匹配集(YAML 配置 + 用户 override);类属性作为 YAML 缺失时的兜底
effective_atf, effective_cmk, _is_cross = _effective_match(
std_instance, match_override,
)
# 找该 indicator 命中的所有字段(applies_to_fields ∪ comment_keywords)
hit_cols, hit_reason = _collect_hit_columns(std_instance, all_columns)
hit_cols, hit_reason = _collect_hit_columns(
std_instance, all_columns,
applies_to_fields=effective_atf,
comment_keywords=effective_cmk,
)
# 单 (table, field) 命中后仍按 MAX_TABLES_PER_FIELD 截断同一字段的表数;
# 不同表名 → 不同字段对 → 都保留
......@@ -552,8 +605,8 @@ def _run_round1_one(
fields_to_check.append(c)
if not fields_to_check:
applicable = list(getattr(std_instance, "applies_to_fields", []) or [])
cmt_kws = [k for k in (getattr(std_instance, "comment_keywords", []) or []) if k]
applicable = effective_atf
cmt_kws = effective_cmk
if log:
log("DEBUG",
f" · {std_id} 未匹配任何字段(字段名={applicable} / "
......@@ -695,6 +748,7 @@ def _run_round1(
all_columns: list[dict],
standards_by_field: dict[str, BaseStandard],
log: Callable | None,
match_overrides: dict[str, dict] | None = None,
) -> dict:
"""Round 1:单值校验(按 group 聚合结果)—— 跑全部 indicator,按 group 输出。"""
per_group: dict[str, dict] = {
......@@ -721,7 +775,8 @@ def _run_round1(
step="standards")
for std_id, std_instance in seen_instances.items():
single = _run_round1_one(cfg, all_columns, std_instance, log)
override = (match_overrides or {}).get(std_id)
single = _run_round1_one(cfg, all_columns, std_instance, log, match_override=override)
group = getattr(std_instance, "group", "通用")
bucket = per_group.setdefault(group, _empty_group_data())
# 合并:violations_by_indicator + violations
......@@ -1429,6 +1484,7 @@ def run_step7_for_indicator(
dict_data: dict,
log: Callable | None = None,
table_filter: set[str] | None = None,
match_override: dict | None = None,
) -> dict:
"""跑单个 indicator(orchestrator 每个 step 调一次)。
......@@ -1474,16 +1530,19 @@ def run_step7_for_indicator(
data = _run_round3_cross_field_one(cfg, by_name, log)
else:
# 单值类:字段名 OR 注释命中都可(applies_to_fields / comment_keywords 二选一非空)
has_name = bool(getattr(std_instance, "applies_to_fields", None))
has_cmt = bool(getattr(std_instance, "comment_keywords", None))
if not (has_name or has_cmt):
# 用 _effective_match 合并 YAML + 用户 override 后再判定
eff_atf, eff_cmk, _ = _effective_match(std_instance, match_override)
if not (eff_atf or eff_cmk):
if log:
log("WARN",
f"{standard_id} 没有 applies_to_fields / comment_keywords 也没匹配 "
f"IND-301/302 模板,跳过",
step=indicator_step_id(standard_id))
return _empty_indicator_step_result(standard_id)
data = _run_round1_one(cfg, columns, std_instance, log)
data = _run_round1_one(
cfg, columns, std_instance, log,
match_override=match_override,
)
return {"section_key": section_key, "data": _wrap_single_indicator(std_instance, data)}
......
This diff is collapsed.
......@@ -356,7 +356,7 @@ body {
/* ── 进度条(上) ── */
.progress-block {
margin-bottom: 20px;
margin: 0;
}
.progress-block .el-progress {
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment