☰
文档OCR全链路:从扫描件预处理到字段抽取与批量续跑
2026/10/24 5:43:35 网站建设 项目流程

简介:这份面向文档数字化与信息提取场景的OCR实践资料包,适合希望快速了解图片文字识别流程的开发者和文档处理人员,重点解决纸质材料、票据图片等转为可编辑文本时的入门与验证问题。压缩包共3个文件,以2个Python脚本和1张JPG示例图片为主,整体约2.03MB;脚本用于演示OCR识别的基本调用与测试流程,示例图片可作为输入样本,方便对照识别效果。已有192人学习下载,可见其入门参考价值。读者可通过脚本了解如何借助开源工具或在线服务完成文字提取,并结合示例图验证识别结果,进而理解图像质量、字体样式等因素对准确率的影响。资料体量轻便,适合作为OCR功能预研、课程示例或文档处理流程原型的起步材料,也便于在此基础上替换图片和调整脚本,延伸到票据、档案等具体业务场景。

1. 一份歪了 3 度的扫描合同,让我重写了整条 OCR 链路

一份 60 页的扫描合同,手工敲成可检索文本要大半天,OCR 跑一遍只要两三分钟。可这两三里,真正花在「识别」上的不到两成,剩下的全在处理歪斜、阴影、印章遮挡和断行。文档 OCR 的真实难点从来不是调一个接口,而是把像素稳定地变成能入库的字段。这篇笔记拆的是一套可复现的文档 OCR 处理链路:开源引擎负责把图像变成字符,预处理和后处理负责把字符变成可信数据。适合正在做票据、合同、档案数字化的开发者,也适合手上压着几千份 PDF 想批量转文本的同学。

2. 先把引擎选对:本地开源、云端接口与混合兜底的取舍

选型这一步最容易犯的错,是拿几张样张比谁识别得准。同一批档案分别用本地引擎和在线接口跑,整页准确率的差距往往不到 4 个百分点,真正把方案筛掉的,是离线能不能跑、数据能不能出域、单页成本随量怎么变这三条硬约束。先把自己的边界条件写下来,再去看引擎,顺序反了后面全是返工。

2.1 三条路线的差异不在识别率,而在边界条件

常见的落地路线就三条:纯本地开源引擎、纯在线接口、本地优先加在线兜底。它们的差别集中在成本曲线、可控性和运维负担上,识别率反而是最容易追平的一项。

维度本地开源引擎在线 OCR 接口混合(本地优先 + 兜底)
单页成本机器折旧加电费,量越大越便宜按调用次数计费,量大需谈接近本地,兜底页单独付费
离线可用完全可用断网即停绝大多数页面可用
数据出域不出出只出判定为疑难的那部分
竖排、手写、表格弱到中等,看引擎普遍更强取决于兜底比例
首字延迟100 毫秒到 2 秒300 毫秒到 1.5 秒,含上传混合
结果稳定性换版本结果会漂供应商升级会漂两边都会漂
运维负担高,装环境、调参、扩机器低,只管超时和重试中

我一般的判断顺序是:先问数据能不能发出去,不能就直接砍掉在线方案;再问日处理量级,每天不到几百页的,本地引擎单机足够,没必要上分布式;最后才看疑难页占比,如果本地引擎在样张上整页准确率已经过 90%,混合方案里兜底那部分通常只占 5% 到 15%,成本可控。

还有一个常被忽略的点:在线接口的计费单位经常是「次」而不是「页」。一份 PDF 拆成 20 页就是 20 次调用,而双面扫描合并成一张图的情况又会被算作 1 次。做预算时按最坏情况估,别按理想页数估,否则月底账单会给你上一课。

2.2 用适配器把引擎隔离开,换供应商不重写业务

