第 8 章

Python 处理 PDF——提取、合并、生成一步到位

第8章:PDF 自动化处理——拆分、合并、提取与生成

PDF 是商业世界的标准格式:合同、发票、报告、规范文档,几乎全用 PDF 交付。但 PDF 并非为机器处理而设计,从中提取数据、批量拆分合并、自动生成报告——这些需求倒逼出了一套成熟的 Python 工具链。本章系统梳理 PDF 处理的四大场景:提取、合并拆分、表格抽取、生成,并以财务报告处理管道作为综合实战。

PDF 库生态:选哪个库做什么事

Python 的 PDF 库较多,功能各有侧重,不存在一个"万能库"。正确的做法是按场景选型:

适合场景 安装命令
pdfplumber 文本提取、表格提取(数字PDF) pip install pdfplumber
pypdf 合并、拆分、旋转、加密/解密 pip install pypdf
reportlab 从零生成PDF(精细排版控制) pip install reportlab
weasyprint HTML模板转PDF(更灵活的排版) pip install weasyprint
tabula-py PDF表格提取到pandas DataFrame pip install tabula-py
PyMuPDF (fitz) 高性能提取、图片转换、注释 pip install pymupdf

**选型原则:**读取用 pdfplumber,生成用 reportlab 或 weasyprint,合并拆分用 pypdf,表格提取用 tabula-py。遇到扫描版 PDF(图片型),需要 OCR 工具(如 pytesseract)配合使用,本章重点处理数字型 PDF。

PDF 文本提取

基础文本提取:pdfplumber

extract_text.py

import pdfplumber
from pathlib import Path

def extract_pdf_text(pdf_path: str) -> str:
    """提取 PDF 全部文本内容"""
    all_text = []
    with pdfplumber.open(pdf_path) as pdf:
        print(f'总页数:{len(pdf.pages)}')
        for page_num, page in enumerate(pdf.pages, start=1):
            text = page.extract_text()
            if text:
                all_text.append(f'--- 第{page_num}页 ---\n{text}')
            else:
                all_text.append(f'--- 第{page_num}页:无可提取文本(可能为扫描图片)---')
    return '\n\n'.join(all_text)

# 单文件提取
text = extract_pdf_text('annual_report.pdf')
print(text[:500])  # 预览前500字符

# 保存到文本文件
Path('extracted_text.txt').write_text(text, encoding='utf-8')

批量提取:100 个 PDF 提取关键信息到 Excel

batch_extract_to_excel.py

import pdfplumber
import pandas as pd
import re
from pathlib import Path

def extract_key_info(pdf_path: Path) -> dict:
    """从财务报告PDF中提取关键信息(以发票为例)"""
    info = {
        'filename':     pdf_path.name,
        'invoice_no':   None,
        'amount':       None,
        'date':         None,
        'vendor':       None,
    }
    with pdfplumber.open(pdf_path) as pdf:
        # 通常关键信息在首页
        text = pdf.pages[0].extract_text() or ''

    # 正则提取发票号
    m = re.search(r'发票号[码::\s]*([A-Z0-9]{8,20})', text)
    if m:
        info['invoice_no'] = m.group(1)

    # 提取金额(匹配"合计:12,345.67"格式)
    m = re.search(r'合计[::\s]*[¥¥]?([\d,]+\.?\d*)', text)
    if m:
        info['amount'] = float(m.group(1).replace(',', ''))

    # 提取日期
    m = re.search(r'(\d{4})[年/\-](\d{1,2})[月/\-](\d{1,2})', text)
    if m:
        info['date'] = f"{m.group(1)}-{m.group(2):>02}-{m.group(3):>02}"

    return info

# 批量处理目录下所有 PDF
pdf_dir = Path('invoices')
results = []
errors = []

for pdf_file in sorted(pdf_dir.glob('*.pdf')):
    try:
        info = extract_key_info(pdf_file)
        results.append(info)
        print(f'  [OK] {pdf_file.name}')
    except Exception as e:
        errors.append({'filename': pdf_file.name, 'error': str(e)})
        print(f'  [错误] {pdf_file.name}: {e}')

# 保存结果
df = pd.DataFrame(results)
df.to_excel('invoice_summary.xlsx', index=False)
print(f'\n完成:{len(results)} 成功,{len(errors)} 失败')
print('结果已保存至 invoice_summary.xlsx')

