Update app.py from anycoder

app.py

@@ -0,0 +1,228 @@
+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"}
+    ]
+)