TwexAPI で Donald Trump の新しい投稿をリアルタイム監視する方法
@realDonaldTrump の公開投稿をチームで追う場合、手動リロードは安定した運用になりません。誰かが X を開き続け、リンクをチャットに貼り、投稿が新しいのか、重複なのか、すでに対応済みなのかを判断する必要があります。
TwexAPI の Advanced Twitter Search by Page は、近リアルタイム監視に使えます。from:realDonaldTrump -filter:retweets を sortBy: "Latest" で polling し、見たことのある tweet_id を保存し、monitor 起動後に現れた新しい公開投稿だけを alert します。
これは streaming socket ではなく polling workflow です。信頼性は state management から生まれます。初回実行で baseline を作り、seen IDs を永続化し、raw post を保存してから alert を送り、同じ tweet_id の重複通知を防ぎます。
このワークフローが合う場面
問いが「Trump が新しい公開投稿をしたかどうか」であれば、この workflow が合います。
用途の例:
- Newsroom、研究者、アナリストが新しい公開投稿に気づく。
- 政策、選挙、市場、メディア narrative を内部で監視する。
- 新しい投稿を Slack、email、Telegram、webhook に流す。
- システムがその公開投稿を最初に観測した時刻を保存する。
完全な履歴アーカイブとして扱わないでください。過去データの回収には pagination export を使い、next_cursor で対象範囲まで進めます。
API Endpoint
POST リクエストを送ります。
https://api.twexapi.io/twitter/advanced_search/page監視用の request body は小さく保ちます。
| Field | Required | Value |
|---|---|---|
searchTerms | Yes | ["from:realDonaldTrump -filter:retweets"] |
sortBy | Yes | "Latest" |
next_cursor | No | 最新ページを polling するだけなら通常は省略します。 |
各ページは最大約 20 件の tweets と data、has_next_page、next_cursor を返します。リアルタイム監視では、最新 1 ページを繰り返し取得し、tweet_id で重複排除します。
curl --request POST \
--url https://api.twexapi.io/twitter/advanced_search/page \
--header 'Authorization: Bearer <token>' \
--header 'Content-Type: application/json' \
--data '{
"searchTerms": [
"from:realDonaldTrump -filter:retweets"
],
"sortBy": "Latest"
}'監視モデル
信頼できる monitor は collection、state、storage、alerting を分けます。
from:realDonaldTrump -filter:retweetsの最新ページを取得する。- 保存済み state がなければ、初回は現在の
tweet_idを保存するだけにする。 - 次回以降、返ってきた ID と保存済み
seen_idsを比較する。 - 新しい raw post を永続ストレージに保存してから alert を送る。
- 新しい
tweet_idごとに 1 回だけ alert し、state を更新する。 - 高優先度監視なら 1 から 5 分程度の間隔で sleep する。
初回 baseline は重要です。これを行わないと、最新ページにすでに存在していた古い投稿を「新規」として通知してしまいます。
Python 監視スクリプト
このスクリプトは、seen IDs をローカル JSON ファイルに保存し、新しい投稿を JSONL に書いてから任意の webhook に alert を送ります。本番では、ローカルファイルを database、Redis、S3、社内 event store などに置き換えてください。
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"]
11WEBHOOK_URL = os.environ.get("ALERT_WEBHOOK_URL")
12
13API_URL = "https://api.twexapi.io/twitter/advanced_search/page"
14QUERY = "from:realDonaldTrump -filter:retweets"
15SORT_BY = "Latest"
16POLL_SECONDS = int(os.environ.get("POLL_SECONDS", "180"))
17
18STATE_PATH = Path("trump-monitor-state.json")
19EVENTS_PATH = Path("trump-post-events.jsonl")
20
21HEADERS = {
22 "Authorization": f"Bearer {TOKEN}",
23 "Content-Type": "application/json",
24}
25
26def load_seen_ids() -> set[str]:
27 if not STATE_PATH.exists():
28 return set()
29 state = json.loads(STATE_PATH.read_text(encoding="utf-8"))
30 return set(state.get("seen_ids") or [])
31
32def save_seen_ids(seen_ids: set[str]) -> None:
33 STATE_PATH.write_text(
34 json.dumps(
35 {
36 "query": QUERY,
37 "sortBy": SORT_BY,
38 "updated_at": datetime.now(timezone.utc).isoformat(),
39 "seen_ids": sorted(seen_ids),
40 },
41 ensure_ascii=False,
42 indent=2,
43 ),
44 encoding="utf-8",
45 )
46
47def fetch_latest_posts() -> list[dict[str, Any]]:
48 response = requests.post(
49 API_URL,
50 headers=HEADERS,
51 json={"searchTerms": [QUERY], "sortBy": SORT_BY},
52 timeout=30,
53 )
54 response.raise_for_status()
55 body = response.json()
56 return [item for item in body.get("data", []) if item]
57
58def post_url(tweet_id: str) -> str:
59 return f"https://x.com/realDonaldTrump/status/{tweet_id}"
60
61def store_event(tweet: dict[str, Any]) -> None:
62 event = {
63 "observed_at": datetime.now(timezone.utc).isoformat(),
64 "query": QUERY,
65 "tweet": tweet,
66 }
67 with EVENTS_PATH.open("a", encoding="utf-8") as f:
68 f.write(json.dumps(event, ensure_ascii=False) + "\n")
69
70def send_alert(tweet: dict[str, Any]) -> None:
71 tweet_id = tweet.get("tweet_id")
72 text = tweet.get("full_text") or tweet.get("text") or ""
73 payload = {
74 "type": "new_trump_post",
75 "tweet_id": tweet_id,
76 "url": post_url(tweet_id),
77 "created_at": tweet.get("created_at_datetime") or tweet.get("created_at"),
78 "text": text,
79 }
80
81 if not WEBHOOK_URL:
82 print("NEW", payload["url"], text[:180])
83 return
84
85 response = requests.post(WEBHOOK_URL, json=payload, timeout=15)
86 response.raise_for_status()
87
88def monitor_once(seen_ids: set[str], initialized: bool) -> bool:
89 latest = fetch_latest_posts()
90
91 if not initialized:
92 for tweet in latest:
93 tweet_id = tweet.get("tweet_id")
94 if tweet_id:
95 seen_ids.add(tweet_id)
96 save_seen_ids(seen_ids)
97 print(f"Initialized baseline with {len(seen_ids)} seen posts")
98 return True
99
100 new_posts = []
101 for tweet in latest:
102 tweet_id = tweet.get("tweet_id")
103 if tweet_id and tweet_id not in seen_ids:
104 new_posts.append(tweet)
105
106 for tweet in reversed(new_posts):
107 tweet_id = tweet["tweet_id"]
108 store_event(tweet)
109 send_alert(tweet)
110 seen_ids.add(tweet_id)
111
112 if new_posts:
113 save_seen_ids(seen_ids)
114
115 print(f"Checked latest page; new posts: {len(new_posts)}")
116 return initialized
117
118def main() -> None:
119 seen_ids = load_seen_ids()
120 initialized = STATE_PATH.exists()
121
122 while True:
123 try:
124 initialized = monitor_once(seen_ids, initialized)
125 except Exception as exc:
126 print(f"Monitor error: {exc}")
127 time.sleep(POLL_SECONDS)
128
129if __name__ == "__main__":
130 main()実行例:
export TWEXAPI_BEARER_TOKEN="<token>"
export ALERT_WEBHOOK_URL="https://example.com/webhook"
export POLL_SECONDS="180"
python monitor_trump_posts.pyALERT_WEBHOOK_URL を設定しなければ、alert は stdout に出力されます。初回の dry run に便利です。
Alert payload の設計
alert は小さく保ちます。下流システムにはイベントをルーティングするための最小コンテキストだけを渡し、詳細分析は保存済み raw payload を読みます。
{
"type": "new_trump_post",
"tweet_id": "1234567890123456789",
"url": "https://x.com/realDonaldTrump/status/1234567890123456789",
"created_at": "Mon Jul 20 12:34:56 +0000 2026",
"text": "<post text returned by the API>"
}基本ルール:
- 1 つの
tweet_idにつき alert は 1 回。 - alert 前に raw post を保存する。
- payload に query を含める。
- webhook 失敗時は retry してもよいが、重複イベントを作らない。
- 解釈は人間の review に残す。
Polling frequency
高優先度監視は 1 から 5 分間隔から始めます。短い間隔は alert を早くしますが、API call と運用ノイズも増えます。
| Need | Suggested interval |
|---|---|
| Newsroom または incident monitoring | 60 から 180 秒 |
| 通常の analyst alert | 3 から 5 分 |
| Research collection | 10 から 15 分 |
| Daily report | scheduled batch job |
同じ state に対して複数の monitor を走らせないでください。重複 alert を生む複数の monitor より、state が明確な 1 つの monitor の方が安全です。
Pagination を使うタイミング
Live monitor では、最新ページを繰り返し取得するだけで十分なことが多いです。next_cursor は downtime 後の catch-up や historical backfill に使います。
catch-up job の流れ:
- 最新ページを取得する。
seen_idsにない ID を保存または alert する。next_cursorを追って限られたページ数だけ進む。- 1 ページすべてが既知、または page limit に達したら止める。
catch-up と live loop は分けておくと、失敗時の原因を追いやすくなります。
よくある落とし穴
- true streaming と呼ばない。これは最新検索ページの polling です。
- backlog 通知が目的でなければ、初回実行で alert しない。
searchTermsを変える場合は state をリセットまたは version 化する。- リアルタイム監視では
TopではなくLatestを使う。 - 本番で
seen_idsを memory だけに置かない。 - monitor は新しい公開投稿を検出するだけで、意味づけは人間が行う。
関連リソース
- Advanced Search by Page API Reference
- Advanced Search で Donald Trump の X 投稿を収集する方法
- Donald Trump の tweets と replies を取得する方法
まとめ
TwexAPI で Trump の新しい投稿を監視する基本は、POST /twitter/advanced_search/page を polling し、query に from:realDonaldTrump -filter:retweets、sort に Latest を使い、tweet_id state を永続化することです。
初回実行で baseline を作り、その後は新しく現れた公開投稿だけに alert すれば、重複通知と state の混乱を避けられます。