**扫描版 PDF 的处理:**如果 page.extract_text() 返回空字符串或乱码,说明 PDF 是扫描图片,需要 OCR。推荐用 pytesseract(需安装 Tesseract 引擎)或调用云端 OCR API(百度、阿里云、腾讯云均有免费额度)。本书第12章会介绍如何调用 AI API 做更高精度的 OCR。

PDF 合并与拆分

安装 pypdf

终端

pip install pypdf

合并多个 PDF

merge_pdfs.py

from pypdf import PdfWriter
from pathlib import Path

def merge_pdfs(input_files: list, output_path: str):
    """按顺序合并多个 PDF 文件"""
    writer = PdfWriter()

    for pdf_path in input_files:
        writer.append(str(pdf_path))
        print(f'  添加:{Path(pdf_path).name}')

    with open(output_path, 'wb') as f:
        writer.write(f)

    total_pages = sum(1 for _ in writer.pages)
    print(f'合并完成:共 {len(input_files)} 个文件,{total_pages} 页
# 合并当月所有部门报告
pdf_files = sorted(Path('monthly_reports/2024-12').glob('*.pdf'))
merge_pdfs(pdf_files, 'monthly_reports/2024-12_合并报告.pdf')

按页码范围拆分

split_pdf.py

from pypdf import PdfReader, PdfWriter
from pathlib import Path

def split_pdf_by_range(input_path: str, ranges: list, output_dir: str):
    """
    按页码范围拆分 PDF。
    ranges: [(开始页, 结束页, 输出文件名), ...] 页码从 1 开始
    """
    reader = PdfReader(input_path)
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    for start_page, end_page, filename in ranges:
        writer = PdfWriter()
        # pypdf 页码索引从 0 开始,用户输入从 1 开始
        for page_num in range(start_page - 1, end_page):
            writer.add_page(reader.pages[page_num])

        out_path = output_dir / filename
        with open(out_path, 'wb') as f:
            writer.write(f)
        print(f'  已提取第{start_page}-{end_page}页
# 使用示例:把一份100页年报拆成四个季度
split_pdf_by_range(
    'annual_report_2024.pdf',
    ranges=[
        (1,  25,  'Q1_report.pdf'),
        (26, 50,  'Q2_report.pdf'),
        (51, 75,  'Q3_report.pdf'),
        (76, 100, 'Q4_report.pdf'),
    ],
    output_dir='split_reports',
)

提取特定页面(如封面页)

extract_pages.py

from pypdf import PdfReader, PdfWriter
from pathlib import Path

def extract_cover_pages(pdf_dir: str, output_path: str):
    """从每份报告中提取第1页(封面),合并为一个汇总文件"""
    writer = PdfWriter()
    pdf_dir = Path(pdf_dir)

    for pdf_file in sorted(pdf_dir.glob('*.pdf')):
        reader = PdfReader(str(pdf_file))
        if len(reader.pages) > 0:
            writer.add_page(reader.pages[0])  # 索引0 = 第1页
            print(f'  提取封面:{pdf_file.name}')

    with open(output_path, 'wb') as f:
        writer.write(f)
    print(f'封面汇总已保存:{output_path}')

# 提取所有月报的封面页
extract_cover_pages('monthly_reports/', 'all_covers_2024.pdf')

PDF 表格提取

pdfplumber 提取表格

pdfplumber 对排版规则的数字 PDF 表格效果较好,支持直接返回列表结构:

extract_table_pdfplumber.py

import pdfplumber
import pandas as pd

def extract_tables_from_pdf(pdf_path: str) -> list[pd.DataFrame]:
    """从 PDF 中提取所有表格,返回 DataFrame 列表"""
    all_tables = []
    with pdfplumber.open(pdf_path) as pdf:
        for page_num, page in enumerate(pdf.pages, start=1):
            tables = page.extract_tables()
            for table_idx, table in enumerate(tables):
                if not table:
                    continue
                # 第一行作为列名,其余作为数据
                df = pd.DataFrame(table[1:], columns=table[0])
                df['_source_page'] = page_num
                df['_table_index'] = table_idx
                all_tables.append(df)
                print(f'  第{page_num}页,表格{table_idx+1}:{len(df)}行×{len(df.columns)}列')
    return all_tables

# 提取并保存
tables = extract_tables_from_pdf('financial_report.pdf')
if tables:
    with pd.ExcelWriter('extracted_tables.xlsx') as writer:
        for i, df in enumerate(tables):
            sheet_name = f'Table_{i+1}'
            df.to_excel(writer, sheet_name=sheet_name, index=False)
    print(f'\n共提取 {len(tables)} 个表格,保存至 extracted_tables.xlsx')

tabula-py 提取 PDF 表格(更强大)

extract_table_tabula.py

# pip install tabula-py
# 依赖 Java 运行时(JRE):https://www.java.com/zh-CN/download/
import tabula
import pandas as pd

# ── 提取所有页面的所有表格 ──
tables = tabula.read_pdf(
    'financial_report.pdf',
    pages='all',        # 'all' 或具体页码 '1,3,5-7'
    multiple_tables=True,
    lattice=True,       # lattice=True 适用于有边框的表格
    # stream=True,      # stream=True 适用于无边框但有空白分隔的表格
)
print(f'共检测到 {len(tables)} 个表格')

# ── 合并并保存 ──
combined = pd.concat(tables, ignore_index=True)
combined.to_excel('all_tables.xlsx', index=False)

# ── 跨页表格处理 ──
# 如果表格跨越多页,使用 pages 指定连续页码范围
cross_page_table = tabula.read_pdf(
    'report.pdf',
    pages='3-5',        # 第3到第5页是同一张跨页表格
    multiple_tables=False,  # 当作一张表格处理
    lattice=True,
)[0]
print(cross_page_table.head())

数据清洗:提取后必做

clean_extracted_data.py

import pandas as pd
import re

def clean_pdf_table(df: pd.DataFrame) -> pd.DataFrame:
    """清洗从PDF提取的表格数据"""
    df = df.copy()

    # 删除全为空/None 的行和列
    df.dropna(how='all', inplace=True)
    df.dropna(axis=1, how='all', inplace=True)

    # 去除列名和数据中的多余空白
    df.columns = [str(c).strip() for c in df.columns]
    df = df.applymap(lambda x: str(x).strip() if pd.notna(x) else x)

    # 将金额列从字符串转为数值(处理"1,234.56"格式)
    for col in df.columns:
        sample = df[col].dropna().head(5)
        if sample.str.match(r'^[\d,]+\.?\d*$').any():
            df[col] = pd.to_numeric(
                df[col].str.replace(',', ''), errors='coerce'
            )

    return df

# 使用
import tabula
tables = tabula.read_pdf('report.pdf', pages='all', multiple_tables=True)
cleaned = [clean_pdf_table(t) for t in tables]
with pd.ExcelWriter('clean_tables.xlsx') as writer:
    for i, df in enumerate(cleaned):
        df.to_excel(writer, sheet_name=f'Sheet{i+1}', index=False)

生成 PDF 报告

方案一:reportlab——从零精细控制

generate_pdf_reportlab.py

from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import cm
from reportlab.lib import colors
from reportlab.platypus import (
    SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle
)

def generate_sales_report(data: list[dict], output_path: str):
    """用 reportlab 生成销售报告PDF"""
    doc = SimpleDocTemplate(
        output_path,
        pagesize=A4,
        rightMargin=2*cm, leftMargin=2*cm,
        topMargin=2*cm, bottomMargin=2*cm,
    )

    styles = getSampleStyleSheet()
    title_style = ParagraphStyle(
        'Title', parent=styles['Title'],
        fontSize=18, spaceAfter=12,
    )
    body_style = styles['Normal']

    elements = []

    # 标题
    elements.append(Paragraph('月度销售报告', title_style))
    elements.append(Spacer(1, 0.5*cm))

    # 摘要段落
    elements.append(Paragraph(
        f'本月共录得销售记录 {len(data)} 条,合计金额详见下表。',
        body_style
    ))
    elements.append(Spacer(1, 0.5*cm))

    # 数据表格
    table_data = [['产品', '数量', '单价', '金额']]  # 表头
    total = 0
    for row in data:
        amount = row['qty'] * row['price']
        total += amount
        table_data.append([
            row['product'],
            str(row['qty']),
            f"¥{row['price']:,.2f}",
            f"¥{amount:,.2f}",
        ])
    table_data.append(['合计', '', '', f"¥{total:,.2f}"])  # 合计行

    t = Table(table_data, colWidths=[6*cm, 3*cm, 4*cm, 4*cm])
    t.setStyle(TableStyle([
        ('BACKGROUND',  (0,0), (-1,0),  colors.HexColor('#1f2d3d')),
        ('TEXTCOLOR',   (0,0), (-1,0),  colors.white),
        ('FONTNAME',    (0,0), (-1,-1), 'Helvetica'),
        ('FONTSIZE',    (0,0), (-1,0),  11),
        ('ALIGN',       (1,0), (-1,-1), 'RIGHT'),
        ('GRID',        (0,0), (-1,-1), 0.5, colors.grey),
        ('ROWBACKGROUNDS', (0,1), (-1,-2), [colors.white, colors.HexColor('#f5f7fa')]),
        ('FONTNAME',    (0,-1), (-1,-1), 'Helvetica-Bold'),
    ]))
    elements.append(t)

    doc.build(elements)
    print(f'PDF 报告已生成:{output_path}')

# 使用示例
sample_data = [
    {'product': '笔记本电脑', 'qty': 50,  'price': 4999},
    {'product': '无线鼠标',   'qty': 200, 'price': 89},
    {'product': '机械键盘',   'qty': 80,  'price': 299},
]
generate_sales_report(sample_data, 'sales_report.pdf')

方案二:weasyprint——HTML 模板转 PDF(更推荐)

用 HTML + CSS 设计报告模板,通过 Jinja2 填入数据,再用 weasyprint 渲染为 PDF。这种方式排版更灵活,前端开发者特别熟悉:

generate_pdf_weasyprint.py

# pip install weasyprint jinja2
from jinja2 import Template
from weasyprint import HTML
from pathlib import Path

HTML_TEMPLATE = """
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<style>
  body { font-family: Arial, sans-serif; margin: 2cm; }
  h1   { color: #1f2d3d; border-bottom: 2px solid #6c63ff; padding-bottom: 8px; }
  table{ width: 100%; border-collapse: collapse; margin-top: 20px; }
  th   { background: #1f2d3d; color: white; padding: 10px 14px; text-align: left; }
  td   { padding: 8px 14px; border-bottom: 1px solid #e2e8f0; }
  tr:nth-child(even) td { background: #f8fafc; }
  .total td { font-weight: bold; background: #eef2ff; }
  .footer    { margin-top: 40px; font-size: 11px; color: #888; text-align: center; }
</style>
</head>
<body>
  <h1>{{"{{"}}title{{"}}"}}</h1>
  <p>报告日期:{{"{{"}}report_date{{"}}"}}  共 {{"{{"}}rows|length{{"}}"}} 条记录</p>
  <table>
    <tr><th>产品名称</th><th>销量</th><th>单价</th><th>金额</th></tr>
    {% for row in rows %}
    <tr>
      <td>{{"{{"}}row.product{{"}}"}}</td>
      <td>{{"{{"}}row.qty{{"}}"}}</td>
      <td>¥{{"{{"}} "%.2f"|format(row.price) {{"}}"}}</td>
      <td>¥{{"{{"}} "%.2f"|format(row.qty * row.price) {{"}}"}}</td>
    </tr>
    {% endfor %}
    <tr class="total">
      <td colspan="3">合计</td>
      <td>¥{{"{{"}} "%.2f"|format(total) {{"}}"}}</td>
    </tr>
  </table>
  <div class="footer">本报告由自动化系统生成,如有疑问请联系财务部</div>
</body>
</html>
"""

def generate_pdf_from_html(data: list[dict], output_path: str):
    from datetime import date
    total = sum(r['qty'] * r['price'] for r in data)
    html_content = Template(HTML_TEMPLATE).render(
        title='月度销售报告',
        report_date=date.today().strftime('%Y年%m月%d日'),
        rows=data,
        total=total,
    )
    HTML(string=html_content).write_pdf(output_path)
    print(f'PDF 已生成:{output_path}')

sample_data = [
    {'product': '笔记本电脑', 'qty': 50,  'price': 4999},
    {'product': '无线鼠标',   'qty': 200, 'price': 89},
]
generate_pdf_from_html(sample_data, 'report_weasyprint.pdf')

**reportlab vs weasyprint:**reportlab 学习曲线陡但对生成的内容精细控制能力强,适合复杂排版;weasyprint 基于 HTML/CSS,易上手,适合内容结构化但设计较简单的报告。两者都支持中文,但需要指定包含中文字符的字体文件。

PDF 加密与权限控制

添加密码保护

encrypt_pdf.py

from pypdf import PdfReader, PdfWriter

def encrypt_pdf(input_path: str, output_path: str,
                user_password: str, owner_password: str = None):
    """
    为 PDF 添加密码保护。
    user_password:  普通用户打开时需要输入的密码
    owner_password: 所有者密码(可以修改权限),不设则与用户密码相同
    """
    reader = PdfReader(input_path)
    writer = PdfWriter()

    for page in reader.pages:
        writer.add_page(page)

    writer.encrypt(
        user_password=user_password,
        owner_password=owner_password or user_password,
        use_128bit=True,
    )

    with open(output_path, 'wb') as f:
        writer.write(f)
    print(f'已加密:{output_path}(用户密码:{user_password})')

# 单文件加密
encrypt_pdf('contract.pdf', 'contract_encrypted.pdf', user_password='Client@2024')

# 批量加密(每个客户不同密码)
import pandas as pd
df = pd.read_excel('clients.xlsx')  # 含"客户名称"和"合同密码"列
for _, row in df.iterrows():
    encrypt_pdf(
        f"contracts/{row['客户名称']}_合同.pdf",
        f"contracts_encrypted/{row['客户名称']}_合同_加密.pdf",
        user_password=str(row['合同密码']),
    )

批量解密(有权限的情况下)

decrypt_pdf.py

from pypdf import PdfReader, PdfWriter
from pathlib import Path

def decrypt_pdf(input_path: str, output_path: str, password: str) -> bool:
    """解密 PDF 文件,返回是否成功"""
    try:
        reader = PdfReader(input_path)
        if reader.is_encrypted:
            result = reader.decrypt(password)
            if result == 0:
                print(f'密码错误:{input_path}')
                return False
        writer = PdfWriter()
        for page in reader.pages:
            writer.add_page(page)
        with open(output_path, 'wb') as f:
            writer.write(f)
        return True
    except Exception as e:
        print(f'解密失败 {input_path}: {e}')
        return False

# 批量解密同一密码的文件
password = 'CompanyInternal2024'
for pdf_file in Path('encrypted').glob('*.pdf'):
    out = Path('decrypted') / pdf_file.name
    out.parent.mkdir(exist_ok=True)
    ok = decrypt_pdf(str(pdf_file), str(out), password)
    print(f'  {"[OK]" if ok else "[失败]"} {pdf_file.name}')

实战项目:财务报告处理管道

场景:每月收到各部门发来的 PDF 财务报告(10-20个),需要:提取每份报告中的关键财务数据

financial_pipeline.py — 财务报告处理管道(完整代码)

"""
财务报告处理管道
功能:批量PDF提取 依赖:pip install pdfplumber pypdf pandas openpyxl weasyprint jinja2
"""
import pdfplumber
import pandas as pd
from pathlib import Path
from datetime import date
import re
import logging
from jinja2 import Template
from weasyprint import HTML

logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
logger = logging.getLogger(__name__)

# ── 配置 ──
PDF_DIR    = Path('reports/monthly')
OUTPUT_DIR = Path('output')
PERIOD     = '2024年12月'

# ── 财务数据提取 ──
def extract_financial_data(pdf_path: Path) -> dict:
    """从单份财务报告提取关键指标"""
    dept = pdf_path.stem.replace('_report', '')
    result = {'部门': dept, '收入': None, '支出': None, '利润': None, '备注': ''}

    with pdfplumber.open(pdf_path) as pdf:
        text = '\n'.join(
            page.extract_text() or '' for page in pdf.pages
        )

    patterns = {
        '收入': r'(?:营业收入|总收入)[::]\s*[¥¥]?([\d,]+\.?\d*)',
        '支出': r'(?:营业支出|总支出)[::]\s*[¥¥]?([\d,]+\.?\d*)',
        '利润': r'(?:净利润|利润总额)[::]\s*[¥¥]?([\d,]+\.?\d*)',
    }
    for key, pattern in patterns.items():
        m = re.search(pattern, text)
        if m:
            result[key] = float(m.group(1).replace(',', ''))
        else:
            result['备注'] += f'{key}未提取;'

    return result

# ── 批量处理 ──
def batch_extract(pdf_dir: Path) -> pd.DataFrame:
    records = []
    for pdf_file in sorted(pdf_dir.glob('*.pdf')):
        try:
            data = extract_financial_data(pdf_file)
            records.append(data)
            logger.info(f'  [OK] {pdf_file.name}')
        except Exception as e:
            records.append({'部门': pdf_file.stem, '备注': f'提取失败: {e}'})
            logger.error(f'  [错误] {pdf_file.name}: {e}')
    return pd.DataFrame(records)

# ── 保存 Excel ──
def save_excel(df: pd.DataFrame, output_path: Path):
    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
        df.to_excel(writer, sheet_name='汇总数据', index=False)
        # 合计行
        summary = pd.DataFrame([{
            '部门': '合计',
            '收入': df['收入'].sum(),
            '支出': df['支出'].sum(),
            '利润': df['利润'].sum(),
        }])
        summary.to_excel(writer, sheet_name='合计', index=False)
    logger.info(f'Excel 已保存:{output_path}')

# ── 生成摘要 PDF ──
SUMMARY_HTML = """
<!DOCTYPE html><html><head><meta charset="utf-8">
<style>
body{font-family:Arial,sans-serif;margin:2cm;color:#1a1a2e}
h1{color:#1f2d3d;font-size:20px;border-bottom:3px solid #6c63ff;padding-bottom:6px}
.meta{color:#666;font-size:12px;margin-bottom:20px}
table{width:100%;border-collapse:collapse;font-size:12px}
th{background:#1f2d3d;color:#fff;padding:8px 12px;text-align:right}
th:first-child{text-align:left}
td{padding:7px 12px;border-bottom:1px solid #e2e8f0;text-align:right}
td:first-child{text-align:left}
tr:nth-child(even) td{background:#f8fafc}
.total-row td{font-weight:bold;background:#eef2ff}
.positive{color:#16a34a}.negative{color:#dc2626}
</style></head><body>
<h1>{{"{{"}}period{{"}}"}} 财务报告摘要</h1>
<div class="meta">生成时间:{{"{{"}}gen_date{{"}}"}} | 共 {{"{{"}}rows|length{{"}}"}} 个部门</div>
<table>
<tr><th>部门</th><th>收入(元)</th><th>支出(元)</th><th>利润(元)</th></tr>
{% for r in rows %}
<tr>
  <td>{{"{{"}}r['部门']{{"}}"}}</td>
  <td>{{"{{"}} "{:,.0f}".format(r['收入'] or 0) {{"}}"}}</td>
  <td>{{"{{"}} "{:,.0f}".format(r['支出'] or 0) {{"}}"}}</td>
  <td class="{{"{{"}}'positive' if (r['利润'] or 0) >= 0 else 'negative'{{"}}"}}">
    {{"{{"}} "{:,.0f}".format(r['利润'] or 0) {{"}}"}}
  </td>
</tr>
{% endfor %}
<tr class="total-row">
  <td>合计</td>
  <td>{{"{{"}} "{:,.0f}".format(total_revenue) {{"}}"}}</td>
  <td>{{"{{"}} "{:,.0f}".format(total_cost) {{"}}"}}</td>
  <td>{{"{{"}} "{:,.0f}".format(total_profit) {{"}}"}}</td>
</tr>
</table>
</body></html>
"""

def generate_summary_pdf(df: pd.DataFrame, output_path: Path, period: str):
    html = Template(SUMMARY_HTML).render(
        period=period,
        gen_date=date.today().strftime('%Y年%m月%d日'),
        rows=df.to_dict('records'),
        total_revenue=df['收入'].sum(),
        total_cost=df['支出'].sum(),
        total_profit=df['利润'].sum(),
    )
    HTML(string=html).write_pdf(str(output_path))
    logger.info(f'摘要PDF已生成:{output_path}')

# ── 主流程 ──
def main():
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

    logger.info(f'开始处理 {PERIOD} 财务报告...')
    df = batch_extract(PDF_DIR)

    save_excel(df, OUTPUT_DIR / f'{PERIOD}_财务汇总.xlsx')
    generate_summary_pdf(df, OUTPUT_DIR / f'{PERIOD}_摘要.pdf', PERIOD)

    logger.info(f'\n管道执行完成。输出目录:{OUTPUT_DIR.resolve()}')

if __name__ == '__main__':
    main()

管道设计要点: 提取、清洗、输出三个阶段完全分离,任何一段出错均有日志记录且不影响其他报告 正则模式按实际 PDF 格式调整,不同来源的报告格式不同需分别适配 Excel 和 PDF 双输出:Excel 方便管理层二次分析,PDF 便于存档和分发 利润正负值用颜色区分,管理层一眼看出亏损部门

上一章

下一章
第9章:网页抓取
本章评分
4.8  / 5  (39 评分)

💬 留言讨论