Twitter——我们的机器人相互交流之地
全文共4713字，预计学习时长9分钟
图片由Jonny Lindner拍摄，源自Pixabay
阅读此文，看看如何用Python创建Twitter机器人来进行一项有争议的实验的。本文包含所有的代码和详细说明。
摘要
本文会详细说明我是如何创建一个Twitter聊天机器人的，它能利用一些人类的基本情感来创造粉丝。这种事在我们身边每天都会发生。我认为重要的是要让大家知道这很容易。我会分享一些有趣的经历，并向大家展示在Twitter中插入机器人的实例。
本文后半部分对机器人的代码和功能进行了详细描述。Github上有当前的产品代码。我在AWS上运行机器人，并通过Serverless架构对其进行部署，这连AWS每月免费层级的5%都用不到。所以值得一试。
动机
图片由Patrick Fore拍摄，源自Unsplash
我想写一些文章，帮助人们学会使用Python，这样也能帮助拓宽他们的技能。部分原因是一些团队成员想学习Python。开始的时候，我告诉自己，先做两个月左右，看看结果如何。如果万不得已，我会找一些结构清晰、脉络清楚的文章来教大家使用Python。相信每个人都可以从中学到一些Python的知识。
图片由Nicholas Green拍摄
Tweepy
图片由Safar Safarov拍摄，源自Unsplas
Tweepy是一个可以访问Twitter API的Python库。 tweepy里的文档看起来很整洁，代码维护得很好。我想试试这个。
凭证
设置身份验证和获取凭证比想象得要更容易：
1.转到Twitter的开发人员页面
2.登录Twitter账号
3.创建一个应用程序并获取凭证（在Keys and Tokens下，参见红色圆圈）
创建一个Twitter应用程序的步骤
机器人的结构
我用了一个简单的API调用测试了Jupyter Notebook中的凭证，一切似乎都没什么问题。现在是时候开始下一步了。我的twitter机器人应该做两件事：
· 创造粉丝
· 宣传Medium的文章
创造粉丝
几年前，我做了一个Instagram机器人的实验，我知道了如何实现第一个目标。这是不道德的，因为它利用了人类渴望被喜欢、被认可的心理，它所创造的一切都是假的。但再说一遍，这是Twitter。此外，我认为有必要聊聊具体的做法。应当向大家演示一下这些机器人是如何工作的，并且其效果如何，这是非常重要的。此外，也很有必要向大家展示，这些机器人每天都在广泛使用。
机器人的工作方式是——给予人们认可和关注:
1.与用户互动（比如转发，评论他们的推文，并关注他们）
2.等待并观察
3.看他们回关你
4.再等一段时间，然后取关他们
因此，暂且抛开所有的伦理问题，下面是相应的代码。
①与用户互动
在项目中，我通常会将配置模块用作配置设定的抽象层。
import os
import yaml as _yaml
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
defget_config():
config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'production.yml')
try:
withopen(config_path) as config_file:
return _yaml.load(config_file)
exceptFileNotFoundError:
logger.error(f'You probably forgot to create a production.yml, as we could not find {config_path}')
raise
defget_post_data():
data_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'post_data.yml')
withopen(data_path) as config_file:
return _yaml.load(config_file)
twitter_config.py hosted with ❤by GitHub
bots.config
配置如下所示:
# Write Access also
API_KEY : "YOUR API KEY HERE"
API_KEY_SECRET : "YOUR API SECRET HERE"
ACCESS_TOKEN : "YOUR ACCESS TOKEN HERE"
ACCESS_TOKEN_SECRET : "YOUR ACCESS TOKEN SECRET HERE"
twitter_sample_config.yml hosted with ❤by GitHub
production.yml
然后可以设置一个模块来提供Twitter API，如下所示:
import tweepy
from bots.config import get_config
__API=None
defconfigure_twitter_api():
API_KEY= get_config()['API_KEY']
API_KEY_SECRET= get_config()['API_KEY_SECRET']
ACCESS_TOKEN= get_config()['ACCESS_TOKEN']
ACCESS_TOKEN_SECRET= get_config()['ACCESS_TOKEN_SECRET']
auth = tweepy.OAuthHandler(API_KEY, API_KEY_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
api = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True)
return api
defget_twitter_api():
global__API
ifnot__API:
__API= configure_twitter_api()
return__API
twitter_api.py hosted with ❤by GitHub
bots.twitter_api
下面的代码包含了交互逻辑。
import tweepy
from bots.twitter_api import get_twitter_api
import bots.utils as _utils
import datetime
import logging
import random
import time
logger = logging.getLogger()
logger.setLevel(logging.INFO)
COMMENTS= [
'Nice piece!', 'Interesting', '', 'I am going to read up on this', 'Thanks for sharing!', 'This is helpful',
'Insightful', 'thought-provoking', 'Will check this out'
]
HASHTAG_SETS= [
{'Python', 'DataScience', 'Machinelearning'},
{'Python', 'Keras'},
{'Python', 'DataScience'},
{'Python', 'Pandas'},
{'Python', 'PyTorch', 'Machinelearning'},
{'Python', 'Scikitlearn'},
{'Python', 'Statisitcs'},
]
deffetch_most_original_tweets(user):
results = []
for tweet in get_twitter_api().user_timeline(user.id, count=20):
ifnot (tweet.retweeted or tweet.in_reply_to_status_id):
tweet.score = score_tweet(tweet)
results.append(tweet)
return results
definteract_with_user(user, following_history, hashtags):
ifnot user.following:
logger.info(f"Following {user.name}")
user.follow()
following_history[user.id_str] = {'followed_at': datetime.datetime.now().isoformat()}
user_tweets =sorted(fetch_most_original_tweets(user), key=lambda x: x.score, reverse=True)
iflen(user_tweets) >0:
interactions =0
for tweet in user_tweets:
tags = {tag['text'].lower() for tag in tweet.entities.get('hashtags')}
lower_given_tag = {tag.lower() for tag in hashtags}
for given_tag in lower_given_tag:
if given_tag in tweet.text.lower():
found_tag_in_text =True
break
else:
found_tag_in_text =False
if (len(tags & lower_given_tag) >0) or found_tag_in_text:
interaction =0
if random.random() >0.95:
comment =f'@{user.screen_name}{random.choice(COMMENTS)}'
logger.info(f"Commenting: {tweet.id} with: {comment}")
get_twitter_api().update_status(
comment,
in_reply_to_status_id=tweet.id_str,
auto_populate_reply_metadata=True
)
time.sleep(random.random()/2)
interaction |=1
ifnot tweet.favorited and (random.random() >.5) and tweet.lang =='en':
logger.info(f"Hearting: {tweet.id} with text: {tweet.text}")
get_twitter_api().create_favorite(tweet.id)
time.sleep(random.random() *5)
interaction |=1
if random.random() >0.95:
logger.info(f"Retweeting: {tweet.id}")
logger.info(f"Text: {tweet.text}")
get_twitter_api().retweet(tweet.id)
time.sleep(random.random())
interaction |=1
interactions += interaction
if interactions ==2:
break
defscore_tweet(tweet):
favorites = _utils.scaled_sigmoid(x=-tweet.favorite_count, stretch=2, max_score=50, center=3)
retweets = _utils.scaled_sigmoid(x=-tweet.retweet_count, stretch=1, max_score=50, center=2)
age = _utils.created_at_score(tweet, stretch=2, max_score=30, center=3)
score = favorites + retweets + age
return score
defscore_user(user):
followed_to_following = _utils.followed_to_following_ratio(user)
followers = _utils.scaled_sigmoid(x=-user.followers_count, stretch=200, max_score=100, center=300)
age = _utils.created_at_score(user, stretch=50, max_score=30, center=60)
score = followed_to_following + followers + age
return score
defget_users_from_recent_tweets(cnt=10, hashtags=None):
q =' AND '.join([f'#{tag}'for tag in hashtags])
users = []
for tweet in tweepy.Cursor(get_twitter_api().search, q=q, lang="en", count=cnt, result_type='recent').items(cnt):
users.append(tweet.user)
return users
deffetchfollow(event=None, context=None):
hashtags = random.choice(HASHTAG_SETS)
# monkey-patch the tweepy User class by adding a hashfunction, which we will need to quickly get unique users
tweepy.models.User.__hash__=lambda self: hash(self.id_str)
users =list(set(get_users_from_recent_tweets(cnt=250, hashtags=hashtags)))
# score users
for user in users:
user.score = score_user(user)
# sort users by score
users =sorted(users, key=lambda x: x.score, reverse=True)
logger.info(f"Found {len(users)}")
following_history = _utils.get_s3_data('following.json')
max_interactions =10
interactions =0
for user in users:
time.sleep(random.random() *10+2)
if user.id_str notin following_history:
try:
logger.info(f"Interacting with {user.name}")
interact_with_user(user, following_history, hashtags)
interactions +=1
exceptExceptionas e:
logger.error(f'Syncing followers history on error: {e}')
_utils.sync_s3_data(following_history)
raise
if interactions >= max_interactions:
break
logger.info('Syncing followers history on ordinary termination')
_utils.sync_s3_data(following_history)
defcomment_tweet(user, tweet):
comment =f'@{user.screen_name}{random.choice(COMMENTS)}'
logger.info(f"Commenting: {tweet.id} with: {comment}")
get_twitter_api().update_status(
comment,
in_reply_to_status_id=tweet.id_str,
auto_populate_reply_metadata=True
)
if__name__=='__main__':
fetchfollow()
fetchfollow.py hosted with ❤by GitHub
bot.fetchfollow
先从两个变量开始：COMMENTS和HASHTAG_SETS，后面也会引用这两个变量，只要给定内容和名称，它们的用法是显而易见的。COMMENTS列表存储了一组通用的积极类的评价，HASHTAG_SETS存储了一系列不同的用于搜索的标签组合。
主要函数是fetchfollow，它执行以下操作:
· 使用HASHTAG_SETS中的随机标签来搜索推特。
· 找到这些推特的用户。根据这些用户的粉丝数量（越少越好），粉丝-关注比率（越低越好）和账号使用时长（越新越好），对用户进行评分，并按得分进行排序，得分最高的（即最有可能会回关你的用户）即为第一，得分最低的即为最后一名。
· 从S3获取following_history，此文件包含了关注每个用户的日期（以及之后对他们取关的日期）。
· 与不在following_history中的用户从最高分到最低分进行互动，（每位最多10次，毕竟我们不想触发机器人警报）。在互动时，给包含我们标签的推特打分，然后随机点赞，评论和转发这些推特。
· 将用户添加到following_history中并更新到S3。毕竟我们不想再关注他们了。
②等待并观察
这个阶段很有趣。这个阶段是把机器人散布到twitter里并观察结果的时候。有时你会觉得很有趣，有时你也会有些困惑。当我在Instagram上试用机器人的时候，我很快就发现了Instagram上有很多色情内容。但这是后话了。
在创建了第一个版本的Twitter机器人之后，我学到了三件事：
ⓐ必须要调整搜索推文的方式，因为最初只搜索Python。
ⓑ必须调整机器人运行的频率，降低行为的确定性。
第一个版本的机器人很快就被拦截了，因为我疯狂地评论并点赞别人的推文，就像《冰河世纪》里的松鼠喝了一杯能量饮料后那样。
第一个应用程序在评论太多之后被限制访问
然而这次，创建一个新的应用程序并采取更谨慎的方法就相当容易了。
ⓒTwitter上有很多机器人。我得到的回复是，“嘿，谢谢你关注我。请查看我经常使用的这项不错的服务：https://xxxbots.xx”。恭喜他们，他们很聪明，采用了一种病毒式的营销方法。
机器人们对我的机器人作出了反应，消息列表还在继续增加
③看他们回关你
在过去的四个星期里，我的Twitter账号积累添加了大约600个粉丝，除了偶尔在发布Medium的帖子列表中添加一个新条目之外，我什么也没做。
④等过段时间，然后取关他们
既然不想关注太多人，那必须时不时地取关一些人，以保持平衡。
import bots.utils as _utils
from dateutil.parser import parse
from bots.twitter_api import get_twitter_api
import random
import logging
import time
import datetime
logger = logging.getLogger()
logger.setLevel(logging.INFO)
defunfollow(event=None, context=None):
if random.random() >.23:
logger.info(f'Doing nothing this time')
else:
following_history = _utils.get_s3_data('following.json')
sorted_by_following_date =sorted(
[elem for elem in following_history.items() if'unfollowed_at'notin elem[1]],
key=lambda x: parse(x[1]['followed_at'])
)
number_to_unfollow = random.randint(1, 3)
for currently_following in sorted_by_following_date[:number_to_unfollow]:
_id = currently_following[0]
try:
print(_id)
get_twitter_api().destroy_friendship(_id)
following_history[_id]['unfollowed_at'] = datetime.datetime.now().isoformat()
logger.info(f'Unfollowing: {_id}')
exceptExceptionas e:
logger.error(f'Unfollowing: {_id} did not work with error {e}')
time.sleep(random.randint(2, 8))
_utils.sync_s3_data(following_history)
twitter_unfollow.py hosted with ❤by GitHub
bots.unfollow
取关函数在执行时，首先获取先前上传的following_history，再根据关注日期，对所有未取关的用户进行升序排序。对于排名前三的用户，调用destroy_friendship （取关功能）。这个名字是我自己取的。那么该函数将更新following_history，然后准备再次调用。
宣传Medium的文章
这一部分直截了当，当然，在伦理道德方面也是无须质疑的。
from collections import namedtuple
from bots.twitter_api import get_twitter_api
import random
import logging
from bots.config import get_post_data
logger = logging.getLogger()
logger.setLevel(logging.INFO)
classMediumPost(namedtuple('MediumPost', ['id', 'url', 'tags', 'text'])):
defmake_post(self):
used_tags =self.tags[:random.randint(1, len(self.tags))]
returnf'{self.text}{" ".join(["#"+ tag for tag in used_tags])}{self.url}'
defpost_to_twitter(self):
api = get_twitter_api()
res = api.update_status(self.make_post())
return res
defpost_random_medium_article(event=None, context=None):
posts = [MediumPost(*v) for k, v in get_post_data().items()]
random_post = random.choice(posts)
logger.info(f'Posting: {random_post}')
random_post.post_to_twitter()
if__name__=='__main__':
#posts = [MediumPost(*v) for k, v in get_post_data().items()]
#print(posts)
post_random_medium_article()
twitter_post.py hosted with ❤by GitHub
bots.post
此脚本从引用列表中随机发布一篇文章参考列表如下所示:
Advanced - Visualize Sales Team:
- Advanced - Visualize Sales Team
- https://towardsdatascience.com/how-to-explore-and-visualize-a-dataset-with-python-7da5024900ef
- - Datascience
- BigData
- DataVisualization
- How to visualize a data set!
....
Advanced - Cat, Dog or Elon Musk:
- Advanced - Cat, Dog or Elon Musk
- https://towardsdatascience.com/cat-dog-or-elon-musk-145658489730
- - Datascience
- BigData
- DataAnalytics
- Python
- Automation
- Machine Learning
- Bots
- Learn how to build an image-recognizing convolutional neural network with Python and Keras in less than 15minutes!
post_data.yml hosted with ❤by GitHub
推文样本
部署
图片由elCarito拍摄，源自Unsplash网
我使用了Serverless 架构，利用Lambda函数和预定义的时间表（在serverless.yml中指定）将机器人部署到AWS。
service: fb-TwitterBot
provider:
name: aws
runtime: python3.6
memorySize: 256
timeout: 900
region: ${opt:region, 'eu-central-1'}
stage: ${opt:stage, 'production'}
environment:
PROJECT: ${self:service}-${self:provider.stage}
ENV: ${self:provider.stage}
iamRoleStatements:
- Effect: "Allow"
Action:
- "s3:*"
Resource: 'arn:aws:s3:::fb-twitterbot'
- Effect: "Allow"
Action:
- "s3:*"
Resource: 'arn:aws:s3:::fb-twitterbot/*'
custom:
pythonRequirements:
dockerizePip: non-linux
plugins:
- serverless-python-requirements
functions:
run:
handler: bots/fetchfollow.fetchfollow
events:
- schedule:
rate: cron(15 */3 * * ? *)
post:
handler: bots/post.post_random_medium_article
events:
- schedule:
rate: cron(37 7,18 * * ? *)
unfollow:
handler: bots/unfollow.unfollow
events:
- schedule:
rate: cron(17,32,45 */2 * * ? *)
serverless.yml hosted with ❤by GitHub
serverless.yml
安装机器人相当简单，但是我会另写一篇文章向大家解释Serverless。如果要更新机器人的话，需要对脚本进行一些更改，然后运行serverless deploy。
结语
我会让机器人运行更长时间，以便大家阅读这篇文章时可以有一个实时的参考。不过，我最终还是会关掉它的。
推荐阅读专题
留言 点赞 关注
我们一起分享AI学习与发展的干货
如需转载，请后台留言，遵守转载规范