feat(workflow): Phase 2 - 重构为 8 步可复用工作流 + 国标插件体系

把 Phase 1 的 7 个独立脚本收敛成一条流水线：

工作流核心 workflow/
- run_governance.py：CLI 入口，支持 --db / --steps / --skip / --offline
- config.example.yaml：阈值 / 数据库 / 输出配置（实际 config.yaml 不入库）
- core/
    db.py      连接与查询辅助
    utils.py   配置加载、目录管理、JSON I/O
    reporter.py Markdown + Word 报告生成
- steps/       8 个 Step 的实现（每个文件一个 Step）

8 步流水线：
  1 数据字典 -> 2 合并/冗余(离线) -> 3 数据验证(连库)
  -> 4 空字段(连库) -> 5 缺注释(离线) -> 6 长度检查(离线)
  -> 7 国标校验(连库) -> 8 报告生成(离线)

国标插件 standards/ 独立可扩展
- base.py + registry.py 自动发现所有标准
- std_001_id_card.py   GB 11643-1999  身份证
- std_002_uscc.py      GB 32100-2015  统一社会信用代码
- std_003_mobile.py    YD/T 1313      手机号
- std_004_xzqh.py      GB/T 2260      行政区划代码

Why：脚本是一次性产物，工作流才是可复用的工程；分离国标后新增规范
     只需要再写一个 std_xxx.py 文件。
How：每个 Step 是独立类，产出统一为 dict，run_governance.py 按编号串行
     执行；国标通过字段名匹配自动加载，registry 提供全局发现。

standards/__init__.py

+"""国家标准校验库（独立插件）
+
+每个标准是一个 Python 文件，定义一个继承 BaseStandard 的类。
+新增标准只需新建一个文件并继承基类，无需修改主流程。
+"""

standards/base.py

+"""标准基类。所有具体标准继承此类。"""
+
+from __future__ import annotations
+
+import re
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+
+
+@dataclass
+class ValidationResult:
+    """单条数据的校验结果"""
+    valid: bool
+    field_name: str
+    sample_value: str
+    reason: str = ""
+    standard: str = ""
+
+    def to_dict(self) -> dict:
+        return {
+            "valid": self.valid,
+            "field": self.field_name,
+            "sample": self.sample_value,
+            "reason": self.reason,
+            "standard": self.standard,
+        }
+
+
+class BaseStandard(ABC):
+    """标准插件基类
+
+    子类需实现：
+      - standard_id:    标准编号，如 "STD-001"
+      - standard_name:  标准中文名
+      - applies_to_fields: 适用的字段名（list）
+      - validate(value):  单值校验逻辑
+      - describe():        给人看的描述（可选）
+    """
+
+    standard_id: str = ""
+    standard_name: str = ""
+    applies_to_fields: list[str] = field(default_factory=list)
+    description: str = ""
+
+    # ── 抽象方法 ──
+    @abstractmethod
+    def validate(self, value: str) -> ValidationResult:
+        """校验单个值，返回 ValidationResult"""
+        ...
+
+    # ── 辅助：批量校验多个值，统计违规率 ──
+    def validate_batch(self, values: list[str], field_name: str = "") -> dict:
+        if not values:
+            return {
+                "standard": self.standard_id,
+                "standard_name": self.standard_name,
+                "total": 0,
+                "valid": 0,
+                "invalid": 0,
+                "violation_rate": 0,
+                "samples": [],
+            }
+
+        valid_count = 0
+        invalid_samples = []
+        for v in values:
+            if v is None or v == "":
+                continue  # 空值不算违规
+            r = self.validate(str(v))
+            if r.valid:
+                valid_count += 1
+            elif len(invalid_samples) < 10:
+                invalid_samples.append({
+                    "field": field_name or self.applies_to_fields[0],
+                    "value": str(v)[:50],
+                    "reason": r.reason,
+                })
+
+        total = sum(1 for v in values if v not in (None, ""))
+        return {
+            "standard": self.standard_id,
+            "standard_name": self.standard_name,
+            "applies_to_fields": self.applies_to_fields,
+            "total": total,
+            "valid": valid_count,
+            "invalid": total - valid_count,
+            "violation_rate": round((total - valid_count) / total, 4) if total else 0,
+            "samples": invalid_samples,
+        }

standards/registry.py

+"""标准注册器 — 自动发现所有标准
+
+新增标准只需在 standards/ 目录下新建一个继承 BaseStandard 的类即可。
+"""
+
+from __future__ import annotations
+
+import importlib
+import pkgutil
+from pathlib import Path
+
+from .base import BaseStandard
+
+
+def discover_standards() -> dict[str, BaseStandard]:
+    """扫描 standards 包下所有子类，返回 {applies_to_field: standard_instance}"""
+    import standards  # noqa: F401
+
+    standards_by_field: dict[str, BaseStandard] = {}
+    seen_classes: set[type] = set()
+
+    # 遍历 standards 包下所有模块
+    package_path = Path(__file__).parent
+    for module_info in pkgutil.iter_modules([str(package_path)]):
+        if module_info.name in ("base", "registry"):
+            continue
+        module = importlib.import_module(f"standards.{module_info.name}")
+
+        # 找 BaseStandard 子类
+        for attr_name in dir(module):
+            attr = getattr(module, attr_name)
+            if (
+                isinstance(attr, type)
+                and issubclass(attr, BaseStandard)
+                and attr is not BaseStandard
+                and attr not in seen_classes
+            ):
+                seen_classes.add(attr)
+                instance = attr()
+                for field in instance.applies_to_fields:
+                    standards_by_field[field] = instance
+
+    return standards_by_field
+
+
+def list_all_standards() -> list[dict]:
+    """返回所有标准的基本信息（用于报告/调试）"""
+    import standards
+
+    out = []
+    seen: set[type] = set()
+    package_path = Path(__file__).parent
+    for module_info in pkgutil.iter_modules([str(package_path)]):
+        if module_info.name in ("base", "registry"):
+            continue
+        module = importlib.import_module(f"standards.{module_info.name}")
+        for attr_name in dir(module):
+            attr = getattr(module, attr_name)
+            if (
+                isinstance(attr, type)
+                and issubclass(attr, BaseStandard)
+                and attr is not BaseStandard
+                and attr not in seen
+            ):
+                seen.add(attr)
+                inst = attr()
+                out.append({
+                    "id": inst.standard_id,
+                    "name": inst.standard_name,
+                    "applies_to_fields": inst.applies_to_fields,
+                    "description": inst.description,
+                })
+    return out

standards/std_001_id_card.py

+"""GB 11643-1999 公民身份号码"""
+
+from __future__ import annotations
+
+import re
+from datetime import date
+
+from .base import BaseStandard, ValidationResult
+
+
+class IdCardStandard(BaseStandard):
+    standard_id = "STD-001"
+    standard_name = "GB 11643-1999 公民身份号码"
+    applies_to_fields = ["id_card", "id_card_no", "id_number", "identity_card"]
+    description = "18位：6位地址码 + 8位出生日期 + 3位顺序码 + 1位校验码（ISO 7064 MOD 11-2）"
+
+    _REGEX = re.compile(r"^[1-9]\d{5}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dXx]$")
+    _WEIGHTS = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
+    _CHECKSUM_MAP = ["1", "0", "X", "9", "8", "7", "6", "5", "4", "3", "2"]
+
+    def validate(self, value: str) -> ValidationResult:
+        if not value:
+            return ValidationResult(True, "", value, "空值跳过", self.standard_id)
+
+        # 长度
+        if len(value) != 18:
+            return ValidationResult(False, "", value, f"长度 {len(value)} ≠ 18", self.standard_id)
+
+        # 正则
+        if not self._REGEX.match(value):
+            return ValidationResult(False, "", value, "格式不符合 6+8+3+1 结构", self.standard_id)
+
+        # 出生日期合法性（含闰年）
+        try:
+            y, m, d = int(value[6:10]), int(value[10:12]), int(value[12:14])
+            date(y, m, d)  # 触发闰年/日期校验
+        except ValueError:
+            return ValidationResult(False, "", value, f"出生日期 {value[6:14]} 不合法", self.standard_id)
+
+        # 校验码
+        expected = self._calc_checksum(value[:17])
+        if value[17].upper() != expected:
+            return ValidationResult(False, "", value, f"校验码错误，应为 {expected}", self.standard_id)
+
+        return ValidationResult(True, "", value, "", self.standard_id)
+
+    def _calc_checksum(self, seventeen: str) -> str:
+        total = sum(int(c) * w for c, w in zip(seventeen, self._WEIGHTS))
+        return self._CHECKSUM_MAP[total % 11]

standards/std_002_uscc.py

+"""GB 32100-2015 法人和其他组织统一社会信用代码"""
+
+from __future__ import annotations
+
+import re
+
+from .base import BaseStandard, ValidationResult
+
+
+class UsccStandard(BaseStandard):
+    standard_id = "STD-002"
+    standard_name = "GB 32100-2015 统一社会信用代码"
+    applies_to_fields = ["uscc", "credit_code", "social_credit_code", "unified_social_credit_code"]
+    description = "18位：登记管理机关(1) + 机构类别(1) + 行政区划(6) + 主体标识码(9) + 校验码(1)"
+
+    # 排除 I、O、Z、S、V
+    _REGEX = re.compile(r"^[1-9A-HJ-NP-RT-Y]\d[0-9A-HJ-NP-RT-Y]{6}[0-9A-HJ-NP-RT-Y]{9}[0-9A-HJ-NP-RT-Y]$")
+    _CHARSET = "0123456789ABCDEFGHJKLMNPQRTUWXY"  # 31 个字符
+    _CHAR_MAP = {c: i for i, c in enumerate(_CHARSET)}
+
+    def validate(self, value: str) -> ValidationResult:
+        if not value:
+            return ValidationResult(True, "", value, "空值跳过", self.standard_id)
+
+        if len(value) != 18:
+            return ValidationResult(False, "", value, f"长度 {len(value)} ≠ 18", self.standard_id)
+
+        # 检查禁用字符
+        forbidden = set("IOZSV")
+        used = set(value)
+        bad = used & forbidden
+        if bad:
+            return ValidationResult(False, "", value, f"包含禁用字符 {bad}", self.standard_id)
+
+        # 正则
+        if not self._REGEX.match(value):
+            return ValidationResult(False, "", value, "格式不符合 1+1+6+9+1 结构", self.standard_id)
+
+        # 校验码 (MOD 31-3)
+        expected = self._calc_checksum(value[:17])
+        if value[17] != expected:
+            return ValidationResult(False, "", value, f"校验码错误，应为 {expected}", self.standard_id)
+
+        return ValidationResult(True, "", value, "", self.standard_id)
+
+    def _calc_checksum(self, seventeen: str) -> str:
+        # 权重为 3^0, 3^1, ..., 3^16 (mod 31)
+        total = 0
+        for i, c in enumerate(seventeen):
+            if c not in self._CHAR_MAP:
+                raise ValueError(f"字符 {c} 不在字符集中")
+            total += self._CHAR_MAP[c] * (3 ** i)
+        idx = (31 - total % 31) % 31
+        return self._CHARSET[idx]

standards/std_003_mobile.py

+"""手机号校验（工信部《电信网编号计划》）"""
+
+from __future__ import annotations
+
+import re
+
+from .base import BaseStandard, ValidationResult
+
+
+class MobileStandard(BaseStandard):
+    standard_id = "STD-003"
+    standard_name = "工信部《电信网编号计划》手机号"
+    applies_to_fields = [
+        "mobile", "phone", "mobile_phone", "tel", "telephone",
+        "contact_phone", "customer_mobile", "legal_person_mobile",
+        "notify_phone", "maintainer_phone", "manager_phone", "receiver_phone",
+    ]
+    description = "11位：以 1 开头，第 2 位 3-9"
+
+    _REGEX = re.compile(r"^1[3-9]\d{9}$")
+    # 详细号段正则（更严格）
+    _DETAILED_REGEX = re.compile(
+        r"^1(3\d|4[5-9]|5[0-35-9]|6[2567]|7[0-8]|8\d|9[0-35-9])\d{8}$"
+    )
+
+    def validate(self, value: str) -> ValidationResult:
+        if not value:
+            return ValidationResult(True, "", value, "空值跳过", self.standard_id)
+
+        # 去除常见格式干扰（空格、横线、+86）
+        v = re.sub(r"[\s\-]", "", value)
+        if v.startswith("+86"):
+            v = v[3:]
+        elif v.startswith("86") and len(v) == 13:
+            v = v[2:]
+
+        if len(v) != 11:
+            return ValidationResult(False, "", value, f"清洗后长度 {len(v)} ≠ 11", self.standard_id)
+
+        if not self._REGEX.match(v):
+            return ValidationResult(False, "", value, "不符合 1[3-9]XXXXXXXXX 格式", self.standard_id)
+
+        if not self._DETAILED_REGEX.match(v):
+            return ValidationResult(False, "", value, "号段不在已知号段表内", self.standard_id)
+
+        return ValidationResult(True, "", value, "", self.standard_id)

standards/std_004_xzqh.py

+"""GB/T 2260 行政区划代码
+
+仅做格式校验（6位或12位全数字），具体编码合法性需对照官方编码表。
+"""
+
+from __future__ import annotations
+
+import re
+
+from .base import BaseStandard, ValidationResult
+
+
+class XzqhStandard(BaseStandard):
+    standard_id = "STD-004"
+    standard_name = "GB/T 2260 行政区划代码"
+    applies_to_fields = ["xzqhbm", "xzqhjb", "adcode", "district_code", "province_code", "city_code", "region_code"]
+    description = "6位标准版或12位扩展版，全数字"
+
+    _REGEX_6 = re.compile(r"^\d{6}$")
+    _REGEX_12 = re.compile(r"^\d{12}$")
+    # 省级代码范围 11-65 + 港澳台
+    _VALID_PROVINCE = set(str(i).zfill(2) for i in list(range(11, 66)) + [71, 81, 82])
+
+    def validate(self, value: str) -> ValidationResult:
+        if not value:
+            return ValidationResult(True, "", value, "空值跳过", self.standard_id)
+
+        if not (self._REGEX_6.match(value) or self._REGEX_12.match(value)):
+            return ValidationResult(False, "", value, "长度必须为 6 或 12 位数字", self.standard_id)
+
+        # 检查省级代码合法性（取前 2 位）
+        province = value[:2]
+        if province not in self._VALID_PROVINCE:
+            return ValidationResult(False, "", value, f"省级代码 {province} 不在 11-65 + 71/81/82 范围", self.standard_id)
+
+        return ValidationResult(True, "", value, "", self.standard_id)

workflow/README.md

+# 数据治理工作流
+
+把 2026-08-03 的数据治理过程抽象成可复用的工作流，支持任意 MySQL 数据库。
+
+## 目录结构
+
+```
+数据治理/
+├── workflow/                         ← 工作流核心
+│   ├── config.yaml                   ← 实际配置（密码走环境变量）
+│   ├── config.example.yaml           ← 配置模板
+│   ├── run_governance.py             ← 主入口
+│   ├── core/
+│   │   ├── db.py                     ← 数据库连接、查询辅助
+│   │   ├── utils.py                  ← 配置加载、目录管理、JSON I/O
+│   │   └── reporter.py               ← 报告生成（Markdown + Word）
+│   └── steps/
+│       ├── base.py                   ← Step 基类
+│       ├── step1_data_dict.py        ← Step 1: 数据字典
+│       ├── step2_merge_redundancy.py ← Step 2: 合并与冗余分析（离线）
+│       ├── step3_verify.py           ← Step 3: 数据验证
+│       ├── step4_empty_fields.py     ← Step 4: 空字段扫描
+│       ├── step5_missing_comments.py ← Step 5: 缺失注释检查（离线）
+│       ├── step6_length_check.py     ← Step 6: 字段长度检查（离线）
+│       ├── step7_standards.py        ← Step 7: 国家标准校验
+│       └── step8_report.py           ← Step 8: 报告生成（离线）
+│
+├── standards/                        ← 国家标准插件（独立、可扩展）
+│   ├── base.py                       ← 标准基类
+│   ├── registry.py                   ← 自动发现所有标准
+│   ├── std_001_id_card.py            ← GB 11643 身份证号
+│   ├── std_002_uscc.py               ← GB 32100 统一社会信用代码
+│   ├── std_003_mobile.py             ← 工信部 手机号
+│   └── std_004_xzqh.py               ← GB/T 2260 行政区划代码
+│
+├── outputs/                          ← 所有运行产出（按数据库+时间戳分目录）
+│   └── <dbname>/
+│       └── <timestamp>/
+│           ├── findings/             ← 各 Step 的 JSON 产出
+│           ├── reports/              ← 最终报告（.md + .docx）
+│           └── run.log
+│
+└── data_dictionary/                  ← 历史脚本与产出（保留作为对比基线）
+    └── ...
+```
+
+## 8 步治理流程
+
+| Step | 名称 | 类型 | 用途 |
+|------|------|------|------|
+| 1 | 数据字典 | 连库 | 抓取所有表/字段的元数据 |
+| 2 | 合并与冗余分析 | **离线** | 基于 Step 1 结果找出可合并表、疑似冗余字段 |
+| 3 | 数据验证 | 连库 | 通过实际查询验证 Step 2 结论 + 检查字典表孤儿编码、字符集 |
+| 4 | 空字段扫描 | 连库 | 找出 ≥80% 空的字段 |
+| 5 | 缺失注释 | **离线** | 找无注释字段，自动推测常见字段含义 |
+| 6 | 字段长度检查 | **离线** | 识别字段定义长度超出标准所需（如身份证用 varchar(50)） |
+| 7 | 国家标准校验 | 连库 | 按字段名匹配标准插件，抽样校验实际数据 |
+| 8 | 报告生成 | **离线** | 汇总所有 findings，生成 Markdown + Word 报告 |
+
+## 快速上手
+
+### 1. 安装依赖
+
+```bash
+pip install pymysql pyyaml python-docx
+```
+
+### 2. 准备配置
+
+```bash
+# 复制配置模板
+cp workflow/config.example.yaml workflow/config.yaml
+
+# 编辑配置（填入数据库连接信息）
+# 推荐用环境变量管理密码：
+#   export MYSQL_PASSWORD=your_password
+```
+
+### 3. 跑流程
+
+```bash
+# 跑完整流程（默认数据库 = 配置文件第一个）
+python workflow/run_governance.py
+
+# 指定数据库
+python workflow/run_governance.py --db smart-build
+
+# 只跑离线分析（不需要数据库）
+python workflow/run_governance.py --steps 2,5,6,8
+
+# 只跑字段长度检查
+python workflow/run_governance.py --steps 6
+
+# 跳过某些步骤
+python workflow/run_governance.py --skip 3,7
+
+# 从已有 JSON 喂入 Step 1 数据（避免重跑数据库，复用历史产出）
+python workflow/run_governance.py --steps 2,5,6,8 --from-json data_dictionary/data_dictionary.json
+```
+
+### 4. 查看产出
+
+```
+outputs/
+└── smart-build/
+    └── 20260804_153022/
+        ├── findings/
+        │   ├── step_1_data_dict.json
+        │   ├── step_2_merge_redundancy.json
+        │   ├── step_3_verify.json
+        │   ├── step_4_empty_fields.json
+        │   ├── step_5_missing_comments.json
+        │   ├── step_6_length_check.json
+        │   ├── step_7_standards.json
+        │   ├── _all_findings.json          ← 汇总
+        │   └── ...
+        ├── reports/
+        │   ├── 数据治理报告_smart-build.md
+        │   └── 数据治理报告_smart-build.docx
+        └── run.log
+```
+
+## 添加新的国家标准
+
+只需要在 `standards/` 下新建一个文件，继承 `BaseStandard`：
+
+```python
+# standards/std_005_email.py
+from standards.base import BaseStandard, ValidationResult
+import re
+
+class EmailStandard(BaseStandard):
+    standard_id = "STD-005"
+    standard_name = "RFC 5322 电子邮件"
+    applies_to_fields = ["email", "e_mail", "mail"]
+    description = "标准邮箱格式"
+
+    _REGEX = re.compile(r"^[\w.+-]+@[\w-]+\.[\w.-]+$")
+
+    def validate(self, value: str) -> ValidationResult:
+        if not value:
+            return ValidationResult(True, "", value, "空值跳过", self.standard_id)
+        if not self._REGEX.match(value):
+            return ValidationResult(False, "", value, "格式错误", self.standard_id)
+        return ValidationResult(True, "", value, "", self.standard_id)
+```
+
+无需修改主流程，重启后 Step 7 会自动发现并应用。
+
+## 扩展自定义 Step
+
+```python
+# workflow/steps/step9_custom.py
+from .base import BaseStep, StepContext
+
+class Step9Custom(BaseStep):
+    name = "9_custom"
+    title = "自定义检查"
+    produces_section = "custom"
+
+    def run(self, ctx: StepContext) -> dict:
+        # 你的逻辑
+        return {"custom_result": [...]}
+```
+
+然后在 `workflow/run_governance.py` 的 `ALL_STEPS` 列表里加上：
+
+```python
+from workflow.steps.step9_custom import Step9Custom
+ALL_STEPS = [..., (Step9Custom(), True)]
+```
+
+## 与历史产出对比
+
+`data_dictionary/` 下保留了 2026-08-03 第一版脚本的产出，可作为对比基线。
+新工作流的产出在 `outputs/<dbname>/<timestamp>/` 下，按时间戳归档。
+
+## 设计原则
+
+1. **配置驱动** — 换数据库只需改配置，无需改代码
+2. **离线/连库分离** — 离线 Step 不连数据库，跑得快、可单独调试
+3. **插件化标准** — 新增标准 = 新建文件，不动主流程
+4. **中间产物 JSON** — 每步产出结构化 JSON，便于二次处理
+5. **多格式报告** — Word（业务方）+ Markdown（编辑）+ JSON（程序）

