app.py

"""
Hugging Face Spaces Entry Point
Improved UI with better logging, status indicators, and results display.
"""

import gradio as gr
import os
import sys
from datetime import datetime
from workflow import create_workflow, run_workflow_step_1, resume_workflow

# Add current directory to path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

# Log file path
LOG_FILE = "app.log"

def clear_logs():
    """Clear the log file for a fresh start."""
    with open(LOG_FILE, "w") as f:
        f.write(f"===== New Session Started at {datetime.now()} =====\n")

def start_test(figma_key, figma_id, url, hf_token):
    """Start the test and stop at the breakpoint."""
    # Clear logs at start of new test
    clear_logs()
    
    print(f"\n{'='*60}")
    print(f"🚀 STARTING NEW TEST")
    print(f"{'='*60}")
    print(f"⏰ Time: {datetime.now()}")
    print(f"🌐 Website URL: {url}")
    print(f"📁 Figma File ID: {figma_id}")
    print(f"🔑 Figma API Key: {'✅ Provided' if figma_key else '❌ Missing'}")
    print(f"🤗 HF Token: {'✅ Provided - AI Analysis will be enhanced' if hf_token else '⚠️ Not provided - Using basic comparison only'}")
    print(f"{'='*60}\n")
    
    execution_id = f"exec_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    thread_id = f"thread_{execution_id}"
    
    try:
        print(f"🧵 Thread ID: {thread_id}")
        print(f"\n📋 STEP 1: Capturing Screenshots...")
        print(f"-" * 40)
        
        state_snapshot = run_workflow_step_1(figma_id, figma_key, url, execution_id, thread_id, hf_token)
        
        if not state_snapshot or not state_snapshot.values:
            print("❌ ERROR: Workflow returned empty state.")
            return "❌ Error: Workflow failed to start or returned empty state.", None, get_current_logs()

        # Get captured screenshots info
        figma_screenshots = state_snapshot.values.get("figma_screenshots", {})
        website_screenshots = state_snapshot.values.get("website_screenshots", {})
        
        print(f"\n{'='*60}")
        print(f"✅ STEP 1 COMPLETE!")
        print(f"{'='*60}")
        print(f"📸 Figma Screenshots: {len(figma_screenshots)}")
        for viewport, path in figma_screenshots.items():
            print(f"   • {viewport}: {path}")
        print(f"📸 Website Screenshots: {len(website_screenshots)}")
        for viewport, path in website_screenshots.items():
            print(f"   • {viewport}: {path}")
        print(f"\n👉 Click 'Step 2: Run AI Analysis' to compare images")
        print(f"{'='*60}\n")
        
        status = f"✅ Step 1 Complete!\n"
        status += f"Thread ID: {thread_id}\n\n"
        status += f"📸 Screenshots Captured:\n"
        status += f"  • Figma: {len(figma_screenshots)} images\n"
        status += f"  • Website: {len(website_screenshots)} images\n\n"
        status += f"🤗 HF Token: {'Active' if hf_token else 'Not provided'}\n\n"
        status += f"👉 Click 'Step 2' button to run AI analysis"
        
        return status, thread_id, get_current_logs()
        
    except Exception as e:
        import traceback
        error_msg = f"❌ Error: {str(e)}"
        print(f"\n{'='*60}")
        print(f"❌ ERROR OCCURRED")
        print(f"{'='*60}")
        print(error_msg)
        traceback.print_exc()
        print(f"{'='*60}\n")
        return error_msg, None, get_current_logs()


