#!/usr/bin/env python # coding: utf-8 import json import os import re import time from random import random import socket from threading import Thread from time import sleep test_html = '''

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%

14+
Partners

55+
Projects Done

89+
Happy Clients

150+
Meetings

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.

''' import gradio as gr import tweepy from fastapi import FastAPI, Request 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') global_oauth1_user_handler = None bearer = os.getenv('BEARER') oauth1_user_handler = tweepy.OAuth1UserHandler( consumer_token, consumer_secret, callback="http://127.0.0.1:7860/" ) target_website = oauth1_user_handler.get_authorization_url(signin_with_twitter=True) block = gr.Blocks(css=".container { max-width: 800px; margin: auto; }") chat_history = [] def get_client_from_tokens(oauth_verifier, oauth_token): new_oauth1_user_handler = tweepy.OAuth1UserHandler( consumer_token, consumer_secret, callback="http://127.0.0.1:7860/" ) 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 ) return their_client def get_oath_headers(): oauth_verifier = None oauth_token = None did_find = False if hasattr(block, "server"): for connection in block.server.server_state.connections: # connection_app_id = connection.app.app.blocks.app_id # if active_app_id == connection_app_id: # print("Its a match") if connection.headers != None: for header in connection.headers: header = header[1].decode() if "oauth_verifier" in header: oauth_verifier = re.search(r"oauth_verifier=(.+)", header).group(1) oauth_token = re.search(r"oauth_token=(.+)&", header).group(1) if oauth_token and oauth_verifier: did_find = True break if did_find: break return oauth_verifier, oauth_token def block_users(client, threshold, dataset): 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 threshold >= user["threshold"]: user = user["username"].strip() user_id = client.get_user(username=user) finished = False while not finished: try: client.block(target_user_id=user_id.data.id) except tweepy.errors.TooManyRequests as e: print(e) time.sleep(240) continue 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 has_oath_header(): headers = get_oath_headers() if headers[0] == None: return False else: return True username_populated = False def chat(radio_score = None, selected_option = None): global client history = [] # app id if radio_score != None and selected_option != None: response = "no blocking" if client != None: chat_history.append(["Model tuned to a '{}%' threshold and is using the '{}' dataset.".format(radio_score, selected_option), "{} Account blocking initialised".format(selected_option.capitalize())]) num_users_blocked = block_users(client,radio_score,selected_option) chat_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"]) return chat_history def infer(prompt): pass have_initialised = False client = None name = None def changed_tab(): 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."]] if client != None and name != "no username": chat_history = [["Welcome {}".format(name), "Initialising WatchTower"]] print("changed tabs - {}".format(name)) chatbot.value = chat_history chatbot.update(value=chat_history) elif has_oath_header() and client==None: tokens = get_oath_headers() if tokens[0] and client==None: client = get_client_from_tokens(tokens[0],tokens[1]) name = client.get_me().data.name have_initialised = True chat_history = [["Welcome {}".format(name), "Initialising WatchTower"]] chatbot.value = chat_history chatbot.update(value=chat_history) elif not has_oath_header() and not have_initialised: chatbot.value = chat_history chatbot.update(value=chat_history) 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(test_html) prediction_tab = gr.TabItem("Getting Started") with prediction_tab: gr.HTML('''

WATCH Tower

''') with gr.Group(): with gr.Box(): with gr.Row().style(mobile_collapse=False, equal_height=True): gr.HTML( value='
'.format( target_website)) 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=80, label="Threshold Certainty Tolerance") chatbot = gr.Chatbot(value=chat_history, label="Watchtower Output").style() btn = gr.Button("Run WatchTower").style(full_width=True) #radio.change(fn=chat, inputs=[radio], outputs=chatbot) #slider.change(fn=chat, inputs=[slider], outputs=chatbot) #text.submit(fn=chat, inputs=[text,text], outputs=chatbot) btn.click(fn=chat, inputs=[slider,radio], outputs=chatbot) tabs.change(fn=changed_tab, inputs=None, outputs=None) gr.Markdown( """___
Created by Boris Dayma et al. 2021-2022
GitHub | Project Report
""" ) block.launch(enable_queue=False)