""" Reusable UI components for the Gradio interface. """ import json import time from typing import Any, Callable, Dict, List, Optional, Tuple import gradio as gr import pandas as pd import plotly.express as px import plotly.graph_objects as go from .themes import ( create_info_card, create_progress_bar, create_status_indicator, format_search_result, ) def create_header() -> gr.HTML: """Create the header section.""" header_html = """

🏆 Intelligent Document Assistant

Transform your documents into intelligent conversations! Upload PDFs, DOCX, and TXT files to create a smart knowledge base. Chat naturally, search semantically, and get AI-powered insights from your content.

New Feature: Conversational AI with hybrid search capabilities | 🚀 Multi-modal responses: Chat, RAG, and Hybrid modes | 📊 Advanced analytics: Track usage and optimize performance

""" return gr.HTML(header_html) def create_file_upload_section() -> Tuple[gr.File, gr.HTML, gr.Button]: """Create file upload section.""" file_upload = gr.File( label="Upload Documents", file_types=[".pdf", ".docx", ".txt"], file_count="multiple", interactive=True, height=150, ) upload_status = gr.HTML( create_status_indicator("ready", "Ready to upload documents"), visible=True ) # Make the button more prominent by putting it in a separate row with gr.Row(): with gr.Column(scale=1): gr.HTML("") # Empty space with gr.Column(scale=2): upload_button = gr.Button( "🚀 Process Documents", variant="primary", size="lg", interactive=False, elem_classes=["process-button"], ) with gr.Column(scale=1): gr.HTML("") # Empty space return file_upload, upload_status, upload_button def create_search_interface() -> Tuple[gr.Textbox, gr.Row, gr.Button]: """Create search interface components.""" search_query = gr.Textbox( label="Search Query", placeholder="Ask a question about your documents...", lines=2, max_lines=4, interactive=True, scale=4, ) with gr.Row() as search_controls: with gr.Column(scale=1): search_mode = gr.Dropdown( choices=["hybrid", "vector", "bm25"], value="hybrid", label="Search Mode", interactive=True, ) with gr.Column(scale=1): num_results = gr.Slider( minimum=1, maximum=20, value=10, step=1, label="Number of Results", interactive=True, ) with gr.Column(scale=1): enable_reranking = gr.Checkbox( label="Enable Re-ranking", value=True, interactive=True ) search_button = gr.Button("Search", variant="primary", size="lg", interactive=False) return ( search_query, search_controls, search_button, search_mode, num_results, enable_reranking, ) def create_results_display() -> Tuple[gr.HTML, gr.JSON, gr.HTML]: """Create results display components.""" results_html = gr.HTML( "
No search results yet. Upload documents and try searching!
", visible=True, ) results_json = gr.JSON( value={}, # Initialize with empty dict to prevent parsing errors label="Detailed Results (JSON)", visible=False, ) search_stats = gr.HTML(visible=False) return results_html, results_json, search_stats def create_document_management() -> Tuple[gr.HTML, gr.Button, gr.Button]: """Create document management interface.""" document_list = gr.HTML( "
No documents uploaded yet.
" ) with gr.Row(): refresh_docs_btn = gr.Button("Refresh List", variant="secondary") clear_docs_btn = gr.Button("Clear All Documents", variant="stop") return document_list, refresh_docs_btn, clear_docs_btn def create_system_status() -> gr.HTML: """Create system status display.""" return gr.HTML( create_status_indicator("loading", "Initializing system..."), visible=True ) def create_analytics_dashboard() -> Tuple[gr.HTML, gr.Plot, gr.Plot, gr.Dataframe]: """Create analytics dashboard components.""" # System overview cards system_overview = gr.HTML( "
" ) # Query analytics chart query_chart = gr.Plot(label="Queries Over Time", visible=False) # Search modes chart search_modes_chart = gr.Plot(label="Search Modes Distribution", visible=False) # Recent activity table activity_table = gr.Dataframe( headers=["Timestamp", "Activity", "Details", "Status"], label="Recent Activity", visible=False, ) return system_overview, query_chart, search_modes_chart, activity_table def format_document_list(documents: List[Dict[str, Any]]) -> str: """Format document list as HTML.""" if not documents: return "
No documents uploaded yet.
" html_parts = ["
"] for doc in documents: filename = doc.get("filename", "Unknown") chunk_count = doc.get("chunk_count", 0) file_type = doc.get("file_type", "unknown").upper() file_size = doc.get("file_size", 0) # Format file size if file_size > 1024 * 1024: size_str = f"{file_size / (1024 * 1024):.1f} MB" elif file_size > 1024: size_str = f"{file_size / 1024:.1f} KB" else: size_str = f"{file_size} bytes" doc_html = f"""
📄 {filename}
""" html_parts.append(doc_html) html_parts.append("
") return "".join(html_parts) def format_search_results( results: List[Dict[str, Any]], search_time: float, query: str ) -> Tuple[str, str]: """Format search results as HTML and create search statistics.""" if not results: results_html = """
🔍
No results found for your query.
Try different keywords or check your search settings.
""" stats_html = f"""
Search completed in {search_time:.2f}s - No results found
""" return results_html, stats_html # Format results results_parts = [ f"

Search Results for: \"{query}\"

" ] for i, result in enumerate(results, 1): result_html = format_search_result(result, i) results_parts.append(result_html) results_html = "".join(results_parts) # Create search statistics avg_score = sum(r.get("scores", {}).get("final_score", 0) for r in results) / len( results ) stats_html = f"""
{len(results)}
Results Found
{search_time:.2f}s
Search Time
{avg_score:.3f}
Avg Score
""" return results_html, stats_html def create_analytics_charts( analytics_data: Dict[str, Any], ) -> Tuple[go.Figure, go.Figure]: """Create analytics charts.""" system_data = analytics_data.get("system", {}) queries_data = analytics_data.get("queries_24h", {}) # Queries over time chart queries_per_hour = queries_data.get("queries_per_hour", []) hours = list(range(len(queries_per_hour))) query_fig = go.Figure() query_fig.add_trace( go.Scatter( x=hours, y=queries_per_hour, mode="lines+markers", name="Queries per Hour", line=dict(color="#667eea", width=3), marker=dict(size=8, color="#667eea"), ) ) query_fig.update_layout( title="Queries Over Time (24 Hours)", xaxis_title="Hours Ago", yaxis_title="Number of Queries", template="plotly_white", height=300, ) # Search modes distribution search_modes = queries_data.get("search_modes", {}) if search_modes: modes = list(search_modes.keys()) counts = list(search_modes.values()) modes_fig = go.Figure( data=[ go.Pie( labels=modes, values=counts, hole=0.3, marker=dict(colors=["#667eea", "#8b5cf6", "#06b6d4"]), ) ] ) modes_fig.update_layout( title="Search Modes Distribution", template="plotly_white", height=300 ) else: modes_fig = go.Figure() modes_fig.update_layout( title="Search Modes Distribution", template="plotly_white", height=300, annotations=[dict(text="No data available", showarrow=False)], ) return query_fig, modes_fig def format_system_overview(analytics_data: Dict[str, Any]) -> str: """Format system overview cards.""" system_data = analytics_data.get("system", {}) conversation_data = analytics_data.get("conversation", {}) cards_html = """
""" # Total queries total_queries = system_data.get("total_queries", 0) cards_html += create_info_card( "Total Queries", str(total_queries), "All-time search queries" ) # Documents processed total_docs = system_data.get("total_documents_processed", 0) cards_html += create_info_card( "Documents", str(total_docs), "Successfully processed" ) # Conversation metrics (if available) if conversation_data: total_conversations = conversation_data.get("total_conversations", 0) if total_conversations > 0: cards_html += create_info_card( "Conversations", str(total_conversations), "Chat sessions completed" ) total_messages = conversation_data.get("total_messages", 0) if total_messages > 0: cards_html += create_info_card( "Messages", str(total_messages), "Chat messages exchanged" ) # Uptime uptime_hours = system_data.get("uptime_hours", 0) uptime_str = ( f"{uptime_hours:.1f}h" if uptime_hours < 24 else f"{uptime_hours/24:.1f}d" ) cards_html += create_info_card("Uptime", uptime_str, "System running time") # Active sessions active_sessions = system_data.get("active_sessions", 0) cards_html += create_info_card( "Active Users", str(active_sessions), "Current sessions" ) cards_html += "
" return cards_html def create_progress_callback() -> Callable: """Create a progress callback function for document processing.""" def progress_callback(message: str, progress: float) -> str: return create_progress_bar(progress, message) return progress_callback def create_error_display(error_message: str) -> str: """Create error display HTML.""" return f"""
⚠️ Error
{error_message}
""" def create_success_display(message: str) -> str: """Create success display HTML.""" return f"""
Success
{message}
""" def create_loading_display(message: str = "Processing...") -> str: """Create loading display HTML.""" return f"""
{message}
"""