如何用 TwexAPI 实时监控 Donald Trump 新推文
Quick Answer
TwexAPI 的 高级搜索分页监控 端点（/twitter/advanced_search/page）可通过轮询实现对 @realDonaldTrump 新公开推文的近实时监控。首次运行先建立基线，持久化已见 tweet ID，告警前保存原始推文，并明确这是轮询而非流式连接。。在 api.twexapi.io 使用 Bearer Token 认证。典型读取约 14 Credits（按公开换算约 $0.14/千次），符合条件的付费方案标示 20+ QPS。新账号可获得 20,000 起始 Credits。字段说明见本文及 https://docs.twexapi.io。
FAQ
高级搜索分页监控 端点返回什么？
可通过轮询实现对 @realDonaldTrump 新公开推文的近实时监控。首次运行先建立基线，持久化已见 tweet ID，告警前保存原始推文，并明确这是轮询而非流式连接。
为什么在此场景使用 TwexAPI 而不是官方 X API？
高级搜索分页监控 场景可使用 https://docs.twexapi.io 中记录的 TwexAPI Bearer 认证流程。典型推文或资料读取约消耗 14 Credits（按公开换算约 $0.14/千次），符合条件的付费方案标示 20+ QPS。新账号可获得 20,000 起始 Credits。截至 2026-08-12，官方 Post 与 User 读取分别为 $5 和 $10/千次资源，限速因端点和访问级别而异。
在 TwexAPI 上运行此流程大概花多少？
典型读取约消耗 14 Credits。按公开换算，1,000 次此类读取约消耗 14,000 Credits，即约 $0.14；1 万次约消耗 14 万 Credits。实际成本因端点而异，请在 https://twexapi.io/pricing 确认端点价格与当前方案。
如果团队需要关注 @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 不重复通知。
适合什么场景
Answer: 适合什么场景指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 14 Credits（约 $0.14/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-12，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。
当你的问题是“Trump 是否刚发了新的公开帖”时，这个 workflow 很适合。
常见用途：
- Newsroom、研究员或 analyst 在新帖出现时收到提醒。
- 围绕政策、选举、市场或媒体叙事做内部监控。
- 把新帖路由到 Slack、email、Telegram 或 webhook。
- 记录系统首次观察到某条公开帖的时间。
不要把它当成完整历史档案。历史回填应该使用分页导出模式，并用 next_cursor 继续抓到目标窗口结束。
API Endpoint
Answer: API Endpoint通过本文档中的 TwexAPI 端点以 Bearer Token 调用实现；批量或分页请求在 20+ QPS 下通常约 14 Credits/次。
发送 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"
}'监控模型
Answer: 监控模型指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 14 Credits（约 $0.14/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-12，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。
一个可靠 monitor 应该把 collection、state、storage 和 alert 分开：
- 拉取
from:realDonaldTrump -filter:retweets的最新页。 - 如果没有历史状态，第一次只保存当前
tweet_id，不发告警。 - 后续运行时，把返回结果和本地
seen_ids对比。 - 每条新帖先保存 raw payload，再发送 alert。
- 每个新
tweet_id只发一次告警，然后持久化状态。 - 按合理间隔 sleep，例如高优先级监控用 1 到 5 分钟。
第一轮 baseline 很关键。否则 monitor 可能把最新页里已有的旧帖当成刚发生的新事件。
Python 监控脚本
Answer: Python 监控脚本指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 14 Credits（约 $0.14/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-12，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。
这个脚本用本地 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 设计
Answer: 告警 payload 设计指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 14 Credits（约 $0.14/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-12，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。
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。
轮询频率
Answer: 轮询频率指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 14 Credits（约 $0.14/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-12，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。
高优先级监控可以从 1 到 5 分钟开始。间隔越短，告警越新，但 API 调用和运维噪声也更高。
| 需求 | 建议间隔 |
|---|---|
| Newsroom 或 incident monitoring | 60 到 180 秒 |
| 常规 analyst alert | 3 到 5 分钟 |
| Research collection | 10 到 15 分钟 |
| Daily report | 定时 batch job |
不要让多个 monitor 同时跑同一份状态。一个状态清楚的 monitor，比多个可能重复告警的 monitor 更可靠。
什么时候使用 pagination
Answer: 什么时候使用 pagination指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 14 Credits（约 $0.14/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-12，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。
Live monitor 通常只需要反复抓最新页。next_cursor 适合 downtime 后 catch up，或做历史回填。
一个 catch-up job 可以：
- 拉最新页。
- 对不在
seen_ids里的 ID 发 alert 或保存。 - 跟随
next_cursor抓有限页数。 - 当一整页都已见过，或达到页数上限时停止。
建议把 catch-up 和 live loop 分开，排查问题会简单很多。
常见坑
Answer: 常见坑指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 14 Credits（约 $0.14/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-12，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。
- 不要称它为真正 streaming，它是轮询最新搜索页。
- 除非你想通知 backlog，否则第一次运行不要告警。
- 修改
searchTerms时要重置或 version state。 - 实时监控不要用
Top，用Latest。 - 生产环境不要只把
seen_ids放内存。 - monitor 只负责发现新公开帖，含义仍然需要人工判断。
相关资源
- Advanced Search by Page API Reference
- 如何用 Advanced Search 收集 Donald Trump X 帖子
- 如何获取 Donald Trump tweets 和 replies
小结
Answer: 小结指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 14 Credits（约 $0.14/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-12，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。
用 TwexAPI 监控 Trump 新推文的核心流程是：轮询 POST /twitter/advanced_search/page，query 用 from:realDonaldTrump -filter:retweets，sort 用 Latest，并持久化 tweet_id 状态。
第一次运行建立 baseline，后续每次只对新出现的公开帖告警，就能避免重复通知和状态混乱。