PDF 转结构化数据转换

概述

基于 DDC 方法论（第 2.4 章），本技能将非结构化的 PDF 文档转换为适合分析与集成的结构化格式。建筑工程会产生大量 PDF 文档——包括技术规范、物料清单（BOM）、进度计划和报告等——这些文档需被提取并处理。

图书参考：《数据转换为结构化形式》

“将数据从非结构化形式转换为结构化形式，既是一门艺术，也是一门科学。该过程常常占据数据工程师工作的大部分时间。”
—《DDC Book》，第 2.4 章

ETL 流程概述

该转换遵循 ETL 模式：

1. 抽取（Extract）：加载 PDF 文档
2. 转换（Transform）：解析并结构化内容
3. 加载（Load）：保存为 CSV、Excel 或 JSON 格式

快速入门

import pdfplumber
import pandas as pd

# 从 PDF 中抽取表格
with pdfplumber.open("construction_spec.pdf") as pdf:
    page = pdf.pages[0]
    table = page.extract_table()
    df = pd.DataFrame(table[1:], columns=table[0])
    df.to_excel("extracted_data.xlsx", index=False)

安装

# 核心库
pip install pdfplumber pandas openpyxl

# 针对扫描版 PDF（OCR）
pip install pytesseract pdf2image
# 同时需安装 Tesseract OCR：https://github.com/tesseract-ocr/tesseract

# 高级 PDF 操作
pip install pypdf

原生 PDF 抽取（pdfplumber）

从 PDF 中抽取全部表格

import pdfplumber
import pandas as pd

def extract_tables_from_pdf(pdf_path):
    """从 PDF 文件中抽取所有表格"""
    all_tables = []

    with pdfplumber.open(pdf_path) as pdf:
        for page_num, page in enumerate(pdf.pages):
            tables = page.extract_tables()
            for table_num, table in enumerate(tables):
                if table and len(table) > 1:
                    # 将首行作为表头
                    df = pd.DataFrame(table[1:], columns=table[0])
                    df['_page'] = page_num + 1
                    df['_table'] = table_num + 1
                    all_tables.append(df)

    if all_tables:
        return pd.concat(all_tables, ignore_index=True)
    return pd.DataFrame()

# 使用示例
df = extract_tables_from_pdf("material_specification.pdf")
df.to_excel("materials.xlsx", index=False)

按布局结构抽取文本

import pdfplumber

def extract_text_with_layout(pdf_path):
    """抽取文本并保留原始布局结构"""
    full_text = []

    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            text = page.extract_text()
            if text:
                full_text.append(text)

    return "\n\n--- Page Break ---\n\n".join(full_text)

# 使用示例
text = extract_text_with_layout("project_report.pdf")
with open("report_text.txt", "w", encoding="utf-8") as f:
    f.write(text)

按坐标区域抽取指定表格

import pdfplumber
import pandas as pd

def extract_table_from_area(pdf_path, page_num, bbox):
    """
    从页面指定区域抽取表格

    参数：
        pdf_path: PDF 文件路径
        page_num: 页面编号（从 0 开始计数）
        bbox: 边界框（x0, top, x1, bottom），单位为点（points；1 英寸 = 72 点）
    """
    with pdfplumber.open(pdf_path) as pdf:
        page = pdf.pages[page_num]
        cropped = page.within_bbox(bbox)
        table = cropped.extract_table()

        if table:
            return pd.DataFrame(table[1:], columns=table[0])
    return pd.DataFrame()

# 使用示例 —— 从指定区域抽取表格
# bbox 格式：(left, top, right, bottom)，单位为点（points）
df = extract_table_from_area("drawing.pdf", 0, (50, 100, 550, 400))

扫描版 PDF 处理（OCR）

从扫描版 PDF 中抽取文本

import pytesseract
from pdf2image import convert_from_path
import pandas as pd

def ocr_scanned_pdf(pdf_path, language='eng'):
    """
    使用 OCR 从扫描版 PDF 中抽取文本

    参数：
        pdf_path: 扫描版 PDF 文件路径
        language: Tesseract 语言代码（如 eng、deu、rus 等）
    """
    # 将 PDF 页面转换为图像
    images = convert_from_path(pdf_path, dpi=300)

    extracted_text = []
    for i, image in enumerate(images):
        text = pytesseract.image_to_string(image, lang=language)
        extracted_text.append({
            'page': i + 1,
            'text': text
        })

    return pd.DataFrame(extracted_text)

