app.py

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("""
        <div style="text-align: center; margin-bottom: 2rem;">
            <h1>🛡️ Telegram Hate Speech Monitor</h1>
            <p>AI-powered monitoring dashboard for public channel safety</p>
            <p style="font-size: 0.9em; color: gray;">
                Built with <a href="https://huggingface.co/spaces/akhaliq/anycoder" target="_blank">anycoder</a>
            </p>
        </div>
    """)

    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"}
    ]
)