⚡

Twitter数据抓取工具 twscrape

twscrape是一个专业的Python库，专门用于Twitter/X社交媒体数据抓取和爬虫开发。通过GraphQL API实现高效数据采集，支持多账号轮换管理、异步并行操作、代理配置和会话持久化。适用于社交媒体监控、舆情分析、用户行为研究、趋势挖掘和大规模数据收集场景。关键词：Twitter爬虫、社交媒体数据采集、Python数据抓取、账号轮换、GraphQL API、异步爬虫、数据挖掘、舆情监控。

数据工程 0 次安装 0 次浏览 更新于 3/1/2026

名称: twscrape 描述: 使用GraphQL API进行Twitter/X数据抓取的Python库，支持账号轮换和会话管理。适用于提取推文、用户资料、粉丝、趋势，或构建社交媒体监控工具。

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"} 用于热门推文）

返回: Tweet对象的异步迭代器

示例:

# 最新推文
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