Spaces:
Configuration error
Configuration error
| """ | |
| Knowledge Cache Service — 3-Layer Assets Knowledge Graph Cache. | |
| Layer 1: In-Memory RAM Dict (0ms Instant Return) | |
| Layer 2: Disk File (Assets Knowledge/<tenant_id>/<user_id>/cache.json) (0ms Disk Load into RAM) | |
| Layer 3: Supabase DB Fetch & 100% Real-Time Dynamic Metadata Enrichment (Zero Static Config Files) | |
| Reads all configurations strictly from live DB connections & .env. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import json | |
| import time | |
| from pathlib import Path | |
| from typing import Any, Optional | |
| import httpx | |
| from dotenv import load_dotenv | |
| # Explicitly load root .env file | |
| root_env = Path(__file__).resolve().parents[3] / ".env" | |
| if root_env.exists(): | |
| load_dotenv(dotenv_path=root_env) | |
| else: | |
| load_dotenv() | |
| SUPABASE_URL = os.getenv("NEXT_PUBLIC_SUPABASE_URL", "") | |
| SUPABASE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY") or os.getenv("NEXT_PUBLIC_SUPABASE_ANON_KEY", "") | |
| BASE_KNOWLEDGE_DIR = os.getenv( | |
| "ASSETS_KNOWLEDGE_DIR", | |
| os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "Assets Knowledge") | |
| ) | |
| # Standard Provider Metadata Mapping for Dynamic Enrichment | |
| DEFAULT_PROVIDER_METADATA: dict[str, dict[str, Any]] = { | |
| "shopify": { | |
| "name": "Shopify E-Commerce", | |
| "category": "store", | |
| "synonyms": ["shopify", "store", "shop", "sales", "revenue", "orders", "checkout"], | |
| "attribute_keys": ["totalRevenue", "totalOrders", "avgOrderValue", "checkouts"], | |
| "display_metrics": ["Total Revenue & Orders", "Average Order Value (AOV)", "Checkout Conversion Rate"], | |
| }, | |
| "ga4": { | |
| "name": "Google Analytics 4 (GA4)", | |
| "category": "analytics", | |
| "synonyms": ["ga4", "google analytics", "analytics", "traffic", "visitors", "sessions", "website"], | |
| "attribute_keys": ["activeUsers", "totalSessions", "bounceRate", "trafficSources"], | |
| "display_metrics": ["Active Users & Total Sessions", "Bounce Rate", "Traffic Channels"], | |
| }, | |
| "meta": { | |
| "name": "Meta Ads (Facebook/Instagram)", | |
| "category": "marketing", | |
| "synonyms": ["meta", "facebook", "instagram", "fb ads", "meta ads", "roas", "ctr", "ad spend"], | |
| "attribute_keys": ["spend", "roas", "ctr", "cpc", "impressions", "clicks"], | |
| "display_metrics": ["Ad Spend & Overall ROAS", "Click-Through Rate (CTR) & CPC", "Campaign Performance"], | |
| }, | |
| "google_ads": { | |
| "name": "Google Ads", | |
| "category": "marketing", | |
| "synonyms": ["google ads", "gads", "search ads", "pmax", "google ad spend"], | |
| "attribute_keys": ["totalImpressions", "totalClicks", "totalSpend", "averageCtr", "averageCpc"], | |
| "display_metrics": ["Google Ad Spend & ROAS", "Click-Through Rate (CTR)", "Search Campaign Breakdown"], | |
| }, | |
| "leads": { | |
| "name": "Leads & Contacts", | |
| "category": "crm", | |
| "synonyms": ["leads", "lead", "contacts", "cpl", "cost per lead", "form leads"], | |
| "attribute_keys": ["totalLeads", "costPerLead", "welcomeEmailSentCount"], | |
| "display_metrics": ["Total Leads Generated", "Cost Per Lead (CPL)", "Email Nurture Status"], | |
| }, | |
| } | |
| class UserKnowledgeCacheManager: | |
| """ | |
| 3-Layer Assets Knowledge Graph Cache Manager: | |
| - self._user_cache: In-memory RAM cache indexed by "tenant_id:user_id". | |
| - Disk: Assets Knowledge/<tenant_id>/<user_id>/cache.json. | |
| - DB: Supabase PostgreSQL tables (user_tenant_roles & connections). | |
| """ | |
| def __init__(self): | |
| self._user_cache: dict[str, dict[str, Any]] = {} | |
| def _get_cache_key(self, tenant_id: str, user_id: str) -> str: | |
| t_id = tenant_id or "default_tenant" | |
| u_id = user_id or "default_user" | |
| return f"{t_id}:{u_id}" | |
| def get_cache_file_path(self, tenant_id: str, user_id: str) -> str: | |
| t_id = tenant_id or "default_tenant" | |
| u_id = user_id or "default_user" | |
| user_dir = os.path.join(BASE_KNOWLEDGE_DIR, t_id, u_id) | |
| os.makedirs(user_dir, exist_ok=True) | |
| return os.path.join(user_dir, "cache.json") | |
| def load_user_assets(self, tenant_id: str, user_id: str, force_refresh: bool = False) -> dict[str, Any]: | |
| """ | |
| 3-Layer Knowledge Graph Resolution: | |
| Step 1 (RAM Check): If key in self._user_cache and not force_refresh -> Return in 0ms! | |
| Step 2 (Disk Check): If cache.json exists on disk and not force_refresh -> Load into RAM and Return in 0ms! | |
| Step 3 (DB Query & Dynamic Build): Fetch connections from DB, enrich in real-time, populate RAM & write to Disk. | |
| """ | |
| cache_key = self._get_cache_key(tenant_id, user_id) | |
| file_path = self.get_cache_file_path(tenant_id, user_id) | |
| if not force_refresh: | |
| # Step 1: In-Memory RAM Check (0ms) | |
| if cache_key in self._user_cache: | |
| print(f"- KnowledgeCache: [Layer 1 RAM Hit] Loaded memory cache for '{cache_key}' (0ms)") | |
| return self._user_cache[cache_key] | |
| # Step 2: Disk File Check (0ms) | |
| if os.path.exists(file_path): | |
| try: | |
| with open(file_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| self._user_cache[cache_key] = data | |
| print(f"- KnowledgeCache: [Layer 2 Disk Hit] Loaded cache.json into RAM for '{cache_key}' (0ms)") | |
| return data | |
| except Exception as e: | |
| print(f"- KnowledgeCache: Warning loading cache.json from disk: {e}") | |
| # Step 3: DB Query & 100% Real-Time Dynamic Build | |
| return self._build_and_save_cache(tenant_id, user_id) | |
| def _build_and_save_cache(self, tenant_id: str, user_id: str) -> dict[str, Any]: | |
| """Fetch active connections from DB, enrich metadata in 100% real-time, and save to RAM + Disk.""" | |
| cache_key = self._get_cache_key(tenant_id, user_id) | |
| file_path = self.get_cache_file_path(tenant_id, user_id) | |
| has_access = True | |
| active_connections = {} | |
| if SUPABASE_KEY and SUPABASE_URL and tenant_id: | |
| headers = { | |
| "apikey": SUPABASE_KEY, | |
| "Authorization": f"Bearer {SUPABASE_KEY}", | |
| "Content-Type": "application/json", | |
| } | |
| # Permission check - supports both snake_case and camelCase | |
| if user_id: | |
| try: | |
| role_url = f"{SUPABASE_URL}/rest/v1/user_tenant_roles" | |
| res = httpx.get(role_url, headers=headers, params={"user_id": f"eq.{user_id}", "tenant_id": f"eq.{tenant_id}", "select": "id"}, timeout=3.0) | |
| if res.status_code != 200 or len(res.json()) == 0: | |
| res2 = httpx.get(role_url, headers=headers, params={"userId": f"eq.{user_id}", "tenantId": f"eq.{tenant_id}", "select": "id"}, timeout=3.0) | |
| if res2.status_code == 200 and len(res2.json()) == 0: | |
| has_access = False | |
| except Exception as e: | |
| print(f"- KnowledgeCache: Permission lookup error: {e}") | |
| # Connection lookup - supports both snake_case and camelCase | |
| try: | |
| conn_url = f"{SUPABASE_URL}/rest/v1/connections" | |
| rows = [] | |
| res1 = httpx.get(conn_url, headers=headers, params={"tenant_id": f"eq.{tenant_id}", "select": "*"}, timeout=3.0) | |
| if res1.status_code == 200: | |
| rows.extend(res1.json()) | |
| res2 = httpx.get(conn_url, headers=headers, params={"tenantId": f"eq.{tenant_id}", "select": "*"}, timeout=3.0) | |
| if res2.status_code == 200: | |
| rows.extend(res2.json()) | |
| for r in rows: | |
| p = r.get("provider") | |
| st = str(r.get("status", "")).lower() | |
| if p and st in ["connected", "active"]: | |
| active_connections[p] = r | |
| except Exception as e: | |
| print(f"- KnowledgeCache: Connections lookup error: {e}") | |
| # Construct Knowledge Graph dynamically from live DB connections | |
| knowledge_graph = [] | |
| for provider_id, conn_data in active_connections.items(): | |
| meta = conn_data.get("metadata") or {} | |
| def_meta = DEFAULT_PROVIDER_METADATA.get(provider_id, { | |
| "name": provider_id.title(), | |
| "category": "general", | |
| "synonyms": [provider_id], | |
| "attribute_keys": list(meta.keys()) if meta else ["status"], | |
| "display_metrics": [f"{provider_id.title()} Performance Metrics"], | |
| }) | |
| item = { | |
| "id": conn_data.get("id", f"conn-{provider_id}"), | |
| "provider": provider_id, | |
| "name": meta.get("name") or def_meta.get("name"), | |
| "category": def_meta.get("category", "general"), | |
| "status": "connected", | |
| "attribute_keys": meta.get("metrics") or def_meta.get("attribute_keys"), | |
| "display_metrics": def_meta.get("display_metrics"), | |
| "synonyms": def_meta.get("synonyms"), | |
| "metadata": { | |
| "propertyId": meta.get("propertyId") or meta.get("property_id") or conn_data.get("externalId"), | |
| "shop": meta.get("shop") or meta.get("myshopifyDomain"), | |
| "adAccountId": conn_data.get("externalId") or conn_data.get("external_id"), | |
| }, | |
| } | |
| knowledge_graph.append(item) | |
| # Fallback to standard platform templates if tenant has no active DB connections yet | |
| if not knowledge_graph: | |
| for p_id, def_meta in DEFAULT_PROVIDER_METADATA.items(): | |
| knowledge_graph.append({ | |
| "id": f"conn-{p_id}", | |
| "provider": p_id, | |
| "name": def_meta["name"], | |
| "category": def_meta["category"], | |
| "status": "disconnected", | |
| "attribute_keys": def_meta["attribute_keys"], | |
| "display_metrics": def_meta["display_metrics"], | |
| "synonyms": def_meta["synonyms"], | |
| }) | |
| cache_data = { | |
| "tenant_id": tenant_id, | |
| "user_id": user_id, | |
| "user_authorized": has_access, | |
| "total_platforms": len(knowledge_graph), | |
| "timestamp_epoch": time.time(), | |
| "cached_at": time.strftime("%Y-%m-%d %H:%M:%S"), | |
| "knowledge_graph": knowledge_graph, | |
| } | |
| # 1. Update In-Memory RAM | |
| self._user_cache[cache_key] = cache_data | |
| # 2. Write to Disk cache.json | |
| try: | |
| with open(file_path, "w", encoding="utf-8") as f: | |
| json.dump(cache_data, f, indent=2) | |
| print(f"- KnowledgeCache: [Layer 3 DB Build] Built Knowledge Graph and saved cache.json for '{cache_key}'") | |
| except Exception as e: | |
| print(f"- KnowledgeCache: Error writing cache.json to disk: {e}") | |
| return cache_data | |
| def invalidate_user_cache(self, tenant_id: str, user_id: str): | |
| """Force refresh / Manual sync: Evicts RAM cache and deletes disk cache.json.""" | |
| cache_key = self._get_cache_key(tenant_id, user_id) | |
| file_path = self.get_cache_file_path(tenant_id, user_id) | |
| if cache_key in self._user_cache: | |
| del self._user_cache[cache_key] | |
| if os.path.exists(file_path): | |
| try: | |
| os.remove(file_path) | |
| print(f"- KnowledgeCache: [Force Refresh] Evicted cache.json from RAM and disk for '{cache_key}'") | |
| except Exception as e: | |
| print(f"- KnowledgeCache: Error deleting cache file: {e}") | |
| knowledge_cache_manager = UserKnowledgeCacheManager() | |
| def load_user_assets(tenant_id: str, user_id: str, force_refresh: bool = False) -> dict[str, Any]: | |
| """Helper function to load user Knowledge Graph cache using 3-layer manager.""" | |
| return knowledge_cache_manager.load_user_assets(tenant_id, user_id, force_refresh=force_refresh) | |
| def invalidate_user_cache(tenant_id: str, user_id: str): | |
| """Helper function to invalidate user cache.""" | |
| knowledge_cache_manager.invalidate_user_cache(tenant_id, user_id) | |