def continue_test(thread_id, hf_token):
    """Resume the test after human approval."""
    if not thread_id:
        print("❌ ERROR: No active thread found. Please run Step 1 first.")
        return "❌ No active thread found. Please run Step 1 first.", get_current_logs()
    
    print(f"\n{'='*60}")
    print(f"🔍 STEP 2: RUNNING AI ANALYSIS")
    print(f"{'='*60}")
    print(f"⏰ Time: {datetime.now()}")
    print(f"🧵 Thread ID: {thread_id}")
    print(f"🤗 HF Token: {'✅ Using enhanced AI analysis' if hf_token else '⚠️ Using basic comparison'}")
    print(f"-" * 40)
    
    try:
        print("🔄 Resuming workflow...")
        state_snapshot = resume_workflow(thread_id, user_approval=True)
        
        if not state_snapshot:
            print("❌ ERROR: Workflow failed to resume.")
            return "❌ Workflow failed to resume. Try running Step 1 again.", get_current_logs()
        
        # Extract results
        score = state_snapshot.values.get("similarity_score", 0)
        differences = state_snapshot.values.get("visual_differences", [])
        hf_analysis = state_snapshot.values.get("hf_analysis", {})
        status = state_snapshot.values.get("status", "unknown")
        error_msg = state_snapshot.values.get("error_message", "")
        
        print(f"\n{'='*60}")
        print(f"📊 ANALYSIS RESULTS")
        print(f"{'='*60}")
        print(f"📈 Similarity Score: {score}/100")
        print(f"🔍 Differences Found: {len(differences)}")
        print(f"📋 Status: {status}")
        
        if error_msg:
            print(f"⚠️ Error: {error_msg}")
        
        # Show difference details
        if differences:
            print(f"\n📋 DIFFERENCE DETAILS:")
            print(f"-" * 40)
            for i, diff in enumerate(differences, 1):
                if isinstance(diff, dict):
                    print(f"\n  [{i}] {diff.get('title', 'Unknown')}")
                    print(f"      Category: {diff.get('category', 'N/A')}")
                    print(f"      Severity: {diff.get('severity', 'N/A')}")
                    print(f"      Viewport: {diff.get('viewport', 'N/A')}")
                    if diff.get('description'):
                        print(f"      Description: {diff.get('description', '')[:100]}...")
        
        # Show HF Analysis results
        if hf_analysis:
            print(f"\n🤗 HF VISION ANALYSIS:")
            print(f"-" * 40)
            for viewport, analysis in hf_analysis.items():
                print(f"  {viewport.upper()}:")
                if isinstance(analysis, dict):
                    for key, value in analysis.items():
                        print(f"    • {key}: {str(value)[:100]}")
        
        print(f"\n{'='*60}")
        print(f"✅ ANALYSIS COMPLETE!")
        print(f"{'='*60}\n")
        
        # Build result string for UI
        result = f"{'='*50}\n"
        result += f"📊 ANALYSIS RESULTS\n"
        result += f"{'='*50}\n\n"
        result += f"📈 Similarity Score: {score}/100\n"
        result += f"🔍 Differences Found: {len(differences)}\n"
        result += f"📋 Status: {status}\n\n"
        
        if differences:
            result += f"📋 DIFFERENCES:\n"
            result += f"-" * 30 + "\n"
            for i, diff in enumerate(differences[:10], 1):  # Show first 10
                if isinstance(diff, dict):
                    result += f"\n[{i}] {diff.get('title', 'Unknown')}\n"
                    result += f"    Severity: {diff.get('severity', 'N/A')}\n"
                    result += f"    Category: {diff.get('category', 'N/A')}\n"
            if len(differences) > 10:
                result += f"\n... and {len(differences) - 10} more differences\n"
        
        if hf_analysis:
            result += f"\n🤗 AI INSIGHTS:\n"
            result += f"-" * 30 + "\n"
            for viewport, analysis in hf_analysis.items():
                result += f"\n{viewport.upper()}:\n"
                if isinstance(analysis, dict):
                    for key, value in list(analysis.items())[:3]:
                        result += f"  • {key}: {str(value)[:80]}\n"
        
        return result, get_current_logs()
        
    except Exception as e:
        import traceback
        error_msg = f"❌ Analysis Error: {str(e)}"
        print(f"\n{'='*60}")
        print(f"❌ ERROR IN ANALYSIS")
        print(f"{'='*60}")
        print(error_msg)
        traceback.print_exc()
        print(f"{'='*60}\n")
        return error_msg, get_current_logs()


def get_current_logs():
    """Get current log contents."""
    try:
        with open(LOG_FILE, "r") as f:
            return f.read()
    except:
        return "No logs available yet."


# Gradio UI
def create_interface():
    with gr.Blocks(title="Advanced UI Regression Testing", theme=gr.themes.Soft()) as demo:
        gr.Markdown("# 🚀 Advanced UI Regression Testing (LangGraph Powered)")
        gr.Markdown("Compare Figma designs with live websites to detect visual differences.")
        
        with gr.Row():
            # LEFT COLUMN - Inputs
            with gr.Column(scale=1):
                gr.Markdown("### 📝 Configuration")
                f_key = gr.Textbox(label="Figma API Key", type="password", placeholder="Enter your Figma API key")
                f_id = gr.Textbox(label="Figma File ID", placeholder="e.g., ENieX2p3Gy3TAtxaB36cZA")
                w_url = gr.Textbox(label="Website URL", placeholder="https://your-website.com")
                hf_token = gr.Textbox(label="Hugging Face API Token (Optional)", type="password", 
                                     placeholder="For enhanced AI analysis")
                
                gr.Markdown("### 🎮 Controls")
                start_btn = gr.Button("1️⃣ Start Capture & Setup", variant="primary", size="lg")
                resume_btn = gr.Button("2️⃣ Run AI Analysis", variant="secondary", size="lg")
                
                # Hidden state
                thread_id_state = gr.State()
        
            # RIGHT COLUMN - Results
            with gr.Column(scale=2):
                gr.Markdown("### 📊 Status & Results")
                status_out = gr.Textbox(label="Current Status", lines=8, interactive=False)
                
                gr.Markdown("### 🎯 Final Analysis Results")
                final_out = gr.Textbox(label="Analysis Results", lines=12, interactive=False)
        
        # BOTTOM - Logs
        gr.Markdown("### 📜 Execution Logs")
        gr.Markdown("*Logs refresh automatically every 2 seconds*")
        log_viewer = gr.Code(label="Live Logs", language="python", interactive=False, lines=20)
        refresh_logs = gr.Timer(value=2)
        
        # Clear button for logs
        clear_btn = gr.Button("🗑️ Clear Logs", variant="secondary", size="sm")

        # Event handlers
        start_btn.click(
            start_test, 
            inputs=[f_key, f_id, w_url, hf_token], 
            outputs=[status_out, thread_id_state, log_viewer]
        )
        
        resume_btn.click(
            continue_test,
            inputs=[thread_id_state, hf_token],
            outputs=[final_out, log_viewer]
        )
        
        clear_btn.click(
            lambda: (clear_logs(), "Logs cleared.", "", get_current_logs()),
            outputs=[status_out, final_out, log_viewer]
        )

        refresh_logs.tick(get_current_logs, outputs=[log_viewer])
        
    return demo


if __name__ == "__main__":
    # Setup logging to file
    class Logger(object):
        def __init__(self):
            self.terminal = sys.stdout
            # Clear log file on startup
            with open(LOG_FILE, "w") as f:
                f.write(f"===== Application Startup at {datetime.now()} =====\n\n")
            self.log = open(LOG_FILE, "a")

        def write(self, message):
            self.terminal.write(message)
            self.log.write(message)
            self.log.flush()

        def flush(self):
            self.terminal.flush()
            self.log.flush()

        def isatty(self):
            return hasattr(self.terminal, 'isatty') and self.terminal.isatty()

    sys.stdout = Logger()
    sys.stderr = Logger()
    
    print(f"--- Application Started at {datetime.now()} ---")
    demo = create_interface()
    demo.launch(server_name="0.0.0.0", server_port=7860)