Commit 272c10ed authored by Data Governance Dev's avatar Data Governance Dev

refactor(web): 数据字典抽到 connect-test + 步骤用 step_id 注册 + 修白屏

## 1. 操作模式重构(核心需求)

按用户要求把 web 端的操作模式从"启动任务时抽数据字典 + 编号步骤"改成
"测试连接时抽数据字典 + 步骤注册式可扩展 + 勾选表后再分析"。

### 后端
- 新增 `web/core/data_dict.py`:`extract_data_dictionary(cfg)` 函数,
  connect-test 与 orchestrator 共用(避免重复查询)
- 新增 `web/core/data_dict_cache.py`:按连接身份 (db_type/host/port/db/user)
  缓存字典,缓存命中走内存,未命中现场再抽
- 删除 `web/core/step_impl/step1_data_dict.py`(从流程中彻底移除)
- `web/core/orchestrator.py`:
  - 流程表从 `[(num, fn, ...), ...]` 改为 `dict[str, StepDef]`
  - 用 `register_step(...)` 装饰器一行注册一个 Step
  - `CRITICAL_STEPS` 改为 `{"merge_redundancy", "missing_comments"}`
  - `run_governance_workflow(..., tables: list[str])` 接受前端勾选的表
- `web/core/models.py`:
  - `ConnectRequest.tables: list[str]`(必填,空时后端返 400)
  - `StepInfo.num: int` → `StepInfo.id: str` + `order: int`
  - `JobStatusResponse.current_step: Optional[int]` → `Optional[str]`
  - 所有步骤列表 `list[int]` → `list[str]`
  - `TestConnectionResponse` 新增 `table_summary` / `data_dictionary` 字段
- `web/core/step_impl/step{2,4,5,6,7}.py`:
  - 全部新增 `table_filter: set[str] | None = None` 参数,开头过滤 `columns`
  - 全部 `step="N"` 字符串改为 `step="step_id"`(merge_redundancy / empty_fields / ...)
- `web/core/job_manager.py`:`Job.current_step` / `steps_planned` / `steps_completed` /
  `steps_failed` 全部从 `int` 改为 `str`,`_sync_step_start/done` 同步
- `web/api/routes.py`:
  - `/api/connect/test` 成功后 `extract_data_dictionary` + `cache.put`,
    返回 `table_summary` + `data_dictionary`
  - `/api/jobs` 校验 `tables` 非空,否则 `HTTP 400`
  - `/api/steps` 从 `get_step_defs()` 动态生成(不再写死编号)

### 前端 (`web/static/index.html` 全量重写)
- 卡片 1:连接数据库(保留)
- 卡片 2(新增):数据字典浏览器
  - 左侧 el-table(reserve-selection):表名 / 注释 / 字段数;
    全选 / 清空 / 反选 按钮 + 搜索框
  - 右侧 el-table:选中表的所有字段,按 (table_name, ordinal_position) 排序
- 卡片 3:分析配置
  - 步骤勾选 + 「开始分析」按钮(label 实时显示 X 张表 / Y 项检查)
  - 表为空时按钮 disabled 并提示
- 卡片 4:分析进度 + 日志(SSE 流式)
- 卡片 5:分析结果(KPI + 多级表格)
  - `TAB_STEP_MAP` 改用 step_id,不再用编号
  - 移除所有 "Step N:" 前缀

## 2. 修复页面打开白屏

`web/static/index.html` 打开后页面全白,根因两层叠加:

### 2.1 多余 `</el-option>`
DB 表单"字符集"里写了 `<el-option ... />` 自闭合 + `</el-option>` 显式闭合,
Vue 模板编译器报 `Invalid end tag` (compiler code 23)。
修复:删除第 102 行的 `</el-option>`。

### 2.2 `v-else-if` 链断裂(Vue 3 compiler code 30)
Vue 3 要求 `v-else` / `v-else-if` 必须是上一个 v-if 元素的紧邻兄弟。模板里:
- 「重新分析」按钮的 `v-else-if` 与「取消」按钮的 `v-if` 中间夹了一个
  无 v-if 的「开始分析」按钮,链断
- 「勾选提示」el-alert 的 `v-else-if` 上方是 `</el-form-item>`,也不是 v-if
修复:去掉 v-else-if,改用独立 v-if 显式写互斥条件。

## 3. 验证

- `python -c "from web.core import ..."` 全部通过
- `python -m web.app` 启动 OK,注册 5 个 step
- `/api/steps` 返回 `{id, order, title, ...}`(无 num)
- `/api/connect/test` 无 DB 时 ok=false,含 `table_summary: []`, `data_dictionary: []`
- `/api/jobs` 空 tables → `HTTP 400 {"detail": "tables 不能为空:..."}`
- 用 jsdom 抓真服务器返回的 `#app.innerHTML`(59914 字符),
  喂给浏览器版 Vue 3.5.40 编译:errors=0, warnings=0
- jsdom 完整挂载:`#app` 渲染 2 子元素 / 8484 字符真实 DOM

## 4. 兼容性

- 旧的 `outputs/<db>/<ts>/findings/step_*_data_dict.json` 文件保留(前端不读)
- `step8_report.py` 消费的 6 个 section key
  (`merge_candidates` / `redundancy_fields` / `empty_fields` /
  `missing_comments` / `length_issues` / `standard_violations`)全部保留

## 5. 文件清单

新增:
- `web/core/data_dict.py`
- `web/core/data_dict_cache.py`
- `docs/WORKLOG.md`

删除:
- `web/core/step_impl/step1_data_dict.py`

