⚡

Twitter/X数据抓取工具 twscrape

twscrape 是一个强大的 Python 库，专门用于高效、大规模地抓取 Twitter（现 X）平台的公开数据。它通过 GraphQL API 和账号轮换机制，支持异步并行抓取推文、用户资料、关注者列表和趋势话题等，是构建社交媒体分析、舆情监控和市场研究工具的利器。关键词：Twitter 数据抓取，X 平台爬虫，社交媒体分析，Python 异步爬虫，账号轮换，GraphQL API，数据采集。

数据工程 0 次安装 0 次浏览 更新于 2/23/2026

twscrape

用于通过 GraphQL API 抓取 Twitter/X 数据的 Python 库，支持账号轮换和会话管理。

何时使用此技能

在以下情况下使用此技能：

  • 处理 Twitter/X 数据提取和抓取
  • 需要通过账号轮换绕过 Twitter API 限制
  • 构建社交媒体监控或分析工具
  • 从 Twitter/X 提取推文、用户资料、关注者、趋势
  • 需要异步/并行抓取操作以进行大规模数据收集
  • 寻找官方 Twitter API 的替代方案

快速参考

安装

pip install twscrape

基本设置

import asyncio
from twscrape import API, gather

async def main():
    api = API()  # 默认使用 accounts.db

    # 添加账号（使用 cookies - 更稳定）
    cookies = "abc=12; ct0=xyz"
    await api.pool.add_account("user1", "pass1", "email@example.com", "mail_pass", cookies=cookies)

    # 或添加账号（使用登录名/密码 - 稳定性较差）
    await api.pool.add_account("user2", "pass2", "email2@example.com", "mail_pass2")
    await api.pool.login_all()

asyncio.run(main())

常见操作

# 搜索推文
await gather(api.search("elon musk", limit=20))

# 获取用户信息
await api.user_by_login("xdevelopers")
user = await api.user_by_id(2244994945)

# 获取用户推文
await gather(api.user_tweets(user_id, limit=20))
await gather(api.user_tweets_and_replies(user_id, limit=20))
await gather(api.user_media(user_id, limit=20))

# 获取关注者/正在关注
await gather(api.followers(user_id, limit=20))
await gather(api.following(user_id, limit=20))

# 推文操作
await api.tweet_details(tweet_id)
await gather(api.retweeters(tweet_id, limit=20))
await gather(api.tweet_replies(tweet_id, limit=20))

# 趋势
await gather(api.trends("news"))

主要特性

1. 多 API 支持

  • 搜索 API：标准 Twitter 搜索功能
  • GraphQL API：高级查询和数据提取
  • 自动切换：基于速率限制和可用性

2. 异步/等待架构

# 并行抓取
async for tweet in api.search("elon musk"):
    print(tweet.id, tweet.user.username, tweet.rawContent)

3. 账号管理

  • 添加多个账号进行轮换
  • 自动处理速率限制
  • 跨运行会话持久化
  • 支持邮件验证（IMAP 或手动）

4. 数据模型

  • 兼容 SNScrape 的模型
  • 易于转换为字典/JSON
  • 可访问原始 API 响应

核心 API 方法

搜索操作

search(query, limit, kv={})

按查询字符串搜索推文。

参数：

  • query (str)：搜索查询（支持 Twitter 搜索语法）
  • limit (int)：要返回的最大推文数
  • kv (dict)：附加参数（例如，{"product": "Top"} 用于热门推文）

返回： 推文对象的异步迭代器

示例：

# 最新推文
async for tweet in api.search("elon musk", limit=20):
    print(tweet.rawContent)

# 热门推文
await gather(api.search("python", limit=20, kv={"product": "Top"}))

用户操作

user_by_login(username)

通过用户名获取用户信息。

示例：

user = await api.user_by_login("xdevelopers")
print(user.id, user.displayname, user.followersCount)

user_by_id(user_id)

通过用户 ID 获取用户信息。

followers(user_id, limit)

获取用户的关注者。

following(user_id, limit)

获取用户关注的用户。

verified_followers(user_id, limit)

仅获取已验证的关注者。

subscriptions(user_id, limit)

获取用户的 Twitter Blue 订阅。

推文操作

tweet_details(tweet_id)

获取特定推文的详细信息。

tweet_replies(tweet_id, limit)

获取对推文的回复。

retweeters(tweet_id, limit)

获取转发特定推文的用户。

user_tweets(user_id, limit)

获取用户的推文（不包括回复）。

user_tweets_and_replies(user_id, limit)

获取用户的推文和回复。

user_media(user_id, limit)

获取用户带有媒体的推文。

其他操作

list_timeline(list_id)

获取 Twitter 列表中的推文。

trends(category)

按类别获取趋势话题。

类别： “news”、“sport”、“entertainment” 等。

账号管理

添加账号

使用 cookies（推荐）：

cookies = "abc=12; ct0=xyz"  # 字符串或 JSON 格式
await api.pool.add_account("user", "pass", "email@example.com", "mail_pass", cookies=cookies)

使用凭据：

await api.pool.add_account("user", "pass", "email@example.com", "mail_pass")
await api.pool.login_all()

CLI 账号管理

# 从文件添加账号
twscrape add_accounts accounts.txt username:password:email:email_password

# 登录所有账号
twscrape login_accounts

