简介:面向需要批量获取雪球内容的 Python 学习者,这份资料实现了登录雪球并抓取收藏文章的全流程脚本,可自动生成文本文件,适合研究模拟登录、会话维持与网页解析等反爬场景。压缩包共 31 个文件、约 32KB,其中 27 个 Python 脚本覆盖爬虫入口、数据处理、热门股票统计、PDF 生成与文件拆分等模块,2 个配置文件用于保存账号与爬虫设置,另有说明文档便于上手。脚本读取配置文件中的用户名和密码完成自动登录,运行主程序即可抓取收藏文章并输出文本,结构清晰,便于替换接口或扩展输出格式。此外还包含数据拆分与格式转换工具,方便生成电子书前做内容清洗与排版。已有 1339 人学习下载,对想独立完成登录、抓取、落地全流程的初学者和中级开发者都有参考价值。
1. 先解决登录再谈爬取:雪球文章抓取的关键是把请求会话变成“熟悉的浏览器”
刚开始抓雪球(xueqiu)文章的人,大概率会遇到一个尴尬的返回:明明网址没抄错,requests 也正常发了,结果拿回来的要么是未登录提醒,要么是一段空 JSON。原因其实很简单:雪球的文章内容接口大多依赖登录态,服务端认的“人”是请求里携带的 Cookie,而不是账号密码本身。只要会话里带着真实浏览器登录后的 Cookie,后面列文章、翻页、拿正文才能一路畅通。
这篇笔记按常见落地路径把整件事拆开讲:先搭一个带登录态的请求会话;再从浏览器里把真正返回文章列表的 JSON 接口找出来;然后写循环翻页的逻辑,把目标范围内的文章一条一条落库;最后专门讲那些最耗时间的坑,包括登录失效、请求头缺失、游标不会翻导致的重复数据。适合想用 Python 自己维护抓取逻辑、又不想引入重型爬虫框架,同时希望代码能长期迭代复用的人。开始之前提醒一句:只抓你自己账号能正常访问的内容,控制好请求频率,别把正常功能挤兑得没法用,这个边界自己心里要有数。
2. 登录会话的两种搭法:浏览器复制 Cookie 与自动化登录交接
抓取之前先解决身份问题。这里有个基础认知:requests 只是一个普通 HTTP 客户端,它本身不带登录信息,服务端靠 Cookie 认人。所以“登录爬取”的本质,是把浏览器登录成功后生成的 Cookie 放进 requests 的 Session 里,让后续请求都带着这个身份标识走。
2.1 方案一:浏览器复制 Cookie,十几分钟就能开始抓
最简单、也不容易被验证码卡住的方式,是从浏览器里直接复制登录后的 Cookie。操作步骤很直白:用你的账号正常登录雪球网页版,打开开发者工具,在 Application 面板里找到 Cookies,把登录相关的核心键值整理出来。更省事的办法是在 Network 面板里任选一个请求,右键 Copy as cURL,然后把 Cookie 那一行单独取出来用。
我一般会把 Cookie 存成 json 文件,这样 requests 启动时可以直接结构化注入,避免手工拼字符串时把引号和空格一起带进去:
import requests import json session = requests.Session() # cookie_file.json 内容形如 {"xq_a_token": "xxx", "u": "xxx", ...} # 键名以你浏览器里实际存在的为准,不同登录状态可能字段不一样 with open("cookie_file.json", "r", encoding="utf-8") as f: cookie_dict = json.load(f) for key, value in cookie_dict.items(): session.cookies.set(key, value) # 请求头尽量保持和真实浏览器一致,别只贴一个 URL session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://xueqiu.com/", "Accept": "application/json, text/plain, */*", })这段代码里值得多说的点有两个。使用session.cookies.set()逐项注入,比直接把 Cookie 拼到头里更可靠,因为 requests 会在后续请求里自动处理域名归属,减少手工拼错的概率。请求头里的Referer要和页面保持一致,不少接口会校验来源;User-Agent最好直接复制浏览器里的完整值,而不是填个通用的,否则服务端可能靠 UA 指纹把请求识别成脚本。
2.2 方案二:用自动化浏览器过登录,再交接 Cookie
如果你要做定时任务,比如每天早上自动抓一次,Cookie 经常过期这件事会让人很焦虑。常见做法是把登录环节交给自动化浏览器跑一次,让 selenium 之类的工具去处理加密参数和验证码,登录完成后把 Cookie 保存下来,再交接给 requests 使用。
import time import pickle from selenium import webdriver options = webdriver.ChromeOptions() options.add_argument("--start-maximized") driver = webdriver.Chrome(options=options) try: driver.get("https://xueqiu.com/") # 留出时间手动登录,扫码或输密码都可以 time.sleep(60) cookies = driver.get_cookies() with open("selenium_cookies.pkl", "wb") as f: pickle.dump(cookies, f) finally: driver.quit()这里的driver.get_cookies()返回的是一个列表,每个元素是包含name、value、domain等信息的字典。后续给 requests 用的时候,要把这个列表转成{name: value}的字典再注入,不能直接把整个列表塞给session.cookies.update()。
为什么我不建议直接对着登录接口做纯模拟登录?因为雪球的登录流程里密码传输有加密逻辑,验证码环节也比较复杂,纯 requests 模拟登录往往要维护一套加密代码,接口稍微升级就全部作废。让 selenium 只负责登录那一小步,后续抓取仍交给 requests,速度和稳定性都能兼顾。
2.3 登录态检测:任务启动前先确认自己没被踢下线
Cookie 会过期,尤其是连续高频请求之后,服务端经常主动把会话踢掉。所以在任务启动时先探一次登录态,比等到请求报错再去补救要省事得多。
def is_logged_in(session): # 找一个登录后才能正常返回列表的接口做探测 resp = session.get( "https://xueqiu.com/statuses/original/timeline.json", params={"user_id": 123456}, # 替换成任意真实用户ID timeout=10, ) if resp.status_code == 401: return False data = resp.json() # 真正返回内容时才有 list 字段;未登录时通常是提示或 error_code return isinstance(data.get("list"), list)我习惯把这段探测放在抓取任务的入口,检测失败就终止任务并发出提醒,避免带着失效的会话空跑几小时。注意探测请求本身也计入频率,不要每翻一页都调它,只在任务启动和运行一段时间后抽检即可。
3. 找出真正返回文章列表的 JSON 接口:抓包优先,别背 URL
登录有了,下一步是找到返回文章列表的接口。这里我踩过一个大坑:以前总想着把网页端文章列表的 URL 记住,以后直接 GET 就行,结果过几个月再看,路径变了、参数也变了,抓回来的数据全是空壳。后来改成浏览器抓包,这个问题就没有再犯过。
3.1 为什么别在网上找别人写死的接口地址
雪球前端和后端接口升级比较频繁,很多教程里给出的地址可能早就失效。更关键的是,不同登录状态下请求的参数都不一样,没登录时可能走的是一个简化接口,登录后才会返回带完整分页字段的版本。所以与其背死地址,不如学会看浏览器到底请求了哪个接口。
操作流程是这样的:登录网页版,按 F12 打开开发者工具,切到 Network 面板,勾选 Fetch/XHR;然后点击页面上文章流相关的按钮,比如“我的关注”或者“自选股”。这时能看到新出现的请求,找到返回内容里包含文章标题的那个 JSON 请求,右键复制成 cURL 命令保存起来。之后要做的就是把 cURL 里的每个 Header 和参数原封不动搬进 requests。
3.2 一次抓包要记录下来的四样东西
真正要把接口从“能看”变成“能跑”,只记 URL 是不够的。我每次接触一个新接口,都会先记四样东西:请求地址、必需的请求头、分页用的参数名、返回 JSON 里的文章列表字段名。把这四样记清楚,后面写解析代码基本不会翻车。
下面是一份常见返回结构的字段说明,具体字段名还是要以你抓包时看到的实际 JSON 为准:
| 字段 | 含义 | 说明 |
|---|---|---|
| id | 文章ID | 去重和排序用的主键 |
| title | 标题 | 有的文章没有标题 |
| description / text | 正文内容 | 通常带 HTML 标签,需要清洗 |
| created_at | 发布时间 | 通常是毫秒时间戳 |
| reply_count / comment_count | 评论数 | 按热度过滤时用 |
| user | 作者信息 | 嵌套的用户对象,包含昵称和 ID |
坦白说,不同时间段抓到的字段名会有差异,比如description可能写成text,comment_count可能写成reply_count。我的习惯是先把一份真实的 JSON 响应存到本地文件,再对着字段写解析代码,而不是直接照搬网络文章里的固定写法。
拿到响应后,先别急着写完整解析,在交互环境里打印第一条看看结构:
import json resp = session.get( "https://xueqiu.com/statuses/original/timeline.json", params={"user_id": 123456, "page": 1}, timeout=10, ) data = resp.json() items = data.get("list", []) print(json.dumps(items[0], ensure_ascii=False, indent=2))这一步能同时验证三件事:接口地址是否有效、登录态是否正常、字段名是否和预期一致。我一般把这条打印结果当作每次开发的起点,确认无误后再继续写循环逻辑。
3.3 请求头为什么那么重要:少一个字段都可能变 400
这类接口的反爬并不算复杂,但很较真。少了Referer,接口可能直接拒绝;少了UA,请求会被当成低版本客户端;Cookie 不完整,返回的就是未登录提示。所以写请求时,我会把 cURL 里的 Header 逐项转换到 requests 中,而不是只贴一个 URL 了事。
还有一个快速验证的办法:先把浏览器复制好的 cURL 命令放在命令行里跑一次,如果能正常返回 JSON,再转成 requests 写法。这样可以把“接口本身的问题”和“代码写法的问题”分开排查,省掉很多不必要的玄学调试时间。
4. 循环翻页抓取“所有”文章:游标、节流与断点落库
拿到列表接口之后,真正定义“所有文章”的其实是翻页逻辑。很多人以为翻页就是让page从 1 加到 100,实际没那么简单。服务端通常基于一个游标值返回下一页,游标不对,翻到一半就会看到数据重复或者直接空列表。
4.1 翻页参数的含义:max_id 与 page 的区别
第一次抓包时我看到参数里既有page也有max_id,一度以为两个都要传。后来实测发现,多数列表接口是以max_id作为游标——你把上一页返回的最后一条文章 ID 传回去,服务端就从这一条之后继续给;page更像一个辅助计数,逻辑上替代不了游标。如果只改page而不维护max_id,服务端很容易把已经返回过的内容再给一遍。
import time import requests def fetch_all_articles(base_url, params_init, session, interval=2): articles = [] cursor = -1 # 游标起始值以抓包实际参数为准 page = 0 while True: page += 1 params = dict(params_init) params["max_id"] = cursor params["page"] = page resp = session.get(base_url, params=params, timeout=10) if resp.status_code != 200: break payload = resp.json() items = payload.get("list") or [] if not items: break articles.extend(items) # 取出下一页游标,字段名以抓包 JSON 为准 next_cursor = payload.get("next_max_id") if next_cursor is None or next_cursor == cursor: break cursor = next_cursor time.sleep(interval) return articles这段代码有几个关键落点。cursor必须在每次循环结束前更新,否则第二次循环会拿同一个游标去请求,得到重复数据。next_cursor == cursor是防止服务端一直返回同一个游标导致死循环的保险。interval控制请求间隔,我一般从 2 秒起步,先跑一批数据观察状态码,再往下调节奏。
4.2 把超时、重试和限流处理统一收口
节流看起来简单,但真正稳定跑下去,靠的是把超时和重试也做进同一套逻辑。如果只写死一个time.sleep(2),遇到网络抖动或者被限流,程序会在同一个错误上反复重试,最后跑出来的数据缺一截。
import time import requests from requests.adapters import HTTPAdapter session = requests.Session() adapter = HTTPAdapter( max_retries=3, pool_connections=5, pool_maxsize=5, ) session.mount("https://", adapter) def request_with_backoff(session, method, url, **kwargs): for attempt in range(3): try: resp = session.request(method, url, **kwargs) if resp.status_code in (200, 304): return resp # 429 限流和 5xx 都值得等一下再重试 if resp.status_code in (429,) or resp.status_code >= 500: time.sleep(2 * (attempt + 1)) continue return resp except requests.RequestException: time.sleep(2 * (attempt + 1)) continue return NoneHTTPAdapter里的max_retries=3只负责连接层的异常,HTTP 状态码这一层还是得自己判断。我会把request_with_backoff作为所有请求的统一入口,后续如果想调整重试策略或者换个请求模块,只改这一处就够了。
4.3 从单用户到“所有文章”:把文章列表落成 JSONL 再逐步扩大范围
标题里的“所有文章”其实要分两层理解。如果关心的是某个用户的历史文章,上面的循环完全覆盖;如果想要的是全站范围的文章,就得换思路:先抓关注流或热帖流列表拿到大量文章 ID,再按 ID 查详情,同时做去重。
全站抓取会面临更大的频率压力,两个问题绕不开:一是不能靠单账号高频并发,往往需要多账号轮换;二是文章量级增长快,存储方案要提前规划。我倾向于先把目标收敛到明确范围,比如某个用户、某个话题、或者某个时间段,跑通流程之后,再逐步扩大。
抓到的数据先落成 JSONL,每条一行 JSON,方便追加和断点续读:
import json def append_to_jsonl(path, articles): with open(path, "a", encoding="utf-8") as f: for item in articles: f.write(json.dumps(item, ensure_ascii=False) + "\n") # 每次翻页结束后增量写入,避免攒到最后内存里堆太多 append_to_jsonl("articles.jsonl", batch_items)边抓边写文件还有个好处:程序中途崩了,已经从文件里读过的那部分数据不必重新抓。这个习惯对长耗时任务特别重要。
5. 登录态爬取避坑手册:5 个真实会遇到的翻车现场
流程能跑通之后,真正折磨人的是那些看着奇怪的小问题。下面几类问题几乎每次都会遇到,我按“现象→原因→解决”的顺序记录下来,方便你对照排查。
5.1 坑:Cookie 明明复制了,几分钟后又提示未登录
现象:刚登录完抓前几页正常,过一会儿突然全部返回未登录。 原因:复制 Cookie 时漏了关键项,或者请求头里的 UA 和浏览器实际不一致。服务端发现请求特征和 Cookie 对应的浏览器对不上号,就会判定会话异常。 解决:用浏览器的 Copy as cURL 把整条请求复制下来,再逐项转换成 requests 的 headers,不要只复制 Cookie 一行,UA 和 Referer 必须一起带上。改完后再用登录态探测函数验证一次。
5.2 坑:接口返回 400,却看不出任何语法错误
现象:URL 看起来没问题,参数也传了,但服务端直接返回 400。 原因:多数情况是缺了请求头里某个自定义字段。这类接口对来源校验很严,不是只查 Cookie 就放行。 解决:把浏览器里原始请求的所有 Header 都搬过来,包括看起来没用的Accept、Sec-Fetch-*等字段,逐一落进 requests 的 headers 里,再逐个删除测试哪些是必需项,最后保留最小可用集合。
5.3 坑:翻页翻到最后,又出现了第一页的数据
现象:明明已经用max_id翻页,数据还是有大量重复。 原因:服务端同时按page和max_id定位,代码里没有正确维护游标,导致下一页请求用的还是上一页的旧值。 解决:把游标更新放到循环结尾,用响应里的next_max_id覆盖旧游标;请求参数里不要同时硬编码page和max_id,让其中一个作为主导。抓完对id做一次去重兜底。
5.4 坑:抓了十分钟后接口返回“操作太频繁”
现象:一开始一切正常,一段时间后返回 429 或者类似的限制错误码。 原因:单账号请求频率超过阈值,服务端对账号做了临时限制。 解决:降低请求频率,把每次请求间隔提高到 5 秒以上;多账号轮番用的时候,要保证同一账号的两次请求间隔稳定,不要并发。遇到限制错误立刻停止该账号的抓取,等一段时间再继续。
5.5 坑:存进数据库的文章正文是一堆转义符号
现象:JSON 里正文没问题,落库后变成\u003c这种转义文本。 原因:接口返回的内容做了 Unicode 转义,或者正文本身是 HTML 片段,直接存字符串没有做反转义。 解决:解析时用json.loads的默认行为先还原,再对正文里的 HTML 标签做简单清理;需要保留样式的,就把整个 HTML 片段转成 Markdown。测试用例里专门留一条含特殊字符的文章,验证显示效果。
6. 让抓取长期稳定运行的三个习惯:自动续期、增量去重、小范围验证
前面五章解决的是“能跑”,最后一章补上真正意义上的“长期跑”。我在维护类似任务时最深的教训是:爬虫出问题一般不是代码炸了,而是“没炸但数据少了”或者“没炸但重复了”。长期任务的核心,就是围绕这两个问题做防御。
先在任务入口做登录态巡检,检测失败就触发一次自动化登录续期,把新 Cookie 写回 Cookie 文件。代码结构上,我会把登录模块和抓取模块拆开,抓取模块只认 Cookie 文件,续期模块只负责生成 Cookie 文件。这样以后想换登录方式,完全不用动抓取代码。
增量去重的策略可以很朴素:以文章id作为主键做集合去重。每次抓完把本批次的id写进一个去重文件,下次任务启动时先加载进来,从已抓的最后一条游标开始续跑。这个方案比每次全量抓再对比要快得多,也不会因为某一次中断导致整个任务前功尽弃。
import json from pathlib import Path def load_seen_ids(path: Path): if not path.exists(): return set() return set(json.loads(path.read_text(encoding="utf-8"))) def save_seen_ids(path: Path, seen_ids): path.write_text(json.dumps(sorted(seen_ids)), encoding="utf-8")最后留一个习惯:不要在没把控频率的情况下直接上全站抓取。我刚开始做类似任务时,被“一口气全量拉下来”的想法带偏过,结果账号被限制,连基础浏览都受影响。后来老老实实把范围切成小目标,单用户、单话题、单时间段逐一跑通,每一段都验证完再合并。这个习惯后来救了我很多次。
如果你准备按这个方案动手,建议先从复制 Cookie 开始,把一章的抓包步骤走一遍,确认接口字段没问题,再写循环翻页。等这两段都稳定了,再考虑自动化登录续期和增量去重。希望这篇笔记能帮你把登录态、接口定位、分页循环这几个关键环节一次搭顺,少走几圈弯路。
本文还有配套的精品资源,点击获取