Last active
December 17, 2015 16:19
-
-
Save wonderb0lt/5638123 to your computer and use it in GitHub Desktop.
The API starts working out!
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import json | |
| import tweetpony | |
| import glob | |
| import raxa | |
| import inspect | |
| import os.path | |
| # Now it gets ugly: | |
| import modules # Parent | |
| import imp # Please don't beat me up. | |
| class DispatchingHandler(tweetpony.api.StreamProcessor): | |
| def __init__(self, api): | |
| tweetpony.api.StreamProcessor.__init__(self, api) | |
| def on_status(self, status): | |
| print 'Got a status, iterating over %d functions' % len(raxa.funcs) | |
| for f in raxa.funcs: | |
| try: | |
| if len(inspect.getargspec(f).args) == 2: | |
| f(status, self.api) | |
| else: | |
| f(status) | |
| except: | |
| pass # Swallow user-side exception | |
| def daemon(dry=False): | |
| consumer_key, consumer_secret, access_key, acces_secret = load_keys() | |
| print 'Loading modules' | |
| for f in glob.glob('modules/example.py'): | |
| name, ext = os.path.splitext(os.path.basename(f)) | |
| imp.load_source('modules.' + name, f) | |
| if dry: | |
| print '%d callbacks registered' % len(raxa.funcs) | |
| return | |
| print 'Authenticating....' | |
| api = tweetpony.API( | |
| consumer_key=consumer_key, consumer_secret=consumer_secret, | |
| access_token=access_key, access_token_secret=acces_secret) | |
| raxa.botname = api.user.screen_name | |
| try: | |
| print 'Starting daemon!' | |
| api.user_stream(processor=DispatchingHandler(api)) | |
| except KeyboardInterrupt: | |
| print '\nShutting down...' | |
| def load_keys(): | |
| with open('conf.json') as f: | |
| keys = json.load(f)['keys'] | |
| return (keys['consumer_key'], keys['consumer_secret'], keys['access_token_key'], keys['access_token_secret']) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import random | |
| from raxa import on_tweet, on_mention | |
| @on_mention(sole_mentionee=True) | |
| def greet(tweet): | |
| n = random.randint(0, 4242) | |
| tweet.reply('Hello. Here, have a random number: %d' % n) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import re | |
| ''' | |
| This module contains the API you need to use for your module files. | |
| ''' | |
| funcs = [] # A internal memory of all on_* instances | |
| botname = None | |
| USER_STREAM = '__userstream__' # Constant for which stream to follow, currently unused. | |
| def isset(var): | |
| return var != None | |
| def get_mentionees(status): | |
| return [user.screen_name for user in status.entities.user_mentions] | |
| class on_tweet(object): | |
| ''' | |
| This is the primary decorator for tweets, use is to annotate your functions. | |
| The tweet is only sent to the annotated method if all of the given criteria | |
| are met. | |
| .. warning:: | |
| Not all may have been implemented. You're on your own! | |
| It accepts a variety of criteria, including the following: | |
| * **author** (str) - Matches if the screen_name of the author | |
| * **is_mention** (bool) - Matches if the tweet contains any mentions to users | |
| * **mentions** (list or str) - Matches if the given users or the given user are mentioned | |
| * **contains** (str) - Matches if the given str is contained in the tweet | |
| * **matches_regex** (str) - Matches if the tweet is matched by the given regex | |
| ''' | |
| def __init__(self, author=None, is_mention=None, mentions=None, contains=None, matches_regex=None): | |
| self.author = author | |
| self.is_mention = is_mention | |
| self.mentions = mentions | |
| self.contains = contains | |
| self.matches_regex = matches_regex | |
| self.f = None | |
| def __call__(self, f): | |
| self.f = f | |
| def wrapped(*args): | |
| if self.passes(args[0]): | |
| f(*args) | |
| funcs.append(wrapped) | |
| return wrapped | |
| def passes(self, status): | |
| # TODO: More pythonic checks. List of functions with all([f() for f in | |
| # checks])? | |
| if isset(self.author) and self.author != status.user.screen_name: | |
| return False | |
| if isset(self.is_mention) and len(status.entities.user_mentions) == 1: | |
| return False | |
| if isset(self.contains) and self.contains in status.text: | |
| return False | |
| if isset(self.matches_regex) and re.search(self.matches_regex, status.text): | |
| return False | |
| # mentions is kind of a special case | |
| mentionees = get_mentionees(status) | |
| if type(self.mentions) == str and self.mentions in mentionees: | |
| return False | |
| elif type(self.mentions) == list and all([x in mentionees for x in self.mentions]): | |
| return False | |
| return True | |
| class on_mention(on_tweet): | |
| ''' | |
| A shortcut for @on_tweet(mentions=*botname*). | |
| Additionally, it has a flag for checking whether the bot is the sole mentionee of the mention. | |
| ''' | |
| def __init__(self, sole_mentionee=False, *args, **kwargs): | |
| on_tweet.__init__(self, mentions=botname) | |
| self.sole_mentionee = sole_mentionee | |
| def __call__(self, f): | |
| return on_tweet.__call__(self, f) | |
| def passes(self, tweet): | |
| if self.sole_mentionee and len(get_mentionees(status)) != 1: | |
| return False | |
| not_self_mention = tweet.user.screen_name != botname # Don't react to your own tweets | |
| return on_tweet.passes(self, tweet) and not_self_mention |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment