Created
August 20, 2021 17:14
-
-
Save impshum/be5fb4e57b26389ea4bd6f8efc108715 to your computer and use it in GitHub Desktop.
None... Bugger off!
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import sqlite3 | |
from sqlite3 import Error | |
import tweepy | |
import configparser | |
# IMPORT CONFIG VARIABLES | |
config = configparser.ConfigParser() | |
config.read('conf.ini') | |
twitter_consumer_key = config['TWITTER']['twitter_consumer_key'] | |
twitter_consumer_secret = config['TWITTER']['twitter_consumer_secret'] | |
twitter_access_token = config['TWITTER']['twitter_access_token'] | |
twitter_access_token_secret = config['TWITTER']['twitter_access_token_secret'] | |
# TWATTER AUTH | |
auth = tweepy.OAuthHandler(twitter_consumer_key, twitter_consumer_secret) | |
auth.set_access_token(twitter_access_token, twitter_access_token_secret) | |
api = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True) | |
# SQLITE CONNECT | |
def db_connect(): | |
try: | |
conn = sqlite3.connect('data.db') | |
create_table = """CREATE TABLE IF NOT EXISTS tweets ( | |
ID INTEGER PRIMARY KEY AUTOINCREMENT, | |
tweet_id TEXT NOT NULL);""" | |
conn.execute(create_table) | |
return conn | |
except Error as e: | |
print(e) | |
# INSERT SQLITE ROW | |
def insert_row(conn, tweet_id): | |
cur = conn.cursor() | |
cur.execute("SELECT * FROM tweets WHERE tweet_id = ?", (tweet_id,)) | |
if not len(cur.fetchall()): | |
conn.execute("INSERT INTO tweets (tweet_id) VALUES (?);", (tweet_id,)) | |
conn.commit() | |
return True | |
# READ ALL SQLITE ROWS | |
def read_db(conn): | |
cur = conn.cursor() | |
cur.execute("SELECT * FROM tweets") | |
rows = cur.fetchall() | |
for row in rows: | |
print(row) | |
# MAIN FUCKERY | |
def main(): | |
conn = db_connect() | |
for tweet in api.user_timeline(): | |
if insert_row(conn, tweet.id_str): | |
print('POST TO TELEGRAM') | |
else: | |
print('OLD POST') | |
def test(): | |
conn = db_connect() | |
read_db(conn) | |
# ROCK & ROLL | |
if __name__ == '__main__': | |
#main() | |
test() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment