My New Twitter Bot

Automatted Tweeting for Fun and Not Profit

Jan 12, 2013 | image credit

Something a Little Bit Different

Now that I have updated my blog and gotten the Obligatory Blogging Like a Hacker Post out of the way, I would like to introduce you to @bucketobytes. @bucketobytes is my new twitter bot. Before we get into the details about him, I’d like to talk about his inspiration. As I was sorting through my RSS feeds over the holidays I came across Random Shopper, a bot that Darius Kazemi wrote. Random Shopper is a bot that randomly buys Darius presents on Amazon. I thought “Well that is just a really cool idea, I wish I had thought of it”. I then went on a hunt for other bots and came across @ticbot and a few others. So I decided I wanted to try my hand at such a thing. So I set out on a quest to build a twitter spam bot, that (hopefully) wouldn’t actually do any spamming.

What Should a Twitter Bot Do?

I have grand plans for @bucketobytes, but to start off with I wanted to keep things simple to ensure that I would actually get him out there. I decided that the bare minimum was that he should follow people that follow him, reply to @mentions, and tweet randomly. As far as content, I came across this github repository that contains a vast number of fortunes. After pairing down the content to the tweet sized fortunes there are about 2000 available. I figured that would be a good start for @bucketobytes so I blatantly stole them and put them in my own project (thank you @brianclapper!). In the future I have been thinking about incorporating chatoms and I’m sure there are other countless resources that I can look into. I thought about other more useful content such as: replying to a mention that includes a movie title with the times that movie was playing. Upon reconsideration, however, I think that the guiding principle I will follow is that @bucketobytes should not really do anything all that useful.

As of this writing the current implementation of @bucketobytes is as follows. If you follow him, he will follow you back. If you @mention him, he will @reply to you with a random fortune. If you @mention him with the hash tag #cc he will retweet what you’ve sent him. And finally he will spontaneously tweet fortunes at random approximately 22 times per day. There are some other implementations I am considering in the future. The first is a mechanism that you can submit suggestions, perhaps with the hashtag #iwantbucketobytesto. Perhaps have him lie, for example subscribe to a Rotten Tomatoes RSS feed and @bucketobytes will say how much he loved such and such movie. I’m excited to implement new ideas as they occur to me. You can find an up to date list of @bucketobytes actions here: @bucketobytes Info Page.

I will now outline his construction, so if you aren’t into reading a lot of code feel free to duck out now.

The Guts of a Twitter Bot

You can find the source code for @bucketobytes on GitHub here: seepel/bucketobytes.

Disclaimer: I would ask that you not do anything nefarious with my code, but I imagine there is other more advanced software for real spam bots. If you arrived at my site looking for a spam bot implementation, you will probably be better served to look elsewhere.

To Python or Not to Python

I decided to use python to write my bot as I am very familiar with it from my Physics days, and it would be easy to get things up and running. For the twitter api I tried a few different GitHub projects and decided on Twython. Twython ended up being the most robust and gave me the most freedom while taking away as many headaches as possible. Other libraries were very good, but ended up being a bit restrictive. In order to get things up and running I did have to modify things quite a bit. At the time of writing Twython was still primarily based in the v1.0 twitter api, and there are some problems with the most recent version of the requests module, in particular dealing with oauth. So I updated all the twitter endpoints to v1.1 and modified the streaming portion of Twython to handle the new oauth user endpoint. You can find my changes here: seepel/Twython. I didn’t end up pushing my changes back to the main repository as my changes are a bit of a hack at the moment. I may come back and revisit this if someone else doesn’t beat me to it.

@bucketobytes: A history

My first implementation was rather fragile and relied entirely upon the REST API. What that meant was that everything that was done, had to be done in discrete chunks. The script would be run periodically, and the bot would have to catch up from its most recent state. Since it was just a prototype I was saving the most recent tweet id in a text file, and stopping processing whenever I hit that tweet. As one would guess this was problematic, if there was ever a problem with that text file @bucketobytes would get out of sync, and potentially re-run any number of actions. This would be annoying for anyone that interacted with him. From here I decided to move to the streaming API.

