ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

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

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 tweepy
git 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_token="YOUR_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_token="YOUR_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( text="This Tweet was Tweeted using Tweepy and Twitter API v2!" ) print(f"https://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_token="YOUR_BEARER_TOKEN") user = client.get_user(username="Twitter", 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 数据流

典型诉求是实时捕获含特定关键词的推文,对应StreamingClient:

class KeywordStream(tweepy.StreamingClient): def on_tweet(self, tweet): print(tweet.text) def on_exception(self, e): print("stream error:", e) stream = KeywordStream(bearer_token="YOUR_BEARER_TOKEN") stream.add_rules(tweepy.StreamRule("python")) stream.filter()

StreamRule定义匹配规则,filter()阻塞运行,命中后回调on_tweet,on_exception负责断线告警。实现细节在 tweepy/streaming.py,文档的 docs/streaming.rst 覆盖了更多回调钩子。

进阶扩展:分页、异步与模型映射

批量采集时手写游标很易错,Paginator把这件事收敛成一个 for 循环,它会自动从响应 meta 里读next_token并在下一轮请求中带上:

for response in tweepy.Paginator(client.get_users_tweets, id=user_id, max_results=100): 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里逐一对照即可。

常见问题与调优

  1. 发布时抛 401/403:现象是create_tweet报Unauthorized或Forbidden;原因是应用缺少 Write 权限,或改过权限后未重新签发令牌。解法是在开发者门户调整权限,重新生成 access token 并再次授权。
  2. 批量请求撞上 429:现象是TooManyRequests异常;原因是超过了单端点配额。建议重试前先确认剩余配额、退避等待并降低频率,处理模式可参考 examples/API_v1/rate_limit_handling.py。
  3. 分页只拿到前 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),仅供参考

返回列表