简介:这是一份面向Python爬虫初学者与金融数据爱好者的实战型资源,聚焦于自动化抓取巨潮资讯网披露的上市公司年报。资源以zip压缩包形式提供,共4个文件,包含1个Python脚本与3个txt文本,整体约4KB,体量轻巧,便于快速阅读与二次修改。其中脚本负责请求页面、解析链接并下载报告,txt文件则承担股票代码、日期参数及使用说明等辅助职责,结构简洁,适合作为爬虫入门练手项目。目前已有3196人学习下载,热度可观。通过阅读代码与说明,读者可掌握HTTP请求、HTML解析、文件保存与异常处理等关键环节,理解如何遍历报告列表、匹配目标链接并落地存储,同时体会遵守robots协议与尊重数据版权的重要性。对于希望用Python自动化收集财务报告、提升爬虫实战能力的开发者而言,这份资源提供了清晰可复用的参考范例。
1. 巨潮资讯年报批量下载:从手工点击到脚本流水线
每年三四月,上市公司年报密集披露,做投研、做数据、做审计的朋友都会经历同一件事:打开巨潮资讯网,输入股票代码,筛选“年度报告”,一页页翻,一个个点,下载下来的文件名还全是乱码似的编号。几十家公司还能忍,几百家就是纯体力活。Python 爬虫批量下载年报这件事,核心不是“能不能爬”,而是“怎么爬得稳、爬得全、爬下来的 PDF 还能对得上号”。巨潮资讯的公告检索接口是公开的 JSON 接口,返回结构清晰,比解析 HTML 页面靠谱得多。这篇笔记面向有 Python 基础、想把手动下载变成可复现流水线的从业者,从接口分析、请求构造、PDF 落盘到断点续传,把每个环节的参数和坑都摊开讲。读完你能拿到一套能直接改股票代码就跑的脚本骨架,也能明白哪些地方一改就翻车。
2. 拆解巨潮公告接口:为什么不能直接爬页面
2.1 页面渲染与接口返回的差别
巨潮资讯的公告列表页看起来是一个带分页的表格,但如果你用 requests 直接请求页面 URL,拿到的 HTML 里往往没有公告数据。原因是列表数据由前端 JavaScript 异步加载,真正的数据源是一个 POST 接口。常见做法是打开浏览器开发者工具的 Network 面板,筛选 XHR/Fetch 请求,翻一页列表,就能看到那个返回 JSON 的请求。它的请求地址通常形如http://www.cninfo.com.cn/new/hisAnnouncement/query,请求体是表单格式,返回结构里announcements字段是一个数组,每个元素包含announcementTitle、adjunctUrl、secCode、secName、announcementTime等字段。直接请求这个接口,比用 Selenium 渲染页面再解析 DOM 要快一个数量级,也更稳定。
这里有一个选型理由:Selenium 或 Playwright 虽然能模拟点击,但资源消耗大、速度慢,而且页面结构一改就全废。接口请求只要参数对,返回的就是结构化数据,解析成本极低。我一般会优先找接口,找不到再考虑无头浏览器。巨潮这个接口多年未大改,参数体系相对固定,适合写成脚本长期用。
2.2 请求参数逐个说明
接口的请求体里几个关键参数决定了你拿到什么数据:
| 参数名 | 含义 | 常用取值 |
|---|---|---|
pageNum | 页码 | 从 1 开始递增 |
pageSize | 每页条数 | 30 或 50,太大可能被截断 |
column | 板块 | szse代表深市,sse代表沪市 |
tabName | 栏目名 | fulltext表示全文检索 |
searchkey | 搜索关键词 | 空字符串表示不限制 |
secid | 证券内部 ID | 可留空,用stock参数替代 |
stock | 股票代码,组织代码 | 格式如000001,gssz0000001 |
category | 公告分类 | category_ndbg_szsh对应年度报告 |
seDate | 日期范围 | 格式2024-01-01~2024-12-31 |
其中category是筛选年报的关键。巨潮的分类编码里,category_ndbg_szsh表示年度报告,category_bndbg_szsh表示半年度报告,category_qybg_szsh表示季度报告。如果你不传这个参数,返回的是所有公告类型,还得自己用标题关键词过滤,容易漏也容易混入摘要。stock参数的格式比较讲究,是“股票代码,组织代码”的拼接,组织代码可以从接口返回的orgId字段拿到,也可以先用股票代码搜索一次,从结果里提取。如果只传股票代码不传组织代码,部分情况下接口也能返回,但为了稳定,建议拼全。
2.3 用 Python 发第一个请求
下面是最小可运行代码,请求某只股票在指定日期范围内的年报列表:
import requests import json url = "http://www.cninfo.com.cn/new/hisAnnouncement/query" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "X-Requested-With": "XMLHttpRequest", "Referer": "http://www.cninfo.com.cn/new/commonUrl?url=disclosure/list/notice" } payload = { "pageNum": 1, "pageSize": 30, "column": "szse", "tabName": "fulltext", "searchkey": "", "secid": "", "stock": "000001,gssz0000001", # 股票代码,组织代码 "category": "category_ndbg_szsh", # 年度报告分类 "seDate": "2024-01-01~2024-12-31", "sortName": "", "sortType": "", "isHLtitle": "true" } resp = requests.post(url, headers=headers, data=payload, timeout=15) data = resp.json() for item in data.get("announcements", []): print(item["secCode"], item["secName"], item["announcementTitle"], item["adjunctUrl"])这段代码的逻辑是:构造表单数据 POST 到接口,拿到 JSON 后遍历announcements数组,打印股票代码、名称、公告标题和附件相对路径。adjunctUrl是 PDF 的下载路径片段,前面需要拼上http://static.cninfo.com.cn/才能访问。timeout=15是必须的,巨潮接口偶尔响应慢,不设超时脚本会卡死。headers里的Referer和X-Requested-With建议带上,虽然部分情况下不带也能返回,但带上更接近浏览器行为,降低被限流的概率。
3. 批量下载与文件命名:把 PDF 稳稳落到本地
3.1 分页遍历与总量控制
接口返回的 JSON 里有一个totalAnnouncement字段,表示符合条件的公告总数。用这个总数除以pageSize就能算出总页数。但要注意,totalAnnouncement有时会略大于实际可翻页的数量,所以更稳妥的做法是循环请求直到announcements为空或页码超过一个上限。下面是一个分页拉取全部年报的函数:
import time def fetch_all_announcements(stock_code, org_id, se_date, category="category_ndbg_szsh"): url = "http://www.cninfo.com.cn/new/hisAnnouncement/query" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "X-Requested-With": "XMLHttpRequest", "Referer": "http://www.cninfo.com.cn/new/commonUrl?url=disclosure/list/notice" } all_items = [] page = 1 while True: payload = { "pageNum": page, "pageSize": 30, "column": "szse", "tabName": "fulltext", "searchkey": "", "secid": "", "stock": f"{stock_code},{org_id}", "category": category, "seDate": se_date, "sortName": "", "sortType": "", "isHLtitle": "true" } try: resp = requests.post(url, headers=headers, data=payload, timeout=15) data = resp.json() except Exception as e: print(f"第 {page} 页请求失败: {e}") break items = data.get("announcements") if not items: break all_items.extend(items) print(f"已拉取第 {page} 页,累计 {len(all_items)} 条") page += 1 time.sleep(0.5) # 控制频率,避免触发限流 return all_items这里time.sleep(0.5)是血泪经验。巨潮接口对高频请求有软限制,连续快速请求几十页后可能返回空数据或 403。加 0.5 秒间隔,几百页也就多花几分钟,但稳定性大幅提升。如果请求失败,不要立刻重试,先 break 出来,记录页码,下次从该页继续。
3.2 PDF 下载与文件名清洗
拿到adjunctUrl后,拼上http://static.cninfo.com.cn/就是完整下载地址。文件名建议用“股票代码_股票名称_公告标题_日期.pdf”的格式,方便后续检索。但公告标题里常含有/、\、:、*、?、"、<、>、|这些在 Windows 下非法的字符,必须替换掉。下面是一个下载函数:
import os import re def download_pdf(item, save_dir="./annual_reports"): os.makedirs(save_dir, exist_ok=True) sec_code = item.get("secCode", "unknown") sec_name = item.get("secName", "unknown") title = item.get("announcementTitle", "untitled") adj_url = item.get("adjunctUrl", "") if not adj_url: return None pdf_url = f"http://static.cninfo.com.cn/{adj_url}" # 清洗文件名中的非法字符 safe_title = re.sub(r'[\\/:*?"<>|]', "_", title) filename = f"{sec_code}_{sec_name}_{safe_title}.pdf" filepath = os.path.join(save_dir, filename) if os.path.exists(filepath): print(f"已存在,跳过: {filename}") return filepath try: r = requests.get(pdf_url, headers={"User-Agent": "Mozilla/5.0"}, timeout=30, stream=True) with open(filepath, "wb") as f: for chunk in r.iter_content(chunk_size=8192): f.write(chunk) print(f"下载完成: {filename}") return filepath except Exception as e: print(f"下载失败 {filename}: {e}") return Nonestream=True配合iter_content是下载大文件的标准做法,避免一次性把整个 PDF 读进内存。os.path.exists判断是断点续传的简易实现:已经下载过的文件直接跳过,脚本中断后重跑不会重复下载。文件名清洗用正则替换,把所有非法字符统一换成下划线,简单有效。注意adjunctUrl有时以/开头,有时不带,拼接时统一处理成http://static.cninfo.com.cn/加路径即可。
3.3 组织代码的获取与股票列表维护
stock参数里的组织代码不是股票代码本身,而是巨潮内部的orgId。获取方式有两种:一是先用股票代码调一次搜索接口,从返回结果里取orgId;二是维护一张股票代码到组织代码的映射表。对于批量任务,建议先写一个小脚本把所有目标股票的组织代码拉下来存成 JSON,后续下载直接读。下面是从搜索结果提取orgId的片段:
def get_org_id(stock_code): url = "http://www.cninfo.com.cn/new/information/topSearch/query" headers = {"User-Agent": "Mozilla/5.0", "Content-Type": "application/x-www-form-urlencoded"} payload = {"keyWord": stock_code, "maxNum": 10} resp = requests.post(url, headers=headers, data=payload, timeout=10) for item in resp.json(): if item.get("code") == stock_code: return item.get("orgId") return None这个搜索接口返回的每条记录里,code是股票代码,orgId是组织代码。拿到后拼成f"{stock_code},{org_id}"传给下载接口。如果股票列表有几百只,建议把这一步的结果缓存到本地文件,避免每次下载都重新查。
4. 避坑与排查:下载年报时最容易翻车的五个地方
4.1 请求返回空数组但页面明明有数据
现象:脚本请求接口,announcements返回空列表,但浏览器里能看到公告。原因通常是category参数不对,或者seDate格式写错。巨潮的日期格式必须是YYYY-MM-DD~YYYY-MM-DD,中间用波浪号连接,两边不能有空格。另外,沪市和深市的column参数不同,深市用szse,沪市用sse,如果股票是沪市的却传了szse,也会返回空。解决方法是先用浏览器抓一次真实请求,把参数原样抄下来,再逐个替换变量。
4.2 下载的 PDF 打不开或只有几 KB
现象:文件下载下来了,但打开提示损坏,或者大小只有几 KB。原因一般是adjunctUrl拼接错误,或者请求 PDF 时被重定向到了登录页。巨潮的静态资源域名是static.cninfo.com.cn,如果拼成了www.cninfo.com.cn,会返回一个 HTML 页面而不是 PDF。解决方法是打印出完整 URL,在浏览器里直接访问验证。另外,部分公告的 PDF 需要登录才能下载,但年报通常不需要,如果遇到 302 跳转,检查headers里是否缺少User-Agent。
4.3 文件名乱码或保存失败
现象:下载下来的文件名是一串乱码,或者保存时报错OSError: [Errno 22] Invalid argument。原因是公告标题里含有 Windows 不允许的字符,或者标题过长超过了文件系统限制。解决方法是下载前用正则替换非法字符,并且截断文件名到 200 个字符以内。另外,如果股票名称里有生僻字,某些文件系统可能不支持,建议统一用 UTF-8 编码保存,并在文件名里保留股票代码作为主键。
4.4 请求频率过高被限流
现象:脚本跑了几十页后,接口开始返回 403 或空数据,过一段时间又恢复。原因是请求间隔太短,触发了巨潮的限流机制。解决方法是在每次请求后加time.sleep(0.5)到1秒的延迟,并且不要并发请求。如果股票数量多,可以分批次跑,比如每跑 50 只股票就休息几分钟。另外,pageSize不要设得太大,30 到 50 比较稳妥,设成 100 容易触发限制。
4.5 断点续传与重复下载
现象:脚本跑到一半中断,重跑时已经下载过的文件又下了一遍,浪费时间。原因是没有做已存在文件的判断。解决方法是在下载函数里加os.path.exists检查,文件存在就跳过。更进一步,可以把已下载的股票代码和公告 ID 记录到一个 JSON 文件里,每次启动时先读取,避免重复请求接口。公告 ID 可以用announcementId字段,它是每条公告的唯一标识。
5. 进阶技巧:把下载脚本变成可复用的数据管道
5.1 用配置文件管理股票列表和日期范围
硬编码股票代码和日期范围不利于复用。我一般会建一个config.json,把股票列表、日期范围、保存目录、请求间隔都放进去:
{ "stocks": [ {"code": "000001", "org_id": "gssz0000001"}, {"code": "600000", "org_id": "gssh0600000"} ], "se_date": "2024-01-01~2024-12-31", "category": "category_ndbg_szsh", "save_dir": "./annual_reports", "sleep_seconds": 0.5 }脚本启动时读取这个文件,遍历stocks列表,逐个调用下载函数。这样换一批股票只需要改配置,不用动代码。org_id可以提前用搜索接口批量拉好,也可以让脚本在运行时自动获取并缓存。
5.2 下载日志与失败重试
批量任务一定要有日志,否则出了问题不知道哪只股票漏了。建议用 Python 的logging模块,把每次请求的页码、返回条数、下载成功与否都记到文件里。对于下载失败的条目,单独存一个failed.json,脚本结束时打印失败列表,下次可以只重试这些。下面是一个简单的日志配置:
import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler("download.log", encoding="utf-8"), logging.StreamHandler() ] )FileHandler把日志写到文件,StreamHandler同时输出到控制台。encoding="utf-8"避免中文日志乱码。每次下载成功或失败都调logging.info或logging.error,排查时直接翻日志文件,比在终端里翻滚动记录高效得多。
5.3 验证下载完整性的两个方法
下载完成后,怎么确认 PDF 是完整的?第一个方法是检查文件大小,年报 PDF 通常几百 KB 到几 MB,如果某个文件只有几 KB,大概率是下载失败或下到了错误页面。第二个方法是检查 PDF 文件头,正常的 PDF 文件前几个字节是%PDF-,可以用 Python 快速校验:
def is_valid_pdf(filepath): try: with open(filepath, "rb") as f: header = f.read(5) return header == b"%PDF-" except Exception: return False遍历下载目录,对每个文件调一次这个函数,把不合法的文件列出来重新下载。这个检查花不了几秒钟,但能避免后续用数据时才发现文件损坏。我一般会在批量下载结束后跑一遍这个校验,把无效文件删掉并记录到失败列表。
5.4 把下载和解析串起来
下载只是第一步,年报 PDF 里的财务数据才是最终目标。如果后续要解析 PDF 提取表格,建议在下载时就按“股票代码/年份”的目录结构存放,比如./annual_reports/000001/2024/000001_平安银行_2024年年度报告.pdf。这样解析脚本可以按目录遍历,不用再从文件名里拆股票代码和年份。目录结构在下载函数里加两行就能实现:
year = item.get("announcementTime", "")[:4] stock_dir = os.path.join(save_dir, sec_code, year) os.makedirs(stock_dir, exist_ok=True) filepath = os.path.join(stock_dir, filename)announcementTime是时间戳,转成日期后取前四位就是年份。按股票代码和年份分目录,后续做数据对齐时非常省事。这个习惯是我踩过坑之后养成的:早期把所有 PDF 堆在一个目录里,几千个文件后找一只股票的年报要翻半天,后来改成按目录分层,效率完全不一样。
5.5 一个容易忽略的细节:公告标题里的年份
有些公司的年报标题写的是“2024年年度报告”,有些写的是“2023年年度报告”,但披露日期是 2024 年。如果你用seDate筛选 2024 年的公告,拿到的是 2024 年披露的所有年报,其中可能包含 2023 年度的报告。所以文件名里的年份应该从公告标题里提取,而不是从披露日期里取。用正则从标题里匹配20\d{2}更准确:
import re match = re.search(r"20\d{2}", title) year = match.group() if match else "unknown"这个细节在批量下载时特别重要,否则你会把 2023 年年报和 2024 年年报混在一起,后续做同比分析时数据全乱。我一般会在下载日志里同时记录披露日期和标题年份,方便核对。
希望帮到你。
本文还有配套的精品资源,点击获取