修改:
- `web/api/routes.py`
- `web/core/job_manager.py`
- `web/core/models.py`
- `web/core/orchestrator.py`
- `web/core/step_impl/step{2,4,5,6,7}_*.py`
- `web/static/index.html`(全量重写)
- `web/README.md`
- `CLAUDE.md`(项目说明改为"数据治理工具,主要开发 web 端")
- `.gitignore`(加 tags.lock / tags.temp)
parent 6c7bc274
...@@ -33,3 +33,5 @@ work-logs/ ...@@ -33,3 +33,5 @@ work-logs/
*.log *.log
!web/logs/.gitkeep !web/logs/.gitkeep
数据治理报告_smart-build.docx 数据治理报告_smart-build.docx
tags.lock
tags.temp
# 说明 # 说明
这是一个数据治理的项目,我需要对一个数据库里面表单以及数据进行分析,然后提出改进意见 这是一个数据治理工具,现在主要开发web端
注意我不会对数据库做任何修改,只是发现问题
最终成果是一个报告文件,以word文档形式
项目数据库:smart-build 项目数据库:smart-build
数据库软件:mysql 数据库软件:mysql
......
# 工作记录
> 任务做完一次记一次。最近的在最上面。
## 2026-08-07 · 修复页面打开空白
### 问题
前端 `web/static/index.html` 打开后页面全白。
### 根因(两层叠加)
#### 第一层:多余 `</el-option>`
DB 连接表单的「字符集」`<el-select>` 里写了:
```html
<el-option label="utf8mb4" value="utf8mb4" />
<el-option label="utf8" value="utf8" />
<el-option label="gbk" value="gbk" />
</el-option> ← 多余的闭合
</el-select>
```
`<el-option ... />` 是自闭合的,再写 `</el-option>` Vue 模板编译器报 `Invalid end tag` (error code 23),整个组件挂载失败 → 空白页。
#### 第二层:`v-else-if` 不与 `v-if` 相邻
修了第一处后页面仍然空白。再次用浏览器版本 Vue 3.5.40 编译模板(用 jsdom 解析后取 `#app.innerHTML`),定位到 compiler error code 30(`X_V_ELSE_NO_ADJACENT_IF`):
[web/static/index.html:245-255](web/static/index.html#L245-L255):
```html
<el-form-item>
<el-button type="success" @click="startJob" ...>开始分析</el-button> ← 永远显示,无 v-if
<el-button v-if="currentJob && jobStatus.status === 'running'" ...>取消</el-button>
<el-button v-else-if="jobResult || ...">重新分析</el-button> ← 报 compiler-30
</el-form-item>
<el-alert v-if="!connected" .../> ← 报 compiler-30
<el-alert v-else-if="form.tables.length === 0" .../> ← 报 compiler-30
```
Vue 3 规定 `v-else` / `v-else-if` **必须是上一个 v-if/v-else-if 元素的紧邻下一个兄弟**。
- 「重新分析」按钮与「取消」按钮之间夹了一个无 v-if 的「开始分析」按钮,链断裂
- 「勾选提示」el-alert 与「请先连接」el-alert 上方是 `</el-form-item>`,也不是 v-if,链断裂
### 验证
用 `jsdom` 拉真服务器返回的 `index.html`,抓 `#app.innerHTML`(59914 字符)后用浏览器版 Vue 3.5.40 `compile()` 编译:
- 修复前:1 个 `compiler-23`(多余 `</el-option>`)+ 2 个 `compiler-30`(v-else-if 链断裂)
- 修复后:`errors: 0, warnings: 0`
- JSDOM 模拟挂载:`#app.children.length=2, innerHTML=8484 字符`(正常渲染)
### 修复
#### 1. 删 `</el-option>` [web/static/index.html:102](web/static/index.html#L102)
保留 3 个自闭合 `<el-option ... />`,删除后面多余的 `</el-option>`。
#### 2. 「重新分析」按钮去掉 `v-else-if` [web/static/index.html:252](web/static/index.html#L252)
```html
<!-- 改前 -->
<el-button v-else-if="jobResult || (currentJob && jobStatus.status !== 'running')" ...>
<!-- 改后 -->
<el-button v-if="!submitting && (jobResult || (currentJob && jobStatus.status !== 'running'))" ...>
```
多加 `!submitting` 保证「开始分析」进行中不显示重新分析。
#### 3. 「勾选提示」el-alert 去掉 `v-else-if` [web/static/index.html:257](web/static/index.html#L257)
```html
<!-- 改前 -->
<el-alert v-else-if="form.tables.length === 0" ...>
<!-- 改后 -->
<el-alert v-if="connected && form.tables.length === 0" ...>
```
显式写 `connected &&` 与上面 `!connected` 的 `v-if` 形成互斥语义,不再依赖链。
### 调试副产物(均已删除)
- `node_modules/`(含 vue@3 + jsdom)
- `check_compile.cjs` / `check_compile_browser.cjs` / `check_compile_parsed.cjs`(用浏览器版 Vue 编译)
- `check_fetch.cjs` / `check_parse.cjs`(jsdom 抓 innerHTML)
- `parsed_template*.html` / `regex_template.html`
### 未提交
按用户要求,待 review 后再提交。
## 2026-08-07 · 操作模式重构
### 需求
1. 测试连接时抽数据字典,前端展示 + 供勾选
2. 双表浏览器:左 = 表清单(带勾选),右 = 选中表的字段
3. 各检查只分析勾选的表
4. Step 不再编号,注册方式可扩展
### 实现要点
**后端**
- 新增 `web/core/data_dict.py`(`extract_data_dictionary` 函数,connect-test 与 orchestrator 共用)
- 新增 `web/core/data_dict_cache.py`(按 `(db_type, host, port, database, user)` 缓存,缺失则现场再抽)
- 删除 `web/core/step_impl/step1_data_dict.py`(彻底从流程移除)
- 重写 `web/core/orchestrator.py`:
- `STEP_REGISTRY` 由 int tuple list 改为 `dict[str, StepDef]`
- 用 `register_step(...)` 一行注册一个 Step
- `CRITICAL_STEPS` 改为 `{"merge_redundancy", "missing_comments"}`
- 主函数签名加 `tables: list[str]`,各 Step 通过 `table_filter` 参数过滤
- `web/core/models.py`:
- `ConnectRequest.tables: list[str]`(必填,空 = 400)
- `StepInfo.num: int` → `StepInfo.id: str` + `order: int`
- `JobStatusResponse.current_step: Optional[int]` → `Optional[str]`
- 所有 `list[int]` 步骤列表 → `list[str]`
- `TestConnectionResponse` 加 `table_summary` / `data_dictionary`
- `web/core/step_impl/step{2,4,5,6,7}.py` 都加了 `table_filter: set[str] | None = None` 参数,开头过滤 `columns`
- `web/core/job_manager.py`:`Job.current_step` / `steps_planned` / `steps_completed` / `steps_failed` 全部改 `list[str]` / `Optional[str]`
- `web/api/routes.py`:
- `/api/connect/test` 成功后调 `extract_data_dictionary` + `cache.put`,返回 `table_summary` + `data_dictionary`
- `/api/jobs` 校验 `tables` 非空,否则 400
- `/api/steps` 从 `get_step_defs()` 动态生成
**前端(`web/static/index.html` 全量重写)**
- 卡片 1:连接数据库(不变)
- 卡片 2(新增):数据字典浏览器
- 左侧 el-table(reserve-selection):表名 / 表注释 / 字段数;带全选 / 清空 / 反选按钮 + 搜索框
- 右侧 el-table:当前勾选表的所有字段,按 `(table_name, ordinal_position)` 排序
- 卡片 3(拆分自原 ①):分析配置 — 步骤勾选 + 「开始分析」按钮
- 按钮 label 带「X 张表 / Y 项检查」
- 空勾选时按钮 disabled 并提示
- 卡片 4:进度 + 日志
- 日志不再渲染 `log.step` 标签(结构化数据仍保留 `step_id`)
- 卡片 5:分析结果(多级表格,不变)
- `TAB_STEP_MAP` 改用 step_id
- 移除所有「Step N:」编号前缀
### 验证
- `python -c "from web.core import ..."` 通过
- `python -m web.app` 启动 OK,注册了 5 个 step
- `/api/steps` 返回 `{id, order, title, ...}`(无 `num`)
- `/api/connect/test` 无 DB 时返回 ok=false,含 `table_summary: [], data_dictionary: []`
- `/api/jobs` 空 tables 返回 `HTTP 400 {"detail": "tables 不能为空:..."}`
### 兼容性
- 旧的 `outputs/<db>/<ts>/findings/step_*_data_dict.json` 文件保留(前端不读)
- `step8_report.py` 消费 6 个 section key(`merge_candidates` / `redundancy_fields` / `empty_fields` / `missing_comments` / `length_issues` / `standard_violations`),全部保留
- 旧数据兼容:`overview_meta` 不再存在,`_build_overview` 改用注入的 `dict_data`
### 未提交
按用户要求,待 review 后再提交。
\ No newline at end of file
...@@ -100,7 +100,7 @@ python -m uvicorn web.app:app --host 0.0.0.0 --port 8765 ...@@ -100,7 +100,7 @@ python -m uvicorn web.app:app --host 0.0.0.0 --port 8765
打开 **http://localhost:8765** 打开 **http://localhost:8765**
- Web 界面:填表 → 测试连接 → 开始治理 - Web 界面:填表 → 测试连接(自动抽取数据字典) → 勾选表 → 选择检查项 → 开始治理
- API 文档:http://localhost:8765/docs - API 文档:http://localhost:8765/docs
- OpenAPI JSON:http://localhost:8765/openapi.json - OpenAPI JSON:http://localhost:8765/openapi.json
...@@ -113,12 +113,14 @@ python -m uvicorn web.app:app --host 0.0.0.0 --port 8765 ...@@ -113,12 +113,14 @@ python -m uvicorn web.app:app --host 0.0.0.0 --port 8765
| 🖥️ **Web 界面** | 浏览器访问,Vue 3 + Element Plus 响应式布局 | | 🖥️ **Web 界面** | 浏览器访问,Vue 3 + Element Plus 响应式布局 |
| 🔌 **双数据库** | MySQL(PyMySQL)+ 达梦(dmPython)统一接口 | | 🔌 **双数据库** | MySQL(PyMySQL)+ 达梦(dmPython)统一接口 |
| 🤖 **LLM 增强** | Claude API 用于字段注释推测 / 合并建议 / 长度推断 / 空字段分类 | | 🤖 **LLM 增强** | Claude API 用于字段注释推测 / 合并建议 / 长度推断 / 空字段分类 |
| 📊 **多级表格** | 7 大章节各一张可展开表格,KPI 卡片概览 | | 📊 **多级表格** | 6 大章节各一张可展开表格,KPI 卡片概览 |
| 📜 **国标展示** | 4 项内置标准可视化展示在页面(无需查询) | | 📜 **国标展示** | 4 项内置标准可视化展示在页面(无需查询) |
| 📡 **实时日志** | SSE 流式推送治理进度,含 INFO/WARN/ERROR 三色 | | 📡 **实时日志** | SSE 流式推送治理进度,含 INFO/WARN/ERROR 三色 |
| 📥 **报告下载** | Markdown / Word 双格式,含历史报告检索 | | 📥 **报告下载** | Markdown / Word 双格式,含历史报告检索 |
| 🗂️ **历史归档** | 每次运行按时间戳归档到 `outputs/<db>/<ts>/` | | 🗂️ **历史归档** | 每次运行按时间戳归档到 `outputs/<db>/<ts>/` |
| 🛡️ **优雅降级** | LLM 失败 / DB 失败不卡死,规则推理兜底 | | 🛡️ **优雅降级** | LLM 失败 / DB 失败不卡死,规则推理兜底 |
| 🔍 **数据字典浏览器** | 测试连接后即可看到所有表/字段;勾选后只分析所选表 |
| 🧩 **可扩展 Step 注册** | `register_step(...)` 一行注册新 Step,无需改 orchestrator |
--- ---
...@@ -138,15 +140,17 @@ python -m uvicorn web.app:app --host 0.0.0.0 --port 8765 ...@@ -138,15 +140,17 @@ python -m uvicorn web.app:app --host 0.0.0.0 --port 8765
│ │ ├── verify/ xzqh/ standards/ empty_fields/ health/ │ │ ├── verify/ xzqh/ standards/ empty_fields/ health/
│ ├── core/ ← 核心模块 │ ├── core/ ← 核心模块
│ │ ├── db_adapter.py ← MySQL + 达梦 统一接口 │ │ ├── db_adapter.py ← MySQL + 达梦 统一接口
│ │ ├── data_dict.py ← 数据字典抽取(connect-test 与 orchestrator 共用)
│ │ ├── data_dict_cache.py ← 数据字典进程内缓存(按连接身份)
│ │ ├── llm.py ← Claude API 封装 │ │ ├── llm.py ← Claude API 封装
│ │ ├── job_manager.py ← 后台任务 + SSE 推送 │ │ ├── job_manager.py ← 后台任务 + SSE 推送
│ │ ├── orchestrator.py ← 8 步流程调度 │ │ ├── orchestrator.py ← 流程调度 + Step 注册(register_step)
│ │ ├── models.py ← Pydantic 契约 │ │ ├── models.py ← Pydantic 契约
│ │ └── step_impl/ ← 8 个 Step 的 Web 版实现 │ │ └── step_impl/ ← 5 个 Step 的实现(merge_redundancy / empty_fields / missing_comments / length_check / standards)
│ ├── api/ │ ├── api/
│ │ └── routes.py ← 12 个 REST 接口 │ │ └── routes.py ← REST 接口
│ ├── static/ │ ├── static/
│ │ ├── index.html ← Vue 3 单页前端 │ │ ├── index.html ← Vue 3 单页前端(连接 / 数据字典浏览器 / 配置 / 进度 / 结果 5 张卡)
│ │ └── style.css │ │ └── style.css
│ ├── configs/ │ ├── configs/
│ │ └── defaults.yaml │ │ └── defaults.yaml
...@@ -172,10 +176,10 @@ python -m uvicorn web.app:app --host 0.0.0.0 --port 8765 ...@@ -172,10 +176,10 @@ python -m uvicorn web.app:app --host 0.0.0.0 --port 8765
| 方法 | 路径 | 说明 | | 方法 | 路径 | 说明 |
|------|------|------| |------|------|------|
| GET | `/api/health` | 健康检查 | | GET | `/api/health` | 健康检查 |
| GET | `/api/steps` | 列出 8 步元信息 | | GET | `/api/steps` | 列出已注册的 Step 元信息(按 step_id) |
| GET | `/api/standards` | 列出已注册国标(页面用) | | GET | `/api/standards` | 列出已注册国标(页面用) |
| POST | `/api/connect/test` | 测试连接(返回表数量) | | POST | `/api/connect/test` | 测试连接;成功后同步抽取数据字典并缓存到 `DataDictCache` |
| POST | `/api/jobs` | 提交治理任务 | | POST | `/api/jobs` | 提交治理任务(必须带非空 `tables`) |
| GET | `/api/jobs/{id}` | 查询任务状态(含进度百分比) | | GET | `/api/jobs/{id}` | 查询任务状态(含进度百分比) |
| DELETE | `/api/jobs/{id}` | 取消任务 | | DELETE | `/api/jobs/{id}` | 取消任务 |
| GET | `/api/jobs/{id}/result` | 获取完整结果 | | GET | `/api/jobs/{id}/result` | 获取完整结果 |
...@@ -190,21 +194,31 @@ python -m uvicorn web.app:app --host 0.0.0.0 --port 8765 ...@@ -190,21 +194,31 @@ python -m uvicorn web.app:app --host 0.0.0.0 --port 8765
LLM 不是「增强」,对部分步骤是核心依赖: LLM 不是「增强」,对部分步骤是核心依赖:
| Step | LLM 模式 | 用途 | 缺 LLM 行为 | | step_id | LLM 模式 | 用途 | 缺 LLM 行为 |
|------|---------|------|------------| |---------|---------|------|------------|
| Step 2 表合并与冗余字段分析 | **required** | 合并 verdict + 高频字段分类 | 任务整体失败 | | `merge_redundancy` 表合并与冗余字段分析 | **required** | 合并 verdict + 高频字段分类 | 任务整体失败 |
| Step 5 缺失注释字段推测 | **required** | 字段语义推测(拼音/英文→中文) | 任务整体失败 | | `missing_comments` 缺失注释字段推测 | **required** | 字段语义推测(拼音/英文→中文) | 任务整体失败 |
| `empty_fields` 大范围空字段扫描 | none | 纯程序化检查 | — |
> Step 1 / 3 / 4 / 6 / 7 / 8 均为纯规则匹配,无 LLM 依赖。其中 Step 6(字段长度检查)目前仅依赖内置 GB/YD 标准规则集,未启用 LLM 自定义字段推断。 | `length_check` 字段长度检查 | none | 内置 GB/YD 标准规则 | — |
| `standards` 国家标准校验 | none | 内置标准插件抽样 | — |
### 设计要点 ### 设计要点
- **required** 步骤:任务启动时预检 LLM 可用性;不可用 → `RuntimeError` → 任务标 `JOB_FAILED` - **required** 步骤:任务启动时预检 LLM 可用性;不可用 → `RuntimeError` → 任务标 `JOB_FAILED`
- **optional** 步骤:LLM 可用就用,失败自动降级到规则推理 - **optional** 步骤:LLM 可用就用,失败自动降级到规则推理
- **批处理**:Step 2 / 5 的 LLM 调用按 15 字段/批,避免 200 次串行 API - **批处理**:`merge_redundancy` / `missing_comments` 的 LLM 调用按 15/30 字段/批,避免 200 次串行 API
- **数据字典与 Step 分离**:连接时已经抽好并缓存到 `DataDictCache`,任务启动直接复用
LLM 调用封装在 [web/core/llm.py](web/core/llm.py)。配置 Key 见上方「配置 LLM Key」 LLM 调用封装在 [web/core/llm.py](web/core/llm.py)。配置 Key 见上方「配置 LLM Key」
小节;未配置时勾选 Step 2/5 的任务会被拦截、整体失败。 小节;未配置时勾选 LLM-required 的任务会被拦截、整体失败。
## 🧩 新增 Step 的步骤
1. 在 [web/core/orchestrator.py](web/core/orchestrator.py) 末尾加一个 `_run_xxx(...)` 适配函数
2. 在同一文件调一次 `register_step(step_id="xxx", title=..., requires_db=..., llm_mode="none", required=False, order=N, fn=_run_xxx)`
3. 在 [web/core/step_impl/](web/core/step_impl/) 里放实现函数,签名为 `(cfg, dict_data, llm, log, cancel_event, table_filter) -> dict`
> Step 注册是声明式 dict + decorator 风格;新增/删除/重排 Step 不需要改 orchestrator 主循环。
--- ---
......
...@@ -14,6 +14,7 @@ from fastapi.responses import FileResponse, JSONResponse ...@@ -14,6 +14,7 @@ from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel from pydantic import BaseModel
from ..core.db_adapter import DBConfig, test_connection from ..core.db_adapter import DBConfig, test_connection
from ..core.data_dict_cache import DataDictCache
from ..core.job_manager import JobManager from ..core.job_manager import JobManager
from ..core.models import ( from ..core.models import (
ConnectRequest, ConnectRequest,
...@@ -67,25 +68,14 @@ async def llm_status(): ...@@ -67,25 +68,14 @@ async def llm_status():
# ── 列出所有步骤 ── # ── 列出所有步骤 ──
@router.get("/steps", response_model=StepsListResponse) @router.get("/steps", response_model=StepsListResponse)
async def list_steps(): async def list_steps():
from ..core.orchestrator import get_step_defs
return StepsListResponse(steps=[ return StepsListResponse(steps=[
StepInfo(num=1, title="获取数据字典", StepInfo(
description="从 information_schema 抓取所有表/字段元数据(必选,作为后续步骤基础)", id=sd.step_id, order=sd.order, title=sd.title,
requires_db=True, llm_mode="none", required=True), description=sd.description, requires_db=sd.requires_db,
StepInfo(num=2, title="表合并与冗余字段分析", llm_mode=sd.llm_mode, required=sd.required,
description="离线:找结构高度相似的表 + 高频字段(LLM 必须:合并 verdict 与冗余分类)", )
requires_db=False, llm_mode="required", required=False), for sd in get_step_defs()
StepInfo(num=3, title="大范围空字段扫描",
description="单表全列扫,统计 NULL/空值比例(纯程序化检查,不依赖 LLM)",
requires_db=True, llm_mode="none", required=False),
StepInfo(num=4, title="缺失注释字段 + 推测",
description="离线:找无注释字段,LLM 必须:批量推测语义",
requires_db=False, llm_mode="required", required=False),
StepInfo(num=5, title="字段长度检查",
description="离线:识别字段定义长度超出国家标准(纯规则匹配,不依赖 LLM)",
requires_db=False, llm_mode="none", required=False),
StepInfo(num=6, title="国家标准校验",
description="连库抽样:身份证/USCC/手机号/行政区划符合性",
requires_db=True, llm_mode="none", required=False),
]) ])
...@@ -122,37 +112,55 @@ async def connect_test(req: TestConnectionRequest): ...@@ -122,37 +112,55 @@ async def connect_test(req: TestConnectionRequest):
logger.warning(f"连接测试失败: {msg}") logger.warning(f"连接测试失败: {msg}")
return TestConnectionResponse(ok=False, message=msg, db_type=req.db_type) return TestConnectionResponse(ok=False, message=msg, db_type=req.db_type)
# 顺便取表数量 # 测试连接成功后:抽取数据字典,灌进缓存,供前端「表/字段浏览器」消费。
tables_count = None # 失败也不阻塞 —— 前端可能只是要看连接是否通。
table_summary: list[dict] = []
data_dictionary: list[dict] = []
tables_count: Optional[int] = None
try: try:
from ..core.db_adapter import open_db from ..core.data_dict import extract_data_dictionary
with open_db(cfg) as db: dict_data = extract_data_dictionary(cfg)
tables_count = len(db.list_tables(req.database)) DataDictCache.get_instance().put(cfg, dict_data)
logger.info(f"连接测试成功: {msg}, 表数 {tables_count}") table_summary = dict_data.get("table_summary", [])
data_dictionary = dict_data.get("data_dictionary", [])
tables_count = len(table_summary)
logger.info(
f"连接测试成功: {msg}, 表数 {tables_count}, 字段数 {len(data_dictionary)}, 已缓存"
)
except Exception as e: except Exception as e:
logger.warning(f"取表数量失败: {e}") logger.warning(f"抽取数据字典失败: {e}")
return TestConnectionResponse( return TestConnectionResponse(
ok=True, message=msg, db_type=req.db_type, ok=True, message=msg, db_type=req.db_type,
database_name=req.database, tables_count=tables_count, database_name=req.database, tables_count=tables_count,
table_summary=table_summary,
data_dictionary=data_dictionary,
) )
# ── 创建治理任务 ── # ── 创建治理任务 ──
@router.post("/jobs", response_model=JobCreatedResponse) @router.post("/jobs", response_model=JobCreatedResponse)
async def create_job(req: ConnectRequest): async def create_job(req: ConnectRequest):
# 前端应该已经拦住,但服务端兜底再校验一次 —— 空表 = 必失败
if not req.tables:
raise HTTPException(status_code=400, detail="tables 不能为空:必须至少勾选一张表才能开始分析")
logger.info( logger.info(
f"POST /api/jobs db={req.database}, db_type={req.db_type}, " f"POST /api/jobs db={req.database}, db_type={req.db_type}, "
f"steps={req.steps or 'all'}, enable_llm={req.enable_llm} " f"steps={req.steps or 'all'}, tables={len(req.tables)} 张, "
f"(注:enable_llm=False 时,Step 2/5 会因 LLM-required 而失败)" f"enable_llm={req.enable_llm} "
f"(注:enable_llm=False 时,LLM-required 步骤会因缺 LLM 而失败)"
) )
jm = JobManager.get_instance() jm = JobManager.get_instance()
job = jm.submit(req) job = jm.submit(req)
# job.steps_planned 初始化要等 _run_job 里赋值;这里给个合理默认值
from ..core.orchestrator import get_step_defs
default_planned = [s.step_id for s in get_step_defs()]
return JobCreatedResponse( return JobCreatedResponse(
job_id=job.job_id, job_id=job.job_id,
status=job.status, status=job.status,
created_at=job.created_at, created_at=job.created_at,
steps_planned=job.steps_planned or [1, 2, 4, 5, 6, 7], steps_planned=job.steps_planned or default_planned,
) )
......
"""Step 1: 获取数据字典(MySQL + 达梦 适配版) """数据字典抽取
通过 db_adapter 统一接口获取: 把 Step 1 的核心逻辑独立成函数,供:
- data_dictionary: 全部字段元数据 - POST /api/connect/test 在前端「测试连接」时调用,把字典灌进缓存供前端预览
- table_summary: 表级汇总 - run_governance_workflow 在启动任务时复用缓存(缺失则现场再抽一次兜底)
- overview_meta: 概览统计
返回结构与旧 Step 1 完全一致(data_dictionary / table_summary / overview_meta),
各分析 Step 已经按这套结构消费;改这里不需要动下游。
""" """
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Callable from typing import Callable, Optional
from ..db_adapter import DBConfig, open_db from .db_adapter import DBConfig, open_db
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def run_step1(cfg: DBConfig, log: Callable | None = None) -> dict: def extract_data_dictionary(cfg: DBConfig, log: Optional[Callable] = None) -> dict:
"""执行 Step 1,返回 data_dictionary + table_summary + overview_meta""" """从 INFORMATION_SCHEMA 抽取数据字典。
Returns:
{
"data_dictionary": [ {table_name, column_name, ordinal_position, column_type,
data_type, char_max_length, numeric_precision, numeric_scale,
is_nullable, column_default, column_comment, extra,
table_comment}, ... ],
"table_summary": [ {table_name, table_type, engine, table_rows, ...}, ... ],
"overview_meta": {database, host, db_type, total_tables, total_fields},
}
"""
if log: if log:
log("INFO", f"连接数据库: {cfg.db_type}://{cfg.user}@{cfg.host}:{cfg.port}/{cfg.database}", step="1") log("INFO", f"连接数据库: {cfg.db_type}://{cfg.user}@{cfg.host}:{cfg.port}/{cfg.database}")
with open_db(cfg) as db: with open_db(cfg) as db:
if log: if log:
log("INFO", "已建立连接,开始读取 information_schema.COLUMNS ...", step="1") log("INFO", "已建立连接,开始读取 information_schema.COLUMNS ...")
columns = db.list_columns(cfg.database) columns = db.list_columns(cfg.database)
if log: if log:
log("INFO", f" · COLUMNS: {len(columns)} 行", step="1") log("INFO", f" · COLUMNS: {len(columns)} 行")
if log: if log:
log("INFO", "读取 information_schema.TABLES ...", step="1") log("INFO", "读取 information_schema.TABLES ...")
tables = db.list_tables(cfg.database) tables = db.list_tables(cfg.database)
if log: if log:
log("INFO", f" · TABLES: {len(tables)} 行", step="1") log("INFO", f" · TABLES: {len(tables)} 行")
# 类型规整 # 类型规整
fixed_int = 0 fixed_int = 0
...@@ -44,7 +57,7 @@ def run_step1(cfg: DBConfig, log: Callable | None = None) -> dict: ...@@ -44,7 +57,7 @@ def run_step1(cfg: DBConfig, log: Callable | None = None) -> dict:
except (TypeError, ValueError): except (TypeError, ValueError):
pass pass
if log and fixed_int: if log and fixed_int:
log("DEBUG", f" · 字段类型规整: 转换 {fixed_int} 个数值字段", step="1") log("DEBUG", f" · 字段类型规整: 转换 {fixed_int} 个数值字段")
for r in tables: for r in tables:
for tf in ("create_time", "update_time"): for tf in ("create_time", "update_time"):
...@@ -57,10 +70,11 @@ def run_step1(cfg: DBConfig, log: Callable | None = None) -> dict: ...@@ -57,10 +70,11 @@ def run_step1(cfg: DBConfig, log: Callable | None = None) -> dict:
except (TypeError, ValueError): except (TypeError, ValueError):
pass pass
# 概览
table_names = {r["table_name"] for r in columns} table_names = {r["table_name"] for r in columns}
if log: if log:
log("INFO", f"获取完成: {len(table_names)} 张表, {len(columns)} 个字段(覆盖 {len(tables)} 条表信息)", step="1") log("INFO",
f"获取完成: {len(table_names)} 张表, {len(columns)} 个字段"
f"(覆盖 {len(tables)} 条表信息)")
overview_meta = { overview_meta = {
"database": cfg.database, "database": cfg.database,
......
"""数据字典缓存(进程内单例)
每次 POST /api/connect/test 成功后,把刚抽到的数据字典放进来;启动任务时
run_governance_workflow 先按「连接身份」查询,缺失则现场再抽一次兜底。
连接身份 = (db_type, host, port, database, user) —— 不含密码。
适用场景:本地工具型 Web 服务,单用户操作;多用户并发不在考虑范围内。
"""
from __future__ import annotations
import logging
import threading
from typing import Optional
from .db_adapter import DBConfig
from .data_dict import extract_data_dictionary
logger = logging.getLogger(__name__)
def _cache_key(cfg: DBConfig) -> tuple:
return (cfg.db_type, cfg.host, cfg.port, cfg.database, cfg.user)
class DataDictCache:
"""进程内单例(线程安全)。"""
_instance: Optional["DataDictCache"] = None
_lock = threading.Lock()
def __init__(self) -> None:
self._store: dict[tuple, dict] = {}
@classmethod
def get_instance(cls) -> "DataDictCache":
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = cls()
return cls._instance
# ── 写 ──
def put(self, cfg: DBConfig, dict_data: dict) -> None:
self._store[_cache_key(cfg)] = dict_data
logger.info(
f"[Cache] 写入数据字典: {_cache_key(cfg)} "
f"(表 {len(dict_data.get('table_summary', []))} 张, "
f"字段 {len(dict_data.get('data_dictionary', []))} 个)"
)
# ── 读 ──
def get(self, cfg: DBConfig) -> Optional[dict]:
return self._store.get(_cache_key(cfg))
# ── 取 / 缺失时现场抽 ──
def get_or_extract(self, cfg: DBConfig, log=None) -> dict:
cached = self.get(cfg)
if cached is not None:
logger.info(f"[Cache] 命中数据字典: {_cache_key(cfg)}")
return cached
logger.info(f"[Cache] 未命中,现场抽取: {_cache_key(cfg)}")
return extract_data_dictionary(cfg, log=log)
# ── 失效 ──
def invalidate(self, cfg: DBConfig) -> None:
self._store.pop(_cache_key(cfg), None)
\ No newline at end of file
...@@ -104,11 +104,11 @@ class Job: ...@@ -104,11 +104,11 @@ class Job:
created_at: datetime = field(default_factory=datetime.now) created_at: datetime = field(default_factory=datetime.now)
started_at: Optional[datetime] = None started_at: Optional[datetime] = None
finished_at: Optional[datetime] = None finished_at: Optional[datetime] = None
current_step: Optional[int] = None current_step: Optional[str] = None # step_id("merge_redundancy" 等)
current_step_title: Optional[str] = None current_step_title: Optional[str] = None
steps_planned: list[int] = field(default_factory=list) steps_planned: list[str] = field(default_factory=list)
steps_completed: list[int] = field(default_factory=list) steps_completed: list[str] = field(default_factory=list)
steps_failed: list[int] = field(default_factory=list) steps_failed: list[str] = field(default_factory=list)
error: Optional[str] = None error: Optional[str] = None
db_info: dict = field(default_factory=dict) db_info: dict = field(default_factory=dict)
logs: LogBuffer = field(default_factory=LogBuffer) logs: LogBuffer = field(default_factory=LogBuffer)
...@@ -187,9 +187,9 @@ class JobManager: ...@@ -187,9 +187,9 @@ class JobManager:
job.run_dir = run_dir job.run_dir = run_dir
logger.info(f"[job {job.job_id}] 运行目录: {run_dir}") logger.info(f"[job {job.job_id}] 运行目录: {run_dir}")
# 收集要跑的步骤(从 STEP_REGISTRY 取,保持单一来源;报告生成在主流程末尾无条件触发,不在此处) # 收集要跑的步骤(从 orchestrator 的注册表取,保持单一来源;报告生成在主流程末尾无条件触发,不在此处)
from .orchestrator import STEP_REGISTRY from .orchestrator import get_step_defs
all_steps = [n for n, *_ in STEP_REGISTRY] all_steps = [s.step_id for s in get_step_defs()]
steps_planned = job.req.steps if job.req.steps else all_steps steps_planned = job.req.steps if job.req.steps else all_steps
job.steps_planned = steps_planned job.steps_planned = steps_planned
logger.info(f"[job {job.job_id}] 计划步骤: {steps_planned}") logger.info(f"[job {job.job_id}] 计划步骤: {steps_planned}")
...@@ -218,18 +218,22 @@ class JobManager: ...@@ -218,18 +218,22 @@ class JobManager:
"ts": datetime.now().isoformat(), "ts": datetime.now().isoformat(),
"level": "INFO", "level": "INFO",
"step": None, "step": None,
"message": f"任务启动(job_id={job.job_id}, db={cfg.database}, steps={steps_planned})", "message": (
f"任务启动(job_id={job.job_id}, db={cfg.database}, "
f"steps={steps_planned}, tables={len(job.req.tables)} 张)"
),
}) })
try: try:
result = await run_governance_workflow( result = await run_governance_workflow(
cfg=cfg, cfg=cfg,
steps=steps_planned, steps=steps_planned,
tables=job.req.tables,
run_dir=run_dir, run_dir=run_dir,
enable_llm=job.req.enable_llm, enable_llm=job.req.enable_llm,
on_log=lambda entry: self._sync_log(job, entry), on_log=lambda entry: self._sync_log(job, entry),
on_step_start=lambda num, title: self._sync_step_start(job, num, title), on_step_start=lambda step_id, title: self._sync_step_start(job, step_id, title),
on_step_done=lambda num, title, ok: self._sync_step_done(job, num, title, ok), on_step_done=lambda step_id, title, ok: self._sync_step_done(job, step_id, title, ok),
on_section_update=lambda section_key, data: self._sync_section_update(job, section_key, data), on_section_update=lambda section_key, data: self._sync_section_update(job, section_key, data),
on_plan_updated=lambda main_steps: self._sync_plan_updated(job, main_steps), on_plan_updated=lambda main_steps: self._sync_plan_updated(job, main_steps),
cancel_event=job._cancel_event, cancel_event=job._cancel_event,
...@@ -271,17 +275,17 @@ class JobManager: ...@@ -271,17 +275,17 @@ class JobManager:
except RuntimeError: except RuntimeError:
job.logs._entries.append(entry) job.logs._entries.append(entry)
def _sync_step_start(self, job: Job, num: int, title: str) -> None: def _sync_step_start(self, job: Job, step_id: str, title: str) -> None:
job.current_step = num job.current_step = step_id
job.current_step_title = title job.current_step_title = title
def _sync_step_done(self, job: Job, num: int, title: str, ok: bool) -> None: def _sync_step_done(self, job: Job, step_id: str, title: str, ok: bool) -> None:
if ok: if ok:
job.steps_completed.append(num) job.steps_completed.append(step_id)
else: else:
job.steps_failed.append(num) job.steps_failed.append(step_id)
def _sync_plan_updated(self, job: Job, main_steps: list[int]) -> None: def _sync_plan_updated(self, job: Job, main_steps: list[str]) -> None:
"""Orchestrator 过滤依赖缺失的步骤后,同步实际计划列表到前端进度分母。 """Orchestrator 过滤依赖缺失的步骤后,同步实际计划列表到前端进度分母。
若不过滤:用户勾了 5 步但其中 1 步因依赖缺失被跳过,结果 若不过滤:用户勾了 5 步但其中 1 步因依赖缺失被跳过,结果
......
...@@ -20,8 +20,10 @@ class ConnectRequest(BaseModel): ...@@ -20,8 +20,10 @@ class ConnectRequest(BaseModel):
charset: str = "utf8mb4" charset: str = "utf8mb4"
connect_timeout: int = 10 connect_timeout: int = 10
# 可选:选择要跑的步骤(默认全 7 步;报告生成不在 steps 内,固定在主流程末尾触发) # 可选:要分析的步骤(按 step_id;为空表示跑全部)
steps: Optional[list[int]] = None steps: Optional[list[str]] = None
# 必填:要分析的表(前端从数据字典里勾选;空 = 禁止启动,由前端拦截)
tables: list[str] = Field(default_factory=list)
# 可选:是否启用 LLM 增强(默认 True,缺 Key 时自动降级) # 可选:是否启用 LLM 增强(默认 True,缺 Key 时自动降级)
enable_llm: bool = True enable_llm: bool = True
# 可选:报告标题 # 可选:报告标题
...@@ -29,8 +31,8 @@ class ConnectRequest(BaseModel): ...@@ -29,8 +31,8 @@ class ConnectRequest(BaseModel):
class TestConnectionRequest(ConnectRequest): class TestConnectionRequest(ConnectRequest):
"""只测连通性,不启动任务""" """只测连通性,不启动任务;前端不需要填 tables(会被忽略)。"""
pass tables: list[str] = Field(default_factory=list)
# ── 响应 ──────────────────────────────────────────────── # ── 响应 ────────────────────────────────────────────────
...@@ -40,13 +42,20 @@ class TestConnectionResponse(BaseModel): ...@@ -40,13 +42,20 @@ class TestConnectionResponse(BaseModel):
db_type: Optional[str] = None db_type: Optional[str] = None
database_name: Optional[str] = None # 改名为避免与 BaseModel.schema() 冲突 database_name: Optional[str] = None # 改名为避免与 BaseModel.schema() 冲突
tables_count: Optional[int] = None tables_count: Optional[int] = None
# 数据字典摘要:测试连接时同步抽出,供前端「表/字段浏览器」消费
# - table_summary 表级元数据(table_name / table_comment / table_rows / ...)
# - data_dictionary 字段级元数据(每行带 table_name)
# 字段太多时只返回表级,让前端按需再 /api/jobs 拉完整 —— 但目前没有这种场景,
# 这里直接给全量,避免再发一次请求。
table_summary: list[dict] = Field(default_factory=list)
data_dictionary: list[dict] = Field(default_factory=list)
class JobCreatedResponse(BaseModel): class JobCreatedResponse(BaseModel):
job_id: str job_id: str
status: str status: str
created_at: datetime created_at: datetime
steps_planned: list[int] steps_planned: list[str]
class JobStatusResponse(BaseModel): class JobStatusResponse(BaseModel):
...@@ -55,23 +64,25 @@ class JobStatusResponse(BaseModel): ...@@ -55,23 +64,25 @@ class JobStatusResponse(BaseModel):
created_at: datetime created_at: datetime
started_at: Optional[datetime] = None started_at: Optional[datetime] = None
finished_at: Optional[datetime] = None finished_at: Optional[datetime] = None
current_step: Optional[int] = None current_step: Optional[str] = None # 改 str:step_id(如 "merge_redundancy")
current_step_title: Optional[str] = None current_step_title: Optional[str] = None
progress_pct: float = 0.0 # 0-100 progress_pct: float = 0.0 # 0-100
steps_planned: list[int] = [] steps_planned: list[str] = []
steps_completed: list[int] = [] steps_completed: list[str] = []
steps_failed: list[int] = [] steps_failed: list[str] = []
error: Optional[str] = None error: Optional[str] = None
db_info: Optional[dict] = None db_info: Optional[dict] = None
class StepInfo(BaseModel): class StepInfo(BaseModel):
num: int # 稳定 ID("merge_redundancy" / "empty_fields" / ...),不再用数字编号 —— 扩展新 step 时
# 只在 orchestrator 的注册表加一行,UI 与后端都按 id 引用。
id: str
# 显示顺序(UI 按 order 升序渲染 checkbox / tab)
order: int = 0
title: str title: str
description: str description: str
requires_db: bool requires_db: bool
# 旧字段保留为向后兼容;新代码请用 llm_mode
llm_enhanced: bool = False
# "none" | "optional" | "required" # "none" | "optional" | "required"
llm_mode: str = "none" llm_mode: str = "none"
# 必选步骤:UI 上禁止取消勾选,后端也会兜底自动追加 # 必选步骤:UI 上禁止取消勾选,后端也会兜底自动追加
......
This diff is collapsed.
"""Step 2: 表合并候选与冗余字段分析(离线 + LLM 增强) """表合并候选与冗余字段分析(离线 + LLM 增强)
基于 Step 1 产出,计算: 基于 orchestrator 注入的数据字典,计算:
- merge_candidates: 结构高度相似的表组 - merge_candidates: 结构高度相似的表组
- decommission_candidates: 0 行表 + 可能的新表替代 - decommission_candidates: 0 行表 + 可能的新表替代
- redundancy_fields: 高频出现的疑似冗余字段 - redundancy_fields: 高频出现的疑似冗余字段
...@@ -29,48 +29,63 @@ OLD_PREFIXES = ["t_", "mall_", "project_"] ...@@ -29,48 +29,63 @@ OLD_PREFIXES = ["t_", "mall_", "project_"]
def run_step2(dict_data: dict, llm: LLMClient | None = None, def run_step2(dict_data: dict, llm: LLMClient | None = None,
log: Callable | None = None) -> dict: log: Callable | None = None,
table_filter: set[str] | None = None) -> dict:
"""合并候选 + 冗余字段分析。
Args:
table_filter: 前端勾选的表名集合(None/空 = 全部)。
"""
columns = dict_data.get("data_dictionary", []) columns = dict_data.get("data_dictionary", [])
table_summary = dict_data.get("table_summary", []) table_summary = dict_data.get("table_summary", [])
if not columns: if not columns:
if log: if log:
log("WARN", "未获取到任何字段元数据,跳过 Step 2", step="2") log("WARN", "未获取到任何字段元数据,跳过", step="merge_redundancy")
return {"merge_candidates": [], "decommission_candidates": {}, "redundancy_fields": []} return {"merge_candidates": [], "decommission_candidates": {}, "redundancy_fields": []}
# 表级过滤:保留被勾选的表(或全部)
if table_filter:
columns = [c for c in columns if c["table_name"] in table_filter]
table_summary = [t for t in table_summary if t["table_name"] in table_filter]
if log:
log("INFO",
f"按 table_filter 过滤后: {len(table_summary)} 张表, {len(columns)} 个字段",
step="merge_redundancy")
by_table: dict[str, list[dict]] = defaultdict(list) by_table: dict[str, list[dict]] = defaultdict(list)
for row in columns: for row in columns:
by_table[row["table_name"]].append(row) by_table[row["table_name"]].append(row)
if log: if log:
log("INFO", f"待分析表数: {len(by_table)}, 字段数: {len(columns)}", step="2") log("INFO", f"待分析表数: {len(by_table)}, 字段数: {len(columns)}", step="merge_redundancy")
# 1. 合并候选 # 1. 合并候选
if log: if log:
log("INFO", "[1/3] 计算合并候选(相似度阈值 ≥95%)...", step="2") log("INFO", "[1/3] 计算合并候选(相似度阈值 ≥95%)...", step="merge_redundancy")
merge = _find_merge(by_table, llm, log) merge = _find_merge(by_table, llm, log)
if log: if log:
log("INFO", f" · 发现 {len(merge)} 组可合并表", step="2") log("INFO", f" · 发现 {len(merge)} 组可合并表", step="merge_redundancy")
# 2. 废弃候选 # 2. 废弃候选
if log: if log:
log("INFO", "[2/3] 计算废弃候选(0 行表)...", step="2") log("INFO", "[2/3] 计算废弃候选(0 行表)...", step="merge_redundancy")
rows_by_table = {t["table_name"]: t.get("table_rows", 0) for t in table_summary} rows_by_table = {t["table_name"]: t.get("table_rows", 0) for t in table_summary}
decom = _find_decommission(rows_by_table, by_table.keys()) decom = _find_decommission(rows_by_table, by_table.keys())
if log: if log:
log("INFO", f" · 强废弃(有新版替代){decom['strong']['count']} 张, 弱废弃(无明确替代){decom['weak']['count']} 张", step="2") log("INFO", f" · 强废弃(有新版替代){decom['strong']['count']} 张, 弱废弃(无明确替代){decom['weak']['count']} 张", step="merge_redundancy")
# 3. 高频字段 # 3. 高频字段
if log: if log:
log("INFO", "[3/3] 计算高频字段(出现 ≥10 张表,LLM 分类)...", step="2") log("INFO", "[3/3] 计算高频字段(出现 ≥10 张表,LLM 分类)...", step="merge_redundancy")
redundancy = _find_redundancy(by_table, llm, log) redundancy = _find_redundancy(by_table, llm, log)
if log: if log:
log("INFO", f" · 高频字段 {len(redundancy)} 个", step="2") log("INFO", f" · 高频字段 {len(redundancy)} 个", step="merge_redundancy")
if log: if log:
log("INFO", log("INFO",
f"合并候选 {len(merge)} 组, 废弃候选 {decom['strong']['count'] + decom['weak']['count']} 张, " f"合并候选 {len(merge)} 组, 废弃候选 {decom['strong']['count'] + decom['weak']['count']} 张, "
f"高频字段 {len(redundancy)} 个", f"高频字段 {len(redundancy)} 个",
step="2") step="merge_redundancy")
return { return {
"merge_candidates": merge, "merge_candidates": merge,
...@@ -124,7 +139,7 @@ def _find_merge(by_table: dict, llm: LLMClient | None, log: Callable | None) -> ...@@ -124,7 +139,7 @@ def _find_merge(by_table: dict, llm: LLMClient | None, log: Callable | None) ->
if llm and llm.available: if llm and llm.available:
try: try:
if log: if log:
log("INFO", f"调用 LLM 评估合并候选 MERGE-{idx:03d}", step="2") log("INFO", f"调用 LLM 评估合并候选 MERGE-{idx:03d}", step="merge_redundancy")
llm_result = llm.suggest_table_merge( llm_result = llm.suggest_table_merge(
tables=[ tables=[
{"name": t, "comment": by_table[t][0].get("table_comment", ""), {"name": t, "comment": by_table[t][0].get("table_comment", ""),
...@@ -143,7 +158,7 @@ def _find_merge(by_table: dict, llm: LLMClient | None, log: Callable | None) -> ...@@ -143,7 +158,7 @@ def _find_merge(by_table: dict, llm: LLMClient | None, log: Callable | None) ->
risk = "; ".join(llm_result["risks"]) risk = "; ".join(llm_result["risks"])
except Exception as e: except Exception as e:
if log: if log:
log("WARN", f"LLM 评估失败,使用规则建议: {e}", step="2") log("WARN", f"LLM 评估失败,使用规则建议: {e}", step="merge_redundancy")
results.append({ results.append({
"id": f"MERGE-{idx:03d}", "id": f"MERGE-{idx:03d}",
...@@ -246,15 +261,15 @@ def _find_redundancy(by_table: dict, llm, log: Callable | None) -> list[dict]: ...@@ -246,15 +261,15 @@ def _find_redundancy(by_table: dict, llm, log: Callable | None) -> list[dict]:
log("INFO", log("INFO",
f" · LLM 分类批次 [{idx}/{total_batches}] " f" · LLM 分类批次 [{idx}/{total_batches}] "
f"({len(batch)} 个字段)", f"({len(batch)} 个字段)",
step="2") step="merge_redundancy")
try: try:
results = llm.classify_redundant_fields_batch(batch) results = llm.classify_redundant_fields_batch(batch)
except Exception as e: except Exception as e:
if log: if log:
log("ERROR", log("ERROR",
f" · LLM 分类批次 [{idx}/{total_batches}] 整体失败: {e}(任务将终止)", f" · LLM 分类批次 [{idx}/{total_batches}] 整体失败: {e}(任务将终止)",
step="2") step="merge_redundancy")
logger.exception("Step 2 LLM 分类批次失败") logger.exception("LLM 分类批次失败")
raise # required step → 抛给上层 raise # required step → 抛给上层
annotated.extend(results) annotated.extend(results)
...@@ -287,5 +302,5 @@ def _find_redundancy(by_table: dict, llm, log: Callable | None) -> list[dict]: ...@@ -287,5 +302,5 @@ def _find_redundancy(by_table: dict, llm, log: Callable | None) -> list[dict]:
cls_count[r["classification"]] = cls_count.get(r["classification"], 0) + 1 cls_count[r["classification"]] = cls_count.get(r["classification"], 0) + 1
summary = ", ".join(f"{k}={v}" for k, v in cls_count.items()) summary = ", ".join(f"{k}={v}" for k, v in cls_count.items())
if log: if log:
log("INFO", f" · LLM 分类结果: {summary}", step="2") log("INFO", f" · LLM 分类结果: {summary}", step="merge_redundancy")
return out return out
\ No newline at end of file
"""Step 3: 大范围空字段扫描(纯程序化检查) """空字段扫描(纯程序化检查)
对每张表跑动态 SQL 统计: 对每张表跑动态 SQL 统计:
- COUNT(*) 总行数 - COUNT(*) 总行数
...@@ -30,32 +30,37 @@ def run_step4(cfg: DBConfig, log: Callable | None = None, ...@@ -30,32 +30,37 @@ def run_step4(cfg: DBConfig, log: Callable | None = None,
min_rows_to_check: int = 5, min_rows_to_check: int = 5,
high_threshold: float = 0.80, high_threshold: float = 0.80,
mid_threshold: float = 0.50, mid_threshold: float = 0.50,
sample_size: int = 3) -> dict: sample_size: int = 3,
table_filter: set[str] | None = None) -> dict:
"""扫描所有表的空字段情况(纯程序化,不依赖 LLM) """扫描所有表的空字段情况(纯程序化,不依赖 LLM)
Args: Args:
dict_data: 复用 Step 1 已拿到的数据字典(节省一次数据库连接/查询)。 dict_data: 必填 —— orchestrator 在任务启动时已通过 DataDictCache 拿到(或现场重抽)。
缺省则调用 Step 1 重取。 table_filter: 前端勾选的表名集合(None/空 = 全部)。
""" """
if log: if log:
log("INFO", log("INFO",
f"Step 3 开始扫描空字段 (high≥{high_threshold:.0%}, mid≥{mid_threshold:.0%}, min_rows={min_rows_to_check})", f"开始扫描空字段 (high≥{high_threshold:.0%}, mid≥{mid_threshold:.0%}, min_rows={min_rows_to_check})",
step="3") step="empty_fields")
if dict_data and dict_data.get("data_dictionary"): if not (dict_data and dict_data.get("data_dictionary")):
# 复用 Step 1 产出(Wave 调度器已保证 Step 1 先完成)
columns = dict_data["data_dictionary"]
table_summary = dict_data.get("table_summary", [])
if log: if log:
log("INFO", " · 复用 Step 1 数据字典", step="3") log("ERROR", "dict_data 为空,请检查 orchestrator 是否正确传入", step="empty_fields")
else: return {"summary": {}, "per_table": [], "skipped": []}
# 兜底:没拿到 Step 1 产出时回退自取(独立跑 Step 3 时用得到)
from .step1_data_dict import run_step1 columns = dict_data["data_dictionary"]
dict_data = run_step1(cfg, log=lambda lvl, msg, **_kw: log(lvl, msg, step="3") if log else None) table_summary = dict_data.get("table_summary", [])
columns = dict_data["data_dictionary"] if log:
table_summary = dict_data.get("table_summary", []) log("INFO", " · 复用 orchestrator 注入的数据字典", step="empty_fields")
# 表级过滤
if table_filter:
columns = [c for c in columns if c["table_name"] in table_filter]
table_summary = [t for t in table_summary if t["table_name"] in table_filter]
if log: if log:
log("INFO", " · 独立拉取数据字典(未复用 Step 1)", step="3") log("INFO",
f"按 table_filter 过滤后: {len(table_summary)} 张表, {len(columns)} 个字段",
step="empty_fields")
by_table: dict[str, list[dict]] = defaultdict(list) by_table: dict[str, list[dict]] = defaultdict(list)
for row in columns: for row in columns:
...@@ -88,15 +93,15 @@ def run_step4(cfg: DBConfig, log: Callable | None = None, ...@@ -88,15 +93,15 @@ def run_step4(cfg: DBConfig, log: Callable | None = None,
table_names = sorted(by_table.keys()) table_names = sorted(by_table.keys())
total = len(table_names) total = len(table_names)
if log: if log:
log("INFO", f"将扫描 {total} 张表(已过滤 TEXT_TYPES)", step="3") log("INFO", f"将扫描 {total} 张表(已过滤 TEXT_TYPES)", step="empty_fields")
try: try:
with open_db(cfg) as db: with open_db(cfg) as db:
for idx, table in enumerate(table_names, 1): for idx, table in enumerate(table_names, 1):
if log and idx % 10 == 0: if log and idx % 10 == 0:
log("INFO", f" 进度 {idx}/{total} - 当前表: {table}", step="3") log("INFO", f" 进度 {idx}/{total} - 当前表: {table}", step="empty_fields")
elif log and idx == 1: elif log and idx == 1:
log("INFO", f" 进度 1/{total} - 当前表: {table}", step="3") log("INFO", f" 进度 1/{total} - 当前表: {table}", step="empty_fields")
# 拿实际行数(避免 TABLE_ROWS 估算不准) # 拿实际行数(避免 TABLE_ROWS 估算不准)
try: try:
...@@ -109,7 +114,7 @@ def run_step4(cfg: DBConfig, log: Callable | None = None, ...@@ -109,7 +114,7 @@ def run_step4(cfg: DBConfig, log: Callable | None = None,
except Exception as e: except Exception as e:
skipped.append({"table": table, "reason": f"统计行数失败: {e}"}) skipped.append({"table": table, "reason": f"统计行数失败: {e}"})
if log: if log:
log("WARN", f" · {table} 行数统计失败: {e}", step="3") log("WARN", f" · {table} 行数统计失败: {e}", step="empty_fields")
continue continue
if row_count < min_rows_to_check: if row_count < min_rows_to_check:
...@@ -134,7 +139,7 @@ def run_step4(cfg: DBConfig, log: Callable | None = None, ...@@ -134,7 +139,7 @@ def run_step4(cfg: DBConfig, log: Callable | None = None,
except Exception as e: except Exception as e:
skipped.append({"table": table, "reason": f"扫描失败: {e}"}) skipped.append({"table": table, "reason": f"扫描失败: {e}"})
if log: if log:
log("WARN", f" · {table} 空字段扫描失败: {e}", step="3") log("WARN", f" · {table} 空字段扫描失败: {e}", step="empty_fields")
continue continue
if not row: if not row:
...@@ -189,8 +194,8 @@ def run_step4(cfg: DBConfig, log: Callable | None = None, ...@@ -189,8 +194,8 @@ def run_step4(cfg: DBConfig, log: Callable | None = None,
logger.debug(f"表 {table}: 高空 {len(high_in_table)}, 中空 {len(mid_in_table)}") logger.debug(f"表 {table}: 高空 {len(high_in_table)}, 中空 {len(mid_in_table)}")
except Exception as e: except Exception as e:
if log: if log:
log("ERROR", f"Step 3 数据库扫描失败: {type(e).__name__}: {e}", step="3") log("ERROR", f"数据库扫描失败: {type(e).__name__}: {e}", step="empty_fields")
logger.exception("Step 3 详细异常") logger.exception("详细异常")
top_tables.sort(key=lambda x: x["high_empty_field_count"], reverse=True) top_tables.sort(key=lambda x: x["high_empty_field_count"], reverse=True)
...@@ -210,7 +215,7 @@ def run_step4(cfg: DBConfig, log: Callable | None = None, ...@@ -210,7 +215,7 @@ def run_step4(cfg: DBConfig, log: Callable | None = None,
log("INFO", log("INFO",
f"扫描完成: 高空 {high_count}, 中空 {mid_count}, 涉及 {len(tables_with_issues)} 张表, " f"扫描完成: 高空 {high_count}, 中空 {mid_count}, 涉及 {len(tables_with_issues)} 张表, "
f"跳过 {len(skipped)} 张", f"跳过 {len(skipped)} 张",
step="3") step="empty_fields")
return { return {
"summary": summary, "summary": summary,
......
"""Step 4: 缺失注释字段检查 + LLM 推测(必跑 LLM) """缺失注释字段检查 + LLM 推测(必跑 LLM)
字段注释的语义判断本质上需要 LLM: 字段注释的语义判断本质上需要 LLM:
- 拼音首字母缩写(如 xzqhbm)、业务缩写、英文组合 → 必须由 LLM 翻译 - 拼音首字母缩写(如 xzqhbm)、业务缩写、英文组合 → 必须由 LLM 翻译
...@@ -23,15 +23,27 @@ logger = logging.getLogger(__name__) ...@@ -23,15 +23,27 @@ logger = logging.getLogger(__name__)
def run_step5(dict_data: dict, llm: LLMClient | None = None, def run_step5(dict_data: dict, llm: LLMClient | None = None,
log: Callable | None = None) -> dict: log: Callable | None = None,
table_filter: set[str] | None = None) -> dict:
"""缺失注释字段 + LLM 推测。
Args:
table_filter: 前端勾选的表名集合(None/空 = 全部)。
"""
columns = dict_data.get("data_dictionary", []) columns = dict_data.get("data_dictionary", [])
if table_filter:
columns = [c for c in columns if c["table_name"] in table_filter]
if log:
log("INFO",
f"按 table_filter 过滤后: {len(columns)} 个字段",
step="missing_comments")
if not columns: if not columns:
if log: if log:
log("WARN", "未获取到任何字段元数据,跳过 Step 4", step="4") log("WARN", "未获取到任何字段元数据,跳过", step="missing_comments")
return {"summary": {}, "by_table": [], "predicted_comments": [], "unpredictable_sample": []} return {"summary": {}, "by_table": [], "predicted_comments": [], "unpredictable_sample": []}
if log: if log:
log("INFO", f"待检查字段总数: {len(columns)}", step="4") log("INFO", f"待检查字段总数: {len(columns)}", step="missing_comments")
# 1. 收集所有无注释字段,按 (table_name, table_comment) 分组 # 1. 收集所有无注释字段,按 (table_name, table_comment) 分组
missing = [] missing = []
...@@ -60,12 +72,12 @@ def run_step5(dict_data: dict, llm: LLMClient | None = None, ...@@ -60,12 +72,12 @@ def run_step5(dict_data: dict, llm: LLMClient | None = None,
if log: if log:
log("INFO", log("INFO",
f" · 缺失注释字段: {len(missing)} 个 (覆盖 {len(by_table)} 张表)", f" · 缺失注释字段: {len(missing)} 个 (覆盖 {len(by_table)} 张表)",
step="4") step="missing_comments")
# 2. 按表分批调用 LLM(必跑;单批失败会让任务终止) # 2. 按表分批调用 LLM(必跑;单批失败会让任务终止)
if not missing: if not missing:
if log: if log:
log("INFO", " · 无缺失注释字段,跳过 LLM", step="4") log("INFO", " · 无缺失注释字段,跳过 LLM", step="missing_comments")
return _empty_result(by_table) return _empty_result(by_table)
BATCH = 30 BATCH = 30
...@@ -81,7 +93,7 @@ def run_step5(dict_data: dict, llm: LLMClient | None = None, ...@@ -81,7 +93,7 @@ def run_step5(dict_data: dict, llm: LLMClient | None = None,
log("INFO", log("INFO",
f" · LLM 推测 [{llm_called}] {tname} " f" · LLM 推测 [{llm_called}] {tname} "
f"({chunk_idx}/{len(chunks)} 批, {len(chunk)} 个字段)", f"({chunk_idx}/{len(chunks)} 批, {len(chunk)} 个字段)",
step="4") step="missing_comments")
try: try:
results = llm.predict_field_comments_batch( results = llm.predict_field_comments_batch(
table_name=tname, table_name=tname,
...@@ -94,8 +106,8 @@ def run_step5(dict_data: dict, llm: LLMClient | None = None, ...@@ -94,8 +106,8 @@ def run_step5(dict_data: dict, llm: LLMClient | None = None,
if log: if log:
log("ERROR", log("ERROR",
f" · LLM 推测失败({tname} 第 {chunk_idx} 批): {e}(任务将终止)", f" · LLM 推测失败({tname} 第 {chunk_idx} 批): {e}(任务将终止)",
step="4") step="missing_comments")
logger.exception("Step 4 LLM 推测批次失败") logger.exception("LLM 推测批次失败")
raise raise
# 把 LLM 结果写回 entry # 把 LLM 结果写回 entry
...@@ -121,11 +133,11 @@ def run_step5(dict_data: dict, llm: LLMClient | None = None, ...@@ -121,11 +133,11 @@ def run_step5(dict_data: dict, llm: LLMClient | None = None,
log("INFO", log("INFO",
f" · LLM 命中 {len(predicted)}/{len(missing)}, " f" · LLM 命中 {len(predicted)}/{len(missing)}, "
f"LLM 解析失败 {len(unpredicted)}", f"LLM 解析失败 {len(unpredicted)}",
step="4") step="missing_comments")
log("INFO", log("INFO",
f"缺失注释 {len(missing)} 个字段,覆盖 {len(by_table)} 张表;" f"缺失注释 {len(missing)} 个字段,覆盖 {len(by_table)} 张表;"
f"LLM 推测成功 {len(predicted)}", f"LLM 推测成功 {len(predicted)}",
step="4") step="missing_comments")
return { return {
"summary": { "summary": {
......
"""Step 5: 字段长度检查(纯规则匹配) """字段长度检查(纯规则匹配)
识别字段定义长度超出标准所需(如身份证号 18 位用 varchar(50) 存储)。 识别字段定义长度超出标准所需(如身份证号 18 位用 varchar(50) 存储)。
仅基于内置规则匹配,不依赖 LLM。 仅基于内置规则匹配,不依赖 LLM。
...@@ -152,17 +152,25 @@ def _match_rule(col_name: str, col_comment: str, rule: LengthRule) -> tuple[bool ...@@ -152,17 +152,25 @@ def _match_rule(col_name: str, col_comment: str, rule: LengthRule) -> tuple[bool
return False, "" return False, ""
def run_step6(dict_data: dict, log: Callable | None = None) -> dict: def run_step6(dict_data: dict, log: Callable | None = None,
table_filter: set[str] | None = None) -> dict:
"""字段长度检查(纯规则匹配)。
Args:
table_filter: 前端勾选的表名集合(None/空 = 全部)。
"""
columns = dict_data.get("data_dictionary", []) columns = dict_data.get("data_dictionary", [])
if table_filter:
columns = [c for c in columns if c["table_name"] in table_filter]
if not columns: if not columns:
if log: if log:
log("WARN", "未获取到任何字段元数据,跳过 Step 5", step="5") log("WARN", "未获取到任何字段元数据,跳过", step="length_check")
return {"summary": {"total_issues": 0}, "issues": []} return {"summary": {"total_issues": 0}, "issues": []}
if log: if log:
log("INFO", log("INFO",
f"内置长度规则: {len(LENGTH_RULES)} 条, 待扫描字段: {len(columns)}", f"内置长度规则: {len(LENGTH_RULES)} 条, 待扫描字段: {len(columns)}",
step="5") step="length_check")
issues = [] issues = []
match_by_name = 0 match_by_name = 0
...@@ -209,7 +217,7 @@ def run_step6(dict_data: dict, log: Callable | None = None) -> dict: ...@@ -209,7 +217,7 @@ def run_step6(dict_data: dict, log: Callable | None = None) -> dict:
log("INFO", log("INFO",
f"规则匹配: 字段名 {match_by_name} 次 + 注释关键字 {match_by_comment} 次," f"规则匹配: 字段名 {match_by_name} 次 + 注释关键字 {match_by_comment} 次,"
f"发现异常 {len(issues)} 条", f"发现异常 {len(issues)} 条",
step="5") step="length_check")
return { return {
"summary": { "summary": {
......
"""Step 6: 国家标准字段校验(连库) """国家标准字段校验(连库)
按字段名匹配 standards 插件,抽样校验实际数据: 按字段名匹配 standards 插件,抽样校验实际数据:
- 身份证号(GB 11643) - 身份证号(GB 11643)
...@@ -45,9 +45,20 @@ SAMPLE_SPECS: list[dict] = [ ...@@ -45,9 +45,20 @@ SAMPLE_SPECS: list[dict] = [
] ]
def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> dict: def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None,
"""执行 Step 6:国标校验""" table_filter: set[str] | None = None) -> dict:
"""执行国标校验
Args:
table_filter: 前端勾选的表名集合(None/空 = 全部)。
"""
columns = dict_data.get("data_dictionary", []) columns = dict_data.get("data_dictionary", [])
if table_filter:
columns = [c for c in columns if c["table_name"] in table_filter]
if log:
log("INFO",
f"按 table_filter 过滤后: {len(columns)} 个字段",
step="standards")
if not columns: if not columns:
return {"summary": {}, "violations_by_standard": [], "violations": []} return {"summary": {}, "violations_by_standard": [], "violations": []}
...@@ -67,7 +78,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di ...@@ -67,7 +78,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di
if not standards_by_field: if not standards_by_field:
if log: if log:
log("WARN", "未发现任何标准插件", step="6") log("WARN", "未发现任何标准插件", step="standards")
return { return {
"summary": {"standards_applied": 0, "fields_checked": 0, "violations": 0}, "summary": {"standards_applied": 0, "fields_checked": 0, "violations": 0},
"violations_by_standard": [], "violations_by_standard": [],
...@@ -79,7 +90,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di ...@@ -79,7 +90,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di
f"已加载 {len(all_standards_info)} 个标准插件, " f"已加载 {len(all_standards_info)} 个标准插件, "
f"匹配字段 {len(standards_by_field)} 个, " f"匹配字段 {len(standards_by_field)} 个, "
f"最大每字段抽样 {SAMPLE_LIMIT} 行 / 每字段最多 {MAX_TABLES_PER_FIELD} 张表", f"最大每字段抽样 {SAMPLE_LIMIT} 行 / 每字段最多 {MAX_TABLES_PER_FIELD} 张表",
step="6") step="standards")
loader = get_sql_loader() loader = get_sql_loader()
...@@ -90,7 +101,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di ...@@ -90,7 +101,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di
if log: if log:
log("DEBUG", log("DEBUG",
f" · [{std_idx}/{len(standards_by_field)}] 标准字段 {std_field} 未出现在任何表中, 跳过", f" · [{std_idx}/{len(standards_by_field)}] 标准字段 {std_field} 未出现在任何表中, 跳过",
step="6") step="standards")
continue continue
tables_with_field = by_name[std_field][:MAX_TABLES_PER_FIELD] tables_with_field = by_name[std_field][:MAX_TABLES_PER_FIELD]
std_id = std_instance.standard_id std_id = std_instance.standard_id
...@@ -107,7 +118,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di ...@@ -107,7 +118,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di
log("INFO", log("INFO",
f" · [{std_idx}/{len(standards_by_field)}] " f" · [{std_idx}/{len(standards_by_field)}] "
f"{std_id} ({std_field}) → {len(tables_with_field)} 张表", f"{std_id} ({std_field}) → {len(tables_with_field)} 张表",
step="6") step="standards")
for col_record in tables_with_field: for col_record in tables_with_field:
table = col_record["table_name"] table = col_record["table_name"]
...@@ -139,7 +150,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di ...@@ -139,7 +150,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di
if log: if log:
log("WARN", log("WARN",
f" - {table}.{field_name} 抽样失败: {e}", f" - {table}.{field_name} 抽样失败: {e}",
step="6") step="standards")
continue continue
sampled = len(rows) sampled = len(rows)
...@@ -188,8 +199,8 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di ...@@ -188,8 +199,8 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di
) )
except Exception as e: except Exception as e:
if log: if log:
log("ERROR", f"Step 6 连库校验失败: {type(e).__name__}: {e}", step="6") log("ERROR", f"连库校验失败: {type(e).__name__}: {e}", step="standards")
logger.exception("Step 6 详细异常") logger.exception("详细异常")
by_standard_list = [] by_standard_list = []
for std_id, rec in by_standard.items(): for std_id, rec in by_standard.items():
...@@ -198,7 +209,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di ...@@ -198,7 +209,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di
if log: if log:
log("INFO", log("INFO",
f" · 标准 {rec['standard']} 抽样 {rec['total_sampled']} 条全部合规,跳过", f" · 标准 {rec['standard']} 抽样 {rec['total_sampled']} 条全部合规,跳过",
step="6") step="standards")
continue continue
violating_fields = [d for d in rec["fields"] if d.get("violations", 0) > 0] violating_fields = [d for d in rec["fields"] if d.get("violations", 0) > 0]
by_standard_list.append({ by_standard_list.append({
...@@ -219,7 +230,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di ...@@ -219,7 +230,7 @@ def run_step7(cfg: DBConfig, dict_data: dict, log: Callable | None = None) -> di
}) })
if log: if log:
log("INFO", f"校验完成: {len(by_standard_list)} 个标准有违规, {total_fields_checked} 个字段, {total_violations} 条违规", step="6") log("INFO", f"校验完成: {len(by_standard_list)} 个标准有违规, {total_fields_checked} 个字段, {total_violations} 条违规", step="standards")
return { return {
"summary": { "summary": {
......
This diff is collapsed.
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment