Spaces:
Running
Running
| import gradio as gr | |
| import json | |
| import time | |
| import random | |
| from datetime import datetime | |
| from typing import Dict, Any, Optional, Tuple | |
| import hashlib | |
| # Import utility functions | |
| from utils import ( | |
| simulate_secure_processing, | |
| get_agents_data, | |
| process_with_agent, | |
| SecurityMetrics | |
| ) | |
| # Initialize global state | |
| security_metrics = SecurityMetrics() | |
| def create_agent_selector(): | |
| """Create agent selection UI with all 6 agents""" | |
| agents = get_agents_data() | |
| agent_info = "" | |
| for agent in agents: | |
| caps = ", ".join(agent["capabilities"][:3]) | |
| agent_info += f"**{agent['name']}** ({agent['type']})\n- Status: {agent['status']}\n- Capabilities: {caps}\n\n" | |
| return agent_info | |
| def process_securely(agent_id: str, operation: str, text_input: str, file_input) -> Tuple[str, str]: | |
| """ | |
| Main processing function that handles secure operations | |
| """ | |
| if not agent_id: | |
| return "❌ Please select an agent first", "" | |
| if not text_input and not file_input: | |
| return "❌ Please provide text input or upload a file", "" | |
| try: | |
| # Process the input based on agent and operation | |
| result = process_with_agent( | |
| agent_id=agent_id, | |
| operation=operation, | |
| text_input=text_input, | |
| file_input=file_input, | |
| metrics=security_metrics | |
| ) | |
| # Format output | |
| output_text = f"""✅ **Processing Successful** | |
| **Agent:** {result['agent_name']} | |
| **Operation:** {operation.title()} | |
| **Security:** {result['security']['encryption']} | |
| **Timestamp:** {result['security']['timestamp']} | |
| **Result:** | |
| {json.dumps(result['data'], indent=2)}""" | |
| security_details = f"""🔒 **Security Details** | |
| - **Hash:** {result['security']['hash'][:32]}... | |
| - **Encryption:** {result['security']['encryption']} | |
| - **Agent ID:** {result['security']['agent_id']} | |
| - **Threats Blocked:** {security_metrics.threats_blocked} | |
| - **Data Units Processed:** {security_metrics.data_processed}""" | |
| return output_text, security_details | |
| except Exception as e: | |
| error_msg = f"❌ Processing failed: {str(e)}" | |
| return error_msg, "" | |
| def update_dashboard(): | |
| """Update privacy dashboard metrics""" | |
| metrics = security_metrics.get_metrics() | |
| dashboard_text = f"""### 📊 **Security Metrics** | |
| - **Encryption Level:** {metrics['encryption_level']} | |
| - **Threats Blocked:** {metrics['threats_blocked']} | |
| - **Data Processed:** {metrics['data_processed']} units | |
| - **Uptime:** {metrics['uptime']} | |
| ### 🛡️ **Privacy Guarantees** | |
| ✓ End-to-end encryption | |
| ✓ Zero data retention | |
| ✓ Local processing priority | |
| ✓ Anonymous usage tracking | |
| ### 📈 **Live Activity** | |
| {metrics['recent_activity']}""" | |
| return dashboard_text | |
| def on_agent_select(agent_name: str) -> str: | |
| """Handle agent selection""" | |
| if not agent_name: | |
| return "No agent selected" | |
| agents = get_agents_data() | |
| agent = next((a for a in agents if a['name'] == agent_name), None) | |
| if agent: | |
| caps = "\n".join([f"- {cap}" for cap in agent['capabilities']]) | |
| return f"""**{agent['name']}** ({agent['type']}) | |
| **Status:** {agent['status']} | |
| **Capabilities:** | |
| {caps}""" | |
| return "" | |
| # Build the Gradio interface | |
| with gr.Blocks() as demo: | |
| # Header with title and anycoder link | |
| gr.Markdown( | |
| """ | |
| # 🔐 Multi-Agent Secure Media Platform | |
| ### Protecting your digital footprint with military-grade security, AI-powered analysis, and creative problem-solving capabilities. Your data never leaves your control. | |
| **[Built with anycoder](https://huggingface.co/spaces/akhaliq/anycoder)** | |
| """, | |
| elem_classes="header" | |
| ) | |
| # Main layout | |
| with gr.Row(): | |
| # Left column: Agent Selection | |
| with gr.Column(scale=1): | |
| gr.Markdown("### 🤖 Select Security Agent") | |
| agents = get_agents_data() | |
| agent_names = [agent['name'] for agent in agents] | |
| agent_names.insert(0, "← Choose an agent") | |
| agent_dropdown = gr.Dropdown( | |
| choices=agent_names, | |
| value="← Choose an agent", | |
| label="Available Agents", | |
| info="Each agent specializes in different security tasks", | |
| interactive=True | |
| ) | |
| agent_info_box = gr.Markdown( | |
| value="Select an agent to view capabilities", | |
| label="Agent Details" | |
| ) | |
| # Bind agent selection | |
| agent_dropdown.change( | |
| fn=on_agent_select, | |
| inputs=agent_dropdown, | |
| outputs=agent_info_box, | |
| api_visibility="private" | |
| ) | |
| # Right column: Secure Processing | |
| with gr.Column(scale=2): | |
| gr.Markdown("### 🔒 Secure Processing") | |
| with gr.Row(): | |
| operation_dropdown = gr.Dropdown( | |
| choices=["analyze", "encrypt", "sanitize", "process"], | |
| value="analyze", | |
| label="Operation Type", | |
| info="Choose how to process your data", | |
| scale=1 | |
| ) | |
| text_input = gr.Textbox( | |
| label="Input Text", | |
| placeholder="Enter text to process securely...", | |
| lines=4, | |
| max_lines=10, | |
| autoscroll=True | |
| ) | |
| file_input = gr.File( | |
| label="Or Upload File", | |
| file_types=["image", "audio", "video", "text", ".pdf", ".doc", ".docx"], | |
| file_count="single", | |
| interactive=True | |
| ) | |
| process_btn = gr.Button( | |
| "🛡️ Process Securely", | |
| variant="primary", | |
| size="lg" | |
| ) | |
| with gr.Row(): | |
| output_text = gr.Textbox( | |
| label="Processing Result", | |
| lines=10, | |
| max_lines=15, | |
| interactive=False, | |
| show_copy_button=True | |
| ) | |
| security_details = gr.Textbox( | |
| label="Security Details", | |
| lines=10, | |
| max_lines=15, | |
| interactive=False, | |
| show_copy_button=True | |
| ) | |
| # Bind processing button | |
| process_btn.click( | |
| fn=process_securely, | |
| inputs=[agent_dropdown, operation_dropdown, text_input, file_input], | |
| outputs=[output_text, security_details], | |
| api_visibility="public", | |
| show_progress="full" | |
| ) | |
| # Bottom section: Privacy Dashboard | |
| gr.Markdown("### 📊 Privacy & Security Dashboard") | |
| dashboard_output = gr.Textbox( | |
| label="Live Security Metrics", | |
| lines=12, | |
| interactive=False, | |
| every=2.0 # Auto-update every 2 seconds | |
| ) | |
| # Timer for live dashboard updates | |
| timer = gr.Timer(value=2.0, active=True) | |
| timer.tick( | |
| fn=update_dashboard, | |
| outputs=dashboard_output, | |
| api_visibility="private", | |
| show_progress="hidden" | |
| ) | |
| # Footer | |
| gr.Markdown( | |
| """ | |
| --- | |
| © 2024 SecureAgent Platform. All data processed with maximum privacy protection. | |
| Built with advanced encryption and zero-knowledge architecture. | |
| """, | |
| elem_classes="footer" | |
| ) | |
| # Launch with Gradio 6 theming | |
| demo.launch( | |
| theme=gr.themes.Soft( | |
| primary_hue="blue", | |
| secondary_hue="indigo", | |
| neutral_hue="slate", | |
| font=gr.themes.GoogleFont("Inter"), | |
| text_size="lg", | |
| spacing_size="lg", | |
| radius_size="md" | |
| ).set( | |
| button_primary_background_fill="*primary_600", | |
| button_primary_background_fill_hover="*primary_700", | |
| block_title_text_weight="600", | |
| block_background_fill="*neutral_50", | |
| block_border_width="1px", | |
| block_border_color="*neutral_200" | |
| ), | |
| css=""" | |
| .header { text-align: center; margin-bottom: 2rem; } | |
| .footer { text-align: center; margin-top: 2rem; font-size: 0.875rem; } | |
| .gradio-container { max-width: 1400px; margin: 0 auto; } | |
| #component-0 { border-bottom: 2px solid #e2e8f0; padding-bottom: 1rem; } | |
| """, | |
| footer_links=[ | |
| {"label": "Built with anycoder", "url": "https://huggingface.co/spaces/akhaliq/anycoder"} | |
| ], | |
| show_error=True, | |
| allowed_paths=["./uploads"] | |
| ) |