真正让我下决心抽一层适配器的,是某次换引擎:业务代码里散着十几处调用,字段名各不相同,改了两天才改完,还漏了一处。后来我把引擎调用统一收成一个插件式接口,上层的预处理、行合并、字段抽取全部只认一个Line结构,换引擎只改一行配置。

# ocr_adapter.py —— 引擎插件层,上层只认 Line import base64 import json import shlex import subprocess import urllib.request from dataclasses import dataclass from pathlib import Path from typing import Callable @dataclass class Line: text: str box: tuple[int, int, int, int] # x0, y0, x1, y1,左上角为原点 conf: float = 0.0 class Backend: name = "base" def recognize(self, image: Path, lang: str = "chi_sim") -> list[Line]: raise NotImplementedError class CliBackend(Backend): """任何带命令行入口的本地引擎都能挂进来,cmd 里用 {img} {lang} 占位""" def __init__(self, name: str, cmd: str, parser: Callable[[str], list[Line]], timeout: int = 180): self.name, self.cmd, self.parser = name, cmd, parser self.timeout = timeout def recognize(self, image: Path, lang: str = "chi_sim") -> list[Line]: argv = shlex.split(self.cmd.format(img=str(image), lang=lang)) r = subprocess.run(argv, capture_output=True, text=True, timeout=self.timeout) if r.returncode != 0: raise RuntimeError(f"{self.name} 退出码 {r.returncode}: {r.stderr[:200]}") return self.parser(r.stdout) class HttpBackend(Backend): """在线接口统一走 HTTP,超时和重试必须自己兜住""" name = "cloud" def __init__(self, endpoint: str, token: str, timeout: int = 30): self.endpoint, self.token, self.timeout = endpoint, token, timeout def recognize(self, image: Path, lang: str = "chi_sim") -> list[Line]: body = json.dumps({ "image": base64.b64encode(image.read_bytes()).decode(), "lang": lang, }).encode() req = urllib.request.Request(self.endpoint, data=body, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {self.token}", }) with urllib.request.urlopen(req, timeout=self.timeout) as resp: data = json.loads(resp.read()) return [Line(t["text"], tuple(t["box"]), t.get("conf", 0.0)) for t in data["lines"]]

这段代码的关键设计点有三个。第一,Line用 dataclass 而不是裸 dict,因为后面的行合并要按box排序和比较,裸 dict 里一个键名拼错就能让你查半小时。第二,CliBackend的cmd是模板字符串,用shlex.split切开而不是拼一个长 shell 字符串,路径里有空格或中文时不会炸。第三,timeout必须显式给:本地引擎遇到异常大的图可能卡死几分钟,在线接口最常见的失败也不是报错而是长时间不返回。

parser的责任边界要划清楚:它只做「引擎原始输出到list[Line]」的转换,不做任何业务判断。别在 parser 里顺手过滤掉低置信度行,那是后处理的事,parser 一旦掺了业务逻辑,你就得为每个业务场景写一个 parser。

2.3 最小闭环:一张图跑到一份 JSON

在铺开整条流水线之前,先让一张图跑通并留下原始结果。这一步别急着接数据库、别急着做字段抽取,只验证两件事:引擎能不能被你的适配器调起来,坐标是不是你期望的那套坐标系。

# run_one.py —— 最小闭环:一张图 -> txt + json import json import sys from pathlib import Path from ocr_adapter import CliBackend, Line _BACKENDS = {} def register_backend(name, instance): _BACKENDS[name] = instance def get_backend(name): if name not in _BACKENDS: raise KeyError(f"未注册的引擎 {name},可选:{list(_BACKENDS)}") return _BACKENDS[name] def parse_tsv(stdout: str) -> list[Line]: """把引擎的 TSV 输出转成 Line,列号以你本地引擎的实际格式为准""" out = [] for row in stdout.splitlines()[1:]: # 跳过表头 cols = row.split("\t") if len(cols) < 12 or not cols[11].strip(): continue x, y, w, h = (int(cols[i]) for i in range(6, 10)) out.append(Line(cols[11], (x, y, x + w, y + h), float(cols[10]))) return out register_backend("local", CliBackend( name="local", cmd="your-ocr-cli -l {lang} --tsv {img}", # 换成你机器上引擎的真实命令 parser=parse_tsv, )) def run(image: Path, engine: str = "local") -> dict: lines = get_backend(engine).recognize(image, lang="chi_sim") image.with_suffix(".txt").write_text( "\n".join(l.text for l in lines), encoding="utf-8") return { "file": image.name, "engine": engine, "lines": len(lines), "raw": [{"t": l.text, "box": l.box, "c": round(l.conf, 3)} for l in lines], } if __name__ == "__main__": print(json.dumps(run(Path(sys.argv[1])), ensure_ascii=False, indent=2))

parse_tsv里的列号(6 到 10 是左、上、宽、高,11 是文本,10 是置信度)是最容易对错的地方,不同引擎的 TSV 列序不一样,跑之前先打一行原始输出看一眼。lang传chi_sim是简体中文的常见代号,中英混排的文档别只给中文,多数引擎会因此把英文字母识别得很差。

输出里一定要保留raw——每个文本行的文本、坐标、置信度都留着。它看着占空间,但这是你唯一的后悔药:后面字段抽取规则写错了,还能拿 raw 重新跑一遍后处理,不用重新识别几万页。

3. 预处理决定上限:倾斜校正、光照归一与二值化参数

识别引擎的输入说到底是一张图,图的几何和灰度被破坏到什么程度,后处理再聪明也救不回来。这一章的三步是最小可用的组合,做完就能把大部分手机拍照件和低质量扫描件拉到能用水平,剩下的交给识别和校验。

3.1 识别率的天花板是预处理定死的

算一笔账就明白了。300 dpi 的 A4 幅面宽约 2480 像素,正文 10.5 磅字在这一分辨率下字高约 44 像素。整页歪 3 度,行首到行尾的水平偏移约 2480 × sin3° ≈ 130 像素——相当于三个字高。这意味着倾斜会让一行的尾部滑进上一行或下一行,行合并阶段就开始串,后面的字段抽取全乱。

二值化同理。全局阈值对均匀扫描件够用,但手机拍照件几乎必然带阴影:页面一侧亮、一侧暗,单一阈值在暗区把笔画吃掉,在亮区把纸纹留成噪点。笔画断一个像素,粗体字还能勉强认出来,宋体的细横就没了。所以常见做法是先把光照拉平,再在局部窗口内做自适应二值化。

还有一个不显眼但影响很大的因素:分辨率。扫描低于 200 dpi 时,小号汉字只有三四十像素高,紧张;超过 400 dpi 后准确率提升非常有限,耗时却线性上涨。300 dpi 是本地引擎的甜点区,差不多够用又不至于慢得难受。

3.2 三步预处理的可抄代码

下面三个函数对应三步:去倾斜、拉平光照、局部二值化。建议按顺序调用,二值化放最后,因为去倾斜和光照归一的算法都依赖灰度连续性。

