Spaces:
Sleeping
Sleeping
| 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='utf8mb4') | |
| 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='utf8mb4') | |
| 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 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`,`followers_count`, `user_info`) VALUES (\""+data['id_str']+"\",\""+data['name']+"\",\""+str(data['followers_count'])+"\", \""+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: | |
| print(e) | |
| result={} | |
| result['id'] = user_id | |
| result['reason'] = str(e) | |
| followers_counts.append(result) | |
| return followers_counts | |
| def predict(user_ids_json): | |
| user_ids = json.loads(user_ids_json) | |
| followers_counts = get_followers_counts(user_ids) | |
| return json.dumps(followers_counts) | |
| with gr.Blocks(css="") as demo: | |
| with gr.Row(): | |
| input_message = gr.Textbox(show_label=False, placeholder="Please enter a list of user ids", visible=True).style(container=False) | |
| out_message = gr.Textbox(show_label=False, visible=True).style(container=False) | |
| with gr.Row(): | |
| btn_submit = gr.Button("Submit") | |
| btn_submit.click(predict,input_message, out_message, api_name='submit_message') | |
| demo.queue(concurrency_count=100) | |
| demo.launch() | |
| # iface = gr.Interface(fn=predict, inputs="text", outputs="text",api_name='getUserInfo') | |
| # iface.launch() |