feat(web): 数据字典右表格字段 checkbox + column_filter 端到端

后端
- models.py: ConnectRequest 加 columns: list[{table_name, column_name}]
  （空 = 走选中表全量字段；非空 = 仅跑用户勾选字段）
- orchestrator.py: run_governance_workflow 接 columns，构造
  column_filter: set[(table, column)] 传给每个 step
- job_manager.py: _run_job 透传 columns
- step2/4/5/6: 加 column_filter 参数 + 按 (table, column) 过滤
- step7 (核心): 删除 _collect_hit_columns 关键字推断
  （applies_to_fields / comment_keywords 不再用于挑选目标字段），
  改为 _select_target_columns 直接返回上游 column_filter 后的字段；
  run_step7_for_indicator 单值类分支去掉 has_name/has_cmt 守门

前端
- index.html: 右表格加 checkbox 列 + 全选/清空/反选 工具栏
- form.columns: Set<table||column>，勾上表时联动勾上表所有字段、
  取消表时移除该表所有字段，用户手动调整不联动回表
- startJob(): Set → [{table_name, column_name}] 后提交
- 启动按钮增加 columns.size === 0 守门 + 新 alert
- disconnectDataDict / testConnection 成功 后清空 form.columns

修复
- testConnection() 不再传 Set 类型的 columns（JSON.stringify 会变 {}
  导致 Pydantic 422），改为显式传空 list
- data_dict.py: 达梦环境下 ALL_TAB_COLUMNS / ALL_COL_COMMENTS 视图
  可能返回 N 倍重复行（B_ZJZX 12 字段实测 78 行），Python 端按
  (table_name, column_name) 兜底去重，命中时打 WARN 日志

docs/WORKLOG.md

@@ -2,6 +2,122 @@
 
 > 任务做完一次记一次。最近的在最上面。
 
+## 2026-08-11 · 数据字典浏览器：右表格加字段 checkbox + 字段级 column_filter（不再按关键字推断）
+
+### 需求
+> "现在开始修改数据字典浏览器，给右边表格的表字段也加上checkbox，然后修改分析逻辑，不要按照关键字或者注释关键字来评判是否分析字段，而是默认都检查用户勾选的字段"
+
+把「分析哪些字段」的决策权完全交给用户：右表格加 checkbox 让用户勾选要分析的字段；分析引擎**不再**按字段名 / 字段注释关键字推断要检查的字段，只跑用户勾选过的。
+
+### 改动
+
+#### 1. 后端 Pydantic 模型 —— 新增 `columns` 字段
+[web/core/models.py](web/core/models.py) `ConnectRequest` 加：
+```python
+columns: list[dict[str, str]] = Field(default_factory=list)
+```
+约定：
+- **空** = 走「选中表的全量字段」向后兼容（前端不勾字段时）
+- **非空** = 仅跑用户勾选的字段，不做关键字推断
+
+#### 2. orchestrator —— 构造 `column_filter` 传给每个 step
+[web/core/orchestrator.py](web/core/orchestrator.py) `run_governance_workflow`：
+- 接 `columns: list[dict[str, str]] | None`
+- 构造 `column_filter: set[tuple[str, str]] | None = {(t, c) for c in columns}`
+- 在 step 调用处把 `column_filter` 作为关键字参数传入 step 函数
+
+#### 3. 6 个 step runner —— 接收 + 应用 column_filter
+[web/core/step_impl/step2_merge_redundancy.py](web/core/step_impl/step2_merge_redundancy.py)、
+[step4_empty_fields.py](web/core/step_impl/step4_empty_fields.py)、
+[step5_missing_comments.py](web/core/step_impl/step5_missing_comments.py)、
+[step6_length_check.py](web/core/step_impl/step6_length_check.py)、
+[step7_standards.py](web/core/step_impl/step7_standards.py)：
+- 函数签名加 `column_filter: set[tuple[str, str]] | None = None`
+- 在 `table_filter` 之后过滤 `(table_name, column_name) in column_filter`
+- 命中后打印 INFO 日志（`按 column_filter 过滤后: N 个字段`）
+
+#### 4. step7 删除关键字推断（核心改动）
+[web/core/step_impl/step7_standards.py](web/core/step_impl/step7_standards.py) **删除** `_collect_hit_columns()`（原逻辑是按 standard 的 `applies_to_fields`（字段名）和 `comment_keywords`（注释 substring）做关键字匹配来推断「这个 indicator 该跑哪些字段」）。
+
+替换为：
+```python
+def _select_target_columns(std_instance, all_columns):
+    """2026-08-11 改：直接返回上游已按 column_filter 过滤的 all_columns。
+    ...
+    改为：用户从右表格勾选什么就检查什么。
+    """
+    return list(all_columns)
+```
+- `_run_round1_one` 调用从 `_collect_hit_columns` 改为 `_select_target_columns`
+- `run_step7_for_indicator` 单值类分支删除 `has_name / has_cmt` 守门（之前没有关键字配置就跳过，现在直接跑）
+- 日志从「字段名匹配 X / 注释匹配 Y」改为「用户勾选字段」
+
+#### 5. job_manager —— 透传 columns
+[web/core/job_manager.py](web/core/job_manager.py) `_run_job` 把 `job.req.columns or None` 传给 `run_governance_workflow`。
+
+#### 6. 前端 —— 右表格 checkbox + 联动 + 提交
+[web/static/index.html](web/static/index.html)：
+- `form` 加 `columns: new Set()`（Set<"table||column">，提交时转 list）
+- 右表格工具栏加 **全选 / 清空 / 反选** 三个按钮
+- 右表格加 `<el-table-column type="selection" reserve-selection />`（放最后一列，规避 Element Plus 的 colspan bug）
+- `@selection-change="onColumnSelectionChange"`：把选中行 → form.columns
+- `:row-key="row => row.table_name + '||' + row.column_name"`：勾选状态可跨筛选保留
+- 新增 `syncFieldsWithTables(newTables, oldTables)` + `watch(() => form.tables, ...)`：
+  - **勾上表 → 该表所有字段默认勾上**
+  - **取消表 → 该表所有字段从勾选集合移除**
+  - 用户在右表格手动调整 → 不联动回表（用户最终决定权）
+- `startJob()` 提交前把 `form.columns` Set 转成后端期望的 `[{table_name, column_name}, ...]`
+- 启动按钮：`disabled` 增加 `form.columns.size === 0`；文案加 `{{ selectedColumnCount }} 个字段`
+- `startJob()` / 启动按钮 / `resetAndStart()` 全部加 `columns.size === 0` 守门 + 新增「请勾选至少一个字段」alert
+
+### 行为变化
+| 场景 | 改前 | 改后 |
+|---|---|---|
+| 用户勾了 2 张表，没勾字段 | 全表所有字段都跑 | 启动被拦下（要勾字段） |
+| 用户勾 2 张表 → 后端自动勾上 30 个字段 | 不支持这层操作 | 启动后跑这 30 个字段 |
+| 用户手动去掉「id_card」字段检查 | 不支持（按关键字推断强制跑） | 真不跑 |
+| step7 哪些字段要跑 | 按 standard 的 `applies_to_fields` + 注释关键字推断 | 用户勾什么跑什么 |
+
+### 没改的地方
+- `web/configs/analysis_tree.json` 不动（步骤元数据未受影响）
+- `db_defaults.yaml` 不动
+- `step7` 的 standard 规则本身（length_check 的字段名正则 / 标准编号 / 描述）保持不变 —— 只改了「匹配范围」（谁被检查），没改「怎么检查」
+
+### 待自测
+1. 浏览器打开 → 选 2 张表 → 右表格应自动出现且所有字段默认勾选。
+2. 手动取消若干字段 → 启动 → 日志应显示「按 column_filter 过滤后: N 个字段」。
+3. step7 中原本会按关键字（如 `身份证`/`手机号`）匹配的字段，被取消勾选后**不应再出现**在异常里。
+
+### 即时发现的小 bug（2026-08-11）
+`POST /api/connect/test` 报 `422 Unprocessable Entity`。
+**根因**：上一轮把 `form.columns` 改成 `Set` 后，`testConnection()` 里仍然写 `JSON.stringify({ ...form, tables: [] })`，JS 的 `Set` 没有可枚举的自有属性，`JSON.stringify` 会把它序列化成 `{}` —— 后端 `columns: list[dict]` 收到 `dict` 直接 422。
+**修复**：[web/static/index.html](web/static/index.html) `testConnection()` 解构出 columns / tables 再发送：
+```js
+const { columns, tables, ...rest } = form;
+body: JSON.stringify({ ...rest, tables: [], columns: [] }),
+```
+（`startJob()` 本来就做了完整 Set → `[{table_name, column_name}]` 转换，不受影响。）
+
+### 又一个数据问题（2026-08-11）—— 达梦下字段元数据重复行
+**现象**：连接达梦 `192.168.20.24:5236/HGWXZJ` 后，左侧 `B_ZJZX` 表字段数显示 **78**（实际应该是 12）；右侧字段表格每个字段重复出现多次（截图里 `PARENT_CODE` × 2、`LEVELS` × 2 … 但右侧顶端仍标"78 个字段"）。
+**根因**：达梦 `ALL_TAB_COLUMNS` / `ALL_COL_COMMENTS` 视图在某些 DM 版本会因内部 metadata 重复而让单条 `(table_name, column_name)` 出现 N 次（具体 N 倍数按数据库实例而定）。LEFT 看到的"78 字段"就是 12 列 × ~6 倍。
+**修复**：后端 Python 层兜底去重（不动 SQL，避免方言分支无限膨胀）：
+
+[web/core/data_dict.py](web/core/data_dict.py) `extract_data_dictionary` 加：
+```python
+# 按 (table_name, column_name) 唯一键去重；首条胜出
+seen, deduped = set(), []
+for r in columns:
+    key = (r.get("table_name"), r.get("column_name"))
+    if key in seen: continue
+    seen.add(key); deduped.append(r)
+```
+并对 `tables` 按 `table_name` 同样去重。命中时打印 `WARN` 日志（`字段去重: 78 → 12 行`），后续可以定位 DM 实例问题。
+**为什么不在 SQL 加 DISTINCT**：DISTINCT 会让字段元数据的 debug 困难（看不到 N 倍的源头），而 Python 去重后还能把真实行数打出来，便于排查 DM 配置。
+**烟测**：mock 13 行（id × 6 + name × 6 + memo × 1）+ 2 张同表 → 13 → 3、2 → 1，日志正常 WARN。
+
+---
+
 ## 2026-08-11 · 隐藏 step 补充修复：前端默认全选仍把 hidden step 带进 form.steps
 
 ### 现象

web/core/data_dict.py

@@ -70,6 +70,42 @@ def extract_data_dictionary(cfg: DBConfig, log: Optional[Callable] = None) -> di
                 except (TypeError, ValueError):
                     pass
 
+    # ── 去重 ──
+    # 2026-08-11 用户反馈：达梦环境下 B_ZJZX 表从 12 字段重复成 78 行。
+    # 根因不明（可能是 ALL_COL_COMMENTS 视图在某些 DM 版本自带 N 倍行；
+    # 也可能是 schema 同名实例 / synonym 跨多 owner 联合）。
+    # 统一在 Python 端按 (table_name, column_name) 去重，取首条 —— 字段元数据
+    # 里这两个字段的组合在单库内必然唯一。
+    if columns:
+        seen = set()
+        deduped = []
+        for r in columns:
+            key = (r.get("table_name"), r.get("column_name"))
+            if key in seen:
+                continue
+            seen.add(key)
+            deduped.append(r)
+        if len(deduped) != len(columns) and log:
+            log("WARN",
+                f"  · 字段去重: {len(columns)} → {len(deduped)} 行 "
+                f"（按 table_name + column_name 唯一键）")
+        columns = deduped
+
+    if tables:
+        seen_t = set()
+        deduped_t = []
+        for r in tables:
+            key = r.get("table_name")
+            if not key or key in seen_t:
+                continue
+            seen_t.add(key)
+            deduped_t.append(r)
+        if len(deduped_t) != len(tables) and log:
+            log("WARN",
+                f"  · 表去重: {len(tables)} → {len(deduped_t)} 行 "
+                f"（按 table_name 唯一键）")
+        tables = deduped_t
+
     table_names = {r["table_name"] for r in columns}
     if log:
         log("INFO",

web/core/job_manager.py

@@ -231,6 +231,7 @@ class JobManager:
                 tables=job.req.tables,
                 run_dir=run_dir,
                 enable_llm=job.req.enable_llm,
+                columns=job.req.columns or 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),

web/core/models.py

@@ -27,6 +27,10 @@ class ConnectRequest(BaseModel):
     steps: Optional[list[str]] = None
     # 必填：要分析的表（前端从数据字典里勾选；空 = 禁止启动，由前端拦截）
     tables: list[str] = Field(default_factory=list)
+    # 可选：要分析的字段（前端从右表格勾选；[(table, column), ...]）。
+    #   - 空 = 走「选中表的全量字段」向后兼容（不勾字段表格时使用）
+    #   - 非空 = 仅跑用户勾选的字段，不再按关键字 / 注释关键字推断（2026-08-11 起改）
+    columns: list[dict[str, str]] = Field(default_factory=list)
     # 可选：是否启用 LLM 增强（默认 True，缺 Key 时自动降级）
     enable_llm: bool = True
     # 可选：报告标题

web/core/orchestrator.py

@@ -95,51 +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, column_filter) ──
 # 每个 _run_* 接收 orchestrator 提供的统一上下文，做最薄的适配（调真正的 step_impl）
 
 
-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, column_filter):
     from .step_impl.step2_merge_redundancy import run_step2
-    data = run_step2(dict_data, llm=llm, log=log, table_filter=table_filter)
+    data = run_step2(dict_data, llm=llm, log=log, table_filter=table_filter,
+                     column_filter=column_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, column_filter):
     from .step_impl.step4_empty_fields import run_step4
-    data = run_step4(cfg, log=log, dict_data=dict_data, table_filter=table_filter)
+    data = run_step4(cfg, log=log, dict_data=dict_data,
+                     table_filter=table_filter, column_filter=column_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,  # noqa: ARG001
+                          table_filter, column_filter):
     """缺失注释字段检查（纯规则，不需要 LLM，llm 参数保留仅为签名统一）"""
     from .step_impl.step5_missing_comments import run_step5
-    data = run_step5(dict_data, log=log, table_filter=table_filter)
+    data = run_step5(dict_data, log=log,
+                     table_filter=table_filter, column_filter=column_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, column_filter):
     from .step_impl.step6_length_check import run_step6
-    data = run_step6(dict_data, log=log, table_filter=table_filter)
+    data = run_step6(dict_data, log=log,
+                     table_filter=table_filter, column_filter=column_filter)
     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, column_filter):
         from .step_impl.step7_standards import run_step7_for_indicator
         return run_step7_for_indicator(
-            standard_id, cfg, dict_data=dict_data, log=log, table_filter=table_filter,
+            standard_id, cfg, dict_data=dict_data, log=log,
+            table_filter=table_filter, column_filter=column_filter,
         )
     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, column_filter):  # noqa: ARG001
     """旧版「国家标准校验」聚合入口 —— 已被 per-indicator 模式取代，保留以兼容历史调用。
     实际 orchestrator 不再注册此 step；用户勾选的是 std_ind_* 单 indicator。
     """
     from .step_impl.step7_standards import run_step7
-    data = run_step7(cfg, dict_data=dict_data, log=log, table_filter=table_filter)
+    data = run_step7(cfg, dict_data=dict_data, log=log,
+                     table_filter=table_filter, column_filter=column_filter)
     return {"section_key": "standard_violations", "data": data}
 
 
@@ -575,6 +582,7 @@ async def run_governance_workflow(
     tables: list[str],
     run_dir: Path,
     enable_llm: bool = True,
+    columns: list[dict[str, 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,
@@ -590,6 +598,9 @@ async def run_governance_workflow(
         tables:     用户勾选的表名列表（不可为空 —— 由前端 / 路由层兜底校验）
         run_dir:    本次运行的输出目录
         enable_llm: 是否允许 Step 调用 LLM
+        columns:    用户从右表格勾选的字段列表 `[{"table_name": ..., "column_name": ...}, ...]`。
+                    空 = 走「选中表的全量字段」向后兼容；
+                    非空 = 仅跑这些字段，**不再按关键字 / 注释关键字推断**（2026-08-11 改）
 
     Returns:
         {
@@ -622,11 +633,18 @@ async def run_governance_workflow(
     log = _make_log_func(on_log)
     log("INFO", f"本次治理配置: db_type={cfg.db_type}, host={cfg.host}:{cfg.port}, db={cfg.database}")
     log("INFO", f"计划执行步骤: {steps}")
-    log("INFO", f"用户勾选表数: {len(tables)} 张")
 
     cache = DataDictCache.get_instance()
     dict_data = cache.get_or_extract(cfg, log=lambda lvl, msg, **_kw: log(lvl, msg))
     table_filter: set[str] | None = set(tables) if tables else None
+    column_filter: set[tuple[str, str]] | None = (
+        {(c["table_name"], c["column_name"]) for c in columns
+         if c.get("table_name") and c.get("column_name")}
+        if columns else None
+    )
+    log("INFO",
+        f"用户勾选表数: {len(tables)} 张, "
+        f"字段数: {len(column_filter) if column_filter else '全部（按 table_filter 全表）'}")
 
     # ── 解析要跑的步骤 ──
     requested = list(steps) if steps else [s.step_id for s in get_step_defs()]
@@ -719,6 +737,7 @@ async def run_governance_workflow(
                     log=log,
                     cancel_event=cancel_event,
                     table_filter=table_filter,
+                    column_filter=column_filter,
                 ),
             )
             section_key = output.get("section_key")

web/core/step_impl/step2_merge_redundancy.py

@@ -137,11 +137,13 @@ OLD_PREFIXES = ["t_", "mall_", "project_"]
 
 def run_step2(dict_data: dict, llm: LLMClient | None = None,
               log: Callable | None = None,
-              table_filter: set[str] | None = None) -> dict:
+              table_filter: set[str] | None = None,
+              column_filter: set[tuple[str, str]] | None = None) -> dict:
     """合并候选 + 冗余字段分析。
 
     Args:
         table_filter: 前端勾选的表名集合（None/空 = 全部）。
+        column_filter: 前端从右表格勾选的字段集合（None/空 = 选中表的全量字段）。
     """
     columns = dict_data.get("data_dictionary", [])
     table_summary = dict_data.get("table_summary", [])
@@ -159,6 +161,15 @@ def run_step2(dict_data: dict, llm: LLMClient | None = None,
                 f"按 table_filter 过滤后: {len(table_summary)} 张表, {len(columns)} 个字段",
                 step="merge_redundancy")
 
+    # 字段级过滤：保留被勾选的字段（或全表字段）
+    if column_filter:
+        columns = [c for c in columns
+                   if (c["table_name"], c["column_name"]) in column_filter]
+        if log:
+            log("INFO",
+                f"按 column_filter 过滤后: {len(columns)} 个字段",
+                step="merge_redundancy")
+
     by_table: dict[str, list[dict]] = defaultdict(list)
     for row in columns:
         by_table[row["table_name"]].append(row)

web/core/step_impl/step4_empty_fields.py

@@ -178,12 +178,14 @@ def run_step4(cfg: DBConfig, log: Callable | None = None,
               high_threshold: float = 0.80,
               mid_threshold: float = 0.50,
               sample_size: int = 3,
-              table_filter: set[str] | None = None) -> dict:
+              table_filter: set[str] | None = None,
+              column_filter: set[tuple[str, str]] | None = None) -> dict:
     """扫描所有表的空字段情况（纯程序化，不依赖 LLM）
 
     Args:
         dict_data: 必填 —— orchestrator 在任务启动时已通过 DataDictCache 拿到（或现场重抽）。
         table_filter: 前端勾选的表名集合（None/空 = 全部）。
+        column_filter: 前端从右表格勾选的字段集合（None/空 = 选中表的全量字段）。
     """
     if log:
         log("INFO",
@@ -209,6 +211,15 @@ def run_step4(cfg: DBConfig, log: Callable | None = None,
                 f"按 table_filter 过滤后: {len(table_summary)} 张表, {len(columns)} 个字段",
                 step="empty_fields")
 
+    # 字段级过滤
+    if column_filter:
+        columns = [c for c in columns
+                   if (c["table_name"], c["column_name"]) in column_filter]
+        if log:
+            log("INFO",
+                f"按 column_filter 过滤后: {len(columns)} 个字段",
+                step="empty_fields")
+
     by_table: dict[str, list[dict]] = defaultdict(list)
     for row in columns:
         by_table[row["table_name"]].append(row)

web/core/step_impl/step5_missing_comments.py

@@ -101,13 +101,15 @@ def _is_missing(comment: str | None) -> bool:
 
 def run_step5(dict_data: dict,
               log: Callable | None = None,
-              table_filter: set[str] | None = None) -> dict:
+              table_filter: set[str] | None = None,
+              column_filter: set[tuple[str, str]] | None = None) -> dict:
     """缺失注释字段检查（纯规则，无 LLM）。
 
     Args:
         dict_data: 数据字典（含 data_dictionary 字段）。
         log: 日志回调（可选）。
         table_filter: 前端勾选的表名集合（None/空 = 全部）。
+        column_filter: 前端从右表格勾选的字段集合（None/空 = 选中表的全量字段）。
     """
     columns = dict_data.get("data_dictionary", [])
     if table_filter:
@@ -116,6 +118,13 @@ def run_step5(dict_data: dict,
             log("INFO",
                 f"按 table_filter 过滤后: {len(columns)} 个字段",
                 step="missing_comments")
+    if column_filter:
+        columns = [c for c in columns
+                   if (c["table_name"], c["column_name"]) in column_filter]
+        if log:
+            log("INFO",
+                f"按 column_filter 过滤后: {len(columns)} 个字段",
+                step="missing_comments")
     if not columns:
         if log:
             log("WARN", "未获取到任何字段元数据，跳过", step="missing_comments")

web/core/step_impl/step6_length_check.py

@@ -229,15 +229,24 @@ 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,
+              column_filter: set[tuple[str, str]] | None = None) -> dict:
     """字段长度检查（纯规则匹配）。
 
     Args:
         table_filter: 前端勾选的表名集合（None/空 = 全部）。
+        column_filter: 前端从右表格勾选的字段集合（None/空 = 选中表的全量字段）。
     """
     columns = dict_data.get("data_dictionary", [])
     if table_filter:
         columns = [c for c in columns if c["table_name"] in table_filter]
+    if column_filter:
+        columns = [c for c in columns
+                   if (c["table_name"], c["column_name"]) in column_filter]
+        if log:
+            log("INFO",
+                f"按 column_filter 过滤后: {len(columns)} 个字段",
+                step="length_check")
     if not columns:
         if log:
             log("WARN", "未获取到任何字段元数据，跳过", step="length_check")

web/core/step_impl/step7_standards.py

@@ -463,53 +463,21 @@ def _empty_indicator_data() -> dict:
     }
 
 
-def _collect_hit_columns(
+def _select_target_columns(
     std_instance: BaseStandard,
     all_columns: list[dict],
-) -> tuple[list[dict], str]:
-    """按 applies_to_fields（字段名） + comment_keywords（字段注释 substring，
-    不区分大小写）两路并集命中字段；返回 (命中字段列表, 命中原因描述)。
-    """
-    applicable = list(getattr(std_instance, "applies_to_fields", []) or [])
-    comment_kws = [k.strip().lower() for k in (getattr(std_instance, "comment_keywords", []) or []) if k]
-    applicable_set = set(applicable)
+) -> list[dict]:
+    """2026-08-11 改：直接返回上游已按 column_filter 过滤的 all_columns。
 
-    hits: list[dict] = []
-    seen: set[tuple[str, str]] = set()
-    hit_kinds: set[str] = set()
+    历史：早期版本会按 standard.applies_to_fields（字段名）和
+    comment_keywords（字段注释 substring）做关键字匹配来「推断这个 indicator
+    该跑哪些字段」。这种推断对用户不可见、经常猜错、且难解释为什么这个字段
+    被检查。
 
-    # 第 1 路：按字段名
-    if applicable_set:
-        for c in all_columns:
-            col_name = c.get("column_name", "")
-            if col_name in applicable_set:
-                key = (c.get("table_name", ""), col_name)
-                if key not in seen:
-                    seen.add(key)
-                    hits.append({**c, "_hit_kind": "name", "_hit_key": col_name})
-                    hit_kinds.add("name")
-
-    # 第 2 路：按字段注释（substring 不区分大小写）
-    if comment_kws:
-        for c in all_columns:
-            cc = (c.get("column_comment") or "").lower()
-            if not cc:
-                continue
-            for kw in comment_kws:
-                if kw and kw in cc:
-                    key = (c.get("table_name", ""), c.get("column_name", ""))
-                    if key not in seen:
-                        seen.add(key)
-                        hits.append({**c, "_hit_kind": "comment", "_hit_key": kw})
-                        hit_kinds.add("comment")
-                    break
-
-    reason_bits = []
-    if "name" in hit_kinds:
-        reason_bits.append(f"字段名匹配 {applicable}")
-    if "comment" in hit_kinds:
-        reason_bits.append(f"注释匹配 {comment_kws}")
-    return hits, " / ".join(reason_bits) or "（无命中条件）"
+    改为：用户从右表格勾选什么就检查什么。前端 columns 已在 orchestrator
+    入口处过滤好 all_columns，这里只做轻量包装。
+    """
+    return list(all_columns)
 
 
 def _run_round1_one(
@@ -532,8 +500,8 @@ def _run_round1_one(
             f"  · {std_id} ({getattr(std_instance, 'group', '通用')}) 启动",
             step=indicator_step_id(std_id))
 
-    # 找该 indicator 命中的所有字段（applies_to_fields ∪ comment_keywords）
-    hit_cols, hit_reason = _collect_hit_columns(std_instance, all_columns)
+    # 取要检查的字段：2026-08-11 起改为「用户勾选字段」，不再按关键字推断
+    hit_cols = _select_target_columns(std_instance, all_columns)
 
     # 单 (table, field) 命中后仍按 MAX_TABLES_PER_FIELD 截断同一字段的表数；
     # 不同表名 → 不同字段对 → 都保留
@@ -548,18 +516,15 @@ 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]
         if log:
             log("DEBUG",
-                f"  · {std_id} 未匹配任何字段（字段名={applicable} / "
-                f"注释关键字={cmt_kws}），跳过",
+                f"  · {std_id} 用户未勾选任何字段，跳过",
                 step=indicator_step_id(std_id))
         return bucket
 
     if log:
         log("DEBUG",
-            f"  · {std_id} 命中 {len(fields_to_check)} 个字段（{hit_reason}）",
+            f"  · {std_id} 待检查 {len(fields_to_check)} 个字段（用户勾选）",
             step=indicator_step_id(std_id))
 
     rec = {
@@ -1344,7 +1309,8 @@ def _run_round2_uniqueness(
 
 
 def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None,
-              table_filter: set[str] | None = None) -> dict:
+              table_filter: set[str] | None = None,
+              column_filter: set[tuple[str, str]] | None = None) -> dict:
     """旧版全量入口（向后兼容）：跑所有 indicator，按 group 输出 4 tab。
 
     新代码应直接调 run_step7_for_indicator(standard_id, ...) 跑单个 indicator。
@@ -1356,6 +1322,13 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None,
             log("INFO",
                 f"按 table_filter 过滤后: {len(columns)} 个字段",
                 step="standards")
+    if column_filter:
+        columns = [c for c in columns
+                   if (c["table_name"], c["column_name"]) in column_filter]
+        if log:
+            log("INFO",
+                f"按 column_filter 过滤后: {len(columns)} 个字段",
+                step="standards")
     if not columns:
         return _EMPTY_FALLBACK()
 
@@ -1425,12 +1398,13 @@ def run_step7_for_indicator(
     dict_data: dict,
     log: Callable | None = None,
     table_filter: set[str] | None = None,
+    column_filter: set[tuple[str, str]] | None = None,
 ) -> dict:
     """跑单个 indicator（orchestrator 每个 step 调一次）。
 
     Args:
         standard_id: 例如 "IND-001-a" / "IND-301" / "IND-302"
-        cfg / dict_data / log / table_filter: 同 run_step7
+        cfg / dict_data / log / table_filter / column_filter: 同 run_step7
 
     Returns:
         {section_key, data} —— data 已注入 _protocol（1 tab）
@@ -1438,6 +1412,13 @@ def run_step7_for_indicator(
     columns = dict_data.get("data_dictionary", [])
     if table_filter:
         columns = [c for c in columns if c["table_name"] in table_filter]
+    if column_filter:
+        columns = [c for c in columns
+                   if (c["table_name"], c["column_name"]) in column_filter]
+        if log:
+            log("INFO",
+                f"按 column_filter 过滤后: {len(columns)} 个字段",
+                step=indicator_step_id(standard_id))
 
     cls = load_standard_class(standard_id)
     if cls is None:
@@ -1469,16 +1450,8 @@ def run_step7_for_indicator(
     elif standard_id == "IND-902":
         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):
-            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)
+        # 单值类：2026-08-11 起不再按 applies_to_fields / comment_keywords 关键字推断，
+        # 直接跑用户勾选的字段（columns 已在 orchestrator 入口按 column_filter 过滤过）。
         data = _run_round1_one(cfg, columns, std_instance, log)
 
     return {"section_key": section_key, "data": _wrap_single_indicator(std_instance, data)}