Update app.py

app.py

@@ -1,77 +1,129 @@
+"""
+WOD Analyzer - Clean and Refactored Version
+A Gradio application for analyzing Work Order Documents with improved code structure.
+"""
+
 import os
 import json
+import time
+from typing import Tuple, Dict, Any, Optional, List
+from dataclasses import dataclass
+from enum import Enum
+
 import gradio as gr
 import pandas as pd
 from python_request import process_wod_document
-
-import time
 from dummy import output_test
 
-# --- Authentication Function ---
-def authenticate_user(username, password):
-    """
-    Simple authentication function.
-    In production, you should use more secure methods like hashed passwords.
-    """
-    return username == "demo" and password == os.environ["PASSWORD"]
+PRODUCTION = True
 
-# --- Core Application Logic ---
-def analyze_wod(file_obj, wod_type):
-    """
-    This function analyzes a Work Order Document using the remote API.
+# === CONSTANTS ===
+class WODType(Enum):
+    """Enum for Work Order Document types."""
+    REPLACEMENT = "REPLACEMENT"
+    THERMAL = "THERMAL"
+    VISIT = "VISIT"
+    PREVENTIVE_MAINTENANCE = "PREVENTIVE_MAINTENANCE"
+    INSTALLATION = "INSTALLATION"
+    WITHDRAWAL = "WITHDRAWAL"
 
-    Args:
-        file_obj: The uploaded file object from Gradio.
-        wod_type: The selected type of Work Order Document.
 
-    Returns:
-        A tuple containing: prediction_display, dataframe, and button_update
-    """
-    # Check if user has selected a valid WOD type
-    if wod_type == "-- WOD type --" or wod_type is None:
-        # Show warning dialog and return empty DataFrame
-        gr.Warning("Please select a WOD type first!")
-        return "", pd.DataFrame(), gr.update(value="Analyze Document", interactive=True)
-    
-    # Check if file is uploaded
-    if file_obj is None:
-        gr.Warning("Please upload a PDF file first!")
-        return "", pd.DataFrame(), gr.update(value="Analyze Document", interactive=True)
-    
-    print(f"Analyzing '{file_obj.name}' (Type: {wod_type})...")
-    
-    try:
-        # In modern Gradio versions, file_obj is already a path string
-        # We can use it directly or get the path from it
-        if hasattr(file_obj, 'name') and os.path.isfile(file_obj.name):
-            # file_obj has a .name attribute pointing to the temporary file
-            temp_file_path = file_obj.name
-            cleanup_needed = False
-        else:
-            # Fallback: assume file_obj is a path string
-            temp_file_path = str(file_obj)
-            cleanup_needed = False
-        
-        # Process the document using the API
-        api_response = process_wod_document(temp_file_path, wod_type)
-        # time.sleep(5)
-        # api_response = json.loads(output_test) 
+class UIConstants:
+    """UI-related constants."""
+    DEFAULT_WOD_TYPE = "-- WOD type --"
+    BUTTON_ANALYZE = "Analyze Document"
+    BUTTON_PROCESSING = "Processing..."
+    BUTTON_RESET = "Reset"
+    
+    # Styling
+    MAX_WIDTH = "960px"
+    TABLE_MAX_HEIGHT = 1250
+    COLUMN_WIDTHS = [30, 60, 10]
+    
+    # Messages
+    TITLE = "# WOD Analyzer"
+    DESCRIPTION = "Upload a Work Order Document to automatically check for requirements."
+    NO_WOD_TYPE_WARNING = "Please select a WOD type first!"
+    NO_FILE_WARNING = "Please upload a PDF file first!"
+
+
+class Config:
+    """Application configuration."""
+    USERNAME = "demo"
+    PASSWORD_ENV_VAR = os.environ["PASSWORD"]
+    DEBUG = True
+    MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB
+
+
+@dataclass
+class AnalysisResult:
+    """Data class for analysis results."""
+    prediction: str
+    dataframe: pd.DataFrame
+    json_data: Dict[str, Any]
+    accordion_visible: bool
+    success: bool
+    error_message: Optional[str] = None
+
+
+# === CORE BUSINESS LOGIC ===
+class WODAnalyzer:
+    """Core business logic for WOD analysis."""
+    
+    @staticmethod
+    def validate_inputs(file_obj: Optional[Any], wod_type: str) -> Tuple[bool, str]:
+        """Validate user inputs."""
+        if wod_type == UIConstants.DEFAULT_WOD_TYPE or wod_type is None:
+            return False, UIConstants.NO_WOD_TYPE_WARNING
         
