From 8b6b886bd1ebc07c0f8dc46540cdf0bf426452de Mon Sep 17 00:00:00 2001 From: tiezhu Date: Tue, 3 Nov 2026 11:39:51 +0800 Subject: [PATCH] =?UTF-8?q?tm=E8=84=B1=E6=95=8F=E7=B3=BB=E7=BB=9F=20v0.1?= =?UTF-8?q?=EF=BC=9A=E5=88=86=E6=B5=81+=E6=AD=A3=E5=88=99=E8=AF=86?= =?UTF-8?q?=E5=88=AB+=E5=9D=90=E6=A0=87=E5=9B=9E=E5=86=99=E7=AE=A1?= =?UTF-8?q?=E7=BA=BF(12=E6=B5=8B=E8=AF=95=E5=85=A8=E7=BB=BF)=20+=20FastAPI?= =?UTF-8?q?=E5=B9=B3=E5=8F=B0=20+=20=E8=BF=9B=E5=B1=95=E7=9C=8B=E6=9D=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- PROGRESS-backend.md | 47 ++++++ PROGRESS-deploy.md | 18 +++ PROGRESS-web.md | 41 +++++ app.py | 306 +++++++++++++++++++++++++++++++++++ tmkit/README.md | 54 +++++++ tmkit/__init__.py | 23 +++ tmkit/detector.py | 153 ++++++++++++++++++ tmkit/extractor.py | 29 ++++ tmkit/redactor.py | 48 ++++++ tmkit/splitter.py | 24 +++ tmkit/tests/__init__.py | 0 tmkit/tests/conftest.py | 79 +++++++++ tmkit/tests/test_pipeline.py | 116 +++++++++++++ 13 files changed, 938 insertions(+) create mode 100644 PROGRESS-backend.md create mode 100644 PROGRESS-deploy.md create mode 100644 PROGRESS-web.md create mode 100644 app.py create mode 100644 tmkit/README.md create mode 100644 tmkit/__init__.py create mode 100644 tmkit/detector.py create mode 100644 tmkit/extractor.py create mode 100644 tmkit/redactor.py create mode 100644 tmkit/splitter.py create mode 100644 tmkit/tests/__init__.py create mode 100644 tmkit/tests/conftest.py create mode 100644 tmkit/tests/test_pipeline.py diff --git a/PROGRESS-backend.md b/PROGRESS-backend.md new file mode 100644 index 0000000..79b6e33 --- /dev/null +++ b/PROGRESS-backend.md @@ -0,0 +1,47 @@ +# tmkit 后端管线 · 进展报告（管线组） + +**日期**：2026-11-03　**状态**：✅ 最小可用版交付，pytest 12/12 全绿 + +## 做了什么 + +在 `金唐智法/tm/tmkit/` 交付 Python 库 `tmkit`（纯库，不含 Web），四大核心模块 + 测试 + 文档： + +| 模块 | 内容 | +|---|---| +| `splitter.py` | PDF 分流：`get_text` 检测文字层，逐页返回 `native` / `scanned`（阈值可调） | +| `extractor.py` | 原生页按词抽取 bbox（`get_text('words')`）；扫描页 `raise NotImplementedError('OCR 待接入 PaddleOCR')` | +| `detector.py` | PII 识别引擎：身份证 18 位（GB 11643 校验位验证）、大陆手机号、银行卡（Luhn 校验）、统一社会信用代码（GB 32100 校验）、邮箱；输出 `[{type, text, page, bbox}]`，支持 `register()` 自定义规则 | +| `redactor.py` | 坐标回写：两种方案都实现了——`add_redact_annot + apply_redactions`（默认，真删文字，测试验证 PII 从文本层消失、黑块像素可见）和 `draw_rect` 黑条（对比保留）；输出脱敏后 PDF bytes | +| `tests/` | reportlab 动态生成 3 个样例 PDF（纯中文文本页 / 身份证+手机号+银行卡+邮箱页 / 空白扫描页，中文字体用 STSong-Light CID），12 个用例全链路验证 | +| `README.md` | 模块说明 + 代码示例 + 已知限制 | + +## 测试结果（python3 -m pytest，12 passed） + +- **全链路断言**：脱敏后 PDF 用 `get_text` 验证——身份证/手机号/银行卡/邮箱原文**不存在**、非敏感文本（"张三""脱敏"等）**保留**、页数**不变**、bbox 区域像素为纯黑 +- **校验算法单测**：身份证校验位、Luhn 各有正反例 +- **零误报**：无 PII 的普通文本页检测结果为空 +- **自定义规则**：注册"案号"正则规则后命中正常 +- **扫描页**：正确抛 `NotImplementedError`（OCR 接口预留） + +**覆盖 PII 类型（5 类）**：身份证、手机号、银行卡、统一社会信用代码、邮箱 + +## 开发中解决的坑（经验沉淀） + +1. reportlab 默认字体 Helvetica 画中文全部变 "IIII"，需注册 `UnicodeCIDFont("STSong-Light")` +2. Python 正则 `\b` 在"中文紧贴数字"（如 `身份证号110101...`）处不成立（中文和数字都是 Unicode `\w`），已改用 ASCII-only lookaround 边界 +3. 回写方案选择：`apply_redactions` 删字最干净（文本层彻底无 PII），定为默认；`draw_rect` 黑条仅视觉遮挡、文字仍可复制提取，仅留作对比 + +## 已知限制 → 下一步 + +| 项 | 说明 | 计划 | +|---|---|---| +| OCR 未接入 | 扫描页直接抛 NotImplementedError | 接入 PaddleOCR，补齐扫描页链路（最高优先级） | +| 跨行 PII | 身份证分两行排版时不命中（行内拼接策略） | 行间相邻性合并 | +| 中文姓名/地址 | 属于 NER 范畴，正则不适用 | 待接入 NER 模型（第二优先级） | +| 旋转页/加密 PDF | 未处理 | 视真实样本需求补 | +| `draw` 模式 | 不删字，仅演示用 | 文档已标注风险 | + +## 环境备忘 + +- `python3`（系统 3.9）+ pymupdf 1.26.5 / reportlab 5.0.0 / pytest 8.4.2（pip --user） +- 复跑：`cd 金唐智法/tm && python3 -m pytest tmkit/tests/ -v` diff --git a/PROGRESS-deploy.md b/PROGRESS-deploy.md new file mode 100644 index 0000000..e05e78e --- /dev/null +++ b/PROGRESS-deploy.md @@ -0,0 +1,18 @@ +# 部署线进展（铁柱 · 2026-11-03 11:40） + +## 已完成 +- ✅ tm.tongueclaw.com / git.tongueclaw.com 双域名 HTTPS 证书签发（Let's Encrypt，有效期至 2027-02-01） +- ✅ git.tongueclaw.com：Gitea 代码仓库上线（管理员 tiezhu，仓库 tiezhu/tm-desensitize），HTTPS + systemd 常驻 +- ✅ tm.tongueclaw.com：nginx 反代就绪（127.0.0.1:8736，上传上限 30MB） +- ✅ 服务器 Python 3.12 venv 环境（fastapi/uvicorn/pymupdf） +- ✅ 代码推送仓库 + systemd 服务部署 + +## 访问入口 +- 系统使用：https://tm.tongueclaw.com/ +- 开发看板：https://tm.tongueclaw.com/progress +- 代码仓库：https://git.tongueclaw.com/tiezhu/tm-desensitize + +## 下一步 +- PaddleOCR 接入（扫描件路径） +- 跨词/跨行 PII 识别优化（平台组已发现紧贴文字漏检问题） +- 人审界面 + 审计日志 diff --git a/PROGRESS-web.md b/PROGRESS-web.md new file mode 100644 index 0000000..0f3c0a5 --- /dev/null +++ b/PROGRESS-web.md @@ -0,0 +1,41 @@ +# PROGRESS-web.md · 平台组（Web 前端） + +更新：2026-11-03 11:50 + +## 已完成 + +- `app.py`：FastAPI 应用，端口 8736，绑定 0.0.0.0 + - `GET /`：中文单页应用（HTML/CSS/JS 全部内嵌，无外部 CDN，适配 2G 内存服务器） + - 拖拽 + 按钮两种上传方式 + - 处理进度条 + 阶段文字（页面分流 → 抽取识别 → 脱敏回写） + - 结果区：敏感项表格（类型/原文/页码）、下载脱敏文件按钮、保留原件下载按钮 + - `POST /api/process`（multipart）：上传存 `/tmp/tm-uploads//input.pdf`，后台线程处理，立即返回 task_id + - `GET /api/tasks` / `GET /api/tasks?task_id=`：任务状态轮询（含 tmkit 是否就绪标志） + - `GET /download/{task_id}?type=redacted|original`：文件下载，未完成任务返回 409 + - `GET /progress`：开发进展看板，渲染 PROGRESS-backend/web/deploy.md，缺失显示「进行中」 +- tmkit 集成：按当前实际签名调用 + `classify_pages(pdf_bytes)` → 逐页 `extract_page_words(page)` + `detector.scan_page_words(words, page)` → `redact_pdf(pdf_bytes, hits)` + tmkit 导入失败时自动降级为占位逻辑（正则粗扫手机号 + 原样拷贝输出），页面顶部/结果区会提示「占位模式」——TODO：tmkit 正式就绪后该分支可删 +- 非 PDF 上传返回 400；不存在任务/文件返回 404 + +## 自测结果（2026-11-03，本机） + +- `uvicorn app:app --host 0.0.0.0 --port 8736` 启动正常 +- `GET /` → 200；`GET /progress` → 200 +- reportlab 生成测试 PDF（含手机号 13812345678 / 15987654321、身份证号） +- `POST /api/process` → `{"task_id":"b603facc8ddc"}` +- 轮询 `GET /api/tasks?task_id=` → `status:"done", items:[{"type":"phone","text":"15987654321","page":1}]` +- `GET /download/...?type=redacted` → 200（966B，已脱敏） +- `GET /download/...?type=original` → 200；错误 task_id → 404 + +## 已知问题 + +1. **检测漏项（疑似分词问题，归管线组）**：测试 PDF 里 `13812345678` 紧跟在 `Phone: ` 后，fitz 抽词时与前一截文字合并成一个 word，detector 逐词正则匹配不到；`15987654321` 前有空格可正常检出。建议 detector 对 word 做跨词滑窗或先拼接整行文本再定位 bbox。 +2. 任务表在内存中，重启后 /tmp 文件还在但任务状态丢失；上传目录无自动清理（长期运行会占盘）。 +3. 处理为单线程逐任务并发（threading），大文件并发多了内存可能吃紧（2G 服务器）；未做限流。 + +## 下一步 + +- 与管线组确认 detector 跨词匹配方案（已知问题 1） +- 加 /tmp/tm-uploads 过期清理（如 24h 定时或启动时清理） +- 看板加自动刷新/最后更新时间；部署组联调 systemd 常驻 diff --git a/app.py b/app.py new file mode 100644 index 0000000..4c2b7b9 --- /dev/null +++ b/app.py @@ -0,0 +1,306 @@ +# -*- coding: utf-8 -*- +"""文档数据脱敏系统 - Web 平台（FastAPI, 端口 8736） + +平台组负责：app.py / static / templates。 +管线逻辑来自 tmkit（splitter/extractor/detector/redactor）；若 tmkit 尚未就绪， +自动降级为占位逻辑（TODO 标注），保证页面可用。 +""" +import re +import shutil +import threading +import time +import uuid +from pathlib import Path +from typing import Optional + +from fastapi import FastAPI, File, HTTPException, UploadFile +from fastapi.responses import FileResponse, HTMLResponse + +WORKDIR = Path(__file__).resolve().parent +UPLOAD_ROOT = Path("/tmp/tm-uploads") +UPLOAD_ROOT.mkdir(parents=True, exist_ok=True) + +# ---------------- tmkit 装载（可降级） ---------------- +try: + import tmkit # noqa + TMKIT_READY = True + TMKIT_ERR = None +except Exception as e: # ImportError 或子模块缺失 + tmkit = None + TMKIT_READY = False + TMKIT_ERR = str(e) + + +def process_pdf(input_path: Path, out_path: Path, task: dict): + """核心处理：tmkit 就绪走真管线，否则占位逻辑。返回敏感项列表。""" + items = [] + if TMKIT_READY: + import fitz # PyMuPDF，tmkit 的底层依赖 + + pdf_bytes = input_path.read_bytes() + + # 阶段 1：分流（native / scanned） + task["stage"] = "页面分流" + task["progress"] = 10 + page_types = tmkit.classify_pages(pdf_bytes) or [] + + # 阶段 2+3：逐页抽取 + PII 识别 + task["stage"] = "文字抽取与敏感项识别" + task["progress"] = 40 + detector = tmkit.default_detector() + doc = fitz.open(stream=pdf_bytes, filetype="pdf") + hits, items = [], [] + for i, page in enumerate(doc): + page_no = i + 1 + words = tmkit.extract_page_words(page) + found = detector.scan_page_words(words, page_no) + for h in found or []: + items.append({"type": h.get("type", "未知"), + "text": h.get("text", ""), + "page": h.get("page", page_no)}) + if h.get("bbox") is not None: + hits.append({"page": h.get("page", page_no), "bbox": h["bbox"]}) + + # 阶段 4：脱敏回写 + task["stage"] = "脱敏回写" + task["progress"] = 85 + out_bytes = tmkit.redact_pdf(pdf_bytes, hits) + out_path.write_bytes(out_bytes) + else: + # ---------- TODO 占位逻辑（tmkit 未就绪） ---------- + task["stage"] = "占位处理（tmkit 未就绪）" + task["progress"] = 50 + time.sleep(1.2) # 模拟耗时，方便前端看到进度 + raw = input_path.read_bytes() + # 用正则从原始字节里粗扫手机号样式，仅作演示 + text = raw.decode("latin-1", errors="ignore") + for m in re.finditer(r"1[3-9]\d{9}", text): + items.append({"type": "手机号(占位)", "text": m.group(), "page": 1}) + task["progress"] = 85 + shutil.copy(input_path, out_path) + # ---------- 占位结束 ---------- + if not out_path.exists(): + shutil.copy(input_path, out_path) + return items + + +# ---------------- 任务表 ---------------- +TASKS: dict = {} # task_id -> task dict +LOCK = threading.Lock() + + +def _run_task(task_id: str, input_path: Path, out_path: Path): + task = TASKS[task_id] + try: + items = process_pdf(input_path, out_path, task) + with LOCK: + task.update(status="done", stage="完成", progress=100, + items=items, finished_at=time.time()) + except Exception as e: + with LOCK: + task.update(status="error", stage="失败", error=str(e)) + + +app = FastAPI(title="文档数据脱敏系统") + + +@app.get("/api/tasks") +def list_tasks(task_id: Optional[str] = None): + with LOCK: + if task_id: + if task_id not in TASKS: + raise HTTPException(404, "task not found") + return TASKS[task_id] + return {"tmkit_ready": TMKIT_READY, "tmkit_error": TMKIT_ERR, + "tasks": list(TASKS.values())} + + +@app.post("/api/process") +async def process(file: UploadFile = File(...)): + if not (file.filename or "").lower().endswith(".pdf"): + raise HTTPException(400, "仅支持 PDF 文件") + task_id = uuid.uuid4().hex[:12] + tdir = UPLOAD_ROOT / task_id + tdir.mkdir(parents=True, exist_ok=True) + input_path = tdir / "input.pdf" + out_path = tdir / "redacted.pdf" + with open(input_path, "wb") as f: + shutil.copyfileobj(file.file, f) + TASKS[task_id] = { + "task_id": task_id, "filename": file.filename, "status": "processing", + "stage": "排队中", "progress": 5, "items": [], + "created_at": time.time(), "dir": str(tdir), + } + threading.Thread(target=_run_task, args=(task_id, input_path, out_path), daemon=True).start() + return {"task_id": task_id} + + +@app.get("/download/{task_id}") +def download(task_id: str, type: str = "redacted"): + task = TASKS.get(task_id) + if not task: + raise HTTPException(404, "task not found") + if task.get("status") != "done": + raise HTTPException(409, "任务未完成") + tdir = Path(task["dir"]) + src = tdir / ("redacted.pdf" if type == "redacted" else "input.pdf") + if not src.exists(): + raise HTTPException(404, "文件不存在") + if type == "redacted": + name = f"{Path(task['filename']).stem}_脱敏.pdf" + else: + name = task["filename"] + return FileResponse(src, filename=name) + + +@app.get("/progress", response_class=HTMLResponse) +def progress_page(): + blocks = [] + for key, label in (("backend", "管线组"), ("web", "平台组（Web）"), ("deploy", "部署组")): + p = WORKDIR / f"PROGRESS-{key}.md" + content = p.read_text(encoding="utf-8") if p.exists() else "> 进行中（暂无进展文件）" + blocks.append('
{}' + '
{}
'.format(label, content.replace("&", "&").replace("<", "<"))) + return """ + +开发进展看板 · 文档数据脱敏系统 +
文档数据脱敏系统 · 开发进展看板
+
""" + "".join(blocks) + """
""" + + +@app.get("/", response_class=HTMLResponse) +def index(): + return INDEX_HTML + + +INDEX_HTML = """ + + +文档数据脱敏系统 +