# preprocess.py import cv2 import numpy as np def deskew(bgr: np.ndarray, max_angle: float = 15.0) -> tuple[np.ndarray, float]: """用前景最小外接矩形的角度估计倾斜,超过阈值的视为误判不转""" gray = cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY) bw = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY_INV | cv2.THRESH_OTSU)[1] coords = cv2.findNonZero(bw) if coords is None: return bgr, 0.0 angle = cv2.minAreaRect(coords)[-1] # OpenCV 4.5 之后返回 [0, 90),之前返回 [-90, 0),两套约定都要照顾 if angle < -45: angle = -(90 + angle) else: angle = -angle if abs(angle) > max_angle: return bgr, 0.0 h, w = bgr.shape[:2] m = cv2.getRotationMatrix2D((w / 2, h / 2), angle, 1.0) rotated = cv2.warpAffine(bgr, m, (w, h), flags=cv2.INTER_CUBIC, borderMode=cv2.BORDER_REPLICATE) return rotated, angle def normalize_lighting(bgr: np.ndarray, ksize: int = 31) -> np.ndarray: """在 LAB 的亮度通道上估计背景并做除法,抹掉阴影和渐晕""" lab = cv2.cvtColor(bgr, cv2.COLOR_BGR2LAB) l, a, b = cv2.split(lab) kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (ksize, ksize)) background = cv2.morphologyEx(l, cv2.MORPH_CLOSE, kernel) background = cv2.GaussianBlur(background, (0, 0), 15) flat = cv2.divide(l, background, scale=255) # 等价于 l / background * 255 return cv2.cvtColor(cv2.merge([flat, a, b]), cv2.COLOR_LAB2BGR) def binarize(bgr: np.ndarray, block: int = 35, c: int = 15) -> np.ndarray: """局部自适应二值化,block 必须是奇数""" gray = cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY) return cv2.adaptiveThreshold( gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, block, c) def preprocess(path, out_stem): img = cv2.imread(str(path), cv2.IMREAD_COLOR) img, angle = deskew(img) img = normalize_lighting(img, ksize=31) bw = binarize(img, block=35, c=15) cv2.imwrite(f"{out_stem}_norm.png", img) cv2.imwrite(f"{out_stem}_bin.png", bw) return img, bw, angle

deskew里最坑的是角度约定:minAreaRect在新旧 OpenCV 版本里返回范围不同,同一份代码换个环境就可能把图转 90 度。稳妥的做法是跑完把中间图存下来,肉眼看一眼,别信代码。max_angle设 15 度是经验值——真实文档的倾斜很少超过 10 度,超过这个数多半是版面里有大片图形或表格边框把外接矩形带偏了,这时候宁可不转。

normalize_lighting里的ksize要跟着分辨率走:300 dpi 的 A4 用 31 到 51 比较合适,太小会连笔画一起当成背景抹掉,太大则阴影去不干净。binarize的block取 31 到 41 之间的奇数,c取 10 到 20;c越大越「保守」,保留的笔画多但噪点也多,做票据这种需要保住细线的场景可以调低到 10 左右。

提示:二值化后的图不要直接喂给识别引擎。多数现代引擎内部会自己做二值化,外部再传一张已经二值化的图,等于做了两次,反而掉点。把灰度图或光照归一后的彩色图交给引擎,二值化结果只用来做版面分析和调参对照。

3.3 三组对照实验:参数怎么落

参数不能凭感觉设,最省事的办法是固定一批 50 到 100 页有代表性的样张,一次只动一个变量,记录整页准确率和单页耗时。下表是这类对照实验里常见的趋势形态,数字只代表量级和方向,别当成基准值照抄。

组输入分辨率光照归一二值化整页准确率趋势单页耗时趋势
A200 dpi否全局 Otsu明显偏低,细横丢失最快
B300 dpi否自适应 35 / 15中等,阴影区断笔中
C300 dpi是自适应 35 / 15较高,推荐起点中
D400 dpi是自适应 41 / 12与 C 接近,提升有限明显变慢,约 1.5 到 2 倍

从这张表能看出两个结论。一是分辨率从 200 提到 300 的收益远大于从 300 提到 400,后者往往用接近两倍的耗时换不到几个点。二是光照归一的收益在手机拍照件上非常明显,在均匀扫描件上几乎为零——所以它应该是个按来源开关的步骤,而不是无脑加进流水线。

实际调参时我一般这样排优先级:先把分辨率统一到 300 dpi,再开光照归一,然后微调block和c,最后才考虑去噪、锐化这类收益更小的操作。每次只动一个参数,改完立刻用同一批样张回归一遍,别攒着一起改——一起改就不知道是哪个起的作用,这种玄学调参最后会把你逼疯。

