Spaces:
Sleeping
Sleeping
| """ | |
| Supabase Memory Layer - Learning & Replay System | |
| Stores configurator states, transitions, and patterns for intelligent replay. | |
| """ | |
| import os | |
| import hashlib | |
| import json | |
| from typing import Optional, Dict, List, Any, Tuple | |
| from datetime import datetime | |
| try: | |
| from supabase import create_client, Client | |
| SUPABASE_AVAILABLE = True | |
| except ImportError: | |
| SUPABASE_AVAILABLE = False | |
| Client = None | |
| class SupabaseMemory: | |
| """ | |
| Memory layer for learning configurator patterns and enabling replay. | |
| - Records page states and transitions | |
| - Tracks success/failure confidence | |
| - Enables replay without LLM calls when confidence is high | |
| - Provides cross-run and cross-site learning | |
| """ | |
| def __init__(self, enabled: bool = True): | |
| """ | |
| Initialize Supabase memory client. | |
| Args: | |
| enabled: Whether to use Supabase (auto-disabled if credentials missing) | |
| """ | |
| self.client: Optional[Client] = None | |
| self.enabled = enabled and SUPABASE_AVAILABLE | |
| if not SUPABASE_AVAILABLE: | |
| print("⚠️ Supabase not available (install: pip install supabase)") | |
| self.enabled = False | |
| return | |
| if not enabled: | |
| print("ℹ️ Supabase memory disabled by configuration") | |
| return | |
| # Try to initialize client | |
| try: | |
| supabase_url = os.getenv("SUPABASE_URL") | |
| supabase_key = os.getenv("SUPABASE_ANON_KEY") | |
| if not supabase_url or not supabase_key: | |
| print("⚠️ Supabase credentials not configured (SUPABASE_URL, SUPABASE_ANON_KEY)") | |
| self.enabled = False | |
| return | |
| self.client = create_client(supabase_url, supabase_key) | |
| print("✓ Supabase memory layer initialized") | |
| except Exception as e: | |
| print(f"⚠️ Failed to initialize Supabase: {e}") | |
| self.enabled = False | |
| self.client = None | |
| def is_enabled(self) -> bool: | |
| """Check if Supabase memory is available and enabled.""" | |
| return self.enabled and self.client is not None | |
| # ============================================================ | |
| # STATE MANAGEMENT | |
| # ============================================================ | |
| def upsert_state( | |
| self, | |
| site_domain: str, | |
| url_path: str, | |
| page_signature: str, | |
| model_name: str, | |
| step_index: int, | |
| is_terminal: bool = False | |
| ) -> Optional[Dict[str, Any]]: | |
| """ | |
| Record or update a page state. | |
| Args: | |
| site_domain: Domain of the site (e.g., "example.com") | |
| url_path: URL path (e.g., "/configure") | |
| page_signature: Computed hash of page structure | |
| model_name: Product model/variant name | |
| step_index: Step number in configuration flow | |
| is_terminal: Whether this is a final state | |
| Returns: | |
| State record with ID, or None if failed | |
| """ | |
| if not self.is_enabled(): | |
| return None | |
| try: | |
| result = self.client.table("states").upsert({ | |
| "site_domain": site_domain, | |
| "url_path": url_path, | |
| "page_signature": page_signature, | |
| "model_name": model_name, | |
| "step_index": step_index, | |
| "is_terminal": is_terminal, | |
| "created_at": datetime.now().isoformat() | |
| }, on_conflict="site_domain,page_signature").execute() | |
| return result.data[0] if result.data else None | |
| except Exception as e: | |
| print(f"⚠️ Failed to upsert state: {e}") | |
| return None | |
| def get_state_by_signature( | |
| self, | |
| site_domain: str, | |
| page_signature: str | |
| ) -> Optional[Dict[str, Any]]: | |
| """Get existing state by signature.""" | |
| if not self.is_enabled(): | |
| return None | |
| try: | |
| result = self.client.table("states").select("*").eq( | |
| "site_domain", site_domain | |
| ).eq( | |
| "page_signature", page_signature | |
| ).execute() | |
| return result.data[0] if result.data else None | |
| except Exception as e: | |
| print(f"⚠️ Failed to get state: {e}") | |
| return None | |
| # ============================================================ | |
| # TRANSITION TRACKING | |
| # ============================================================ | |
| def record_transition( | |
| self, | |
| from_state_id: str, | |
| to_state_id: str, | |
| action_type: str, | |
| selector_pattern: Dict[str, Any], | |
| success: bool = True | |
| ) -> bool: | |
| """ | |
| Record a transition between states. | |
| Args: | |
| from_state_id: UUID of source state | |
| to_state_id: UUID of destination state | |
| action_type: Type of action (e.g., "click", "select") | |
| selector_pattern: Pattern used to find element (for replay) | |
| success: Whether the transition succeeded | |
| Returns: | |
| True if recorded successfully | |
| """ | |
| if not self.is_enabled(): | |
| return False | |
| try: | |
| # Check if transition exists | |
| existing = self.client.table("transitions").select("*").eq( | |
| "from_state", from_state_id | |
| ).eq( | |
| "to_state", to_state_id | |
| ).eq( | |
| "action_type", action_type | |
| ).execute() | |
| if existing.data: | |
| # Update existing transition | |
| transition_id = existing.data[0]["id"] | |
| field = "success_count" if success else "failure_count" | |
| current_value = existing.data[0][field] | |
| self.client.table("transitions").update({ | |
| field: current_value + 1, | |
| "last_used": datetime.now().isoformat() | |
| }).eq("id", transition_id).execute() | |
| else: | |
| # Create new transition | |
| self.client.table("transitions").insert({ | |
| "from_state": from_state_id, | |
| "to_state": to_state_id, | |
| "action_type": action_type, | |
| "selector_pattern": json.dumps(selector_pattern), | |
| "success_count": 1 if success else 0, | |
| "failure_count": 0 if success else 1, | |
| "last_used": datetime.now().isoformat() | |
| }).execute() | |
| return True | |
| except Exception as e: | |
| print(f"⚠️ Failed to record transition: {e}") | |
| return False | |
| def get_confident_transition( | |
| self, | |
| from_state_id: str, | |
| min_success_rate: float = 0.8, | |
| min_seen_count: int = 3 | |
| ) -> Optional[Dict[str, Any]]: | |
| """ | |
| Get a high-confidence transition for replay (avoids LLM call). | |
| Args: | |
| from_state_id: Source state UUID | |
| min_success_rate: Minimum success rate (default: 80%) | |
| min_seen_count: Minimum times seen (default: 3) | |
| Returns: | |
| Transition data if confident, None otherwise | |
| """ | |
| if not self.is_enabled(): | |
| return None | |
| try: | |
| # Get all transitions from this state | |
| result = self.client.table("transitions").select("*").eq( | |
| "from_state", from_state_id | |
| ).execute() | |
| if not result.data: | |
| return None | |
| # Find best confident transition | |
| best_transition = None | |
| best_confidence = 0.0 | |
| for transition in result.data: | |
| success = transition["success_count"] | |
| failure = transition["failure_count"] | |
| total = success + failure | |
| if total < min_seen_count: | |
| continue | |
| success_rate = success / total if total > 0 else 0.0 | |
| if success_rate >= min_success_rate and success_rate > best_confidence: | |
| best_confidence = success_rate | |
| best_transition = transition | |
| return best_transition | |
| except Exception as e: | |
| print(f"⚠️ Failed to get confident transition: {e}") | |
| return None | |
| # ============================================================ | |
| # PATTERN LEARNING | |
| # ============================================================ | |
| def upsert_continue_pattern( | |
| self, | |
| site_domain: str, | |
| pattern: Dict[str, Any], | |
| success: bool = True | |
| ) -> bool: | |
| """ | |
| Record a "continue" button pattern with success tracking. | |
| Args: | |
| site_domain: Domain of the site | |
| pattern: Pattern dict (text, classes, position hints) | |
| success: Whether the pattern led to success | |
| Returns: | |
| True if recorded successfully | |
| """ | |
| if not self.is_enabled(): | |
| return False | |
| try: | |
| pattern_json = json.dumps(pattern) | |
| # Check if pattern exists | |
| existing = self.client.table("continue_patterns").select("*").eq( | |
| "site_domain", site_domain | |
| ).eq( | |
| "pattern", pattern_json | |
| ).execute() | |
| if existing.data: | |
| # Update existing | |
| record = existing.data[0] | |
| seen_count = record["seen_count"] + 1 | |
| success_count = record.get("success_count", 0) + (1 if success else 0) | |
| success_rate = success_count / seen_count if seen_count > 0 else 0.0 | |
| self.client.table("continue_patterns").update({ | |
| "success_rate": success_rate, | |
| "seen_count": seen_count, | |
| "success_count": success_count | |
| }).eq("id", record["id"]).execute() | |
| else: | |
| # Create new | |
| self.client.table("continue_patterns").insert({ | |
| "site_domain": site_domain, | |
| "pattern": pattern_json, | |
| "success_rate": 1.0 if success else 0.0, | |
| "seen_count": 1, | |
| "success_count": 1 if success else 0 | |
| }).execute() | |
| return True | |
| except Exception as e: | |
| print(f"⚠️ Failed to upsert pattern: {e}") | |
| return False | |
| def get_best_continue_pattern( | |
| self, | |
| site_domain: str, | |
| min_success_rate: float = 0.8, | |
| min_seen_count: int = 3 | |
| ) -> Optional[Dict[str, Any]]: | |
| """ | |
| Get the most reliable "continue" button pattern for a site. | |
| Args: | |
| site_domain: Domain of the site | |
| min_success_rate: Minimum success rate | |
| min_seen_count: Minimum times seen | |
| Returns: | |
| Pattern dict or None | |
| """ | |
| if not self.is_enabled(): | |
| return None | |
| try: | |
| result = self.client.table("continue_patterns").select("*").eq( | |
| "site_domain", site_domain | |
| ).gte("success_rate", min_success_rate).gte( | |
| "seen_count", min_seen_count | |
| ).order("success_rate", desc=True).limit(1).execute() | |
| if result.data: | |
| pattern_str = result.data[0]["pattern"] | |
| return json.loads(pattern_str) | |
| return None | |
| except Exception as e: | |
| print(f"⚠️ Failed to get continue pattern: {e}") | |
| return None | |
| # ============================================================ | |
| # UTILITIES | |
| # ============================================================ | |
| def get_stats(self, site_domain: Optional[str] = None) -> Dict[str, Any]: | |
| """ | |
| Get memory statistics. | |
| Args: | |
| site_domain: Optional domain filter | |
| Returns: | |
| Stats dict with counts and metrics | |
| """ | |
| if not self.is_enabled(): | |
| return {"enabled": False} | |
| try: | |
| stats = {"enabled": True} | |
| # State count | |
| if site_domain: | |
| states = self.client.table("states").select("id", count="exact").eq( | |
| "site_domain", site_domain | |
| ).execute() | |
| else: | |
| states = self.client.table("states").select("id", count="exact").execute() | |
| stats["states"] = states.count if hasattr(states, 'count') else len(states.data) | |
| # Transition count | |
| if site_domain: | |
| transitions = self.client.table("transitions").select("id", count="exact").execute() | |
| # Filter would require joining with states - skip for simplicity | |
| stats["transitions"] = transitions.count if hasattr(transitions, 'count') else len(transitions.data) | |
| else: | |
| transitions = self.client.table("transitions").select("id", count="exact").execute() | |
| stats["transitions"] = transitions.count if hasattr(transitions, 'count') else len(transitions.data) | |
| # Pattern count | |
| if site_domain: | |
| patterns = self.client.table("continue_patterns").select("id", count="exact").eq( | |
| "site_domain", site_domain | |
| ).execute() | |
| else: | |
| patterns = self.client.table("continue_patterns").select("id", count="exact").execute() | |
| stats["patterns"] = patterns.count if hasattr(patterns, 'count') else len(patterns.data) | |
| return stats | |
| except Exception as e: | |
| print(f"⚠️ Failed to get stats: {e}") | |
| return {"enabled": True, "error": str(e)} | |
| # ============================================================ | |
| # UTILITY FUNCTIONS | |
| # ============================================================ | |
| def compute_page_signature(page_state: Dict[str, Any]) -> str: | |
| """ | |
| Compute a stable signature for a page state. | |
| Uses element structure (not content) to identify similar pages. | |
| Args: | |
| page_state: Page state dict with 'elements' list | |
| Returns: | |
| Hex hash string (signature) | |
| """ | |
| sig_parts = [] | |
| elements = page_state.get("elements", []) | |
| for elem in elements[:50]: # Limit to first 50 elements for performance | |
| # Use tag, classes, and approximate position | |
| tag = elem.get("tag", "") | |
| classes = elem.get("classes", "")[:30] # Truncate long class strings | |
| position = elem.get("position", {}) | |
| y_bucket = position.get("y", 0) // 100 # Group by 100px vertical bands | |
| sig_parts.append(f"{tag}|{classes}|{y_bucket}") | |
| # Sort for stability (order may vary slightly) | |
| sig_parts.sort() | |
| # Hash the signature | |
| signature_str = "|".join(sig_parts) | |
| return hashlib.sha256(signature_str.encode()).hexdigest() | |
| def extract_site_domain(url: str) -> str: | |
| """ | |
| Extract domain from URL. | |
| Args: | |
| url: Full URL | |
| Returns: | |
| Domain string (e.g., "example.com") | |
| """ | |
| from urllib.parse import urlparse | |
| parsed = urlparse(url) | |
| return parsed.netloc.replace("www.", "") | |
| def create_selector_pattern(element: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Create a reusable selector pattern from an element. | |
| Args: | |
| element: Element dict from page state | |
| Returns: | |
| Pattern dict for replay | |
| """ | |
| return { | |
| "tag": element.get("tag"), | |
| "classes": element.get("classes", "")[:50], | |
| "text": element.get("text", "")[:50], | |
| "attributes": { | |
| k: v for k, v in element.get("attributes", {}).items() | |
| if k in ["id", "name", "data-*", "aria-label"] | |
| } | |
| } | |