Skip to content

Instantly share code, notes, and snippets.

@wonderb0lt
Last active December 17, 2015 16:19
Show Gist options
  • Save wonderb0lt/5638123 to your computer and use it in GitHub Desktop.
Save wonderb0lt/5638123 to your computer and use it in GitHub Desktop.
The API starts working out!
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'])
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)
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