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

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

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

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"
  }'

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

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

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}")