简介面向Python爬虫开发者和雪球内容研究者以自动登录雪球并抓取个人收藏文章为核心功能解决长期积累社区内容时只能逐篇复制保存的繁琐问题。包体共31个文件、体积约32KB其中27个Python脚本构成主要逻辑另有cfg配置文件用于账号参数设置以及Markdown文档辅助说明主抓取脚本读取配置中的用户与密码后即可完成登录并输出TXT文本配套的策略分析与行情数据模块可扩展数据处理还有脚本便于对生成的大文件进行拆分。目前已有1339人学习下载适合初中级爬虫学习者作为登录模拟、会话维护和批量导出的综合样例。压缩包内含完整的账号配置模板、清晰的脚本分工说明和可直接运行的抓取入口结合说明文档能快速定位各模块用途稍加改造便可适配收藏夹管理、文章备份等更多个性化场景是一套小巧但实用的雪球数据采集工具集合。1. 为什么“登录后”爬雪球才是真正的工程起点网上能搜到的雪球爬虫教程大多扔给你几行代码抓回来几十条数据就收工。这类 Demo 最大的问题是没登录未登录会话拿到的文章流并不完整很多用户的置顶长文和特定分类的讨论在匿名状态下根本不会出现在结果里抓完你也不知道少了什么。标题里的“登录并爬取所有文章”其实在讲一个工程问题如何用一个可信的登录态把雪球的公开文章做成一份可以持续爬、断了能续、日积月累的本地数据资产。它解决的痛点是“今天跑通明天还在”而不是“跑一次就扔”。这篇文章适合有爬虫基础、想把这套链路搭成长期服务的开发者照着做能把账号登录、全量采集、增量更新和防封节奏一次串起来。2. 登录这关Cookie、会话与验证码兜底很多人一上来就在代码里硬编码一个 Cookie 字符串过期了就去浏览器里复制一份新的。这种做法撑不过三天服务端会话的有效期是动态的有的账号活跃度高会话能活两周有的隔夜就失效。真正的做法是让程序自己维护登录态用 requests.Session 模拟登录把服务端下发的 Cookie 序列化到本地文件每次启动先检测会话有效性失效再重新走登录流程。这套思路是给后续所有接口请求打地基下面按步骤拆开讲。2.1 先看清登录态要带哪些字段网络面板里的三个关键点第一次做千万别急着写代码先打开浏览器登录一次在开发者工具的网络面板里找三条信息。第一是 Cookie 里哪些字段是登录后新增的比如会话令牌、用户编号这类名称第二是请求头里有没有自定义的鉴权字段雪球这类平台经常单独放一个 header 而不是把所有身份塞进 Cookie第三是当前业务接口的完整路径列表、详情、用户信息分别走哪个地址。把这三样记录到本地备忘里后面的代码才有的放矢。常见误用是只盯着“登录接口响应里的 token”以为拿到 token 就万事大吉。实际上雪球的接口都是整套 Cookie 配合的单拿一个 token 拼不出可用请求。还有的人从浏览器复制所有请求头连 Sec-Fetch 这类浏览器专属字段也带进来爬虫代码和浏览器环境对不上请求直接失败。正确做法是先小规模试两个接口确认复制过来的完整 Header 能服务化复用再开始封装 Session。2.2 账号密码登录用 Session 保存 Cookie 的最小实现账号密码登录是最常规的入口也是手机上最容易触发验证码的方式。写代码时有一个关键动作全程用同一个 Session 对象登录成功后的 Cookie 会被会话自动带上不需要每次手动拼接。核心实现如下import requests import pickle from pathlib import Path session requests.Session() # 模拟登录接口实际路径请在浏览器网络面板里确认 login_url https://api.example.com/account/login headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Referer: https://xueqiu.com/, Origin: https://xueqiu.com, } payload { username: 你的账号, password: 你的密码, remember: 1, } try: resp session.post(login_url, datapayload, headersheaders, timeout10) # 很多站点业务失败也是200必须检查返回体里的业务字段 if resp.json().get(ok) is not True: print(登录失败, resp.json().get(msg)) else: # 把整个cookie jar持久化到本地文件下次直接复用 Path(cookies.pkl).write_bytes(pickle.dumps(session.cookies)) print(登录成功Cookie 已保存) except Exception as e: print(请求异常, e)这段代码里有两个值得注意的地方。一是 timeout 务必设置不要写 0 或省略雪球的接口高峰期慢起来能拖到几十秒没有超时控制整个任务会被一个请求卡死。二是 pickle 序列化的是整个 cookie jar里面包含多个域名下的不同字段比手动拼接 Cookie 字符串靠谱得多因为 Session 会根据请求 URL 自动挑出对应的 Cookie 字段不用你关心作用域问题。header 里的 Referer 和 Origin 别乱删。有的接口会校验来源来源不对直接拒绝有的服务在反爬策略里专门盯缺失来源的请求。如果你照抄上面的结构登录失败先去确认这两个字段的值是否和浏览器里登录时一致尤其 Origin 里带的是完整站点名还是二级域名差一个字符都会影响结果。2.3 验证码兜底扫码登录让账号密码不再成为单点故障账号密码登录最烦人的是验证码登录频率一高或者网络环境有异常服务端就会弹验证码。别急着找打码平台对个人维护的采集脚本来说扫码登录是更省事的兜底方案。它的流程是请求一个二维码把二维码图片展示出来手机扫码确认之后轮询接口确认状态趁会话有效时把 Cookie 存下来。核心轮询逻辑如下import requests import time import qrcode import pickle from pathlib import Path session requests.Session() base_url https://api.example.com # 1. 获取二维码图片保存到本地 qr_init session.get(base_url /qr/init, timeout10).json() qr_code qr_init.get(qr_code) qrcode.make(qr_code).save(login_qr.png) # 2. 轮询扫码状态确认后保存 Cookie while True: status session.get( base_url /qr/status, params{code: qr_code}, timeout10 ).json() if status.get(state) confirmed: Path(cookies.pkl).write_bytes(pickle.dumps(session.cookies)) print(扫码登录完成Cookie 已保存) break elif status.get(state) expired: print(二维码已过期请重新生成) break time.sleep(2)轮询间隔不要小于 2 秒扫码确认到服务端状态更新之间有延迟太频繁容易触发频率限制。二维码过期是正常现象日志里打印一句提示然后退出即可不要在同一会话里无限重试生成新码。实际接口路径可能和这里略有差异字段名也可能不同你需要对着网络面板改一遍。登录成功后马上做一次轻量健康检查用一个信息查询接口验证 Cookie 是否可用确认没问题再入库。否则后续业务接口跑一半才发现会话失效排查起来会多花很多时间。3. “所有文章”怎么做列表、详情与增量更新标题里的“所有文章”听起来唬人落地时得先定义边界。物理意义上的全站所有文章不存在工程上做的是“自某起始时间点之后的全量加上后续增量”。这意味着要分两层抓第一层抓文章流的 ID 列表拿到文章编号、作者、发布时间这些元数据第二层再按 ID 抓详情正文。两层分开的好处是列表接口体积小、速度快可以先做一次广度扫描把 ID 全部落库详情层按需慢慢抓哪条失败就重试哪条不用整篇重跑。3.1 文章流接口先把 ID 列表批量拿下来文章流是所有后续数据的地基。抓它的核心是翻页参数页码、每页数量、排序方式、时间窗口。雪球这类社区一般会提供两种流全部动态流和长文流你要根据目标决定用哪一种。下面是翻页拉取 ID 的框架import requests SESSION requests.Session() # 登录后从 cookies.pkl 恢复会话 import pickle import time with open(cookies.pkl, rb) as f: SESSION.cookies.update(pickle.load(f)) def fetch_article_ids(page: int 1): # 示意接口实际路径在浏览器里确认后再改 url https://api.example.com/statuses/list params { page: page, size: 20, type: 0, # 0 全部1 长文按你的目标调 sort: time, # 按发布时间排序方便做增量 } headers {User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64)} resp SESSION.get(url, paramsparams, headersheaders, timeout10) data resp.json() if not data.get(ok): print(列表接口业务报错, data.get(message)) return [], False items data[data][items] ids [item[id] for item in items if item.get(id)] # 返回是否还有下一页方便 while 循环继续翻 has_next len(items) 20 return ids, has_next翻页逻辑别写死 for page in range(1, 999)要依据返回数量判断是否该停。常见反模式是页面无数据还继续递增 page直到请求到一个超大页码被网关断开。另外注意 size 别贪大很多接口对单页数量有隐藏上限你传 100 它也只给你 20还有可能直接报参数错误。先用 20 跑通再试 50逐步加大不要一上来拉满。排序参数选 time 是为了给增量更新留后路每次抓完记录当前最大发布时间下次从这个时间点往后扫能少爬很多无用数据。如果你的目标是某个时间区间可以在参数里加起止时间但多数接口只提供相对时间窗口这时就得靠任务表做过滤。3.2 详情正文入库MD5 指纹去重用 SQLite 做主库ID 列表拿到后接下来按 ID 逐个请求详情接口。详情响应里的正文字段格式比较杂可能是 HTML 片段、JSON 嵌套文本、带转义符的字符串入库前先做指纹去重避免同一篇文章重复落库。核心写入代码如下import hashlib import sqlite3 import time def get_article_detail(article_id): # 示意接口实际请抓包确认 url fhttps://api.example.com/statuses/show params {id: article_id} resp SESSION.get(url, paramsparams, timeout10) data resp.json() if not data.get(ok): raise ValueError(f详情失败: {article_id}, msg{data.get(message)}) body data[data][content] # MD5 指纹用于检测内容变化同一ID编辑后指纹会变 md5 hashlib.md5(body.encode(utf-8)).hexdigest() return { id: article_id, text: body, md5: md5, created_at: data[data].get(created_at, 0), author: data[data].get(user, {}).get(screen_name, ), } def upsert_article(conn, row): sql INSERT INTO article(id, text_md5, text, created_at, author, ts) VALUES(?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET text_md5excluded.text_md5, textexcluded.text, created_atexcluded.created_at conn.execute(sql, ( row[id], row[md5], row[text], row[created_at], row[author], time.time() )) conn.commit()这里有两个设计点值得你细看。一是入库字段里保留了 id 和 text_md5 两个指纹id 保证不重复插入同一篇文章text_md5 用于检测文章被编辑过的情况正文变了指纹就变你可以通过比对 md5 发现哪些文章被原始作者修改过。二是 ON CONFLICT 做的是更新而不是忽略有意为之同一个 id 下内容变了旧数据要被覆盖这样后续数据分析拿到的永远是较新版本。正文抽取别用正则去抠 HTML。有的正文里带 div 嵌套、换行符、图片链接正则写起来又脆又难维护。正确姿势是直接存原始返回字段需要展示或导出时再交给 BeautifulSoup 之类的解析库处理入库环节保持原始是最省力的选择。3.3 让“所有”能续下去任务表与断点续爬列表扫描和详情抓取分成两个环节之后任务表就成了连接两者的管道。每拿到一个新 ID就向任务表插入一条 pending 记录“待抓”状态的任务被消费时先查状态再看是否更新。这样程序崩溃、断网、被封都只需要把失败任务重置回 pending不需要从头翻列表。CREATE TABLE IF NOT EXISTS article_task ( id INTEGER PRIMARY KEY, status TEXT DEFAULT pending, retry INTEGER DEFAULT 0, updated_at INTEGER );def dispatch(conn, batch_ids): cur conn.cursor() for aid in batch_ids: cur.execute( INSERT OR IGNORE INTO article_task(id, status, updated_at) VALUES(?, pending, ?), (aid, int(time.time())) ) conn.commit() def run_once(conn, limit200): cur conn.cursor() cur.execute( SELECT id FROM article_task WHERE statuspending ORDER BY id ASC LIMIT ?, (limit,) ) tasks cur.fetchall() for (aid,) in tasks: try: row get_article_detail(aid) upsert_article(conn, row) cur.execute( UPDATE article_task SET statusok, updated_at? WHERE id?, (int(time.time()), aid) ) except Exception as e: cur.execute( UPDATE article_task SET statusfailed, retryretry1, updated_at? WHERE id?, (int(time.time()), aid) ) conn.commit() time.sleep(1)run_once 循环体的最后一行 sleep 是防封节奏的一部分别看它简单忘掉的后果是详情接口连续高频请求直接把会话打到冻结。任务表的另一大好处是支持失败任务重试你只需要写一个 SQL 把 failed 状态重置回 pending再跑一次 run_once 即可。如果某个任务 retry 超过 5 次还失败说明接口或文章结构有变化这时应把它单独挑出来人工排查而不是无限重试。任务表就是整个采集链路的后悔药。4. 限速与防封请求节奏写进代码里个人账号被临时限制绝大多数情况不是因为你抓的量大而是请求波形像机器。服务端常用的限频策略有分钟窗口和小时窗口两层统计同窗口内请求间隔均匀与否往往比总量更敏感。常见做法是两层保险一个全局限速器保证任意两秒内只有一次请求一个随机延时把节奏打散。别小看这两样没有它们再大的账号也熬不过一小时。4.1 一个简单的限速器随机延时与并发控制多线程场景下最怕各线程各自 sleep 然后同时发请求。用一个带锁的限速器统一管理下一次可发起请求的时间点是可控且简单的方案import threading import time import random class RateLimiter: def __init__(self, min_interval1.0, max_interval3.0): self.min_interval min_interval self.max_interval max_interval self._lock threading.Lock() self._next_at 0 def wait(self): with self._lock: now time.time() if now self._next_at: time.sleep(self._next_at - now) # 随机打散下一次请求的间隔避免固定节奏 gap random.uniform(self.min_interval, self.max_interval) self._next_at time.time() gap用法是在每次 HTTP 请求前调用 limiter.wait()比如列表接口设 min_interval2.0、max_interval4.0详情接口设 1.0 到 2.0。这里的参数不是玄学而是个人账号在合理频率区间的经验值低于这条线很快会触发验证。注意不要把 min_interval 设成 0哪怕 max 再大也没用服务端抓的就是最低间隔。如果你开多线程把同一个限速器实例传给所有线程。它的锁机制保证全局只有一个时间窗口不会出现线程 A 和线程 B 间隔很小的情况。这个设计要尽早做晚了再改造就麻烦很多。4.2 重试要区分限流和业务失败不要见错就重试重试逻辑写不好爬虫会陷入无效循环。最常见的错误是看到非 200 状态码就重试但雪球的业务失败往往隐藏在 200 响应里重试多少次都白搭。经验是分三类处理HTTP 429 说明触发限频要按 Retry-After 头等待业务字段 okfalse 且 message 提示操作频繁要指数退避登录态失效的错误则应该终止采集并触发重新登录。def request_with_retry(func, max_retry3, base_delay2.0): last_exc None for attempt in range(max_retry): try: resp func() if resp.status_code 429: # 按服务端给的等待时长休息多数是30秒到几分钟 wait_sec float(resp.headers.get(Retry-After, 30)) time.sleep(wait_sec) continue data resp.json() if data.get(ok): return data if data.get(message) in (操作频繁, 请稍后再试): time.sleep(base_delay * (2 ** attempt)) continue # 其他业务失败多半是参数或登录态问题不建议在这里重试 return data except Exception as e: last_exc e time.sleep(base_delay * (2 ** attempt)) raise RuntimeError(f最终失败: {last_exc})这段代码里有个细节业务消息里的操作频繁错误才退避重试其他业务错误直接返回。因为一个报错的参数重试一万次也报同样的错反而会白白消耗请求量。429 分支单独处理是因为服务端已经明确告诉你超限了这时候正常退避可能不够得按它给的时间等。日志里要记录清楚是哪类失败否则事后排查只能猜。4.3 会话体温做一次轻量健康检查登录态是会过期的与其等业务接口开始报错再回头处理不如每天正式任务前检查一次会话健康度def check_login_valid(): if not SESSION.cookies: return False resp SESSION.get(https://api.example.com/user/info, timeout10) data resp.json() if data.get(ok) and data[data].get(id): return True return False如果返回 False不要立刻重试先清空本地 Cookie 走扫码登录流程。健康检测接口选最轻量、返回字段最少的那一个别用文章详情当体温计白白消耗请求。健康检查加定时任务配合重登录逻辑这套链路就可以无人值守跑很久。5. 避坑爬雪球文章最常见的五个翻车现场5.1 翻车一昨天还好好的第二天一跑全是登录失效现象健康检查能过但业务接口跑几分钟后全返回登录失效的错误码。原因我排了半天发现是某个来源字段在请求中被服务端校验浏览器里登录时它带了完整信息而代码里复制的 Session 缺了这一项。解决重新抓包核对所有请求头字段登录成功后先跑三个不同类型接口做冒烟测试确认没问题再开始批量任务。5.2 翻车二列表能翻页详情却大量返回空正文现象ID 列表一切正常但按 ID 请求详情时 content 字段经常为空。原因文章流里混入了删帖和站内短消息这类对象在详情接口里没有正文另外未登录或会话过期时详情接口会返回空正文但状态码依然正常。解决详情返回空时先做一次会话健康检查排除登录态问题然后把这类 ID 单独标记为“无正文”不纳入文章主库。5.3 翻车三任务表状态设计失误一崩就全量重来现象程序崩溃后重启发现昨天跑过的数据又全部插入一遍。原因任务表里只有 pending 和 ok 两个状态程序崩溃时正在跑的任务还停留在 pending重启后又被重新调度。解决引入 processing 中间状态每次调度任务前先执行一条 SQL把 processing 且超过 5 分钟未更新的记录重置回 pending这样崩溃恢复不需要手动干预。5.4 翻车四频繁换 UA 反而更容易被限制现象请求头里随机切换不同浏览器的 UA第二天直接被要求验证。原因服务端把 UA 看作了身份的一部分同一会话短时间内 UA 跳变会触发异常标记。解决固定一个常用的 UA 字符串不要随机换会话失效时再考虑换新身份重新登录。5.5 翻车五正文直接落 Excel字段全乱现象导出到 Excel 后长文变成一堆带 tag 的字符串换行符也没了。原因正文本质是富文本截断的 tag 在表格里显示不出来。解决入库保持原始 HTML导出时用解析器把纯文本抽出来再做清洗。清洗逻辑独立放在导出阶段不要污染主库数据。6. 进阶把临时脚本装进定时任务做成能长期运转的小系统前面的代码拼起来已经能跑但离“长期运转”还差三件事定时调度、日志留痕、数据校验。定时调度用操作系统的 crontab 是最轻量的方案不需要额外引入任务队列中间件。每天凌晨三点跑一次批量任务白天再安排一次增量任务基本覆盖一天的文章更新节奏0 3 * * * cd /home/user/xueqiu /home/user/venv/bin/python main.py --task daily logs/crawl.log 21 0 13 * * * cd /home/user/xueqiu /home/user/venv/bin/python main.py --task daily logs/crawl.log 21日志直接追加到同一个文件里配合 python 内置的 logging 模块把每次任务的起始时间、成功条数、失败条数、异常详情都打出来。排错的时候最怕看到日志里只有一行“error”没有上下文所以在代码里至少打印当前任务 ID 和接口 URL定位会快很多。数据校验是常被人忽略的环节。我一般会在每天任务结束后加一步抽样校验从昨天入库的文章里随机抽 10 条拿它们的 ID 去列表接口比对看标题、时间、作者是否一致。不一致就说明解析逻辑有偏差需要立刻查。这一步不复杂但能及时发现接口变更避免脏数据堆积def sample_check(): cur conn.cursor() rows cur.execute( SELECT id FROM article WHERE created_at ? ORDER BY random() LIMIT 10, (int(time.time()) - 86400,) ).fetchall() for (aid,) in rows: detail get_article_detail(aid) if not detail or detail[id] ! aid: print(f校验失败: {aid})我自己的习惯是在每个采集脚本的主目录放一个 check.py只做健康检查和抽样校验这两件事全部通过才发通知。这个习惯帮我止损过很多次尤其是接口悄悄大版本更新的时候。校验逻辑别写太复杂只要能回答“数据有没有在正常增长、格式有没有变质”就够。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/12 19:14:39 · 阅读 1.2k 雪球全量文章爬虫：登录态管理与增量更新实战
简介面向Python爬虫开发者和雪球内容研究者以自动登录雪球并抓取个人收藏文章为核心功能解决长期积累社区内容时只能逐篇复制保存的繁琐问题。包体共31个文件、体积约32KB其中27个Python脚本构成主要逻辑另有cfg配置文件用于账号参数设置以及Markdown文档辅助说明主抓取脚本读取配置中的用户与密码后即可完成登录并输出TXT文本配套的策略分析与行情数据模块可扩展数据处理还有脚本便于对生成的大文件进行拆分。目前已有1339人学习下载适合初中级爬虫学习者作为登录模拟、会话维护和批量导出的综合样例。压缩包内含完整的账号配置模板、清晰的脚本分工说明和可直接运行的抓取入口结合说明文档能快速定位各模块用途稍加改造便可适配收藏夹管理、文章备份等更多个性化场景是一套小巧但实用的雪球数据采集工具集合。1. 为什么“登录后”爬雪球才是真正的工程起点网上能搜到的雪球爬虫教程大多扔给你几行代码抓回来几十条数据就收工。这类 Demo 最大的问题是没登录未登录会话拿到的文章流并不完整很多用户的置顶长文和特定分类的讨论在匿名状态下根本不会出现在结果里抓完你也不知道少了什么。标题里的“登录并爬取所有文章”其实在讲一个工程问题如何用一个可信的登录态把雪球的公开文章做成一份可以持续爬、断了能续、日积月累的本地数据资产。它解决的痛点是“今天跑通明天还在”而不是“跑一次就扔”。这篇文章适合有爬虫基础、想把这套链路搭成长期服务的开发者照着做能把账号登录、全量采集、增量更新和防封节奏一次串起来。2. 登录这关Cookie、会话与验证码兜底很多人一上来就在代码里硬编码一个 Cookie 字符串过期了就去浏览器里复制一份新的。这种做法撑不过三天服务端会话的有效期是动态的有的账号活跃度高会话能活两周有的隔夜就失效。真正的做法是让程序自己维护登录态用 requests.Session 模拟登录把服务端下发的 Cookie 序列化到本地文件每次启动先检测会话有效性失效再重新走登录流程。这套思路是给后续所有接口请求打地基下面按步骤拆开讲。2.1 先看清登录态要带哪些字段网络面板里的三个关键点第一次做千万别急着写代码先打开浏览器登录一次在开发者工具的网络面板里找三条信息。第一是 Cookie 里哪些字段是登录后新增的比如会话令牌、用户编号这类名称第二是请求头里有没有自定义的鉴权字段雪球这类平台经常单独放一个 header 而不是把所有身份塞进 Cookie第三是当前业务接口的完整路径列表、详情、用户信息分别走哪个地址。把这三样记录到本地备忘里后面的代码才有的放矢。常见误用是只盯着“登录接口响应里的 token”以为拿到 token 就万事大吉。实际上雪球的接口都是整套 Cookie 配合的单拿一个 token 拼不出可用请求。还有的人从浏览器复制所有请求头连 Sec-Fetch 这类浏览器专属字段也带进来爬虫代码和浏览器环境对不上请求直接失败。正确做法是先小规模试两个接口确认复制过来的完整 Header 能服务化复用再开始封装 Session。2.2 账号密码登录用 Session 保存 Cookie 的最小实现账号密码登录是最常规的入口也是手机上最容易触发验证码的方式。写代码时有一个关键动作全程用同一个 Session 对象登录成功后的 Cookie 会被会话自动带上不需要每次手动拼接。核心实现如下import requests import pickle from pathlib import Path session requests.Session() # 模拟登录接口实际路径请在浏览器网络面板里确认 login_url https://api.example.com/account/login headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Referer: https://xueqiu.com/, Origin: https://xueqiu.com, } payload { username: 你的账号, password: 你的密码, remember: 1, } try: resp session.post(login_url, datapayload, headersheaders, timeout10) # 很多站点业务失败也是200必须检查返回体里的业务字段 if resp.json().get(ok) is not True: print(登录失败, resp.json().get(msg)) else: # 把整个cookie jar持久化到本地文件下次直接复用 Path(cookies.pkl).write_bytes(pickle.dumps(session.cookies)) print(登录成功Cookie 已保存) except Exception as e: print(请求异常, e)这段代码里有两个值得注意的地方。一是 timeout 务必设置不要写 0 或省略雪球的接口高峰期慢起来能拖到几十秒没有超时控制整个任务会被一个请求卡死。二是 pickle 序列化的是整个 cookie jar里面包含多个域名下的不同字段比手动拼接 Cookie 字符串靠谱得多因为 Session 会根据请求 URL 自动挑出对应的 Cookie 字段不用你关心作用域问题。header 里的 Referer 和 Origin 别乱删。有的接口会校验来源来源不对直接拒绝有的服务在反爬策略里专门盯缺失来源的请求。如果你照抄上面的结构登录失败先去确认这两个字段的值是否和浏览器里登录时一致尤其 Origin 里带的是完整站点名还是二级域名差一个字符都会影响结果。2.3 验证码兜底扫码登录让账号密码不再成为单点故障账号密码登录最烦人的是验证码登录频率一高或者网络环境有异常服务端就会弹验证码。别急着找打码平台对个人维护的采集脚本来说扫码登录是更省事的兜底方案。它的流程是请求一个二维码把二维码图片展示出来手机扫码确认之后轮询接口确认状态趁会话有效时把 Cookie 存下来。核心轮询逻辑如下import requests import time import qrcode import pickle from pathlib import Path session requests.Session() base_url https://api.example.com # 1. 获取二维码图片保存到本地 qr_init session.get(base_url /qr/init, timeout10).json() qr_code qr_init.get(qr_code) qrcode.make(qr_code).save(login_qr.png) # 2. 轮询扫码状态确认后保存 Cookie while True: status session.get( base_url /qr/status, params{code: qr_code}, timeout10 ).json() if status.get(state) confirmed: Path(cookies.pkl).write_bytes(pickle.dumps(session.cookies)) print(扫码登录完成Cookie 已保存) break elif status.get(state) expired: print(二维码已过期请重新生成) break time.sleep(2)轮询间隔不要小于 2 秒扫码确认到服务端状态更新之间有延迟太频繁容易触发频率限制。二维码过期是正常现象日志里打印一句提示然后退出即可不要在同一会话里无限重试生成新码。实际接口路径可能和这里略有差异字段名也可能不同你需要对着网络面板改一遍。登录成功后马上做一次轻量健康检查用一个信息查询接口验证 Cookie 是否可用确认没问题再入库。否则后续业务接口跑一半才发现会话失效排查起来会多花很多时间。3. “所有文章”怎么做列表、详情与增量更新标题里的“所有文章”听起来唬人落地时得先定义边界。物理意义上的全站所有文章不存在工程上做的是“自某起始时间点之后的全量加上后续增量”。这意味着要分两层抓第一层抓文章流的 ID 列表拿到文章编号、作者、发布时间这些元数据第二层再按 ID 抓详情正文。两层分开的好处是列表接口体积小、速度快可以先做一次广度扫描把 ID 全部落库详情层按需慢慢抓哪条失败就重试哪条不用整篇重跑。3.1 文章流接口先把 ID 列表批量拿下来文章流是所有后续数据的地基。抓它的核心是翻页参数页码、每页数量、排序方式、时间窗口。雪球这类社区一般会提供两种流全部动态流和长文流你要根据目标决定用哪一种。下面是翻页拉取 ID 的框架import requests SESSION requests.Session() # 登录后从 cookies.pkl 恢复会话 import pickle import time with open(cookies.pkl, rb) as f: SESSION.cookies.update(pickle.load(f)) def fetch_article_ids(page: int 1): # 示意接口实际路径在浏览器里确认后再改 url https://api.example.com/statuses/list params { page: page, size: 20, type: 0, # 0 全部1 长文按你的目标调 sort: time, # 按发布时间排序方便做增量 } headers {User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64)} resp SESSION.get(url, paramsparams, headersheaders, timeout10) data resp.json() if not data.get(ok): print(列表接口业务报错, data.get(message)) return [], False items data[data][items] ids [item[id] for item in items if item.get(id)] # 返回是否还有下一页方便 while 循环继续翻 has_next len(items) 20 return ids, has_next翻页逻辑别写死 for page in range(1, 999)要依据返回数量判断是否该停。常见反模式是页面无数据还继续递增 page直到请求到一个超大页码被网关断开。另外注意 size 别贪大很多接口对单页数量有隐藏上限你传 100 它也只给你 20还有可能直接报参数错误。先用 20 跑通再试 50逐步加大不要一上来拉满。排序参数选 time 是为了给增量更新留后路每次抓完记录当前最大发布时间下次从这个时间点往后扫能少爬很多无用数据。如果你的目标是某个时间区间可以在参数里加起止时间但多数接口只提供相对时间窗口这时就得靠任务表做过滤。3.2 详情正文入库MD5 指纹去重用 SQLite 做主库ID 列表拿到后接下来按 ID 逐个请求详情接口。详情响应里的正文字段格式比较杂可能是 HTML 片段、JSON 嵌套文本、带转义符的字符串入库前先做指纹去重避免同一篇文章重复落库。核心写入代码如下import hashlib import sqlite3 import time def get_article_detail(article_id): # 示意接口实际请抓包确认 url fhttps://api.example.com/statuses/show params {id: article_id} resp SESSION.get(url, paramsparams, timeout10) data resp.json() if not data.get(ok): raise ValueError(f详情失败: {article_id}, msg{data.get(message)}) body data[data][content] # MD5 指纹用于检测内容变化同一ID编辑后指纹会变 md5 hashlib.md5(body.encode(utf-8)).hexdigest() return { id: article_id, text: body, md5: md5, created_at: data[data].get(created_at, 0), author: data[data].get(user, {}).get(screen_name, ), } def upsert_article(conn, row): sql INSERT INTO article(id, text_md5, text, created_at, author, ts) VALUES(?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET text_md5excluded.text_md5, textexcluded.text, created_atexcluded.created_at conn.execute(sql, ( row[id], row[md5], row[text], row[created_at], row[author], time.time() )) conn.commit()这里有两个设计点值得你细看。一是入库字段里保留了 id 和 text_md5 两个指纹id 保证不重复插入同一篇文章text_md5 用于检测文章被编辑过的情况正文变了指纹就变你可以通过比对 md5 发现哪些文章被原始作者修改过。二是 ON CONFLICT 做的是更新而不是忽略有意为之同一个 id 下内容变了旧数据要被覆盖这样后续数据分析拿到的永远是较新版本。正文抽取别用正则去抠 HTML。有的正文里带 div 嵌套、换行符、图片链接正则写起来又脆又难维护。正确姿势是直接存原始返回字段需要展示或导出时再交给 BeautifulSoup 之类的解析库处理入库环节保持原始是最省力的选择。3.3 让“所有”能续下去任务表与断点续爬列表扫描和详情抓取分成两个环节之后任务表就成了连接两者的管道。每拿到一个新 ID就向任务表插入一条 pending 记录“待抓”状态的任务被消费时先查状态再看是否更新。这样程序崩溃、断网、被封都只需要把失败任务重置回 pending不需要从头翻列表。CREATE TABLE IF NOT EXISTS article_task ( id INTEGER PRIMARY KEY, status TEXT DEFAULT pending, retry INTEGER DEFAULT 0, updated_at INTEGER );def dispatch(conn, batch_ids): cur conn.cursor() for aid in batch_ids: cur.execute( INSERT OR IGNORE INTO article_task(id, status, updated_at) VALUES(?, pending, ?), (aid, int(time.time())) ) conn.commit() def run_once(conn, limit200): cur conn.cursor() cur.execute( SELECT id FROM article_task WHERE statuspending ORDER BY id ASC LIMIT ?, (limit,) ) tasks cur.fetchall() for (aid,) in tasks: try: row get_article_detail(aid) upsert_article(conn, row) cur.execute( UPDATE article_task SET statusok, updated_at? WHERE id?, (int(time.time()), aid) ) except Exception as e: cur.execute( UPDATE article_task SET statusfailed, retryretry1, updated_at? WHERE id?, (int(time.time()), aid) ) conn.commit() time.sleep(1)run_once 循环体的最后一行 sleep 是防封节奏的一部分别看它简单忘掉的后果是详情接口连续高频请求直接把会话打到冻结。任务表的另一大好处是支持失败任务重试你只需要写一个 SQL 把 failed 状态重置回 pending再跑一次 run_once 即可。如果某个任务 retry 超过 5 次还失败说明接口或文章结构有变化这时应把它单独挑出来人工排查而不是无限重试。任务表就是整个采集链路的后悔药。4. 限速与防封请求节奏写进代码里个人账号被临时限制绝大多数情况不是因为你抓的量大而是请求波形像机器。服务端常用的限频策略有分钟窗口和小时窗口两层统计同窗口内请求间隔均匀与否往往比总量更敏感。常见做法是两层保险一个全局限速器保证任意两秒内只有一次请求一个随机延时把节奏打散。别小看这两样没有它们再大的账号也熬不过一小时。4.1 一个简单的限速器随机延时与并发控制多线程场景下最怕各线程各自 sleep 然后同时发请求。用一个带锁的限速器统一管理下一次可发起请求的时间点是可控且简单的方案import threading import time import random class RateLimiter: def __init__(self, min_interval1.0, max_interval3.0): self.min_interval min_interval self.max_interval max_interval self._lock threading.Lock() self._next_at 0 def wait(self): with self._lock: now time.time() if now self._next_at: time.sleep(self._next_at - now) # 随机打散下一次请求的间隔避免固定节奏 gap random.uniform(self.min_interval, self.max_interval) self._next_at time.time() gap用法是在每次 HTTP 请求前调用 limiter.wait()比如列表接口设 min_interval2.0、max_interval4.0详情接口设 1.0 到 2.0。这里的参数不是玄学而是个人账号在合理频率区间的经验值低于这条线很快会触发验证。注意不要把 min_interval 设成 0哪怕 max 再大也没用服务端抓的就是最低间隔。如果你开多线程把同一个限速器实例传给所有线程。它的锁机制保证全局只有一个时间窗口不会出现线程 A 和线程 B 间隔很小的情况。这个设计要尽早做晚了再改造就麻烦很多。4.2 重试要区分限流和业务失败不要见错就重试重试逻辑写不好爬虫会陷入无效循环。最常见的错误是看到非 200 状态码就重试但雪球的业务失败往往隐藏在 200 响应里重试多少次都白搭。经验是分三类处理HTTP 429 说明触发限频要按 Retry-After 头等待业务字段 okfalse 且 message 提示操作频繁要指数退避登录态失效的错误则应该终止采集并触发重新登录。def request_with_retry(func, max_retry3, base_delay2.0): last_exc None for attempt in range(max_retry): try: resp func() if resp.status_code 429: # 按服务端给的等待时长休息多数是30秒到几分钟 wait_sec float(resp.headers.get(Retry-After, 30)) time.sleep(wait_sec) continue data resp.json() if data.get(ok): return data if data.get(message) in (操作频繁, 请稍后再试): time.sleep(base_delay * (2 ** attempt)) continue # 其他业务失败多半是参数或登录态问题不建议在这里重试 return data except Exception as e: last_exc e time.sleep(base_delay * (2 ** attempt)) raise RuntimeError(f最终失败: {last_exc})这段代码里有个细节业务消息里的操作频繁错误才退避重试其他业务错误直接返回。因为一个报错的参数重试一万次也报同样的错反而会白白消耗请求量。429 分支单独处理是因为服务端已经明确告诉你超限了这时候正常退避可能不够得按它给的时间等。日志里要记录清楚是哪类失败否则事后排查只能猜。4.3 会话体温做一次轻量健康检查登录态是会过期的与其等业务接口开始报错再回头处理不如每天正式任务前检查一次会话健康度def check_login_valid(): if not SESSION.cookies: return False resp SESSION.get(https://api.example.com/user/info, timeout10) data resp.json() if data.get(ok) and data[data].get(id): return True return False如果返回 False不要立刻重试先清空本地 Cookie 走扫码登录流程。健康检测接口选最轻量、返回字段最少的那一个别用文章详情当体温计白白消耗请求。健康检查加定时任务配合重登录逻辑这套链路就可以无人值守跑很久。5. 避坑爬雪球文章最常见的五个翻车现场5.1 翻车一昨天还好好的第二天一跑全是登录失效现象健康检查能过但业务接口跑几分钟后全返回登录失效的错误码。原因我排了半天发现是某个来源字段在请求中被服务端校验浏览器里登录时它带了完整信息而代码里复制的 Session 缺了这一项。解决重新抓包核对所有请求头字段登录成功后先跑三个不同类型接口做冒烟测试确认没问题再开始批量任务。5.2 翻车二列表能翻页详情却大量返回空正文现象ID 列表一切正常但按 ID 请求详情时 content 字段经常为空。原因文章流里混入了删帖和站内短消息这类对象在详情接口里没有正文另外未登录或会话过期时详情接口会返回空正文但状态码依然正常。解决详情返回空时先做一次会话健康检查排除登录态问题然后把这类 ID 单独标记为“无正文”不纳入文章主库。5.3 翻车三任务表状态设计失误一崩就全量重来现象程序崩溃后重启发现昨天跑过的数据又全部插入一遍。原因任务表里只有 pending 和 ok 两个状态程序崩溃时正在跑的任务还停留在 pending重启后又被重新调度。解决引入 processing 中间状态每次调度任务前先执行一条 SQL把 processing 且超过 5 分钟未更新的记录重置回 pending这样崩溃恢复不需要手动干预。5.4 翻车四频繁换 UA 反而更容易被限制现象请求头里随机切换不同浏览器的 UA第二天直接被要求验证。原因服务端把 UA 看作了身份的一部分同一会话短时间内 UA 跳变会触发异常标记。解决固定一个常用的 UA 字符串不要随机换会话失效时再考虑换新身份重新登录。5.5 翻车五正文直接落 Excel字段全乱现象导出到 Excel 后长文变成一堆带 tag 的字符串换行符也没了。原因正文本质是富文本截断的 tag 在表格里显示不出来。解决入库保持原始 HTML导出时用解析器把纯文本抽出来再做清洗。清洗逻辑独立放在导出阶段不要污染主库数据。6. 进阶把临时脚本装进定时任务做成能长期运转的小系统前面的代码拼起来已经能跑但离“长期运转”还差三件事定时调度、日志留痕、数据校验。定时调度用操作系统的 crontab 是最轻量的方案不需要额外引入任务队列中间件。每天凌晨三点跑一次批量任务白天再安排一次增量任务基本覆盖一天的文章更新节奏0 3 * * * cd /home/user/xueqiu /home/user/venv/bin/python main.py --task daily logs/crawl.log 21 0 13 * * * cd /home/user/xueqiu /home/user/venv/bin/python main.py --task daily logs/crawl.log 21日志直接追加到同一个文件里配合 python 内置的 logging 模块把每次任务的起始时间、成功条数、失败条数、异常详情都打出来。排错的时候最怕看到日志里只有一行“error”没有上下文所以在代码里至少打印当前任务 ID 和接口 URL定位会快很多。数据校验是常被人忽略的环节。我一般会在每天任务结束后加一步抽样校验从昨天入库的文章里随机抽 10 条拿它们的 ID 去列表接口比对看标题、时间、作者是否一致。不一致就说明解析逻辑有偏差需要立刻查。这一步不复杂但能及时发现接口变更避免脏数据堆积def sample_check(): cur conn.cursor() rows cur.execute( SELECT id FROM article WHERE created_at ? ORDER BY random() LIMIT 10, (int(time.time()) - 86400,) ).fetchall() for (aid,) in rows: detail get_article_detail(aid) if not detail or detail[id] ! aid: print(f校验失败: {aid})我自己的习惯是在每个采集脚本的主目录放一个 check.py只做健康检查和抽样校验这两件事全部通过才发通知。这个习惯帮我止损过很多次尤其是接口悄悄大版本更新的时候。校验逻辑别写太复杂只要能回答“数据有没有在正常增长、格式有没有变质”就够。希望帮到你。本文还有配套的精品资源点击获取