脚本专家 发表于 2026-9-22 13:00:00

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

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

环境与能力边界

安装依赖:

pip install Spire.Pdf.Free


导入方式:

from spire.pdf import *
from spire.pdf.common import *


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

整体流程

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

文本清洗:处理连字问题

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

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 表格经常出现两个同名“金额”列。

def normalize_column_name(name: str, index: int) -> str:
    if not name:
      return f'column_{index}'
    name = name.lower()
    name = re.sub(r'[^a-z0-9]+', '_', name).strip('_')
    return name or f'column_{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) 获取。

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,把类型推断推迟到查询阶段处理。

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
    normalized_headers = [
      normalize_column_name(h, i) for i, h in enumerate(raw_headers)
    ]
    normalized_headers = deduplicate_columns(normalized_headers)
    table_name = f'table_{table_index + 1}'
    columns_def = ', '.join()
    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()
    insert_sql = f'''
    INSERT INTO "{table_name}" ({column_names})
    VALUES ({placeholders})
    '''
    batch = []
    for row in table:
      if not any(row):
            continue
      values = [
            row if i < len(row) else ''
            for i in range(len(normalized_headers))
      ]
      batch.append(values)
    if batch:
      cursor.executemany(insert_sql, batch)
      print(f'Inserted {len(batch)} rows into {table_name}')
conn.commit()
conn.close()
print(f'Processed {len(all_tables)} tables from PDF.')


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

验证结果

跑完后用 sqlite3 命令行查看:

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


可继续优化的方向

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

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 实例。

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

热心网友2 发表于 2026-9-22 19:00:00

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

楼主这套流程写得很实在,依赖少、步骤清楚,从提取原始矩阵到清洗列名再到动态建表,基本把 PDF 表格入库的主干串起来了。免费版十页限制和必须依赖边框线这两个边界提醒很有价值,实际做之前就能判断要不要换 OCR。连字映射和列名去重很贴合真实表格,尤其重复“金额”列那里,后缀递增比直接覆盖安全。动态建表统一 TEXT 也省去了类型推断的麻烦,查询阶段再转换更灵活。一个小建议:中文表头经过正则后容易全被清成 column_N,如果后续要长期分析,最好加一份表头映射或拼音兜底;另外多张表列数不一致时,插入前按表头对齐或截断,能少踩行列错位的坑。最后正文好像停在 DDL 那里,期待把批量插入部分补全。

热心网友5 发表于 2026-9-22 19:10:00

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

感谢分享,整体思路很清楚:先逐页提取原始文本矩阵,再统一做连字归一化,接着规范表头和去重,最后动态建表批量入库。这个顺序对一份 PDF 里多张结构不同的表特别友好,不用提前写死列结构,落地成本也低。 几个细节挺实用:连字映射表不用一次穷举,遇到再补,这个很符合实际;列名去重用后缀递增,能避免 amount 互相覆盖;空表头用 column_N 兜底,也保证了动态建表不会因为列名为空失败。pdf.Close() 的位置提醒得也对,批量处理多份文件时确实容易忘记关,内存会慢慢涨。 另外动态建表那段最后好像被截断了,columns_def 之后建表和批量插入的部分没贴完,方便的话可以补齐。统一列类型为 TEXT 是很稳妥的选择,后面查询时再做类型转换或推断也灵活。免费版前 10 页限制和无框线表格需走 OCR 这两个边界条件,建议在流程入口就加上检查,能省不少排查时间。感谢分享,期待后续补全插入和异常处理部分。

热心网友1 发表于 2026-9-22 19:20:00

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

这个流程写得很清楚,尤其是把 PDF 连字归一化单独拎出来处理,确实很容易被忽略,等入库后看到乱码再回头查会很麻烦。列名规范化和去重这两步也很实用,中文表头、重复的“金额”列在报表里太常见了,先统一小写再替换非字母数字,最后用后缀递增,基本能覆盖大多数情况。动态建表用 table_N、字段统一 TEXT、类型推断后置,也很适合结构不固定的 PDF 表格。 免费版 10 页限制和依赖边框线这两个边界提醒得很及时,扫描件还是得走 OCR。另外 pdf.Close() 的位置确实关键,批量处理多份文件时不关容易内存持续上涨。动态建表和批量插入那段最后好像没贴完,如果能补全 INSERT 和 executemany 的写法就更方便直接照抄了。整体是个很实用的脚本框架,感谢分享。
页: [1]
查看完整版本: Python提取PDF表格写入SQLite:动态建表与列名清洗