Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
twitter_bot.py
#!/usr/bin/env python3
# MIT license
from tweepy import OAuthHandler, Stream, API
from tweepy.streaming import StreamListener
import sqlite3
import secrets
# auth
# add a file called secrets.py next to bot.py and enter keys and tokens into variables.
auth = OAuthHandler(secrets.consumer_key, secrets.consumer_secret)
auth.set_access_token(secrets.access_token, secrets.access_token_secret)
api = API(auth)
# auth end
class StdOutListener(StreamListener):
def on_status(self, status):
# establish db
conn = sqlite3.connect("db.sqlite3") # db created if no db.sqlite3 found.
c = conn.cursor()
# create db table for data.
c.execute("CREATE TABLE IF NOT EXISTS twitter(ID INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT, screen_name TEXT, lang TEXT, text TEXT)")
# write data.
if status.text.find("RT @") == -1 or status.text.startswith("@"): # ignore manual RTs.
print("\n" + status.lang, status.created_at, status.user.screen_name, status.text)
c.execute("INSERT INTO twitter(created_at, screen_name, lang, text) VALUES (?, ?, ?, ?)",
(status.created_at, status.user.screen_name, status.lang, status.text))
conn.commit() # save to db.
# rename "print" to "api.update_status" in production.
print("Tweets: @" + status.user.screen_name + " tweet text.")
def on_error(self, status_code):
# https://dev.twitter.com/overview/api/response-codes
print("Error: " + status_code)
return True
def on_timeout(self):
print("Timeout Tweepy on_timeout")
return True
if __name__ == "__main__":
stream = Stream(auth, StdOutListener())
stream.filter(track=["search"], async=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment