用 Python 把 PDF 表格批量导入 SQLite

处理 PDF 表格数据的场景很常见季度报表、对账单、业务台账业务方给一份 PDF 过来需要结构化后进数据库做后续分析。本文分享一个完整的 Python 实现覆盖从表格提取、字段清洗、动态建表到批量入库的全流程。代码依赖免费库 Free Spire.PDF 完成 PDF 解析其余全部基于标准库部署成本很低。环境与库代码语言Bash自动换行AI代码解释pip install Spire.Pdf.Free导入方式代码语言Python自动换行AI代码解释from spire.pdf import * from spire.pdf.common import *需要提前说明两个能力边界避免后面踩坑免费版单文件最多处理前 10 页超出部分不会返回。表格提取依赖 PDF 内的边框线结构扫描件或无框线表格不适用需走 OCR 方案。整体流程处理流程分三步逐页提取所有表格的原始文本矩阵对表头做规范化、去重生成合法列名每张表动态建表并批量插入这样做的好处是一份 PDF 里包含多张结构不同的表格时不需要提前预知列结构也能全部落入数据库。文本清洗先处理连字问题PDF 里常见的一类坑是连字符ligature被替换成 Unicode 私有区字符例如fi、ff会变成\ue005、\ue000之类的编码。直接入库就是一堆乱码方块所以第一步要做归一化。代码语言Python自动换行AI代码解释def normalize_text(text: str) - str: if not text: return ligature_map { \ue000: ff, \ue001: ft, \ue002: ffi, \ue003: ffl, \ue004: ti, \ue005: fi, } for k, v in ligature_map.items(): text text.replace(k, v) return text.strip()映射表不需要一次穷举实践中遇到新字符再补充即可。列名规范化与去重原始表头往往带空格、大小写混杂、甚至包含中文和符号不能直接作为数据库列名。这里做两件事用正则把非字母数字字符统一替换成下划线空表头用column_N兜底处理重复列名PDF 表格经常出现两个同名金额列代码语言Python自动换行AI代码解释def normalize_column_name(name: str, index: int) - str: if not name: return fcolumn_{index} name name.lower() name re.sub(r[^a-z0-9], _, name).strip(_) return name or fcolumn_{index} def deduplicate_columns(columns): seen set() result [] for col in columns: base col count 1 while col in seen: col f{base}_{count} count 1 seen.add(col) result.append(col) return result去重逻辑采用后缀递增方式amount、amount_1、amount_2不会互相覆盖。提取表格PdfTableExtractor是核心对象按页调用ExtractTable(page_index)返回该页所有表格对象。每张表通过GetRowCount()/GetColumnCount()遍历单元格文本用GetText(row, col)获取。代码语言Python自动换行AI代码解释pdf PdfDocument() pdf.LoadFromFile(Quarterly Sales.pdf) extractor PdfTableExtractor(pdf) all_tables [] for i in range(pdf.Pages.Count): tables extractor.ExtractTable(i) if not tables: continue for table in tables: table_rows [] for row in range(table.GetRowCount()): row_data [ normalize_text(table.GetText(row, col)) for col in range(table.GetColumnCount()) ] table_rows.append(row_data) if table_rows: all_tables.append(table_rows) pdf.Close() if not all_tables: raise ValueError(No tables found in PDF.)注意pdf.Close()的位置——批量处理多份文件时这一步很关键否则内存占用会持续累积。动态建表 批量插入不同表格的列结构不同用table_N命名并动态生成 DDL 是最省事的做法。所有列统一声明为TEXT把类型推断推迟到查询阶段处理。代码语言Python自动换行AI代码解释conn sqlite3.connect(sales_data.db) cursor conn.cursor() for table_index, table in enumerate(all_tables): if len(table) 2: continue # 只有表头没有数据行的表直接跳过 raw_headers table[0] normalized_headers [ normalize_column_name(h, i) for i, h in enumerate(raw_headers) ] normalized_headers deduplicate_columns(normalized_headers) table_name ftable_{table_index 1} columns_def , .join([f{col} TEXT for col in normalized_headers]) cursor.execute(f CREATE TABLE IF NOT EXISTS {table_name} ( id INTEGER PRIMARY KEY AUTOINCREMENT, {columns_def} ) ) placeholders , .join([? for _ in normalized_headers]) column_names , .join([f{col} for col in normalized_headers]) insert_sql f INSERT INTO {table_name} ({column_names}) VALUES ({placeholders}) batch [] for row in table[1:]: if not any(row): continue values [ row[i] if i len(row) else for i in range(len(normalized_headers)) ] batch.append(values) if batch: cursor.executemany(insert_sql, batch) print(fInserted {len(batch)} rows into {table_name}) conn.commit() conn.close() print(fProcessed {len(all_tables)} tables from PDF.)几个实现细节值得展开if len(table) 2跳过空表。PDF 中有时会提取出只有表头、甚至只有一行空字符串的伪表格提前过滤掉能避免无意义建表。列数对齐取最小值。当某行的列数少于表头时row[i] if i len(row) else 补齐空字符串。反过来如果某行列数多于表头多出的部分会被直接丢弃——这在合并单元格场景下会出现实践中建议加一条日志记录列数异常的行号方便后续人工核对。executemany批量插入。单条INSERT在一张几百行的表上就会明显变慢批量执行能显著降低 SQLite 事务开销。验证结果跑完后直接用 sqlite3 命令行看数据代码语言Bash自动换行AI代码解释sqlite3 sales_data.db .tables sqlite3 sales_data.db SELECT * FROM table_1 LIMIT 5;可以继续优化的方向类型转换。目前全部以 TEXT 入库如果后续要按数值排序或聚合可以在插入前做一次类型推断代码语言Python自动换行AI代码解释def try_parse(value: str): try: return float(value.replace(,, )) except (ValueError, AttributeError): return value把values列表在入 batch 前做一次映射能让纯数字列以 REAL 类型存储查询时就不需要CAST了。表名语义化。用table_1、table_2只是最省事的方案。如果 PDF 中每张表上方有标题文本可以提取后作为表名读起来更直观。多文件批处理。把主流程封装成process_pdf(path, conn)函数外层套一层文件遍历即可。注意每个文件都要LoadFromFile→ 提取 →Close()不要复用PdfDocument实例。