import gradio as gr import pandas as pd import random import time from datetime import datetime # --- Simulated Backend Logic --- # In a real production environment, this would use libraries like `python-telegram-bot` or `Telethon` # to connect to the Telegram API and a model like Hugging Face's `toxic-bert` for detection. class TelegramMonitorSimulator: def __init__(self): self.hate_keywords = ["hate", "violence", "attack", "stupid", "kill", "idiot"] self.sample_messages = [ "Great meeting today everyone!", "I really hate the way this is going.", "Check out this new article.", "This is just stupid, nobody cares.", "Lovely weather we are having.", "We should attack this problem head on.", "Hello world!", "I want to kill this process." ] self.monitored_channels = [] def add_channel(self, channel_name): if channel_name and channel_name not in self.monitored_channels: self.monitored_channels.append(channel_name) return f"Added '{channel_name}' to monitoring list." return "Channel already exists or name is empty." def process_incoming_message(self): """Simulates the bot receiving a message from a monitored channel.""" if not self.monitored_channels: return None, "No channels monitored. Add a channel first." # Pick random channel and message channel = random.choice(self.monitored_channels) text = random.choice(self.sample_messages) # Simple detection logic (Simulating an AI model) detected = any(word in text.lower() for word in self.hate_keywords) score = random.uniform(0.7, 0.99) if detected else random.uniform(0.0, 0.3) return { "id": int(time.time() * 1000), "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "channel": channel, "message": text, "score": round(score, 2), "status": "Pending Review" }, None # Initialize the simulator monitor = TelegramMonitorSimulator() # --- Gradio Application --- def update_dashboard(logs): """Helper to convert logs state to dataframe for display.""" if not logs: return pd.DataFrame(columns=["ID", "Time", "Channel", "Message", "Score", "Status"]) return pd.DataFrame(logs)[["id", "timestamp", "channel", "message", "score", "status"]] def add_channel_handler(channel_input, current_channels): result = monitor.add_channel(channel_input) # Return updated list of channels for the display/State updated_channels = monitor.monitored_channels.copy() return result, updated_channels def simulate_traffic_handler(current_logs): """Simulates the bot detecting a hate speech event.""" new_entry, error = monitor.process_incoming_message() if error: return current_logs, update_dashboard(current_logs), error # Only log if it looks like hate speech (score > 0.6) for the demo purposes # or log everything if you want a full history. Let's log flagged only. if new_entry['score'] > 0.6: updated_logs = current_logs.copy() updated_logs.insert(0, new_entry) # Add to top return updated_logs, update_dashboard(updated_logs), "⚠️ New Potential Hate Speech Detected!" return current_logs, update_dashboard(current_logs), "Message received (clean)." def assess_message(logs, selected_id, assessment): """Updates the status of a selected message.""" if not logs or selected_id is None: return logs, update_dashboard(logs), gr.Dropdown(choices=[], value=None), "No message selected." updated_logs = [] found = False for log in logs: if log['id'] == selected_id: log['status'] = assessment found = True updated_logs.append(log) if found: # Refresh dropdown choices (remove the assessed item from pending) pending_ids = [l['id'] for l in updated_logs if l['status'] == "Pending Review"] return updated_logs, update_dashboard(updated_logs), gr.Dropdown(choices=pending_ids, value=None), f"Message ID {selected_id} marked as {assessment}." return logs, update_dashboard(logs), gr.Dropdown(choices=[l['id'] for l in logs if l['status'] == "Pending Review"], value=None), "Message not found." # --- Gradio UI Construction --- # Note: In Gradio 6, Blocks takes no parameters. Configuration goes in launch(). with gr.Blocks() as demo: # Header gr.HTML("""

🛡️ Telegram Hate Speech Monitor

AI-powered monitoring dashboard for public channel safety

Built with anycoder

""") with gr.Row(): # Left Column: Controls with gr.Column(scale=1): gr.Markdown("### ⚙️ Configuration") with gr.Group(): channel_input = gr.Textbox(label="Channel Username", placeholder="@example_channel") add_channel_btn = gr.Button("Add Channel", variant="primary") channel_status = gr.Textbox(label="Status", interactive=False) monitored_list = gr.JSON(label="Monitored Channels", value=[]) gr.Markdown("### 📡 Live Monitor (Simulated)") gr.Markdown("*Click to simulate incoming messages from monitored channels.*") simulate_btn = gr.Button("Simulate Incoming Traffic", variant="secondary") monitor_status = gr.Textbox(label="Monitor Log", interactive=False) # Right Column: Dashboard with gr.Column(scale=3): gr.Markdown("### 📋 Flagged Content Review") # Dataframe to show logs logs_df = gr.Dataframe( label="Detected Messages", headers=["ID", "Time", "Channel", "Message", "Score", "Status"], datatype=["number", "str", "str", "str", "number", "str"], interactive=False, wrap=True ) gr.Markdown("### ✍️ Human Assessment") with gr.Row(): msg_selector = gr.Dropdown( label="Select Pending Message ID", choices=[], info="Select a message from the list above to review" ) with gr.Row(): approve_btn = gr.Button("✅ Confirm Hate Speech", variant="stop") reject_btn = gr.Button("❌ False Positive", variant="secondary") assessment_feedback = gr.Textbox(label="Assessment Result", interactive=False) # --- State Management --- # We store the raw logs in state so we can manipulate them logs_state = gr.State(value=[]) # --- Event Listeners --- # 1. Add Channel add_channel_btn.click( fn=add_channel_handler, inputs=[channel_input, monitored_list], outputs=[channel_status, monitored_list], api_visibility="private" ) # 2. Simulate Traffic (The Bot Loop) simulate_btn.click( fn=simulate_traffic_handler, inputs=[logs_state], outputs=[logs_state, logs_df, monitor_status], api_visibility="private" ) # 3. Assessment Logic # Helper to update dropdown choices based on current state def update_dropdown(logs): pending = [l['id'] for l in logs if l['status'] == "Pending Review"] return gr.Dropdown(choices=pending, value=None) # When logs change (via simulation), update the dropdown options logs_state.change( fn=update_dropdown, inputs=[logs_state], outputs=[msg_selector] ) approve_btn.click( fn=assess_message, inputs=[logs_state, msg_selector, gr.State("Confirmed Hate Speech")], outputs=[logs_state, logs_df, msg_selector, assessment_feedback], api_visibility="private" ) reject_btn.click( fn=assess_message, inputs=[logs_state, msg_selector, gr.State("False Positive")], outputs=[logs_state, logs_df, msg_selector, assessment_feedback], api_visibility="private" ) # --- Launch --- # In Gradio 6, theme and other app-level params are in launch() demo.launch( theme=gr.themes.Soft( primary_hue="red", secondary_hue="orange", neutral_hue="slate" ), footer_links=[ {"label": "View Source", "url": "https://github.com/gradio-app/gradio"}, {"label": "Built with anycoder", "url": "https://huggingface.co/spaces/akhaliq/anycoder"} ] )