|
|
| (9 intermediate revisions by the same user not shown) |
| Line 1: |
Line 1: |
| ( ) ( )<br/>
| | https://twitter.com/pfftbot |
| (>•.•<)<br/>
| |
| <nowiki> (") (")</nowiki>
| |
|
| |
|
| | removes phase information from images |
|
| |
|
| == @_pil_bot_ ==
| | [[Image:CSEhtXUXIAAuFvM.png]] |
| | [[Image:CSEgxHFWUAA32HG.png]] |
| | [[Image:CSEhNiDWwAAJtnb.png]] |
|
| |
|
| [[File:pil__bot.png|500px|thumb|left]]
| |
|
| |
|
| === pilbot.py ===
| | <nowiki>#</nowiki>pfft |
| | |
| <source lang="python"> | |
| # -*- coding: utf-8 -*- # | |
| | |
| from twitterbot import TwitterBot
| |
| import keys
| |
| | |
| from PIL import Image
| |
| from sys import argv
| |
| from random import random
| |
| | |
| from pileval import PILEval
| |
| | |
| class PilBot(TwitterBot):
| |
|
| |
| def __init__(self):
| |
| TwitterBot.__init__(self)
| |
| self.pillermann = PILEval()
| |
|
| |
| def bot_init(self):
| |
| """ Initialize and configure the bot """
| |
| | |
| ############################
| |
| # REQUIRED: LOGIN DETAILS! #
| |
| ############################
| |
| self.config['api_key'] = keys.consumer_key
| |
| self.config['api_secret'] = keys.consumer_secret
| |
| self.config['access_key'] = keys.access_token
| |
| self.config['access_secret'] = keys.access_token_secret
| |
| | |
| | |
| ######################################
| |
| # SEMI-OPTIONAL: OTHER CONFIG STUFF! #
| |
| ######################################
| |
| | |
| # how often to tweet, in seconds
| |
| self.config['tweet_interval'] = 1 * 5 # default: 1 minutes
| |
| | |
| # use this to define a (min, max) random range of how often to tweet
| |
| # e.g., self.config['tweet_interval_range'] = (5*60, 10*60) # tweets every 5-10 minutes
| |
| self.config['tweet_interval_range'] = None
| |
| | |
| # only reply to tweets that specifically mention the bot
| |
| self.config['reply_direct_mention_only'] = True
| |
| | |
| # only include bot followers (and original tweeter) in @-replies
| |
| self.config['reply_followers_only'] = False
| |
| | |
| # fav any tweets that mention this bot?
| |
| self.config['autofav_mentions'] = False
| |
| | |
| # fav any tweets containing these keywords?
| |
| self.config['autofav_keywords'] = []
| |
| | |
| # follow back all followers?
| |
| self.config['autofollow'] = False
| |
| | |
| | |
| def on_scheduled_tweet(self):
| |
| """ Make a public tweet to the bot's own timeline. """
| |
| # We might take senteces from somewhere and tweet them on a regular basis ...
| |
| pass # don't do anything here ...
| |
| | |
| | |
| def on_mention(self, tweet, prefix):
| |
| """ Actions to take when a mention is received. """
| |
| text = tweet.text.split()[1]
| |
| self.action(text)
| |
|
| |
|
| |
| def on_timeline(self, tweet, prefix):
| |
| """ Actions to take on a timeline tweet. """
| |
| pass
| |
|
| |
| def action(self, code):
| |
| result = self.pillermann.eval(code)
| |
|
| |
| if result: # post the image
| |
| self.post_tweet(code, media="img.png", file=self.pillermann.get_file())
| |
| else: # post the error
| |
| self.post_tweet(self.pillermann.last_error)
| |
|
| |
| if __name__ == '__main__':
| |
| | |
| bot = PilBot()
| |
|
| |
| if len(argv) == 1:
| |
| bot.run()
| |
| else:
| |
| tweet_text = "text((300, 400), 'botsNplots', fill='#ff0')"
| |
| bot.action(tweet_text)
| |
| | |
| | |
| </source>
| |
| | |
| | |
| === pileval.py ===
| |
| | |
| <source lang="python">
| |
| from PIL import Image, ImageDraw
| |
| | |
| from os import path
| |
| from io import BytesIO
| |
| | |
| class PILEval():
| |
|
| |
| def __init__(self, filename='test.png'):
| |
| self.filename = filename
| |
| try:
| |
| self.image = Image.open(filename)
| |
| except FileNotFoundError as e:
| |
| self.image = Image.new("RGBA", (640, 480), 'white')
| |
| self.last_error = ''
| |
|
| |
| def __repr__(self):
| |
| return "<PILEval: %s>" % self.filename
| |
|
| |
| def eval(self, code):
| |
| draw = ImageDraw.Draw(self.image, "RGBA")
| |
| code = "draw.%s" % code
| |
| | |
| try:
| |
| code_obj = compile(code, '<string>', 'exec')
| |
| exec(code_obj)
| |
| except Exception as e:
| |
| self.last_error = str(e)
| |
| return False
| |
|
| |
| self.image.save(self.filename)
| |
|
| |
| return True
| |
|
| |
| def get_file(self):
| |
| img = Image.open(self.filename)
| |
| file = BytesIO()
| |
| img.save(file, format='PNG')
| |
| return file
| |
|
| |
| if __name__ == '__main__':
| |
| # some q'n'dirty tests
| |
| print("main")
| |
| pillermann = PILEval()
| |
| | |
| # will work
| |
| # result = pillermann.eval("rectangle([20, 10, 120, 200], fill='#f00')")
| |
| # if result:
| |
| # print('good')
| |
|
| |
| # will also work
| |
| # result = pillermann.eval("ellipse([200, 50, 500, 500], fill='#def')")
| |
| # result = pillermann.eval("point((123, 234), fill='#345')")
| |
| # result = pillermann.eval("line([23, 34, 345, 456], fill='#eca', width=10)")
| |
| # result = pillermann.eval("arc([500, 10, 600, 440], 10, 95, fill='#3f9')")
| |
| # result = pillermann.eval("text((22, 202), '_pil_bot_!', fill='#a4f')")
| |
|
| |
| result = pillermann.eval("text((100, 100), text=code, fill='#000')")
| |
| if result:
| |
| print("ok")
| |
| else:
| |
| print(pillermann.last_error)
| |
| # will yield a compiler error
| |
| # result = pillermann.eval("rectangle([20, 10, 120, 200], fill='#f00'")
| |
| # if not result:
| |
| # print("error result: %s" % pillermann.last_error)
| |
| | |
| # f = pillermann.get_file()
| |
| # print(f)
| |
|
| |
| </source>
| |
| | |
| === twitterbot/bot.py ===
| |
| | |
| adapts thricedotted's (http://lczzz.me) twitterbot for python 3
| |
| | |
| <source lang="python">
| |
| | |
| import os
| |
| import codecs
| |
| import json
| |
| import logging
| |
| import tweepy
| |
| import time
| |
| import re
| |
| import random
| |
| import pickle
| |
| | |
| def ignore(method):
| |
| """
| |
| Use the @ignore decorator on TwitterBot methods you wish to leave
| |
| unimplemented, such as on_timeline and on_mention.
| |
| """
| |
| method.not_implemented = True
| |
| return method
| |
| | |
| | |
| class TwitterBot:
| |
| | |
| def __init__(self):
| |
| self.config = {}
| |
| | |
| self.custom_handlers = []
| |
| | |
| self.config['reply_direct_mention_only'] = False
| |
| self.config['reply_followers_only'] = True
| |
| | |
| self.config['autofav_mentions'] = False
| |
| self.config['autofav_keywords'] = []
| |
| | |
| self.config['autofollow'] = False
| |
| | |
| self.config['tweet_interval'] = 30 * 60
| |
| self.config['tweet_interval_range'] = None
| |
| | |
| self.config['reply_interval'] = 10
| |
| self.config['reply_interval_range'] = None
| |
| | |
| self.config['ignore_timeline_mentions'] = True
| |
| | |
| self.config['logging_level'] = logging.DEBUG
| |
| self.config['storage'] = FileStorage()
| |
| | |
| self.state = {}
| |
| | |
| # call the custom initialization
| |
| self.bot_init()
| |
| | |
| auth = tweepy.OAuthHandler(self.config['api_key'], self.config['api_secret'])
| |
| auth.set_access_token(self.config['access_key'], self.config['access_secret'])
| |
| self.api = tweepy.API(auth)
| |
| | |
| self.id = self.api.me().id
| |
| self.screen_name = self.api.me().screen_name
| |
| | |
| logging.basicConfig(format='%(asctime)s | %(levelname)s: %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p',
| |
| filename=self.screen_name + '.log',
| |
| level=self.config['logging_level'])
| |
| | |
| logging.info('Initializing bot...')
| |
| | |
| try:
| |
| with self.config['storage'].read(self.screen_name) as f:
| |
| self.state = pickle.load(f)
| |
|
| |
| except IOError:
| |
| self.state['last_timeline_id'] = 1
| |
| self.state['last_mention_id'] = 1
| |
| | |
| self.state['last_timeline_time'] = 0
| |
| self.state['last_mention_time'] = 0
| |
| | |
| self.state['last_tweet_id'] = 1
| |
| self.state['last_tweet_time'] = 1
| |
| | |
| self.state['last_reply_id'] = 0
| |
| self.state['last_reply_time'] = 0
| |
| | |
| self.state['recent_timeline'] = []
| |
| self.state['mention_queue'] = []
| |
| | |
| self.state['friends'] = self.api.friends_ids(self.id)
| |
| self.state['followers'] = self.api.followers_ids(self.id)
| |
| self.state['new_followers'] = []
| |
| self.state['last_follow_check'] = 0
| |
| | |
| logging.info('Bot initialized!')
| |
| | |
| | |
| def bot_init(self):
| |
| """
| |
| Initialize custom state values for your bot.
| |
| """
| |
| raise NotImplementedError("You MUST have bot_init() implemented in your bot! What have you DONE!")
| |
| | |
| | |
| def log(self, message, level=logging.INFO):
| |
| if level == logging.ERROR:
| |
| logging.error(message)
| |
| else:
| |
| logging.info(message)
| |
| | |
| | |
| def _log_tweepy_error(self, message, e):
| |
| try:
| |
| e_message = e.message[0]['message']
| |
| code = e.message[0]['code']
| |
| self.log("{}: {} ({})".format(message, e_message, code), level=logging.ERROR)
| |
| except:
| |
| self.log(message, e)
| |
| | |
| | |
| def _tweet_url(self, tweet):
| |
| return "http://twitter.com/" + tweet.author.screen_name + "/status/" + str(tweet.id)
| |
| | |
| | |
| def _save_state(self):
| |
| with self.config['storage'].write(self.screen_name) as f:
| |
| pickle.dump(self.state, f)
| |
| self.log('Bot state saved')
| |
| | |
| | |
| def on_scheduled_tweet(self):
| |
| """
| |
| Post a general tweet to own timeline.
| |
| """
| |
| #self.post_tweet(text)
| |
| raise NotImplementedError("You need to implement this to tweet to timeline (or pass if you don't want to)!")
| |
| | |
| | |
| def on_mention(self, tweet, prefix):
| |
| """
| |
| Perform some action upon receiving a mention.
| |
| """
| |
| #self.post_tweet(text)
| |
| raise NotImplementedError("You need to implement this to reply to/fav mentions (or pass if you don't want to)!")
| |
| | |
| | |
| | |
| def on_timeline(self, tweet, prefix):
| |
| """
| |
| Perform some action on a tweet on the timeline.
| |
| """
| |
| #self.post_tweet(text)
| |
| raise NotImplementedError("You need to implement this to reply to/fav timeline tweets (or pass if you don't want to)!")
| |
| | |
| | |
| def on_follow(self, f_id):
| |
| """
| |
| Perform some action when followed.
| |
| """
| |
| if self.config['autofollow']:
| |
| try:
| |
| self.api.create_friendship(f_id, follow=True)
| |
| self.state['friends'].append(f_id)
| |
| logging.info('Followed user id {}'.format(f_id))
| |
| except tweepy.TweepError as e:
| |
| self._log_tweepy_error('Unable to follow user', e)
| |
| | |
| time.sleep(3)
| |
| | |
| self.state['followers'].append(f_id)
| |
| | |
| | |
| | |
| def post_tweet(self, text, reply_to=None, media=None, file=None):
| |
| """
| |
| Post a tweet containing the text.
| |
| If you provide a filename the file will be added to the tweet.
| |
| You can also provide the filedata, for example when using dynamically created images.
| |
| """
| |
|
| |
| kwargs = { "status": text }
| |
| args = []
| |
| | |
| if media is not None:
| |
| cmd = self.api.update_with_media
| |
| args.insert(0, media)
| |
| | |
| if file is not None:
| |
| kwargs["file"] = file
| |
| else:
| |
| cmd = self.api.update_status
| |
| | |
| try:
| |
| self.log('Tweeting "{}"'.format(text))
| |
| if reply_to:
| |
| self.log("-- Responding to status {}".format(self._tweet_url(reply_to)))
| |
| kwargs['in_reply_to_status_id'] = reply_to.id
| |
| else:
| |
| self.log("-- Posting to own timeline")
| |
| | |
| tweet = cmd(*args, **kwargs)
| |
| self.log('Status posted at {}'.format(self._tweet_url(tweet)))
| |
| return True
| |
| | |
| except tweepy.TweepError as e:
| |
| self._log_tweepy_error('Can\'t post status', e)
| |
| return False
| |
| | |
| | |
| def favorite_tweet(self, tweet):
| |
| try:
| |
| logging.info('Faving ' + self._tweet_url(tweet))
| |
| self.api.create_favorite(tweet.id)
| |
| | |
| except tweepy.TweepError as e:
| |
| self._log_tweepy_error('Can\'t fav status', e)
| |
| | |
| | |
| def _ignore_method(self, method):
| |
| return hasattr(method, 'not_implemented') and method.not_implemented
| |
| | |
| | |
| def _handle_timeline(self):
| |
| """
| |
| Reads the latest tweets in the bots timeline and perform some action.
| |
| self.recent_timeline
| |
| """
| |
| for tweet in self.state['recent_timeline']:
| |
| prefix = self.get_mention_prefix(tweet)
| |
| self.on_timeline(tweet, prefix)
| |
| | |
| words = tweet.text.lower().split()
| |
| if any(w in words for w in self.config['autofav_keywords']):
| |
| self.favorite_tweet(tweet)
| |
| | |
| #time.sleep(self.config['reply_interval'])
| |
| | |
| | |
| def _handle_mentions(self):
| |
| """
| |
| Performs some action on the mentions in self.mention_queue
| |
| """
| |
| # TODO: only handle a certain number of mentions at a time?
| |
| for mention in iter(self.state['mention_queue']):
| |
| prefix = self.get_mention_prefix(mention)
| |
| self.on_mention(mention, prefix)
| |
| self.state['mention_queue'].remove(mention)
| |
| | |
| if self.config['autofav_mentions']:
| |
| self.favorite_tweet(mention)
| |
| | |
| #time.sleep(self.config['reply_interval'])
| |
| | |
| | |
| def get_mention_prefix(self, tweet):
| |
| """
| |
| Returns a string of users to @-mention when responding to a tweet.
| |
| """
| |
| mention_back = ['@' + tweet.author.screen_name]
| |
| mention_back += [s for s in re.split('[^@\w]', tweet.text) if len(s) > 2 and s[0] == '@' and s[1:] != self.screen_name]
| |
| | |
| if self.config['reply_followers_only']:
| |
| mention_back = [s for s in mention_back if s[1:] in self.state['followers'] or s == '@' + tweet.author.screen_name]
| |
| | |
| return ' '.join(mention_back)
| |
| | |
| | |
| def _check_mentions(self):
| |
| """
| |
| Checks mentions and loads most recent tweets into the mention queue
| |
| """
| |
| if self._ignore_method(self.on_mention):
| |
| logging.debug('Ignoring mentions')
| |
| return
| |
| | |
| try:
| |
| current_mentions = self.api.mentions_timeline(since_id=self.state['last_mention_id'], count=100)
| |
| | |
| # direct mentions only?
| |
| if self.config['reply_direct_mention_only']:
| |
| current_mentions = [t for t in current_mentions if re.split('[^@\w]', t.text)[0] == '@' + self.screen_name]
| |
| | |
| if len(current_mentions) != 0:
| |
| self.state['last_mention_id'] = current_mentions[0].id
| |
|
| |
| self.state['last_mention_time'] = time.time()
| |
| | |
| self.state['mention_queue'] += reversed(current_mentions)
| |
| | |
| logging.info('Mentions updated ({} retrieved, {} total in queue)'.format(len(current_mentions), len(self.state['mention_queue'])))
| |
| | |
| except tweepy.TweepError as e:
| |
| self._log_tweepy_error('Can\'t retrieve mentions', e)
| |
| | |
| except IncompleteRead as e:
| |
| self.log('Incomplete read error -- skipping mentions update')
| |
| | |
| | |
| def _check_timeline(self):
| |
| """
| |
| Checks timeline and loads most recent tweets into recent timeline
| |
| """
| |
| if self._ignore_method(self.on_timeline):
| |
| logging.debug('Ignoring timeline')
| |
| return
| |
| | |
| try:
| |
| current_timeline = self.api.home_timeline(count=200, since_id=self.state['last_timeline_id'])
| |
| | |
| # remove my tweets
| |
| current_timeline = [t for t in current_timeline if t.author.screen_name.lower() != self.screen_name.lower()]
| |
| | |
| # remove all tweets mentioning me
| |
| current_timeline = [t for t in current_timeline if not re.search('@'+self.screen_name, t.text, flags=re.IGNORECASE)]
| |
| | |
| if self.config['ignore_timeline_mentions']:
| |
| # remove all tweets with mentions (heuristically)
| |
| current_timeline = [t for t in current_timeline if '@' not in t.text]
| |
| | |
| if len(current_timeline) != 0:
| |
| self.state['last_timeline_id'] = current_timeline[0].id
| |
|
| |
| self.state['last_timeline_time'] = time.time()
| |
| | |
| self.state['recent_timeline'] = list(reversed(current_timeline))
| |
| | |
| logging.info('Timeline updated ({} retrieved)'.format(len(current_timeline)))
| |
| | |
| except tweepy.TweepError as e:
| |
| self._log_tweepy_error('Can\'t retrieve timeline', e)
| |
| | |
| except IncompleteRead as e:
| |
| self.log('Incomplete read error -- skipping timeline update')
| |
| | |
| | |
| def _check_followers(self):
| |
| """
| |
| Checks followers.
| |
| """
| |
| logging.info("Checking for new followers...")
| |
| | |
| try:
| |
| self.state['new_followers'] = [f_id for f_id in self.api.followers_ids(self.id) if f_id not in self.state['followers']]
| |
| | |
| self.config['last_follow_check'] = time.time()
| |
| | |
| except tweepy.TweepError as e:
| |
| self._log_tweepy_error('Can\'t update followers', e)
| |
| | |
| except IncompleteRead as e:
| |
| self.log('Incomplete read error -- skipping followers update')
| |
| | |
|
| |
| def _handle_followers(self):
| |
| """
| |
| Handles new followers.
| |
| """
| |
| for f_id in self.state['new_followers']:
| |
| self.on_follow(f_id)
| |
| | |
| def register_custom_handler(self, action, interval):
| |
| """
| |
| Register a custom action to run at some interval.
| |
| """
| |
| handler = {}
| |
| | |
| handler['action'] = action
| |
| handler['interval'] = interval
| |
| handler['last_run'] = 0
| |
| | |
| self.custom_handlers.append(handler)
| |
| | |
| def run(self):
| |
| """
| |
| Runs the bot! This probably shouldn't be in a "while True" lol.
| |
| """
| |
| while True:
| |
|
| |
| # check followers every 15 minutes
| |
| #if self.autofollow and (time.time() - self.last_follow_check) > (15 * 60):
| |
| if self.state['last_follow_check'] > (15 * 60):
| |
| self._check_followers()
| |
| self._handle_followers()
| |
| | |
| # check mentions every minute-ish
| |
| #if self.reply_to_mentions and (time.time() - self.last_mention_time) > 60:
| |
| if (time.time() - self.state['last_mention_time']) > 60:
| |
| self._check_mentions()
| |
| self._handle_mentions()
| |
| | |
| # tweet to timeline
| |
| #if self.reply_to_timeline and (time.time() - self.last_mention_time) > 60:
| |
| if (time.time() - self.state['last_timeline_time']) > 60:
| |
| self._check_timeline()
| |
| self._handle_timeline()
| |
| | |
| # tweet to timeline on the correct interval
| |
| if (time.time() - self.state['last_tweet_time']) > self.config['tweet_interval']:
| |
| self.on_scheduled_tweet()
| |
| | |
| # TODO: maybe this should only run if the above is successful...
| |
| if self.config['tweet_interval_range'] is not None:
| |
| self.config['tweet_interval'] = random.randint(*self.config['tweet_interval_range'])
| |
| | |
| self.log("Next tweet in {} seconds".format(self.config['tweet_interval']))
| |
| self.state['last_tweet_time'] = time.time()
| |
| | |
| # run custom action
| |
| for handler in self.custom_handlers:
| |
| if (time.time() - handler['last_run']) > handler['interval']:
| |
| handler['action']()
| |
| handler['last_run'] = time.time()
| |
| | |
| # save current state
| |
| self._save_state()
| |
| | |
| logging.info("Sleeping for a bit...")
| |
| time.sleep(30)
| |
| | |
| | |
| class FileStorage(object):
| |
| """
| |
| Default storage adapter.
| |
| | |
| Adapters must implement two methods: read(name) and write(name).
| |
| """
| |
| | |
| | |
| def read(self, name):
| |
| """
| |
| Return an IO-like object that will produce binary data when read from.
| |
| If nothing is stored under the given name, raise IOError.
| |
| """
| |
| filename = self._get_filename(name)
| |
| if os.path.exists(filename):
| |
| logging.debug("Reading from {}".format(filename))
| |
| else:
| |
| logging.debug("{} doesn't exist".format(filename))
| |
| return open(filename, 'rb')
| |
| | |
| | |
| def write(self, name):
| |
| """
| |
| Return an IO-like object that will store binary data written to it.
| |
| """
| |
| filename = self._get_filename(name)
| |
| if os.path.exists(filename):
| |
| logging.debug("Overwriting {}".format(filename))
| |
| else:
| |
| logging.debug("Creating {}".format(filename))
| |
| return open(filename, 'wb')
| |
| | |
| | |
| def _get_filename(self, name):
| |
| return '{}_state.pkl'.format(name)
| |
| | |
| </source> | |
