twitter_bot.py

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)