如何用 TwexAPI 实时监控 Donald Trump 新推文

如果团队需要关注 @realDonaldTrump 的公开帖子，手动刷新不是一个可靠流程。有人要不断打开 X，复制链接到群里，再判断这条是不是新帖、是不是重复、是不是已经处理过。

TwexAPI 的 Advanced Twitter Search by Page 可以用来做近实时监控。用 from:realDonaldTrump -filter:retweets 和 sortBy: "Latest" 轮询最新页，保存已见过的 tweet_id，只对 monitor 启动后出现的新公开帖告警。

这不是 streaming socket，而是 polling workflow。可靠性来自状态管理：第一次运行建立 baseline，持久化 seen IDs，先保存 raw post 再告警，并确保同一 tweet_id 不重复通知。

适合什么场景

当你的问题是“Trump 是否刚发了新的公开帖”时，这个 workflow 很适合。

常见用途：

Newsroom、研究员或 analyst 在新帖出现时收到提醒。
围绕政策、选举、市场或媒体叙事做内部监控。
把新帖路由到 Slack、email、Telegram 或 webhook。
记录系统首次观察到某条公开帖的时间。

不要把它当成完整历史档案。历史回填应该使用分页导出模式，并用 next_cursor 继续抓到目标窗口结束。

API Endpoint

发送 POST 请求：

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

监控任务的请求体很小：

字段	必填	本场景取值
`searchTerms`	是	`["from:realDonaldTrump -filter:retweets"]`
`sortBy`	是	`"Latest"`
`next_cursor`	否	监控最新页时通常不传。

每页最多返回约 20 条 tweets，并带有 data、has_next_page 和 next_cursor。实时监控时，反复查询最新第一页，然后按 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 和 alert 分开：

拉取 from:realDonaldTrump -filter:retweets 的最新页。
如果没有历史状态，第一次只保存当前 tweet_id，不发告警。
后续运行时，把返回结果和本地 seen_ids 对比。
每条新帖先保存 raw payload，再发送 alert。
每个新 tweet_id 只发一次告警，然后持久化状态。
按合理间隔 sleep，例如高优先级监控用 1 到 5 分钟。

第一轮 baseline 很关键。否则 monitor 可能把最新页里已有的旧帖当成刚发生的新事件。

Python 监控脚本

这个脚本用本地 JSON 文件保存 seen IDs，并把每条新帖写入 JSONL 后再发送 webhook。生产环境可以把本地文件替换成数据库、Redis、S3 或内部事件存储。

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。

告警 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>"
}

建议规则：

每个 tweet_id 只发一次 alert。
alert 前先保存 raw post。
payload 里带上 query。
webhook 失败可以重试，但不能生成重复事件。
解读仍然保留人工 review。

轮询频率

高优先级监控可以从 1 到 5 分钟开始。间隔越短，告警越新，但 API 调用和运维噪声也更高。

需求	建议间隔
Newsroom 或 incident monitoring	60 到 180 秒
常规 analyst alert	3 到 5 分钟
Research collection	10 到 15 分钟
Daily report	定时 batch job

不要让多个 monitor 同时跑同一份状态。一个状态清楚的 monitor，比多个可能重复告警的 monitor 更可靠。

什么时候使用 pagination

Live monitor 通常只需要反复抓最新页。next_cursor 适合 downtime 后 catch up，或做历史回填。

一个 catch-up job 可以：

拉最新页。
对不在 seen_ids 里的 ID 发 alert 或保存。
跟随 next_cursor 抓有限页数。
当一整页都已见过，或达到页数上限时停止。

建议把 catch-up 和 live loop 分开，排查问题会简单很多。

常见坑

不要称它为真正 streaming，它是轮询最新搜索页。
除非你想通知 backlog，否则第一次运行不要告警。
修改 searchTerms 时要重置或 version state。
实时监控不要用 Top，用 Latest。
生产环境不要只把 seen_ids 放内存。
monitor 只负责发现新公开帖，含义仍然需要人工判断。

小结

用 TwexAPI 监控 Trump 新推文的核心流程是：轮询 POST /twitter/advanced_search/page，query 用 from:realDonaldTrump -filter:retweets，sort 用 Latest，并持久化 tweet_id 状态。

