#!/usr/bin/env python # coding: utf-8 # Imports import json import os import time import gradio as gr import tweepy # The HTML body used for the WatchTower page html_body = '''
Architecture

WATCH Tower

Block Violent Content Before It Reaches Your Feed

WatchTower identifies, blocks, and filters out violent and radical content before it reaches your Twitter feed.


WatchTower works to protect you from violent, misinformation, hate speech and other malicious communication by using a suite of machine learning models to identify user accounts that post content that commonly falls into these categories. WatchTower is broken down into two components, the first utilises the Twitter streaming API and applies a suite of machine learning models to identify users that commonly post malicious information, while the second element provides a web UI where users can authenticaate with Twitter and tailor the types and thresholds for the accounts they block.


WatchTower was developed solely by James Stevenson and primarily uses Pinpoint, a machine learning model also developed by James. The future roadmap sees WatchTower incoperate other models for identifying contrent such as misinformation and hate speech. More on Pinpoint and the model WatchTower uses to identify violent extremism can be seen below.

Model Accuracy:

Machine learning models can be validated based on several statistics. These statistics for Pinpoint the main ML model used by WatchTower can be seen below.


Accuracy

73%

Recall

62%

Precision

78%

F-Measure

69%

Chirp Development Challenge 2022

WatchTower was developed for the Chirp 2022 Twitter API Developer Challenge

Watchtower was developed solely by James Stevenson for the Chirp 2022 Twitter API Developer Challenge. More infomration of this can be found below.


Architecture

''' # Twitter keys consumer_token = os.getenv('CONSUMER_TOKEN') consumer_secret = os.getenv('CONSUMER_SECRET') my_access_token = os.getenv('ACCESS_TOKEN') my_access_secret = os.getenv('ACCESS_SECRET') bearer = os.getenv('BEARER') # Setup the gradio block and add some generic CSS block = gr.Blocks(css=".container { max-width: 800px; margin: auto; }", title="WatchTower") # Chat history variable used for the chatbot prompt on the 'getting started' page. chat_history = [] def get_client_from_tokens(oauth_verifier, oauth_token): ''' This function is used for generating a Tweepy client object based on Oauth verifier and token paramiters :param oauth_verifier: :param oauth_token: :return: A Tweepy client object ''' new_oauth1_user_handler = tweepy.OAuth1UserHandler( consumer_token, consumer_secret, callback="https://hf.space/embed/User1342/WatchTower/" ) new_oauth1_user_handler.request_token = { "oauth_token": oauth_token, "oauth_token_secret": consumer_secret } access_token, access_token_secret = new_oauth1_user_handler.get_access_token( oauth_verifier ) their_client = tweepy.Client( bearer_token=bearer, consumer_key=consumer_token, consumer_secret=consumer_secret, access_token=access_token, access_token_secret=access_token_secret ) # TODO: The below is not necessary and can be removed. global client client = their_client return their_client def block_users(client, threshold, dataset): ''' Used for blocking a series of users based on the threshold and datasets provided. Here the users folder is used. TODO: Datasets not implemented. :param client: :param threshold: :param dataset: :return: The number of blocked users. ''' num_users_blocked = 0 for filename in os.listdir("users"): filename = os.path.join("users", filename) user_file = open(filename, "r") users = json.load(user_file) for user in users: if user["threshold"] >= threshold: user_id = str(user["username"]) finished = False while not finished: try: client.block(target_user_id=user_id) except tweepy.errors.TooManyRequests as e: print(e) time.sleep(240) continue time.sleep(1) finished = True me = client.get_me() print("{} blocked {}".format(me.data["username"], user)) num_users_blocked = num_users_blocked + 1 return num_users_blocked def chat(selected_option=None, radio_score=None, url_params=None): ''' This function is used to initialise blocking users once the user has authenticated with Twitter. :param selected_option: :param radio_score: :param url_params: :return: the chatbot history is returned (including information on blocked accounts). ''' global client global chat_history history = [] # app id if "oauth_verifier" in url_params and "oauth_token" in url_params and client is None: client = get_client_from_tokens(url_params["oauth_verifier"], url_params["oauth_token"]) if radio_score != None and selected_option != None: if client != None: history.append( ["Model tuned to a '{}%' threshold and is using the '{}' dataset.".format(radio_score, selected_option), "{} Account blocking initialised".format(str(selected_option).capitalize())]) num_users_blocked = block_users(client, radio_score, selected_option) history.append( ["Blocked {} user account(s).".format(num_users_blocked), "Thank you for using Watchtower."]) elif radio_score != None or selected_option != None: chat_history.append(["Initialisation error!", "Please tune the model by using the above options"]) history = chat_history + history chatbot.value = history chatbot.update(value=history) return history def infer(prompt): pass have_initialised = False client = None name = None def button_pressed(slider_value, url_params): #print(url_params) return [None, chat(radio.value, slider_value, url_params)] # The website that the user will visit to authenticate WatchTower. target_website = None def update_target_website(): ''' Updates the URL used to authenticate WatchTower with Twitter. #TODO this function is full of old code and can be optimised. :return: ''' global have_initialised global chatbot global chat_history global client global name name = "no username" chat_history = [ ["Welcome to Watchtower.".format(name), "Log in via Twitter and configure your blocking options above."]] chatbot.value = chat_history chatbot.update(value=chat_history) twitter_auth_button.value = 'Log In With Twitter
'.format( get_target_website()) twitter_auth_button.update( value='Log In With Twitter
'.format( get_target_website())) return 'Log In With Twitter
'.format( get_target_website()) # The below is a JS blob used to retrieve the URL params. # Thanks to here: https://discuss.huggingface.co/t/hugging-face-and-gradio-url-paramiters/21110/2 get_window_url_params = """ function(text_input, url_params) { console.log(text_input, url_params); const params = new URLSearchParams(window.location.search); url_params = Object.fromEntries(params); return [text_input, url_params]; } """ def get_target_website(): ''' A wrapper function used for retrieving the URL a user will use to authenticate WatchTower with Twitter. :return: ''' oauth1_user_handler = tweepy.OAuth1UserHandler( consumer_token, consumer_secret, callback="https://hf.space/embed/User1342/WatchTower/" ) target_website = oauth1_user_handler.get_authorization_url(signin_with_twitter=True) return target_website # The Gradio HTML component used for the 'sign in with Twitter' button twitter_auth_button = gr.HTML( value='Log In With Twitter
'.format( get_target_website())) # The main chunk of code that uses Gradio blocks to create the UI with block: gr.HTML('''

