feat(web): 自定义规则端到端 + 空 steps 区分 None/[]

将「数据字典浏览器 → 字段表 + 号 → 累积自定义规则 → 后端跑 LIKE 检索」打通完整链路。

【新增】
- web/core/step_impl/step9_custom_rules.py：每条规则 SELECT col FROM tbl
  WHERE col LIKE '%kw%' ESCAPE '!' LIMIT 200；跨 DB 占位符
  (mysql=%s / dmPython=oracledb=?)、统一 '!' 避免达梦 [CODE:-6106]；
  每条 SQL 命中行 = 1 个 entry（不做 distinct，重复值并列展示）
- orchestrator 注册 custom_value_check 步骤（hidden=True，前端不展示；
  提交 non-empty custom_rules 时自动追加）
- models.ConnectRequest.custom_rules 字段

【修改】
- job_manager：区分 None（跑全部）vs []（显式空选择），同步 orchestrator 语义
- orchestrator.run_governance_workflow：同上 + 接受 custom_rules 参数
- step8_report：自定义规则报告渲染改读 matches[]（7 列：规则名称/表名/
  字段名/匹配值/规则/关键字/错误；删除「命中行」列）
- index.html：分析配置卡 + 自定义规则卡 UI 联动
- docs/WORKLOG.md：上述变更的工作记录

【为什么需要整体提交】
orchestrator 的 steps 语义修正 + custom_rules 注入 + job_manager 同步 +
step9 执行 + step8 渲染 + models 字段，每一块独立提交都会留下 broken
intermediate commit。

web/core/job_manager.py

@@ -190,7 +190,13 @@ class JobManager:
         # 收集要跑的步骤（从 orchestrator 的注册表取，保持单一来源；报告生成在主流程末尾无条件触发，不在此处）
         from .orchestrator import get_step_defs
         all_steps = [s.step_id for s in get_step_defs()]
-        steps_planned = job.req.steps if job.req.steps else all_steps
+        # 区分 None（= 跑全部，向后兼容老调用） vs []（= 显式空选择，2026-08-12 起的语义）
+        # 必须同步 orchestrator.run_governance_workflow 的语义，
+        # 否则前端提交 steps=[] 时仍会跑全部 → 结果页塞满。
+        if job.req.steps is None:
+            steps_planned = list(all_steps)
+        else:
+            steps_planned = list(job.req.steps)
         job.steps_planned = steps_planned
         logger.info(f"[job {job.job_id}] 计划步骤: {steps_planned}")
 
@@ -232,6 +238,7 @@ class JobManager:
                 run_dir=run_dir,
                 enable_llm=job.req.enable_llm,
                 match_overrides=getattr(job.req, "match_overrides", None),
+                custom_rules=getattr(job.req, "custom_rules", 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

@@ -38,6 +38,13 @@ class ConnectRequest(BaseModel):
     #       用于本次任务（不写回配置文件）。
     # 跨字段 / 跨表 indicator（IND-301/302/901/902）即使提交了 override 也会被忽略。
     match_overrides: dict[str, dict[str, list[str]]] = Field(default_factory=dict)
+    # 可选：用户在前端「自定义规则」卡片里累积的规则（点字段表 + 号 → 输入关键字）
+    #   每项: {table_name, table_comment, column_name, column_comment,
+    #          rule_type: 'contains_keyword', user_input: '<keyword>'}
+    # 后端 step8_custom_rules 会按每条规则跑 SQL：
+    #   SELECT <col> FROM <table> WHERE <col> LIKE '%<keyword>%' ESCAPE '\\' LIMIT 200
+    # user_input 为空的项会被跳过（不报错）
+    custom_rules: list[dict] = Field(default_factory=list)
 
 
 class TestConnectionRequest(ConnectRequest):

web/core/orchestrator.py

@@ -155,6 +155,32 @@ def _run_standards(*, cfg, dict_data, llm, log, cancel_event, table_filter, matc
     return {"section_key": "standard_violations", "data": data}
 
 
+def _run_custom_value_check_factory(custom_rules: list[dict] | None):
+    """自定义规则 step 的 runner 工厂 —— 通过闭包捕获 custom_rules
+
+    dispatch 循环调 runner 时只传 (cfg, dict_data, llm, log, cancel_event, table_filter, match_overrides)，
+    没有 custom_rules 参数；这里通过工厂模式在注册时把 custom_rules 钉死。
+    """
+    def _runner(*, cfg, dict_data, llm, log, cancel_event, table_filter, match_overrides):  # noqa: ARG001
+        from .step_impl.step9_custom_rules import run_step_custom_value_check
+        data = run_step_custom_value_check(cfg, custom_rules or [], log=log)
+        return {"section_key": "custom_value_check", "data": data}
+    return _runner
+
+
+# ── 自定义规则上下文（workflow 启动时由 run_governance_workflow 写入） ──
+# dispatch 循环不传 custom_rules 参数，所以用一个 module-level 容器承载本次运行的规则列表。
+# 仅 run_governance_workflow 内部写，外面只读。
+_CUSTOM_RULES_CTX: list[dict] | None = None
+
+
+def _run_custom_value_check(*, cfg, dict_data, llm, log, cancel_event, table_filter, match_overrides):  # noqa: ARG001
+    """dispatch 调用的 runner —— 读 _CUSTOM_RULES_CTX（workflow 启动时注入）"""
+    from .step_impl.step9_custom_rules import run_step_custom_value_check
+    data = run_step_custom_value_check(cfg, _CUSTOM_RULES_CTX or [], log=log)
+    return {"section_key": "custom_value_check", "data": data}
+
+
 # ── 注册（按 order 升序：UI / 日志展示顺序） ──────────────────────
 register_step(
     step_id="merge_redundancy",
@@ -564,6 +590,28 @@ def _register_indicator_steps() -> None:
 _register_indicator_steps()
 
 
+register_step(
+    step_id="custom_value_check",
+    title="自定义规则（字段值包含关键字）",
+    description="用户在「数据字典浏览器」字段表右侧 + 号累积的规则：按 (表, 字段, 关键字) 跑 LIKE 检索，返回命中样本",
+    requires_db=True, llm_mode="none", required=False, order=900,
+    fn=_run_custom_value_check,
+    hidden=True,   # 2026-08-12 起：不在 UI 展示；触发方式：前端提交 custom_rules 时自动追加（见 run_governance_workflow）
+    detail=StepDetail(
+        purpose="执行用户在前端「自定义规则」卡片里累积的规则，对每条 (表, 字段, 关键字) 跑一次 LIKE 检索，返回命中样本写入报告。",
+        target="用户在数据字典浏览器里 + 号选中的字段（不受 table_filter 限制）",
+        check=(
+            "1. 对每条 rule 执行 SQL：SELECT `<col>` FROM `<table>` WHERE `<col>` LIKE '%<kw>%' ESCAPE '\\\\' LIMIT 200\n"
+            "2. 关键字里的 % / _ / \\ 先 \\ 反转义，避免被 LIKE 当通配符吃掉\n"
+            "3. user_input 为空 → 跳过该条（不报错）\n"
+            "4. SQL 执行失败 → 写 error 字段继续下一条，不中断 step\n"
+            "5. 样本 distinct 后最多保留 50 个"
+        ),
+        format="纯 SQL LIKE 检索，无国标对应；输出每条规则的命中数 + 样本值",
+    ),
+)
+
+
 # ── 关键 Step（失败直接终止任务） ──────────────────────────────
 #   merge_redundancy：LLM-required，缺 LLM = 关键失败
 #   missing_comments：2026-08-11 起改为纯规则（不再依赖 LLM），失败不再终止流程
@@ -588,6 +636,7 @@ async def run_governance_workflow(
     run_dir: Path,
     enable_llm: bool = True,
     match_overrides: dict[str, dict[str, list[str]]] | None = None,
+    custom_rules: list[dict] | 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,
@@ -603,6 +652,9 @@ async def run_governance_workflow(
         tables:     用户勾选的表名列表（不可为空 —— 由前端 / 路由层兜底校验）
         run_dir:    本次运行的输出目录
         enable_llm: 是否允许 Step 调用 LLM
+        match_overrides: 每 step 的「字段名 / 字段注释」用户运行时 override
+        custom_rules:   用户在前端累积的自定义规则（点字段表 + 号 → 输入关键字）
+                        非空时自动追加 custom_value_check 步骤到 main_steps
 
     Returns:
         {
@@ -641,8 +693,20 @@ async def run_governance_workflow(
     table_filter: set[str] | None = set(tables) if tables else None
     log("INFO", f"用户勾选表数: {len(tables)} 张")
 
+    # ── 注入自定义规则上下文（runner 通过 _CUSTOM_RULES_CTX 读） ──
+    global _CUSTOM_RULES_CTX
+    _CUSTOM_RULES_CTX = list(custom_rules) if custom_rules else None
+    if _CUSTOM_RULES_CTX:
+        log("INFO",
+            f"自定义规则: {len(_CUSTOM_RULES_CTX)} 条，将自动追加 custom_value_check 步骤",
+            step="custom_value_check")
+
     # ── 解析要跑的步骤 ──
-    requested = list(steps) if steps else [s.step_id for s in get_step_defs()]
+    # 区分 None（= 跑全部，向后兼容老调用） vs []（= 显式空选择，2026-08-12 起的语义）
+    if steps is None:
+        requested = [s.step_id for s in get_step_defs()]
+    else:
+        requested = list(steps)
     unknown = [s for s in requested if s not in _STEPS]
     if unknown:
         msg = f"未知的 step_id: {unknown}（已注册的：{[s.step_id for s in get_step_defs()]}）"
@@ -651,6 +715,16 @@ async def run_governance_workflow(
     main_steps = requested
     log("INFO", f"调度计划: 主步骤={main_steps}")
 
+    # ── 自动追加：用户提交了 custom_rules → 追加 custom_value_check 步骤 ──
+    # 仅当至少有一条非空 user_input 的规则（空规则的项会在 step 内被跳过）
+    has_effective_rule = any(
+        (r.get("user_input") or "").strip()
+        for r in (_CUSTOM_RULES_CTX or [])
+    )
+    if has_effective_rule and "custom_value_check" not in main_steps:
+        main_steps.append("custom_value_check")
+        log("INFO", "已自动追加自定义规则步骤 custom_value_check", step="custom_value_check")
+
     # ── 启动预检：LLM-required 步骤 + LLM 不可用 → fail-fast ──
     llm = get_llm_client() if enable_llm else None
     required_planned = [s for s in main_steps if _STEPS[s].llm_mode == "required"]

web/core/step_impl/step8_report.py

@@ -26,6 +26,7 @@ SECTION_ORDER = [
     ("missing_comments",     "缺失注释的字段"),
     ("length_issues",        "字段长度异常"),
     ("standard_violations",  "不符合国家标准的字段"),
+    ("custom_value_check",   "自定义规则命中"),
 ]
 
 
@@ -567,6 +568,48 @@ def _md_standards(lines: list[str], data: dict):
         lines.append("")
 
 
+def _md_custom_value_check(lines: list[str], data: dict):
+    """自定义规则（字段值包含关键字）— 命中明细
+
+    输出格式：每条 SQL 命中行 = 一行（**不做 distinct**，重复值并列出现）。
+    """
+    summary = data.get("summary") or {}
+    matches = data.get("matches") or []
+
+    total = summary.get("total_rules", 0)
+    executed = summary.get("executed_rules", 0)
+    skipped_empty = summary.get("skipped_empty_keyword", 0)
+    failed = summary.get("failed_rules", 0)
+    matched = summary.get("total_matched_rows", 0)
+
+    lines.append(f"规则数: **{total}** | "
+                 f"有效: **{executed}** | "
+                 f"空关键字跳过: **{skipped_empty}** | "
+                 f"失败: **{failed}** | "
+                 f"命中行数: **{matched}**（每条 SQL 命中行 = 一行，不去重）")
+    lines.append("")
+
+    if not matches:
+        lines.append("（未提供自定义规则 / 无命中）")
+        lines.append("")
+        return
+
+    # 保持 SQL 返回顺序（不再按 matched_rows 排序 —— 没有这个字段了）
+    lines.append(_md_row(["规则名称", "表名", "字段名", "匹配值", "规则", "关键字", "错误"]))
+    lines.append("|---|---|---|---|---|---|---|")
+    for m in matches:
+        lines.append(_md_row([
+            m.get("rule_name", "") or "-",
+            m.get("table_name", ""),
+            m.get("column_name", ""),
+            "" if m.get("value") is None else str(m.get("value")),
+            m.get("rule_type", ""),
+            m.get("keyword", ""),
+            m.get("error", "") or "-",
+        ]))
+    lines.append("")
+
+
 _MD_SECTION_HANDLERS = {
     "merge_candidates":    _md_merge,
     "redundancy_fields":   _md_redundancy,
@@ -574,6 +617,7 @@ _MD_SECTION_HANDLERS = {
     "missing_comments":    _md_missing,
     "length_issues":       _md_length,
     "standard_violations": _md_standards,
+    "custom_value_check":  _md_custom_value_check,
 }
 
 
@@ -988,6 +1032,40 @@ def _docx_standards(doc, data: dict):
         ])
 
 
+def _docx_custom_value_check(doc, data: dict):
+    """自定义规则 — 命中明细（Word）
+
+    输出格式：每条 SQL 命中行 = 一行（**不做 distinct**，重复值并列出现）。
+    """
+    summary = data.get("summary") or {}
+    matches = data.get("matches") or []
+
+    doc.add_paragraph(
+        f"规则数: {summary.get('total_rules', 0)} | "
+        f"有效: {summary.get('executed_rules', 0)} | "
+        f"空关键字跳过: {summary.get('skipped_empty_keyword', 0)} | "
+        f"失败: {summary.get('failed_rules', 0)} | "
+        f"命中行数: {summary.get('total_matched_rows', 0)}（每条 SQL 命中行 = 一行，不去重）"
+    )
+
+    if not matches:
+        doc.add_paragraph("（未提供自定义规则 / 无命中）")
+        return
+
+    rows = []
+    for m in matches:
+        rows.append([
+            m.get("rule_name", "") or "-",
+            m.get("table_name", ""),
+            m.get("column_name", ""),
+            "" if m.get("value") is None else str(m.get("value")),
+            m.get("rule_type", ""),
+            m.get("keyword", ""),
+            m.get("error", "") or "-",
+        ])
+    _add_table(doc, ["规则名称", "表名", "字段名", "匹配值", "规则", "关键字", "错误"], rows)
+
+
 _DOCX_SECTION_HANDLERS = {
     "merge_candidates":    _docx_merge,
     "redundancy_fields":   _docx_redundancy,
@@ -995,4 +1073,5 @@ _DOCX_SECTION_HANDLERS = {
     "missing_comments":    _docx_missing,
     "length_issues":       _docx_length,
     "standard_violations": _docx_standards,
+    "custom_value_check":  _docx_custom_value_check,
 }
\ No newline at end of file