Tweepy API：把 Python 接进 Twitter 的读写与实时流

Tweepy API把 Python 接进 Twitter 的读写与实时流【免费下载链接】tweepyTwitter for Python!项目地址: https://gitcode.com/gh_mirrors/tw/tweepy当你的 Python 应用需要拉取推文、查询用户资料、发布内容或者监听实时关键词时直接用 Tweepy API 去对接 Twitter API v2 端点是最省事的路线。Tweepy 是 Python 生态里使用最广的 Twitter 第三方封装同步 Client、异步 AsyncClient、StreamingClient 与分页 Paginator 都是内置件签名、响应解析这类脏活不必自己写。项目定位Tweepy API 为什么值得选Tweepy 在这个领域几乎没有直接竞品它的差异化不在性能而在覆盖面与 API 设计一套库同时维护 v1.1 与 v2 两套端点Client的方法名基本与 Twitter 官方文档一一对应降低查文档成本返回的不是裸 JSON而是解析好的模型对象tweet.text、user.name这类属性访问比手动取字典舒服得多。同步与异步客户端方法名保持一致迁移成本低。阅读源码建议从 tweepy/client.py 入手它是 v2 同步客户端的完整实现[tweepy/__init__.py](https://link.gitcode.com/i/2a4baacf83c32516db2580012ff37da1)则汇总了Client、Paginator、StreamingClient等全部对外导出扫一眼即可建立全局印象。环境搭建与首次运行前置条件不多Python 3.6 及以上版本Twitter 开发者账号并在开发者门户创建应用只读场景准备 bearer token 即可涉及发布等写操作则需要 OAuth 1.0a 的 consumer key/secret 与 access token/secret 四件套安装与获取源码各一条命令pip install tweepygit clone https://gitcode.com/gh_mirrors/tw/tweepy仓库顶层结构很清晰核心封装在 tweepy/客户端、流、分页各自拆成独立模块examples/ 按 API 版本分API_v1与API_v2两档docs/ 是 Sphinx 文档源tests/ 是测试用例配合cassettes/里录制的请求/响应数据离线也能跑集成测试。认证在构建客户端时完成两种方式按场景二选一import tweepy # 只读场景bearer token 即可 client tweepy.Client(bearer_tokenYOUR_BEARER_TOKEN) # 写操作OAuth 1.0a 四件套 client tweepy.Client(consumer_key..., consumer_secret..., access_token..., access_token_secret...)最小验证用 bearer token 调一次get_me能打印出账号名即环境就绪import tweepy client tweepy.Client(bearer_tokenYOUR_BEARER_TOKEN) print(tweepy.__version__) print(client.get_me().data.username)核心能力实战Tweepy API 的读、写、流发布推文从构造客户端到拿到新 ID典型诉求是内容自动发布或机器人回帖对应create_tweet方法client tweepy.Client( consumer_key..., consumer_secret..., access_token..., access_token_secret... ) response client.create_tweet( textThis Tweet was Tweeted using Tweepy and Twitter API v2! ) print(fhttps://twitter.com/user/status/{response.data[id]})create_tweet返回Response对象response.data[id]即新推文 ID出错时它不返回错误码而是抛异常401/403/429 等各有对应类型异常体系定义在 tweepy/errors.py。完整可运行版本参考 examples/API_v2/create_tweet.py其中还演示了发布到 Community 的写法。拉取推文与用户档案用 fields 控制返回典型诉求是采集推文文本或用户粉丝数对应get_tweet/get_tweets/get_user这一族读方法client tweepy.Client(bearer_tokenYOUR_BEARER_TOKEN) user client.get_user(usernameTwitter, user_fields[public_metrics]).data print(user.name, user.public_metrics[followers_count]) tweets client.get_tweets(ids[1234567890], tweet_fields[created_at]).data print(tweets[0].text)user_fields/tweet_fields参数决定服务端返回哪些字段不传时默认只有 id、name 等基础项基础字段可以属性访问额外字段以字典形式挂在对象上。示例见 examples/API_v2/get_users.py。监听关键词搭一条 Twitter 数据流典型诉求是实时捕获含特定关键词的推文对应StreamingClientclass KeywordStream(tweepy.StreamingClient): def on_tweet(self, tweet): print(tweet.text) def on_exception(self, e): print(stream error:, e) stream KeywordStream(bearer_tokenYOUR_BEARER_TOKEN) stream.add_rules(tweepy.StreamRule(python)) stream.filter()StreamRule定义匹配规则filter()阻塞运行命中后回调on_tweeton_exception负责断线告警。实现细节在 tweepy/streaming.py文档的 docs/streaming.rst 覆盖了更多回调钩子。进阶扩展分页、异步与模型映射批量采集时手写游标很易错Paginator把这件事收敛成一个 for 循环它会自动从响应 meta 里读next_token并在下一轮请求中带上for response in tweepy.Paginator(client.get_users_tweets, iduser_id, max_results100): for tweet in response.data: print(tweet.id, tweet.text)源码见 tweepy/pagination.py。并发量上来后可以把同样的代码平移到 tweepy/asynchronous/client.py 中的AsyncClient方法名与同步版一致加个await即可流式也有对应的异步版本。至于tweet、user这些返回对象背后的Tweet/User/Space类定义在tweepy/models.py里逐一对照即可。常见问题与调优发布时抛 401/403现象是create_tweet报Unauthorized或Forbidden原因是应用缺少 Write 权限或改过权限后未重新签发令牌。解法是在开发者门户调整权限重新生成 access token 并再次授权。批量请求撞上 429现象是TooManyRequests异常原因是超过了单端点配额。建议重试前先确认剩余配额、退避等待并降低频率处理模式可参考 examples/API_v1/rate_limit_handling.py。分页只拿到前 100 条现象是循环第一轮后就没有数据原因是没有回传next_token。用Paginator自动处理或手动把meta[next_token]传给下一轮请求。延伸学习地图 docs/Sphinx 文档源client、streaming、pagination 各有独立章节docs/changelog.md版本变更升级前建议过一遍cassettes/录制的请求/响应数据调试时可直接对照tweepy/asynchronous/异步客户端、分页与流的完整实现熟悉以上路径后一个合理的下一步是先用Paginator把某位用户的近期推文落成一份采集脚本再叠加StreamingClient做实时监听等并发成为瓶颈时再整体切换到AsyncClient。【免费下载链接】tweepyTwitter for Python!项目地址: https://gitcode.com/gh_mirrors/tw/tweepy创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考