Python自动化工具：从巨潮资讯网抓取年报PDF并转换为可分析文本

简介：这是一套面向金融数据分析初学者与Python实践者的自动化年报处理工具，专为解决巨潮资讯网上市公司年报PDF难以批量词频分析的痛点而设计。资源共13个文件，含4个核心Python脚本（年报链接抓取、PDF下载、PDF转TXT、通用文本分析）、4个XML配置文件（IDEA项目结构）、1个覆盖2004–2023年年报链接的Excel数据源、1个依赖说明txt及README.md等，整体压缩包仅2.14MB，轻量易部署。已有201人学习下载，适合需快速构建年报文本挖掘流水线的用户。读者可直接运行三步脚本完成“网页抓取→PDF下载→文本提取”全流程，获得标准化TXT文本用于后续关键词统计、情感分析或LDA主题建模；配套xlsx链接库与模块化代码结构，便于按行业/年份筛选目标公司，并支持自定义清洗规则扩展。

1. 项目概述与核心价值

最近在做一个金融文本分析的小研究，需要批量处理上市公司的年报。源头数据自然想到了巨潮资讯网，这是国内最权威的上市公司信息披露平台。但实际操作起来，发现几个痛点非常明显：手动一个个搜索、点开、下载PDF效率极低；下载下来的PDF文件格式混乱，直接做文本分析就像在沙子里淘金，噪音太多；最后还得想办法把PDF里的文字抠出来转成干净的TXT。这一套流程走下来，半天时间就没了，还容易出错。于是，我就动手写了这个Python小工具，目标很明确：一键搞定从巨潮资讯网抓取指定公司年报、下载PDF、并自动转换为规整TXT文本的全流程，为后续的词频分析、情感分析或机器学习建模提供高质量的数据原料。

这个小工具的核心用户，是像我一样需要处理大量金融文本的研究员、学生、量化分析爱好者或者任何对上市公司信息披露文本感兴趣的人。它解决的问题非常具体：将非结构化的、散落在网络上的PDF公告，转化为结构化的、纯净的文本数据。你不需要懂复杂的网络协议，也不需要去研究各种PDF解析库的坑，工具已经把这条流水线搭建好了。你只需要提供目标公司的股票代码和想要的年份，它就能帮你把脏活累活都干了，让你能专注于更有价值的分析工作本身。

2. 整体设计思路与技术选型

做这个工具，我的设计原则就八个字：流程自动化，结果可用化。整个工具被设计成一条清晰的流水线，每个环节解决一个具体问题，并且环环相扣。

2.1 核心流程拆解

整个工具的工作流可以分解为四个核心阶段：

1. 信息检索与链接抓取：根据用户输入的股票代码和年份，在巨潮资讯网上定位到对应的年报PDF公告页面，并提取出PDF文件的真实下载地址。
2. 文件下载与本地管理：将PDF文件可靠地下载到本地指定目录，并按照“公司代码/年份”的层级结构进行存储，便于后续管理。
3. PDF文本内容提取：解析下载的PDF文件，尽可能准确、完整地提取出其中的文字内容，并处理PDF中常见的格式干扰问题。
4. 文本清洗与格式化输出：对提取出的原始文本进行深度清洗，去除页眉页脚、无关字符、多余空格和换行，生成适合进行自然语言处理（NLP）的纯净TXT文件。

2.2 关键技术选型与考量

为什么用这些库？每个选择背后都有实际踩坑后的考量。

• 网络请求与解析：requests+lxml

  • requests是Python HTTP库的绝对首选，其API简洁直观，会话保持、超时设置、异常处理都非常完善。相比urllib，代码可读性和开发效率高出一大截。
  • 为什么选lxml而不是BeautifulSoup？在解析像巨潮资讯网这类结构复杂、动态内容可能较多的网页时，lxml的解析速度和XPath表达能力更强。巨潮网的页面结构虽然可能变动，但其公告列表和链接的DOM路径通常有规律可循，用XPath可以更精准、更稳定地定位元素。BeautifulSoup更适合解析不规范的HTML，而lxml在速度和精准度上更适合这个生产数据抓取的场景。

• PDF文本提取：pdfplumber

  • 这是选型中最重要的决定。我对比过PyPDF2/PyMuPDF(fitz)/pdfminer.six等多个库。
  • PyPDF2对中文支持 historically 不太友好，提取复杂版式PDF时容易乱码或丢字。
  • PyMuPDF性能极佳，但对某些内嵌字体或特殊编码的PDF，提取文本需要额外配置，不够“傻瓜化”。
  • pdfminer.six功能强大，但API相对复杂，需要更多的代码来处理布局分析。
  • pdfplumber正是在pdfminer.six的基础上进行了高层封装，它提供了极其友好的API（比如.extract_text()），并且在表格提取、字符位置识别上表现突出。对于年报这种以连续文字为主、夹杂少量表格的文档，pdfplumber能在保证提取精度的前提下，极大简化代码，是平衡了易用性与能力的绝佳选择。

• 文本清洗与正则表达式：re

  • 从PDF里提出的文本是“毛坯房”，充满了换行符（因为PDF排版导致的断行）、多余空格、页码标记（如“- 1 -”）、页眉页脚信息等。re（正则表达式）是进行批量文本模式匹配和替换的瑞士军刀，是深度清洗不可或缺的工具。

• 路径与文件操作：os,pathlib

  • 用于创建按公司和年份分类的目录结构，管理文件路径。使用pathlib可以让路径操作更面向对象、更清晰，跨平台兼容性也更好。

注意：关于动态内容与反爬。巨潮资讯网的部分页面数据可能是通过JavaScript动态加载的。经过实测，其核心的公告列表和PDF链接在初始HTML响应中通常就已包含，直接使用requests获取页面内容并用lxml解析即可抓到。如果未来网站改版，链接变为动态加载，则需要考虑使用Selenium或Playwright等浏览器自动化工具来模拟浏览器行为获取完整页面。本工具基于当前可稳定访问的静态结构设计。

3. 核心模块详解与实操要点

接下来，我们深入每个核心模块，看看代码具体怎么写，以及有哪些必须注意的细节。

3.1 链接抓取模块：如何精准定位PDF下载地址

这是第一步，也是最容易因为网站结构变动而失效的一步。我们的目标是：输入股票代码（如000001平安银行）和年份（如2023），得到对应年报PDF的最终下载链接。

import requests from lxml import etree import re def fetch_annual_report_url(stock_code, year): """ 根据股票代码和年份，从巨潮资讯网获取年报PDF的下载链接。 Args: stock_code (str): 6位股票代码，例如 '000001' year (str or int): 年份，例如 '2023' Returns: str: 年报PDF文件的直接下载链接，如果未找到则返回None。 """ # 构造搜索请求的URL模板（实际URL需根据网站当前结构调整） # 这里是一个示例，巨潮网通常通过特定接口或搜索页面进行查询 base_url = f"http://www.cninfo.com.cn/new/disclosure/stock?stockCode={stock_code}" # 注意：实际URL可能需要包含更多参数，如公告类型、年份范围等。 # 更常见的做法是模拟其搜索接口的POST请求或解析搜索结果页。 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' } try: resp = requests.get(base_url, headers=headers, timeout=10) resp.raise_for_status() # 检查HTTP请求是否成功 resp.encoding = 'utf-8' # 或根据网页实际编码调整，如'gbk' # 使用lxml解析HTML html = etree.HTML(resp.text) # **关键步骤：使用XPath定位包含指定年份年报的链接** # 示例XPath，需要根据巨潮网实际页面结构进行修改！！！ # 思路：寻找所有公告链接，然后过滤出文本中包含“年报”和指定年份的链接。 # 例如：//a[contains(@href, '.pdf') and contains(text(), '年报') and contains(text(), '2023')] year_str = str(year) # 一个更稳健的XPath可能如下（需开发者手动分析页面后确定）: # 假设每个公告项在一个class为‘announcement’的div里，标题在a标签内 link_xpath = f"//div[contains(@class, 'announcement')]//a[contains(., '年度报告') and contains(., '{year_str}')]/@href" pdf_links = html.xpath(link_xpath) if not pdf_links: print(f"未找到股票{stock_code}在{year}年的年度报告链接。") return None # 通常第一个链接就是最新的或最相关的年报 target_relative_url = pdf_links[0] # 将相对URL补全为绝对URL # 需要观察巨潮网链接是相对路径还是绝对路径 if target_relative_url.startswith('http'): pdf_url = target_relative_url else: # 这里需要根据网站基础URL进行拼接，例如： pdf_url = f"http://www.cninfo.com.cn{target_relative_url}" print(f"成功找到PDF链接: {pdf_url}") return pdf_url except requests.RequestException as e: print(f"网络请求失败: {e}") return None except Exception as e: print(f"解析页面时发生未知错误: {e}") return None

