Python+AI实现自动提取PDF财报数据的完整指南_Python

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

一、项目背景

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

打开pdf财报
逐页翻阅找数据
手动复制粘贴到excel
整理格式、计算指标

这种方式有几个痛点：

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

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

二、技术架构

整体架构如下：pdf财报 → 文本提取 → ai解析 → 结构化数据 → excel输出

涉及技术栈：

pymupdf：pdf文本提取
paddleocr：ocr识别（处理扫描版）
dashscope/qwen3：ai解析财务数据
pandas：数据处理和excel输出

三、环境准备

安装依赖

pip install pymupdf paddleocr dashscope pandas openpyxl

配置api key

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

四、核心实现

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)