图片由Jonny Lindner拍摄，源自Pixabay

阅读此文，看看如何用Python创建Twitter机器人来进行一项有争议的实验的。本文包含所有的代码和详细说明。

摘 要

本文会详细说明我是如何创建一个Twitter聊天机器人的，它能利用一些人类的基本情感来创造粉丝。这种事在我们身边每天都会发生。我认为重要的是要让大家知道这很容易。我会分享一些有趣的经历，并向大家展示在Twitter中插入机器人的实例。

本文后半部分对机器人的代码和功能进行了详细描述。Github上有当前的产品代码。我在AWS上运行机器人，并通过Serverless架构对其进行部署，这连AWS每月免费层级的5%都用不到。所以值得一试。

动 机

图片由Patrick Fore拍摄，源自Unsplash

我想写一些文章，帮助人们学会使用Python，这样也能帮助拓宽他们的技能。部分原因是一些团队成员想学习Python。开始的时候，我告诉自己，先做两个月左右，看看结果如何。如果万不得已，我会找一些结构清晰、脉络清楚的文章来教大家使用Python。相信每个人都可以从中学到一些Python的知识。

推 特

Twitter

图片由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,, 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。

结 语

我会让机器人运行更长时间，以便大家阅读这篇文章时可以有一个实时的参考。不过，我最终还是会关掉它的。

————————————————

版权声明：本文为CSDN博主「读芯术」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。

原文链接：https://blog.csdn.net/duxinshuxiaobian/article/details/102987845