refactoreed descover schema

app/tasks/analytics_worker.py

@@ -332,29 +332,32 @@ class AnalyticsWorker:
     
     # app/tasks/analytics_worker.py - Replace your _discover_schema method
 
-    def _discover_schema(self):
-        """Schema discovery with proper caching and error handling"""
-        if not self._entity_type:
-            logger.error("[SCHEMA] ❌ Cannot discover schema: entity_type not set")
-            raise ValueError("entity_type must be set before schema discovery")
+    # app/tasks/analytics_worker.py - Replace line ~95
+
+    async def _discover_schema(self, df: pd.DataFrame) -> Dict[str, str]:
+        """Schema discovery with entity context (NOW ACCEPTS df)"""
         try:
             logger.info("[SCHEMA] 🧠 Cache miss, discovering...")
         
             from app.schemas.org_schema import OrgSchema
         
-            # Initialize schema discoverer with entity context
-            self.schema = OrgSchema(self.org_id, self._entity_type)
-            mapping = self.schema.get_mapping()
+            # Ensure entity_type is set (from STEP 2)
+            if not getattr(self, '_entity_type', None):
+                raise ValueError("entity_type must be set in STEP 2")
+        
+            # Run sync discovery in thread pool (non-blocking)
+            def sync_discover():
+                schema = OrgSchema(self.org_id, self._entity_type)
+                return schema.get_mapping()
+        
+            mapping = await asyncio.to_thread(sync_discover)
         
             if not mapping:
                 raise ValueError("Empty mapping returned")
             
-            # ✅ FIX: Define cache_key BEFORE using it
+            # Cache for 24h
             cache_key = f"schema:{self.org_id}:{self._entity_type}:worker_cache"
-        
-            # ✅ FIX: Save to Redis with proper TTL
-            event_hub.setex(cache_key, 86400, json.dumps(mapping))
-            logger.info(f"[SCHEMA] 💾 Cached mapping for 24h: {cache_key}")
+            await asyncio.to_thread(event_hub.setex, cache_key, 86400, json.dumps(mapping))
         
             self._schema_cache = mapping
             logger.info(f"[SCHEMA] ✅ Discovery complete: {len(mapping)} columns")
@@ -363,13 +366,13 @@ class AnalyticsWorker:
         except Exception as e:
             logger.error(f"[SCHEMA] ❌ Discovery failed: {e}")
         
-            # 🚀 EMERGENCY FALLBACK: Map columns to themselves (SaaS-ready)
+            # 🚀 EMERGENCY FALLBACK: Map df columns to themselves
             logger.warning("[SCHEMA] 🚨 Using fallback - mapping columns as-is")
-            stealth_mapping = {col: col for col in self.df.columns}
+            stealth_mapping = {col: col for col in df.columns}
         
-            # ✅ Cache the fallback too
-            cache_key = f"schema:{self.org_id}:{self._entity_type}:worker_cache:fallback"
-            event_hub.setex(cache_key, 3600, json.dumps(stealth_mapping))
+            if getattr(self, '_entity_type', None):
+                cache_key = f"schema:{self._entity_type}:fallback"
+                await asyncio.to_thread(event_hub.setex, cache_key, 3600, json.dumps(stealth_mapping))
         
             self._schema_cache = stealth_mapping
             return stealth_mapping