The streaming version was a bit better. It consisted of two scripts: the first was a script that would listen to the user stream and respond accordingly, the second was a separate script that when run would post a random fortune. So the respond script would act as a long running process to respond to actions, and the post script would be run from a crontab periodically. This process worked a lot better, and was stable enough to let run for a while. But I had grander plans, what if I wanted to incorporate the random posting into the replying and have finer control over how many tweets @bucketobytes was making. This method also led to these scripts being rather lengthy, it would be nice to refactor things.

So I setup a modular design. To do this I had to learn about threading in python. Luckily I found this article over at IBM that was dead simple to understand. The script is centered around two long running threads. Of course the first is listening to the user streaming endpoint to trigger replies and follows. Setting up the stream didn’t change much from the respond script of the previous iteration. What changed was how I funneled input into twitter. When a json object cames down from the user stream, it is placed into a queue to be processed. To process the pending queue, I setup a Thread class PostScheduler that would be responsible for the coordination of posting and following. Here is how that class is setup.

1 class PostScheduler(threading.Thread):
2   def __init__(self, api, simulate=False, controllers=None, default_time_to_sleep=60):
3     threading.Thread.__init__(self)
4     self.api = api
5     self.controllers = controllers
6     self.queue = Queue.Queue()
7     self.post_objects = []
8     self.default_time_to_sleep = default_time_to_sleep
9     self.setDaemon(True)

The class has an instance of a Twython object in the api variable, a list of controllers (we’ll get to that in a bit), a Queue, and a list of post_objects. The heartbeat of the bot is determined by the run method.

 1 def run(self):
 2     while True:
 3       queue_object = self.queue.get()
 5       if self.queue.empty():
 6         self.queue.put(self.default_time_to_sleep)
 8       if isinstance(queue_object, (int, long, float)):
 9         time_to_sleep = queue_object
10         if time_to_sleep > 0:
11           time.sleep(time_to_sleep)
12           self.evaluate_tweets()
13       else:
14         self.post_objects.append(queue_object)
16       self.queue.task_done()

What happens here is that the scheduler will remove the first item in the queue. If that item is a number, then that signals to the scheduler that it needs to wait before it handles more actions. If that object is not a number, then it is assumed that it is a dictionary object representing some twitter output (post_object) that needs to be handled. For example: a json object representing a mention from the streaming endpoint. That dictionary is then appended to post_objects for later processing by evaluate_tweets().

 1   def evaluate_tweets(self):
 2     self.count += 1
 3     seconds_from_midnight = ( - datetime.min).seconds
 4     post_objects_to_remove = []
 6     for post_object in self.post_objects:
 7       can_be_handled = False
 8       for controller in self.controllers:
 9         if controller.can_handle_object(post_object):
10           can_be_handled = True
11           break
12       if not can_be_handled:
13         post_objects_to_remove.append(post_object)
15     for post_object in post_objects_to_remove:
16       self.post_objects.remove(post_object)
18     for controller in self.controllers:
19       chosen_object = None
20       for post_object in self.post_objects:
21         if self.evaluate_tweet(controller, post_object, seconds_from_midnight):
22           chosen_object = post_object
23           break
24       if chosen_object != None:
25         self.post_objects.remove(chosen_object)
26         break
27       self.evaluate_tweet(controller, { }, seconds_from_midnight)

The evaluate_tweets() method is where the controllers come in. The controllers allow the whole bot to be configured. The first thing evaluate_tweets does, is figure how many seconds it has been since midnight. This way, one can configure the action to be dependent on the time of day. The next thing it does is for each post_object, determine if one of its controllers can handle the object, if not the object is removed. The problem is that there is a lot of stuff that comes in from the user stream that is not a mention or follow, this sequence removes the junk that will never be responded to. The next chunk of code runs through all the controllers and gives them an opportunity to act on the post_object via evaluate_tweet. If a controller handles the object, it is of course removed from the list of post_objects to be handled. The very last line evaluates a post_object being represented by an empty dictionary. This is how spontaneous tweets are generated.

  def evaluate_tweet(self, controller, post_object, seconds_from_midnight):
    probability = controller.probabilityToPost(post_object, seconds_from_midnight, self.default_time_to_sleep, self.simulate)
    if probability == 0:
      return False
    steps = 10000.0
    random_number = random.randrange(steps)/steps
    if random_number <= probability:
      self.posts += 1
      print controller
      print controller.composePost(self.api, post_object, self.simulate)
      controller.postUpdateStatus(self.api, post_object)
      return True
    return False

The evaluate_tweet() method asks a controller for the probability to respond to an object based on the object, the time of day, and the size of the time step that governs the schedulers heartbeat. The method then generates a random number and if the number is less than the probability, prompts the controller to respond to the object. This allows a controller to do something such as make tweets happen at certain times of day, while still being somewhat random. The second thing that is helpful about this method is that it allows a controller to back off on certain actions. For example, say there is a controller that handles replies, and someone decides to spam the account with 1000 mentions. If my bot were to respond to all those mentions at once, it would probably hit the twitter limit and potentially get blocked. This allows the bot to cut certain users off from replies.

Now that we see how the bot handles the flow of tweets, let’s talk controllers. Controllers are responsible for determining how and when the bot should tweet. For example there is a PostController that is responsible for spontaneously posting tweets, the ReplyController is responsible for dealing with mentions, the RetweetController is responsible for handling instances where someone retweets one of @bucketobytes tweets, and etc. Here is the PostController:

 1 class PostController(object):
 2   def __init__(self, post_composers = [], postControllers = None, current_user=None):
 3     self.post_composers = post_composers
 4     self.current_user = current_user
 6   def can_handle_object(self, post_object):
 7     return len(post_object) == 0
 9   def probabilityToPost(self, post_object, seconds_from_midnight, time_step, simulate=False):
10     if len(post_object) != 0:
11       return 0
12     if self.isCurrentUser(post_object):
13       return 0
14     # flat distribution 22 tweets per day
15     one_day = 60.*60.*24./float(time_step)
16     if simulate:
17       one_day /= 60
18     return 22./one_day
20   def isCurrentUser(self, post_object):
21     if self.current_user == None:
22       print 'No current user skipping'
23       return False
24     # don't respond if the tweet belongs to the current user -- would be infinite loop!
25     if post_object.has_key('user'):
26       if post_object['user'].has_key('id_str'):
27         return post_object['user']['id_str'] == self.current_user['id_str']
28     return False
30   def choosePostComposer(self):
31     post_composers = []
32     total_percent = 0
33     for post_composer in self.post_composers:
34       if post_composer.percent() == 100:
35         return post_composer
36       post_composers.append(post_composer)
37       total_percent += post_composer.percent()
38     probability = random.randrange(total_percent)
39     threshold = 0
40     for post_composer in post_composers:
41       if threshold <= post_composer.percent():
42         return post_composer
43       threshold += post_composer.percent()
44     return post_composer
46   def composePost(self, api, post_object, simulate):
47     return self.choosePostComposer().compose(api, post_object, simulate)

The post controller handles post objects that are empty dictionaries, this happens at the end of each scheduler cycle. Thus far, it has a constant probability to post such that the average should be about 22 tweets per day. In the future I will look into making it more likely to tweet at certain times of day.

Each controller has a list of PostComposers. This will later give me the ability to tweet different things. For example: spontaneously tweet a fortune 40% of the time and post a chatom 60% of the time. The controller object can also decide if it should respond to a post_object based on its content. For example here is the ReplyController that only handles objects which mention @bucketobytes.

 1 class ReplyController(post.PostController):
 2   def __init__(self, post_composers = [], postControllers = None, current_user=None):
 3     post.PostController(post_composers, postControllers, current_user)
 4     self.post_composers = post_composers
 5     self.current_user = current_user
 6     self.reply_ids = { }
 8   def can_handle_object(self, post_object):
 9     if self.isCurrentUser(post_object):
