commit c19e9fcebf48893d1a7b29368efb2ba31678d186 Author: Bepr4 <63661977@qq.com> Date: Thu Aug 20 17:04:10 2026 +0800 govdoc-md-cleaner v0.1.0: 政务文档 Markdown 清洗工具（规则引擎 + CLI + 测试） - 8 条 YAML 声明规则：页码/页眉页脚/目录点线/图片/HTML表格/散落标签/行尾空白/空行 - 防误伤设计：protect 正则 + 内容形态豁免 + OCR burst 检测 - md-clean single/batch CLI，JSON 清洗报告 - 18 个单元测试 Co-Authored-By: Claude diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0f71e77 --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +# 数据与产物 + +*.cleaned.md +report.json +__pycache__/ +*.pyc +.venv/ +dist/ +*.egg-info/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..fd29ead --- /dev/null +++ b/README.md @@ -0,0 +1,93 @@ +# govdoc-md-cleaner + +政务文档（招标 / 投标 / 采购 / 合同）**PDF→Markdown 产物**的规则化清洗工具。 + +上游解析管线（MinerU / OCR）把 PDF 转成 Markdown 后，会带入大量非正文噪音： +逐页重复的页眉页脚、页码行、目录点线、死链图片引用、压成单行的 HTML 表格、 +行尾硬换行双空格，以及 OCR 重复崩坏段。本工具用一套 **YAML 声明、按序应用、 +逐条统计** 的规则把它们清掉，输出可直接进入比对 / RAG / 审核管线的干净 Markdown。 + +## 数据来源 + +清洗目标为 `/home/lihaoze/gov_test_data`（律所上传的真实政务文件的筛选子集， +详见其 `compare/readme.md`），未来会持续接入更多批次的 Markdown 文件。 +**本项目只包含清洗代码与规则，不包含任何业务数据**——测试数据不入库。 + +## 清洗规则（rules/default.yaml） + +| 顺序 | 规则 | 处理对象 | 示例 | +|---|---|---|---| +| 10 | `strip_page_lines` | 页码行 | `第 3 页 共 53 页`、`第4页共4页`、`Page 3 of 10`、`- 4 -` | +| 20 | `strip_repeated_short_lines` | 页眉页脚 + OCR 崩坏 | 全篇重复 ≥3 次的短行；连续刷屏 ≥5 次（实测"审计程序"×1359） | +| 30 | `strip_toc_dots` | 目录点线 | `第一章 投标邀请函 ………… 2` → 保留标题，删点线页码 | +| 40 | `drop_images` | 死链图片 | `![](images/xxx.jpg)`、base64 内嵌图 | +| 50 | `normalize_tables` | HTML 表格 | `
…` 单行压缩 → Markdown 管道表格 | +| 60 | `strip_stray_html` | 散落标签 | `
` | +| 70 | `rstrip_lines` | 行尾空白 | MinerU 每行行尾的双空格硬换行 | +| 80 | `collapse_blank_lines` | 空行折叠 | 连续空行 → 1 行；裁掉文首文末 | + +### 防误伤设计（来自真实数据踩坑） + +- **protect 正则**：`投标人：（公章）`、`法定代表人签名:`、`日期: 年 月 日`、 + `致：xxx`、声明函结尾句——标书里逐章重复，**是正文模板不是页眉**，默认保护。 +- **内容形态豁免**：编号条款 `（1）…`、`1、…`、`一、…`、列表/表格行、 + `乙方：xxx` 字段行——平行结构天然重复，引擎侧直接不参与页眉判定。 +- **burst 检测**：同一短行连续刷屏 ≥5 次（间隔 ≤2 行）判为 OCR 重复崩坏， + 直接删除（003-10 案例出现"审计程序"连续 1359 行）。 + +## 安装 + +```bash +pip install -e . # 或直接 python -m cleaner.cli（仅需 PyYAML） +``` + +## 使用 + +```bash +# 单文件：清洗 + 打印统计 +md-clean single input.md -o output.md --diff + +# 批量：递归清洗目录下所有 .md，输出到平行目录 + JSON 报告 +md-clean batch /path/to/uploads -o /path/to/uploads_cleaned \ + --pattern "*.md" --report report.json + +# 用自定义规则集（复制 default.yaml 改参数即可） +md-clean batch uploads/ --rules rules/strict.yaml +``` + +`--diff` 输出 unified diff，`report.json` 记录每个文件每条规则的命中数， +清洗过程完全可审计、可回滚（重跑即得原结果的对照）。 + +## 项目结构 + +``` +govdoc-md-cleaner/ +├── cleaner/ +│ ├── cleaner.py # 引擎：按序应用规则，输出 CleanResult(text, stats) +│ ├── rules.py # 规则实现 + REGISTRY + YAML 加载 +│ └── cli.py # single / batch 两个子命令 +├── rules/ +│ └── default.yaml # 默认规则集（顺序、开关、参数、protect 列表） +├── tests/ +│ └── test_rules.py # 每条规则的最小样例 + 防误伤回归 +└── docs/ + └── RULES.md # 规则编写指南（新增规则的方法） +``` + +## 新增一条规则 + +1. `cleaner/rules.py`：实现 `_your_rule(text, params, stats) -> text`，注册进 `REGISTRY`； +2. `rules/default.yaml`：追加 `- name: your_rule / order: / params:`； +3. `tests/test_rules.py`：加一个真实数据浓缩出的最小样例。 + +原则：**规则只删噪音不删正文**；拿不准的形态默认保留，靠 protect/exempt 收紧。 + +## 测试 + +```bash +python -m unittest discover tests -v +``` + +## License + +MIT diff --git a/cleaner/__init__.py b/cleaner/__init__.py new file mode 100644 index 0000000..2785bf4 --- /dev/null +++ b/cleaner/__init__.py @@ -0,0 +1,11 @@ +"""govdoc-md-cleaner: 政务文档 Markdown 清洗工具包。 + +针对 PDF→Markdown 转换产物（MinerU / OCR 管线输出）的常见脏数据， +提供基于 YAML 规则的、可复现、可审计的清洗能力。 +""" + +from cleaner.cleaner import MarkdownCleaner, CleanResult, load_rules +from cleaner.cli import main + +__version__ = "0.1.0" +__all__ = ["MarkdownCleaner", "CleanResult", "load_rules", "main"] diff --git a/cleaner/cleaner.py b/cleaner/cleaner.py new file mode 100644 index 0000000..7764ed9 --- /dev/null +++ b/cleaner/cleaner.py @@ -0,0 +1,56 @@ +"""清洗引擎：按规则顺序应用，输出清洗后文本 + 统计报告。""" + +from __future__ import annotations + +import difflib +from dataclasses import dataclass, field +from pathlib import Path +from typing import Dict, List, Optional + +from cleaner.rules import REGISTRY, Rule, load_rules + + +@dataclass +class CleanResult: + """一次清洗的结果：产物 + 可审计的统计。""" + + text: str + stats: Dict[str, int] = field(default_factory=dict) + rules_applied: List[str] = field(default_factory=list) + + @property + def total_hits(self) -> int: + return sum(v for k, v in self.stats.items() if k != "collapsed_blanks") + + +class MarkdownCleaner: + def __init__(self, rules: Optional[List[Rule]] = None, rules_path: Optional[Path] = None): + self.rules = rules if rules is not None else load_rules(rules_path) + + def clean_text(self, text: str) -> CleanResult: + stats: Dict[str, int] = {} + applied: List[str] = [] + for rule in self.rules: + if not rule.enabled: + continue + func = REGISTRY[rule.name] + text = func(text, rule.params, stats) + applied.append(rule.name) + return CleanResult(text=text, stats=stats, rules_applied=applied) + + def clean_file(self, src: Path, dst: Optional[Path] = None) -> CleanResult: + raw = Path(src).read_text(encoding="utf-8") + result = self.clean_text(raw) + if dst is not None: + Path(dst).parent.mkdir(parents=True, exist_ok=True) + Path(dst).write_text(result.text, encoding="utf-8") + return result + + @staticmethod + def diff(before: str, after: str, context: int = 1) -> str: + return "\n".join( + difflib.unified_diff( + before.splitlines(), after.splitlines(), + fromfile="before", tofile="after", lineterm="", n=context, + ) + ) diff --git a/cleaner/cli.py b/cleaner/cli.py new file mode 100644 index 0000000..d656411 --- /dev/null +++ b/cleaner/cli.py @@ -0,0 +1,89 @@ +"""命令行入口。 + +用法: + md-clean single [-o output.md] [--diff] [--rules rules.yaml] + md-clean batch [-o outdir] [--pattern "*.md"] [--report report.json] +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import List, Optional + +from cleaner.cleaner import MarkdownCleaner +from cleaner.rules import load_rules + + +def _build(path: Optional[Path]) -> MarkdownCleaner: + return MarkdownCleaner(rules=load_rules(path)) + + +def cmd_single(args: argparse.Namespace) -> int: + cleaner = _build(args.rules) + src = Path(args.input) + raw = src.read_text(encoding="utf-8") + result = cleaner.clean_text(raw) + if args.output: + Path(args.output).parent.mkdir(parents=True, exist_ok=True) + Path(args.output).write_text(result.text, encoding="utf-8") + print(f"已写入 {args.output}") + if args.diff: + print(MarkdownCleaner.diff(raw, result.text)) + print(json.dumps(result.stats, ensure_ascii=False, indent=2)) + return 0 + + +def cmd_batch(args: argparse.Namespace) -> int: + cleaner = _build(args.rules) + src_dir = Path(args.directory) + files = sorted(p for p in src_dir.rglob(args.pattern) if p.is_file()) + if not files: + print(f"在 {src_dir} 下未找到匹配 {args.pattern} 的文件", file=sys.stderr) + return 1 + out_dir = Path(args.output) if args.output else src_dir.parent / (src_dir.name + "_cleaned") + report = [] + for f in files: + rel = f.relative_to(src_dir) + dst = out_dir / rel + result = cleaner.clean_file(f, dst) + report.append({"file": str(rel), "stats": result.stats}) + hits = result.total_hits + print(f"[ok] {rel} 命中 {hits} 处") + if args.report: + Path(args.report).parent.mkdir(parents=True, exist_ok=True) + Path(args.report).write_text( + json.dumps({"files": report}, ensure_ascii=False, indent=2), encoding="utf-8" + ) + print(f"报告已写入 {args.report}") + print(f"共清洗 {len(files)} 个文件 → {out_dir}") + return 0 + + +def main(argv: Optional[List[str]] = None) -> int: + parser = argparse.ArgumentParser(prog="md-clean", description="政务文档 Markdown 清洗工具") + sub = parser.add_subparsers(dest="command", required=True) + + p1 = sub.add_parser("single", help="清洗单个文件") + p1.add_argument("input", help="输入 .md 文件") + p1.add_argument("-o", "--output", help="输出路径（缺省打印统计不写文件）") + p1.add_argument("--diff", action="store_true", help="打印 unified diff") + p1.add_argument("--rules", type=Path, help="规则 YAML 路径") + + p2 = sub.add_parser("batch", help="批量清洗目录（递归）") + p2.add_argument("directory", help="输入目录") + p2.add_argument("-o", "--output", help="输出目录（缺省 _cleaned）") + p2.add_argument("--pattern", default="*.md", help="文件 glob（默认 *.md）") + p2.add_argument("--report", help="清洗报告 JSON 输出路径") + p2.add_argument("--rules", type=Path, help="规则 YAML 路径") + + args = parser.parse_args(argv) + if args.command == "single": + return cmd_single(args) + return cmd_batch(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/cleaner/rules.py b/cleaner/rules.py new file mode 100644 index 0000000..5e0a25a --- /dev/null +++ b/cleaner/rules.py @@ -0,0 +1,281 @@ +"""规则模型：一条清洗规则 = 名称 + 开关 + 参数 + 应用顺序。 + +规则用 YAML 声明（rules/*.yaml），引擎按 order 依次应用， +这样清洗过程可复现、可 diff、可回滚。 +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional + +import yaml + + +@dataclass +class Rule: + """一条清洗规则。 + + name: 唯一标识（报告里引用） + order: 应用顺序，小的先执行 + enabled: 开关，方便对某个用例单独关掉 + params: 传给处理函数的额外参数 + """ + + name: str + order: int = 100 + enabled: bool = True + params: Dict[str, Any] = field(default_factory=dict) + + @staticmethod + def from_dict(d: Dict[str, Any]) -> "Rule": + return Rule( + name=d["name"], + order=int(d.get("order", 100)), + enabled=bool(d.get("enabled", True)), + params=dict(d.get("params") or {}), + ) + + +# --------------------------------------------------------------------------- +# 每条规则的具体实现。函数签名统一为 (text, params, stats) -> text。 +# stats 是 {rule_name: 删改行数}，用于生成清洗报告。 +# --------------------------------------------------------------------------- + +RuleFunc = Callable[[str, Dict[str, Any], Dict[str, int]], str] + +# 页码类：第X页 共Y页 / 第X页共Y页 / Page x of y / - 3 - 等 +_RE_PAGE_CN = re.compile( + r"^[ \t]*第\s*[0-9０-９]+\s*页\s*(?:[,，/]?\s*共\s*[0-9０-９]+\s*页)?[ \t]*$" +) +_RE_PAGE_EN = re.compile( + r"^[ \t]*(?:[-–—]?\s*Page\s+\d+(?:\s+of\s+\d+)?\s*[-–—]?" + r"|[-–—]\s*\d{1,4}\s*[-–—]" + r"|\d+\s*/\s*\d+)[ \t]*$", + re.IGNORECASE, +) +# 纯页码数字行（单独一行只有 1-4 位数字，且不是标题编号场景） +_RE_PAGE_BARE = re.compile(r"^[ \t]*\d{1,4}[ \t]*$") + + +def _strip_page_lines(text: str, params: Dict[str, Any], stats: Dict[str, int]) -> str: + keep_bare_numbers = bool(params.get("keep_bare_numbers", True)) + out, n = [], 0 + for line in text.splitlines(): + if _RE_PAGE_CN.match(line) or _RE_PAGE_EN.match(line): + n += 1 + continue + if not keep_bare_numbers and _RE_PAGE_BARE.match(line): + n += 1 + continue + out.append(line) + stats["page_lines"] = stats.get("page_lines", 0) + n + return "\n".join(out) + + +# 页眉/页脚：同一短行在全篇重复出现 >= N 次（默认 3），视为页眉页脚删除。 +# 两道保险避免误伤正文： +# 1. protect 正则 —— 标书里逐章重复的模板行（签章/日期/声明结尾） +# 2. 内容形态行直接豁免 —— 编号条款 "(1)…"/"1、…"/"一、…"/列表/表格行 +# 在平行结构的标书里天然重复，但它们是正文不是页眉。 +_RE_CONTENT_LIKE = re.compile( + r"^[\d０-９(（\[【\-–—*•·①-⑳一二三四五六七八九十百第章节条款、,.。:：|]" + r"|[一-鿿]{1,6}[：:]\s*\S" # "乙方：xxx" / "地址：xxx" 这类字段行 + r"|^[一-鿿A-Za-z]{1,6}[：:]\s*$" # "乙方：" / "注：" 字段标签行 +) + + +def _strip_repeated_short_lines( + text: str, params: Dict[str, Any], stats: Dict[str, int] +) -> str: + threshold = int(params.get("threshold", 3)) + max_len = int(params.get("max_len", 40)) + # 同一短行在文中连续出现 >= burst_limit 次（间隔 <= burst_gap 行）视为 + # OCR 重复崩坏（如 003-10 案例"审计程序"连续刷屏 1359 次），无论阈值直接删。 + burst_limit = int(params.get("burst_limit", 5)) + burst_gap = int(params.get("burst_gap", 2)) + protect = [re.compile(p) for p in params.get("protect", [])] + lines = text.splitlines() + counts: Dict[str, int] = {} + positions: Dict[str, List[int]] = {} + for i, line in enumerate(lines): + s = line.strip() + if ( + 0 < len(s) <= max_len + and not s.startswith("#") + and not _RE_CONTENT_LIKE.match(s) + and "![" not in s + ): + counts[s] = counts.get(s, 0) + 1 + positions.setdefault(s, []).append(i) + repeated = { + s + for s, c in counts.items() + if c >= threshold and not any(rx.search(s) for rx in protect) + } + # OCR 崩坏连续段：即使该行被 protect/内容豁免，连续刷屏也删 + for s, pos in positions.items(): + best_run = run = 1 + for a, b in zip(pos, pos[1:]): + run = run + 1 if b - a <= burst_gap else 1 + best_run = max(best_run, run) + if best_run >= burst_limit: + repeated.add(s) + if not repeated: + return text + out, n = [], 0 + for line in lines: + if line.strip() in repeated: + n += 1 + continue + out.append(line) + stats["header_footer_lines"] = stats.get("header_footer_lines", 0) + n + return "\n".join(out) + + +# 目录点线：标题文字 ………… 12 / ······ 3 之类（… U+2026 也算；# 前缀可选） +_RE_TOC_DOTS = re.compile( + r"^(#{1,6}\s+)?.*?[ \t]*[\.。·•‧…]{6,}[ \t]*[\d０-９]*[ \t]*$" +) + + +def _strip_toc_dots(text: str, params: Dict[str, Any], stats: Dict[str, int]) -> str: + out, n = [], 0 + for line in text.splitlines(): + m = _RE_TOC_DOTS.match(line) + if m: + # 去掉点线和页码，保留标题文字；纯点线+页码的目录行整行删 + title = re.sub(r"[ \t]*[\.。·•‧…]{6,}[ \t]*[\d０-９]*[ \t]*$", "", line).rstrip() + if title.strip() and not re.fullmatch(r"[\.。·•‧…\d０-９\s]+", title): + out.append(title) + n += 1 + continue + out.append(line) + stats["toc_dot_lines"] = stats.get("toc_dot_lines", 0) + n + return "\n".join(out) + + +# 图片引用：![](images/xxx.jpg) —— 图片目录不在交付物里，引用是死链 +_RE_IMAGE = re.compile(r"[ \t]*!\[[^\]]*\]\([^)]*\)[ \t]*") + + +def _drop_images(text: str, params: Dict[str, Any], stats: Dict[str, int]) -> str: + placeholder = params.get("placeholder") # None=整行删除；否则替换为占位文本 + out, n = [], 0 + for line in text.splitlines(): + if _RE_IMAGE.fullmatch(line): + n += 1 + if placeholder: + out.append(str(placeholder)) + continue + new = _RE_IMAGE.sub("", line) + if new != line: + n += 1 + line = new.rstrip() + out.append(line) + stats["image_refs"] = stats.get("image_refs", 0) + n + return "\n".join(out) + + +# HTML 表格规范化：
…
压缩为合法 Markdown 管道表格 +_RE_TABLE = re.compile(r"]*>(.*?)
", re.DOTALL | re.IGNORECASE) +_RE_TR = re.compile(r"]*>(.*?)", re.DOTALL | re.IGNORECASE) +_RE_TD = re.compile(r"]*>(.*?)", re.DOTALL | re.IGNORECASE) + + +def _cell_text(raw: str) -> str: + cell = re.sub(r"", " ", raw, flags=re.IGNORECASE) + cell = re.sub(r"<[^>]+>", "", cell) + return " ".join(cell.split()).replace("|", "\\|") + + +def _table_to_md(tbl_html: str) -> str: + rows: List[List[str]] = [] + for tr in _RE_TR.findall(tbl_html): + cells = [_cell_text(td) for td in _RE_TD.findall(tr)] + if cells: + rows.append(cells) + if not rows: + return "" + width = max(len(r) for r in rows) + rows = [r + [""] * (width - len(r)) for r in rows] + lines = ["| " + " | ".join(rows[0]) + " |", "|" + "---|" * width] + lines.extend("| " + " | ".join(r) + " |" for r in rows[1:]) + return "\n".join(lines) + + +def _normalize_tables(text: str, params: Dict[str, Any], stats: Dict[str, int]) -> str: + def _sub(m: re.Match[str]) -> str: + md = _table_to_md(m.group(1)) + return md if md else "" + + new, n = _RE_TABLE.subn(_sub, text) + # 残缺兜底：文档截断导致 未闭合时，把剩余 行也转掉 + if "" + m.group(0) + "
"), + tail, + ) + tail = re.sub(r"]*>", "", tail) + new = head + tail + n += 1 + stats["html_tables"] = stats.get("html_tables", 0) + n + return new + + +# 表格内
会被上面规则拍平；这里处理散落的 HTML 换行/空白标签 +def _strip_stray_html(text: str, params: Dict[str, Any], stats: Dict[str, int]) -> str: + new = re.sub(r"", " ", text, flags=re.IGNORECASE) + if new != text: + stats["stray_html"] = stats.get("stray_html", 0) + text.count(" str: + out, n = [], 0 + for line in text.splitlines(): + stripped = line.rstrip() + if stripped != line: + n += 1 + out.append(stripped) + stats["trailing_ws_lines"] = stats.get("trailing_ws_lines", 0) + n + return "\n".join(out) + + +# 连续空行压成一行；文首文末空白裁掉 +def _collapse_blank_lines( + text: str, params: Dict[str, Any], stats: Dict[str, int] +) -> str: + text = re.sub(r"[ \t]*\n(?:[ \t]*\n){2,}", "\n\n", text) + text = text.strip("\n") + "\n" if text.strip() else "" + stats["collapsed_blanks"] = stats.get("collapsed_blanks", 1) + return text + + +REGISTRY: Dict[str, RuleFunc] = { + "strip_page_lines": _strip_page_lines, + "strip_repeated_short_lines": _strip_repeated_short_lines, + "strip_toc_dots": _strip_toc_dots, + "drop_images": _drop_images, + "normalize_tables": _normalize_tables, + "strip_stray_html": _strip_stray_html, + "rstrip_lines": _rstrip_lines, + "collapse_blank_lines": _collapse_blank_lines, +} + + +def load_rules(path: Optional[Path] = None) -> List[Rule]: + """从 YAML 加载规则；未指定路径时用包内默认规则。""" + if path is None: + path = Path(__file__).resolve().parent.parent / "rules" / "default.yaml" + data = yaml.safe_load(Path(path).read_text(encoding="utf-8")) or {} + rules = [Rule.from_dict(d) for d in data.get("rules", [])] + unknown = [r.name for r in rules if r.name not in REGISTRY] + if unknown: + raise ValueError(f"未知规则: {unknown}，可用规则: {sorted(REGISTRY)}") + return sorted(rules, key=lambda r: r.order) diff --git a/docs/RULES.md b/docs/RULES.md new file mode 100644 index 0000000..2882943 --- /dev/null +++ b/docs/RULES.md @@ -0,0 +1,45 @@ +# 规则编写指南 + +一条清洗规则 = `cleaner/rules.py` 里的一个函数 + `rules/*.yaml` 里的一条声明。 + +## 函数签名 + +```python +def _your_rule(text: str, params: dict, stats: dict) -> str: + ... + stats["your_hits"] = stats.get("your_hits", 0) + n # 计入报告 + return new_text +``` + +- 输入输出都是**整篇文本**；引擎按 `order` 从小到大依次调用。 +- `params` 来自 YAML，改参数不用改代码。 +- `stats` 的 key 会出现在 `report.json`，命名用蛇形复数（如 `page_lines`）。 + +## YAML 声明 + +```yaml +rules: + - name: your_rule # 必须与 REGISTRY 键一致，加载时校验 + order: 55 # 应用顺序；同段处理尽量插在相关规则之间 + enabled: true # 某用例不适用的规则可单关 + params: + threshold: 3 + protect: ["正则1", "正则2"] +``` + +## 设计守则（从 gov_test_data 踩坑总结） + +1. **只删噪音，不删正文**。拿不准的形态默认保留，宁可漏删不可误删。 +2. **重复 ≠ 页眉**。标书是平行模板文档：签章栏、日期栏、声明结尾句、 + 编号条款都会重复出现。判定页眉前先过： + - `protect` 正则（业务模板白名单） + - 内容形态豁免（编号/列表/表格/字段行） + - burst 检测（连续刷屏才是 OCR 崩坏） +3. **每个规则独立可测**。tests/ 里用真实数据浓缩的最小样例做回归， + 尤其是防误伤样例（protect 命中、编号条款保留）。 +4. **统计必须可见**。每条规则报告命中数，批量清洗后扫一眼 report.json + 就能发现某条规则突然命中异常（多半是误伤）。 + +## 已知规则明细 + +见 `rules/default.yaml` 内注释与 README 规则表。 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..0dac4ab --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,21 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "govdoc-md-cleaner" +version = "0.1.0" +description = "政务文档（招标/投标/采购/合同）PDF→Markdown 产物的清洗工具" +readme = "README.md" +requires-python = ">=3.10" +license = { text = "MIT" } +dependencies = ["PyYAML>=6.0"] + +[project.scripts] +md-clean = "cleaner.cli:main" + +[tool.setuptools.packages.find] +include = ["cleaner*"] + +[tool.setuptools.package-data] +cleaner = ["../rules/*.yaml"] diff --git a/rules/default.yaml b/rules/default.yaml new file mode 100644 index 0000000..8d7b760 --- /dev/null +++ b/rules/default.yaml @@ -0,0 +1,55 @@ +# 政务文档 Markdown 默认清洗规则集 +# 每条规则: name(必填，须在引擎 REGISTRY 中注册) / order(应用顺序) / enabled / params +# 针对 PDF→Markdown 管线（MinerU/OCR）产物的典型脏数据。 + +rules: + # 1) 页码行："第 X 页 共 Y 页" / "第X页共4页" / "Page 3 of 10" / "3 / 10" + - name: strip_page_lines + order: 10 + params: + keep_bare_numbers: true # 单独一行纯数字（可能是页码也可能是编号），默认保留 + + # 2) 页眉页脚 + OCR 重复崩坏行： + # a) 同一短行全篇重复 >= threshold 次（如逐页出现的项目名、"正本"） + # b) 同一短行连续刷屏 >= burst_limit 次（OCR 崩坏，如"审计程序"×1359） + # protect 列出"重复但属于正文模板"的保护正则（标书里逐章出现的签章/日期栏） + # 内容形态行（编号条款/列表/表格行）天然重复，已在引擎侧豁免 + - name: strip_repeated_short_lines + order: 20 + params: + threshold: 3 + max_len: 40 + burst_limit: 5 + burst_gap: 2 + protect: + - "公章" # 投标人：（公章） + - "签名|签字|盖章" # 法定代表人签名: + - "日期|年.*月.*日" # 日期: / 日期: 年 月 日 + - "^致[:：]" # 致：xxx（投标函收件人） + - "负责|声明" # 声明函固定结尾句（中小企业声明函等） + + # 3) 目录点线："第一章 投标邀请函 ……………… 2"（保留标题文字，去掉点线和页码） + - name: strip_toc_dots + order: 30 + + # 4) 图片引用：![](images/xxx.jpg) 为死链，整行删除 + - name: drop_images + order: 40 + params: + placeholder: null # 需要保留位置时改为 "[图]" 之类 + + # 5) HTML 表格 → Markdown 管道表格（
单行压缩形态） + - name: normalize_tables + order: 50 + + # 6) 散落的
标签 + - name: strip_stray_html + order: 60 + + # 7) 行尾空白（MinerU 输出每行带双空格硬换行符） + - name: rstrip_lines + order: 70 + + # 8) 连续空行压为 1 行，裁掉文首文末空白 + - name: collapse_blank_lines + order: 80 diff --git a/tests/test_rules.py b/tests/test_rules.py new file mode 100644 index 0000000..49d5b17 --- /dev/null +++ b/tests/test_rules.py @@ -0,0 +1,154 @@ +"""清洗规则单元测试。 + +fixtures 里是各脏数据模式的最小样例，跑一遍断言规则命中且正文无损。 +""" + +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from cleaner.cleaner import MarkdownCleaner +from cleaner.rules import ( + _drop_images, + _normalize_tables, + _rstrip_lines, + _strip_page_lines, + _strip_repeated_short_lines, + _strip_toc_dots, + load_rules, +) + +S = {} # 每个用例独立 stats + + +def st(): + return {} + + +class TestPageLines(unittest.TestCase): + def test_cn_page(self): + text = "正文A\n第 3 页 共 53 页\n正文B\n第4页共4页\n结尾" + out = _strip_page_lines(text, {}, st()) + self.assertEqual(out, "正文A\n正文B\n结尾") + + def test_en_page(self): + text = "foo\nPage 3 of 10\nbar\n- 4 -\nbaz" + out = _strip_page_lines(text, {"keep_bare_numbers": False}, st()) + self.assertNotIn("Page 3", out) + self.assertNotIn("- 4 -", out) + + def test_bare_number_kept_by_default(self): + text = "条款\n12\n下文" + out = _strip_page_lines(text, {}, st()) + self.assertIn("12", out) + + +class TestRepeatedLines(unittest.TestCase): + def test_header_removed(self): + lines = ["某某采购项目招标文件"] + ["内容%d" % i for i in range(5)] + text = "\n".join(("某某采购项目招标文件 \n" + l) for l in lines) + out = _strip_repeated_short_lines(text, {"threshold": 3, "max_len": 40}, st()) + self.assertNotIn("某某采购项目招标文件", out) + self.assertIn("内容1", out) + + def test_protected_signature_kept(self): + text = "\n".join(["投标人：（公章）"] * 4 + ["正文"]) + out = _strip_repeated_short_lines( + text, {"threshold": 3, "max_len": 40, "protect": ["公章"]}, st() + ) + self.assertIn("投标人：（公章）", out) + + def test_numbered_clause_kept(self): + # 编号条款是正文不是页眉 + text = "\n".join(["（1）乙方须接受甲方监督。"] * 4 + ["正文"]) + out = _strip_repeated_short_lines(text, {"threshold": 3, "max_len": 40}, st()) + self.assertIn("（1）乙方须接受甲方监督。", out) + + def test_ocr_burst_removed(self): + text = "\n".join(["审计程序"] * 30) + out = _strip_repeated_short_lines( + text, + {"threshold": 999, "max_len": 40, "burst_limit": 5, "burst_gap": 2}, + st(), + ) + self.assertNotIn("审计程序", out) + + +class TestTocDots(unittest.TestCase): + def test_toc_line(self): + text = "第一章 投标邀请函 ……………………………………… 2" + out = _strip_toc_dots(text, {}, st()) + self.assertEqual(out, "第一章 投标邀请函") + + def test_plain_toc_line_keeps_text(self): + # 无标题目录行：去掉点线页码，保留"序号+标题"文字 + text = "21 迷交的投标文件 ………………………………………………… 25" + out = _strip_toc_dots(text, {}, st()) + self.assertEqual(out, "21 迷交的投标文件") + + def test_body_with_ellipsis_kept(self): + text = "此处省略部分内容……后续" + out = _strip_toc_dots(text, {}, st()) + self.assertIn("后续", out) + + +class TestImages(unittest.TestCase): + def test_image_line_dropped(self): + text = "![](images/abc.jpg) \n正文" + out = _drop_images(text, {}, st()) + self.assertEqual(out, "正文") + + def test_placeholder(self): + text = "![](images/abc.jpg)\n正文" + out = _drop_images(text, {"placeholder": "[图]"}, st()) + self.assertIn("[图]", out) + + def test_base64_image_dropped(self): + text = "![名称](data:image/png;base64,AAAA)\n正文" + out = _drop_images(text, {}, st()) + self.assertNotIn("base64", out) + + +class TestTables(unittest.TestCase): + def test_table_to_pipe(self): + html = "
序号名称
1保洁
" + out = _normalize_tables(html, {}, st()) + self.assertIn("| 序号 | 名称 |", out) + self.assertIn("| 1 | 保洁 |", out) + self.assertIn("|---|---|", out) + + def test_pipe_escaped(self): + html = "
a|b
" + out = _normalize_tables(html, {}, st()) + self.assertIn("a\\|b", out) + + def test_br_in_cell(self): + html = "
品
目
" + out = _normalize_tables(html, {}, st()) + self.assertIn("品 目", out) + + +class TestEngine(unittest.TestCase): + def test_full_pipeline(self): + cleaner = MarkdownCleaner(rules=load_rules()) + raw = ( + "# 标题\n![](images/x.jpg) \n第 1 页 共 2 页 \n" + "正文一段。 \n
a
\n\n\n\n尾部\n" + ) + result = cleaner.clean_text(raw) + self.assertNotIn("images/", result.text) + self.assertNotIn("第 1 页", result.text) + self.assertNotIn("", result.text) + self.assertIn("| a |", result.text) + self.assertNotIn("\n\n\n", result.text) + + def test_default_rules_load(self): + rules = load_rules() + self.assertTrue(len(rules) >= 8) + self.assertEqual(rules, sorted(rules, key=lambda r: r.order)) + + +if __name__ == "__main__": + unittest.main(verbosity=2)