-        # Clean up temporary file if we created one
-        if cleanup_needed:
-            os.unlink(temp_file_path)
+        if file_obj is None:
+            return False, UIConstants.NO_FILE_WARNING
         
-        # Check if API call was successful
+        return True, ""
+    
+    @staticmethod
+    def get_file_path(file_obj: Any) -> str:
+        """Extract file path from Gradio file object."""
+        if hasattr(file_obj, 'name') and os.path.isfile(file_obj.name):
+            return file_obj.name
+        return str(file_obj)
+    
+    @staticmethod
+    def process_api_response(api_response: Dict[str, Any]) -> AnalysisResult:
+        """Process API response and convert to AnalysisResult."""
         if api_response.get("status") != "success":
             error_msg = api_response.get("message", "Unknown error occurred")
-            gr.Error(f"API Error: {error_msg}")
-            return "", pd.DataFrame(), gr.update(value="Analyze Document", interactive=True)
+            return AnalysisResult(
+                prediction="",
+                dataframe=pd.DataFrame(),
+                json_data={},
+                accordion_visible=False,
+                success=False,
+                error_message=f"API Error: {error_msg}"
+            )
         
         # Parse the API response
         results = api_response.get("results", {})
         summary = results.get("summary", {})
+        extracted_data = api_response.get("extracted_data", {})
+        
+        # Convert summary to DataFrame
+        df = WODAnalyzer._create_summary_dataframe(summary)
         
-        # Convert API response to DataFrame format
+        # Get prediction
+        prediction = results.get("prediction", "Unknown")
+        
+        return AnalysisResult(
+            prediction=prediction,
+            dataframe=df,
+            json_data=extracted_data,
+            accordion_visible=bool(extracted_data),
+            success=True
+        )
+    
+    @staticmethod
+    def _create_summary_dataframe(summary: Dict[str, Any]) -> pd.DataFrame:
+        """Create DataFrame from summary data."""
         requirements = []
         reasons = []
         statuses = []
@@ -79,6 +131,7 @@ def analyze_wod(file_obj, wod_type):
         for requirement_name, details in summary.items():
             requirements.append(requirement_name)
             reasons.append(details.get("reasoning", ""))
+            
             # Convert true/false to PASS/FAIL
             status_bool = details.get("status", "false")
             if isinstance(status_bool, str):
@@ -87,148 +140,285 @@ def analyze_wod(file_obj, wod_type):
                 status = "PASS" if status_bool else "FAIL"
             statuses.append(status)
         