第一次运行建立 baseline，后续每次只对新出现的公开帖告警，就能避免重复通知和状态混乱。

如何用 TwexAPI 实时监控 Donald Trump 新推文

适合什么场景

当你的问题是“Trump 是否刚发了新的公开帖”时，这个 workflow 很适合。

常见用途：

Newsroom、研究员或 analyst 在新帖出现时收到提醒。
围绕政策、选举、市场或媒体叙事做内部监控。
把新帖路由到 Slack、email、Telegram 或 webhook。
记录系统首次观察到某条公开帖的时间。

不要把它当成完整历史档案。历史回填应该使用分页导出模式，并用 next_cursor 继续抓到目标窗口结束。

API Endpoint

发送 POST 请求：

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

监控任务的请求体很小：

字段	必填	本场景取值
`searchTerms`	是	`["from:realDonaldTrump -filter:retweets"]`
`sortBy`	是	`"Latest"`
`next_cursor`	否	监控最新页时通常不传。

每页最多返回约 20 条 tweets，并带有 data、has_next_page 和 next_cursor。实时监控时，反复查询最新第一页，然后按 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 和 alert 分开：

拉取 from:realDonaldTrump -filter:retweets 的最新页。
如果没有历史状态，第一次只保存当前 tweet_id，不发告警。
后续运行时，把返回结果和本地 seen_ids 对比。
每条新帖先保存 raw payload，再发送 alert。
每个新 tweet_id 只发一次告警，然后持久化状态。
按合理间隔 sleep，例如高优先级监控用 1 到 5 分钟。

第一轮 baseline 很关键。否则 monitor 可能把最新页里已有的旧帖当成刚发生的新事件。

Python 监控脚本

这个脚本用本地 JSON 文件保存 seen IDs，并把每条新帖写入 JSONL 后再发送 webhook。生产环境可以把本地文件替换成数据库、Redis、S3 或内部事件存储。

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。

告警 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>"
}

建议规则：

每个 tweet_id 只发一次 alert。
alert 前先保存 raw post。
payload 里带上 query。
webhook 失败可以重试，但不能生成重复事件。
解读仍然保留人工 review。

轮询频率

高优先级监控可以从 1 到 5 分钟开始。间隔越短，告警越新，但 API 调用和运维噪声也更高。

需求	建议间隔
Newsroom 或 incident monitoring	60 到 180 秒
常规 analyst alert	3 到 5 分钟
Research collection	10 到 15 分钟
Daily report	定时 batch job

不要让多个 monitor 同时跑同一份状态。一个状态清楚的 monitor，比多个可能重复告警的 monitor 更可靠。

什么时候使用 pagination

Live monitor 通常只需要反复抓最新页。next_cursor 适合 downtime 后 catch up，或做历史回填。

一个 catch-up job 可以：

拉最新页。
对不在 seen_ids 里的 ID 发 alert 或保存。
跟随 next_cursor 抓有限页数。
当一整页都已见过，或达到页数上限时停止。

建议把 catch-up 和 live loop 分开，排查问题会简单很多。

常见坑

不要称它为真正 streaming，它是轮询最新搜索页。
除非你想通知 backlog，否则第一次运行不要告警。
修改 searchTerms 时要重置或 version state。
实时监控不要用 Top，用 Latest。
生产环境不要只把 seen_ids 放内存。
monitor 只负责发现新公开帖，含义仍然需要人工判断。

小结

第一次运行建立 baseline，后续每次只对新出现的公开帖告警，就能避免重复通知和状态混乱。

如何用 TwexAPI 实时监控 Donald Trump 新推文

如何用 TwexAPI 实时监控 Donald Trump 新推文

适合什么场景

API Endpoint

监控模型

Python 监控脚本

告警 payload 设计

轮询频率

什么时候使用 pagination

常见坑

相关资源

小结

如何用 TwexAPI 实时监控 Donald Trump 新推文

如何用 TwexAPI 实时监控 Donald Trump 新推文

适合什么场景

API Endpoint

监控模型

Python 监控脚本

告警 payload 设计

轮询频率

什么时候使用 pagination

常见坑

相关资源

小结