Just created a twitter bot with python and tweepy (a python library to interact with twitter, http://www.tweepy.org/) which is following a defined set of users. This users are gathered from defined user’s followers. After a defined lease of days this followers are became unfollowed if they did not followed back during a new execution of the script.
This script also handles the max API calls per execution. –> normally the max limit is set to 300/15min.
All data will be stored and updated in a local CSV.
Basic steps:
1. Create an twitter application and describe the purpose and details. (https://developer.twitter.com/en/apps/)
App Name: Choose a unique name
Description: Description of your application
Sign in with Twitter: Enabled
Callback URL: http://127.0.0.1
App usage: Purpose and description of your application
2. Get the Keys and Tokens
Get the API keys from tab: “Keys and tokens” and store them in a different file
# Variables that contain the user credentaials to access Tiwtter APIself. ACCESS_TOKEN = "1...i" ACCESS_TOKEN_SECRET = "d...6" CONSUMER_KEY = "M...L" CONSUMER_SECRET = "6...X"
3. Explore basic functions and API docs
Tweepy docs: http://docs.tweepy.org/en/v3.5.0/
Authentication
import tweepy from tweepy import OAuthHandler import twitter_creds auth = OAuthHandler(twitter_creds.CONSUMER_KEY, twitter_creds.CONSUMER_SECRET) auth.set_access_token(twitter_creds.ACCESS_TOKEN, twitter_creds.ACCESS_TOKEN_SECRET)
Create a Tweet
tweet = api.update_status(‘Hello World!’)
Delete a Tweet
tweet = api.update_status('something')
api.destroy_status(tweet.id_str)
Follow
api.create_friendship('@vmwarecloudaws')
Unfollow
api.destroy_friendship('@vmwarecloudaws')
Structure of CSV:
id,screen_name,scannedOn,scannedFrom,followedOn,followedBackOn,unfollowedOn,state,recordState 2466userId6,screen_name,2019-03-18 11:49:30,screen_name,2019-03-23 11:49:30,2019-03-21 11:49:30,,bidirectional,True
Python script:
import csv #interact with CSV
import sys #import sys for terminating script
import datetime
import time
import random #to pick a user to get followers from - defined in list "toGetFollowersFrom"
import tweepy
from tweepy import OAuthHandler
from tweepy.parsers import RawParser
import twitter_creds
#global list with user objects imported from CSV
twitters = list()
users = list()
apiInteractions = 1 #starts on 1 because of the authentication
#parameters
maxAllowedApiInteractions = 200 #script will end after this number is reached
ownScreen_Name = 'raphi_gassmann' #without (at)
daysForFollowersToFollowBack = 7
toGetFollowersFrom = ["user1", "user2"]
#csv filepath --> first row needs to be headers: id, screen_name, scannedOn, scannedFrom, followedOn, followedBackOn, unfollowedOn, state, recordState
csvFilepath = '/Users/rgassmann/Desktop/twitter-bot/twitterBot.csv'
#authentication - https://developer.twitter.com/en/apps/
auth = OAuthHandler(twitter_creds.CONSUMER_KEY, twitter_creds.CONSUMER_SECRET)
auth.set_access_token(twitter_creds.ACCESS_TOKEN, twitter_creds.ACCESS_TOKEN_SECRET)
api = tweepy.API(auth)
#api = tweepy.API(auth_handler = auth, parser = rawParser)
# ========================== classes ==========================
class User:
def __init__(self, id, screen_name, scannedOn, scannedFrom, followedOn, followedBackOn, unfollowedOn, state, recordState):
self.id = id
self.screen_name = screen_name
self.scannedOn = scannedOn
self.scannedFrom = scannedFrom
self.followedOn = followedOn
self.followedBackOn = followedBackOn
self.unfollowedOn = unfollowedOn
self.state = state
self.recordState = recordState
def follow(self):
'''
will create follow friendship with defined user
'''
global apiInteractions
try:
apiInteractionAllowed()
apiUser = api.create_friendship(self.id) #add as follower
apiInteractions += 1 #increase apiInteraction
print 'Followed user: %s, userId: %s' % (apiUser.screen_name, apiUser.id)
now = datetime.datetime.now()
global newUser
newUser = User(apiUser.id, apiUser.screen_name, self.scannedOn, self.scannedFrom, now.strftime('%Y-%m-%d %H:%M:%S'), self.followedBackOn, self.unfollowedOn, 'followed', True)
#update CSV
setRowToDeleted(self.id, csvFilepath)
newUserToCsv(newUser, csvFilepath)
twitters.append(newUser)
except tweepy.TweepError as e:
if (e.message[0]['code'] == 160): #already requested to follow
print 'Alredy requested to follow user: %s, updating CSV...' % (self.screen_name)
now = datetime.datetime.now()
newUser = User(self.id, self.screen_name, self.scannedOn, self.scannedFrom, now.strftime('%Y-%m-%d %H:%M:%S'), self.followedBackOn, self.unfollowedOn, 'followed', True)
#update CSV
setRowToDeleted(self.id, csvFilepath)
newUserToCsv(newUser, csvFilepath)
twitters.append(newUser)
if (e.message[0]['code'] == 89): #expired ACCESS_TOKEN
args = ('Invalid or expired token. Failed to unfollow user: %s, id: %s') % (self.screen_name, self.id)
sys.exit(args) #end script
if (e.meassage[0]['code'] == 161): #followed more than 1000 this day
args = ('The limit to follow more than 1000 account this day is reached. More information here: https://help.twitter.com/en/using-twitter/twitter-follow-limit')
sys.exit(args) #end script
if (e.message[0]['code'] == 34) or (e.message[0]['code'] == 108): #page does not exist, err108=cannot find specific user
#delete user, seems is not existing any more
setRowToDeleted(self.id, csvFilepath)
else:
print 'Following failed for user.screen_name: %s, user.id: %s' % (self.screen_name, self.id)
print e
pass
def unfollow(self):
'''
will destroy / unfollow defined user
'''
global apiInteractions
try:
apiInteractionAllowed()
apiUser = api.destroy_friendship(self.id) #remove as followed
apiInteractions += 1 #increase apiInteraction
print 'Unfollowed user: %s, userId: %s' % (apiUser.screen_name, apiUser.id)
now = datetime.datetime.now()
global newUser
newUser = User(apiUser.id, apiUser.screen_name, self.scannedOn, self.scannedFrom, self.followedOn, self.followedBackOn, now.strftime('%Y-%m-%d %H:%M:%S'), 'unfollowed', True)
#update CSV
setRowToDeleted(self.id, csvFilepath)
newUserToCsv(newUser, csvFilepath)
twitters.append(newUser)
except tweepy.TweepError as e:
if (e.message[0]['code'] == 89): #expired ACCESS_TOKEN
args = ('Invalid or expired token. Failed to unfollow user: %s, id: %s') % (self.screen_name, self.id)
sys.exit(args) #end script
if (e.message[0]['code'] == 34) or (e.message[0]['code'] == 108): #page does not exist, err108=cannot find specific user
#delete user, seems is not existing any more
setRowToDeleted(self.id, csvFilepath)
else:
print e
pass
# ========================== pre checks ==========================
def followedCleanUpNeeded(daysSinceFollowedOn):
"""
Is checking CSV if there are user in state=followed but not yet confirmed to followed back
AND if they are followed more than 7 days ago
AND if there are more than 10 users to perform the action on
returning list of users which did not followed back
"""
print ' ====== CheckCleanUpNeeded ====== '
global apiInteractions
apiInteractions += 1
apiInteractionAllowed()
followers = tweepy.Cursor(api.followers, screen_name=ownScreen_Name, count=200).items()
while True:
try:
follower = next(followers)
except tweepy.TweepError as e:
apiInteractionAllowed()
print 'request of followers of %s timed out' % (ownScreen_Name)
print e
apiInteractions += 1
time.sleep(60*15)
follower = next(followers)
except StopIteration:
break
for user in twitters:
if (str(user.id) == str(follower.id) and user.recordState == True and user.state != 'bidirectional'):
now = datetime.datetime.now()
newUser = User(follower.id, follower.screen_name, user.scannedOn, user.scannedFrom, user.followedOn, now.strftime('%Y-%m-%d %H:%M:%S'), user.unfollowedOn, 'bidirectional', True)
setRowToDeleted(str(follower.id), csvFilepath)
newUserToCsv(newUser, csvFilepath)
print 'new follower found and updated - screen_name: %s' % (follower.screen_name)
compareDate = datetime.datetime.now()+datetime.timedelta(days=-daysSinceFollowedOn, hours=0) #today minus daysSinceFollowedOn pararmeter
cleanUpUsers = list()
print 'compareDate: %s' % (compareDate)
for user in twitters:
if (user.followedOn != '' and user.state == 'followed' and user.followedBackOn == '' and user.recordState == True): #check first if followedOn has content
followedOnDatetime = datetime.datetime.strptime(user.followedOn, '%Y-%m-%d %H:%M:%S')
if (followedOnDatetime < compareDate):
cleanUpUsers.append(user)
print 'CleanUp needed for %s users' % (len(cleanUpUsers))
return cleanUpUsers
def newUsersToFollow():
"""
Is checking CSV if there are user in state=new to followed
AND if there are more than threshold parameter defined users to perform the action on
returning list of users objects
"""
print ' ====== CheckIfNewUsersToFollowExists ====== '
counter = 0
toFollow = list()
for user in twitters:
if (user.state == 'new' and user.recordState == True and user.followedOn == ''):
toFollow.append(user)
return toFollow
def apiInteractionAllowed():
global apiInteractions
global maxAllowedApiInteractions
if (apiInteractions>=maxAllowedApiInteractions):
args = ('Script terminated, because of: max API calls reached, API interactions: %s') % (apiInteractions)
sys.exit(args) #end script
# ========================== CSV interactions ==========================
def readCsv(filepath):
'''
reads defined CSV and creates user objects
'''
file = open(filepath)
global users
users = list(csv.reader(file, delimiter=','))
#checking headers row
if (users[0][0] == 'id' and users[0][1] == 'screen_name' and users[0][2] == 'scannedOn' and users[0][3] == 'scannedFrom' and users[0][4] == 'followedOn' and users[0][5] == 'followedBackOn' and users[0][6] == 'unfollowedOn' and users[0][7] == 'state' and users[0][8] == 'recordState'):
#slicing data rows
#users = users[1:][:] # not needed
print 'valid CSV provided as input'
else:
print 'invalid CSV provided as input - please check headers row'
pass
for idx, val in enumerate(users):
#check if record is set as active
if (str_to_bool(users[idx][8]) == True):
if (users[idx][0] != 'id' and users[idx][1] != 'screen_name'): # exclude headers
#create user object
user = User(users[idx][0], users[idx][1], users[idx][2], users[idx][3], users[idx][4], users[idx][5], users[idx][6], users[idx][7], str_to_bool(users[idx][8]))
#append user to global twitters list
twitters.append(user)
else:
#record is set as deleted, nothing to do
pass
def setRowToDeleted(userid, filepath):
'''
sets a user record csv to inactive / deleted
'''
#update object in twitters list
for user in twitters:
if (user.id == userid):
user.recordState = False
#update CSV
for idx, val in enumerate(users):
if (users[idx][0]==userid and str_to_bool(users[idx][8]) == True):
users[idx][8] = 'False'
writer = csv.writer(open(filepath, 'w'))
writer.writerows(users)
def newUserToCsv(user, filepath):
'''
writes a new user object to CSV
'''
newUser = []
newUser.append(user.id)
newUser.append(user.screen_name)
newUser.append(user.scannedOn)
newUser.append(user.scannedFrom)
newUser.append(user.followedOn)
newUser.append(user.followedBackOn)
newUser.append(user.unfollowedOn)
newUser.append(user.state)
newUser.append(user.recordState)
users.append(newUser)
writer = csv.writer(open(filepath, 'w'))
writer.writerows(users)
def str_to_bool(s):
'''
converts a string to a boolean and retuning it
'''
if s.upper() == 'TRUE':
return True
elif s.upper() == 'FALSE':
return False
else:
return s
# ========================== gathering new followers from ... ==========================
def gatheringUsersToFollow(username):
global apiInteractions
apiInteractions += 1
followers = tweepy.Cursor(api.followers, screen_name=username, count=200).items()
while True:
try:
follower = next(followers)
except tweepy.TweepError:
apiInteractionAllowed()
apiInteractions += 1
time.sleep(60*15)
follower = next(followers)
except StopIteration:
break
userAlreadyExist = bool(False)
for user in twitters:
if (str(user.id)==str(follower.id)):
userAlreadyExist = bool(True)
if not userAlreadyExist:
now = datetime.datetime.now()
newUser = User(follower.id, follower.screen_name, now.strftime('%Y-%m-%d %H:%M:%S'), username, '', '', '', 'new', True)
newUserToCsv(newUser, csvFilepath)
print 'new user added as state: new to follow - screen_name: %s' % (follower.screen_name)
# ========================== main ==========================
if __name__ == '__main__':
#open CSV
readCsv(csvFilepath)
# PRIO 1: CleanUp
toUnfollow = followedCleanUpNeeded(daysForFollowersToFollowBack)
if (len(toUnfollow)>0):
print 'CleanUp needed! --> running CleanUp'
for user in toUnfollow:
print 'user for cleanup: %s' % (user.screen_name)
user.unfollow() # --> just be sure !!!!sys.exit('just check')
# PRIO 2: follow
toFollow = newUsersToFollow()
if len(toFollow)>20:
print 'Following needed --> found %s users to follow' % (len(toFollow))
for user in toFollow:
user.follow()
# PRIO 3: get followers of
gatheringUsersToFollow(random.choice(toGetFollowersFrom))
args = ('Script terminated, because of: All actions performed (no limit reached), API interactions: %s') % (apiInteractions)
sys.exit(args) #end script