实操要点与避坑指南：

1. XPath是动态的：上面代码中的link_xpath只是一个示例。你必须亲自打开巨潮资讯网，使用浏览器的开发者工具（F12），查看目标公告列表的HTML结构，然后编写适合当前页面结构的XPath。这是爬虫项目最核心、最需要手动适配的部分。
2. 处理反爬机制：巨潮网对频繁访问可能会有限制。务必在请求头headers中设置一个真实的User-Agent。更进阶的做法是使用requests.Session()保持会话，并在请求间添加随机延时（如time.sleep(random.uniform(1, 3))），模拟人类操作。
3. 链接有效性验证：获取到的pdf_url可能是一个中间页面链接，而不是直接的PDF地址。有些网站会先跳转到一个预览或确认页面。你可能需要再次访问这个链接，从响应内容中提取最终的.pdf链接。可以通过检查URL是否以.pdf结尾，或者分析第二次请求返回的HTML/JS代码来判断。
4. 错误处理要周全：网络超时、页面结构变化、链接失效等情况都要考虑到，并用try...except包裹，给出明确的错误提示，避免程序因单个任务失败而完全崩溃。

3.2 文件下载与管理模块：稳定下载与有序存储

拿到正确的PDF链接后，下一步就是把它可靠地下载下来，并放到合适的位置。

import os from pathlib import Path def download_pdf(pdf_url, stock_code, year, save_dir='./reports'): """ 下载PDF文件并保存到本地结构化目录。 Args: pdf_url (str): PDF文件的直接下载链接。 stock_code (str): 股票代码，用于创建子目录。 year (str or int): 年份，用于创建子目录和命名文件。 save_dir (str): 根存储目录。 Returns: str or None: 成功则返回保存的文件路径，失败返回None。 """ if not pdf_url: return None # 创建按股票代码和年份组织的目录结构 stock_dir = Path(save_dir) / stock_code year_dir = stock_dir / str(year) year_dir.mkdir(parents=True, exist_ok=True) # parents=True允许创建多层目录 # 从URL中提取文件名，或自己构造 # 方法1：使用URL的最后一部分作为文件名（可能不友好） # filename = pdf_url.split('/')[-1] # 方法2：自己构造清晰的文件名（推荐） filename = f"{stock_code}_{year}_年报.pdf" file_path = year_dir / filename headers = { 'User-Agent': 'Mozilla/5.0 ...', # 有时需要添加Referer头，表明请求来源，防止盗链 'Referer': 'http://www.cninfo.com.cn/' } try: print(f"正在下载: {pdf_url}") # stream=True用于流式下载大文件，避免内存占用过高 response = requests.get(pdf_url, headers=headers, stream=True, timeout=30) response.raise_for_status() # 检查响应内容是否是PDF content_type = response.headers.get('content-type', '') if 'pdf' not in content_type.lower(): print(f"警告：下载的内容类型不是PDF，而是 {content_type}。链接可能已失效或需要处理。") # 这里可以尝试处理非PDF响应（如跳转页面） # 以二进制写入模式保存文件 with open(file_path, 'wb') as f: # 分块写入，对于大文件更安全 for chunk in response.iter_content(chunk_size=8192): if chunk: f.write(chunk) print(f"文件已保存至: {file_path}") return str(file_path) except requests.RequestException as e: print(f"下载PDF失败: {e}") # 可选：删除可能已部分下载的损坏文件 if file_path.exists(): file_path.unlink() return None except IOError as e: print(f"文件写入失败: {e}") return None

实操要点与避坑指南：

1. 流式下载（stream=True）：对于几MB甚至几十MB的年报PDF，务必使用流式下载。这允许你分块读取响应内容并写入文件，而不是一次性将整个文件加载到内存中，对于内存管理和大文件下载至关重要。
2. 目录结构设计：./reports/000001/2023/000001_2023_年报.pdf这样的结构一目了然，后续管理和批量处理都非常方便。使用pathlib的Path对象进行路径操作，比用字符串拼接更优雅、更安全（自动处理不同操作系统的路径分隔符）。
3. 设置超时和重试：网络环境不稳定，下载可能中断。可以为requests.get设置timeout参数。对于重要的批量任务，可以考虑实现一个简单的重试机制（例如，使用tenacity库或在循环中重试几次）。
4. 验证文件完整性：简单的验证可以检查文件大小是否过小（可能下载失败），或者尝试用pdfplumber打开一下，看是否会报“非PDF文件”错误。

4. PDF转TXT的核心实现与深度清洗

下载好PDF只是拿到了原材料，转换成干净文本才是提升数据质量的关键。这一步的挑战在于，PDF本质上是面向打印的格式，提取文本时会携带大量排版信息。

4.1 基础文本提取

我们先看看如何用pdfplumber把文字“读”出来。

import pdfplumber def pdf_to_raw_text(pdf_path): """ 使用pdfplumber提取PDF中的所有文本。 Args: pdf_path (str): PDF文件的路径。 Returns: str: 提取出的原始文本。 """ raw_text = "" try: with pdfplumber.open(pdf_path) as pdf: for page_num, page in enumerate(pdf.pages): # 提取单页文本 page_text = page.extract_text() if page_text: # 在每页文本间添加一个换行符作为分隔（可选） raw_text += page_text + "\n" else: print(f"警告：第 {page_num+1} 页可能为扫描图像或未提取到文本。") return raw_text except Exception as e: print(f"打开或解析PDF文件失败: {e}") return ""

4.2 深度文本清洗：从“毛坯”到“精装”

raw_text直接可读，但充满了问题。下面是一个清洗函数的示例，它处理了多种常见噪音：

import re def clean_extracted_text(raw_text): """ 对提取的原始文本进行深度清洗。 Args: raw_text (str): 从PDF提取的原始文本。 Returns: str: 清洗后的纯净文本。 """ if not raw_text: return "" text = raw_text # 1. 移除PDF中常见的页码标记，如 “- 1 -”, “1 / 100”, “[1]” # 这类标记通常在行首或行尾，且单独成行或与少量字符在一起 text = re.sub(r'^\s*[-–—]?\s*\d+\s*[-–—]?\s*$', '', text, flags=re.MULTILINE) # 处理 “- 1 -” text = re.sub(r'^\s*\d+\s*/\s*\d+\s*$', '', text, flags=re.MULTILINE) # 处理 “1 / 100” text = re.sub(r'^\s*\[\d+\]\s*$', '', text, flags=re.MULTILINE) # 处理 “[1]” # 2. 移除常见的页眉页脚关键词（需根据年报特点调整） header_footer_keywords = [ r'巨潮资讯网', r'www\.cninfo\.com\.cn', r'公司年度报告', r'年度报告摘要', r'第\s*\d+\s*页\s*共\s*\d+\s*页', r'股票代码：\d+', r'股票简称：.*', r'^.*\d{4}年\s*年度报告.*$' # 匹配整行为页眉的情况 ] for pattern in header_footer_keywords: text = re.sub(pattern, '', text, flags=re.MULTILINE | re.IGNORECASE) # 3. 修复因PDF排版导致的错误换行：将行末不是句号、分号、感叹号、问号的中文换行连接起来 # 中文段落内换行通常没有结束符，直接连接。英文则需谨慎。 # 这是一个简化策略，针对中文文本效果较好。 lines = text.split('\n') cleaned_lines = [] i = 0 while i < len(lines): line = lines[i].strip() if not line: # 跳过空行 i += 1 continue # 如果当前行不以中文句子结束符结尾，且下一行非空，则合并 if i + 1 < len(lines): next_line = lines[i+1].strip() if next_line and not re.search(r'[。？！；.””]$', line): line = line + next_line i += 1 # 跳过下一行，因为它已被合并 cleaned_lines.append(line) i += 1 text = '\n'.join(cleaned_lines) # 4. 合并过多的空白字符（空格、制表符等）为单个空格 text = re.sub(r'\s+', ' ', text) # 5. 移除首尾空白 text = text.strip() # 6. (可选但推荐) 移除或替换特殊字符、不可见字符 # 移除ASCII控制字符（除了换行和制表符） text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text) # 将中文全角空格、不间断空格等替换为普通空格 text = re.sub(r'[\u3000\u00A0]', ' ', text) return text