workflow/__init__.py

+"""数据治理工作流包"""

workflow/config.example.yaml

+# ============================================================
+# 数据治理工作流 — 配置文件模板
+# 复制为 config.yaml 并填入实际值（密码建议用环境变量）
+# ============================================================
+
+# 待治理的数据库（支持多个，一份配置可治理多库）
+databases:
+  smart-build:
+    host: 192.168.20.10
+    port: 3306
+    user: root
+    password: ${MYSQL_PASSWORD}    # 支持 ${ENV_VAR} 占位符
+    dbname: smart-build
+    charset: utf8mb4
+    connect_timeout: 10
+
+# 采样与阈值
+thresholds:
+  # 空字段检查
+  empty_field:
+    high: 0.80                   # >=80% 空值标记为"高空"
+    mid: 0.50                    # >=50% 空值标记为"中空"
+    min_rows_to_check: 5         # 低于此行数的表跳过
+
+  # 表合并相似度
+  similarity:
+    merge_threshold: 0.80        # 字段结构相似度 >= 此值认为可合并
+
+# 输出配置
+output:
+  base_dir: ./outputs           # 产物根目录
+  formats:                      # 报告输出格式
+    - json                      # 结构化中间产物（每步）
+    - markdown                  # 给人看的 Markdown 摘要
+    - docx                      # 给业务方/领导的 Word 报告
+  keep_history: true            # 是否保留历次运行结果（按时间戳归档）
+
+# 标准库校验开关
+standards:
+  enabled: true
+  auto_apply: true              # 自动按字段名匹配并应用对应标准
+  strict_mode: false            # true=严格校验；false=只标记问题不阻断报告生成
+
+# 报告模板（可选，自定义模板放在 templates/ 下）
+report:
+  title: "数据治理报告"
+  organization: ""
+  include_sections:             # 报告包含的章节
+    - overview                  # 概述
+    - merge_candidates          # 表合并候选
+    - redundancy_fields         # 冗余字段
+    - empty_fields              # 大范围空字段
+    - missing_comments          # 缺失注释
+    - length_issues             # 字段长度问题
+    - standard_violations       # 不符合国家标准的字段

workflow/core/__init__.py

+"""工作流核心模块：db / utils / reporter"""

workflow/core/db.py

+"""数据库连接工具
+
+提供统一的 pymysql 连接管理、查询辅助、密码获取能力。
+所有 Step 共享，避免每个脚本重复连接逻辑。
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from contextlib import contextmanager
+from getpass import getpass
+from typing import Any, Iterable
+
+import pymysql
+from pymysql.cursors import DictCursor
+
+
+# ── 密码解析 ────────────────────────────────────────────
+_ENV_PATTERN = re.compile(r"\$\{([A-Z_][A-Z0-9_]*)\}")
+
+
+def resolve_password(value: str) -> str:
+    """把 '${MYSQL_PASSWORD}' 这种占位符替换为环境变量值。
+    未设置时返回原字符串，由调用方决定是否提示输入。"""
+    if not isinstance(value, str):
+        return value
+
+    def _replace(match: re.Match) -> str:
+        env_name = match.group(1)
+        env_value = os.environ.get(env_name)
+        if env_value is None:
+            return match.group(0)  # 保留占位符，后续提示输入
+        return env_value
+
+    return _ENV_PATTERN.sub(_replace, value)
+
+
+def prompt_password(db_config: dict) -> str:
+    """交互式获取密码（不回显）。连接信息仅用于显示。"""
+    host = db_config.get("host", "?")
+    user = db_config.get("user", "?")
+    dbname = db_config.get("database", "?")
+    print("=" * 60)
+    print(f"  连接目标: {user}@{host}")
+    print(f"  数据库:   {dbname}")
+    print("=" * 60)
+    pwd = getpass("请输入数据库密码（输入不显示）: ")
+    if not pwd:
+        raise RuntimeError("密码不能为空")
+    return pwd
+
+
+# ── 连接 ────────────────────────────────────────────────
+def connect(db_config: dict, password: str | None = None) -> pymysql.connections.Connection:
+    """根据配置建立连接。密码可显式传入，也可从 ${ENV} 自动解析，缺失则交互式询问。"""
+    cfg = dict(db_config)
+    pwd = cfg.pop("password", None) or password
+
+    # 解析 ${ENV} 占位符
+    if isinstance(pwd, str):
+        pwd = resolve_password(pwd)
+
+    # 仍未拿到密码则询问
+    if not pwd:
+        pwd = prompt_password(cfg)
+
+    try:
+        conn = pymysql.connect(password=pwd, cursorclass=DictCursor, **cfg)
+        print("✓ 数据库连接成功")
+        return conn
+    except pymysql.err.OperationalError as e:
+        raise RuntimeError(f"无法连接数据库: {e}") from e
+
+
+@contextmanager
+def connection(db_config: dict, password: str | None = None):
+    """上下文管理器，自动关闭连接。"""
+    conn = connect(db_config, password)
+    try:
+        yield conn
+    finally:
+        conn.close()
+        print("数据库连接已关闭。")
+
+
+# ── 查询辅助 ────────────────────────────────────────────
+def query_all(cursor, sql: str, params: Iterable | None = None) -> list[dict[str, Any]]:
+    cursor.execute(sql, params)
+    return cursor.fetchall()
+
+
+def query_one(cursor, sql: str, params: Iterable | None = None) -> dict[str, Any] | None:
+    cursor.execute(sql, params)
+    row = cursor.fetchone()
+    return row
+
+
+def query_scalar(cursor, sql: str, params: Iterable | None = None) -> Any:
+    cursor.execute(sql, params)
+    row = cursor.fetchone()
+    return row[0] if row else None
+
+
+# ── JSON 序列化兜底 ────────────────────────────────────
+def json_serializer(obj: Any) -> Any:
+    if hasattr(obj, "isoformat"):
+        return obj.isoformat()
+    if isinstance(obj, bytes):
+        return obj.decode("utf-8", errors="replace")
+    if isinstance(obj, set):
+        return sorted(obj)
+    return str(obj)

workflow/core/utils.py

+"""通用工具：配置加载、目录管理、日志、JSON I/O"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from .db import json_serializer
+
+# ── 配置加载 ────────────────────────────────────────────
+def load_config(config_path: str | os.PathLike) -> dict:
+    """加载 YAML 配置文件"""
+    path = Path(config_path)
+    if not path.exists():
+        raise FileNotFoundError(f"配置文件不存在: {path}")
+    with open(path, encoding="utf-8") as f:
+        cfg = yaml.safe_load(f) or {}
+    return cfg
+
+
+def get_db_config(cfg: dict, db_name: str) -> dict:
+    """从配置中取指定数据库的连接配置"""
+    dbs = cfg.get("databases", {})
+    if db_name not in dbs:
+        available = ", ".join(dbs.keys())
+        raise KeyError(f"配置中找不到数据库 '{db_name}'，可选: {available}")
+    return dbs[db_name]
+
+
+# ── 输出目录 ────────────────────────────────────────────
+def output_paths(cfg: dict, db_name: str, step_name: str) -> dict[str, Path]:
+    """生成某一步的输出路径（json / md / log）。"""
+    base = Path(cfg.get("output", {}).get("base_dir", "./outputs"))
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    run_dir = base / db_name / ts
+    findings_dir = run_dir / "findings"
+    findings_dir.mkdir(parents=True, exist_ok=True)
+
+    return {
+        "run_dir": run_dir,
+        "findings_dir": findings_dir,
+        "json": findings_dir / f"step_{step_name}.json",
+        "markdown": findings_dir / f"step_{name_clean(step_name)}.md",
+        "log": run_dir / "run.log",
+    }
+
+
+def name_clean(name: str) -> str:
+    """把 step 名规范成文件友好形式"""
+    return re.sub(r"[^a-z0-9_]+", "_", name.lower()).strip("_")
+
+
+# ── JSON I/O ────────────────────────────────────────────
+def save_json(data: Any, filepath: os.PathLike) -> Path:
+    p = Path(filepath)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    with open(p, "w", encoding="utf-8") as f:
+        json.dump(data, f, ensure_ascii=False, indent=2, default=json_serializer)
+    return p
+
+
+def load_json(filepath: os.PathLike) -> Any:
+    with open(filepath, encoding="utf-8") as f:
+        return json.load(f)
+
+
+# ── 日志 ────────────────────────────────────────────────
+class StepLogger:
+    """带文件输出的简易 logger。每个 step 一个实例。"""
+
+    def __init__(self, log_file: os.PathLike | None = None):
+        self.log_file = Path(log_file) if log_file else None
+        if self.log_file:
+            self.log_file.parent.mkdir(parents=True, exist_ok=True)
+
+    def log(self, msg: str, *, level: str = "INFO") -> None:
+        ts = datetime.now().strftime("%H:%M:%S")
+        line = f"[{ts}] [{level}] {msg}"
+        print(line)
+        if self.log_file:
+            with open(self.log_file, "a", encoding="utf-8") as f:
+                f.write(line + "\n")
+
+    def section(self, title: str) -> None:
+        bar = "─" * 60
+        self.log(f"\n{bar}\n{title}\n{bar}")
+
+    def ok(self, msg: str) -> None:
+        self.log(f"✓ {msg}")
+
+    def warn(self, msg: str) -> None:
+        self.log(f"⚠ {msg}", level="WARN")
+
+    def error(self, msg: str) -> None:
+        self.log(f"✗ {msg}", level="ERROR")

