Skip to content

Instantly share code, notes, and snippets.

@eliliam
Last active July 27, 2017 22:05
Show Gist options
  • Save eliliam/a021da423c540868a51eaaf150807c34 to your computer and use it in GitHub Desktop.
Save eliliam/a021da423c540868a51eaaf150807c34 to your computer and use it in GitHub Desktop.
A cool script that streams to all the live tweets on twitter based on a length and filter, tons of fun to watch!
from tweepy.streaming import StreamListener
from tweepy import OAuthHandler
from tweepy import Stream
import json, sys, argparse, signal
#
# Written by Eli Smith, known as @Plunkinguitar on GitHub
#
# USER CONFIGS
consumer_key=""
consumer_secret=""
access_token_key=""
access_token_secret=""
# END USER CONFIGS
if (consumer_key=="" or consumer_secret=="" or access_token_key=="" or access_token_secret==""):
print "Please configure your access to twitter at the top of this file"
sys.exit(1)
# Command line argument parser
parser = argparse.ArgumentParser(description="Fun little script to pull live comments from twitter based on search terms and length")
parser.add_argument('-l', '--length', type=int, default=140, required=True, help="Max length of tweets(<40 is a lot of fun)")
parser.add_argument('-t', '--term', type=str, required=True, help="Terms to search for in the live tweets")
parser.add_argument('--lang', type=str, default="en", help="Language code to use, default en")
args = vars(parser.parse_args())
def nice_close(signal, frame):
print '\nExiting...'
sys.exit(0)
signal.signal(signal.SIGINT, nice_close)
def is_ascii(s):
return all(ord(c) < 128 for c in s)
class StdOutListener(StreamListener):
def __init__(self):
super(StdOutListener, self).__init__()
self.allTweets = []
def on_data(self, raw_data):
raw_data = json.loads(raw_data)
if "limit" not in raw_data:
if raw_data['lang'] == args['lang'] \
and "@" not in raw_data['text'] \
and is_ascii(raw_data['text'])\
and len(raw_data['text']) < int(args['length'])\
and "http" not in raw_data['text']:
print raw_data['text']
return True
def on_error(self, status_code):
print status_code
def all_tweets(self):
return self.allTweets
l = StdOutListener()
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token_key, access_token_secret)
stream = Stream(auth, l)
stream.filter(track=args['term'])
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment