后端开发教程 - Java、Spring Boot 实战 - msg200.com
使用 Twitter APIv2 🚀🚀 在 Python 中创建 Twitter 机器人
当我将@poet_this 的Twitter 机器人迁移到 Twitter API v2 时，我找不到任何好的教程，至少没有一个能正常工作的。在机器人成功之后，很多人联系我，问我如何自己创建一个 Twitter 机器人！我很乐意帮忙 :)
🤔 我们正在构建什么
让我们创建一个 Twitter 机器人，每当有人提及它时，它都会回复。为了简单起见，我们只用相同的文字回复，但要大写。
我们想要做的：
（original_uploader）：这是一条很酷的推文
（评论者）：@ 机器人（*提及机器人）
（机器人）：这是一条很酷的推文
在本教程的最后，我还将介绍如何添加更多酷炫功能，以及一些有关您自己的酷炫 Twitter 机器人的想法！
⚡ 入门
首先，你需要一个 Twitter 开发者账号。在Twitter 开发者面板上注册一个。
好吧，一旦我们完成了......
让我们创建一个新的python项目！
设置项目
我建议在虚拟环境中工作。创建一个新文件夹，打开终端并使用此命令
python -m venv env
env/Scripts/activate // on windows
source env/bin/activate // on linux
最后我们只需要安装依赖项，
pip install python-dotenv python-twitter-v2
这是我们现在真正需要的仅有的两个依赖项。python-dotenv 用于使用环境变量，python-twitter-v2 用于与 twitter API 交互！
🔐 使用 Twitter 进行身份验证
要通过机器人的 Twitter 帐户对 Twitter API 进行身份验证，我们需要一些东西......
- 消费者密钥（来自开发者仪表板）
- 消费者秘密（来自开发者仪表板）
- 访问令牌
- 访问令牌秘密
从技术上讲，您也可以使用承载令牌，但现在让我们保持简单。
要获取访问令牌和密钥（用于测试），您可以在此处的仪表板中创建它们：
确保您具有读取和写入权限，我们需要这些权限用于机器人。
（如果您想控制另一个帐户，例如机器人帐户，请按照以下步骤获取另一个帐户的访问令牌/密钥 - https://developer.twitter.com/en/docs/authentication/oauth-1-0a/obtaining-user-access-tokens）
一旦我们拥有了所有这些令牌，就创建一个文件，.env并以这种方式填写它 -
CONSUMER_KEY=...
CONSUMER_SECRET=...
ACCESS_TOKEN=...
ACCESS_TOKEN_SECRET=...
👨🏻💻现在就开始代码吧！
让我们把问题分解成小部分......
我们希望，
- 使用 Twitter API 进行身份验证
- 聆听所有提及
- 回复！
太棒了，首先，让我们进行身份验证
import pytwitter
from os import environ as env
from dotenv import load_dotenv
load_dotenv() # Loads the .env file we created earlier
api = pytwitter.Api(
consumer_key=env["CONSUMER_KEY"],
consumer_secret=env["CONSUMER_SECRET"],
access_token=env["OAUTH_TOKEN"],
access_secret=env["OAUTH_TOKEN_SECRET"],
)
您可以使用某些端点检查机器人是否已正确验证，例如，
print(api.get_user(username="Twitter"))
# Should print Response(data=User(id='783214', name='Twitter', username='Twitter'))
现在让我们来听听提及。
不过有一点需要注意——我们不想回复所有提及。这是因为 Twitter 的运作方式。
假设我发了一条推文，有人对其评论@DhravyaShah：Hello world！@SomeOtherRandomUser：Hi Dhravya！
但是，它不仅仅是“Hi Dhravya”，而是看起来像这样@DhravyaShah：Hello world！@SomeOtherRandomUser：@DhravyaShah Hi Dhravya！
注意到区别了吗？这在 Twitter 机器人中可能是一个问题，因为它甚至会回复机器人下方的评论。
我们会通过检查文本是否只提到了机器人，而没有其他内容来解决这个问题。
怎么做？只需删除所有提及项，然后检查是否为空字符串即可！
last_tweet_id = 0 # Just a variable to keep track of the last tweet we replied to.
if __name__ == "__main__":
while True:
try:
mentions = api.get_mentions(
user_id=str(BOT_ID), # get bot id from tweeterid.com
return_json=True,
since_id=str(last_tweet_id),
tweet_fields=[
"conversation_id",
"entities"
]
)
if not isinstance(mentions, dict):
continue
太棒了，现在我们得到了源源不断的提及，其实不然。我们得到了之后的所有提及since_id。
这些提及实际上包含大量数据。为了简洁起见，我不会详细介绍，但您可以随意print()浏览一下所有提供的字段！
现在让我们检查一下是否有任何data并且它includes也有（includes有所有额外的实体）
if not "data" in mentions.keys():
continue # There's no more tweets
for tweet in mentions["data"]:
text = tweet["text"]
reply_to = tweet["id"]
# If it's not a reply to another tweet
if not (tweet["conversation_id"] == tweet["id"]):
if str(tweet["in_reply_to_user_id"]) == str(BOT_ID):
continue
# Get the parent tweet
tweet_ = api.get_tweet(return_json=True,tweet_id=tweet["referenced_tweets"][0]["id"])
text = tweet_["text"]
# If tweet is a reply, it's in the format "@user @user @bot"
users = [
mention["username"] for mention in tweet["entities"]["mentions"]
]
new_txt = tweet["text"]
# Remove all mentions in the start
for user in users:
new_txt = new_txt.replace(f"@{user}", "")
new_txt = new_txt.strip()
如果new_txt为“”（空），则表示这是“直接提及”——仅仅是一条“ @bot ”评论。如果是直接提及，则表示已明确调用该机器人，这意味着我们可以做出回应。
if (new_txt == ""):
api.create_tweet(text=text.upper(), reply_in_reply_to_tweet_id = reply_to
最后，让我们结束循环，并将 last_tweet_id 更新为新的 last_tweet_id
if "meta" in mentions.keys():
if "newest_id" in mentions["meta"].keys():
last_tweet_id = mentions["meta"]["newest_id"]
time.sleep(10)
基本上就是这样！以下是Github gist中的完整代码
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
| import pytwitter | |
| from os import environ as env | |
| from dotenv import load_dotenv | |
| load_dotenv() # Loads the .env file we created earlier | |
| api = pytwitter.Api( | |
| consumer_key=env["CONSUMER_KEY"], | |
| consumer_secret=env["CONSUMER_SECRET"], | |
| access_token=env["OAUTH_TOKEN"], | |
| access_secret=env["OAUTH_TOKEN_SECRET"], | |
| ) | |
| # print(api.get_user(username="Twitter")) # To test the api connection | |
| last_tweet_id = 0 # Just a variable to keep track of the last tweet we replied to. | |
| if __name__ == "__main__": | |
| while True: | |
| mentions = api.get_mentions( | |
| user_id=str(BOT_ID), # get bot id from tweeterid.com | |
| return_json=True, | |
| since_id=str(last_tweet_id), | |
| tweet_fields=[ | |
| "conversation_id", | |
| "entities", | |
| "in_reply_to_user_id", | |
| "referenced_tweets" | |
| ] | |
| ) | |
| if not isinstance(mentions, dict): | |
| continue | |
| if not "data" in mentions.keys(): | |
| continue # There's no more tweets | |
| for tweet in mentions["data"]: | |
| text = tweet["text"] | |
| reply_to = tweet["id"] | |
| # If it's not a reply to another tweet | |
| if not (tweet["conversation_id"] == tweet["id"]): | |
| if str(tweet["in_reply_to_user_id"]) == str(BOT_ID): | |
| continue | |
| # Get the parent tweet | |
| tweet_ = api.get_tweet(return_json=True,tweet_id=tweet["referenced_tweets"][0]["id"]) | |
| text = tweet_["text"] | |
| # If tweet is a reply, it's in the format "@user @user @bot" | |
| users = [ | |
| mention["username"] for mention in tweet["entities"]["mentions"] | |
| ] | |
| new_txt = tweet["text"] | |
| # Remove all mentions in the start | |
| for user in users: | |
| new_txt = new_txt.replace(f"@{user}", "") | |
| new_txt = new_txt.strip() | |
| if (new_txt == ""): | |
| api.create_tweet(text=text.upper(), reply_in_reply_to_tweet_id = reply_to) | |
| if "meta" in mentions.keys(): | |
| if "newest_id" in mentions["meta"].keys(): | |
| last_tweet_id = mentions["meta"]["newest_id"] | |
| time.sleep(10) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import pytwitter | |
| from os import environ as env | |
| from dotenv import load_dotenv | |
| load_dotenv() # Loads the .env file we created earlier | |
| api = pytwitter.Api( | |
| consumer_key=env["CONSUMER_KEY"], | |
| consumer_secret=env["CONSUMER_SECRET"], | |
| access_token=env["OAUTH_TOKEN"], | |
| access_secret=env["OAUTH_TOKEN_SECRET"], | |
| ) | |
| # print(api.get_user(username="Twitter")) # To test the api connection | |
| last_tweet_id = 0 # Just a variable to keep track of the last tweet we replied to. | |
| if __name__ == "__main__": | |
| while True: | |
| mentions = api.get_mentions( | |
| user_id=str(BOT_ID), # get bot id from tweeterid.com | |
| return_json=True, | |
| since_id=str(last_tweet_id), | |
| tweet_fields=[ | |
| "conversation_id", | |
| "entities", | |
| "in_reply_to_user_id", | |
| "referenced_tweets" | |
| ] | |
| ) | |
| if not isinstance(mentions, dict): | |
| continue | |
| if not "data" in mentions.keys(): | |
| continue # There's no more tweets | |
| for tweet in mentions["data"]: | |
| text = tweet["text"] | |
| reply_to = tweet["id"] | |
| # If it's not a reply to another tweet | |
| if not (tweet["conversation_id"] == tweet["id"]): | |
| if str(tweet["in_reply_to_user_id"]) == str(BOT_ID): | |
| continue | |
| # Get the parent tweet | |
| tweet_ = api.get_tweet(return_json=True,tweet_id=tweet["referenced_tweets"][0]["id"]) | |
| text = tweet_["text"] | |
| # If tweet is a reply, it's in the format "@user @user @bot" | |
| users = [ | |
| mention["username"] for mention in tweet["entities"]["mentions"] | |
| ] | |
| new_txt = tweet["text"] | |
| # Remove all mentions in the start | |
| for user in users: | |
| new_txt = new_txt.replace(f"@{user}", "") | |
| new_txt = new_txt.strip() | |
| if (new_txt == ""): | |
| api.create_tweet(text=text.upper(), reply_in_reply_to_tweet_id = reply_to) | |
| if "meta" in mentions.keys(): | |
| if "newest_id" in mentions["meta"].keys(): | |
| last_tweet_id = mentions["meta"]["newest_id"] | |
| time.sleep(10) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
| import pytwitter | |
| from os import environ as env | |
| from dotenv import load_dotenv | |
| load_dotenv() # Loads the .env file we created earlier | |
| api = pytwitter.Api( | |
| consumer_key=env["CONSUMER_KEY"], | |
| consumer_secret=env["CONSUMER_SECRET"], | |
| access_token=env["OAUTH_TOKEN"], | |
| access_secret=env["OAUTH_TOKEN_SECRET"], | |
| ) | |
| # print(api.get_user(username="Twitter")) # To test the api connection | |
| last_tweet_id = 0 # Just a variable to keep track of the last tweet we replied to. | |
| if __name__ == "__main__": | |
| while True: | |
| mentions = api.get_mentions( | |
| user_id=str(BOT_ID), # get bot id from tweeterid.com | |
| return_json=True, | |
| since_id=str(last_tweet_id), | |
| tweet_fields=[ | |
| "conversation_id", | |
| "entities", | |
| "in_reply_to_user_id", | |
| "referenced_tweets" | |
| ] | |
| ) | |
| if not isinstance(mentions, dict): | |
| continue | |
| if not "data" in mentions.keys(): | |
| continue # There's no more tweets | |
| for tweet in mentions["data"]: | |
| text = tweet["text"] | |
| reply_to = tweet["id"] | |
| # If it's not a reply to another tweet | |
| if not (tweet["conversation_id"] == tweet["id"]): | |
| if str(tweet["in_reply_to_user_id"]) == str(BOT_ID): | |
| continue | |
| # Get the parent tweet | |
| tweet_ = api.get_tweet(return_json=True,tweet_id=tweet["referenced_tweets"][0]["id"]) | |
| text = tweet_["text"] | |
| # If tweet is a reply, it's in the format "@user @user @bot" | |
| users = [ | |
| mention["username"] for mention in tweet["entities"]["mentions"] | |
| ] | |
| new_txt = tweet["text"] | |
| # Remove all mentions in the start | |
| for user in users: | |
| new_txt = new_txt.replace(f"@{user}", "") | |
| new_txt = new_txt.strip() | |
| if (new_txt == ""): | |
| api.create_tweet(text=text.upper(), reply_in_reply_to_tweet_id = reply_to) | |
| if "meta" in mentions.keys(): | |
| if "newest_id" in mentions["meta"].keys(): | |
| last_tweet_id = mentions["meta"]["newest_id"] | |
| time.sleep(10) |
您可以通过添加“关键词”为机器人添加更多功能，就像@poet_this在 Twitter 上的工作方式一样 - 评论“ @poet_this blue 2”（蓝色 2 是关键词）生成特定风格的图像。
最后...
以上只是简单的概述，现在……
不妨自己动手做一个机器人！这是一个非常有趣且回报丰厚的过程，你会从中学到很多东西。
感谢阅读！
您可以通过 Github 赞助我来支持我 - https://github.com/sponsors/Dhravya
您也可以在 Twitter 上关注我 - https://twitter.com/dhravyashah