WATCH Tower

''') gr.HTML("


") # todo check if user signed in user_message = "Log in via Twitter and configure your blocking options above." chat_history.append(["Welcome to Watchtower.", user_message]) tabs = gr.Tabs() with tabs: intro_tab = gr.TabItem("Introduction") with intro_tab: gr.HTML(html_body) prediction_tab = gr.TabItem("Getting Started") with prediction_tab: html_button = gr.HTML('''
Architecture

WATCH Tower

''') with gr.Group(): with gr.Box(): url_params = gr.JSON({}, visible=False, label="URL Params") text_input = gr.Text(label="Input", visible=False) text_output = gr.Text(label="Output", visible=False) with gr.Row().style(mobile_collapse=False, equal_height=True): twitter_auth_button with gr.Row().style(mobile_collapse=False, equal_height=True): radio = gr.CheckboxGroup(value="Violent", choices=["Violent", "Hate Speech", "Misinformation"], interactive=False, label="Behaviour To Block") slider = gr.Slider(value=15, interactive=True, label="Threshold Certainty Tolerance") chatbot = gr.Chatbot(value=chat_history, label="Watchtower Output").style() btn = gr.Button("Run WatchTower").style(full_width=True) btn.click(fn=button_pressed, inputs=[slider, url_params], outputs=[text_output, chatbot], _js=get_window_url_params) tabs.change(fn=update_target_website, inputs=None, outputs=None) gr.Markdown( """___

Created by et al. 2021-2022
GitHub

""" ) block.__enter__() block.set_event_trigger( event_name="load", fn=update_target_website, inputs=None, outputs=[html_button], no_target=True ) block.launch(enable_queue=False)