Spaces:
Sleeping
Sleeping
| """ | |
| 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) |