如何下载某个 X 用户的全部推文与回复

下载一个用户的 X 历史内容，不应该只是“传一个很大的数量，然后希望接口一次返回完”。真正可复用的导出任务需要 cursor 循环、原始数据落盘，以及少量去重逻辑，否则重试时很容易产生重复行。

本文使用 TwexAPI 的分页端点 POST /twitter/tweets-replies/page，逐页导出某个用户自己发布的帖子与回复。对于归档、研究、合规复核和可重复的数据任务，它比单次 count 请求更稳。

导出策略

建议同时保留两份文件：

原始 JSONL：每行一个 tweet 对象，作为数据源。
CSV 投影：只保留分析和人工复核需要的字段。

两者都要保留。CSV 更方便查看，但原始 JSON 能保留后续可能需要的媒体、引用、回复关系和作者信息。

API 端点

POST https://api.twexapi.io/twitter/tweets-replies/page

Authorization: Bearer <your_token>
Content-Type: application/json

请求体：

字段	是否必填	说明
`screen_name`	是	目标 X 用户名，可带或不带 `@`
`next_cursor`	否	上一页响应返回的 cursor

每次响应都会返回当前页 data 和分页信息：

字段	含义
`data`	当前页的推文与回复
`has_next_page`	是否还有下一页
`next_cursor`	下一次请求要传入的 cursor

第一页不需要传 next_cursor：

curl --request POST \
  --url https://api.twexapi.io/twitter/tweets-replies/page \
  --header 'Authorization: Bearer <your_token>' \
  --header 'Content-Type: application/json' \
  --data '{
    "screen_name": "elonmusk"
  }'

继续翻页时，把返回的 cursor 传回去：

{
  "screen_name": "elonmusk",
  "next_cursor": "1234567890"
}

Python Cursor 导出脚本

下面的脚本会把原始结果写入 tweets-replies.jsonl。它按 tweet_id 去重，在 has_next_page 为 false 时停止，并把 cursor 循环写得很明确，便于恢复和排查。

1import json
2import os
3import time
4
5import requests
6
7API_URL = "https://api.twexapi.io/twitter/tweets-replies/page"
8TOKEN = os.environ["TWEXAPI_TOKEN"]
9SCREEN_NAME = "elonmusk"
10OUTPUT_PATH = "tweets-replies.jsonl"
11
12def fetch_page(next_cursor=None):
13    body = {"screen_name": SCREEN_NAME}
14    if next_cursor:
15        body["next_cursor"] = next_cursor
16
17    response = requests.post(
18        API_URL,
19        headers={
20            "Authorization": f"Bearer {TOKEN}",
21            "Content-Type": "application/json",
22        },
23        json=body,
24        timeout=45,
25    )
26    response.raise_for_status()
27    payload = response.json()
28    if payload.get("code") not in (None, 200):
29        raise RuntimeError(payload.get("msg", "TwexAPI request failed"))
30    return payload
31
32seen_ids = set()
33next_cursor = None
34page_count = 0
35saved_count = 0
36
37with open(OUTPUT_PATH, "w", encoding="utf-8") as output:
38    while True:
39        payload = fetch_page(next_cursor)
40        page_count += 1
41
42        items = payload.get("data") or []
43        for tweet in items:
44            if not tweet:
45                continue
46
47            tweet_id = tweet.get("tweet_id") or tweet.get("id")
48            if not tweet_id or tweet_id in seen_ids:
49                continue
50
51            seen_ids.add(tweet_id)
52            output.write(json.dumps(tweet, ensure_ascii=False) + "\n")
53            saved_count += 1
54
55        print(f"page={page_count} fetched={len(items)} saved={saved_count}")
56
57        if not payload.get("has_next_page") or not payload.get("next_cursor"):
58            break
59
60        next_cursor = payload["next_cursor"]
61        time.sleep(0.5)
62
63print(f"Finished: {saved_count} unique tweets and replies saved to {OUTPUT_PATH}")

运行方式：

TWEXAPI_TOKEN="your_twexapi_bearer_token" python export-tweets-replies.py

生成 CSV 投影

保存原始 JSONL 后，再生成一个字段更少的 CSV，方便团队查看和标注。

1import csv
2import json
3
4INPUT_PATH = "tweets-replies.jsonl"
5OUTPUT_PATH = "tweets-replies.csv"
6
7fields = [
8    "tweet_id",
9    "created_at_datetime",
10    "text",
11    "lang",
12    "favorite_count",
13    "retweet_count",
14    "reply_count",
15    "quote_count",
16    "view_count",
17    "in_reply_to",
18    "in_reply_to_screen_name",
19    "is_quote_status",
20]
21
22with open(INPUT_PATH, encoding="utf-8") as source, open(
23    OUTPUT_PATH, "w", encoding="utf-8", newline=""
24) as target:
25    writer = csv.DictWriter(target, fieldnames=fields)
26    writer.writeheader()
27
28    for line in source:
29        tweet = json.loads(line)
30        row = {field: tweet.get(field) for field in fields}
31        row["tweet_id"] = tweet.get("tweet_id") or tweet.get("id")
32        row["text"] = tweet.get("full_text") or tweet.get("text") or ""
33        writer.writerow(row)
34
35print(f"Wrote {OUTPUT_PATH}")

建议保留哪些字段

用于研究或审计时，不要只保存正文。至少要留下能解释每一行来源的上下文。

字段	为什么重要
`tweet_id`	稳定的去重和引用键
`created_at` / `created_at_datetime`	时间线排序和时间窗口分析
`text` / `full_text`	可读正文
`in_reply_to`, `in_reply_to_screen_name`	区分回复和独立帖子
`favorite_count`, `retweet_count`, `reply_count`, `quote_count`	互动表现复核
`media`, `urls`, `hashtags`, `cashtags`	内容补充信息
`user`	导出时的作者快照

生产环境注意事项

每页返回后立即写入文件，不要等全部抓完再保存。
如果任务需要恢复，保存最后一次成功的 next_cursor。
按 tweet_id 去重，避免重试或 cursor 变化产生重复行。
即使当前只用 CSV，也保留原始 JSONL。
在输出文件旁记录导出时间、目标 screen_name 和脚本版本。
做周期性导出时，用已保存的 tweet_id 判断新增内容。

Count 端点还是分页端点

TwexAPI 也提供 GET /twitter/{screen_name}/tweets-replies/{count}。如果只是小规模、一次性的获取，并且你已经知道目标数量，它会比较方便。

但正式导出建议使用 POST /twitter/tweets-replies/page。Cursor 分页能让进度更清楚，单次失败的成本更低，也更方便控制落盘和重试。

如何下载某个 X 用户的全部推文与回复

导出策略

建议同时保留两份文件：

原始 JSONL：每行一个 tweet 对象，作为数据源。
CSV 投影：只保留分析和人工复核需要的字段。

两者都要保留。CSV 更方便查看，但原始 JSON 能保留后续可能需要的媒体、引用、回复关系和作者信息。

API 端点

POST https://api.twexapi.io/twitter/tweets-replies/page

Authorization: Bearer <your_token>
Content-Type: application/json

请求体：

字段	是否必填	说明
`screen_name`	是	目标 X 用户名，可带或不带 `@`
`next_cursor`	否	上一页响应返回的 cursor

每次响应都会返回当前页 data 和分页信息：

字段	含义
`data`	当前页的推文与回复
`has_next_page`	是否还有下一页
`next_cursor`	下一次请求要传入的 cursor

第一页不需要传 next_cursor：

curl --request POST \
  --url https://api.twexapi.io/twitter/tweets-replies/page \
  --header 'Authorization: Bearer <your_token>' \
  --header 'Content-Type: application/json' \
  --data '{
    "screen_name": "elonmusk"
  }'

继续翻页时，把返回的 cursor 传回去：

{
  "screen_name": "elonmusk",
  "next_cursor": "1234567890"
}

Python Cursor 导出脚本

下面的脚本会把原始结果写入 tweets-replies.jsonl。它按 tweet_id 去重，在 has_next_page 为 false 时停止，并把 cursor 循环写得很明确，便于恢复和排查。

1import json
2import os
3import time
4
5import requests
6
7API_URL = "https://api.twexapi.io/twitter/tweets-replies/page"
8TOKEN = os.environ["TWEXAPI_TOKEN"]
9SCREEN_NAME = "elonmusk"
10OUTPUT_PATH = "tweets-replies.jsonl"
11
12def fetch_page(next_cursor=None):
13    body = {"screen_name": SCREEN_NAME}
14    if next_cursor:
15        body["next_cursor"] = next_cursor
16
17    response = requests.post(
18        API_URL,
19        headers={
20            "Authorization": f"Bearer {TOKEN}",
21            "Content-Type": "application/json",
22        },
23        json=body,
24        timeout=45,
25    )
26    response.raise_for_status()
27    payload = response.json()
28    if payload.get("code") not in (None, 200):
29        raise RuntimeError(payload.get("msg", "TwexAPI request failed"))
30    return payload
31
32seen_ids = set()
33next_cursor = None
34page_count = 0
35saved_count = 0
36
37with open(OUTPUT_PATH, "w", encoding="utf-8") as output:
38    while True:
39        payload = fetch_page(next_cursor)
40        page_count += 1
41
42        items = payload.get("data") or []
43        for tweet in items:
44            if not tweet:
45                continue
46
47            tweet_id = tweet.get("tweet_id") or tweet.get("id")
48            if not tweet_id or tweet_id in seen_ids:
49                continue
50
51            seen_ids.add(tweet_id)
52            output.write(json.dumps(tweet, ensure_ascii=False) + "\n")
53            saved_count += 1
54
55        print(f"page={page_count} fetched={len(items)} saved={saved_count}")
56
57        if not payload.get("has_next_page") or not payload.get("next_cursor"):
58            break
59
60        next_cursor = payload["next_cursor"]
61        time.sleep(0.5)
62
63print(f"Finished: {saved_count} unique tweets and replies saved to {OUTPUT_PATH}")

运行方式：

TWEXAPI_TOKEN="your_twexapi_bearer_token" python export-tweets-replies.py

生成 CSV 投影

保存原始 JSONL 后，再生成一个字段更少的 CSV，方便团队查看和标注。

1import csv
2import json
3
4INPUT_PATH = "tweets-replies.jsonl"
5OUTPUT_PATH = "tweets-replies.csv"
6
7fields = [
8    "tweet_id",
9    "created_at_datetime",
10    "text",
11    "lang",
12    "favorite_count",
13    "retweet_count",
14    "reply_count",
15    "quote_count",
16    "view_count",
17    "in_reply_to",
18    "in_reply_to_screen_name",
19    "is_quote_status",
20]
21
22with open(INPUT_PATH, encoding="utf-8") as source, open(
23    OUTPUT_PATH, "w", encoding="utf-8", newline=""
24) as target:
25    writer = csv.DictWriter(target, fieldnames=fields)
26    writer.writeheader()
27
28    for line in source:
29        tweet = json.loads(line)
30        row = {field: tweet.get(field) for field in fields}
31        row["tweet_id"] = tweet.get("tweet_id") or tweet.get("id")
32        row["text"] = tweet.get("full_text") or tweet.get("text") or ""
33        writer.writerow(row)
34
35print(f"Wrote {OUTPUT_PATH}")

建议保留哪些字段

用于研究或审计时，不要只保存正文。至少要留下能解释每一行来源的上下文。

字段	为什么重要
`tweet_id`	稳定的去重和引用键
`created_at` / `created_at_datetime`	时间线排序和时间窗口分析
`text` / `full_text`	可读正文
`in_reply_to`, `in_reply_to_screen_name`	区分回复和独立帖子
`favorite_count`, `retweet_count`, `reply_count`, `quote_count`	互动表现复核
`media`, `urls`, `hashtags`, `cashtags`	内容补充信息
`user`	导出时的作者快照

生产环境注意事项

每页返回后立即写入文件，不要等全部抓完再保存。
如果任务需要恢复，保存最后一次成功的 next_cursor。
按 tweet_id 去重，避免重试或 cursor 变化产生重复行。
即使当前只用 CSV，也保留原始 JSONL。
在输出文件旁记录导出时间、目标 screen_name 和脚本版本。
做周期性导出时，用已保存的 tweet_id 判断新增内容。

Count 端点还是分页端点

TwexAPI 也提供 GET /twitter/{screen_name}/tweets-replies/{count}。如果只是小规模、一次性的获取，并且你已经知道目标数量，它会比较方便。

但正式导出建议使用 POST /twitter/tweets-replies/page。Cursor 分页能让进度更清楚，单次失败的成本更低，也更方便控制落盘和重试。

如何下载某个 X 用户的全部推文与回复

如何下载某个 X 用户的全部推文与回复

导出策略

API 端点

Python Cursor 导出脚本

生成 CSV 投影

建议保留哪些字段

生产环境注意事项

Count 端点还是分页端点

相关资源

如何下载某个 X 用户的全部推文与回复

如何下载某个 X 用户的全部推文与回复

导出策略

API 端点

Python Cursor 导出脚本

生成 CSV 投影

建议保留哪些字段

生产环境注意事项

Count 端点还是分页端点

相关资源