如何通过 API 实时监控 Donald Trump 新推文

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

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

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

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