app.py

import tweepy
import os
import json
import mysql.connector
import gradio as gr

bearer_token=os.environ["BEARER_KEY"]

consumer_key = os.environ["CONSUMER_KEY"]
consumer_secret = os.environ["CONSUMER_SECRET"]
access_token = os.environ["ACCESS_TOKEN"]
access_token_secret = os.environ["ACCESS_TOKEN_SECRET"]
db_name = os.environ["DB_NAME"]
db_host = os.environ["DB_HOST"]
db_port = os.environ["DB_PORT"]
db_user = os.environ["DB_USER"]
db_ps = os.environ["DB_PS"]


auth = tweepy.OAuth1UserHandler(
    consumer_key=consumer_key, consumer_secret=consumer_secret,
    access_token=access_token, access_token_secret=access_token_secret
)
api = tweepy.API(auth)

def mysql_exe_insertorupdate(sql):
    db = mysql.connector.connect(host = db_host,port = db_port ,user =db_user, password=db_ps, database=db_name, charset='utf8')
    cursor  = db.cursor()
    cursor.execute(sql)
    db.commit()
    db.close()

def mysql_exe_select(sql):
    db = mysql.connector.connect(host = db_host,port = db_port ,user =db_user, password=db_ps, database=db_name, charset='utf8')
    cursor  = db.cursor()
    cursor.execute(sql)
    results = cursor.fetchall()
    db.commit()
    db.close()
    return results

def get_twitter_user_followers(user_id):
    select_by_id = "SELECT user_id, user_name, JSON_EXTRACT(user_info, '$.followers_count') AS followers_count FROM tw_user_info WHERE user_id = '"+user_id+"' and create_time >= DATE_SUB(NOW(), INTERVAL 15 DAY)"
    select_data = mysql_exe_select(select_by_id)
    result = {}
    if(select_data):
        result['id'] = select_data[0][0]
        result['name'] = select_data[0][1]
        result['followers_count'] = select_data[0][2]
        return json.dumps(result)
    else:
        response = api.get_user(user_id=user_id)
        data = response._json
        data_str = json.dumps(json.dumps(data))[1:-1]
        sql = "INSERT INTO tw_user_info(`user_id`, `user_name`, `user_info`) VALUES ('"+data['id_str']+"','"+data['name']+"', '"+data_str+"')"
        mysql_exe_insertorupdate(sql)
    
        if "followers_count" in data:
            result['id'] = data['id']
            result['name'] = data['name']
            result['followers_count'] = data['followers_count']
            return result
        else:
            raise Exception("Unable to retrieve user data")


def get_followers_counts(user_ids):
    followers_counts = []
    for user_id in user_ids:
        try:
            followers_count = get_twitter_user_followers(user_id)
            followers_counts.append(followers_count)
        except Exception as e:
            followers_counts.append(str(e))
    return followers_counts

def predict(user_ids_json):
    user_ids = json.loads(user_ids_json)
    followers_counts = get_followers_counts(user_ids)
    return followers_counts

iface = gr.Interface(fn=predict, inputs="text", outputs="text")
iface.launch()