# enhanced_verification_interface.py """ Enhanced Verification Interface Integration. Integrates the enhanced verification modes with the existing Gradio application. Provides mode selection, session resumption, and progress preservation. Requirements: 1.1, 1.2, 1.3, 1.4, 1.5, 6.1 """ import gradio as gr from typing import List, Dict, Tuple, Optional, Any, Union from datetime import datetime import uuid from src.core.verification_models import ( EnhancedVerificationSession, VerificationRecord, TestMessage, TestDataset, ) from src.core.verification_store import JSONVerificationStore from src.core.test_datasets import TestDatasetManager from src.interface.enhanced_verification_ui import EnhancedVerificationUIComponents # Import configuration with fallback defaults try: from app_config import ( ENHANCED_VERIFICATION_CONFIG, FEATURE_FLAGS, is_feature_enabled ) except ImportError: ENHANCED_VERIFICATION_CONFIG = {"enabled": True, "default_mode": None} FEATURE_FLAGS = { "manual_input_mode_enabled": True, "file_upload_mode_enabled": True, "dataset_editing_enabled": True, "show_incomplete_session_prompts": True, } def is_feature_enabled(feature_name: str) -> bool: return FEATURE_FLAGS.get(feature_name, False) class EnhancedVerificationInterface: """Main interface controller for enhanced verification modes.""" def __init__(self, store: JSONVerificationStore = None, config: dict = None): """ Initialize the enhanced verification interface. Args: store: Verification data store (optional, creates default if not provided) config: Configuration dictionary (optional, uses ENHANCED_VERIFICATION_CONFIG if not provided) """ self.store = store or JSONVerificationStore() self.config = config or ENHANCED_VERIFICATION_CONFIG self.current_mode = self.config.get("default_mode", None) self.current_session = None self.incomplete_sessions = [] # Feature flags for mode availability self.manual_input_enabled = is_feature_enabled("manual_input_mode_enabled") self.file_upload_enabled = is_feature_enabled("file_upload_mode_enabled") self.dataset_editing_enabled = is_feature_enabled("dataset_editing_enabled") self.show_incomplete_prompts = is_feature_enabled("show_incomplete_session_prompts") def create_interface(self) -> gr.Blocks: """ Create the complete enhanced verification interface. Returns: Gradio Blocks component with mode selection and all verification modes """ with gr.Blocks(title="Enhanced Verification Modes") as interface: # Application state current_mode_state = gr.State(value=None) current_session_state = gr.State(value=None) incomplete_sessions_state = gr.State(value=[]) pending_mode_switch_state = gr.State(value=None) selected_session_state = gr.State(value=None) # Model overrides (populated by the main app's AI Model Configuration, if wired) model_overrides_state = gr.State(value={}) # Main container with gr.Column(): # Header gr.Markdown("# 🔍 Enhanced Verification Modes") gr.Markdown("Choose your verification approach based on your testing needs and data source.") # Status message status_message = gr.Markdown("", visible=True, label="Status") # Incomplete sessions section incomplete_sessions_section = gr.Row(visible=False) with incomplete_sessions_section: with gr.Column(): gr.Markdown("## 📋 Resume Previous Sessions") gr.Markdown("You have incomplete verification sessions. You can resume where you left off or start a new session.") incomplete_sessions_display = gr.HTML( value="", label="Incomplete Sessions" ) with gr.Row(): resume_session_btn = gr.Button( "▶️ Resume Selected Session", variant="primary", scale=2 ) clear_sessions_btn = gr.Button( "🗑️ Clear All Sessions", variant="secondary", scale=1 ) # Mode selection section mode_selection_section = gr.Row(visible=True) with mode_selection_section: with gr.Column(): gr.Markdown("## 🎯 Select Verification Mode") with gr.Row(): # Standard Verification Mode (replaced Enhanced Dataset) # NOTE: Per UI simplification request, we keep the implementation # but hide this tile by default. show_standard_verification_tile = False with gr.Column(scale=1, visible=show_standard_verification_tile): gr.Markdown("### ✓ Standard Verification") gr.Markdown("Use predefined test datasets to verify classifier accuracy. Quick and straightforward verification workflow.") gr.Markdown("**Features:**") gr.Markdown("• Pre-classified test datasets") gr.Markdown("• Step-by-step message review") gr.Markdown("• Accuracy tracking and statistics") gr.Markdown("• Export verification results") gr.Markdown("• Session progress preservation") standard_verification_btn = gr.Button( "✓ Start Standard Verification", variant="primary", size="lg" ) # Manual Input Mode with gr.Column(scale=1): mode_info = EnhancedVerificationUIComponents.MODE_OPTIONS["manual_input"] gr.Markdown(f"### {mode_info['icon']} {mode_info['title']}") gr.Markdown(mode_info["description"]) gr.Markdown("**Features:**") for feature in mode_info["features"]: gr.Markdown(f"• {feature}") manual_input_btn = gr.Button( f"{mode_info['icon']} Start Manual Input Mode", variant="primary", size="lg" ) # File Upload Mode with gr.Column(scale=1): mode_info = EnhancedVerificationUIComponents.MODE_OPTIONS["file_upload"] gr.Markdown(f"### {mode_info['icon']} {mode_info['title']}") gr.Markdown(mode_info["description"]) gr.Markdown("**Features:**") for feature in mode_info["features"]: gr.Markdown(f"• {feature}") file_upload_btn = gr.Button( f"{mode_info['icon']} Start File Upload Mode", variant="primary", size="lg" ) # Mode switch confirmation dialog mode_switch_dialog = gr.Row(visible=False) with mode_switch_dialog: with gr.Column(): gr.Markdown("### ⚠️ Switch Mode Confirmation") switch_warning_text = gr.Markdown( "You have unsaved progress in the current mode. What would you like to do?", label="Warning" ) with gr.Row(): save_and_switch_btn = gr.Button( "💾 Save Progress & Switch", variant="primary", scale=2 ) discard_and_switch_btn = gr.Button( "🗑️ Discard & Switch", variant="secondary", scale=1 ) cancel_switch_btn = gr.Button( "❌ Cancel", scale=1 ) # Individual mode interfaces (initially hidden) # Standard Verification interface (replaced Enhanced Dataset) standard_verification_interface = gr.Row(visible=False) with standard_verification_interface: with gr.Column(): gr.Markdown("# ✓ Standard Verification Mode") gr.Markdown("Review classified messages and provide feedback to improve the spiritual distress classifier.") back_from_standard_btn = gr.Button("← Back to Mode Selection", size="sm") gr.Markdown("---") # Create dataset interface components from src.interface.enhanced_dataset_interface import EnhancedDatasetInterfaceController dataset_controller = EnhancedDatasetInterfaceController() # Pre-load dataset choices for initialization try: initial_choices, _, _, _ = dataset_controller.initialize_interface() except Exception: initial_choices = [] # Dataset selection interface with gr.Row(): with gr.Column(scale=2): gr.Markdown("## 📊 Select Dataset") dataset_selector = gr.Dropdown( choices=initial_choices, value=initial_choices[0] if initial_choices else None, label="Available Datasets", info="Choose a dataset to verify", interactive=True ) load_dataset_btn = gr.Button("📥 Load Dataset", variant="primary") with gr.Column(scale=1): gr.Markdown("## 📊 Dataset Information") dataset_info_display = gr.Markdown( "Select a dataset to view details", label="Dataset Details" ) # Verification setup section (initially hidden) verification_section = gr.Row(visible=False) with verification_section: with gr.Column(): gr.Markdown("## 🔍 Dataset Verification") # Verification controls with gr.Row(): with gr.Column(scale=2): verifier_name_input = gr.Textbox( label="Verifier Name", placeholder="Enter your name...", interactive=True ) with gr.Column(scale=1): start_verification_btn = gr.Button( "🚀 Start Verification", variant="primary", size="lg" ) # Progress display verification_progress = gr.Markdown( "Ready to start verification", label="Progress" ) # Message review section (initially hidden) message_review_section = gr.Row(visible=False) with message_review_section: with gr.Column(scale=2): gr.Markdown("## 📝 Message Review") # Current message display current_message_display = gr.Textbox( label="Patient Message", interactive=False, lines=4 ) # Classification result with gr.Row(): classifier_decision_display = gr.Markdown( "🔄 Loading...", label="Classifier Decision" ) classifier_confidence_display = gr.Markdown( "", label="Confidence" ) # Feedback buttons gr.Markdown("### Is this classification correct?") with gr.Row(): correct_btn = gr.Button("✓ Correct", variant="primary", scale=1) incorrect_btn = gr.Button("✗ Incorrect", variant="stop", scale=1) # Correction section (initially hidden) correction_section = gr.Row(visible=False) with correction_section: with gr.Column(): gr.Markdown("### Select correct classification:") correction_selector = gr.Radio( choices=["green", "yellow", "red"], label="Correct Classification", value=None ) correction_notes = gr.Textbox( label="Notes (Optional)", placeholder="Why is this incorrect?", lines=2 ) with gr.Row(): submit_correction_btn = gr.Button("✓ Submit Correction", variant="primary") cancel_correction_btn = gr.Button("✗ Cancel", variant="secondary") with gr.Column(scale=1): gr.Markdown("## 📊 Session Statistics") session_stats_display = gr.Markdown( "**Progress:** 0/0\n**Correct:** 0\n**Incorrect:** 0\n**Accuracy:** 0%" ) # Export section gr.Markdown("## 💾 Export Results") with gr.Column(): export_csv_btn = gr.Button("📄 Export CSV", variant="secondary", size="sm") export_json_btn = gr.Button("📋 Export JSON", variant="secondary", size="sm") export_xlsx_btn = gr.Button("📊 Export Excel", variant="secondary", size="sm") # Navigation gr.Markdown("## 🎮 Session Control") with gr.Row(): skip_btn = gr.Button("⏭️ Skip", variant="secondary") finish_btn = gr.Button("🏁 Finish Session", variant="primary") # Status message for dataset interface dataset_status_message = gr.Markdown("", visible=True) # Dataset state current_dataset_state = gr.State(value=None) verification_session_state = gr.State(value=None) current_message_index_state = gr.State(value=0) current_classification_state = gr.State(value=None) manual_input_interface = gr.Row(visible=False) with manual_input_interface: with gr.Column(): gr.Markdown("# ✏️ Manual Input Mode") gr.Markdown("Enter individual messages for immediate classification and verification.") back_from_manual_btn = gr.Button("← Back to Mode Selection", size="sm") gr.Markdown("---") # Embed the actual interface manual_input_ui = EnhancedVerificationUIComponents.create_manual_input_interface(model_overrides_state) file_upload_interface = gr.Row(visible=False) with file_upload_interface: with gr.Column(): gr.Markdown("# 📁 File Upload Mode") gr.Markdown("Upload CSV or XLSX files for batch processing.") back_from_file_btn = gr.Button("← Back to Mode Selection", size="sm") gr.Markdown("---") # Embed the actual interface file_upload_ui = EnhancedVerificationUIComponents.create_file_upload_interface(model_overrides_state) # Event handlers def initialize_interface(): """Initialize the interface and check for incomplete sessions.""" try: has_incomplete, sessions, display_html = EnhancedVerificationUIComponents.check_for_incomplete_sessions(self.store) if has_incomplete: return ( gr.Row(visible=True), # Show incomplete sessions section display_html, # Display sessions HTML sessions, # Store sessions in state "✨ Welcome back! You have incomplete sessions. You can resume where you left off or start a new session." ) else: return ( gr.Row(visible=False), # Hide incomplete sessions section "", # Empty display [], # Empty sessions list "✨ Welcome to Enhanced Verification Modes! Choose a mode to get started." ) except Exception as e: return ( gr.Row(visible=False), "", [], f"❌ Error initializing interface: {str(e)}" ) def switch_to_mode( mode_type: str, current_mode_val: Optional[str], current_session_val: Optional[EnhancedVerificationSession] ): """Handle mode switching with progress preservation.""" try: # Check if we need to show progress preservation warning has_progress = ( current_session_val is not None and not current_session_val.is_complete and current_session_val.verified_count > 0 ) if has_progress and current_mode_val != mode_type: # Show confirmation dialog warning_msg, show_dialog = EnhancedVerificationUIComponents.create_mode_switch_confirmation( current_mode_val, mode_type, has_progress ) return ( gr.Row(visible=True), # Show confirmation dialog warning_msg, # Warning message mode_type, # Store pending mode switch current_mode_val, # Keep current mode current_session_val, # Keep current session f"⚠️ Confirm mode switch to {EnhancedVerificationUIComponents.MODE_OPTIONS[mode_type]['title']}" ) else: # Direct switch (no progress to preserve) # Return only the 6 values expected by the button click handler result = perform_mode_switch(mode_type, current_session_val) # Extract only the values needed for the 6-output handler return ( result[4], # mode_switch_dialog visibility result[5], # switch_warning_text result[6], # pending_mode_switch_state result[7], # current_mode_state result[8], # current_session_state result[9] # status_message ) except Exception as e: return ( gr.Row(visible=False), # Hide confirmation dialog "", # Clear warning None, # Clear pending switch current_mode_val, # Keep current mode current_session_val, # Keep current session f"❌ Error switching modes: {str(e)}" ) def perform_mode_switch( mode_type: str, session_to_save: Optional[EnhancedVerificationSession] = None, save_progress: bool = True ): """Perform the actual mode switch.""" try: # Save current session if exists and requested if session_to_save and not session_to_save.is_complete and save_progress: self.store.save_session(session_to_save) # Update interface visibility mode_selection_visible = mode_type is None standard_verification_visible = mode_type == "standard_verification" manual_input_visible = mode_type == "manual_input" file_upload_visible = mode_type == "file_upload" # Create status message if mode_type: mode_titles = { "standard_verification": "Standard Verification", "manual_input": "Manual Input", "file_upload": "File Upload" } mode_title = mode_titles.get(mode_type, 'Unknown') status_msg = f"✅ Switched to {mode_title} mode" else: status_msg = "✅ Returned to mode selection" return ( gr.Row(visible=mode_selection_visible), # Mode selection section gr.Row(visible=standard_verification_visible), # Standard verification interface gr.Row(visible=manual_input_visible), # Manual input interface gr.Row(visible=file_upload_visible), # File upload interface gr.Row(visible=False), # Hide confirmation dialog "", # Clear warning message None, # Clear pending mode switch mode_type, # Set current mode None, # Clear current session (will be set by mode interface) status_msg # Status message ) except Exception as e: return ( gr.Row(visible=True), # Show mode selection on error gr.Row(visible=False), # Hide standard verification gr.Row(visible=False), # Hide manual input gr.Row(visible=False), # Hide file upload gr.Row(visible=False), # Hide confirmation dialog "", # Clear warning None, # Clear pending switch None, # Clear current mode None, # Clear current session f"❌ Error performing mode switch: {str(e)}" ) def resume_selected_session( sessions: List[EnhancedVerificationSession], selected_session: Optional[EnhancedVerificationSession] ): """Resume a selected session.""" try: if not selected_session: return ( None, # Current mode None, # Current session "⚠️ No session selected. Please select a session first." ) # Switch to the appropriate mode for this session mode_type = selected_session.mode_type # Update current session self.current_session = selected_session # Perform mode switch to resume session return perform_mode_switch(mode_type, None, False) + (selected_session,) except Exception as e: return ( None, None, f"❌ Error resuming session: {str(e)}" ) def clear_all_sessions(sessions: List[EnhancedVerificationSession]): """Clear all incomplete sessions.""" try: cleared_count = 0 for session in sessions: if self.store.delete_session(session.session_id): cleared_count += 1 return ( gr.Row(visible=False), # Hide incomplete sessions section "", # Clear display [], # Clear sessions list f"✅ Cleared {cleared_count} incomplete session{'s' if cleared_count != 1 else ''}" ) except Exception as e: return ( gr.Row(visible=True), # Keep section visible "Error clearing sessions", # Error display sessions, # Keep sessions f"❌ Error clearing sessions: {str(e)}" ) # Bind initialization interface.load( initialize_interface, outputs=[ incomplete_sessions_section, incomplete_sessions_display, incomplete_sessions_state, status_message ] ) # Helper function for direct mode switch (no confirmation needed) def direct_mode_switch(mode_type: str): """Directly switch to a mode without confirmation.""" mode_selection_visible = False standard_verification_visible = mode_type == "standard_verification" manual_input_visible = mode_type == "manual_input" file_upload_visible = mode_type == "file_upload" mode_titles = { "standard_verification": "Standard Verification", "manual_input": "Manual Input", "file_upload": "File Upload" } mode_title = mode_titles.get(mode_type, 'Unknown') status_msg = f"✅ Switched to {mode_title} mode" return ( gr.Row(visible=mode_selection_visible), # mode_selection_section gr.Row(visible=standard_verification_visible), # standard_verification_interface gr.Row(visible=manual_input_visible), # manual_input_interface gr.Row(visible=file_upload_visible), # file_upload_interface mode_type, # current_mode_state status_msg # status_message ) # Special function for standard verification mode initialization def switch_to_standard_verification(): """Switch to standard verification mode with proper initialization.""" try: # Get first dataset info for display dataset_choices, _, _, _ = dataset_controller.initialize_interface() first_dataset = None dataset_info = "Select a dataset to view details" if dataset_choices: first_info, first_dataset = dataset_controller.get_dataset_info(dataset_choices[0]) dataset_info = first_info return ( gr.Row(visible=False), # mode_selection_section gr.Row(visible=True), # standard_verification_interface gr.Row(visible=False), # manual_input_interface gr.Row(visible=False), # file_upload_interface "standard_verification", # current_mode_state "✅ Switched to Standard Verification mode", # status_message dataset_info, # dataset_info_display first_dataset # current_dataset_state ) except Exception as e: return ( gr.Row(visible=False), # mode_selection_section gr.Row(visible=True), # standard_verification_interface gr.Row(visible=False), # manual_input_interface gr.Row(visible=False), # file_upload_interface "standard_verification", # current_mode_state f"❌ Error initializing verification mode: {str(e)}", # status_message "Error loading datasets", # dataset_info_display None # current_dataset_state ) # Bind mode selection buttons standard_verification_btn.click( switch_to_standard_verification, inputs=[], outputs=[ mode_selection_section, standard_verification_interface, manual_input_interface, file_upload_interface, current_mode_state, status_message, dataset_info_display, current_dataset_state ] ) manual_input_btn.click( lambda: direct_mode_switch("manual_input"), inputs=[], outputs=[ mode_selection_section, standard_verification_interface, manual_input_interface, file_upload_interface, current_mode_state, status_message ] ) file_upload_btn.click( lambda: direct_mode_switch("file_upload"), inputs=[], outputs=[ mode_selection_section, standard_verification_interface, manual_input_interface, file_upload_interface, current_mode_state, status_message ] ) # Bind confirmation dialog buttons save_and_switch_btn.click( lambda pms, cs: perform_mode_switch(pms, cs, True), inputs=[pending_mode_switch_state, current_session_state], outputs=[ mode_selection_section, standard_verification_interface, manual_input_interface, file_upload_interface, mode_switch_dialog, switch_warning_text, pending_mode_switch_state, current_mode_state, current_session_state, status_message ] ) discard_and_switch_btn.click( lambda pms, cs: perform_mode_switch(pms, cs, False), inputs=[pending_mode_switch_state, current_session_state], outputs=[ mode_selection_section, standard_verification_interface, manual_input_interface, file_upload_interface, mode_switch_dialog, switch_warning_text, pending_mode_switch_state, current_mode_state, current_session_state, status_message ] ) cancel_switch_btn.click( lambda: ( gr.Row(visible=False), # Hide dialog "", # Clear warning None, # Clear pending switch "❌ Mode switch cancelled" ), outputs=[ mode_switch_dialog, switch_warning_text, pending_mode_switch_state, status_message ] ) # Helper function to go back to mode selection def go_back_to_mode_selection(): """Return to mode selection screen.""" return ( gr.Row(visible=True), # mode_selection_section gr.Row(visible=False), # standard_verification_interface gr.Row(visible=False), # manual_input_interface gr.Row(visible=False), # file_upload_interface None, # current_mode_state "✅ Returned to mode selection" # status_message ) # Bind Back buttons for each mode back_from_standard_btn.click( go_back_to_mode_selection, outputs=[ mode_selection_section, standard_verification_interface, manual_input_interface, file_upload_interface, current_mode_state, status_message ] ) back_from_manual_btn.click( go_back_to_mode_selection, outputs=[ mode_selection_section, standard_verification_interface, manual_input_interface, file_upload_interface, current_mode_state, status_message ] ) back_from_file_btn.click( go_back_to_mode_selection, outputs=[ mode_selection_section, standard_verification_interface, manual_input_interface, file_upload_interface, current_mode_state, status_message ] ) # Bind session resumption buttons resume_session_btn.click( resume_selected_session, inputs=[incomplete_sessions_state, selected_session_state], outputs=[ current_mode_state, current_session_state, status_message ] ) clear_sessions_btn.click( clear_all_sessions, inputs=[incomplete_sessions_state], outputs=[ incomplete_sessions_section, incomplete_sessions_display, incomplete_sessions_state, status_message ] ) # Dataset interface event handlers def on_dataset_selection_change(dataset_selection): """Handle dataset selection change.""" try: dataset_info, dataset_obj = dataset_controller.get_dataset_info(dataset_selection) return ( dataset_info, # dataset_info_display dataset_obj # current_dataset_state ) except Exception as e: return ( f"❌ Error loading dataset info: {str(e)}", # dataset_info_display None # current_dataset_state ) def on_load_dataset(current_dataset): """Handle load dataset for verification.""" try: if not current_dataset: return ( gr.Row(visible=False), # verification_section "❌ No dataset selected" # dataset_status_message ) return ( gr.Row(visible=True), # verification_section f"✅ Dataset '{current_dataset.name}' loaded for verification" # dataset_status_message ) except Exception as e: return ( gr.Row(visible=False), # verification_section f"❌ Error loading dataset: {str(e)}" # dataset_status_message ) def on_start_verification(current_dataset, verifier_name): """Handle starting verification session.""" try: if not current_dataset: return ( None, # verification_session_state gr.Row(visible=False), # message_review_section "", # current_message_display "", # classifier_decision_display "", # classifier_confidence_display "", # session_stats_display 0, # current_message_index_state None, # current_classification_state "❌ No dataset selected" # dataset_status_message ) if not verifier_name or not verifier_name.strip(): return ( None, # verification_session_state gr.Row(visible=False), # message_review_section "", # current_message_display "", # classifier_decision_display "", # classifier_confidence_display "", # session_stats_display 0, # current_message_index_state None, # current_classification_state "❌ Please enter your name" # dataset_status_message ) success, message, session = dataset_controller.start_verification_session( current_dataset, verifier_name ) if success: # Get first message for verification first_message, classification_result = dataset_controller.get_current_message_for_verification() if first_message: # Format classification display decision = classification_result.get('decision', 'unknown').upper() confidence = classification_result.get('confidence', 0) * 100 decision_colors = {'GREEN': '🟢', 'YELLOW': '🟡', 'RED': '🔴'} decision_badge = f"{decision_colors.get(decision, '❓')} **{decision}**" confidence_text = f"Confidence: **{confidence:.1f}%**" stats_text = f"""**Progress:** 1/{len(current_dataset.messages)} **Correct:** 0 **Incorrect:** 0 **Accuracy:** N/A""" return ( session, # verification_session_state gr.Row(visible=True), # message_review_section first_message.text, # current_message_display decision_badge, # classifier_decision_display confidence_text, # classifier_confidence_display stats_text, # session_stats_display 0, # current_message_index_state classification_result, # current_classification_state message # dataset_status_message ) else: return ( session, # verification_session_state gr.Row(visible=False), # message_review_section "", # current_message_display "", # classifier_decision_display "", # classifier_confidence_display "", # session_stats_display 0, # current_message_index_state None, # current_classification_state "❌ No messages in dataset" # dataset_status_message ) else: return ( None, # verification_session_state gr.Row(visible=False), # message_review_section "", # current_message_display "", # classifier_decision_display "", # classifier_confidence_display "", # session_stats_display 0, # current_message_index_state None, # current_classification_state message # dataset_status_message ) except Exception as e: return ( None, # verification_session_state gr.Row(visible=False), # message_review_section "", # current_message_display "", # classifier_decision_display "", # classifier_confidence_display "", # session_stats_display 0, # current_message_index_state None, # current_classification_state f"❌ Error starting verification: {str(e)}" # dataset_status_message ) def on_correct_feedback(current_dataset, session, msg_index, classification): """Handle correct classification feedback.""" try: success, message, stats = dataset_controller.submit_verification_feedback(True) if success and not stats.get('is_complete', False): # Get next message next_message, next_classification = dataset_controller.get_current_message_for_verification() if next_message: decision = next_classification.get('decision', 'unknown').upper() confidence = next_classification.get('confidence', 0) * 100 decision_colors = {'GREEN': '🟢', 'YELLOW': '🟡', 'RED': '🔴'} decision_badge = f"{decision_colors.get(decision, '❓')} **{decision}**" confidence_text = f"Confidence: **{confidence:.1f}%**" stats_text = f"""**Progress:** {stats['processed'] + 1}/{stats['total']} **Correct:** {stats['correct']} **Incorrect:** {stats['incorrect']} **Accuracy:** {stats['accuracy']:.1f}%""" return ( next_message.text, # current_message_display decision_badge, # classifier_decision_display confidence_text, # classifier_confidence_display stats_text, # session_stats_display msg_index + 1, # current_message_index_state next_classification, # current_classification_state gr.Row(visible=False), # correction_section message # dataset_status_message ) # Session complete stats_text = f"""**🎉 Session Complete!** **Total:** {stats['processed']} **Correct:** {stats['correct']} **Incorrect:** {stats['incorrect']} **Final Accuracy:** {stats['accuracy']:.1f}%""" return ( "✅ Verification session complete!", # current_message_display "🎉 **COMPLETE**", # classifier_decision_display "", # classifier_confidence_display stats_text, # session_stats_display msg_index, # current_message_index_state None, # current_classification_state gr.Row(visible=False), # correction_section f"✅ Session complete! Accuracy: {stats['accuracy']:.1f}%" # dataset_status_message ) except Exception as e: return ( gr.update(), # current_message_display gr.update(), # classifier_decision_display gr.update(), # classifier_confidence_display gr.update(), # session_stats_display msg_index, # current_message_index_state classification, # current_classification_state gr.Row(visible=False), # correction_section f"❌ Error: {str(e)}" # dataset_status_message ) def on_incorrect_feedback(): """Show correction section.""" return ( gr.Row(visible=True), # correction_section "⚠️ Please select the correct classification" # dataset_status_message ) def on_cancel_correction(): """Cancel correction.""" return ( gr.Row(visible=False), # correction_section "❌ Correction cancelled" # dataset_status_message ) def on_submit_correction(current_dataset, session, msg_index, classification, correction, notes): """Submit correction feedback.""" try: if not correction: return ( gr.update(), # current_message_display gr.update(), # classifier_decision_display gr.update(), # classifier_confidence_display gr.update(), # session_stats_display msg_index, # current_message_index_state classification, # current_classification_state gr.Row(visible=True), # correction_section "❌ Please select a classification" # dataset_status_message ) success, message, stats = dataset_controller.submit_verification_feedback(False, correction, notes) if success and not stats.get('is_complete', False): # Get next message next_message, next_classification = dataset_controller.get_current_message_for_verification() if next_message: decision = next_classification.get('decision', 'unknown').upper() confidence = next_classification.get('confidence', 0) * 100 decision_colors = {'GREEN': '🟢', 'YELLOW': '🟡', 'RED': '🔴'} decision_badge = f"{decision_colors.get(decision, '❓')} **{decision}**" confidence_text = f"Confidence: **{confidence:.1f}%**" stats_text = f"""**Progress:** {stats['processed'] + 1}/{stats['total']} **Correct:** {stats['correct']} **Incorrect:** {stats['incorrect']} **Accuracy:** {stats['accuracy']:.1f}%""" return ( next_message.text, # current_message_display decision_badge, # classifier_decision_display confidence_text, # classifier_confidence_display stats_text, # session_stats_display msg_index + 1, # current_message_index_state next_classification, # current_classification_state gr.Row(visible=False), # correction_section message # dataset_status_message ) # Session complete stats_text = f"""**🎉 Session Complete!** **Total:** {stats['processed']} **Correct:** {stats['correct']} **Incorrect:** {stats['incorrect']} **Final Accuracy:** {stats['accuracy']:.1f}%""" return ( "✅ Verification session complete!", # current_message_display "🎉 **COMPLETE**", # classifier_decision_display "", # classifier_confidence_display stats_text, # session_stats_display msg_index, # current_message_index_state None, # current_classification_state gr.Row(visible=False), # correction_section f"✅ Session complete! Accuracy: {stats['accuracy']:.1f}%" # dataset_status_message ) except Exception as e: return ( gr.update(), # current_message_display gr.update(), # classifier_decision_display gr.update(), # classifier_confidence_display gr.update(), # session_stats_display msg_index, # current_message_index_state classification, # current_classification_state gr.Row(visible=True), # correction_section f"❌ Error: {str(e)}" # dataset_status_message ) def on_finish_session(session): """Finish verification session.""" return ( gr.Row(visible=False), # message_review_section "✅ Session finished. You can start a new verification or select another dataset." # dataset_status_message ) def on_export_results(format_type, session): """Export verification results.""" try: if not session: return "❌ No active session to export" success, message, file_path = dataset_controller.export_session_results(format_type) if success: return f"✅ Results exported to {format_type.upper()} format" else: return message except Exception as e: return f"❌ Export error: {str(e)}" def on_skip_message(current_dataset, session, msg_index, classification): """Skip current message without feedback.""" try: if not session or not current_dataset: return ( gr.update(), # current_message_display gr.update(), # classifier_decision_display gr.update(), # classifier_confidence_display gr.update(), # session_stats_display msg_index, # current_message_index_state classification, # current_classification_state "❌ No active session" # dataset_status_message ) # Move to next message without recording feedback dataset_controller.current_message_index += 1 if dataset_controller.current_message_index >= len(current_dataset.messages): # Session complete stats_text = f"""**🎉 Session Complete!** **Total:** {session.verified_count} **Correct:** {session.correct_count} **Incorrect:** {session.incorrect_count} **Skipped:** {dataset_controller.current_message_index - session.verified_count} **Final Accuracy:** {(session.correct_count / session.verified_count * 100) if session.verified_count > 0 else 0:.1f}%""" return ( "✅ Verification session complete!", # current_message_display "🎉 **COMPLETE**", # classifier_decision_display "", # classifier_confidence_display stats_text, # session_stats_display dataset_controller.current_message_index, # current_message_index_state None, # current_classification_state "✅ Session complete!" # dataset_status_message ) # Get next message next_message, next_classification = dataset_controller.get_current_message_for_verification() if next_message: decision = next_classification.get('decision', 'unknown').upper() confidence = next_classification.get('confidence', 0) * 100 decision_colors = {'GREEN': '🟢', 'YELLOW': '🟡', 'RED': '🔴'} decision_badge = f"{decision_colors.get(decision, '❓')} **{decision}**" confidence_text = f"Confidence: **{confidence:.1f}%**" skipped_count = dataset_controller.current_message_index - session.verified_count stats_text = f"""**Progress:** {dataset_controller.current_message_index + 1}/{len(current_dataset.messages)} **Correct:** {session.correct_count} **Incorrect:** {session.incorrect_count} **Skipped:** {skipped_count} **Accuracy:** {(session.correct_count / session.verified_count * 100) if session.verified_count > 0 else 0:.1f}%""" return ( next_message.text, # current_message_display decision_badge, # classifier_decision_display confidence_text, # classifier_confidence_display stats_text, # session_stats_display dataset_controller.current_message_index, # current_message_index_state next_classification, # current_classification_state "⏭️ Message skipped" # dataset_status_message ) else: return ( "No more messages", # current_message_display "", # classifier_decision_display "", # classifier_confidence_display "All messages processed", # session_stats_display msg_index, # current_message_index_state None, # current_classification_state "✅ All messages processed" # dataset_status_message ) except Exception as e: return ( gr.update(), # current_message_display gr.update(), # classifier_decision_display gr.update(), # classifier_confidence_display gr.update(), # session_stats_display msg_index, # current_message_index_state classification, # current_classification_state f"❌ Error skipping: {str(e)}" # dataset_status_message ) # Bind dataset interface events dataset_selector.change( on_dataset_selection_change, inputs=[dataset_selector], outputs=[dataset_info_display, current_dataset_state] ) load_dataset_btn.click( on_load_dataset, inputs=[current_dataset_state], outputs=[verification_section, dataset_status_message] ) start_verification_btn.click( on_start_verification, inputs=[current_dataset_state, verifier_name_input], outputs=[ verification_session_state, message_review_section, current_message_display, classifier_decision_display, classifier_confidence_display, session_stats_display, current_message_index_state, current_classification_state, dataset_status_message ] ) correct_btn.click( on_correct_feedback, inputs=[current_dataset_state, verification_session_state, current_message_index_state, current_classification_state], outputs=[ current_message_display, classifier_decision_display, classifier_confidence_display, session_stats_display, current_message_index_state, current_classification_state, correction_section, dataset_status_message ] ) incorrect_btn.click( on_incorrect_feedback, outputs=[correction_section, dataset_status_message] ) cancel_correction_btn.click( on_cancel_correction, outputs=[correction_section, dataset_status_message] ) submit_correction_btn.click( on_submit_correction, inputs=[ current_dataset_state, verification_session_state, current_message_index_state, current_classification_state, correction_selector, correction_notes ], outputs=[ current_message_display, classifier_decision_display, classifier_confidence_display, session_stats_display, current_message_index_state, current_classification_state, correction_section, dataset_status_message ] ) skip_btn.click( on_skip_message, inputs=[current_dataset_state, verification_session_state, current_message_index_state, current_classification_state], outputs=[ current_message_display, classifier_decision_display, classifier_confidence_display, session_stats_display, current_message_index_state, current_classification_state, dataset_status_message ] ) finish_btn.click( on_finish_session, inputs=[verification_session_state], outputs=[message_review_section, dataset_status_message] ) # Export buttons export_csv_btn.click( lambda session: on_export_results("csv", session), inputs=[verification_session_state], outputs=[dataset_status_message] ) export_json_btn.click( lambda session: on_export_results("json", session), inputs=[verification_session_state], outputs=[dataset_status_message] ) export_xlsx_btn.click( lambda session: on_export_results("xlsx", session), inputs=[verification_session_state], outputs=[dataset_status_message] ) return interface def create_enhanced_verification_tab() -> gr.Blocks: """ Create enhanced verification tab for integration with existing application. Returns: Gradio Blocks component for enhanced verification modes """ interface_controller = EnhancedVerificationInterface() return interface_controller.create_interface()