10       return False
11     if not post_object.has_key('entities'):
12       return False
13     if not post_object['entities'].has_key('user_mentions'):
14       return False
15     for user_mention in post_object['entities']['user_mentions']:
16       if user_mention['id_str'] == self.current_user['id_str']:
17         return True
18     return False
20   def probabilityToPost(self, post_object, seconds_from_midnight, time_step, simulate=False):
21     if self.isCurrentUser(post_object):
22       return 0
23     if not post_object.has_key('entities'):
24       return 0
25     if not post_object['entities'].has_key('user_mentions'):
26       return 0
27     for user_mention in post_object['entities']['user_mentions']:
28       if user_mention['id_str'] == self.current_user['id_str']:
29         return self.probabilityForId(post_object, seconds_from_midnight, time_step)
30     return 0
32   def probabilityForId(self, post_object, seconds_from_midnight, time_step):
33     if not post_object.has_key('user'):
34       return 0
35     if not post_object['user'].has_key('id_str'):
36       return 0
37     user_id = post_object['user']['id_str']
38     if not self.reply_ids.has_key(user_id):
39       self.reply_ids[user_id] = { 'probability' : 1, 'first_reply' :, 'last_attempt' : datetime.min }
41     current_datetime =
42     if (current_datetime - self.reply_ids[user_id]['first_reply']).seconds > 1:#60*60*24:
43       self.reply_ids[user_id] = { 'probability' : 1, 'first_reply' :, 'last_attempt' : datetime.min }
44       return 1
46     probability = self.reply_ids[user_id]['probability']
47     delta = ( - self.reply_ids[user_id]['last_attempt'])
48     if delta.microseconds < 500:
49       probability = 0
51     self.reply_ids[user_id]['last_attempt'] =
53     return probability
55   def postUpdateStatus(self, api, post_object):
56     user_id = post_object['user']['id_str']
57     probability = float(self.reply_ids[user_id]['probability'])
58     self.reply_ids[user_id]['probability'] = probability * 0.5

Finally, the flow trickles down to a PostComposer which creates the actual tweet. Here is the FortuneComposer

 1 class FortuneComposer(PostComposer):
 2   def __init__(self):
 3     self.fortunes = open('fortunes').read().split('\n%\n')
 4     for fortune in self.fortunes:
 5       if len(fortune) > 140:
 6         self.fortunes.remove(fortune)
 8   def compose(self, api, post_object, simulate):
 9     fortune = None
10     screen_name = None
11     if post_object.has_key('user'):
12       if post_object['user'].has_key('screen_name'):
13         screen_name = post_object['user']['screen_name']
14     if screen_name != None:
15       fortune = self.chooseFortune(140, screen_name)
16     else:
17       fortune = self.chooseFortune()
18     if fortune == None:
19       return None
20     if simulate:
21       return fortune
22     if post_object.has_key('id_str') and screen_name != None:
23       return api.updateStatus(status=fortune, in_reply_to_status_id=post_object['id_str'])
24     else:
25       return api.updateStatus(status=fortune)
27   def chooseFortune(self, max_len=140, screen_name=None):
28     fortune = ''
29     if screen_name != None:
30       fortune += '@' + screen_name + ' '
31       max_len -= len(fortune)
32     tmp_fortune = random.choice(self.fortunes)
33     count = 0
34     while len(tmp_fortune) > max_len:
35       if count > 1000:
36         return None
37       tmp_fortune = random.choice(self.fortunes)
38       count += 1
39     fortune += tmp_fortune
40     return fortune

This should be pretty self explanatory. It chooses a random fortune, prepending a screen name at the beginning if it is constructing a reply. It then ensures that the tweet will fit in the allocated 140 characters, and finally uses the provided Twython api object to send it to twitter.

To sum everything up, I feel that I have a nice implementation of a twitter bot that I can expand on down the line. It should be relatively easy to add new actions as I think of them. So why not give it a try and send @bucketobytes an @mention?