4.3 保存清洗后的文本

最后，将清洗好的文本保存为TXT文件。

def save_clean_text(clean_text, pdf_file_path): """ 将清洗后的文本保存为TXT文件，与PDF文件同名同目录。 Args: clean_text (str): 清洗后的文本内容。 pdf_file_path (str): 原始PDF文件的路径。 Returns: str: 保存的TXT文件路径。 """ if not clean_text: print("清洗后的文本为空，跳过保存。") return None pdf_path = Path(pdf_file_path) # 生成对应的TXT文件路径，例如将 .pdf 替换为 .txt txt_path = pdf_path.with_suffix('.txt') try: # 使用UTF-8编码保存，确保中文字符正确 with open(txt_path, 'w', encoding='utf-8') as f: f.write(clean_text) print(f"清洗文本已保存至: {txt_path}") return str(txt_path) except IOError as e: print(f"保存TXT文件失败: {e}") return None

实操要点与避坑指南：

1. 清洗规则是迭代出来的：上面的清洗函数是一个通用起点。不同的PDF模板（不同公司、不同年份、不同排版软件生成）会产生不同的噪音。最好的方法是：先处理几份典型的年报，观察raw_text的输出，找出其中特定的页眉、页脚、页码、无关水印的规律，然后补充或修改正则表达式模式。这是一个“观察-分析-添加规则”的循环过程。
2. “修复错误换行”是双刃剑：这个步骤对提高中文文本的连贯性至关重要，因为它把被PDF硬换行拆散的句子重新拼接起来。但算法（如上面的简单合并逻辑）可能误判，例如将表格中不同单元格的内容错误地合并。如果后续分析对段落结构要求极高，可能需要更复杂的启发式规则，或者考虑保留原始换行，在后续分析中再处理。
3. 编码问题：务必使用utf-8编码读写TXT文件，这是处理中文文本的标准。pdfplumber提取的文本通常是Unicode字符串，直接写入即可。
4. 处理扫描件PDF：如果年报是扫描版图片（这在早期年报中常见），pdfplumber的.extract_text()将返回空或极少文本。此时需要OCR（光学字符识别）技术。你可以集成pytesseract库（Google Tesseract的Python封装）和pdf2image库（将PDF页面转为图片）来处理。但这会极大增加复杂度和运行时间，仅当确定目标PDF为扫描件时才需启用。

5. 集成与主流程控制

将上述模块串联起来，形成一个完整的工具。我们还需要一个主函数来协调整个流程，并处理批量任务。

