カーソルページングで Serenity のツイートとリプライをエクスポートする方法
Serenity の最新投稿を数件見るだけなら、ページングなしの取得で十分です。ページ単位のエクスポートが必要になるのは、公開投稿とリプライをローカルに保存し、失敗後に再開できるアーカイブを作りたい場合です。
この記事では、Serenity は公開 X アカウント @aleabitoreddit を指します。
TwexAPI の Get All Tweets and Replies by User by Page は、1 回のリクエストで最大 20 件の公開項目を返し、has_next_page と next_cursor を含みます。重要なのは cursor を保存することです。ページ処理に成功するたびに next_cursor を記録すれば、失敗時に最初からやり直さず、最後の checkpoint から再開できます。
ページングを使う場面
カーソルページングは、単なるプレビューではなく再現可能なデータセットが必要なときに使います。たとえば、Serenity の公開投稿を毎日保存する、後でリプライ分析を行うためのデータを作る、前回実行後に増えた投稿だけ取得する、といった用途です。
最初はローカル JSON への書き出しで十分です。本番ジョブでは、ページごとに保存し、tweet_id で重複排除し、最後に成功した cursor を checkpoint テーブルまたはファイルに残してください。
API エンドポイント
POST https://api.twexapi.io/twitter/tweets-replies/page
Authorization: Bearer <your_token>
Content-Type: application/json最初のリクエストは screen_name のみです。続きは直前のレスポンスにある next_cursor を渡します。has_next_page が false になった場合、または 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":"aleabitoreddit"}'Python カーソルループ
次の例は、各ページを JSONL に追記し、次の cursor を checkpoint ファイルに保存します。すべてをメモリに集めて最後に 1 ファイルへ書くより、長いエクスポートで安全です。
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} 件を追記しました")本番運用の注意点
- 現在のページを保存できた後に
next_cursorを保存します。ページ保存前に cursor だけ進めないでください。 - データベースや分析テーブルへ保存する前に
tweet_idで重複排除します。 created_at_datetime、cashtags、言語フィールド、エンゲージメント数を保存します。- 元の JSON レスポンスも短期間保管しておくと、後でフィールドマッピングを変更したときに再処理できます。
- 削除済みまたは取得不能な投稿は、公開データアーカイブ上の通常の欠落として扱います。
関連リソース
免責事項
本記事は公開 X データの保存とアーカイブ方法を説明するもので、投資助言ではありません。取得した投稿、リプライ、cashtag、エンゲージメント指標だけを投資判断の根拠にしないでください。