Upload 2 files

app.py

@@ -0,0 +1,140 @@
+"""
+Hugging Face Spaces Entry Point
+Fixed for Gradio 4.12.0 compatibility.
+"""
+
+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__)))
+
+def start_test(figma_key, figma_id, url, hf_token):
+    """Start the test and stop at the breakpoint."""
+    print(f"🚀 Starting test for URL: {url}")
+    execution_id = f"exec_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+    thread_id = f"thread_{execution_id}"
+    
+    try:
+        # Run until breakpoint
+        print(f"  🧵 Thread ID: {thread_id}")
+        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, []
+
+        status = f"✅ Step 1 Complete. Thread ID: {thread_id}\n"
+        status += f"Screenshots captured. Please review them below before proceeding to AI analysis."
+        
+        # Extract screenshots from state
+        screenshots = state_snapshot.values.get("website_screenshots", {})
+        print(f"  📸 Captured {len(screenshots)} screenshots.")
+        
+        return status, thread_id, list(screenshots.values())
+    except Exception as e:
+        import traceback
+        error_msg = f"❌ Error in start_test: {str(e)}"
+        print(error_msg)
+        traceback.print_exc()
+        return error_msg, None, []
+
+def continue_test(thread_id):
+    """Resume the test after human approval."""
+    if not thread_id:
+        return "❌ No active thread found.", None
+    
+    state_snapshot = resume_workflow(thread_id, user_approval=True)
+    
+    if not state_snapshot:
+        return "❌ Workflow failed to resume.", None
+    
+    score = state_snapshot.values.get("similarity_score", 0)
+    diffs = len(state_snapshot.values.get("visual_differences", []))
+    
+    result = f"✅ Analysis Complete!\n"
+    result += f"Similarity Score: {score}/100\n"
+    result += f"Differences Found: {diffs}"
+    
+    return result, None
+
+# Gradio UI
+def create_interface():
+    # For Gradio 4.12.0, theme must be in the Blocks constructor
+    with gr.Blocks(title="Advanced UI Regression Testing", theme=gr.themes.Soft()) as demo:
+        gr.Markdown("# 🚀 Advanced UI Regression Testing (LangGraph Powered)")
+        gr.Markdown("This version uses **LangGraph Persistence**, **Breakpoints**, and **Subgraphs**.")
+        
+        with gr.Row():
+            with gr.Column():
+                f_key = gr.Textbox(label="Figma API Key", type="password")
+                f_id = gr.Textbox(label="Figma File ID")
+                w_url = gr.Textbox(label="Website URL")
+                hf_token = gr.Textbox(label="Hugging Face API Token", type="password")
+                start_btn = gr.Button("1. Start Capture & Setup", variant="primary")
+            
+            with gr.Column():
+                status_out = gr.Textbox(label="Status")
+                thread_id_state = gr.State()
+                gallery = gr.Gallery(label="Captured Screenshots")
+                resume_btn = gr.Button("2. Approve & Run AI Analysis", variant="secondary")
+                final_out = gr.Textbox(label="Final Results")
+                log_viewer = gr.Code(label="Execution Logs", language="python", interactive=False, lines=10)
+                refresh_logs = gr.Timer(value=2)
+
+        def on_start_click():
+            return "⏳ Processing... Please check the logs below for real-time updates.", None, []
+
+        start_btn.click(
+            on_start_click,
+            outputs=[status_out, thread_id_state, gallery]
+        ).then(
+            start_test, 
+            inputs=[f_key, f_id, w_url, hf_token], 
+            outputs=[status_out, thread_id_state, gallery]
+        )
+        
+        resume_btn.click(
+            continue_test,
+            inputs=[thread_id_state],
+            outputs=[final_out, thread_id_state]
+        )
+
+        def update_logs():
+            # This is a simplified way to show stdout in the UI
+            # In a real HF Space, you'd read from a log file
+            try:
+                with open("app.log", "r") as f:
+                    return f.readlines()[-20:] # Show last 20 lines
+            except:
+                return "No logs available yet."
+
+        refresh_logs.tick(update_logs, outputs=[log_viewer])
+    return demo
+
+if __name__ == "__main__":
+    import sys
+    
+    class Logger(object):
+        def __init__(self):
+            self.terminal = sys.stdout
+            self.log = open("app.log", "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()
+
+    sys.stdout = Logger()
+    sys.stderr = Logger()
+    
+    print(f"--- Application Started at {datetime.now()} ---")
+    demo = create_interface()
+    demo.launch()

workflow.py

@@ -0,0 +1,83 @@
+"""
+LangGraph Workflow Orchestration
+Updated for dictionary-based state and persistence.
+"""
+
+from typing import Dict, Any, List
+from langgraph.graph import StateGraph, END
+from langgraph.checkpoint.memory import MemorySaver
+from state_schema import WorkflowState, create_initial_state
+from agents import (
+    agent_0_node,
+    agent_1_node,
+    agent_2_node,
+    agent_3_integrated_node
+)
+
+def create_workflow():
+    """Create the LangGraph workflow."""
+    workflow = StateGraph(WorkflowState)
+    
+    # Add Nodes
+    workflow.add_node("setup", agent_0_node)
+    workflow.add_node("capture_design", agent_1_node)
+    workflow.add_node("capture_website", agent_2_node)
+    workflow.add_node("analysis", agent_3_integrated_node)
+    
+    # Define Edges
+    workflow.set_entry_point("setup")
+    workflow.add_edge("setup", "capture_design")
+    workflow.add_edge("capture_design", "capture_website")
+    
+    # Human-in-the-loop breakpoint before analysis
+    workflow.add_edge("capture_website", "analysis")
+    workflow.add_edge("analysis", END)
+    
+    # Add persistence
+    checkpointer = MemorySaver()
+    
+    return workflow.compile(
+        checkpointer=checkpointer,
+        interrupt_before=["analysis"]
+    )
+
+def run_workflow_step_1(figma_id, figma_key, url, execution_id, thread_id, hf_token=""):
+    """Run the first part of the workflow until the breakpoint."""
+    print(f"  ⚙️  Initializing workflow for thread: {thread_id}")
+    app = create_workflow()
+    config = {"configurable": {"thread_id": thread_id}}
+    
+    initial_state = create_initial_state(
+        figma_file_key=figma_id,
+        figma_access_token=figma_key,
+        website_url=url,
+        hf_token=hf_token,
+        execution_id=execution_id
+    )
+    
+    # Run until interrupt
+    print("  🏃 Running workflow nodes...")
+    try:
+        for event in app.stream(initial_state, config, stream_mode="values"):
+            # Log the current node if possible
+            if event:
+                print(f"    📍 Current state updated")
+    except Exception as e:
+        print(f"  ❌ Error during workflow execution: {str(e)}")
+        raise
+        
+    return app.get_state(config)
+
+def resume_workflow(thread_id, user_approval=True):
+    """Resume the workflow after human approval."""
+    app = create_workflow()
+    config = {"configurable": {"thread_id": thread_id}}
+    
+    # Update state with approval
+    app.update_state(config, {"user_approval": user_approval})
+    
+    # Resume execution
+    for event in app.stream(None, config, stream_mode="values"):
+        state = event
+        
+    return app.get_state(config)