4. 后处理与结构化:把散装文本行拼成能入库的字段

引擎吐出来的是几百个带坐标的文本片段,顺序、断行、空格全是乱的,直接拼接出来的东西人能看懂,程序没法用。这一章解决三件事:把碎片还原成正确的阅读顺序、用模板和校验把字段抽出来、让批量任务能断点续跑。

4.1 行合并与阅读顺序还原

多数引擎按检测框输出,一个长句可能被切成三段,页眉和正文的第一行也可能因为坐标接近被并到同一行。合并的核心是按纵坐标聚类成行、行内按横坐标排序、再按间距决定要不要补空格。

# layout.py from ocr_adapter import Line def group_lines(lines: list[Line], y_tol: float = 0.55, x_gap: float = 1.8) -> list[str]: """y_tol 控制纵向合并的松紧,x_gap 控制行内补空格的阈值(单位:字高)""" rows: list[dict] = [] for ln in sorted(lines, key=lambda v: (v.box[1], v.box[0])): h = ln.box[3] - ln.box[1] for row in rows: rh = row["y1"] - row["y0"] cy_new = (ln.box[1] + ln.box[3]) / 2 cy_row = (row["y0"] + row["y1"]) / 2 if abs(cy_new - cy_row) < y_tol * max(h, rh): row["items"].append(ln) row["y0"] = min(row["y0"], ln.box[1]) row["y1"] = max(row["y1"], ln.box[3]) break else: rows.append({"y0": ln.box[1], "y1": ln.box[3], "items": [ln]}) rows.sort(key=lambda r: r["y0"]) out = [] for row in rows: items = sorted(row["items"], key=lambda v: v.box[0]) buf = items[0].text for prev, cur in zip(items, items[1:]): gap = cur.box[0] - prev.box[2] avg_h = ((prev.box[3] - prev.box[1]) + (cur.box[3] - cur.box[1])) / 2 buf += (" " if gap > x_gap * avg_h else "") + cur.text out.append(buf) return out

y_tol是最需要按文档类型调的参数。0.55 意味着两个框的中心纵向距离小于自身高度的 0.55 倍就并成一行,对常规正文合适;遇到行距很紧的表格或者上下标(比如金额里的角分),调到 0.4 更稳。x_gap控制空格:中文文档里字间距本来就小,1.8 倍字高基本只在真正的词间空隙处触发;英文文档要调到 0.3 到 0.5,否则整行会连成一片。

双栏排版是这套逻辑的天敌。两栏的同一行在纵向上几乎重合,会被合并成一行交错文本。处理办法是在合并之前先做一次竖向投影,找中间的空白谷把页面切成左右两栏,分别跑group_lines再拼接。别指望识别引擎帮你分栏,它只认框,不认语义。

4.2 正则模板 + 交叉校验

字段抽取最直接的办法是按文档类型写模板,模板里放正则和校验函数。关键不在于正则写得多花哨,而在于每抽一个字段都配一个能证伪的校验:金额要能和中文大写金额对上,日期要能通过合法性检查,编号要能过校验位。

# extract.py import re from datetime import date from decimal import Decimal MONEY = re.compile(r"[¥¥]\s*([0-9][0-9,]*(?:\.[0-9]{1,2})?)") DATE = re.compile(r"(20\d{2})\s*[-/年.]\s*(\d{1,2})\s*[-/月.]\s*(\d{1,2})") TAXNO = re.compile(r"\b([0-9A-Z]{18})\b") CODE = "0123456789ABCDEFGHJKLMNPQRTUWXY" WEIGHT = [1, 3, 9, 27, 19, 26, 16, 17, 20, 29, 25, 13, 8, 24, 10, 30, 28] def parse_money(text: str) -> Decimal | None: m = MONEY.search(text.replace(" ", "")) return Decimal(m.group(1).replace(",", "")) if m else None def parse_date(text: str) -> date | None: m = DATE.search(text) if not m: return None try: return date(int(m.group(1)), int(m.group(2)), int(m.group(3))) except ValueError: return None def check_taxno(no: str) -> bool: """18 位统一社会信用代码的校验位,专门用来挡 O/0、I/1、S/5 这类混淆""" if len(no) != 18 or any(c not in CODE for c in no): return False s = sum(CODE.index(c) * WEIGHT[i] for i, c in enumerate(no[:17])) return CODE[(31 - s % 31) % 31] == no[17]

check_taxno是这套后处理里性价比最高的一段。识别引擎在无衬线字体上把0认成O、把1认成I是家常便饭,而校验位能把这些错误几乎全部挡下来,代价只是一次除法和查表。同样的思路可以用在身份证号、银行卡号、发票代码上,凡是带校验位的字段都应该走一遍,比调引擎参数省事得多。

parse_date用try/except ValueError而不是自己写月份天数判断,是因为date构造器已经把闰年和大小月处理掉了,自己写十有八九会漏掉 2 月 29 日。parse_money里先replace(" ", "")再匹配,是因为引擎经常在货币符号和数字之间插一个空格或换行。

模板本身要跟文档版本绑定。合同改版、票据换样式之后,旧模板会静默失配——不报错,只是抽出来一堆None。所以每份结果里要带上模板版本号,抽取失败率突然抬头时,第一个要看的就是这个字段。

4.3 批量任务:manifest 与断点续跑

跑几千份文件时,最怕的不是慢,是跑到 80% 崩了要重来。解决办法是给每个文件算一个内容哈希当主键,边跑边往manifest.jsonl追加,重启时先读 manifest 跳过已完成的。这是整条链路里唯一的后悔药,别省。

# batch.py import hashlib import json from concurrent.futures import ProcessPoolExecutor from pathlib import Path SUFFIX = {".png", ".jpg", ".jpeg", ".tif", ".tiff", ".bmp"} def file_id(p: Path) -> str: """按内容而不是文件名做标识,改名、挪目录都不会重复处理""" h = hashlib.sha1() with p.open("rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) return h.hexdigest()[:16] def load_done(manifest: Path) -> set[str]: if not manifest.exists(): return set() return {json.loads(l)["id"] for l in manifest.read_text(encoding="utf-8").splitlines() if l} def process_one(p: Path) -> dict: """单页全流程:预处理 -> 识别 -> 行合并 -> 字段抽取""" return {"id": file_id(p), "file": str(p), "ok": True} def run_batch(root: Path, out: Path, workers: int = 4): out.mkdir(parents=True, exist_ok=True) manifest = out / "manifest.jsonl" done = load_done(manifest) todo = [p for p in sorted(root.rglob("*")) if p.suffix.lower() in SUFFIX and file_id(p) not in done] with ProcessPoolExecutor(max_workers=workers) as ex, \ manifest.open("a", encoding="utf-8") as mf: for rec in ex.map(process_one, todo): mf.write(json.dumps(rec, ensure_ascii=False) + "\n") mf.flush() # 每写一条就落盘,进程被 kill 也不丢进度

workers不要设成 CPU 核数。识别引擎本身经常是多线程的,进程数开满会互相抢核,实测下来 4 到 8 个进程通常是吞吐的拐点,再往上收益递减甚至倒退。给每个 worker 设一个超时,遇到卡死的文件就跳过并记一条失败记录,别让一个坏文件拖垮整批。

mf.flush()看起来很浪费,但它是断点续跑能成立的前提。不加 flush 时数据还在缓冲区里,进程被 kill 或机器断电就全丢了,你会以为跑了 80%,实际 manifest 里只有 20%。这个坑我踩过一次,两千多页白跑。

注意:file_id在筛选和process_one里各算了一次,等于每个文件读两遍。文件量在万级以内影响不大

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询