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