API で Donald Trump の tweets と replies をページごとに取得する方法

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

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

1{
2  "code": 200,
3  "msg": "success",
4  "data": [
5    {
6      "tweet_id": "1803006263529541838",
7      "text": "<post text returned by the API>",
8      "created_at": "Mon Jun 17 03:51:48 +0000 2024",
9      "created_at_datetime": "2024-06-17T03:51:48.000Z",
10      "favorite_count": 0,
11      "retweet_count": 0,
12      "reply_count": 0,
13      "quote_count": 0,
14      "in_reply_to": null,
15      "in_reply_to_screen_name": null,
16      "is_quote_status": false,
17      "user": {
18        "screen_name": "realDonaldTrump"
19      }
20    }
21  ],
22  "has_next_page": true,
23  "next_cursor": "1234567890"
24}

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

1import json
2import os
3import time
4from datetime import datetime, timezone
5from pathlib import Path
6from typing import Any
7
8import requests
9
10TOKEN = os.environ["TWEXAPI_BEARER_TOKEN"]
11API_URL = "https://api.twexapi.io/twitter/tweets-replies/page"
12SCREEN_NAME = "realDonaldTrump"
13
14OUT = Path("realDonaldTrump-tweets-replies.jsonl")
15META = Path("realDonaldTrump-tweets-replies-meta.json")
16
17HEADERS = {
18    "Authorization": f"Bearer {TOKEN}",
19    "Content-Type": "application/json",
20}
21
22def fetch_page(next_cursor: str | None = None) -> dict[str, Any]:
23    payload: dict[str, Any] = {"screen_name": SCREEN_NAME}
24    if next_cursor:
25        payload["next_cursor"] = next_cursor
26
27    response = requests.post(
28        API_URL,
29        headers=HEADERS,
30        json=payload,
31        timeout=45,
32    )
33    response.raise_for_status()
34    body = response.json()
35
36    if body.get("code") not in (None, 200):
37        raise RuntimeError(body.get("msg") or "TwexAPI request failed")
38
39    return body
40
41seen_ids: set[str] = set()
42cursor: str | None = None
43page_count = 0
44saved_count = 0
45cursor_log = []
46
47with OUT.open("w", encoding="utf-8") as output:
48    while True:
49        body = fetch_page(cursor)
50        page_count += 1
51
52        items = [item for item in body.get("data", []) if item]
53        for tweet in items:
54            tweet_id = tweet.get("tweet_id") or tweet.get("id")
55            if not tweet_id or tweet_id in seen_ids:
56                continue
57
58            seen_ids.add(tweet_id)
59            output.write(json.dumps(tweet, ensure_ascii=False) + "\n")
60            saved_count += 1
61
62        next_cursor = body.get("next_cursor")
63        cursor_log.append(
64            {
65                "page": page_count,
66                "fetched_items": len(items),
67                "saved_unique_items": saved_count,
68                "has_next_page": body.get("has_next_page"),
69                "next_cursor": next_cursor,
70            }
71        )
72
73        print(
74            f"page={page_count} fetched={len(items)} "
75            f"saved_unique={saved_count}"
76        )
77
78        if not body.get("has_next_page") or not next_cursor:
79            break
80
81        cursor = next_cursor
82        time.sleep(0.5)
83
84META.write_text(
85    json.dumps(
86        {
87            "screen_name": SCREEN_NAME,
88            "exported_at": datetime.now(timezone.utc).isoformat(),
89            "unique_items": saved_count,
90            "pages": cursor_log,
91        },
92        ensure_ascii=False,
93        indent=2,
94    ),
95    encoding="utf-8",
96)
97
98print(f"Saved {saved_count} unique tweets and replies to {OUT}")

export TWEXAPI_BEARER_TOKEN="<token>"
python export_trump_tweets_replies.py

1import csv
2import json
3
4INPUT_PATH = "realDonaldTrump-tweets-replies.jsonl"
5OUTPUT_PATH = "realDonaldTrump-tweets-replies.csv"
6
7FIELDS = [
8    "tweet_id",
9    "post_type",
10    "created_at_datetime",
11    "text",
12    "lang",
13    "favorite_count",
14    "retweet_count",
15    "reply_count",
16    "quote_count",
17    "view_count",
18    "in_reply_to",
19    "in_reply_to_screen_name",
20    "is_quote_status",
21    "url",
22]
23
24def post_type(tweet: dict) -> str:
25    if tweet.get("in_reply_to") or tweet.get("in_reply_to_screen_name"):
26        return "reply"
27    if tweet.get("is_quote_status") or tweet.get("quoted_status_id_str"):
28        return "quote"
29    return "original"
30
31with open(INPUT_PATH, encoding="utf-8") as source, open(
32    OUTPUT_PATH,
33    "w",
34    encoding="utf-8",
35    newline="",
36) as target:
37    writer = csv.DictWriter(target, fieldnames=FIELDS)
38    writer.writeheader()
39
40    for line in source:
41        tweet = json.loads(line)
42        tweet_id = tweet.get("tweet_id") or tweet.get("id")
43        row = {
44            "tweet_id": tweet_id,
45            "post_type": post_type(tweet),
46            "created_at_datetime": tweet.get("created_at_datetime"),
47            "text": tweet.get("full_text") or tweet.get("text") or "",
48            "lang": tweet.get("lang"),
49            "favorite_count": tweet.get("favorite_count") or 0,
50            "retweet_count": tweet.get("retweet_count") or 0,
51            "reply_count": tweet.get("reply_count") or 0,
52            "quote_count": tweet.get("quote_count") or 0,
53            "view_count": tweet.get("view_count"),
54            "in_reply_to": tweet.get("in_reply_to"),
55            "in_reply_to_screen_name": tweet.get("in_reply_to_screen_name"),
56            "is_quote_status": tweet.get("is_quote_status"),
57            "url": f"https://x.com/realDonaldTrump/status/{tweet_id}",
58        }
59        writer.writerow(row)
60
61print(f"Wrote {OUTPUT_PATH}")