如何用高级搜索采集 Donald Trump 的 X 帖子

如果你要追踪 X 上的公众人物，真正难的通常不是第一条请求，而是让查询可复现、分页不中断，并在后续分析时还能说明数据是怎么来的。

这篇文章使用 TwexAPI 的 Advanced Twitter Search by Page 端点采集 @realDonaldTrump 的公开帖子。同样的流程也适用于其他公开账号：先写清楚高级搜索语句，请求第一页，保存返回的 next_cursor，再继续请求下一页，直到 has_next_page 为 false 或达到你的采集范围。

什么时候用这个流程

当你需要稳定采集符合高级搜索表达式的公开 X 帖子时，适合使用分页高级搜索。以 Trump 监控或研究场景为例，常见目标包括：

按指定时间范围采集 from:realDonaldTrump 的帖子。
用 economy、election、trade 等主题词缩小数据集。
用 -filter:retweets 排除转推，只保留原创帖和引用评论。
在做摘要、情绪分析或仪表盘之前，先保存原始 API 响应。

不要把单页结果当成完整归档。每一页只是当前结果集的一段，只有按计划走完整条 cursor 链，数据集才算完成。

端点与请求体

发送 POST 请求到：

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

请求体包含三个字段：

字段	是否必填	用法
`searchTerms`	是	高级搜索语句数组。此场景建议从 `from:realDonaldTrump -filter:retweets` 开始。
`sortBy`	是	用 `Latest` 按时间倒序采集；只有明确需要排序结果时再用 `Top`。
`next_cursor`	否	第一页留空。下一页传入上一次响应里的 `next_cursor`。

响应会返回 data、has_next_page 和 next_cursor。建议把这三个值和导出元信息一起保存。

请求第一页

先使用清晰、窄一点的查询，并设置 sortBy: "Latest"。这样后续分页更容易解释，因为下一页会沿着最新结果集继续往后取。

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

简化后的响应结构如下：

1{
2  "code": 200,
3  "msg": "success",
4  "data": [
5    {
6      "tweet_id": "1234567890123456789",
7      "created_at": "Mon Jul 01 12:34:56 +0000 2025",
8      "text": "<API 返回的帖子正文>",
9      "favorite_count": 1000,
10      "retweet_count": 200
11    }
12  ],
13  "has_next_page": true,
14  "next_cursor": "cursor_from_this_response"
15}

使用 next_cursor 继续分页

当 has_next_page 为 true 时，把上一次响应中的 cursor 传入下一次请求。

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",
    "next_cursor": "cursor_from_this_response"
  }'

不要自己拼 cursor，也不要把别的查询返回的 cursor 混用。cursor 只对应生成它的查询条件和排序方式。

Python 导出脚本

下面的脚本会把每条帖子以 JSONL 形式写入文件，按 tweet_id 去重，并把 cursor 流程写入一个元信息文件。

1import json
2import time
3from datetime import datetime, timezone
4from pathlib import Path
5
6import requests
7
8TOKEN = "<your_bearer_token>"
9URL = "https://api.twexapi.io/twitter/advanced_search/page"
10QUERY = "from:realDonaldTrump -filter:retweets"
11SORT_BY = "Latest"
12OUT = Path("realDonaldTrump-posts.jsonl")
13META = Path("realDonaldTrump-posts-meta.json")
14
15headers = {
16    "Authorization": f"Bearer {TOKEN}",
17    "Content-Type": "application/json",
18}
19
20seen_ids = set()
21cursor = None
22page = 0
23cursor_log = []
24
25with OUT.open("w", encoding="utf-8") as f:
26    while True:
27        payload = {
28            "searchTerms": [QUERY],
29            "sortBy": SORT_BY,
30        }
31        if cursor:
32            payload["next_cursor"] = cursor
33
34        response = requests.post(URL, headers=headers, json=payload, timeout=30)
35        response.raise_for_status()
36        body = response.json()
37
38        page += 1
39        items = body.get("data") or []
40        for item in items:
41            tweet_id = item.get("tweet_id")
42            if not tweet_id or tweet_id in seen_ids:
43                continue
44            seen_ids.add(tweet_id)
45            f.write(json.dumps(item, ensure_ascii=False) + "\n")
46
47        cursor_log.append({
48            "page": page,
49            "items": len(items),
50            "next_cursor": body.get("next_cursor"),
51            "has_next_page": body.get("has_next_page"),
52        })
53
54        if not body.get("has_next_page") or not body.get("next_cursor"):
55            break
56
57        cursor = body["next_cursor"]
58        time.sleep(1)
59
60META.write_text(json.dumps({
61    "query": QUERY,
62    "sortBy": SORT_BY,
63    "exported_at": datetime.now(timezone.utc).isoformat(),
64    "unique_tweets": len(seen_ids),
65    "pages": cursor_log,
66}, ensure_ascii=False, indent=2), encoding="utf-8")
67
68print(f"Saved {len(seen_ids)} unique posts to {OUT}")

如果任务会长时间运行，建议加入临时网络错误的重试逻辑，并在每次请求下一页之前持久化最后一个成功 cursor。这样中断后可以继续，而不是从第一页重来。

查询语句示例

先从简单语句开始，再一次只加一个筛选条件。这样结果异常时，更容易判断是哪一段查询改变了数据集。

目标	`searchTerms` 值
获取该账号的最新公开帖子	`from:realDonaldTrump -filter:retweets`
获取提到经济的帖子	`from:realDonaldTrump economy -filter:retweets`
固定月份窗口	`from:realDonaldTrump since:2024-01-01 until:2024-02-01 -filter:retweets`
只看带媒体的帖子	`from:realDonaldTrump filter:media -filter:retweets`
关注投票相关表达	`from:realDonaldTrump (vote OR voting OR ballot) -filter:retweets`

如果要比较多个主题，建议每个主题单独跑查询，并把查询语句和每条结果一起保存。把太多主题塞进一个查询，会让后续分析更难解释。

应该保存哪些字段

至少保存：

API 返回的原始 tweet 对象。
tweet_id、created_at、正文、作者字段，以及响应中存在的互动计数字段。
完整的 searchTerms、sortBy、导出时间和 cursor 链。
规范化后的帖子 URL，例如 https://x.com/realDonaldTrump/status/<tweet_id>。

做分析时，不建议只依赖某个时刻抓到的互动数。互动数会变化，所以每次导出都应该带时间戳，并明确后续刷新是覆盖旧值，还是追加一个新快照。

常见坑

分页中途修改 searchTerms，会让 cursor 链失去可解释性。
用 Top 做历史导出，可能得到更难审计的排序结果。除非你的分析目标就是排名，否则优先用 Latest。
过早丢掉原始响应，会让后续字段映射和复查变困难。
把公众人物帖子当成完全中性的语料，会忽略上下文。请把日期、查询语句和筛选决策一起保存。

小结

采集 Trump 帖子时，TwexAPI 更清晰的流程是：写好高级搜索语句，调用 POST /twitter/advanced_search/page，用 next_cursor 持续分页，同时保存原始结果和元信息。这样得到的数据集更容易复跑、检查和解释。

第一次导出可以作为基线。之后再根据产品需求决定是否做定时刷新、主题数据集，或对新增帖子做提醒。