用 Python 抓推特：一个能扛住生产环境的脚本
用 Python 抓推特：一个能扛住生产环境的脚本
每篇教程都给你二十行代码把推文取下来。它们能跑，然后你拿去跑一个真实任务，才发现那二十行是简单的部分。
★这篇讲的是剩下的 80%：★ 不会截断的翻页、能区分限流和真实故障的重试、能扛住重启的去重，以及崩溃后的断点续跑。没有一样是难的，每一样都很琐碎，而漏掉任何一样都会产生一个静默失败的脚本。
(如果你还在决定要不要自建，四种方案的对比在这里。本篇假设你已经选了 Python。)
库怎么选
| 选项 | 你得到什么 | 你要负责什么 |
|---|---|---|
snscrape |
不需要密钥，直接读公开端点 | ★X 一改动，你来修★ |
tweepy |
官方 API 封装，文档完善 | 一个 API key 和它的费用 |
requests + 数据接口 |
纯 HTTP，没有封装 | 什么都不用 —— 坏了是服务商的事 |
★真正的问题不是"哪个库最好"，而是"X 改动时谁来修"。★
用 snscrape 那个人是你。它免费、能力确实不错，而维护成本是真实的 —— 去看任何 X 抓取库的 issue 列表，你会看到同样的节奏：一波"突然什么都不返回"、修复、安静、然后又一波。
用 tweepy 你走的是官方 API，所以很少坏 —— 但定价才是约束，超出免费档的读取是按条计费的。
用纯 requests 对接数据接口，没有封装会坏，也没有 SDK 版本要跟。⚠️ 它要花钱，这是诚实的代价。
我们是第三行。★没有 SDK —— 就是 HTTP 和 JSON★，意味着没有东西要装，也没有东西要升级。
让它能上生产的四件事
1. 不会截断的翻页
★最常见的那个 bug。★ 一页返回的条数少于你要的，并不代表你到底了 —— 列表中间出现短页非常常见。
# ❌ 静默丢数据
if len(batch) < 100:
break
# ✅ 唯一正确的条件
cursor = body.get("meta", {}).get("next_cursor")
if not cursor:
break
为什么它难缠： 它产生的结果看起来很合理。你本该拿到 900 条却拿到 340 条，什么错都不报，而你在几周后某个总数对不上时才发现。
2. 能区分失败类型的重试
不是所有失败都该用同一种反应：
| 状态码 | 含义 | 动作 |
|---|---|---|
| 429 | 被限流 | ★等待后重试 —— 这个一定会成功★ |
| 5xx | 服务端问题 | 退避后重试 |
| 404 | 不存在 | ★不要重试 —— 它永远不会成功★ |
| 400/401 | 你的请求错了 | 不要重试，去改代码 |
⚠️ ★对 404 无限重试是真的有人上线过的 bug。★ 同样地，对 429 直接放弃也是 —— 而那恰恰是唯一保证会自行恢复的失败。更多关于限流处理。
3. 能扛住重启的去重
内存里的 set() 在重启后是空的，所以续跑的任务会把一切重新处理一遍。要持久化。
4. 断点续跑
长任务会被打断。把游标和已见 ID 存下来，重启就是继续而不是重来。
完整脚本
import json, pathlib, time
import requests
BASE = "https://api.socialapi.tech"
KEY = "your_api_key"
HDRS = {"X-API-Key": KEY}
STATE = pathlib.Path("state.json")
OUT = pathlib.Path("posts.jsonl")
def load_state():
if STATE.exists():
s = json.loads(STATE.read_text())
return s.get("cursor"), set(s.get("seen", []))
return None, set()
def save_state(cursor, seen):
# 给已见列表封顶 —— 只有最近的尾部有用
STATE.write_text(json.dumps({"cursor": cursor,
"seen": sorted(seen)[-50_000:]}))
def fetch(path, params, attempts=5):
"""一次请求,带「区分失败类型」的重试。"""
for attempt in range(attempts):
try:
r = requests.get(f"{BASE}{path}", params=params,
headers=HDRS, timeout=60)
except requests.RequestException:
time.sleep(2 ** attempt) # 网络抖动
continue
if r.status_code == 200:
return r.json()
if r.status_code == 404:
return None # ★永远不重试★
if r.status_code in (400, 401, 403):
raise RuntimeError(f"{r.status_code}: {r.text[:200]}")
# 429 与 5xx 值得等
time.sleep(2 ** attempt)
raise RuntimeError(f"重试 {attempts} 次后放弃")
def scrape(username, max_pages=100):
cursor, seen = load_state()
total = 0
with OUT.open("a", encoding="utf-8") as out:
for _ in range(max_pages):
params = {"username": username, "limit": 100}
if cursor:
params["cursor"] = cursor
body = fetch("/v1/user/last_tweets", params)
if body is None:
break # 账号没了
for post in body["data"]:
if post["id"] in seen:
continue
seen.add(post["id"])
out.write(json.dumps(post, ensure_ascii=False) + "\n")
total += 1
cursor = body.get("meta", {}).get("next_cursor")
save_state(cursor, seen) # ★每一页之后都存★
if not cursor:
break # ★唯一正确的停止条件★
return total
print(f"新写入 {scrape('nasa')} 条")
每一块在干什么：
- ★每一页之后都
save_state★ —— 崩溃只损失一页，不是整轮 - ★404 时
return None★ —— 账号没了，重试解决不了 - ★400/401 时
raise★ —— 你的请求有问题，重试只是白烧配额 - ★429/5xx 指数退避★ —— 这些会自行恢复
- ★没有东西要装或升级★ —— 就是纯 HTTP，没有 SDK 版本要跟
- ★JSONL 追加写★ —— 能扛中断，而且不需要把整个数据集放内存
值得点名的常见错误
每次请求之间 time.sleep(1)。 它并不能防止限流，只是让慢任务更慢。该在 429 来的时候处理它，而不是猜一个"安全"的节奏。
先攒进 list 再统一写。 长任务会占几百 MB，而且崩溃时全丢。要流式写盘。
把 ID 当整数。 ⚠️ X 的 ID 超出某些语言的安全整数范围，★JavaScript 在 2^53 以上会静默取整★ —— 见为什么 ID 必须是字符串。
请求不设超时。 一个卡住的连接会无限期阻塞任务。永远要设。
写 CSV 不做转义。 推文正文里有逗号、引号和换行 —— 导出的坑在这里。
大家会问的问题
怎么用 Python 抓推特？
要么用读公开端点的开源库，要么用 requests 对接数据接口。上面那四件生产化的事对两者都适用。
Python 上最好的推特库是哪个？
官方 API 用 tweepy，公开端点用 snscrape，数据接口用纯 requests。区分点是 X 改动时谁来维护。
snscrape 现在还能用吗？ 它在 X 的两次改动之间能用，改动之后就坏。 依赖它做时效性任务之前，先看它最近的 issue。
不用 API key 能抓推特吗？ 开源库读公开端点不需要 key。那正是它们的用途，也正是它们会坏的原因。
Python 里怎么取某个用户的推文？ 翻页拉他的时间线。上面那个脚本是完整可跑的。
翻页该怎么处理？ ★跟着游标翻到空为止。绝不要在短页处停。★
我的爬虫为什么只返回一部分推文？ 几乎总是在短页处停了，而不是在游标为空时停。
Python 里怎么避免限流？ 你不是避免它，而是处理它。 捕获 429 并指数退避。
tweepy 和 snscrape 有什么区别？
tweepy 封装官方 API 且需要密钥；snscrape 不需要密钥，直接读公开端点。
Python 里怎么按关键词抓推文？ 用搜索调用传你的查询，然后翻页。
Python 里能按日期抓吗？
能，用 since: 和 until: —— 但要把区间分块，否则结果上限会截断你。
抓下来的推文怎么保存？ 逐行追加 JSONL。 需要人读的时候再转成 CSV。
崩溃后怎么续跑？ 每一页之后都持久化游标和已见 ID。 上面的脚本就是这么做的。
我的脚本为什么越跑越慢？ 通常是内存里的 list 无限增长。改成流式写盘。
怎么抓多个账号？ 循环处理，每个账号维护独立的状态，这样一个失败不会丢掉其它账号的进度。
这个需要用 async 吗？ 很少需要。 瓶颈是限流不是并发 —— async 在这里增加复杂度但不增加速度。
Python 里怎么抓粉丝？ 对粉丝端点用同样的翻页模式 —— 见粉丝列表处理。
怎么拿到互动数据？ 它随每条推文一起返回 —— 点赞、转发、回复、引用、浏览量。不需要第二次调用。
能抓图片和视频吗？
媒体 URL 随推文数据一起返回，在 photos 和 videos 里。⚠️ ★两者形状不同 —— 见媒体字段。★
能抓多少数据？ 受索引能回溯多远限制，不受你的代码限制 —— 大多数账号是几个月而不是几年。
用 Python 抓推特合法吗？ 读取公开数据是被广泛实践的。对账号做自动化操作才是有风险的那一类 —— 合规讨论。
需要代理吗？ 自建爬虫上了量之后通常需要。这正是人们低估的维护成本之一。
怎么测试我的爬虫？ 先拿一个小账号跑，并核对条数和界面上看到的一致。
为什么我收到 401？ 密钥错误或缺失。★不要重试 401 —— 它永远不会成功。★
超时该设多少？ API 调用 60 秒；媒体下载要更长，因为文件大得多。
该记什么日志？ 记页码、游标和每页条数。一旦发生截断，这些正是能告诉你截在哪里的东西。
该用数据库还是文件？ 文件起步足够。 JSONL 追加干净，以后导进任何东西都行。
跨多次运行怎么去重？ 持久化已见 ID。 内存里的 set 在重启后是空的。
能定时跑吗？ 能 —— 这就是正常形态。在运行之间保存状态，每次都从上次停的地方继续。
已删除的账号怎么处理？ 404 就是没了。★直接返回，不要重试。★
需要什么 Python 版本？ 任何当前版本。这里没有依赖新语言特性的东西。
要装什么？
直接调 HTTP 的话装 requests。走库的方案则要装那个库和它的依赖。
能改用 JavaScript 吗？ 能，逻辑一样。⚠️ ★注意 ID 精度问题 —— 它在 JavaScript 里咬得更狠。★
怎么扩展到几千个账号？ 把任务放进队列，用适度的并发处理。 决定上限的是限流，不是 CPU。
抓推特最快的方式是什么？ 速度由限流决定。 写得好的爬虫之间差别很小，限流是大家共享的。
怎么知道我的爬虫漏了东西？ 拿一个已知账号的总数和界面对比。 要找的失败模式是静默截断。
如果你要把它放进生产
上面那个脚本是可用的起点。让它活下来的是那些不体面的部分： ★按游标翻页、按失败类型重试、每页之后持久化状态。★
这里每一种失败都是静默的 —— 任务"完成"了、文件里有数据，而数据是残缺的。
我们的 API 是纯 HTTP，带游标翻页，每条推文都带完整互动数据，所以没有 SDK 要装也没有版本要跟。只读、固定按次价格，而且没有属于你的限流要管 —— 这直接消掉了上面四件事中的一件。