【实战教程】用Python+AI自动提取PDF财报数据：从0到1完整指南

作为一名财务分析师，我每周都要处理大量PDF财报。手工复制粘贴太痛苦，直到我用Python+AI搭了一套自动化工具，现在10分钟就能搞定原来需要2小时的工作。这篇文章分享完整实现方案，包含可直接运行的代码。

---

一、项目背景

财务分析的第一步是数据提取。传统方式是：

1. 打开PDF财报
2. 逐页翻阅找数据
3. 手动复制粘贴到Excel
4. 整理格式、计算指标

这种方式有几个痛点：

• 效率低：一份财报要处理30分钟
• 易出错：手工复制容易遗漏或出错
• 格式乱：不同公司财报格式不统一
• 无法批量：100份财报就要重复100次

本文介绍如何用Python+AI实现自动化提取。

---

二、技术架构

整体架构如下：
PDF财报 → 文本提取 → AI解析 → 结构化数据 → Excel输出

涉及技术栈：
- **pymupdf**：PDF文本提取
- **paddleocr**：OCR识别（处理扫描版）
- **DashScope/Qwen3**：AI解析财务数据
- **pandas**：数据处理和Excel输出

---

#  三、环境准备

### 1. 安装依赖

```bash
pip install pymupdf paddleocr dashscope pandas openpyxl

2. 配置API Key

# config.py
DASHSCOPE_API_KEY = "your-api-key-here"

获取地址：https://dashscope.aliyun.com

四、核心实现

4.1 PDF文本提取
财报PDF分两种：文字版和扫描版。

import fitz  # pymupdf
from paddleocr import PaddleOCR

class PDFExtractor:
def __init__(self):
        self.ocr = PaddleOCR(use_angle_cls=True, lang='ch')
  
def extract_text(self, pdf_path):
"""提取PDF文本，自动判断文字版/扫描版"""
        doc = fitz.open(pdf_path)
        text = ""
      
# 尝试直接提取文字
        for page in doc:
            text += page.get_text()
      
# 文字太少说明是扫描版，启用OCR
        if len(text) < 1000:
print("检测到扫描版PDF，启动OCR...")
            text = self._extract_with_ocr(doc)
      
        return text
  
def _extract_with_ocr(self, doc):
"""OCR识别扫描版PDF"""
        text = ""
        for page_num in range(len(doc)):
# 页面转图片
            pix = doc[page_num].get_pixmap(matrix=fitz.Matrix(2, 2))
            img_path = f"temp_page_{page_num}.png"
pix.save(img_path)
          
# OCR识别
            result = self.ocr.ocr(img_path, cls=True)
            for line in result[0]:
                text += line[1][0] + "\n"
      
        return text

4.2 AI解析财务数据
用Qwen3提取关键财务指标。

import dashscope
import json
import re

class FinancialDataExtractor:
def __init__(self, api_key):
        dashscope.api_key = api_key
  
def extract(self, text, company_name):
"""提取财务数据"""
        prompt = f"""
请从以下财报文本中提取关键财务指标，以JSON格式返回。
      
需要提取的指标：
1. 营业收入（本期、上期、同比增幅）
2. 净利润（本期、上期、同比增幅）
3. 总资产
4. 净资产
5. 经营现金流
6. 毛利率
7. 净利率
8. ROE（净资产收益率）
      
财报文本：
        {text[:8000]}
      
请严格按以下JSON格式返回：
        {{
            "company": "{company_name}",
            "revenue": {{
"current": "本期营收（亿元）",
"previous": "上期营收（亿元）",
"yoy": "同比增幅（%）"
            }},
            "net_profit": {{
"current": "本期净利润（亿元）",
"previous": "上期净利润（亿元）",
"yoy": "同比增幅（%）"
            }},
"total_assets": "总资产（亿元）",
"net_assets": "净资产（亿元）",
"operating_cash_flow": "经营现金流（亿元）",
"gross_margin": "毛利率（%）",
"net_margin": "净利率（%）",
"roe": "ROE（%）"
        }}
      
注意：
- 如果某项数据未找到，填"未披露"
- 金额统一转换为亿元
- 百分比保留2位小数
        """
      
        response = dashscope.Generation.call(
            model="qwen3-72b",
            messages=[{"role": "user", "content": prompt}],
            result_format="message"
        )
      
# 解析JSON
        content = response.output.choices[0].message.content
        json_match = re.search(r'\{.*\}', content, re.DOTALL)
      
        if json_match:
            return json.loads(json_match.group())
else:
            return {"error": "解析失败", "raw": content}

4.3 数据计算与导出
提取原始数据后，计算衍生指标并导出Excel。

import pandas as pd

class FinancialCalculator:
def calculate(self, data):
"""计算衍生指标"""
        revenue = self._parse_amount(data['revenue']['current'])
        net_profit = self._parse_amount(data['net_profit']['current'])
        total_assets = self._parse_amount(data['total_assets'])
        net_assets = self._parse_amount(data['net_assets'])
      
        metrics = {
'营业收入(亿元)': revenue,
'净利润(亿元)': net_profit,
'总资产(亿元)': total_assets,
'净资产(亿元)': net_assets,
            '净利润率(%)': round(net_profit / revenue * 100, 2) if revenue > 0 else 0,
            'ROA(%)': round(net_profit / total_assets * 100, 2) if total_assets > 0 else 0,
            '权益乘数': round(total_assets / net_assets, 2) if net_assets > 0 else 0,
        }
      
        return metrics
  
def _parse_amount(self, amount_str):
"""解析金额字符串"""
        if '未披露' in str(amount_str):
            return 0
      
# 移除单位，提取数字
        import re
        numbers = re.findall(r'[\d.]+', str(amount_str))
        return float(numbers[0]) if numbers else 0
  
def to_excel(self, metrics, output_path):
"""导出Excel"""
        df = pd.DataFrame([metrics])
        df.to_excel(output_path, index=False)
        print(f"数据已导出: {output_path}")

五、完整流程整合

def process_financial_report(pdf_path, company_name, api_key):
"""处理财报完整流程"""
    print(f"开始处理 {company_name} 的财报...")
  
# 1. 提取PDF文本
    extractor = PDFExtractor()
    text = extractor.extract_text(pdf_path)
    print(f"✅ 文本提取完成，共 {len(text)} 字符")
  
# 2. AI提取财务数据
    data_extractor = FinancialDataExtractor(api_key)
    raw_data = data_extractor.extract(text, company_name)
print("✅ 财务数据提取完成")
  
# 3. 计算衍生指标
    calculator = FinancialCalculator()
    metrics = calculator.calculate(raw_data)
print("✅ 指标计算完成")
  
# 4. 导出Excel
    output_path = f"{company_name}_财务数据.xlsx"
calculator.to_excel(metrics, output_path)
  
    return metrics

# 使用示例
if __name__ == "__main__":
    API_KEY = "your-api-key"
    result = process_financial_report(
        pdf_path="company_report.pdf",
        company_name="某某公司",
        api_key=API_KEY
    )
print(result)

六、效果对比

七、踩坑经验

PDF格式混乱：不同公司财报排版差异大，建议先检测PDF类型（文字/扫描/混合），再选择解析方式。

OCR识别率：扫描版财报识别率约95%，复杂表格容易出错。关键数据建议人工复核。

AI提取误差：特别是正负号和小数点，AI偶尔会搞错。建议设置数据校验规则。

API成本：大量调用大模型API费用不低，建议本地部署小模型或使用缓存。

数据安全：财报涉及敏感信息，注意脱敏处理。建议本地处理，数据不出内网。

八、后续优化

批量处理：支持同时处理多个PDF，自动汇总到一张Excel。
数据库存储：将提取的数据存入数据库，支持历史对比。
可视化分析：自动生成财务图表和趋势分析。
异常检测：自动识别财务数据异常，预警风险。