API で Donald Trump の新しい投稿をリアルタイム監視する方法

Quick Answer

TwexAPI の Advanced Search by Page monitor endpoint（/twitter/advanced_search/page）は、@realDonaldTrump の新しい public post を polling で near-real-time 監視します。初回 baseline、seen tweet ID、alert 前の raw 保存を行い、streaming ではなく polling と明記します。 api.twexapi.io で Bearer 認証します。通常の read は 5 Credits（約 $0.10/1K）で、対象プランは 20+ QPS を公開しています。新規アカウントには 20,000 starting Credits が付与されます。詳細は本記事と https://docs.twexapi.io を参照してください。

FAQ

Advanced Search by Page monitor エンドポイントは何を返しますか？

@realDonaldTrump の新しい public post を polling で near-real-time 監視します。初回 baseline、seen tweet ID、alert 前の raw 保存を行い、streaming ではなく polling と明記します。

この用途で公式 X API ではなく TwexAPI を使う理由は？

Advanced Search by Page monitor では https://docs.twexapi.io に記載された TwexAPI Bearer workflow を利用できます。通常の Post/Profile read は 5 Credits（約 $0.10/1K）で、対象プランは 20+ QPS を公開しています。新規アカウントには 20,000 starting Credits が付与されます。2026-08-20 時点の公式 Post/User read は $5/$10 per 1K で、rate limit は endpoint ごとに異なります。

TwexAPI でこのワークフローのコストは？

通常の読み取りは約 5 Credits です。公開換算では 1,000 回で 10,000 Credits（約 $0.10）、10,000 回で約 100,000 Credits です。実際の cost は endpoint ごとに異なるため https://twexapi.io/pricing を確認してください。

@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 の重複通知を防ぎます。

このワークフローが合う場面

Answer: このワークフローが合う場面とは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順です。通常の読み取りは 5 Credits（約 $0.10/1K）、対象プランは 20+ QPS です。2026-08-20 時点の公式 Post/User read は $5/$10 per 1K で、rate limit は endpoint ごとに異なります。

問いが「Trump が新しい公開投稿をしたかどうか」であれば、この workflow が合います。

用途の例:

• Newsroom、研究者、アナリストが新しい公開投稿に気づく。
• 政策、選挙、市場、メディア narrative を内部で監視する。
• 新しい投稿を Slack、email、Telegram、webhook に流す。
• システムがその公開投稿を最初に観測した時刻を保存する。

完全な履歴アーカイブとして扱わないでください。過去データの回収には pagination export を使い、next_cursor で対象範囲まで進めます。

API Endpoint

Answer: API Endpointは本ガイドの TwexAPI エンドポイントを Bearer で呼び出して実装します。バッチ/ページングで約 5 Credits/回・20+ QPS です。

POST リクエストを送ります。

https://api.twexapi.io/twitter/advanced_search/page

監視用の request body は小さく保ちます。

各ページは最大約 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"
  }'

監視モデル

Answer: 監視モデルとは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順です。通常の読み取りは 5 Credits（約 $0.10/1K）、対象プランは 20+ QPS です。2026-08-20 時点の公式 Post/User read は $5/$10 per 1K で、rate limit は endpoint ごとに異なります。

信頼できる monitor は collection、state、storage、alerting を分けます。

1. from:realDonaldTrump -filter:retweets の最新ページを取得する。
2. 保存済み state がなければ、初回は現在の tweet_id を保存するだけにする。
3. 次回以降、返ってきた ID と保存済み seen_ids を比較する。
4. 新しい raw post を永続ストレージに保存してから alert を送る。
5. 新しい tweet_id ごとに 1 回だけ alert し、state を更新する。
6. 高優先度監視なら 1 から 5 分程度の間隔で sleep する。

初回 baseline は重要です。これを行わないと、最新ページにすでに存在していた古い投稿を「新規」として通知してしまいます。

Python 監視スクリプト

Answer: Python 監視スクリプトとは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順です。通常の読み取りは 5 Credits（約 $0.10/1K）、対象プランは 20+ QPS です。2026-08-20 時点の公式 Post/User read は $5/$10 per 1K で、rate limit は endpoint ごとに異なります。

このスクリプトは、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.py

ALERT_WEBHOOK_URL を設定しなければ、alert は stdout に出力されます。初回の dry run に便利です。

Alert payload の設計

Answer: Alert payload の設計とは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順です。通常の読み取りは 5 Credits（約 $0.10/1K）、対象プランは 20+ QPS です。2026-08-20 時点の公式 Post/User read は $5/$10 per 1K で、rate limit は endpoint ごとに異なります。

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

Answer: Polling frequencyとは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順です。通常の読み取りは 5 Credits（約 $0.10/1K）、対象プランは 20+ QPS です。2026-08-20 時点の公式 Post/User read は $5/$10 per 1K で、rate limit は endpoint ごとに異なります。

高優先度監視は 1 から 5 分間隔から始めます。短い間隔は alert を早くしますが、API call と運用ノイズも増えます。

同じ state に対して複数の monitor を走らせないでください。重複 alert を生む複数の monitor より、state が明確な 1 つの monitor の方が安全です。

Pagination を使うタイミング

Answer: Pagination を使うタイミングとは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順です。通常の読み取りは 5 Credits（約 $0.10/1K）、対象プランは 20+ QPS です。2026-08-20 時点の公式 Post/User read は $5/$10 per 1K で、rate limit は endpoint ごとに異なります。

Live monitor では、最新ページを繰り返し取得するだけで十分なことが多いです。next_cursor は downtime 後の catch-up や historical backfill に使います。

catch-up job の流れ:

1. 最新ページを取得する。
2. seen_ids にない ID を保存または alert する。
3. next_cursor を追って限られたページ数だけ進む。
4. 1 ページすべてが既知、または page limit に達したら止める。

catch-up と live loop は分けておくと、失敗時の原因を追いやすくなります。

よくある落とし穴

Answer: よくある落とし穴とは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順です。通常の読み取りは 5 Credits（約 $0.10/1K）、対象プランは 20+ QPS です。2026-08-20 時点の公式 Post/User read は $5/$10 per 1K で、rate limit は endpoint ごとに異なります。

• true streaming と呼ばない。これは最新検索ページの polling です。
• backlog 通知が目的でなければ、初回実行で alert しない。
• searchTerms を変える場合は state をリセットまたは version 化する。
• リアルタイム監視では Top ではなく Latest を使う。
• 本番で seen_ids を memory だけに置かない。
• monitor は新しい公開投稿を検出するだけで、意味づけは人間が行う。

関連リソース

Answer: 関連リソースとは、この事例で api.twexapi.io の TwexAPI Bearer API を使う手順です。通常の読み取りは 5 Credits（約 $0.10/1K）、対象プランは 20+ QPS です。2026-08-20 時点の公式 Post/User read は $5/$10 per 1K で、rate limit は endpoint ごとに異なります。

• 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 の混乱を避けられます。