Spaces:
Sleeping
Sleeping
| """ | |
| Agent 3: Integrated Difference Analyzer | |
| Updated for LangGraph dictionary-based state and Store integration. | |
| """ | |
| import os | |
| import sys | |
| from typing import Dict, Any, List | |
| from pathlib import Path | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| from hf_vision_analyzer import create_hf_analyzer | |
| from storage_manager import get_storage_manager | |
| from agents.agent_3_difference_analyzer import ScreenshotComparator | |
| class IntegratedDifferenceAnalyzer: | |
| """ | |
| Integrated analyzer combining multiple detection methods. | |
| """ | |
| def __init__(self, hf_token: str = None): | |
| self.hf_token = hf_token or os.getenv('HUGGINGFACE_API_KEY') | |
| self.hf_analyzer = create_hf_analyzer(self.hf_token, model_type="captioning") | |
| self.storage = get_storage_manager() | |
| self.screenshot_comparator = ScreenshotComparator() | |
| def analyze_differences(self, state: Dict[str, Any]) -> Dict[str, Any]: | |
| """Comprehensive difference analysis.""" | |
| print("\n🔍 Agent 3: Integrated Difference Analysis...") | |
| figma_screenshots = state.get("figma_screenshots", {}) | |
| website_screenshots = state.get("website_screenshots", {}) | |
| execution_id = state.get("execution_id", "unknown") | |
| all_differences = [] | |
| hf_results = {} | |
| try: | |
| for viewport in ["desktop", "mobile"]: | |
| if viewport not in figma_screenshots or viewport not in website_screenshots: | |
| continue | |
| figma_path = figma_screenshots[viewport] | |
| website_path = website_screenshots[viewport] | |
| # 1. Pixel Comparison | |
| # Note: screenshot_comparator should return dicts or be converted to dicts | |
| diffs = self.screenshot_comparator.compare_screenshots(figma_path, website_path, viewport) | |
| # Ensure diffs are serializable dicts | |
| serializable_diffs = [] | |
| for d in diffs: | |
| if hasattr(d, 'to_dict'): | |
| serializable_diffs.append(d.to_dict()) | |
| elif isinstance(d, dict): | |
| serializable_diffs.append(d) | |
| else: | |
| # Fallback conversion | |
| serializable_diffs.append({ | |
| "category": getattr(d, 'category', 'visual'), | |
| "severity": getattr(d, 'severity', 'Medium'), | |
| "title": getattr(d, 'title', 'Difference'), | |
| "description": getattr(d, 'description', ''), | |
| "viewport": viewport, | |
| "location": getattr(d, 'location', None) | |
| }) | |
| all_differences.extend(serializable_diffs) | |
| # 2. HF Vision | |
| if self.hf_analyzer: | |
| comparison = self.hf_analyzer.compare_images(figma_path, website_path) | |
| hf_results[viewport] = comparison | |
| # 3. Storage | |
| from PIL import Image | |
| self.storage.save_screenshot(Image.open(figma_path), execution_id, viewport, "figma") | |
| self.storage.save_screenshot(Image.open(website_path), execution_id, viewport, "website") | |
| # Calculate similarity | |
| high_count = len([d for d in all_differences if d.get('severity') == "High"]) | |
| score = max(0, 100 - (high_count * 10)) | |
| return { | |
| "visual_differences": all_differences, | |
| "similarity_score": score, | |
| "hf_analysis": hf_results, | |
| "status": "analysis_complete" | |
| } | |
| except Exception as e: | |
| print(f" ❌ Analysis failed: {str(e)}") | |
| return {"status": "analysis_failed", "error_message": str(e)} | |
| def agent_3_integrated_node(state: Dict[str, Any]) -> Dict[str, Any]: | |
| """LangGraph node for integrated Agent 3.""" | |
| analyzer = IntegratedDifferenceAnalyzer(hf_token=state.get("hf_token")) | |
| return analyzer.analyze_differences(state) | |