# 使用示例
df = ocr_scanned_pdf("scanned_specification.pdf", language='eng')
df.to_csv("ocr_results.csv", index=False)

OCR 表格抽取

import pytesseract
from pdf2image import convert_from_path
import pandas as pd
import cv2
import numpy as np

def ocr_table_from_scanned_pdf(pdf_path, page_num=0):
    """使用 OCR 并结合表格检测，从扫描版 PDF 中抽取表格"""
    # 将指定页面转换为图像
    images = convert_from_path(pdf_path, first_page=page_num+1,
                                last_page=page_num+1, dpi=300)
    image = np.array(images[0])

    # 转为灰度图
    gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)

    # 应用二值化阈值处理
    _, binary = cv2.threshold(gray, 150, 255, cv2.THRESH_BINARY_INV)

    # 使用自定义配置抽取带表格结构的文本
    custom_config = r'--oem 3 --psm 6'
    text = pytesseract.image_to_string(gray, config=custom_config)

    # 将文本解析为表格结构
    lines = text.strip().split('\n')
    data = [line.split() for line in lines if line.strip()]

    if data:
        # 假设首行为表头
        df = pd.DataFrame(data[1:], columns=data[0] if len(data[0]) > 0 else None)
        return df
    return pd.DataFrame()

# 使用示例
df = ocr_table_from_scanned_pdf("scanned_bom.pdf")
print(df)

面向建筑工程的专用抽取

物料清单（BOM）抽取

import pdfplumber
import pandas as pd
import re

def extract_bom_from_pdf(pdf_path):
    """从建筑工程 PDF 中抽取物料清单（BOM）"""
    all_items = []

    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            tables = page.extract_tables()
            for table in tables:
                if not table or len(table) < 2:
                    continue

                # 查找表头行（匹配常见 BOM 字段关键词）
                header_keywords = ['item', 'description', 'quantity', 'unit', 'material']
                for i, row in enumerate(table):
                    if row and any(keyword in str(row).lower() for keyword in header_keywords):
                        # 找到表头，处理后续数据行
                        headers = [str(h).strip() for h in row]
                        for data_row in table[i+1:]:
                            if data_row and any(cell for cell in data_row if cell):
                                item = dict(zip(headers, data_row))
                                all_items.append(item)
                        break

    return pd.DataFrame(all_items)

# 使用示例
bom = extract_bom_from_pdf("project_bom.pdf")
bom.to_excel("bom_extracted.xlsx", index=False)

项目进度计划抽取

import pdfplumber
import pandas as pd
from datetime import datetime

def extract_schedule_from_pdf(pdf_path):
    """从 PDF 中抽取项目进度计划 / 甘特图数据"""
    with pdfplumber.open(pdf_path) as pdf:
        all_tasks = []

        for page in pdf.pages:
            tables = page.extract_tables()
            for table in tables:
                if not table:
                    continue

                # 寻找类似进度计划的表格
                headers = table[0] if table else []

                # 判断是否为进度计划类表格
                schedule_keywords = ['task', 'activity', 'start', 'end', 'duration']
                if any(kw in str(headers).lower() for kw in schedule_keywords):
                    for row in table[1:]:
                        if row and any(cell for cell in row if cell):
                            task = dict(zip(headers, row))
                            all_tasks.append(task)

    df = pd.DataFrame(all_tasks)

    # 尝试解析日期字段
    date_columns = ['Start', 'End', 'Start Date', 'End Date', 'Finish']
    for col in date_columns:
        if col in df.columns:
            df[col] = pd.to_datetime(df[col], errors='coerce')

    return df

# 使用示例
schedule = extract_schedule_from_pdf("project_schedule.pdf")
print(schedule)

技术规范解析

import pdfplumber
import pandas as pd
import re

