Bepr4
c19e9fcebf
- 8 条 YAML 声明规则：页码/页眉页脚/目录点线/图片/HTML表格/散落标签/行尾空白/空行 - 防误伤设计：protect 正则 + 内容形态豁免 + OCR burst 检测 - md-clean single/batch CLI，JSON 清洗报告 - 18 个单元测试 Co-Authored-By: Claude <noreply@anthropic.com>
282 lines
10 KiB
Python
282 lines
10 KiB
Python
"""规则模型：一条清洗规则 = 名称 + 开关 + 参数 + 应用顺序。
|
||
|
||
规则用 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)
|
||
|
||
|
||
# 图片引用： —— 图片目录不在交付物里，引用是死链
|
||
_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 表格规范化：<table ...><tr><td>…</td></tr></table> 压缩为合法 Markdown 管道表格
|
||
_RE_TABLE = re.compile(r"<table[^>]*>(.*?)</table>", re.DOTALL | re.IGNORECASE)
|
||
_RE_TR = re.compile(r"<tr[^>]*>(.*?)</tr>", re.DOTALL | re.IGNORECASE)
|
||
_RE_TD = re.compile(r"<t[dh][^>]*>(.*?)</t[dh]>", re.DOTALL | re.IGNORECASE)
|
||
|
||
|
||
def _cell_text(raw: str) -> str:
|
||
cell = re.sub(r"<br\s*/?>", " ", 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)
|
||
# 残缺兜底：文档截断导致 <table> 未闭合时，把剩余 <tr> 行也转掉
|
||
if "<table" in new:
|
||
tail_i = new.rfind("<table")
|
||
head, tail = new[:tail_i], new[tail_i:]
|
||
tail = _RE_TR.sub(
|
||
lambda m: "\n" + _table_to_md("<table>" + m.group(0) + "</table>"),
|
||
tail,
|
||
)
|
||
tail = re.sub(r"</?table[^>]*>", "", tail)
|
||
new = head + tail
|
||
n += 1
|
||
stats["html_tables"] = stats.get("html_tables", 0) + n
|
||
return new
|
||
|
||
|
||
# 表格内 <br/> 会被上面规则拍平；这里处理散落的 HTML 换行/空白标签
|
||
def _strip_stray_html(text: str, params: Dict[str, Any], stats: Dict[str, int]) -> str:
|
||
new = re.sub(r"<br\s*/?>", " ", text, flags=re.IGNORECASE)
|
||
if new != text:
|
||
stats["stray_html"] = stats.get("stray_html", 0) + text.count("<br")
|
||
return new
|
||
|
||
|
||
# 行尾双空格（MinerU 每行都带，语义是硬换行，清洗后统一去掉）
|
||
def _rstrip_lines(text: str, params: Dict[str, Any], stats: Dict[str, int]) -> 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)
|