Python 按用户名获取 Twitter 推文：时间线 API 与游标分页
做海外竞品监测、机构公告归档或红人内容研究，常见起点是一份 Twitter 用户名名单。你需要的不只是推文正文，还包括发布时间、作者、公开互动数，以及可重复运行的分页流程。本文用 Twitter 时间线 API 演示如何从公开用户名开始采集近期内容，避免重复入库，也避免把关键词命中误认成目标账号发帖。
这是一项独立的只读数据服务，并非 X 官方服务；不提供私信、发帖或私密内容访问，也不保证完整历史档案。
先选对接口
| 目标 | 操作 | 必填查询参数 |
|---|---|---|
| 用户名转账号 ID、读取公开资料 | user-info | screenname |
| 获取账号近期时间线 | user-timeline | screenname |
| 搜索品牌关键词或话题标签 | search | query |
| 按推文 ID 补全详情 | tweet-info | id |
| 获取账号公开媒体帖子 | users-media | screenname |
完整参数见 Twitter 数据 API。时间线按账号选取，搜索按关键词匹配，两者不能互相替代。
准备工作与首次请求
在控制台取得自己有权使用的 API Key，准备 Python 3 和 requests。密钥只放在服务端环境变量 OPENDATA_API_KEY 中，不要写进网页、公开仓库或截图。查询参数可能进入日志，应对 key 脱敏。
下面使用文档中的公开账号 NASA，无需提供 X 登录凭据。YOUR_API_KEY 是占位符；阅读、打开页面或复制代码不会调用数据接口，替换密钥并主动执行后才会发起请求、消耗调用额度。
curl --get 'https://api.opendata-api.com/api/project/twitter/user-info?key=YOUR_API_KEY' \
--data-urlencode 'screenname=NASA'
curl --get 'https://api.opendata-api.com/api/project/twitter/user-timeline?key=YOUR_API_KEY' \
--data-urlencode 'screenname=NASA'用户名不带 @。第一页不传 cursor；后续把返回的 next_cursor 原样传入 cursor，不要拆解或拼接。user-info 和 user-timeline 的可选 rest_id 优先于用户名，但传了 ID 仍须保留必填的 screenname，翻页期间不要换账号。
响应字段与三层状态
先检查 HTTP 状态，再检查外层 {code, message, data} 的 code，最后检查 data.status。HTTP 200 与外层 code=200 都不等于查到了账号或推文。 资料命中为 active，未找到为 notfound；已验证的时间线成功状态为 ok。未知状态或结构应停止解析，而不是冒充空结果。
时间线业务数据包含 timeline、user、next_cursor 和 prev_cursor。常用字段如下，本文不伪造成功响应：
| 字段 | 处理方式 |
|---|---|
tweet_id | 推文唯一标识，按字符串保存、关联和去重 |
text、created_at | 正文与发布时间；另存采集时间 |
favorites、retweets、replies、quotes、bookmarks | 点赞、转发、回复、引用、收藏的公开计数快照 |
views | 可能是数字字符串，缺失不补零 |
author、entities、media | 作者、文本实体和媒体；作者字段可能为空 |
retweeted、retweeted_tweet | 转发相关记录，不能全部算作目标账号原创 |
公开浏览量不是后台曝光量，认证标签也不等于身份核验。不同接口的计数字段并不完全一致。
Python：限制页数、原样回传游标、按 ID 去重
示例先确认账号，再最多读取三页时间线；只输出作者 ID 匹配且不含转发标记的记录。作者缺失或不匹配会跳过，因此输出数量可能少于接口返回数量。网络、HTTP、平台或业务异常均停止，不自动重试，也不输出含密钥的请求 URL。
import os
import time
import requests
BASE = "https://api.opendata-api.com/api/project/twitter"
KEY = os.environ["OPENDATA_API_KEY"].strip()
if not KEY:
raise RuntimeError("Missing API key")
def read(operation, **params):
try:
response = requests.get(
f"{BASE}/{operation}", params={"key": KEY, **params}, timeout=30
)
except requests.RequestException:
raise RuntimeError("Network request failed") from None
if not 200 <= response.status_code < 300:
raise RuntimeError(f"HTTP {response.status_code}; collection stopped")
try:
envelope = response.json()
except ValueError:
raise RuntimeError("Invalid JSON response") from None
if not isinstance(envelope, dict) or str(envelope.get("code")) != "200":
raise RuntimeError("Platform request failed")
data = envelope.get("data")
if not isinstance(data, dict):
raise RuntimeError("Unexpected business response")
return data
def iter_tweets(screenname, max_pages=3):
profile = read("user-info", screenname=screenname)
if profile.get("status") == "notfound":
return
if profile.get("status") != "active" or not profile.get("rest_id"):
raise RuntimeError("Profile could not be resolved")
if profile.get("protected") is True:
return # Private content is out of scope
target_id = str(profile["rest_id"])
cursor = None
seen_ids, seen_cursors = set(), set()
for _ in range(max_pages):
params = {"screenname": screenname, "rest_id": target_id}
if cursor is not None:
params["cursor"] = cursor # Pass the opaque cursor unchanged
data = read("user-timeline", **params)
if data.get("status") != "ok" or not isinstance(data.get("timeline"), list):
raise RuntimeError("Timeline status or structure is unexpected")
for tweet in data["timeline"]:
if not isinstance(tweet, dict):
raise RuntimeError("Unexpected post record")
tweet_id = tweet.get("tweet_id")
author = tweet.get("author")
if tweet_id is None or not isinstance(author, dict):
continue
tweet_id = str(tweet_id)
if not tweet_id or tweet_id in seen_ids:
continue
if str(author.get("rest_id")) != target_id:
continue
if tweet.get("retweeted") or tweet.get("retweeted_tweet"):
continue
seen_ids.add(tweet_id)
yield tweet
next_cursor = data.get("next_cursor")
if next_cursor is None or next_cursor == "":
return
if not isinstance(next_cursor, str):
raise RuntimeError("Unexpected cursor type")
if next_cursor in seen_cursors:
return
seen_cursors.add(next_cursor)
cursor = next_cursor
time.sleep(1) # Deliberate pacing, not a documented rate limit
if __name__ == "__main__":
for tweet in iter_tweets("NASA"):
print(str(tweet["tweet_id"]), tweet.get("created_at"), tweet.get("text"))采集任务怎样落地
- 竞品内容观察：定时从第一页轮询，按推文 ID 建唯一索引，记录本次页数和时间窗口；不是实时推送，也不能承诺两次轮询之间没有遗漏。
- 公告归档：保存账号 ID、来源推文 ID、发布时间与采集时间；只有返回且通过筛选的记录进入归档。
- 活动复盘：对选定推文另调
tweet-info更新互动快照，保留快照日期，不用今天的计数代替发布当天表现。
上线前设定请求预算、超时和最大页数。遇到限流或临时故障再做有上限的退避；先保存已成功页和游标，再按原账号恢复。游标不保证永久有效，跨任务仍要靠数据库去重。使用公开数据也须遵守适用法律、平台条款和内容使用权要求。
常见问题
如何用 Python 获取指定 Twitter 用户的推文？
调用 user-timeline 并传 screenname；上面的代码额外用资料接口核对账号 ID，并过滤转发和未知作者。
Twitter 时间线接口的 next_cursor 怎么分页？
首请求省略 cursor，之后原样回传 next_cursor。游标缺失、为空或重复时停止，并设置最大页数；不要假定固定页长。
怎么通过 Twitter 用户名查询用户 ID？
向 user-info 传不带 @ 的 screenname，确认 status=active 后读取 rest_id，以字符串保存。
如何根据推文 ID 获取正文、媒体和互动数据？
把字符串 ID 传给 tweet-info 的 id，核对 status=active。详情中的点赞字段是 likes，时间线则是 favorites；不可用推文的返回形态尚未验证。
怎么采集包含品牌关键词或话题标签的推文？
用 search 的 query，例如品牌词或标签，已验证的模式为 search_type=Latest。按相同查询翻页，检查正文与 user_info.rest_id；品牌词命中不代表品牌账号身份，情绪分析须另做。
Twitter 粉丝和关注列表怎么分页导出？
分别用 followers、following，传 screenname 并续传游标。按字符串 user_id 去重后写 CSV，保留空值、采集时间与覆盖说明；总粉丝数不等于本页行数。
时间线 API 能获取账号的全部历史推文吗？ 不能保证。可用历史和页长会变化，应把结果描述为已采集窗口，而非完整历史档案。
分页采集推文时如何去重，避免重复数据？
本次运行用 tweet_id 集合，跨运行用数据库唯一键。重复推文可更新内容，但不同日期的互动快照应单独保存。
Twitter 用户资料返回 notfound 应该怎么处理？
记为本次未找到，停止该账号后续采集。相关字段可能为 null，不要写成零粉丝，也不要把同一输入立即反复重试。
如何获取某条推文的回复和对话上下文？
latest-replies 按 id 取近期回复，tweet-thread 取基础上下文与可用回复；两者字段及分页输出尚未实测验证。user-replies 查的是某账号发出的回复，不是收到的评论。
怎么获取 Twitter 账号公开发布的图片链接？
用 users-media，已观察到照片位于 media.photo，链接字段为 media_url_https。保留来源推文 ID；链接不代表转载许可，也不是完整媒体库。
X Spaces 接口返回音频还是基础信息？
spaces-info 接收 Space 自身的 ID，可含字母数字，只用于查询可用基础信息。具体字段仍须验证，不承诺音频、录音、转写或听众身份。
文档与更多支持
继续阅读 user-timeline、user-info、search 与 tweet-info，调用成本以定价页面为准。
如果还需要 Instagram、TikTok、Reddit、YouTube、LinkedIn、Bluesky、Threads 或 Facebook 数据接口，可通过联系我们提出需求，按需评估新增；也支持源码级私有化部署。