import time import random def process_single_report(stock_code, year, base_save_dir='./reports'): """ 处理单只股票单一年份的年报：抓取->下载->转换->清洗->保存。 Args: stock_code (str): 股票代码。 year (str or int): 年份。 base_save_dir (str): 基础保存目录。 Returns: bool: 是否成功处理。 """ print(f"\n{'='*50}") print(f"开始处理: 股票{stock_code}, {year}年年报") print(f"{'='*50}") # 步骤1: 获取PDF下载链接 pdf_url = fetch_annual_report_url(stock_code, year) if not pdf_url: print("获取PDF链接失败，终止流程。") return False # 可选：添加随机延时，避免请求过快 time.sleep(random.uniform(1, 3)) # 步骤2: 下载PDF pdf_path = download_pdf(pdf_url, stock_code, year, base_save_dir) if not pdf_path: print("下载PDF失败，终止流程。") return False # 步骤3: 提取原始文本 raw_text = pdf_to_raw_text(pdf_path) if not raw_text: print("从PDF提取文本失败，可能为扫描件或加密文档。") # 这里可以尝试调用OCR备用流程 return False # 步骤4: 清洗文本 clean_text = clean_extracted_text(raw_text) # 步骤5: 保存清洗后文本 txt_path = save_clean_text(clean_text, pdf_path) success = txt_path is not None status = "成功" if success else "失败" print(f"处理完成，状态: {status}") return success def batch_process(stock_list, year_range, base_save_dir='./reports'): """ 批量处理多只股票多个年份的年报。 Args: stock_list (list): 股票代码列表，如 ['000001', '000002']。 year_range (list): 年份列表，如 [2022, 2023]。 base_save_dir (str): 基础保存目录。 """ results = [] total = len(stock_list) * len(year_range) processed = 0 for stock_code in stock_list: for year in year_range: processed += 1 print(f"\n进度: ({processed}/{total})") try: success = process_single_report(stock_code, year, base_save_dir) results.append((stock_code, year, success)) except Exception as e: print(f"处理{stock_code}-{year}时发生未捕获的异常: {e}") results.append((stock_code, year, False)) # 批量任务中，更长的随机延时以示友好 time.sleep(random.uniform(2, 5)) # 打印汇总报告 print(f"\n{'='*50}") print("批量处理完成！汇总报告：") success_count = sum(1 for _, _, s in results if s) print(f"总计任务: {len(results)}， 成功: {success_count}， 失败: {len(results)-success_count}") if success_count < len(results): print("失败详情:") for code, yr, suc in results: if not suc: print(f" - {code} ({yr})") # 示例：主程序入口 if __name__ == "__main__": # 单次处理示例 # process_single_report('000001', 2023) # 批量处理示例 my_stocks = ['000001', '600519'] # 平安银行， 贵州茅台 my_years = [2022, 2023] batch_process(my_stocks, my_years, base_save_dir='./annual_reports')

6. 常见问题排查与实战技巧

在实际运行中，你几乎一定会遇到下面这些问题。这里是我踩过坑后总结的排查清单和应对技巧。

6.1 网络与请求相关问题

6.2 数据提取与解析问题

6.3 文本清洗与格式问题

6.4 效率与稳定性优化技巧

1. 使用连接池与会话：在批量处理时，使用requests.Session()可以复用TCP连接，显著提升请求效率。
2. 实现增量抓取：在本地维护一个记录已成功处理(股票代码, 年份)的日志文件或数据库。每次运行前先检查，避免重复下载和转换。
3. 加入健壮的日志系统：不要只靠print。使用Python的logging模块，将运行信息、错误、警告记录到文件，便于后期排查问题。
4. 设置全局超时与重试：除了单个请求的超时，可以为整个脚本设置一个运行超时，防止因某个任务卡死而无限等待。使用retrying或tenacity库可以优雅地实现重试逻辑。
5. 处理编码探测：巨潮网页面编码可能是UTF-8或GBK。如果resp.text出现乱码，可以尝试resp.content.decode('gbk')，或者使用chardet库自动探测编码：import chardet; encoding = chardet.detect(resp.content)['encoding']。

这个工具链搭建起来后，你就拥有了一套自动化的金融文本数据生产线。从几十份到上千份年报，都可以在设定好任务后让程序自动运行。产出的规整TXT文件，可以直接喂给像jieba、snowNLP这样的中文文本分析库，或者sklearn、TensorFlow等机器学习框架，进行更深度的洞察挖掘。

本文还有配套的精品资源，点击获取