# 手动邮件验证
twscrape login_accounts --manual

# 列出账号和状态
twscrape accounts

# 重新登录特定账号
twscrape relogin user1 user2

# 重试失败的登录
twscrape relogin_failed

代理配置

每个账号的代理

proxy = "http://login:pass@example.com:8080"
await api.pool.add_account("user", "pass", "email@example.com", "mail_pass", proxy=proxy)

全局代理

api = API(proxy="http://login:pass@example.com:8080")

环境变量

export TWS_PROXY=socks5://user:pass@127.0.0.1:1080
twscrape search "elon musk"

动态代理更改

api.proxy = "socks5://user:pass@127.0.0.1:1080"
doc = await api.user_by_login("elonmusk")
api.proxy = None  # 禁用代理

优先级： api.proxy > TWS_PROXY 环境变量 > 账号特定代理

CLI 使用

搜索操作

twscrape search "QUERY" --limit=20
twscrape search "elon musk lang:es" --limit=20 > data.txt
twscrape search "python" --limit=20 --raw  # 原始 API 响应

用户操作

twscrape user_by_login USERNAME
twscrape user_by_id USER_ID
twscrape followers USER_ID --limit=20
twscrape following USER_ID --limit=20
twscrape verified_followers USER_ID --limit=20
twscrape user_tweets USER_ID --limit=20

推文操作

twscrape tweet_details TWEET_ID
twscrape tweet_replies TWEET_ID --limit=20
twscrape retweeters TWEET_ID --limit=20

趋势

twscrape trends sport
twscrape trends news

自定义数据库

twscrape --db custom-accounts.db <command>

高级用法

原始 API 响应

async for response in api.search_raw("elon musk"):
    print(response.status_code, response.json())

停止迭代

from contextlib import aclosing

async with aclosing(api.search("elon musk")) as gen:
    async for tweet in gen:
        if tweet.id < 200:
            break

将模型转换为字典/JSON

user = await api.user_by_id(user_id)
user_dict = user.dict()
user_json = user.json()

启用调试日志

from twscrape.logger import set_log_level
set_log_level("DEBUG")

环境变量

  • TWS_PROXY：所有账号的全局代理 示例：socks5://user:pass@127.0.0.1:1080

  • TWS_WAIT_EMAIL_CODE：邮件验证的超时时间（默认：30 秒）

  • TWS_RAISE_WHEN_NO_ACCOUNT：当没有可用账号时引发异常而不是等待 值：false、0、true、1（默认：false）

速率限制与局限性

速率限制

  • 每个端点的速率限制每 15 分钟重置一次
  • 每个账号针对不同操作有独立的限制
  • 当达到限制时，账号会自动轮换

推文限制

  • user_tweets 和 user_tweets_and_replies 每个用户最多返回约 3,200 条推文
  • 这是 Twitter/X 平台的限制

账号状态

  • 速率限制因以下因素而异：
    • 账号年龄
    • 账号验证状态
    • 账号活动历史

处理速率限制

该库自动：

  • 切换到下一个可用账号
  • 如果所有账号都用尽，则等待速率限制重置
  • 跟踪每个端点的速率限制状态

常见模式

大规模数据收集

async def collect_user_data(username):
    user = await api.user_by_login(username)

    # 收集推文
    tweets = await gather(api.user_tweets(user.id, limit=100))

    # 收集关注者
    followers = await gather(api.followers(user.id, limit=100))

    # 收集正在关注
    following = await gather(api.following(user.id, limit=100))

    return {
        'user': user,
        'tweets': tweets,
        'followers': followers,
        'following': following
    }

带过滤器的搜索

# 语言过滤器
await gather(api.search("python lang:en", limit=20))

# 日期过滤器
await gather(api.search("AI since:2024-01-01", limit=20))

# 来自特定用户
await gather(api.search("from:elonmusk", limit=20))

# 带媒体
await gather(api.search("cats filter:media", limit=20))

批处理

async def process_users(usernames):
    tasks = []
    for username in usernames:
        task = api.user_by_login(username)
        tasks.append(task)

    users = await asyncio.gather(*tasks)
    return users

故障排除

登录问题

  • 使用 cookies 而不是凭据以获得更稳定的身份验证
  • 使用 --manual 标志启用手动邮件验证
  • 检查邮件密码是否正确以进行 IMAP 访问

速率限制问题

  • 添加更多账号以更好地轮换
  • 增加请求之间的等待时间
  • 使用 twscrape accounts 监控账号状态

未返回数据

  • 检查账号状态 - 它们可能被暂停或受到速率限制
  • 验证查询语法 - 使用 Twitter 搜索语法
  • 尝试不同的账号 - 有些可能有更好的访问权限

连接问题

  • 配置代理（如果在防火墙后面）
  • 检查网络连接
  • 验证 Twitter/X 是否可以从您的位置访问

资源

参考资料

有关详细的 API 文档和示例，请参阅 references/ 目录中的参考文件：

  • references/installation.md - 安装和设置
  • references/api_methods.md - 完整的 API 方法参考
  • references/account_management.md - 账号配置和管理
  • references/cli_usage.md - 命令行界面指南
  • references/proxy_config.md - 代理配置选项
  • references/examples.md - 代码示例和模式

仓库：https://github.com/vladkens/twscrape 星标：1998+ 语言：Python 许可证：MIT