Python提取PDF表格写入SQLite：动态建表与列名清洗

脚本专家

PDF表格数据入库是常见数据处理任务：季度报表、对账单、业务台账等 PDF 需要结构化后进入数据库分析。本文基于 Python 与 Free Spire.PDF 展示从表格提取、字段清洗、动态建表到批量写入 SQLite 的完整脚本流程。除 PDF 解析库外，其余使用标准库，部署成本低。

环境与能力边界

安装依赖：

2. pip install Spire.Pdf.Free

复制代码

导入方式：

2. from spire.pdf import *
3. from spire.pdf.common import *

复制代码

两个限制需要提前注意：免费版单文件最多处理前 10 页，超出部分不会返回；表格提取依赖 PDF 内的边框线结构，扫描件或无框线表格不适用，需改用 OCR 方案。

整体流程

处理分三步：逐页提取所有表格的原始文本矩阵；对表头做规范化、去重，生成合法列名；每张表动态建表并批量插入。这样一份 PDF 里包含多张结构不同的表格时，不需要提前预知列结构，也能全部落入数据库。

文本清洗：处理连字问题

PDF 中常见连字符被替换成 Unicode 私有区字符，例如 fi、ff 会变成 \ue005、\ue000 之类编码，直接入库会出现乱码方块。第一步需要归一化：

2. def normalize_text(text: str) -> str:
3.     if not text:
4.         return ''
5.     ligature_map = {
6.         '\\ue000': 'ff', '\\ue001': 'ft', '\\ue002': 'ffi',
7.         '\\ue003': 'ffl', '\\ue004': 'ti', '\\ue005': 'fi',
8.     }
9.     for k, v in ligature_map.items():
10.         text = text.replace(k, v)
11.     return text.strip()

复制代码

映射表不需要一次穷举，实践中遇到新字符再补充即可。

列名规范化与去重

原始表头常带空格、大小写混杂，甚至包含中文和符号，不能直接作为数据库列名。这里做两件事：用正则把非字母数字字符统一替换成下划线，空表头用 column_N 兜底；处理重复列名，PDF 表格经常出现两个同名“金额”列。

2. def normalize_column_name(name: str, index: int) -> str:
3.     if not name:
4.         return f'column_{index}'
5.     name = name.lower()
6.     name = re.sub(r'[^a-z0-9]+', '_', name).strip('_')
7.     return name or f'column_{index}'
9. def deduplicate_columns(columns):
10.     seen = set()
11.     result = []
12.     for col in columns:
13.         base = col
14.         count = 1
15.         while col in seen:
16.             col = f'{base}_{count}'
17.             count += 1
18.         seen.add(col)
19.         result.append(col)
20.     return result

复制代码

去重逻辑采用后缀递增方式，生成 amount、amount_1、amount_2，不会互相覆盖。

提取表格

PdfTableExtractor 是核心对象，按页调用 ExtractTable(page_index) 返回该页所有表格对象。每张表通过 GetRowCount() / GetColumnCount() 遍历，单元格文本用 GetText(row, col) 获取。

2. pdf = PdfDocument()
3. pdf.LoadFromFile('Quarterly Sales.pdf')
4. extractor = PdfTableExtractor(pdf)
5. all_tables = []
6. for i in range(pdf.Pages.Count):
7.     tables = extractor.ExtractTable(i)
8.     if not tables:
9.         continue
10.     for table in tables:
11.         table_rows = []
12.         for row in range(table.GetRowCount()):
13.             row_data = [
14.                 normalize_text(table.GetText(row, col))
15.                 for col in range(table.GetColumnCount())
16.             ]
17.             table_rows.append(row_data)
18.         if table_rows:
19.             all_tables.append(table_rows)
20. pdf.Close()
21. if not all_tables:
22.     raise ValueError('No tables found in PDF.')

复制代码

注意 pdf.Close() 的位置。批量处理多份文件时这一步很关键，否则内存占用会持续累积。

动态建表与批量插入

不同表格列结构不同，用 table_N 命名并动态生成 DDL 是最省事的做法。所有列统一声明为 TEXT，把类型推断推迟到查询阶段处理。

2. conn = sqlite3.connect('sales_data.db')
3. cursor = conn.cursor()
4. for table_index, table in enumerate(all_tables):
5.     if len(table) < 2:
6.         continue
7.     raw_headers = table[0]
8.     normalized_headers = [
9.         normalize_column_name(h, i) for i, h in enumerate(raw_headers)
10.     ]
11.     normalized_headers = deduplicate_columns(normalized_headers)
12.     table_name = f'table_{table_index + 1}'
13.     columns_def = ', '.join([f'"{col}" TEXT' for col in normalized_headers])
14.     cursor.execute(f'''
15.     CREATE TABLE IF NOT EXISTS "{table_name}" (
16.         id INTEGER PRIMARY KEY AUTOINCREMENT,
17.         {columns_def}
18.     )
19.     ''')
20.     placeholders = ', '.join(['?' for _ in normalized_headers])
21.     column_names = ', '.join([f'"{col}"' for col in normalized_headers])
22.     insert_sql = f'''
23.     INSERT INTO "{table_name}" ({column_names})
24.     VALUES ({placeholders})
25.     '''
26.     batch = []
27.     for row in table[1:]:
28.         if not any(row):
29.             continue
30.         values = [
31.             row[i] if i < len(row) else ''
32.             for i in range(len(normalized_headers))
33.         ]
34.         batch.append(values)
35.     if batch:
36.         cursor.executemany(insert_sql, batch)
37.         print(f'Inserted {len(batch)} rows into {table_name}')
38. conn.commit()
39. conn.close()
40. print(f'Processed {len(all_tables)} tables from PDF.')

复制代码

几个实现细节：if len(table) < 2 会跳过只有表头或空行的伪表格，避免无意义建表。列数对齐时，当某行列数少于表头，使用 row if i < len(row) else '' 补空字符串；如果某行列数多于表头，多出的部分会被直接丢弃。这在合并单元格场景下会出现，实践中建议加日志记录列数异常的行号，便于人工核对。executemany 批量插入能显著降低 SQLite 事务开销，单条 INSERT 在几百行表上就会明显变慢。

验证结果

跑完后用 sqlite3 命令行查看：

2. sqlite3 sales_data.db ".tables"
3. sqlite3 sales_data.db "SELECT * FROM table_1 LIMIT 5;"

复制代码

可继续优化的方向

类型转换：目前全部以 TEXT 入库。如果后续要按数值排序或聚合，可以在插入前做类型推断：

2. def try_parse(value: str):
3.     try:
4.         return float(value.replace(',', ''))
5.     except (ValueError, AttributeError):
6.         return value

复制代码

把 values 列表在入 batch 前做一次映射，能让纯数字列以 REAL 类型存储，查询时就不需要 CAST。

表名语义化：table_1、table_2 只是最省事方案。如果 PDF 中每张表上方有标题文本，可以提取后作为表名，读起来更直观。

多文件批处理：把主流程封装成 process_pdf(path, conn) 函数，外层套一层文件遍历即可。注意每个文件都要 LoadFromFile → 提取 → Close()，不要复用 PdfDocument 实例。

整套方案除 PDF 解析库外没有第三方依赖，适合嵌入现有数据处理脚本，也方便在容器里运行。如果场景是页数不多、结构相对规范的 PDF 报表，这份代码基本可以直接改几个常量投入使用。