workflow/run_governance.py

+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+数据治理工作流主入口
+
+用法：
+  # 跑完整流程（默认数据库、全部步骤）
+  python workflow/run_governance.py
+
+  # 指定数据库
+  python workflow/run_governance.py --db smart-build
+
+  # 只跑某几步
+  python workflow/run_governance.py --steps 1,4,7
+
+  # 跳过某些步骤
+  python workflow/run_governance.py --skip 3,7
+
+  # 指定配置文件
+  python workflow/run_governance.py --config workflow/config.yaml
+
+  # 干跑模式（不连数据库，只跑离线分析步骤）
+  python workflow/run_governance.py --steps 2,5,6 --offline
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+import time
+import traceback
+from pathlib import Path
+
+# 把项目根目录加入 sys.path
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from workflow.core.utils import load_config, get_db_config, output_paths, StepLogger
+from workflow.steps.base import StepContext
+from workflow.steps.step1_data_dict import Step1DataDict
+from workflow.steps.step2_merge_redundancy import Step2MergeRedundancy
+from workflow.steps.step3_verify import Step3Verify
+from workflow.steps.step4_empty_fields import Step4EmptyFields
+from workflow.steps.step5_missing_comments import Step5MissingComments
+from workflow.steps.step6_length_check import Step6LengthCheck
+from workflow.steps.step7_standards import Step7Standards
+from workflow.steps.step8_report import Step8Report
+
+
+# ── Step 注册表 ──
+# offline=True 表示此 Step 不连数据库，可独立运行
+ALL_STEPS = [
+    (Step1DataDict(), True),
+    (Step2MergeRedundancy(), True),       # 离线分析
+    (Step3Verify(), False),
+    (Step4EmptyFields(), False),
+    (Step5MissingComments(), True),       # 离线
+    (Step6LengthCheck(), True),           # 离线
+    (Step7Standards(), False),
+    (Step8Report(), True),                # 离线
+]
+
+
+def main():
+    parser = argparse.ArgumentParser(description="数据治理工作流")
+    parser.add_argument("--config", default=str(PROJECT_ROOT / "workflow" / "config.yaml"))
+    parser.add_argument("--db", default=None, help="要治理的数据库名（默认取配置第一个）")
+    parser.add_argument("--steps", default=None, help="要运行的步骤编号，逗号分隔（如 1,4,7）")
+    parser.add_argument("--skip", default=None, help="要跳过的步骤编号，逗号分隔")
+    parser.add_argument("--offline", action="store_true", help="干跑模式：只跑离线步骤")
+    parser.add_argument("--from-json", dest="from_json", default=None,
+                        help="从已有 JSON 加载 Step 1 数据（避免重新跑数据库）")
+    args = parser.parse_args()
+
+    # 1. 加载配置
+    cfg = load_config(args.config)
+    db_name = args.db or next(iter(cfg.get("databases", {})))
+    db_config = get_db_config(cfg, db_name)
+
+    print("=" * 60)
+    print(f"  数据治理工作流")
+    print(f"  数据库: {db_name}")
+    print(f"  配置:   {args.config}")
+    print("=" * 60)
+
+    # 2. 确定要跑的步骤
+    wanted = set(int(x) for x in args.steps.split(",")) if args.steps else None
+    skipped = set(int(x) for x in args.skip.split(",")) if args.skip else set()
+
+    # 3. 准备运行目录
+    paths = output_paths(cfg, db_name, "all")
+    logger = StepLogger(paths["log"])
+
+    # 4. 顺序执行
+    step_outputs: dict = {}
+    findings: dict = {}
+
+    # 4.0 如果指定了 --from-json，先加载作为 Step 1 的输入
+    if args.from_json:
+        from workflow.core.utils import load_json
+        from_json_path = Path(args.from_json)
+        dict_data = load_json(from_json_path)
+        # 兼容两种结构：纯 dict_data 列表，或者带 meta 的 dict
+        if isinstance(dict_data, dict) and "data_dictionary" in dict_data:
+            data_list = dict_data["data_dictionary"]
+            summary = dict_data.get("table_summary", [])
+        else:
+            data_list = dict_data
+            summary = []
+        step_outputs["1_data_dict"] = {
+            "data_dictionary": data_list,
+            "table_summary": summary,
+            "overview_meta": {
+                "database": db_name,
+                "host": db_config.get("host", ""),
+                "total_tables": len(set(r["table_name"] for r in data_list)),
+                "total_fields": len(data_list),
+            },
+        }
+        logger.log(f"已从 JSON 加载 Step 1 数据: {len(data_list)} 个字段")
+
+    start = time.time()
+
+    for step_num, (step, offline_ok) in enumerate(ALL_STEPS, 1):
+        if wanted and step_num not in wanted:
+            continue
+        if step_num in skipped:
+            logger.log(f"[Step {step_num}] {step.title} — 跳过")
+            continue
+        if args.offline and not offline_ok:
+            logger.log(f"[Step {step_num}] {step.title} — 离线模式跳过")
+            continue
+
+        logger.section(f"Step {step_num}: {step.title}")
+        ctx = StepContext(
+            cfg=cfg,
+            db_name=db_name,
+            db_config=db_config,
+            logger=logger,
+            paths=paths,
+            findings=findings,
+            step_outputs=step_outputs,
+        )
+        try:
+            output = step.run(ctx)
+            # 把产出存到 step_outputs
+            step_outputs[step.name] = output
+            # 把 section 数据也聚合到 findings（供报告使用）
+            if step.produces_section:
+                findings[step.produces_section] = output
+            # 把 Step 自身的 JSON 持久化
+            step.save(ctx, output)
+            logger.ok(f"[Step {step_num}] 完成")
+        except Exception as e:
+            logger.error(f"[Step {step_num}] 失败: {e}")
+            if wanted:  # 如果是用户指定的步骤，失败就中断
+                raise
+            traceback.print_exc()
+
+    elapsed = time.time() - start
+    print("\n" + "=" * 60)
+    print(f"  ✓ 流程完成，耗时 {elapsed:.1f}s")
+    print(f"  产出目录: {paths['run_dir']}")
+    print(f"  日志: {paths['log']}")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    main()

workflow/steps/__init__.py

+"""各 Step 实现。所有 Step 继承 BaseStep。"""

workflow/steps/base.py

+"""Step 基类。
+
+每个 Step 必须实现：
+  - name:        步骤标识（如 "1_data_dict"）
+  - title:       步骤中文标题
+  - run(ctx):    主逻辑，ctx 包含 cfg/db_config/logger/paths
+
+Step 之间通过 ctx.findings 共享数据。
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+from pathlib import Path
+
+import pymysql
+
+from ..core.db import connect, json_serializer
+from ..core.utils import StepLogger, output_paths, save_json
+from ..core.reporter import MarkdownBuilder
+
+
+@dataclass
+class StepContext:
+    """运行时上下文，跨 Step 传递"""
+    cfg: dict
+    db_name: str
+    db_config: dict
+    logger: StepLogger
+    paths: dict
+    findings: dict  # 累计的各 Step 发现，key=section 名（如 "merge_candidates"）
+    step_outputs: dict  # 原始输出，key=step name，value=StepResult 写入的 dict
+
+    def get_section_data(self, section: str) -> dict:
+        return self.findings.get(section, {})
+
+
+class BaseStep(ABC):
+    name: str = ""
+    title: str = ""
+    produces_section: str = ""  # 此 Step 产出的 section 名（用于汇总到 findings）
+
+    @abstractmethod
+    def run(self, ctx: StepContext) -> dict:
+        """返回此 Step 的结构化输出（dict），会自动写入 JSON"""
+        ...
+
+    # ── 辅助方法 ──
+    def connect_db(self, ctx: StepContext) -> pymysql.connections.Connection:
+        return connect(ctx.db_config)
+
+    def save(self, ctx: StepContext, data: dict) -> Path:
+        """把 Step 输出写入 JSON 文件"""
+        enriched = {
+            "meta": {
+                "step": self.name,
+                "title": self.title,
+                "database": ctx.db_name,
+                "executed_at": datetime.now().isoformat(),
+            },
+            **data,
+        }
+        return save_json(enriched, ctx.paths["json"])
+
+    def log(self, ctx: StepContext, msg: str) -> None:
+        ctx.logger.log(f"[{self.name}] {msg}")

workflow/steps/step1_data_dict.py

+"""Step 1: 获取数据字典
+
+输出：
+  - data_dictionary.json: 全部字段详情（INFORMATION_SCHEMA.COLUMNS）
+  - table_summary.json:   表级别汇总（INFORMATION_SCHEMA.TABLES）
+
+产出 section: overview（overview_meta）
+"""
+
+from __future__ import annotations
+
+from .base import BaseStep, StepContext
+
+
+class Step1DataDict(BaseStep):
+    name = "1_data_dict"
+    title = "获取数据字典"
+    produces_section = "overview_meta"
+
+    def run(self, ctx: StepContext) -> dict:
+        self.log(ctx, "开始获取数据字典...")
+        conn = self.connect_db(ctx)
+        try:
+            with conn.cursor() as cur:
+                columns = self._fetch_columns(cur, ctx.db_config["database"])
+                tables = self._fetch_tables(cur, ctx.db_config["database"])
+
+            # 汇总
+            table_names = set(r["table_name"] for r in columns)
+            ctx.logger.ok(
+                f"获取完成：{len(table_names)} 张表，{len(columns)} 个字段"
+            )
+
+            return {
+                "data_dictionary": columns,
+                "table_summary": tables,
+                "overview_meta": {
+                    "database": ctx.db_name,
+                    "host": ctx.db_config.get("host", ""),
+                    "total_tables": len(table_names),
+                    "total_fields": len(columns),
+                },
+            }
+        finally:
+            conn.close()
+
+    # ── 查询 ──
+    @staticmethod
+    def _fetch_columns(cur, schema: str) -> list[dict]:
+        sql = """
+            SELECT
+                c.TABLE_NAME                                       AS table_name,
+                c.COLUMN_NAME                                      AS column_name,
+                c.ORDINAL_POSITION                                 AS ordinal_position,
+                c.COLUMN_TYPE                                      AS column_type,
+                c.DATA_TYPE                                        AS data_type,
+                c.CHARACTER_MAXIMUM_LENGTH                         AS char_max_length,
+                c.NUMERIC_PRECISION                                AS numeric_precision,
+                c.NUMERIC_SCALE                                    AS numeric_scale,
+                c.IS_NULLABLE                                      AS is_nullable,
+                c.COLUMN_DEFAULT                                   AS column_default,
+                c.COLUMN_COMMENT                                   AS column_comment,
+                c.EXTRA                                            AS extra,
+                t.TABLE_COMMENT                                    AS table_comment
+            FROM INFORMATION_SCHEMA.COLUMNS c
+            LEFT JOIN INFORMATION_SCHEMA.TABLES t
+                ON c.TABLE_SCHEMA = t.TABLE_SCHEMA
+               AND c.TABLE_NAME   = t.TABLE_NAME
+            WHERE c.TABLE_SCHEMA = %s
+            ORDER BY c.TABLE_NAME, c.ORDINAL_POSITION
+        """
+        cur.execute(sql, (schema,))
+        rows = cur.fetchall()
+        # 类型转换
+        for r in rows:
+            for k in ("char_max_length", "numeric_precision", "numeric_scale", "ordinal_position"):
+                if r.get(k) is not None:
+                    r[k] = int(r[k])
+        return rows
+
+    @staticmethod
+    def _fetch_tables(cur, schema: str) -> list[dict]:
+        sql = """
+            SELECT TABLE_NAME, TABLE_TYPE, ENGINE, TABLE_ROWS,
+                   DATA_LENGTH, INDEX_LENGTH, TABLE_COMMENT,
+                   CREATE_TIME, UPDATE_TIME
+            FROM INFORMATION_SCHEMA.TABLES
+            WHERE TABLE_SCHEMA = %s
+            ORDER BY TABLE_NAME
+        """
+        cur.execute(sql, (schema,))
+        rows = cur.fetchall()
+        for r in rows:
+            for tf in ("CREATE_TIME", "UPDATE_TIME"):
+                if r.get(tf) is not None and hasattr(r[tf], "strftime"):
+                    r[tf] = r[tf].strftime("%Y-%m-%d %H:%M:%S")
+            for nf in ("TABLE_ROWS", "DATA_LENGTH", "INDEX_LENGTH"):
+                if r.get(nf) is not None:
+                    r[nf] = int(r[nf])
+        return rows

workflow/steps/step2_merge_redundancy.py

+"""Step 2: 表合并与冗余字段分析（离线分析）
+
+完全基于 Step 1 产出的 data_dictionary，不需要再查数据库。
+- 识别结构相同的表（可合并候选）
+- 识别结构完全为空且无新表替代的废弃候选
+- 统计高频出现的字段（疑似冗余）
+
+产出 section: merge_candidates
+"""
+
+from __future__ import annotations
+
+from collections import Counter, defaultdict
+
+from .base import BaseStep, StepContext
+
+
+class Step2MergeRedundancy(BaseStep):
+    name = "2_merge_redundancy"
+    title = "表合并与冗余字段分析"
+    produces_section = "merge_candidates"
+
+    def run(self, ctx: StepContext) -> dict:
+        # 依赖 Step 1 的产出；如果没有则自己连库拉一次
+        dict_data = ctx.step_outputs.get("1_data_dict", {}).get("data_dictionary", [])
+        table_summary = ctx.step_outputs.get("1_data_dict", {}).get("table_summary", [])
+        if not dict_data:
+            self.log(ctx, "未找到 Step 1 产出，自动连库获取...")
+            from .step1_data_dict import Step1DataDict
+            ctx.step_outputs["1_data_dict"] = Step1DataDict().run(ctx)
+            dict_data = ctx.step_outputs["1_data_dict"].get("data_dictionary", [])
+            table_summary = ctx.step_outputs["1_data_dict"].get("table_summary", [])
+        if not dict_data:
+            self.log(ctx, "仍未拿到数据，跳过")
+            return {"merge_candidates": [], "decommission_candidates": []}
+
+        # 按表分组
+        by_table: dict[str, list[dict]] = defaultdict(list)
+        for row in dict_data:
+            by_table[row["table_name"]].append(row)
+
+        # 1. 表合并候选
+        merge_candidates = self._find_merge_candidates(by_table)
+
+        # 2. 废弃候选
+        decommission = self._find_decommission_candidates(by_table, table_summary)
+
+        # 3. 高频字段（疑似冗余）
+        redundancy = self._find_redundancy_fields(by_table)
+
+        self.log(
+            ctx,
+            f"发现 {len(merge_candidates)} 组可合并候选, "
+            f"{decommission['strong']['count'] + decommission['weak']['count']} 张废弃候选, "
+            f"{len(redundancy)} 个高频字段",
+        )
+
+        return {
+            "merge_candidates": merge_candidates,
+            "decommission_candidates": decommission,
+            "redundancy_fields": redundancy,
+        }
+
+    # ── 工具方法 ──
+    def _table_columns(self, columns: list[dict]) -> set[str]:
+        return {c["column_name"] for c in columns}
+
+    def _similarity(self, cols_a: set[str], cols_b: set[str]) -> float:
+        if not cols_a or not cols_b:
+            return 0.0
+        return len(cols_a & cols_b) / len(cols_a | cols_b)
+
+    def _find_merge_candidates(self, by_table: dict) -> list[dict]:
+        """找结构高度相似的表"""
+        table_names = sorted(by_table.keys())
+        groups: list[list[str]] = []
+        visited: set[str] = set()
+
+        for i, t in enumerate(table_names):
+            if t in visited:
+                continue
+            cols_i = self._table_columns(by_table[t])
+            cluster = [t]
+            for j in range(i + 1, len(table_names)):
+                t2 = table_names[j]
+                if t2 in visited:
+                    continue
+                cols_j = self._table_columns(by_table[t2])
+                sim = self._similarity(cols_i, cols_j)
+                if sim >= 0.95:  # 95% 以上视为结构相同
+                    cluster.append(t2)
+            if len(cluster) >= 2:
+                visited.update(cluster)
+                groups.append(cluster)
+
+        # 转成 findings 格式
+        results = []
+        for idx, group in enumerate(groups, 1):
+            ref_cols = self._table_columns(by_table[group[0]])
+            results.append({
+                "id": f"MERGE-{idx:03d}",
+                "type": "结构高度相似",
+                "tables": group,
+                "common_columns": len(ref_cols),
+                "similarity": "≥95%",
+                "columns": sorted(ref_cols),
+                "suggestion": self._suggest_merge(group),
+            })
+        return results
+
+    def _suggest_merge(self, group: list[str]) -> str:
+        if not group:
+            return "合并候选"
+        if len(group) == 2:
+            base = group[0]
+            return f"合并为 {base}，新增类型字段区分"
+        if len(group) <= 5:
+            return f"合并为 {group[0]}，新增 record_type / type 字段区分"
+        return f"合并 {len(group)} 张同类表"
+
+    def _find_decommission_candidates(
+        self, by_table: dict, table_summary: list[dict]
+    ) -> dict:
+        """找 0 行表"""
+        rows_by_table = {t["TABLE_NAME"]: t.get("TABLE_ROWS", 0) for t in table_summary}
+        zero_rows = [t for t, n in rows_by_table.items() if n == 0 and t in by_table]
+
+        # 弱/强分类：是否有同前缀的新版表
+        strong, weak = [], []
+        for t in sorted(zero_rows):
+            replacement = self._find_replacement(t, rows_by_table.keys())
+            entry = {"table": t, "replacement": replacement}
+            if replacement:
+                strong.append(entry)
+            else:
+                weak.append(entry)
+
+        return {
+            "strong": {"description": "有新版表替代", "tables": strong, "count": len(strong)},
+            "weak": {"description": "无明确替代", "tables": weak, "count": len(weak)},
+        }
+
+    def _find_replacement(self, table: str, all_tables) -> str | None:
+        """启发式：t_xxx 是否被 t_talent_xxx 或 t_ai_xxx 等替代"""
+        prefixes = ["t_", "mall_", "project_"]
+        for prefix in prefixes:
+            if table.startswith(prefix):
+                stem = table[len(prefix):]
+                # 尝试常见新前缀
+                for new_prefix in ["t_talent_", "t_ai_", "t_design_"]:
+                    cand = new_prefix + stem
+                    if cand in all_tables and cand != table:
+                        return cand
+        return None
+
+    def _find_redundancy_fields(self, by_table: dict) -> list[dict]:
+        """统计字段出现频次，频次过高视为疑似冗余
+
+        通用字段（id / create_time / update_time / 状态类）几乎每张表都有，
+        不算"冗余"，过滤掉。
+        """
+        # 通用字段白名单
+        COMMON = {
+            "id", "create_time", "update_time", "create_by", "update_by",
+            "remark", "del_flag", "tenant_id", "dept_id", "status",
+            "sort_order", "version",
+        }
+        field_counter: Counter = Counter()
+        for cols in by_table.values():
+            for c in cols:
+                field_counter[c["column_name"]] += 1
+
+        # 出现 >= 10 次 且不在通用集合里
+        suspects = [
+            {"field": f, "table_count": n}
+            for f, n in field_counter.most_common()
+            if n >= 10 and f not in COMMON
+        ]
+        return suspects

workflow/steps/step3_verify.py

+"""Step 3: 数据验证（需连数据库）
+
+通过实际查询验证 Step 2 的分析结论：
+- 合并表的实际数据量
+- 高频冗余字段的填充率 / 一致性
+- 行政区划字典表的孤儿编码
+
+产出 section: redundancy_fields + data_quality
+"""
+
+from __future__ import annotations
+
+from .base import BaseStep, StepContext
+from ..core.db import query_all, query_scalar
+
+
+class Step3Verify(BaseStep):
+    name = "3_verify"
+    title = "数据验证（连库查询）"
+    produces_section = "redundancy_fields"
+
+    def run(self, ctx: StepContext) -> dict:
+        self.log(ctx, "开始数据验证...")
+
+        merge_out = ctx.step_outputs.get("2_merge_redundancy", {})
+        merge_candidates = merge_out.get("merge_candidates", [])
+
+        conn = self.connect_db(ctx)
+        try:
+            with conn.cursor() as cur:
+                # 1. 验证合并表的实际行数
+                confirmed_merge = self._verify_merge(cur, merge_candidates)
+
+                # 2. 验证高频字段（取 Top 5）
+                redundancy_out = merge_out.get("redundancy_fields", [])
+                redundancy_findings = self._verify_redundancy(cur, redundancy_out[:5])
+
+                # 3. 行政区划字典表孤儿编码（如果有 xzqhbm 字段的话）
+                xzqh_orphans = self._verify_xzqh(cur)
+
+                # 4. 字符集不一致检查
+                charset_issues = self._check_charset(cur)
+
+            self.log(
+                ctx,
+                f"验证完成：合并候选 {len(confirmed_merge)}, "
+                f"冗余字段 {len(redundancy_findings)}, "
+                f"行政区划孤儿 {xzqh_orphans.get('orphan_total', 0)}, "
+                f"字符集异常 {len(charset_issues)}",
+            )
+
+            return {
+                "confirmed_merge_candidates": confirmed_merge,
+                "redundancy_fields": redundancy_findings,
+                "xzqh_orphan_check": xzqh_orphans,
+                "data_quality_issues": [
+                    {"id": "DQ-001", "type": "字符集不一致", "tables": charset_issues},
+                ] if charset_issues else [],
+            }
+        finally:
+            conn.close()
+
+    # ── 验证合并候选 ──
+    def _verify_merge(self, cur, candidates: list[dict]) -> list[dict]:
+        out = []
+        for c in candidates:
+            counts = []
+            for t in c["tables"]:
+                try:
+                    n = query_scalar(cur, f"SELECT COUNT(*) FROM `{t}`")
+                    counts.append({"table": t, "rows": n})
+                except Exception as e:
+                    counts.append({"table": t, "error": str(e)})
+            total = sum(x.get("rows", 0) for x in counts if isinstance(x.get("rows"), int))
+            out.append({
+                **c,
+                "row_counts": counts,
+                "total_rows": total,
+            })
+        return out
+
+    # ── 验证冗余字段 ──
+    def _verify_redundancy(self, cur, fields: list[dict]) -> list[dict]:
+        out = []
+        for f in fields:
+            field_name = f["field"]
+            # 找所有含该字段的表
+            tables_with_field = query_all(
+                cur,
+                """
+                SELECT TABLE_NAME FROM INFORMATION_SCHEMA.COLUMNS
+                WHERE TABLE_SCHEMA = DATABASE() AND COLUMN_NAME = %s
+                """,
+                (field_name,),
+            )
+            out.append({
+                "field": field_name,
+                "occurrences": f["table_count"],
+                "tables_with_field": [t["TABLE_NAME"] for t in tables_with_field],
+            })
+        return out
+
+    # ── 行政区划孤儿编码 ──
+    def _verify_xzqh(self, cur) -> dict:
+        try:
+            dict_count = query_scalar(cur, "SELECT COUNT(*) FROM c_bri_xzqh")
+        except Exception:
+            return {"orphan_total": 0, "note": "c_bri_xzqh 表不存在，跳过"}
+
+        # 找含 xzqhbm 的表
+        tables = query_all(
+            cur,
+            """
+            SELECT c.TABLE_NAME, t.TABLE_ROWS
+            FROM INFORMATION_SCHEMA.COLUMNS c
+            JOIN INFORMATION_SCHEMA.TABLES t
+              ON c.TABLE_SCHEMA = t.TABLE_SCHEMA AND c.TABLE_NAME = t.TABLE_NAME
+            WHERE c.TABLE_SCHEMA = DATABASE() AND c.COLUMN_NAME = 'xzqhbm' AND t.TABLE_ROWS > 0
+            """,
+        )
+
+        orphan_total = 0
+        per_table = []
+        for t in tables[:5]:  # 只查前 5 张大表避免太慢
+            try:
+                orphans = query_all(
+                    cur,
+                    f"""
+                    SELECT g.xzqhbm FROM `{t['TABLE_NAME']}` g
+                    LEFT JOIN c_bri_xzqh x ON g.xzqhbm = x.xzqhbm
+                    WHERE g.xzqhbm IS NOT NULL AND g.xzqhbm != '' AND x.xzqhbm IS NULL
+                    LIMIT 100
+                    """,
+                )
+                per_table.append({
+                    "table": t["TABLE_NAME"],
+                    "orphan_count": len(orphans),
+                    "sample": orphans[:5],
+                })
+                orphan_total += len(orphans)
+            except Exception as e:
+                per_table.append({"table": t["TABLE_NAME"], "error": str(e)})
+
+        return {
+            "dict_table_rows": dict_count,
+            "orphan_total": orphan_total,
+            "per_table": per_table,
+        }
+
+    # ── 字符集检查 ──
+    def _check_charset(self, cur) -> list[dict]:
+        rows = query_all(
+            cur,
+            """
+            SELECT TABLE_NAME, TABLE_COLLATION
+            FROM INFORMATION_SCHEMA.TABLES
+            WHERE TABLE_SCHEMA = DATABASE() AND TABLE_COLLATION IS NOT NULL
+            GROUP BY TABLE_COLLATION
+            """,
+        )
+        collations = {r["TABLE_COLLATION"] for r in rows}
+        if len(collations) > 1:
+            return [{"collation": c, "tables": [r["TABLE_NAME"] for r in rows if r["TABLE_COLLATION"] == c][:5]}
+                    for c in collations]
+        return []

workflow/steps/step4_empty_fields.py

+"""Step 4: 大范围空字段扫描
+
+逐表扫描所有字段，统计 NULL 率 + 空字符串率，识别填充率低的字段。
+
+产出 section: empty_fields
+"""
+
+from __future__ import annotations
+
+from .base import BaseStep, StepContext
+from ..core.db import query_all, query_scalar
+
+
+class Step4EmptyFields(BaseStep):
+    name = "4_empty_fields"
+    title = "大范围空字段扫描"
+    produces_section = "empty_fields"
+
+    def run(self, ctx: StepContext) -> dict:
+        thresholds = ctx.cfg.get("thresholds", {}).get("empty_field", {})
+        high_th = thresholds.get("high", 0.8)
+        mid_th = thresholds.get("mid", 0.5)
+        min_rows = thresholds.get("min_rows_to_check", 5)
+
+        self.log(ctx, f"阈值：高空>={int(high_th*100)}%, 中空>={int(mid_th*100)}%, 最小行数={min_rows}")
+
+        conn = self.connect_db(ctx)
+        try:
+            with conn.cursor() as cur:
+                # 1. 取所有表的列信息
+                cols_info = self._get_all_columns(cur, ctx.db_config["database"])
+
+                # 2. 按表分组
+                by_table: dict[str, list[dict]] = {}
+                for c in cols_info:
+                    by_table.setdefault(c["TABLE_NAME"], []).append(c)
+
+                # 3. 扫描
+                all_results = []
+                skipped = []
+                per_table = []
+                total = len(by_table)
+
+                for idx, (tname, cols) in enumerate(by_table.items(), 1):
+                    actual = query_scalar(cur, f"SELECT COUNT(*) FROM `{tname}`") or 0
+                    if actual < min_rows:
+                        skipped.append({"table": tname, "rows": actual})
+                        continue
+
+                    # 单 SQL 统计所有列
+                    sql, col_order = self._build_sql(tname, cols)
+                    try:
+                        row = query_all(cur, sql)
+                        if not row:
+                            continue
+                        stats = row[0]
+                        table_issues = []
+                        for cn in col_order:
+                            ek = f"{cn}__empty"
+                            nk = f"{cn}__null"
+                            if ek in stats:
+                                ec, label = stats[ek] or 0, "NULL或空字符串"
+                            elif nk in stats:
+                                ec, label = stats[nk] or 0, "NULL"
+                            else:
+                                continue
+                            rate = round(ec / actual, 4)
+                            col_info = next((c for c in cols if c["COLUMN_NAME"] == cn), {})
+                            fr = {
+                                "table_name": tname,
+                                "column_name": cn,
+                                "data_type": col_info.get("DATA_TYPE", ""),
+                                "char_max_length": col_info.get("CHARACTER_MAXIMUM_LENGTH"),
+                                "column_comment": col_info.get("COLUMN_COMMENT", ""),
+                                "total_rows": actual,
+                                "empty_count": ec,
+                                "empty_rate": rate,
+                                "level": "高空" if rate >= high_th else ("中空" if rate >= mid_th else "正常"),
+                            }
+                            all_results.append(fr)
+                            if rate >= mid_th:
+                                table_issues.append(fr)
+
+                        if table_issues:
+                            per_table.append({
+                                "table_name": tname,
+                                "table_comment": cols[0].get("TABLE_COMMENT", ""),
+                                "total_rows": actual,
+                                "high_empty_fields": sum(1 for f in table_issues if f["level"] == "高空"),
+                                "mid_empty_fields": sum(1 for f in table_issues if f["level"] == "中空"),
+                            })
+                    except Exception as e:
+                        skipped.append({"table": tname, "error": str(e)})
+
+                    if idx % 20 == 0:
+                        self.log(ctx, f"  进度 {idx}/{total}")
+
+                # 4. 汇总
+                summary = self._build_summary(all_results, per_table, skipped, high_th, mid_th, min_rows)
+                self.log(
+                    ctx,
+                    f"扫描完成：{summary['high_empty_fields_count']} 高空 + {summary['mid_empty_fields_count']} 中空",
+                )
+
+                return {
+                    "summary": summary,
+                    "per_table": per_table,
+                    "high_empty_fields": [f for f in all_results if f["level"] == "高空"],
+                }
+        finally:
+            conn.close()
+
+    # ── 工具 ──
+    @staticmethod
+    def _get_all_columns(cur, schema: str) -> list[dict]:
+        sql = """
+            SELECT c.TABLE_NAME, c.COLUMN_NAME, c.DATA_TYPE,
+                   c.CHARACTER_MAXIMUM_LENGTH, c.IS_NULLABLE, c.COLUMN_COMMENT,
+                   c.ORDINAL_POSITION, t.TABLE_ROWS, t.TABLE_COMMENT
+            FROM INFORMATION_SCHEMA.COLUMNS c
+            JOIN INFORMATION_SCHEMA.TABLES t
+              ON c.TABLE_SCHEMA = t.TABLE_SCHEMA AND c.TABLE_NAME = t.TABLE_NAME
+            WHERE c.TABLE_SCHEMA = %s AND t.TABLE_TYPE = 'BASE TABLE'
+            ORDER BY c.TABLE_NAME, c.ORDINAL_POSITION
+        """
+        return query_all(cur, sql, (schema,))
+
+    @staticmethod
+    def _build_sql(table: str, cols: list[dict]) -> tuple[str, list[str]]:
+        string_types = {"varchar", "char", "text", "longtext", "mediumtext", "tinytext", "enum", "set"}
+        exprs, order = [], []
+        for c in cols:
+            cn = c["COLUMN_NAME"]
+            dt = (c["DATA_TYPE"] or "").lower()
+            order.append(cn)
+            if dt in string_types:
+                exprs.append(
+                    f"SUM(CASE WHEN `{cn}` IS NULL OR `{cn}` = '' THEN 1 ELSE 0 END) AS `{cn}__empty`"
+                )
+            else:
+                exprs.append(f"SUM(CASE WHEN `{cn}` IS NULL THEN 1 ELSE 0 END) AS `{cn}__null`")
+        sql = f"SELECT COUNT(*) AS __total__, {', '.join(exprs)} FROM `{table}`"
+        return sql, order
+
+    @staticmethod
+    def _build_summary(all_fields, per_table, skipped, high_th, mid_th, min_rows) -> dict:
+        high = [f for f in all_fields if f["level"] == "高空"]
+        mid = [f for f in all_fields if f["level"] == "中空"]
+        # Top 20
+        high_by_table = {}
+        for f in high:
+            high_by_table[f["table_name"]] = high_by_table.get(f["table_name"], 0) + 1
+        top_high = sorted(high_by_table.items(), key=lambda x: x[1], reverse=True)[:20]
+        return {
+            "total_tables_scanned": len(per_table) + len(skipped),
+            "total_fields_checked": len(all_fields),
+            "high_empty_fields_count": len(high),
+            "mid_empty_fields_count": len(mid),
+            "tables_with_issues_count": len(per_table),
+            "skipped_tables_count": len(skipped),
+            "thresholds": {
+                "high_empty": f">={int(high_th*100)}%",
+                "mid_empty": f">={int(mid_th*100)}%",
+                "min_rows_to_check": min_rows,
+            },
+            "top_tables_by_high_empty_fields": [
+                {"table_name": t, "high_empty_field_count": c} for t, c in top_high
+            ],
+            "high_empty_fields": high,
+        }

workflow/steps/step5_missing_comments.py

+"""Step 5: 缺失注释检查 + 推测注释
+
+基于 Step 1 的数据字典，识别没有字段注释的字段。
+对于可推测的字段（基于字段名规律）自动给出推测注释。
+
+产出 section: missing_comments
+"""
+
+from __future__ import annotations
+
+from .base import BaseStep, StepContext
+
+# 字段名 → 推测注释 的映射表（基于常见业务命名习惯）
+# 维护这个表就能持续扩展推测能力
+COMMENT_HINTS = {
+    # 通用
+    "id": "主键ID",
+    "create_time": "创建时间",
+    "update_time": "更新时间",
+    "create_by": "创建人",
+    "update_by": "更新人",
+    "remark": "备注",
+    "del_flag": "删除标记（0=未删 1=已删）",
+    "tenant_id": "租户ID（多租户隔离用）",
+    "dept_id": "部门ID",
+    # 项目相关
+    "project_id": "项目ID",
+    "project_name": "项目名称",
+    "project_code": "项目编号",
+    "project_type": "项目类型",
+    # 企业/单位
+    "enterprise_id": "企业ID",
+    "enterprise_name": "企业名称",
+    "unit_id": "单位ID",
+    # 工地/场地
+    "site_id": "工地ID",
+    "site_code": "工地编号",
+    "site_name": "工地名称",
+    # 行政/地理
+    "xzqhbm": "行政区划编码",
+    "xzqhmc": "行政区划名称",
+    "lat": "纬度",
+    "lng": "经度",
+    "address": "地址",
+    # 时间字段
+    "start_time": "开始时间",
+    "end_time": "结束时间",
+    "plan_start_date": "计划开始日期",
+    "plan_end_date": "计划结束日期",
+    # 状态/标志
+    "status": "状态",
+    "sort_order": "排序",
+    # 业务字段
+    "flow": "流量",
+    "pressure": "压力",
+    "code": "编码",
+    "stage": "阶段",
+    "type": "类型",
+}
+
+
+class Step5MissingComments(BaseStep):
+    name = "5_missing_comments"
+    title = "缺失注释检查 + 推测"
+    produces_section = "missing_comments"
+
+    def run(self, ctx: StepContext) -> dict:
+        dict_data = ctx.step_outputs.get("1_data_dict", {}).get("data_dictionary", [])
+        if not dict_data:
+            self.log(ctx, "未找到 Step 1 产出，自动连库获取...")
+            from .step1_data_dict import Step1DataDict
+            ctx.step_outputs["1_data_dict"] = Step1DataDict().run(ctx)
+            dict_data = ctx.step_outputs["1_data_dict"].get("data_dictionary", [])
+        if not dict_data:
+            self.log(ctx, "仍未拿到数据，跳过")
+            return {"summary": {"total_missing_comments": 0, "tables_affected": 0, "predicted_count": 0}}
+
+        # 找缺失注释的字段
+        missing = []
+        predicted = []
+        for r in dict_data:
+            comment = (r.get("column_comment") or "").strip()
+            if not comment:
+                hint = COMMENT_HINTS.get(r["column_name"])
+                entry = {
+                    "table_name": r["table_name"],
+                    "column_name": r["column_name"],
+                    "data_type": r.get("data_type", ""),
+                }
+                if hint:
+                    entry["predicted"] = hint
+                    entry["reason"] = "字段名匹配常见命名映射"
+                    predicted.append(entry)
+                else:
+                    entry["predicted"] = None
+                missing.append(entry)
+
+        # 按表聚合
+        by_table: dict[str, int] = {}
+        for m in missing:
+            by_table[m["table_name"]] = by_table.get(m["table_name"], 0) + 1
+
+        self.log(
+            ctx,
+            f"缺失注释字段 {len(missing)} 个，覆盖 {len(by_table)} 张表；其中 {len(predicted)} 个可自动推测",
+        )
+
+        return {
+            "summary": {
+                "total_missing_comments": len(missing),
+                "tables_affected": len(by_table),
+                "predicted_count": len(predicted),
+            },
+            "by_table": [
+                {"table_name": t, "missing_count": c}
+                for t, c in sorted(by_table.items(), key=lambda x: x[1], reverse=True)[:20]
+            ],
+            "predicted_comments": predicted[:50],
+            "unpredictable_sample": [m for m in missing if not m.get("predicted")][:50],
+        }

workflow/steps/step6_length_check.py

+"""Step 6: 字段长度检查
+
+基于数据字典，识别字段定义长度超出标准所需长度的字段（如身份证号用 varchar(50)）。
+实际长度通过采样数据估算（如果数据库连得上的话）。
+
+产出 section: length_issues
+"""
+
+from __future__ import annotations
+
+from .base import BaseStep, StepContext
+from ..core.db import query_scalar
+
+# 字段名 → 期望长度的启发式规则
+# 这里只列出标准字段；其他字段如果出现在字典表里会被忽略
+LENGTH_RULES = {
+    "id_card": 18,            # 身份证号
+    "id_card_no": 18,
+    "id_number": 18,
+    "identity_card": 18,
+    "uscc": 18,               # 统一社会信用代码
+    "credit_code": 18,
+    "social_credit_code": 18,
+    "mobile": 11,             # 手机号
+    "phone": 11,
+    "tel": 11,
+    "telephone": 11,
+    "mobile_phone": 11,
+    "xzqhbm": 12,             # 行政区划编码（12位扩展版）
+    "district_code": 12,
+    "adcode": 12,
+    "postal_code": 6,         # 邮政编码
+    "zip_code": 6,
+    "postcode": 6,
+}
+
+
+class Step6LengthCheck(BaseStep):
+    name = "6_length_check"
+    title = "字段长度异常检查"
+    produces_section = "length_issues"
+
+    def run(self, ctx: StepContext) -> dict:
+        dict_data = ctx.step_outputs.get("1_data_dict", {}).get("data_dictionary", [])
+        if not dict_data:
+            self.log(ctx, "未找到 Step 1 产出，自动连库获取...")
+            from .step1_data_dict import Step1DataDict
+            ctx.step_outputs["1_data_dict"] = Step1DataDict().run(ctx)
+            dict_data = ctx.step_outputs["1_data_dict"].get("data_dictionary", [])
+        if not dict_data:
+            self.log(ctx, "仍未拿到数据，跳过")
+            return {"summary": {"total_issues": 0, "rules_applied": 0}, "issues": []}
+
+        issues = []
+        for r in dict_data:
+            col = r["column_name"]
+            if col not in LENGTH_RULES:
+                continue
+            expected = LENGTH_RULES[col]
+            actual = r.get("char_max_length")
+
+            # 只对字符串类型有意义
+            if r.get("data_type") not in ("varchar", "char"):
+                continue
+
+            if actual is None:
+                continue
+            if actual > expected:
+                issues.append({
+                    "table_name": r["table_name"],
+                    "column_name": col,
+                    "data_type": r.get("data_type"),
+                    "actual": str(actual),
+                    "expected": str(expected),
+                    "wasted_bytes_per_row": actual - expected,
+                    "issue": f"字段定义 varchar({actual})，标准仅需 {expected}，超出 {actual - expected}",
+                })
+
+        # 按浪费空间排序
+        issues.sort(key=lambda x: x["wasted_bytes_per_row"], reverse=True)
+
+        self.log(ctx, f"发现 {len(issues)} 个字段长度异常")
+
+        return {
+            "summary": {
+                "total_issues": len(issues),
+                "rules_applied": len(LENGTH_RULES),
+            },
+            "issues": issues[:200],  # Top 200
+        }

workflow/steps/step7_standards.py

+"""Step 7: 国家标准校验
+
+根据字段名自动匹配相应的国家标准，对实际数据进行抽样校验。
+- 身份证号 (GB 11643)
+- 统一社会信用代码 (GB 32100)
+- 手机号 (工信部号标)
+- 行政区划代码 (GB/T 2260)
+
+产出 section: standard_violations
+"""
+
+from __future__ import annotations
+
+from .base import BaseStep, StepContext
+from ..core.db import query_all, query_scalar
+from standards.registry import discover_standards
+
+
+SAMPLE_LIMIT = 500  # 每张表最多抽样
+
+
+class Step7Standards(BaseStep):
+    name = "7_standards"
+    title = "国家标准校验"
+    produces_section = "standard_violations"
+
+    def run(self, ctx: StepContext) -> dict:
+        # 1. 发现所有标准
+        standards_by_field = discover_standards()
+        if not standards_by_field:
+            self.log(ctx, "未发现任何标准插件，跳过")
+            return {"summary": {"standards_applied": 0}}
+
+        self.log(ctx, f"加载了 {len(standards_by_field)} 个标准插件，覆盖 {len(set(standards_by_field.values()))} 个标准")
+
+        # 2. 从数据字典找待校验字段
+        dict_data = ctx.step_outputs.get("1_data_dict", {}).get("data_dictionary", [])
+        # field → list of (table_name)
+        field_to_tables: dict[str, list[str]] = {}
+        for r in dict_data:
+            cn = r["column_name"]
+            if cn in standards_by_field:
+                field_to_tables.setdefault(cn, []).append(r["table_name"])
+
+        if not field_to_tables:
+            self.log(ctx, "未找到匹配任何已知标准的字段")
+            return {"summary": {"standards_applied": len(standards_by_field), "fields_checked": 0}}
+
+        # 3. 连库抽样校验
+        conn = self.connect_db(ctx)
+        results = []
+        try:
+            with conn.cursor() as cur:
+                for field_name, tables in field_to_tables.items():
+                    standard = standards_by_field[field_name]
+                    for tname in tables[:5]:  # 每个字段最多取前 5 张表
+                        try:
+                            self._validate_field(cur, standard, field_name, tname, results)
+                        except Exception as e:
+                            self.log(ctx, f"  ✗ {tname}.{field_name}: {e}")
+        finally:
+            conn.close()
+
+        # 4. 汇总
+        by_std: dict[str, list] = {}
+        for r in results:
+            by_std.setdefault(r["standard"], []).append(r)
+
+        violations_summary = []
+        total_violations = 0
+        for std_id, items in by_std.items():
+            total_invalid = sum(x["invalid"] for x in items)
+            total_violations += total_invalid
+            violations_summary.append({
+                "standard": std_id,
+                "standard_name": items[0]["standard_name"],
+                "fields_count": len(items),
+                "total_invalid": total_invalid,
+                "details": items,
+            })
+
+        self.log(ctx, f"校验完成：发现 {total_violations} 条违规")
+
+        return {
+            "summary": {
+                "standards_applied": len(set(standards_by_field.values())),
+                "fields_checked": sum(len(t) for t in field_to_tables.values()),
+                "violations": total_violations,
+            },
+            "violations_by_standard": violations_summary,
+        }
+
+    def _validate_field(self, cur, standard, field_name: str, tname: str, results: list) -> None:
+        # 先看表是否有数据
+        total = query_scalar(cur, f"SELECT COUNT(*) FROM `{tname}`") or 0
+        if total == 0:
+            return
+
+        # 抽样（避免对大表全扫）
+        sql = f"SELECT `{field_name}` FROM `{tname}` WHERE `{field_name}` IS NOT NULL AND `{field_name}` != '' LIMIT %s"
+        rows = query_all(cur, sql, (SAMPLE_LIMIT,))
+        values = [r[field_name] for r in rows]
+
+        result = standard.validate_batch(values, field_name)
+        result["table"] = tname
+        results.append(result)

workflow/steps/step8_report.py

+"""Step 8: 报告生成
+
+汇总所有 Step 的 findings，生成 Markdown 和 Word 报告。
+
+产出: outputs/<db>/<ts>/reports/ 数据治理报告_<db>.docx + .md
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from .base import BaseStep, StepContext
+from ..core.reporter import build_markdown, build_docx, save_docx
+from ..core.utils import save_json
+
+
+class Step8Report(BaseStep):
+    name = "8_report"
+    title = "生成报告"
+    produces_section = ""  # 报告 Step 不写入 findings
+
+    def run(self, ctx: StepContext) -> dict:
+        # 从所有 Step 输出汇总成 findings 字典
+        findings = self._collect_findings(ctx)
+
+        # 1. 生成 Markdown
+        overview_meta = ctx.step_outputs.get("1_data_dict", {}).get("overview_meta", {})
+        md_text = build_markdown(ctx.cfg, findings, overview_meta)
+        reports_dir = ctx.paths["run_dir"] / "reports"
+        reports_dir.mkdir(parents=True, exist_ok=True)
+
+        md_path = reports_dir / f"数据治理报告_{ctx.db_name}.md"
+        md_path.write_text(md_text, encoding="utf-8")
+        self.log(ctx, f"Markdown 报告: {md_path}")
+
+        # 2. 生成 Word（如果启用）
+        formats = ctx.cfg.get("output", {}).get("formats", ["json", "markdown"])
+        docx_path = None
+        if "docx" in formats:
+            try:
+                title = ctx.cfg.get("report", {}).get("title", "数据治理报告")
+                doc = build_docx(title, md_text)
+                docx_path = reports_dir / f"数据治理报告_{ctx.db_name}.docx"
+                save_docx(doc, docx_path)
+                self.log(ctx, f"Word 报告: {docx_path}")
+            except Exception as e:
+                self.log(ctx, f"Word 报告生成失败: {e}")
+
+        return {
+            "report_files": {
+                "markdown": str(md_path),
+                "docx": str(docx_path) if docx_path else None,
+            },
+            "findings_summary": {k: bool(v) for k, v in findings.items()},
+        }
+
+    def _collect_findings(self, ctx: StepContext) -> dict:
+        """把所有 Step 产出按 section 名汇总"""
+        findings: dict = {}
+
+        # Step 1: overview meta 直接放进 overview
+        overview_meta = ctx.step_outputs.get("1_data_dict", {}).get("overview_meta")
+        if overview_meta:
+            findings["overview"] = {"meta": overview_meta}
+
+        # Step 2: merge_candidates + decommission + redundancy
+        s2 = ctx.step_outputs.get("2_merge_redundancy", {})
+        if s2:
+            findings["merge_candidates"] = {
+                "merge_candidates": s2.get("merge_candidates", []),
+                "decommission_candidates": s2.get("decommission_candidates", {}),
+            }
+            findings["redundancy_fields"] = {
+                "redundancy_fields": s2.get("redundancy_fields", []),
+            }
+
+        # Step 3: 验证后的合并/冗余 + 数据质量问题
+        s3 = ctx.step_outputs.get("3_verify", {})
+        if s3:
+            # 用 Step 3 验证后的数据覆盖 Step 2 的初步分析
+            if s3.get("confirmed_merge_candidates"):
+                findings["merge_candidates"]["confirmed_merge_candidates"] = s3["confirmed_merge_candidates"]
+            if s3.get("redundancy_fields"):
+                findings["redundancy_fields"]["confirmed_redundancy_fields"] = s3["redundancy_fields"]
+            if s3.get("data_quality_issues"):
+                findings["data_quality"] = {"issues": s3["data_quality_issues"]}
+
+        # Step 4: 空字段
+        s4 = ctx.step_outputs.get("4_empty_fields", {})
+        if s4:
+            findings["empty_fields"] = s4
+
+        # Step 5: 缺失注释
+        s5 = ctx.step_outputs.get("5_missing_comments", {})
+        if s5:
+            findings["missing_comments"] = s5
+
+        # Step 6: 字段长度
+        s6 = ctx.step_outputs.get("6_length_check", {})
+        if s6:
+            findings["length_issues"] = s6
+
+        # Step 7: 标准违规
+        s7 = ctx.step_outputs.get("7_standards", {})
+        if s7:
+            findings["standard_violations"] = s7
+
+        # 保存合并后的 findings
+        findings_path = ctx.paths["findings_dir"] / "_all_findings.json"
+        save_json(findings, findings_path)
+        self.log(ctx, f"汇总 findings 已保存: {findings_path}")
+
+        return findings