如何用游标分页导出 Serenity 的推文与回复
如果只是看 Serenity 最近几条公开内容，非分页端点已经够用。分页导出解决的是另一个问题：把账号公开帖子和回复稳定保存下来，让任务可以中断后继续，而不是失败一次就从头跑。
这里的 Serenity 指公开 X 账号 @aleabitoreddit。如果你已经用非分页流程试跑过 100 条，并确认字段结构没问题，下一步就应该换成 TwexAPI 的 Get All Tweets and Replies by User by Page 端点做可恢复归档。
这个端点每页最多返回 20 条公开内容，并在响应里给出 has_next_page 和 next_cursor。真正要保存的不是“当前页编号”，而是“下一页 cursor”。每成功处理一页，就记录一次 next_cursor；任务失败后，从最近一次 checkpoint 继续。
什么时候用分页
先判断你是在做一次性预览，还是在做可重复运行的数据任务。只想快速看最近内容时，用非分页接口更简单；要稳定归档、重跑或增量更新时，用游标分页。
常见场景有三类：
| 场景 | 为什么需要分页 |
|---|---|
| 每天归档 Serenity 公开内容 | 任务可能被网络、限流或部署中断，需要能恢复 |
| 为回复分析准备候选帖子 | 先拿到一批源帖，再按 reply_count 或主题挑选 |
| 后续做 cashtag 监控 | 先从源帖中提取 cashtags，再扩大到全站讨论 |
第一次可以先导出到本地 JSONL 文件。生产任务中，建议每抓到一页就写入存储，按 tweet_id 去重，并把最近一次成功的 cursor 写入 checkpoint 表或本地文件。
API 端点
分页导出的请求入口是 POST /twitter/tweets-replies/page。请求体必填 screen_name，首次请求不传 next_cursor；从第二页开始，把上一页返回的 next_cursor 传回去。
POST https://api.twexapi.io/twitter/tweets-replies/page
Authorization: Bearer <your_token>
Content-Type: application/jsoncurl --request POST \
--url https://api.twexapi.io/twitter/tweets-replies/page \
--header 'Authorization: Bearer <your_token>' \
--header 'Content-Type: application/json' \
--data '{"screen_name":"aleabitoreddit"}'Python 游标循环
这个 Python 游标循环会按页请求、按页落盘、按页保存 checkpoint。这样任务在第 30 页失败，也只需要从第 31 页继续，而不是重新导出前 30 页。
下面示例把每条内容追加到 JSONL 文件，并把下一页 cursor 保存到 serenity-checkpoint.json：
1import json
2import os
3from pathlib import Path
4
5import requests
6
7TOKEN = os.environ["TWEXAPI_BEARER_TOKEN"]
8URL = "https://api.twexapi.io/twitter/tweets-replies/page"
9SCREEN_NAME = "aleabitoreddit"
10HEADERS = {"Authorization": f"Bearer {TOKEN}"}
11OUTPUT = Path("serenity-tweets-pages.jsonl")
12CHECKPOINT = Path("serenity-checkpoint.json")
13
14def load_cursor():
15 if not CHECKPOINT.exists():
16 return None
17 return json.loads(CHECKPOINT.read_text(encoding="utf-8")).get("next_cursor")
18
19def save_cursor(cursor):
20 CHECKPOINT.write_text(
21 json.dumps(
22 {"screen_name": SCREEN_NAME, "next_cursor": cursor},
23 ensure_ascii=False,
24 indent=2,
25 ),
26 encoding="utf-8",
27 )
28
29def append_items(items):
30 with OUTPUT.open("a", encoding="utf-8") as output:
31 for item in items:
32 output.write(json.dumps(item, ensure_ascii=False) + "\n")
33
34cursor = load_cursor()
35page_count = 0
36item_count = 0
37
38while True:
39 body = {"screen_name": SCREEN_NAME}
40 if cursor:
41 body["next_cursor"] = cursor
42
43 response = requests.post(URL, headers=HEADERS, json=body, timeout=30)
44 response.raise_for_status()
45 payload = response.json()
46
47 items = [item for item in payload.get("data", []) if item]
48 append_items(items)
49 page_count += 1
50 item_count += len(items)
51
52 next_cursor = payload.get("next_cursor")
53 if not payload.get("has_next_page") or not next_cursor:
54 if CHECKPOINT.exists():
55 CHECKPOINT.unlink()
56 break
57
58 save_cursor(next_cursor)
59 cursor = next_cursor
60
61print(f"已处理 {page_count} 页，追加 {item_count} 条内容")这个脚本没有在内存里累积所有结果，适合长任务。实际写数据库时，把 tweet_id 设置为唯一键；同一条内容重复出现时更新互动指标，而不是插入重复行。
生产环境注意事项
生产环境的目标是让导出任务可以长期跑，而不是只在本地成功一次。最重要的是处理顺序：先写入当前页，再保存下一页 cursor。
| 注意事项 | 建议 |
|---|---|
| checkpoint | 每成功处理一页就保存 next_cursor，不要只在任务结束时保存 |
| 去重 | 写入数据库或分析表前按 tweet_id 去重 |
| 字段保留 | 至少保留 created_at_datetime、cashtags、语言字段和互动指标 |
| 原始数据 | 原始 JSON 建议短期保留，字段映射变更时可以重新解析 |
| 缺口处理 | 已删除或不可访问的帖子视为公开数据归档中的正常缺口 |
如果你每天定时跑一次，可以把“本次新增多少条”“最后一个 cursor 是什么”“是否正常结束”写入任务日志。这样排查问题时，不需要先翻完整数据表。
相关资源
免责声明
本文只介绍公开 X 数据的导出和归档方法。归档到的帖子、回复、cashtag 或互动指标都不能直接作为投资依据。请遵守适用法律与 X 平台规则。