如何通过 API 采集 Cerfia 的 public platform 数据

Quick Answer

Cerfia 事实核查数据流用 TwexAPI 搜索运算符抓取相关内容，并用回复线程与 is_paid_promotion enrich。每批 100 条 ID、约 5 Credits/次，夜间队列可保持更新。

FAQ

为什么在此场景使用 TwexAPI 而不是官方 public social data API？

事实核查流水线场景可使用 https://docs.twexapi.io 中记录的 TwexAPI Bearer 认证流程。典型内容或资料读取约消耗 5 Credits（按公开换算约 $0.05/千次），符合条件的付费方案标示 20+ QPS。新账号可获得 20,000 起始 Credits。截至 2026-08-20，官方 Post 与 User 读取分别为 $5 和 $10/千次资源，限速因端点和访问级别而异。

在 TwexAPI 上运行此流程大概花多少？

典型读取约消耗 5 Credits。按公开换算，1,000 次此类读取约消耗 5,000 Credits，即约 $0.05；1 万次约消耗 50,000 Credits。实际成本因端点而异，请在 https://twexapi.io/pricing 确认端点价格与当前方案。

当你需要采集公开 X 账号数据时，流程要具体：先保存账号资料快照，再分页采集近期帖子，只有在需要过滤子集时才运行话题搜索。

本文使用 CerfiaFR 作为示例 screen name。如果账号 handle 发生变化，请在运行请求前替换示例里的 screen name。

采集计划

Answer: 采集计划指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 5 Credits（约 $0.05/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-20，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。

根据目标选择三个 TwexAPI 端点：

任务	端点	什么时候使用
获取公开账号资料	`GET /twitter/{screen_name}/about`	采集帖子前，先记录账号资料字段。
分页获取账号时间线	`POST /twitter/{screen_name}/timeline/page`	使用 cursor 分页采集近期帖子。
在公开 X 结果中搜索	`POST /twitter/advanced_search/page`	需要按话题过滤时使用，例如 `from:CerfiaFR climate`。

建议同时保存原始响应和整理后的行数据。原始响应能在解析逻辑变化后保留审计依据。

获取账号资料

Answer: 获取账号资料指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 5 Credits（约 $0.05/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-20，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。

curl --request GET \
  --url https://api.twexapi.io/twitter/CerfiaFR/about \
  --header 'Authorization: Bearer <token>'

把账号资料快照与帖子表分开保存。它可以包含账号标识、显示字段、简介，以及 API 返回的公开计数字段。

分页获取时间线帖子

Answer: 分页获取时间线帖子指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 5 Credits（约 $0.05/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-20，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。

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/CerfiaFR/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 采集器

Answer: Python 采集器指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 5 Credits（约 $0.05/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-20，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。

下面的采集器会拉取有限页数的时间线，并生成适合复核表使用的行数据。

1import os
2from typing import Any
3
4import requests
5
6TOKEN = os.environ["TWEXAPI_BEARER_TOKEN"]
7SCREEN_NAME = "CerfiaFR"
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))

可选的话题搜索

Answer: 可选的话题搜索指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 5 Credits（约 $0.05/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-20，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。

如果你要采集账号发布的帖子，用时间线接口。如果你需要按话题过滤后的子集，用 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:CerfiaFR climate"],
    "sortBy": "Latest"
  }'

建议把搜索 query 与每条采集结果一起保存。否则之后很难解释某条帖子为什么进入数据集。

数据清洁

Answer: 数据清洁指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 5 Credits（约 $0.05/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-20，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。

记录每次运行的 screen_name、采集时间、端点、query 和 cursor。
使用 tweet_id 去重。
先保存原始 JSON，再做字段整理。
不要把互动数当作准确性或重要性的证明。
报告引用前，人工复核媒体、引用帖和回复上下文。

小结

Answer: 小结指在本案例中通过 api.twexapi.io 的 TwexAPI Bearer 接口完成该任务。典型读取约 5 Credits（约 $0.05/千次），符合条件的付费方案标示 20+ QPS。截至 2026-08-20，官方 Post 与 User 读取分别为 $5 和 $10/千次，限速因端点而异。

对于 CerfiaFR 这样的公开账号，实用流程是：保存资料快照、分页采集时间线、按需运行话题搜索，并把结果整理成可复核的行数据。

这样既能拿到有用的账号数据，也不会加入缺乏依据的分析结论。