Commit 5d3c7883 authored by Data Governance Dev's avatar Data Governance Dev

feat(web): 报告内容完整化(Markdown + docx 全量明细)

原 step8_report.py 两个问题:

1. Markdown 每个章节都截前 10-20 条样本,详细信息缺失
2. docx 后半部分只是 stub:标题 + f"{summary}" 把整个 dict 塞进段落,
   浏览器/Word 里看到的就是一坨 JSON 风格的字符串,没有表格

重写方案:两个版本共用 SECTION_ORDER + 同名渲染函数(_md_xxx / _docx_xxx),
保证两边结构对齐。

新增/补齐的章节内容:

- 表合并候选:完整候选明细(ID/优先级/相似度/公共列数/差异列/建议/风险)
  + 各候选组的公共列清单 + 强/弱废弃候选分开列表
- 冗余字段:全部高频字段(不再截 20)+ LLM 失败标「⚠️ 待人工核对」
- 大范围空字段:阈值说明 + 按表汇总(高空数倒序) + 字段明细按表分组
  (每张表高空/中空字段单独子表,含空值率) + 跳过表清单
- 缺失注释字段:summary(涉及表/LLM 调用次数)+ 按表分布 + 推测注释明细
  (全列) + 未推测样本
- 字段长度异常:6 个汇总指标 + 全部异常字段(按 wasted_bytes desc 排,
  10 列含国标/规则/依据)
- 国标违规:按标准统计 + 每标准下的违规字段明细 + 违规样本(最多 500 条)

辅助细节:
- Markdown 表格换行用 <br>、管道符转义 \|、空值显示 -、浮点格式 xx.xx%
- docx 用原生 add_table + style 'Light Grid Accent 1',Heading 1/2/3 分层
- 顶层 docstring 说明新结构;SECTION_ORDER 单一来源驱动两版渲染