-        # Create DataFrame
-        df = pd.DataFrame({
+        return pd.DataFrame({
             "Requirement": requirements,
             "Reason": reasons,
             "Status": statuses
         })
+    
+    @classmethod
+    def analyze_document(cls, file_obj: Any, wod_type: str) -> AnalysisResult:
+        """Main analysis function."""
+        # Validate inputs
+        is_valid, error_msg = cls.validate_inputs(file_obj, wod_type)
+        if not is_valid:
+            return AnalysisResult(
+                prediction="",
+                dataframe=pd.DataFrame(),
+                json_data={},
+                accordion_visible=False,
+                success=False,
+                error_message=error_msg
+            )
         
-        # Get prediction for display
-        prediction = results.get("prediction", "Unknown")
-        gr.Info(f"Analysis completed! Overall prediction: {prediction}")
+        try:
+            print(f"Analyzing '{file_obj.name}' (Type: {wod_type})...")
+            
+            # Get file path
+            file_path = cls.get_file_path(file_obj)
+            
+            # Process the document using the API (currently mocked)
+            #
+
+            if PRODUCTION:
+                 api_response = process_wod_document(file_path, wod_type)
+            else:
+                time.sleep(1)  # Simulate processing time
+                api_response = json.loads(output_test)
+            
+            # Process the response
+            return cls.process_api_response(api_response)
+            
+        except Exception as e:
+            error_msg = f"Error processing document: {str(e)}"
+            print(error_msg)
+            return AnalysisResult(
+                prediction="",
+                dataframe=pd.DataFrame(),
+                json_data={},
+                accordion_visible=False,
+                success=False,
+                error_message=error_msg
+            )
+
+
+# === AUTHENTICATION ===
+class AuthManager:
+    """Handle user authentication."""
+    
+    @staticmethod
+    def authenticate_user(username: str, password: str) -> bool:
+        """
+        Simple authentication function.
+        In production, use more secure methods like hashed passwords.
+        """
+        expected_password = Config.PASSWORD_ENV_VAR
+        if not expected_password:
+            print("Warning: PASSWORD environment variable not set")
+            return False
+        
+        return username == Config.USERNAME and password == expected_password
+
+
+# === UI COMPONENTS ===
+class UIBuilder:
+    """Builds and manages UI components."""
+    
+    @staticmethod
+    def get_wod_type_options() -> List[str]:
+        """Get WOD type dropdown options."""
+        return [UIConstants.DEFAULT_WOD_TYPE] + [wod_type.value for wod_type in WODType]
+    
+    @staticmethod
+    def create_custom_css() -> str:
+        """Create custom CSS for the application."""
+        return f"""
+        .gradio-container {{
+            max-width: {UIConstants.MAX_WIDTH} !important;
+            margin: auto !important;
+        }}
+        .progress-text {{
+            display: none !important;
+        }}
+        """
+    
+    @staticmethod
+    def create_theme() -> gr.Theme:
+        """Create custom theme for the application."""
+        return gr.themes.Default(primary_hue="blue", secondary_hue="sky")
+    
+    @staticmethod
+    def format_prediction_display(prediction: str) -> str:
+        """Format prediction for display."""
+        if prediction and prediction != "Unknown":
+            return f"<h1 style='text-align: center; color: #1f77b4;'>{prediction}</h1>"
+        return ""
+
+
+# === EVENT HANDLERS ===
+class EventHandlers:
+    """Handle UI events and interactions."""
+    
+    @staticmethod
+    def handle_analyze_button(
+        file_obj: Any, 
+        wod_type: str, 
+        current_button_value: str
+    ) -> Tuple[str, pd.DataFrame, Dict[str, Any], gr.update, gr.update]:
+        """Handle analyze button click."""
+        # Check if this is a reset action
+        if current_button_value == UIConstants.BUTTON_RESET:
+            return (
+                "",
+                pd.DataFrame(),
+                {},
+                gr.update(visible=False),
+                gr.update(value=UIConstants.BUTTON_ANALYZE, interactive=True)
+            )
+        
+        # Perform analysis
+        result = WODAnalyzer.analyze_document(file_obj, wod_type)
         
-        # Format prediction as centered H1 for display
-        prediction_display = f"<h1 style='text-align: center; color: #1f77b4;'>{prediction}</h1>" if prediction != "Unknown" else ""
+        if not result.success:
+            if result.error_message:
+                if "Please select" in result.error_message or "Please upload" in result.error_message:
+                    gr.Warning(result.error_message)
+                else:
+                    gr.Error(result.error_message)
+            
+            return (
+                "",
+                pd.DataFrame(),
+                {},
+                gr.update(visible=False),
+                gr.update(value=UIConstants.BUTTON_ANALYZE, interactive=True)
+            )
         
-        # Return with Reset button
-        return prediction_display, df, gr.update(value="Reset", interactive=True)
+        # Success case
+        gr.Info(f"Analysis completed! Overall prediction: {result.prediction}")
+        prediction_display = UIBuilder.format_prediction_display(result.prediction)
         
-    except Exception as e:
-        error_msg = f"Error processing document: {str(e)}"
-        print(error_msg)
-        gr.Error(error_msg)
-        return "", pd.DataFrame(), gr.update(value="Analyze Document", interactive=True)
+        return (
+            prediction_display,
+            result.dataframe,
+            result.json_data,
+            gr.update(visible=result.accordion_visible),
+            gr.update(value=UIConstants.BUTTON_RESET, interactive=True)
+        )
+    
+    @staticmethod
+    def set_processing_state(current_button_value: str) -> gr.update:
+        """Set button to processing state."""
+        if current_button_value == UIConstants.BUTTON_RESET:
+            return gr.update()  # Don't change if it's reset
+        return gr.update(value=UIConstants.BUTTON_PROCESSING, interactive=False)
+
 
-def reset_app():
+# === JAVASCRIPT FUNCTIONS ===
+class JavaScriptFunctions:
+    """JavaScript functions for the UI."""
+    
+    THEME_SETUP = """
+    function refresh() {
+        const url = new URL(window.location);
+        if (url.searchParams.get('__theme') !== 'light') {
+            url.searchParams.set('__theme', 'light');
+            window.location.href = url.href;
+        }
+    }
     """
-    Reset function that refreshes the page using JavaScript
+    
+    REFRESH_ON_RESET = """
+    function(button_value) {
+        if (button_value === "Reset") {
+            window.location.reload();
+            return false;
+        }
+        return true;
+    }
     """
-    return gr.update(value="Analyze Document", interactive=True)
 
-js_func = """
-function refresh() {
-    const url = new URL(window.location);
 
-    if (url.searchParams.get('__theme') !== 'light') {
-        url.searchParams.set('__theme', 'light');
-        window.location.href = url.href;
-    }
-}
-"""
-
-# JavaScript function to refresh the page when Reset is clicked
-refresh_js = """
-function(button_value) {
-    if (button_value === "Reset") {
-        window.location.reload();
-        return false; // Prevent default action
-    }
-    return true;
-}
-"""
+# === MAIN APPLICATION ===
+class WODAnalyzerApp:
+    """Main application class."""
+    
+    def __init__(self):
+        self.ui_builder = UIBuilder()
+        self.event_handlers = EventHandlers()
+    
+    def create_interface(self) -> gr.Blocks:
+        """Create the Gradio interface."""
+        with gr.Blocks(
+            theme=self.ui_builder.create_theme(),
+            js=JavaScriptFunctions.THEME_SETUP,
+            css=self.ui_builder.create_custom_css()
+        ) as demo:
+            
+            # Header
+            gr.Markdown(f"{UIConstants.TITLE}\n{UIConstants.DESCRIPTION}")
+            
+            # Input Section
+            with gr.Row():
+                file_input = gr.File(label="Upload WOD PDF")
+                type_input = gr.Dropdown(
+                    choices=self.ui_builder.get_wod_type_options(),
+                    label="Type",
+                    value=UIConstants.DEFAULT_WOD_TYPE,
+                    info="Select the type of work order."
+                )
+            
+            # Action Button
+            analyze_btn = gr.Button(UIConstants.BUTTON_ANALYZE, variant="primary")
+            
+            # Results Section
+            gr.Markdown("---\n## Results")
+            
+            # Prediction display
+            prediction_output = gr.Markdown(value="", visible=True)
+            
+            # JSON display for extracted data
+            with gr.Accordion("Extraction Result from Page 1", open=False, visible=False) as json_accordion:
+                json_output = gr.JSON(label="Extracted Data", show_label=False, open=True)
+            
+            # Results table
+            results_output = gr.DataFrame(
+                headers=["Requirement", "Reason", "Status"],
+                datatype=["str", "str", "str"],
+                interactive=False,
+                max_height=UIConstants.TABLE_MAX_HEIGHT,
+                column_widths=UIConstants.COLUMN_WIDTHS,
+                wrap=True
+            )
+            
+            # Event handling
+            analyze_btn.click(
+                fn=self.event_handlers.set_processing_state,
+                inputs=[analyze_btn],
+                outputs=[analyze_btn],
+                show_progress=False,
+                js=JavaScriptFunctions.REFRESH_ON_RESET
+            ).then(
+                fn=self.event_handlers.handle_analyze_button,
+                inputs=[file_input, type_input, analyze_btn],
+                outputs=[prediction_output, results_output, json_output, json_accordion, analyze_btn],
+                show_progress=True
+            )
+        
+        return demo
+    
+    def launch(self, enable_auth: bool = False) -> None:
+        """Launch the application."""
+        demo = self.create_interface()
+        
+        if enable_auth:
+            demo.launch(
+                auth=AuthManager.authenticate_user,
+                auth_message="Please enter your credentials to access the WOD Analyzer",
+                debug=Config.DEBUG,
+                ssr_mode=False
+            )
+        else:
+            demo.launch(debug=Config.DEBUG)
 
-# --- Gradio User Interface Definition ---
-# Using gr.Blocks() for a custom layout that matches the elegant design.
-with gr.Blocks(
-    #theme=gr.themes.Soft(primary_hue="blue", secondary_hue="sky"), 
-    theme=gr.themes.Default(primary_hue="blue", secondary_hue="sky"), 
-    js=js_func,
-    css=".gradio-container {max-width: 960px !important; margin: auto !important;} .progress-text { display: none !important; }"
-) as demo:
-    
-    # Main Title and Description
-    gr.Markdown(
-        """
-        # WOD Analyzer
-        Upload a Work Order Document to automatically check for requirements.
-        """
-    )
-
-    # Input Section
-    with gr.Row():
-        # File Upload Component
-        file_input = gr.File(label="Upload WOD PDF")
-        
-        # Dropdown for WOD Type
-        type_input = gr.Dropdown(
-            ["-- WOD type --", "REPLACEMENT", "THERMAL", "VISIT", "PREVENTIVE_MAINTENANCE", "INSTALLATION", "WITHDRAWAL"],
-            label="Type",
-            value="-- WOD type --",
-            info="Select the type of work order."
-        )
 
-    # Action Button
-    analyze_btn = gr.Button("Analyze Document", variant="primary")
-
-    # Results Section
-    gr.Markdown("---")
-    gr.Markdown("## Results")
-    
-    # Prediction display (centered H1)
-    prediction_output = gr.Markdown(value="", visible=True)
-    
-    # DataFrame to display the output, with styling for the 'Status' column
-    results_output = gr.DataFrame(
-        headers=["Requirement", "Reason", "Status"],
-        datatype=["str", "str", "str"],
-        interactive=False,
-        max_height=1250,
-        column_widths=[30, 60, 10],
-        wrap=True
-    )
-
-    # Define the interaction: clicking the button calls the function
-    def handle_button_click(file_obj, wod_type, current_button_value):
-        """Handle button click with state management"""
-        # Check if this is a reset action (this shouldn't be reached due to JS)
-        if current_button_value == "Reset":
-            return "", pd.DataFrame(), gr.update(value="Analyze Document", interactive=True)
-        
-        # Otherwise, this is an analyze action  
-        prediction_display, df, button_update = analyze_wod(file_obj, wod_type)
-        return prediction_display, df, button_update
-    
-    # Set button to Processing state when clicked (only if not Reset)
-    def set_processing_state(current_button_value):
-        if current_button_value == "Reset":
-            return gr.update()  # Don't change if it's reset
-        return gr.update(value="Processing...", interactive=False)
-    
-    # Chain the events: first set processing state, then analyze
-    analyze_btn.click(
-        fn=set_processing_state,
-        inputs=[analyze_btn],
-        outputs=[analyze_btn],
-        show_progress=False,
-        js=refresh_js  # Add JavaScript to handle reset
-    ).then(
-        fn=handle_button_click,
-        inputs=[file_input, type_input, analyze_btn],
-        outputs=[prediction_output, results_output, analyze_btn],
-        show_progress=True
-    )
-
-# --- Launch the Application with Authentication ---
-if __name__ == "__main__":
-    # The launch() command creates a web server with authentication enabled
-    # Users must provide the correct username and password to access the app
+# === MAIN ENTRY POINT ===
+def main():
+    """Main entry point."""
+    app = WODAnalyzerApp()
+    
+    app.launch(enable_auth=PRODUCTION)
 
-    # demo.launch(debug=True)
 
-    demo.launch(
-        auth=authenticate_user,  # Enable authentication
-        auth_message="Please enter your credentials to access the WOD Analyzer",
-        share=True,
-        ssr_mode=False,
-    )
+if __name__ == "__main__":
+    main()