如何用 TwexAPI 采集 Donald Trump 的 X 数据
采集公开账号数据时,不建议只靠一次关键词搜索。更稳的结构是:先保存账号资料,再用 cursor 分页获取时间线帖子,最后在需要更窄范围时使用搜索。
本文使用 realDonaldTrump 作为示例 screen name。公开账号可能更换 handle,尤其在做长期归档时,运行前请先确认当前 screen name。
采集计划
这个流程使用三个 TwexAPI 端点:
| 任务 | 端点 | 什么时候使用 |
|---|---|---|
| 获取公开账号资料 | GET /twitter/{screen_name}/about | 采集帖子前保存账号资料快照。 |
| 分页获取账号帖子 | POST /twitter/{screen_name}/timeline/page | 使用 cursor 分页采集时间线帖子。 |
| 搜索公开 X 结果 | POST /twitter/advanced_search/page | 需要按话题、短语或搜索操作符过滤时使用。 |
建议同时保存原始响应和整理后的行数据。这样字段映射或导出规则变化时,仍然能回到原始 JSON 复核。
获取账号资料
curl --request GET \
--url https://api.twexapi.io/twitter/realDonaldTrump/about \
--header 'Authorization: Bearer <token>'把账号资料快照与帖子表分开保存。它可以包含账号标识、展示字段、简介字段,以及 API 返回的公开计数字段。
分页获取时间线帖子
POST /twitter/{screen_name}/timeline/page 接收 count 和 next_cursor。第一页不传 next_cursor。如果响应里的 has_next_page 为 true,就把返回的 next_cursor 放进下一次请求。
curl --request POST \
--url https://api.twexapi.io/twitter/realDonaldTrump/timeline/page \
--header 'Authorization: Bearer <token>' \
--header 'Content-Type: application/json' \
--data '{
"count": 20
}'响应会包含当前页 data 和分页字段。
{
"code": 200,
"msg": "success",
"data": [],
"has_next_page": true,
"next_cursor": "DAAHCgAB...",
"requested_count": 20,
"effective_count": 20,
"parsed_count": 20
}Python 采集器
下面的采集器会拉取有限页数的时间线,并把每条帖子整理成方便复核的行数据。
1import os
2from typing import Any
3
4import requests
5
6TOKEN = os.environ["TWEXAPI_BEARER_TOKEN"]
7SCREEN_NAME = "realDonaldTrump"
8BASE_URL = "https://api.twexapi.io"
9
10def headers() -> dict[str, str]:
11 return {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}
12
13def get_profile(screen_name: str) -> dict[str, Any] | None:
14 response = requests.get(
15 f"{BASE_URL}/twitter/{screen_name}/about",
16 headers=headers(),
17 timeout=30,
18 )
19 response.raise_for_status()
20 return response.json().get("data")
21
22def get_timeline_page(
23 screen_name: str,
24 *,
25 count: int = 20,
26 next_cursor: str | None = None,
27) -> dict[str, Any]:
28 payload: dict[str, Any] = {"count": count}
29 if next_cursor:
30 payload["next_cursor"] = next_cursor
31
32 response = requests.post(
33 f"{BASE_URL}/twitter/{screen_name}/timeline/page",
34 headers=headers(),
35 json=payload,
36 timeout=30,
37 )
38 response.raise_for_status()
39 return response.json()
40
41def normalize_post(post: dict[str, Any]) -> dict[str, Any]:
42 tweet_id = post.get("tweet_id") or post.get("id")
43 return {
44 "tweet_id": tweet_id,
45 "created_at": post.get("created_at_datetime") or post.get("created_at"),
46 "text": post.get("full_text") or post.get("text") or "",
47 "favorite_count": post.get("favorite_count") or 0,
48 "retweet_count": post.get("retweet_count") or 0,
49 "reply_count": post.get("reply_count") or 0,
50 "quote_count": post.get("quote_count") or 0,
51 "url": f"https://x.com/i/status/{tweet_id}",
52 }
53
54def collect_timeline(screen_name: str, max_pages: int = 3) -> list[dict[str, Any]]:
55 rows: list[dict[str, Any]] = []
56 cursor = None
57
58 for _ in range(max_pages):
59 page = get_timeline_page(screen_name, count=20, next_cursor=cursor)
60 rows.extend(normalize_post(post) for post in page.get("data", []))
61
62 if not page.get("has_next_page"):
63 break
64 cursor = page.get("next_cursor")
65 if not cursor:
66 break
67
68 return rows
69
70if __name__ == "__main__":
71 profile = get_profile(SCREEN_NAME)
72 posts = collect_timeline(SCREEN_NAME, max_pages=2)
73 print(profile.get("screen_name") if profile else SCREEN_NAME, len(posts))可选的话题搜索
如果目标是账号归档,用时间线分页接口。如果只需要某个话题或关键词下的子集,再使用 advanced search by 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 economy"],
"sortBy": "Latest"
}'把完整 query 与每条结果一起保存。之后有人问为什么某条帖子进入或没有进入数据集时,这个字段会很关键。
数据清洁
- 记录每次运行的
screen_name、采集时间、端点、query 和 cursor。 - 使用
tweet_id作为去重主键。 - 先保存原始 JSON,再展开字段。
- 不要把互动数当作准确性、重要性或公众态度的证明。
- 在报告引用前,人工复核引用帖、媒体内容和回复上下文。
小结
对于 realDonaldTrump 这样的公开账号,实用流程是:账号资料、时间线分页、可选话题搜索,以及严格的数据导出记录。
这样文章给读者的是可复用的采集方法,而不是无法验证的价格宣传、伪造响应或 API 没有提供的分析结论。