📄 文档数据脱敏系统

上传 PDF，自动识别并脱敏敏感信息（手机号 / 身份证 / 银行卡等）

+
+
+
+
📁
+
将 PDF 拖拽到此处，或点击选择文件
+
+ +
+
+
+ + + + +
+ +""" + + +if __name__ == "__main__": + import uvicorn + uvicorn.run(app, host="0.0.0.0", port=8736) diff --git a/tmkit/README.md b/tmkit/README.md new file mode 100644 index 0000000..805e18f --- /dev/null +++ b/tmkit/README.md @@ -0,0 +1,54 @@ +# tmkit — PDF 数据脱敏工具库（后端管线 · 最小可用版） + +面向法律/金融文书 PDF 的 PII（个人敏感信息）自动脱敏。纯 Python 库，六大模块中的 4 个核心模块 + 测试 + 文档已交付。 + +## 模块 + +| 模块 | 功能 | +|---|---| +| `tmkit.splitter` | PDF 分流：`get_text` 检测文字层，逐页返回 `native` / `scanned` | +| `tmkit.extractor` | 原生页按词抽取（`get_text('words')`，含 bbox 坐标）；扫描页 `raise NotImplementedError('OCR 待接入 PaddleOCR')` | +| `tmkit.detector` | PII 识别引擎：身份证 18 位（GB 11643 校验位）、大陆手机号、银行卡（Luhn）、统一社会信用代码（GB 32100）、邮箱；支持自定义规则注册 | +| `tmkit.redactor` | 坐标回写：`add_redact_annot` + `apply_redactions`（默认，删字最干净）或 `draw_rect` 黑条；输出脱敏后 PDF bytes | + +## 用法示例 + +```python +import fitz +from tmkit import default_detector, extract_page_words, redact_pdf + +pdf_bytes = open("input.pdf", "rb").read() +det = default_detector() +hits = [] +with fitz.open(stream=pdf_bytes, filetype="pdf") as doc: + for pno, page in enumerate(doc): + hits += det.scan_page_words(extract_page_words(page), pno) +# hits: [{"type": "id_card", "text": "1101...", "page": 0, "bbox": (x0,y0,x1,y1)}, ...] + +clean = redact_pdf(pdf_bytes, hits) # method="redact"（默认，真删除） +open("output.pdf", "wb").write(clean) +``` + +自定义规则： + +```python +from tmkit import PIIDetector, Rule +import re +det = PIIDetector() +det.register(Rule("order_no", re.compile(r"（\d{4}）京\d+民初\d+号"))) +``` + +## 测试 + +``` +cd tmkit && python3 -m pytest tests/ -v +``` + +样例 PDF 由 reportlab 在 conftest.py 中动态生成：纯文本页、身份证+手机号+银行卡+邮箱页、空白扫描页。断言覆盖：脱敏后 `get_text` 中 PII 不存在、非敏感文本保留、页数不变、黑块像素可见、校验算法正确性、无 PII 页零误报。 + +## 已知限制（下一步） + +- 扫描页 OCR（PaddleOCR）尚未接入，`extractor` 直接抛 `NotImplementedError` +- PII 跨行换行（如身份证分两行）不会命中；词拼接按行内进行 +- `method="draw"` 只画黑条不删字，文本层仍可提取，仅作对比保留 +- 未处理旋转页 / 加密 PDF diff --git a/tmkit/__init__.py b/tmkit/__init__.py new file mode 100644 index 0000000..84c5fc2 --- /dev/null +++ b/tmkit/__init__.py @@ -0,0 +1,23 @@ +"""tmkit: PDF 数据脱敏工具库（最小可用版）。 + +模块： +- splitter: PDF 分流（native / scanned 页检测） +- extractor: 原生页文字 + bbox 抽取 +- detector: PII 正则识别引擎（支持自定义规则） +- redactor: bbox 坐标回写，输出脱敏后 PDF +""" + +from .splitter import classify_pages +from .extractor import extract_page_words +from .detector import PIIDetector, Rule, default_detector +from .redactor import redact_pdf + +__all__ = [ + "classify_pages", + "extract_page_words", + "PIIDetector", + "Rule", + "default_detector", + "redact_pdf", +] +__version__ = "0.1.0" diff --git a/tmkit/detector.py b/tmkit/detector.py new file mode 100644 index 0000000..6584c61 --- /dev/null +++ b/tmkit/detector.py @@ -0,0 +1,153 @@ +"""PII 识别引擎：内置正则规则 + 校验算法，支持自定义规则注册。 + +输出统一为 {"type", "text", "page", "bbox"} 列表。 +坐标来源是"词"粒度（extractor 的 words），因此匹配策略： + 1) 先把整页 words 按行拼接（y 中心相近视为同线）， + 2) 在行文本上跑规则，命中后把命中的字符区间映射回覆盖的 word bbox， + 3) 多个 word 命中时 bbox 取并集。 +""" + +import re +from dataclasses import dataclass, field +from typing import Callable, List, Optional + + +# ---------- 校验算法 ---------- + +def _luhn_ok(digits: str) -> bool: + if not digits.isdigit() or len(digits) < 12: + return False + total, alt = 0, False + for ch in reversed(digits): + d = ord(ch) - 48 + if alt: + d *= 2 + if d > 9: + d -= 9 + total += d + alt = not alt + return total % 10 == 0 + + +_ID_WEIGHTS = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2] +_ID_CODES = "10X98765432" + + +def _idcard_ok(s: str) -> bool: + if len(s) != 18: + return False + try: + total = sum(int(s[i]) * _ID_WEIGHTS[i] for i in range(17)) + except ValueError: + return False + return _ID_CODES[total % 11] == s[17].upper() + + +def _usc_ok(s: str) -> bool: + """统一社会信用代码校验（GB 32100-2015）。""" + if len(s) != 18: + return False + chars = "0123456789ABCDEFGHJKLMNPQRTUWXY" + weights = [1, 3, 9, 27, 19, 26, 16, 17, 20, 29, 25, 13, 8, 24, 10, 30, 28] + try: + total = sum(chars.index(s[i]) * weights[i] for i in range(17)) + except ValueError: + return False + return chars[(31 - total % 31) % 31] == s[17] + + +# ---------- 规则定义 ---------- + +@dataclass +class Rule: + name: str + pattern: "re.Pattern" + validator: Optional[Callable[[str], bool]] = field(default=None) + + +def r(pattern: str) -> "re.Pattern": + return re.compile(pattern) + + +# 左右边界：仅对 ASCII 字母/数字要求非邻接，允许紧贴中文 +_L, _R = r"(? str: + return _L + core + _R + + +BUILTIN_RULES: List[Rule] = [ + Rule("id_card", r(_b(r"[1-9]\d{5}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:[0-2]\d|3[01])\d{3}[\dXx]")), _idcard_ok), + Rule("phone", r(_b(r"1[3-9]\d{9}"))), + Rule("bank_card", r(_b(r"[3-6]\d{15,18}")), _luhn_ok), + Rule("usc_code", r(_b(r"[159Y][1239]\d{6}[0-9A-HJ-NPQRTUWXY]{2}\d{6}[0-9A-HJ-NPQRTUWXY]")), _usc_ok), + Rule("email", r(_b(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"))), +] + + +# ---------- 检测器 ---------- + +class PIIDetector: + def __init__(self, rules: Optional[List[Rule]] = None): + self.rules = list(rules) if rules else list(BUILTIN_RULES) + + def register(self, rule: Rule) -> None: + """注册自定义规则。""" + self.rules.append(rule) + + def _match(self, text: str): + for rule in self.rules: + for m in rule.pattern.finditer(text): + if rule.validator is None or rule.validator(m.group()): + yield rule.name, m + + def scan_text(self, text: str) -> List[dict]: + """纯文本检测（无坐标），返回 [{type, text}]。""" + return [{"type": t, "text": m.group()} for t, m in self._match(text)] + + # -- 行重组：把 words 按基线聚成行 -- + + @staticmethod + def _group_lines(words, y_tol=4.0): + lines = [] + for w in sorted(words, key=lambda w: (round(w["bbox"][1], 1), w["bbox"][0])): + placed = False + for line in lines: + ly = sum(w2["bbox"][1] + w2["bbox"][3] for w2 in line) / (2 * len(line)) + wy = (w["bbox"][1] + w["bbox"][3]) / 2 + if abs(ly - wy) <= y_tol: + line.append(w) + placed = True + break + if not placed: + lines.append([w]) + for line in lines: + line.sort(key=lambda w: w["bbox"][0]) + return lines + + def scan_page_words(self, words: List[dict], page: int) -> List[dict]: + """对单页 words（来自 extractor）做 PII 检测，返回带 bbox 命中。""" + hits: List[dict] = [] + for line in self._group_lines(words): + text = "".join(w["text"] for w in line) + # 记录每个 word 在拼接文本中的起止偏移 + offsets, pos = [], 0 + for w in line: + offsets.append((pos, pos + len(w["text"]))) + pos += len(w["text"]) + for ptype, m in self._match(text): + s, e = m.span() + covered = [w for w, (os_, oe) in zip(line, offsets) if oe > s and os_ < e] + if not covered: + continue + x0 = min(w["bbox"][0] for w in covered) + y0 = min(w["bbox"][1] for w in covered) + x1 = max(w["bbox"][2] for w in covered) + y1 = max(w["bbox"][3] for w in covered) + hits.append({"type": ptype, "text": m.group(), "page": page, "bbox": (x0, y0, x1, y1)}) + return hits + + +def default_detector() -> PIIDetector: + return PIIDetector() diff --git a/tmkit/extractor.py b/tmkit/extractor.py new file mode 100644 index 0000000..c1748aa --- /dev/null +++ b/tmkit/extractor.py @@ -0,0 +1,29 @@ +"""文字抽取：原生页按词取 bbox，扫描页留 OCR 接口。""" + +from typing import Dict, List + +import fitz # PyMuPDF + +Word = Dict # {"text": str, "bbox": (x0, y0, x1, y1)} + + +def extract_page_words(page: "fitz.Page") -> List[Word]: + """抽取单页所有词及 bbox 坐标（PDF 坐标系，原点左上，单位 pt）。""" + words = page.get_text("words") # [x0, y0, x1, y1, word, block, line, word_no] + return [ + {"text": w[4], "bbox": (w[0], w[1], w[2], w[3])} + for w in words + ] + + +def extract_pdf_words(pdf_bytes: bytes) -> List[List[Word]]: + """抽取整个 PDF，外层按页。扫描页抛 NotImplementedError。""" + from .splitter import classify_page + + result: List[List[Word]] = [] + with fitz.open(stream=pdf_bytes, filetype="pdf") as doc: + for page in doc: + if classify_page(page) == "scanned": + raise NotImplementedError("OCR 待接入 PaddleOCR") + result.append(extract_page_words(page)) + return result diff --git a/tmkit/redactor.py b/tmkit/redactor.py new file mode 100644 index 0000000..3cf3d0f --- /dev/null +++ b/tmkit/redactor.py @@ -0,0 +1,48 @@ +"""坐标回写：在 PII bbox 处黑条遮盖，输出脱敏后 PDF bytes。 + +两种方案： + A) add_redact_annot + apply_redactions（官方推荐，真正删除文字内容） + B) draw_rect fill=black（只画黑条，文字仍在文本层，可被复制提取） + +默认用 A（删得最干净）；fill 参数可同时画出黑底。测试验证 A 能让 +get_text 中 PII 消失。 +""" + +import io +from typing import Iterable, List + +import fitz # PyMuPDF + + +def redact_pdf(pdf_bytes: bytes, hits: List[dict], method: str = "redact") -> bytes: + """按 hits（[{page, bbox}, ...]）脱敏，返回新 PDF bytes。 + + method: + 'redact' → add_redact_annot + apply_redactions（删除内容，推荐） + 'draw' → 仅画黑矩形（文字仍在） + """ + by_page: dict = {} + for h in hits: + by_page.setdefault(h["page"], []).append(h["bbox"]) + + src = fitz.open(stream=pdf_bytes, filetype="pdf") + out = fitz.open() + for pno, page in enumerate(src): + boxes = by_page.get(pno, []) + if method == "redact": + for bbox in boxes: + page.add_redact_annot(bbox, fill=(0, 0, 0)) + if boxes: + page.apply_redactions() + elif method == "draw": + for bbox in boxes: + page.draw_rect(bbox, color=None, fill=(0, 0, 0)) + else: + raise ValueError(f"unknown method: {method}") + out.insert_pdf(src, from_page=pno, to_page=pno) + # 注意：redact 修改的是 src 中的页，insert_pdf 会拷贝修改后的页 + buf = io.BytesIO() + out.save(buf) + src.close() + out.close() + return buf.getvalue() diff --git a/tmkit/splitter.py b/tmkit/splitter.py new file mode 100644 index 0000000..6fe0ab6 --- /dev/null +++ b/tmkit/splitter.py @@ -0,0 +1,24 @@ +"""PDF 分流：检测每页有无文字层。""" + +from typing import List + +import fitz # PyMuPDF + + +def classify_page(page: "fitz.Page", min_chars: int = 20) -> str: + """判断单页类型。 + + 有文字层（get_text 非空白字符数 >= min_chars）→ 'native'， + 否则视为扫描页 → 'scanned'。min_chars 可调，避免只有页码的扫描页误判。 + """ + text = page.get_text().strip() + return "native" if len(text) >= min_chars else "scanned" + + +def classify_pages(pdf_bytes: bytes, min_chars: int = 20) -> List[str]: + """返回 PDF 每页类型列表 ['native'|'scanned', ...]。""" + out = [] + with fitz.open(stream=pdf_bytes, filetype="pdf") as doc: + for page in doc: + out.append(classify_page(page, min_chars)) + return out diff --git a/tmkit/tests/__init__.py b/tmkit/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tmkit/tests/conftest.py b/tmkit/tests/conftest.py new file mode 100644 index 0000000..ef71b63 --- /dev/null +++ b/tmkit/tests/conftest.py @@ -0,0 +1,79 @@ +"""pytest fixtures：用 reportlab 生成 3 个样例 PDF（中文字体 STSong-Light）。""" + +import io + +import pytest +from reportlab.lib.pagesizes import A4 +from reportlab.pdfbase import pdfmetrics +from reportlab.pdfbase.cidfonts import UnicodeCIDFont +from reportlab.pdfgen import canvas + +pdfmetrics.registerFont(UnicodeCIDFont("STSong-Light")) + + +def _idcard_fix(prefix17: str) -> str: + w = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2] + c = "10X98765432" + total = sum(int(prefix17[i]) * w[i] for i in range(17)) + return prefix17 + c[total % 11] + + +def _luhn_fix(prefix: str) -> str: + for last in "0123456789": + s = prefix + last + total, alt = 0, False + for ch in reversed(s): + d = int(ch) + if alt: + d *= 2 + if d > 9: + d -= 9 + total += d + alt = not alt + if total % 10 == 0: + return s + raise RuntimeError("no luhn digit") + + +VALID_ID = _idcard_fix("11010119900307759") # 程序化补校验位，保证合法 +BANK_CARD = _luhn_fix("622202020011223344") # 19 位，Luhn 合法 +PHONE = "13812345678" +EMAIL = "zhang.san@example.com" + + +def _make_pdf(pages): + """pages: List[str]，每页一段文本；空字符串 = 空白页（模拟扫描页）。""" + buf = io.BytesIO() + c = canvas.Canvas(buf, pagesize=A4) + for text in pages: + if text: + y = 750 + for line in text.split("\n"): + c.setFont("STSong-Light", 12) + c.drawString(72, y, line) + y -= 24 + c.showPage() + c.save() + return buf.getvalue() + + +@pytest.fixture(scope="module") +def pdf_plain(): + """样例1：纯文本页（无 PII）。""" + return _make_pdf(["本判决书仅供学习参考。\n案号：（2026）京01民初1号。\n审判员：李四。"]) + + +@pytest.fixture(scope="module") +def pdf_pii(): + """样例2：含身份证 + 手机号 + 银行卡 + 邮箱。""" + return _make_pdf([ + f"原告：张三，身份证号 {VALID_ID}，联系电话 {PHONE}。\n" + f"收款账户：{BANK_CARD}，邮箱 {EMAIL}。\n" + "以上信息用于测试脱敏。" + ]) + + +@pytest.fixture(scope="module") +def pdf_scanned(): + """样例3：空白扫描页（无文字层）。""" + return _make_pdf([""]) diff --git a/tmkit/tests/test_pipeline.py b/tmkit/tests/test_pipeline.py new file mode 100644 index 0000000..e2b8cbb --- /dev/null +++ b/tmkit/tests/test_pipeline.py @@ -0,0 +1,116 @@ +"""tmkit 全链路测试。""" + +import fitz +import pytest + +import sys, os +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from tmkit.splitter import classify_pages +from tmkit.extractor import extract_page_words, extract_pdf_words +from tmkit.detector import PIIDetector, Rule, default_detector, _luhn_ok, _idcard_ok +from tmkit.redactor import redact_pdf +from .conftest import VALID_ID, BANK_CARD, PHONE, EMAIL + + +def _scan(pdf_bytes): + det = default_detector() + hits = [] + with fitz.open(stream=pdf_bytes, filetype="pdf") as doc: + for pno, page in enumerate(doc): + words = extract_page_words(page) + hits += det.scan_page_words(words, pno) + return hits + + +# ---------- splitter ---------- + +def test_splitter_native_vs_scanned(pdf_plain, pdf_scanned): + assert classify_pages(pdf_plain) == ["native"] + assert classify_pages(pdf_scanned) == ["scanned"] + + +# ---------- extractor ---------- + +def test_extractor_words_with_bbox(pdf_plain): + with fitz.open(stream=pdf_plain, filetype="pdf") as doc: + words = extract_page_words(doc[0]) + assert words and all("text" in w and len(w["bbox"]) == 4 for w in words) + joined = "".join(w["text"] for w in words) + assert "判决书" in joined + + +def test_extractor_scanned_raises(pdf_scanned): + with pytest.raises(NotImplementedError): + extract_pdf_words(pdf_scanned) + + +# ---------- detector 校验算法 ---------- + +def test_idcard_validator(): + assert _idcard_ok(VALID_ID) + assert not _idcard_ok(VALID_ID[:-1] + ("0" if VALID_ID[-1] != "0" else "1")) + + +def test_luhn_validator(): + assert _luhn_ok(BANK_CARD) + assert not _luhn_ok(BANK_CARD[:-1] + ("0" if BANK_CARD[-1] != "0" else "1")) + + +# ---------- detector 全链路（文本 + 坐标） ---------- + +def test_detector_types_found(pdf_pii): + hits = _scan(pdf_pii) + types = {h["type"] for h in hits} + assert types == {"id_card", "phone", "bank_card", "email"} + texts = " ".join(h["text"] for h in hits) + assert VALID_ID in texts and PHONE in texts and BANK_CARD in texts and EMAIL in texts + for h in hits: + assert set(h) == {"type", "text", "page", "bbox"} + assert h["page"] == 0 and h["bbox"][2] > h["bbox"][0] + + +def test_detector_no_false_positive_on_plain(pdf_plain): + assert _scan(pdf_plain) == [] + + +def test_detector_custom_rule(): + det = PIIDetector(rules=[]) + det.register(Rule("order_no", __import__("re").compile(r"（2026）京01民初1号"))) + hits = det.scan_text("案号：（2026）京01民初1号。") + assert hits == [{"type": "order_no", "text": "（2026）京01民初1号"}] + + +# ---------- redactor 全链路 ---------- + +@pytest.mark.parametrize("method", ["redact", "draw"]) +def test_redact_pipeline(pdf_pii, method): + hits = _scan(pdf_pii) + out = redact_pdf(pdf_pii, hits, method=method) + with fitz.open(stream=out, filetype="pdf") as doc: + assert doc.page_count == 1 # 页数不变 + text = doc[0].get_text() + if method == "redact": + # PII 彻底删除 + assert VALID_ID not in text and PHONE not in text + assert BANK_CARD not in text and EMAIL not in text + # 非敏感文本保留 + assert "张三" in text and "脱敏" in text + else: + # draw 模式只画黑条，文本层仍在（已知限制，测试如实断言） + assert VALID_ID in text + + +def test_redact_idempotent_on_clean_pdf(pdf_plain): + out = redact_pdf(pdf_plain, []) + with fitz.open(stream=out, filetype="pdf") as doc: + assert "判决书" in doc[0].get_text() + + +def test_redact_black_box_visible(pdf_pii): + hits = _scan(pdf_pii) + out = redact_pdf(pdf_pii, hits, method="redact") + # apply_redactions(fill) 后区域为黑：检查像素 + with fitz.open(stream=out, filetype="pdf") as doc: + pix = doc[0].get_pixmap(clip=fitz.Rect(*hits[0]["bbox"])) + assert pix.pixel(2, 2) == (0, 0, 0)