smoke test:拿真实 outputs/smart-build/20260804_085733/findings/_all_findings.json
跑过,Markdown 11425 字符 / 255 行,docx 24 段 + 7 张表格,数据正确。
parent 3687f62d
......@@ -3,6 +3,8 @@
读取前 7 步的 sections 汇总,生成:
- {dbname}.md - Markdown 报告(可读、可编辑)
- {dbname}.docx - Word 报告(业务方用)
两个版本都按章节铺完整明细表格,不再截断前 N 条样本。
"""
from __future__ import annotations
......@@ -15,6 +17,18 @@ from typing import Any, Callable
logger = logging.getLogger(__name__)
# ── 报告章节配置 ──
# 每个章节的标题 + section key + 渲染函数(md / docx 各一个)
SECTION_ORDER = [
("merge_candidates", "表合并候选"),
("redundancy_fields", "冗余字段"),
("empty_fields", "大范围空字段"),
("missing_comments", "缺失注释的字段"),
("length_issues", "字段长度异常"),
("standard_violations", "不符合国家标准的字段"),
]
def generate_reports(run_dir: Path, sections: dict, overview: dict,
log: Callable | None = None) -> tuple[str | None, str | None]:
"""生成 Markdown + docx 报告。返回 (md_path, docx_path)。"""
......@@ -40,8 +54,6 @@ def generate_reports(run_dir: Path, sections: dict, overview: dict,
if log:
log("INFO", f"[2/2] 生成 Word 报告 → {docx_path.name}", step="8")
from docx import Document
from docx.shared import Pt, Cm, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
doc = Document()
_render_docx(doc, sections, overview)
doc.save(str(docx_path))
......@@ -59,15 +71,21 @@ def generate_reports(run_dir: Path, sections: dict, overview: dict,
return str(md_path), docx_result
# ════════════════════════════════════════════════════════════════
# Markdown 渲染
# ════════════════════════════════════════════════════════════════
def _build_markdown(sections: dict, overview: dict) -> str:
"""拼装完整 Markdown 报告"""
lines: list[str] = []
# 标题
# 标题 + 头部
lines.append(f"# 数据治理报告 — {overview.get('database', '')}")
lines.append("")
lines.append(f"> 生成时间: {overview.get('executed_at', datetime.now().isoformat())}")
lines.append(f"> 数据库类型: {overview.get('db_type', '-')}")
lines.append(f"> 主机: {overview.get('host', '-')}")
lines.append("")
# 概览
......@@ -75,204 +93,906 @@ def _build_markdown(sections: dict, overview: dict) -> str:
lines.append("")
lines.append("| 项目 | 值 |")
lines.append("|------|-----|")
lines.append(f"| 数据库名 | {overview.get('database', '-')} |")
lines.append(f"| 数据库类型 | {overview.get('db_type', '-')} |")
lines.append(f"| 主机 | {overview.get('host', '-')} |")
lines.append(f"| 表总数 | {overview.get('total_tables', 0)} |")
lines.append(f"| 字段总数 | {overview.get('total_fields', 0)} |")
lines.append(f"| 大范围空字段(≥80%) | {overview.get('high_empty_fields', 0)} |")
lines.append(f"| 中范围空字段(≥50%) | {overview.get('mid_empty_fields', 0)} |")
lines.append(f"| 缺失注释字段 | {overview.get('missing_comments', 0)} |")
lines.append(f"| 字段长度异常 | {overview.get('length_issues', 0)} |")
lines.append(f"| 国标违规记录 | {overview.get('standard_violations', 0)} |")
lines.append(f"| 表合并候选 | {overview.get('merge_candidates', 0)} |")
lines.append(f"| 冗余字段 | {overview.get('redundancy_fields', 0)} |")
overview_rows = [
("数据库名", overview.get("database", "-")),
("数据库类型", overview.get("db_type", "-")),
("主机", overview.get("host", "-")),
("表总数", overview.get("total_tables", 0)),
("字段总数", overview.get("total_fields", 0)),
("大范围空字段(≥80%)", overview.get("high_empty_fields", 0)),
("中范围空字段(≥50%)", overview.get("mid_empty_fields", 0)),
("缺失注释字段", overview.get("missing_comments", 0)),
("字段长度异常", overview.get("length_issues", 0)),
("国标违规记录", overview.get("standard_violations", 0)),
("表合并候选", overview.get("merge_candidates", 0)),
("冗余字段", overview.get("redundancy_fields", 0)),
]
for k, v in overview_rows:
lines.append(f"| {k} | {v} |")
lines.append("")
# 表合并
_section_merge(lines, sections.get("merge_candidates", {}))
# 章节索引
lines.append("## 章节索引")
lines.append("")
for idx, (key, title) in enumerate(SECTION_ORDER, start=2):
lines.append(f"- [{idx}、{title}](#{_md_anchor(title)})")
lines.append("")
# 冗余字段
_section_redundancy(lines, sections.get("redundancy_fields", {}))
# 各章节
for idx, (key, title) in enumerate(SECTION_ORDER, start=2):
lines.append(f"## {idx}、{title}")
lines.append("")
_render_section_md(lines, key, sections.get(key, {}))
lines.append("")
# 空字段
_section_empty(lines, sections.get("empty_fields", {}))
return "\n".join(lines)
# 缺失注释
_section_missing(lines, sections.get("missing_comments", {}))
# 字段长度
_section_length(lines, sections.get("length_issues", {}))
def _md_anchor(title: str) -> str:
"""生成 markdown anchor(去掉中文标点 + 小写)"""
return title.replace(" ", "-").replace("、", "").replace("(", "-").replace(")", "")
# 国标违规
_section_standards(lines, sections.get("standard_violations", {}))
return "\n".join(lines)
def _render_section_md(lines: list[str], key: str, data: dict):
"""按 key 调用对应渲染函数"""
handler = _MD_SECTION_HANDLERS.get(key)
if handler:
handler(lines, data)
else:
lines.append("(无渲染函数)")
def _section_merge(lines: list[str], data: dict):
lines.append("## 二、表合并候选")
cands = data.get("merge_candidates", [])
if not cands:
lines.append("(未发现可合并表)")
def _md_row(values: list) -> str:
"""构造一行 markdown 表格,| 默认转义"""
return "| " + " | ".join(_md_cell(v) for v in values) + " |"
def _md_cell(v: Any) -> str:
"""单格转义:None → '-';包含 | / 换行 → 用 <br> 包裹并替换换行"""
if v is None or v == "":
return "-"
s = str(v)
# 替换换行为 <br>,避免破坏 markdown 行
s = s.replace("\r\n", "<br>").replace("\n", "<br>").replace("\r", "<br>")
# 转义管道符
s = s.replace("|", "\\|")
return s
# ── 各章节 markdown 渲染 ─────────────────────────────────────────
def _md_merge(lines: list[str], data: dict):
candidates = data.get("merge_candidates") or []
decommission = data.get("decommission_candidates") or {}
strong = (decommission.get("strong") or {}).get("tables", []) if isinstance(decommission, dict) else []
weak = (decommission.get("weak") or {}).get("tables", []) if isinstance(decommission, dict) else []
lines.append(f"候选合并组: **{len(candidates)}** | "
f"强废弃候选: **{len(strong)}** | "
f"弱废弃候选: **{len(weak)}**")
lines.append("")
if not candidates:
lines.append("(未发现可合并的表)")
lines.append("")
return
lines.append("| ID | 优先级 | 表 | 相似度 | 建议 |")
lines.append("|----|--------|------|--------|------|")
for c in cands:
# 合并候选明细
lines.append("### 合并候选明细")
lines.append("")
lines.append(_md_row(["ID", "优先级", "相似度", "表数", "公共列数", "表", "差异列", "建议", "风险"]))
lines.append("|---|---|---|---|---|---|---|---|---|")
for c in candidates:
tables = ", ".join(c.get("tables", []))
lines.append(f"| {c.get('id','')} | {c.get('priority','')} | {tables} | "
f"{c.get('similarity', '')} | {c.get('suggestion', '')[:60]} |")
diff_cols = ", ".join(c.get("diff_columns", []) or []) or "-"
lines.append(_md_row([
c.get("id", ""),
c.get("priority", ""),
c.get("similarity", ""),
len(c.get("tables", [])),
c.get("common_columns", ""),
tables,
diff_cols,
(c.get("suggestion", "") or "")[:120],
c.get("risk", ""),
]))
lines.append("")
decom = data.get("decommission_candidates", {})
strong = decom.get("strong", {}).get("tables", [])
# 公共列明细
lines.append("### 各候选组的公共列")
lines.append("")
for c in candidates:
cols = c.get("columns", []) or []
lines.append(f"- **{c.get('id', '')}** ({', '.join(c.get('tables', []))}): {', '.join(cols) if cols else '(无)'}")
lines.append("")
# 强废弃候选
if strong:
lines.append("### 废弃候选(有新版替代)")
lines.append("| 旧表 | 建议替代 |")
lines.append("|------|----------|")
lines.append("### 强废弃候选(有明确新表替代)")
lines.append("")
lines.append(_md_row(["旧表", "建议替代", "证据"]))
lines.append("|---|---|---|")
for d in strong:
lines.append(f"| {d.get('table','')} | {d.get('replacement','')} |")
lines.append(_md_row([d.get("table", ""), d.get("replacement", ""), d.get("evidence", "")]))
lines.append("")
# 弱废弃候选
if weak:
lines.append("### 弱废弃候选(疑似可废弃)")
lines.append("")
lines.append(_md_row(["表", "相似表", "相似度"]))
lines.append("|---|---|---|")
for d in weak:
lines.append(_md_row([d.get("table", ""), d.get("similar_to", ""), d.get("similarity", "")]))
lines.append("")
def _section_redundancy(lines: list[str], data: dict):
lines.append("## 三、冗余字段")
fields = data.get("redundancy_fields", [])
def _md_redundancy(lines: list[str], data: dict):
fields = data.get("redundancy_fields") or []
lines.append(f"高频字段数(出现在 ≥10 张表): **{len(fields)}**")
lines.append("")
if not fields:
lines.append("(未发现明显冗余字段)")
lines.append("")
return
lines.append("| 字段 | 出现表数 | 风险 | 建议 |")
lines.append("|------|----------|------|------|")
lines.append(_md_row(["字段", "出现表数", "分类", "LLM 推理", "建议处置"]))
lines.append("|---|---|---|---|---|")
for f in fields:
lines.append(f"| {f.get('field','')} | {f.get('table_count','')} | "
f"{f.get('risk','')} | {f.get('suggestion','')[:60]} |")
source = f.get("source", "")
reasoning = f.get("reasoning", "") or ""
if source == "llm_failed":
reasoning = "⚠️ LLM 推理失败(待人工核对)"
lines.append(_md_row([
f.get("field", ""),
f.get("table_count", ""),
f.get("classification", ""),
reasoning,
f.get("recommendation", ""),
]))
lines.append("")
def _section_empty(lines: list[str], data: dict):
summary = data.get("summary", {})
lines.append("## 四、大范围空字段")
lines.append(f"扫描表数: **{summary.get('total_tables_scanned', 0)}**, "
f"高空字段: **{summary.get('high_empty_fields_count', 0)}**, "
f"中空字段: **{summary.get('mid_empty_fields_count', 0)}**")
def _md_empty(lines: list[str], data: dict):
summary = data.get("summary") or {}
per_table = data.get("per_table") or []
skipped = data.get("skipped") or []
total_tables = summary.get("total_tables_scanned", 0)
high_count = summary.get("high_empty_fields_count", 0)
mid_count = summary.get("mid_empty_fields_count", 0)
issues_tables = summary.get("tables_with_issues_count", 0)
skipped_count = summary.get("skipped_tables_count", 0)
lines.append(f"扫描表数: **{total_tables}** | "
f"涉及问题表: **{issues_tables}** | "
f"高空字段: **{high_count}** | "
f"中空字段: **{mid_count}** | "
f"跳过表数: **{skipped_count}**")
lines.append("")
top = summary.get("top_tables_by_high_empty_fields", [])[:10]
if top:
lines.append("### 高空字段最多的表 Top 10")
lines.append("| 表名 | 高空字段数 |")
lines.append("|------|------------|")
for t in top:
lines.append(f"| {t.get('table_name','')} | {t.get('high_empty_field_count', 0)} |")
thresholds = summary.get("thresholds") or {}
if thresholds:
lines.append(f"阈值: 高空 ≥ {thresholds.get('high', 0)*100:.0f}% | "
f"中空 ≥ {thresholds.get('mid', 0)*100:.0f}%")
lines.append("")
if not per_table:
lines.append("(未发现空字段问题)")
lines.append("")
return
# 按表汇总(高空数倒序)
lines.append("### 按表汇总(高空字段数倒序)")
lines.append("")
lines.append(_md_row(["表名", "表注释", "总行数", "字段数", "高空字段数", "中空字段数"]))
lines.append("|---|---|---|---|---|---|")
# 排序:高空数 desc,再按总字段数 desc
sorted_tables = sorted(
per_table,
key=lambda r: (-(r.get("high_count") or 0), -(r.get("total_fields") or 0)),
)
for r in sorted_tables:
lines.append(_md_row([
r.get("table_name", ""),
r.get("table_comment", "") or "-",
r.get("total_rows", 0),
r.get("total_fields", 0),
r.get("high_count", 0),
r.get("mid_count", 0),
]))
lines.append("")
def _section_missing(lines: list[str], data: dict):
summary = data.get("summary", {})
lines.append("## 五、缺失注释的字段")
lines.append(f"缺失注释字段: **{summary.get('total_missing_comments', 0)}**, "
f"涉及表数: **{summary.get('tables_affected', 0)}**, "
f"已推测: **{summary.get('predicted_count', 0)}**")
# 字段明细(按表分组)
lines.append("### 字段明细(按表分组)")
lines.append("")
samples = data.get("predicted_comments", [])[:20]
if samples:
lines.append("### 推测注释样本")
lines.append("| 表 | 字段 | 推测注释 | 置信度 | 依据 |")
lines.append("|----|------|----------|--------|------|")
for p in samples:
lines.append(f"| {p.get('table_name','')} | {p.get('column_name','')} | "
f"{p.get('predicted','')} | {p.get('confidence','')} | {p.get('reason','')} |")
for r in sorted_tables:
lines.append(f"#### {r.get('table_name', '')}")
if r.get("table_comment"):
lines.append(f"> 表注释: {r.get('table_comment', '')}")
else:
lines.append("> 表注释: (无)")
lines.append("")
high_list = r.get("high_empty_fields") or []
mid_list = r.get("mid_empty_fields") or []
if high_list:
lines.append(f"**🔴 高空字段({len(high_list)})**")
lines.append("")
lines.append(_md_row(["字段名", "类型", "注释", "空值数", "总行数", "空值率"]))
lines.append("|---|---|---|---|---|---|")
for f in sorted(high_list, key=lambda x: -(x.get("empty_rate") or 0)):
lines.append(_md_row([
f.get("column_name", ""),
f.get("data_type", ""),
f.get("column_comment", "") or "-",
f.get("empty_count", 0),
f.get("total_rows", 0),
f"{(f.get('empty_rate') or 0)*100:.2f}%",
]))
lines.append("")
if mid_list:
lines.append(f"**🟡 中空字段({len(mid_list)})**")
lines.append("")
lines.append(_md_row(["字段名", "类型", "注释", "空值数", "总行数", "空值率"]))
lines.append("|---|---|---|---|---|---|")
for f in sorted(mid_list, key=lambda x: -(x.get("empty_rate") or 0)):
lines.append(_md_row([
f.get("column_name", ""),
f.get("data_type", ""),
f.get("column_comment", "") or "-",
f.get("empty_count", 0),
f.get("total_rows", 0),
f"{(f.get('empty_rate') or 0)*100:.2f}%",
]))
lines.append("")
if not high_list and not mid_list:
lines.append("(该表无空字段记录)")
lines.append("")
# 跳过的表
if skipped:
lines.append("### 跳过的表")
lines.append("")
lines.append(_md_row(["表名", "原因", "记录数(如有)"]))
lines.append("|---|---|---|")
for s in skipped:
lines.append(_md_row([
s.get("table", ""),
s.get("reason", ""),
s.get("rows", "") if s.get("rows") is not None else "-",
]))
lines.append("")
def _section_length(lines: list[str], data: dict):
summary = data.get("summary", {})
lines.append("## 六、字段长度异常")
lines.append(f"异常字段数: **{summary.get('total_issues', 0)}**, "
f"应用规则数: **{summary.get('rules_applied', 0)}**")
def _md_missing(lines: list[str], data: dict):
summary = data.get("summary") or {}
by_table = data.get("by_table") or []
predicted = data.get("predicted_comments") or []
unpredicted = data.get("unpredictable_sample") or []
total = summary.get("total_missing_comments", 0)
tables_affected = summary.get("tables_affected", 0)
predicted_count = summary.get("predicted_count", 0)
predicted_by_llm = summary.get("predicted_by_llm", 0)
unpredicted_count = summary.get("unpredicted_count", 0)
llm_calls = summary.get("llm_calls", 0)
lines.append(f"缺失注释字段: **{total}** | "
f"涉及表: **{tables_affected}** | "
f"已推测: **{predicted_count}**(LLM: {predicted_by_llm}) | "
f"未推测: **{unpredicted_count}** | "
f"LLM 调用: **{llm_calls}**")
lines.append("")
issues = data.get("issues", [])[:20]
if issues:
lines.append("### 异常样本")
lines.append("| 表 | 字段 | 实际 | 标准 | 规则 | 建议 |")
lines.append("|----|------|------|------|------|------|")
for i in issues:
lines.append(f"| {i.get('table_name','')} | {i.get('column_name','')} | "
f"{i.get('actual_length','')} | "
f"{i.get('required_length','?')} | {i.get('rule','')} | "
f"{i.get('suggestion','')[:60]} |")
# 按表分布
if by_table:
lines.append("### 缺失注释字段按表分布")
lines.append("")
lines.append(_md_row(["表名", "缺失字段数"]))
lines.append("|---|---|")
for r in by_table:
lines.append(_md_row([r.get("table_name", ""), r.get("missing_count", 0)]))
lines.append("")
# 推测注释明细
if predicted:
lines.append(f"### 推测注释明细(共 {len(predicted)} 条)")
lines.append("")
lines.append(_md_row(["表名", "字段名", "类型", "推测注释", "置信度", "依据"]))
lines.append("|---|---|---|---|---|---|")
for p in predicted:
lines.append(_md_row([
p.get("table_name", ""),
p.get("column_name", ""),
p.get("data_type", ""),
p.get("predicted", ""),
p.get("confidence", ""),
p.get("reason", ""),
]))
lines.append("")
def _section_standards(lines: list[str], data: dict):
summary = data.get("summary", {})
lines.append("## 七、不符合国家标准的字段")
lines.append(f"应用标准数: **{summary.get('standards_applied', 0)}**, "
f"校验字段数: **{summary.get('fields_checked', 0)}**, "
f"违规记录数: **{summary.get('violations', 0)}**")
# 未推测样本
if unpredicted:
lines.append(f"### 未推测样本(共 {len(unpredicted)} 条)")
lines.append("")
lines.append(_md_row(["表名", "字段名", "类型", "依据"]))
lines.append("|---|---|---|---|")
for p in unpredicted:
lines.append(_md_row([
p.get("table_name", ""),
p.get("column_name", ""),
p.get("data_type", ""),
p.get("reason", ""),
]))
lines.append("")
if not predicted and not unpredicted and not by_table:
lines.append("(未发现缺失注释字段)")
lines.append("")
def _md_length(lines: list[str], data: dict):
summary = data.get("summary") or {}
issues = data.get("issues") or []
total = summary.get("total_issues", 0)
rules = summary.get("rules_applied", 0)
matched_name = summary.get("matched_by_name", 0)
matched_comment = summary.get("matched_by_comment", 0)
wasted_bytes = summary.get("wasted_bytes_total", 0)
lines.append(f"异常字段数: **{total}** | "
f"应用规则数: **{rules}** | "
f"按列名匹配: **{matched_name}** | "
f"按注释匹配: **{matched_comment}** | "
f"冗余字节: **{wasted_bytes:,}**")
lines.append("")
if not issues:
lines.append("(未发现字段长度异常)")
lines.append("")
return
lines.append(_md_row(["表名", "字段名", "类型", "实际", "标准", "单行冗余字节", "国标", "规则", "依据", "建议"]))
lines.append("|---|---|---|---|---|---|---|---|---|---|")
# 排序:按 wasted_bytes desc(影响最大排前)
for i in sorted(issues, key=lambda x: -(x.get("wasted_bytes_per_row") or 0)):
lines.append(_md_row([
i.get("table_name", ""),
i.get("column_name", ""),
i.get("data_type", ""),
i.get("actual_length", ""),
i.get("required_length", "?"),
i.get("wasted_bytes_per_row", 0),
i.get("standard", ""),
i.get("rule", ""),
i.get("basis", ""),
i.get("suggestion", ""),
]))
lines.append("")
def _md_standards(lines: list[str], data: dict):
summary = data.get("summary") or {}
by_std = data.get("violations_by_standard") or []
violations = data.get("violations") or []
applied = summary.get("standards_applied", 0)
used = summary.get("standards_used", 0)
fields_checked = summary.get("fields_checked", 0)
total_invalid = summary.get("violations", 0)
lines.append(f"应用标准: **{applied}** | "
f"命中标准: **{used}** | "
f"校验字段: **{fields_checked}** | "
f"违规记录: **{total_invalid}**")
lines.append("")
if not by_std:
lines.append("(未发现国标违规)")
lines.append("")
return
# 按标准统计
lines.append("### 按标准统计")
lines.append("")
lines.append(_md_row(["标准编号", "标准名", "字段数", "违规字段数", "采样总数", "违规数", "违规率"]))
lines.append("|---|---|---|---|---|---|---|")
for v in by_std:
rate = v.get("violation_rate") or 0
lines.append(_md_row([
v.get("standard", ""),
v.get("standard_name", ""),
v.get("fields_count", 0),
v.get("violating_fields_count", 0),
v.get("total_sampled", 0),
v.get("total_invalid", 0),
f"{rate*100:.2f}%",
]))
lines.append("")
# 各标准下的违规字段明细
lines.append("### 各标准下的违规字段明细")
lines.append("")
by_std = data.get("violations_by_standard", [])
if by_std:
lines.append("### 按标准统计")
lines.append("| 标准 | 标准名 | 字段数 | 违规数 | 违规率 |")
lines.append("|------|--------|--------|--------|--------|")
for v in by_std:
lines.append(f"| {v.get('standard','')} | {v.get('standard_name','')[:30]} | "
f"{v.get('fields_count',0)} | {v.get('total_invalid',0)} | "
f"{v.get('violation_rate', 0):.2%} |")
for v in by_std:
details = v.get("details_with_violations") or []
if not details:
continue
lines.append(f"#### {v.get('standard', '')} — {v.get('standard_name', '')}")
lines.append("")
lines.append(_md_row(["表", "字段", "采样数", "违规数", "违规率"]))
lines.append("|---|---|---|---|---|")
for d in details:
dr = d.get("violation_rate") or 0
lines.append(_md_row([
d.get("table", ""),
d.get("field", ""),
d.get("sampled", 0),
d.get("violations", 0),
f"{dr*100:.2f}%",
]))
# 样本展示(前 5 条)
any_samples = False
for d in details:
samples = d.get("samples") or []
if samples:
if not any_samples:
lines.append("")
lines.append("**违规样本**")
lines.append("")
any_samples = True
lines.append(f"- `{d.get('table','')}.{d.get('field','')}`: " + "; ".join(
f"`{s.get('value','')}` ({s.get('reason','')})" for s in samples[:5]
))
lines.append("")
# 违规样本(平铺)
if violations:
lines.append(f"### 违规样本(前 {len(violations)} 条)")
lines.append("")
lines.append(_md_row(["表名", "字段名", "规则类型", "标准名", "字段类型", "样本值", "错误原因"]))
lines.append("|---|---|---|---|---|---|---|")
for v in violations[:500]:
lines.append(_md_row([
v.get("table_name", ""),
v.get("column_name", ""),
v.get("rule_type", ""),
v.get("standard_name", ""),
v.get("column_type", ""),
v.get("value", ""),
v.get("error", ""),
]))
lines.append("")
_MD_SECTION_HANDLERS = {
"merge_candidates": _md_merge,
"redundancy_fields": _md_redundancy,
"empty_fields": _md_empty,
"missing_comments": _md_missing,
"length_issues": _md_length,
"standard_violations": _md_standards,
}
# ════════════════════════════════════════════════════════════════
# docx 渲染
# ════════════════════════════════════════════════════════════════
def _render_docx(doc, sections: dict, overview: dict):
"""简易 docx 渲染(标题 + 段落 + 表格)"""
from docx.shared import Pt, Cm, RGBColor
"""Word 报告渲染:每个章节用原生表格铺完整明细"""
from docx.shared import Pt, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
# 标题
h = doc.add_heading(f"数据治理报告 — {overview.get('database', '')}", level=0)
h.alignment = WD_ALIGN_PARAGRAPH.CENTER
# ── 标题区 ──
title = doc.add_heading(f"数据治理报告 — {overview.get('database', '')}", level=0)
title.alignment = WD_ALIGN_PARAGRAPH.CENTER
doc.add_paragraph(f"生成时间: {overview.get('executed_at', '')}")
doc.add_paragraph(f"数据库类型: {overview.get('db_type', '-')}")
info = doc.add_paragraph()
info.add_run(f"生成时间: {overview.get('executed_at', '')}\n").bold = False
info.add_run(f"数据库类型: {overview.get('db_type', '-')}\n")
info.add_run(f"主机: {overview.get('host', '-')}\n")
# 概览表格
# ── 一、概览 ──
doc.add_heading("一、概览", level=1)
overview_table = doc.add_table(rows=1, cols=2)
overview_table.style = "Light Grid Accent 1"
overview_table.rows[0].cells[0].text = "项目"
overview_table.rows[0].cells[1].text = "值"
for k, v in [
("数据库", overview.get("database", "-")),
("数据库类型", overview.get("db_type", "-")),
("主机", overview.get("host", "-")),
overview_rows = [
("数据库名", str(overview.get("database", "-"))),
("数据库类型", str(overview.get("db_type", "-"))),
("主机", str(overview.get("host", "-"))),
("表总数", str(overview.get("total_tables", 0))),
("字段总数", str(overview.get("total_fields", 0))),
("高空字段(≥80%)", str(overview.get("high_empty_fields", 0))),
("中空字段(≥50%)", str(overview.get("mid_empty_fields", 0))),
("大范围空字段(≥80%)", str(overview.get("high_empty_fields", 0))),
("中范围空字段(≥50%)", str(overview.get("mid_empty_fields", 0))),
("缺失注释字段", str(overview.get("missing_comments", 0))),
("字段长度异常", str(overview.get("length_issues", 0))),
("国标违规记录", str(overview.get("standard_violations", 0))),
("表合并候选", str(overview.get("merge_candidates", 0))),
("冗余字段", str(overview.get("redundancy_fields", 0))),
]:
row = overview_table.add_row().cells
row[0].text = k
row[1].text = str(v)
# 简单添加各章节标题(详细表格由 Web 前端展示,Word 只给概览)
for sec_title, sec_key in [
("二、表合并候选", "merge_candidates"),
("三、冗余字段", "redundancy_fields"),
("四、大范围空字段", "empty_fields"),
("五、缺失注释的字段", "missing_comments"),
("六、字段长度异常", "length_issues"),
("七、不符合国家标准的字段", "standard_violations"),
]:
doc.add_heading(sec_title, level=1)
data = sections.get(sec_key, {})
summary = data.get("summary", {})
if summary:
doc.add_paragraph(f"本章节发现: {summary}")
doc.add_paragraph("(完整明细请查看 Web 页面或 Markdown 报告)")
\ No newline at end of file
]
_add_table(doc, ["项目", "值"], overview_rows)
# ── 各章节 ──
for idx, (key, title) in enumerate(SECTION_ORDER, start=2):
doc.add_heading(f"{idx}、{title}", level=1)
handler = _DOCX_SECTION_HANDLERS.get(key)
if handler:
handler(doc, sections.get(key, {}))
else:
doc.add_paragraph("(无渲染函数)")
def _add_table(doc, headers: list[str], rows: list, style: str = "Light Grid Accent 1"):
"""通用表格:headers + rows(每行是可迭代的值列表)。空行就只输出表头。"""
if not rows:
# 没有数据也加表头,提示用户
table = doc.add_table(rows=1, cols=len(headers))
table.style = style
for i, h in enumerate(headers):
table.rows[0].cells[i].text = h
doc.add_paragraph("(无数据)")
return table
table = doc.add_table(rows=1, cols=len(headers))
table.style = style
hdr = table.rows[0].cells
for i, h in enumerate(headers):
hdr[i].text = str(h)
for row in rows:
cells = table.add_row().cells
for i, v in enumerate(row):
cells[i].text = _docx_cell(v)
return table
def _docx_cell(v: Any) -> str:
"""单元格文本:None/空 → '-';其他 str()"""
if v is None or v == "":
return "-"
s = str(v)
return s
# ── 各章节 docx 渲染 ────────────────────────────────────────────
def _docx_merge(doc, data: dict):
candidates = data.get("merge_candidates") or []
decommission = data.get("decommission_candidates") or {}
strong = (decommission.get("strong") or {}).get("tables", []) if isinstance(decommission, dict) else []
weak = (decommission.get("weak") or {}).get("tables", []) if isinstance(decommission, dict) else []
doc.add_paragraph(
f"候选合并组: {len(candidates)} | 强废弃候选: {len(strong)} | 弱废弃候选: {len(weak)}"
)
if candidates:
doc.add_heading("合并候选明细", level=2)
rows = []
for c in candidates:
tables = ", ".join(c.get("tables", []))
diff_cols = ", ".join(c.get("diff_columns", []) or []) or "-"
rows.append([
c.get("id", ""),
c.get("priority", ""),
c.get("similarity", ""),
len(c.get("tables", [])),
c.get("common_columns", ""),
tables,
diff_cols,
(c.get("suggestion", "") or "")[:200],
c.get("risk", ""),
])
_add_table(doc, ["ID", "优先级", "相似度", "表数", "公共列数", "表", "差异列", "建议", "风险"], rows)
# 公共列明细
doc.add_heading("各候选组的公共列", level=2)
for c in candidates:
cols = c.get("columns", []) or []
doc.add_paragraph(
f"{c.get('id', '')} ({', '.join(c.get('tables', []))}): {', '.join(cols) if cols else '(无)'}"
)
if strong:
doc.add_heading("强废弃候选", level=2)
_add_table(doc, ["旧表", "建议替代", "证据"], [
[d.get("table", ""), d.get("replacement", ""), d.get("evidence", "")]
for d in strong
])
if weak:
doc.add_heading("弱废弃候选", level=2)
_add_table(doc, ["表", "相似表", "相似度"], [
[d.get("table", ""), d.get("similar_to", ""), d.get("similarity", "")]
for d in weak
])
if not candidates and not strong and not weak:
doc.add_paragraph("(未发现可合并的表)")
def _docx_redundancy(doc, data: dict):
fields = data.get("redundancy_fields") or []
doc.add_paragraph(f"高频字段数(出现在 ≥10 张表): {len(fields)}")
if not fields:
doc.add_paragraph("(未发现明显冗余字段)")
return
rows = []
for f in fields:
source = f.get("source", "")
reasoning = f.get("reasoning", "") or ""
if source == "llm_failed":
reasoning = "⚠️ LLM 推理失败(待人工核对)"
rows.append([
f.get("field", ""),
f.get("table_count", ""),
f.get("classification", ""),
reasoning,
f.get("recommendation", ""),
])
_add_table(doc, ["字段", "出现表数", "分类", "LLM 推理", "建议处置"], rows)
def _docx_empty(doc, data: dict):
summary = data.get("summary") or {}
per_table = data.get("per_table") or []
skipped = data.get("skipped") or []
doc.add_paragraph(
f"扫描表数: {summary.get('total_tables_scanned', 0)} | "
f"涉及问题表: {summary.get('tables_with_issues_count', 0)} | "
f"高空字段: {summary.get('high_empty_fields_count', 0)} | "
f"中空字段: {summary.get('mid_empty_fields_count', 0)} | "
f"跳过表数: {summary.get('skipped_tables_count', 0)}"
)
thresholds = summary.get("thresholds") or {}
if thresholds:
doc.add_paragraph(
f"阈值: 高空 ≥ {thresholds.get('high', 0)*100:.0f}% | 中空 ≥ {thresholds.get('mid', 0)*100:.0f}%"
)
if not per_table:
doc.add_paragraph("(未发现空字段问题)")
return
# 按表汇总
doc.add_heading("按表汇总(高空字段数倒序)", level=2)
sorted_tables = sorted(
per_table,
key=lambda r: (-(r.get("high_count") or 0), -(r.get("total_fields") or 0)),
)
_add_table(doc, ["表名", "表注释", "总行数", "字段数", "高空字段数", "中空字段数"], [
[
r.get("table_name", ""),
r.get("table_comment", "") or "-",
r.get("total_rows", 0),
r.get("total_fields", 0),
r.get("high_count", 0),
r.get("mid_count", 0),
]
for r in sorted_tables
])
# 字段明细
doc.add_heading("字段明细(按表分组)", level=2)
for r in sorted_tables:
doc.add_heading(r.get("table_name", ""), level=3)
if r.get("table_comment"):
doc.add_paragraph(f"表注释: {r.get('table_comment', '')}")
else:
doc.add_paragraph("表注释: (无)")
high_list = r.get("high_empty_fields") or []
mid_list = r.get("mid_empty_fields") or []
if high_list:
doc.add_paragraph(f"🔴 高空字段({len(high_list)})")
rows = []
for f in sorted(high_list, key=lambda x: -(x.get("empty_rate") or 0)):
rows.append([
f.get("column_name", ""),
f.get("data_type", ""),
f.get("column_comment", "") or "-",
f.get("empty_count", 0),
f.get("total_rows", 0),
f"{(f.get('empty_rate') or 0)*100:.2f}%",
])
_add_table(doc, ["字段名", "类型", "注释", "空值数", "总行数", "空值率"], rows)
if mid_list:
doc.add_paragraph(f"🟡 中空字段({len(mid_list)})")
rows = []
for f in sorted(mid_list, key=lambda x: -(x.get("empty_rate") or 0)):
rows.append([
f.get("column_name", ""),
f.get("data_type", ""),
f.get("column_comment", "") or "-",
f.get("empty_count", 0),
f.get("total_rows", 0),
f"{(f.get('empty_rate') or 0)*100:.2f}%",
])
_add_table(doc, ["字段名", "类型", "注释", "空值数", "总行数", "空值率"], rows)
if not high_list and not mid_list:
doc.add_paragraph("(该表无空字段记录)")
if skipped:
doc.add_heading("跳过的表", level=2)
_add_table(doc, ["表名", "原因", "记录数(如有)"], [
[
s.get("table", ""),
s.get("reason", ""),
s.get("rows", "") if s.get("rows") is not None else "-",
]
for s in skipped
])
def _docx_missing(doc, data: dict):
summary = data.get("summary") or {}
by_table = data.get("by_table") or []
predicted = data.get("predicted_comments") or []
unpredicted = data.get("unpredictable_sample") or []
doc.add_paragraph(
f"缺失注释字段: {summary.get('total_missing_comments', 0)} | "
f"涉及表: {summary.get('tables_affected', 0)} | "
f"已推测: {summary.get('predicted_count', 0)} "
f"(LLM: {summary.get('predicted_by_llm', 0)}) | "
f"未推测: {summary.get('unpredicted_count', 0)} | "
f"LLM 调用: {summary.get('llm_calls', 0)}"
)
if by_table:
doc.add_heading("缺失注释字段按表分布", level=2)
_add_table(doc, ["表名", "缺失字段数"], [
[r.get("table_name", ""), r.get("missing_count", 0)]
for r in by_table
])
if predicted:
doc.add_heading(f"推测注释明细(共 {len(predicted)} 条)", level=2)
_add_table(doc, ["表名", "字段名", "类型", "推测注释", "置信度", "依据"], [
[
p.get("table_name", ""),
p.get("column_name", ""),
p.get("data_type", ""),
p.get("predicted", ""),
p.get("confidence", ""),
p.get("reason", ""),
]
for p in predicted
])
if unpredicted:
doc.add_heading(f"未推测样本(共 {len(unpredicted)} 条)", level=2)
_add_table(doc, ["表名", "字段名", "类型", "依据"], [
[
p.get("table_name", ""),
p.get("column_name", ""),
p.get("data_type", ""),
p.get("reason", ""),
]
for p in unpredicted
])
if not predicted and not unpredicted and not by_table:
doc.add_paragraph("(未发现缺失注释字段)")
def _docx_length(doc, data: dict):
summary = data.get("summary") or {}
issues = data.get("issues") or []
doc.add_paragraph(
f"异常字段数: {summary.get('total_issues', 0)} | "
f"应用规则数: {summary.get('rules_applied', 0)} | "
f"按列名匹配: {summary.get('matched_by_name', 0)} | "
f"按注释匹配: {summary.get('matched_by_comment', 0)} | "
f"冗余字节: {summary.get('wasted_bytes_total', 0):,}"
)
if not issues:
doc.add_paragraph("(未发现字段长度异常)")
return
rows = []
for i in sorted(issues, key=lambda x: -(x.get("wasted_bytes_per_row") or 0)):
rows.append([
i.get("table_name", ""),
i.get("column_name", ""),
i.get("data_type", ""),
i.get("actual_length", ""),
i.get("required_length", "?"),
i.get("wasted_bytes_per_row", 0),
i.get("standard", ""),
i.get("rule", ""),
i.get("basis", ""),
i.get("suggestion", ""),
])
_add_table(doc, ["表名", "字段名", "类型", "实际", "标准", "单行冗余字节", "国标", "规则", "依据", "建议"], rows)
def _docx_standards(doc, data: dict):
summary = data.get("summary") or {}
by_std = data.get("violations_by_standard") or []
violations = data.get("violations") or []
doc.add_paragraph(
f"应用标准: {summary.get('standards_applied', 0)} | "
f"命中标准: {summary.get('standards_used', 0)} | "
f"校验字段: {summary.get('fields_checked', 0)} | "
f"违规记录: {summary.get('violations', 0)}"
)
if not by_std:
doc.add_paragraph("(未发现国标违规)")
return
doc.add_heading("按标准统计", level=2)
_add_table(doc, ["标准编号", "标准名", "字段数", "违规字段数", "采样总数", "违规数", "违规率"], [
[
v.get("standard", ""),
v.get("standard_name", ""),
v.get("fields_count", 0),
v.get("violating_fields_count", 0),
v.get("total_sampled", 0),
v.get("total_invalid", 0),
f"{(v.get('violation_rate') or 0)*100:.2f}%",
]
for v in by_std
])
doc.add_heading("各标准下的违规字段明细", level=2)
for v in by_std:
details = v.get("details_with_violations") or []
if not details:
continue
doc.add_heading(f"{v.get('standard', '')} — {v.get('standard_name', '')}", level=3)
_add_table(doc, ["表", "字段", "采样数", "违规数", "违规率"], [
[
d.get("table", ""),
d.get("field", ""),
d.get("sampled", 0),
d.get("violations", 0),
f"{(d.get('violation_rate') or 0)*100:.2f}%",
]
for d in details
])
# 样本
for d in details:
samples = d.get("samples") or []
if samples:
doc.add_paragraph(
f"样本({d.get('table','')}.{d.get('field','')}): "
+ "; ".join(f"`{s.get('value','')}` ({s.get('reason','')})" for s in samples[:5])
)
if violations:
doc.add_heading(f"违规样本(前 {len(violations)} 条)", level=2)
_add_table(doc, ["表名", "字段名", "规则类型", "标准名", "字段类型", "样本值", "错误原因"], [
[
v.get("table_name", ""),
v.get("column_name", ""),
v.get("rule_type", ""),
v.get("standard_name", ""),
v.get("column_type", ""),
v.get("value", ""),
v.get("error", ""),
]
for v in violations[:500]
])
_DOCX_SECTION_HANDLERS = {
"merge_candidates": _docx_merge,
"redundancy_fields": _docx_redundancy,
"empty_fields": _docx_empty,
"missing_comments": _docx_missing,
"length_issues": _docx_length,
"standard_violations": _docx_standards,
}
\ No newline at end of file
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