def parse_specification_pdf(pdf_path):
    """解析建筑工程技术规范文档"""
    specs = []

    with pdfplumber.open(pdf_path) as pdf:
        full_text = ""
        for page in pdf.pages:
            text = page.extract_text()
            if text:
                full_text += text + "\n"

    # 解析章节（常见规范格式）
    section_pattern = r'(\d+\.\d+(?:\.\d+)?)\s+([A-Z][^\n]+)'
    sections = re.findall(section_pattern, full_text)

    for num, title in sections:
        specs.append({
            'section_number': num,
            'title': title.strip(),
            'level': len(num.split('.'))
        })

    return pd.DataFrame(specs)

# 使用示例
specs = parse_specification_pdf("technical_spec.pdf")
print(specs)

批量处理

批量处理多个 PDF 文件

import pdfplumber
import pandas as pd
from pathlib import Path

def batch_extract_tables(folder_path, output_folder):
    """批量处理文件夹内所有 PDF，并抽取其中的表格"""
    pdf_files = Path(folder_path).glob("*.pdf")
    results = []

    for pdf_path in pdf_files:
        print(f"Processing: {pdf_path.name}")
        try:
            with pdfplumber.open(pdf_path) as pdf:
                for page_num, page in enumerate(pdf.pages):
                    tables = page.extract_tables()
                    for table_num, table in enumerate(tables):
                        if table and len(table) > 1:
                            df = pd.DataFrame(table[1:], columns=table[0])
                            df['_source_file'] = pdf_path.name
                            df['_page'] = page_num + 1

                            # 保存单个表格
                            output_name = f"{pdf_path.stem}_p{page_num+1}_t{table_num+1}.xlsx"
                            df.to_excel(Path(output_folder) / output_name, index=False)
                            results.append(df)
        except Exception as e:
            print(f"Error processing {pdf_path.name}: {e}")

    # 合并输出
    if results:
        combined = pd.concat(results, ignore_index=True)
        combined.to_excel(Path(output_folder) / "all_tables.xlsx", index=False)

    return len(results)

# 使用示例
count = batch_extract_tables("./pdf_documents/", "./extracted/")
print(f"Extracted {count} tables")

抽取后数据清洗

import pandas as pd

def clean_extracted_data(df):
    """清洗 PDF 抽取数据中的常见问题"""
    # 删除完全为空的行
    df = df.dropna(how='all')

    # 清理字符串列首尾空白字符
    for col in df.select_dtypes(include=['object']).columns:
        df[col] = df[col].str.strip()

    # 删除所有单元格均为空字符串的行
    df = df[df.apply(lambda row: any(cell != '' for cell in row), axis=1)]

    # 尝试转换数值列
    for col in df.columns:
        # 尝试转为数值类型
        numeric_series = pd.to_numeric(df[col], errors='coerce')
        if numeric_series.notna().sum() > len(df) * 0.5:  # 若超过 50% 为有效数值，则执行转换
            df[col] = numeric_series

    return df

# 使用示例
df = extract_tables_from_pdf("document.pdf")
df_clean = clean_extracted_data(df)
df_clean.to_excel("clean_data.xlsx", index=False)

导出选项

import pandas as pd
import json

def export_to_multiple_formats(df, base_name):
    """将 DataFrame 导出为多种格式"""
    # Excel
    df.to_excel(f"{base_name}.xlsx", index=False)

    # CSV
    df.to_csv(f"{base_name}.csv", index=False, encoding='utf-8-sig')

    # JSON
    df.to_json(f"{base_name}.json", orient='records', indent=2)

    # JSON Lines（适用于大数据集）
    df.to_json(f"{base_name}.jsonl", orient='records', lines=True)

# 使用示例
df = extract_tables_from_pdf("document.pdf")
export_to_multiple_formats(df, "extracted_data")

速查表

故障排查

参考资料

• 图书：Artem Boiko《Data-Driven Construction》，第 2.4 章
• 网站：https://datadrivenconstruction.io
• pdfplumber 文档：https://github.com/jsvine/pdfplumber
• Tesseract OCR：https://github.com/tesseract-ocr/tesseract

下一步

• 参阅 image-to-data 获取图像处理相关内容
• 参阅 cad-to-data 获取 CAD/BIM 数据抽取相关内容
• 参阅 etl-pipeline 获取自动化处理工作流相关内容
• 参阅 data-quality-check 获取抽取数据质量校验相关内容