cnaged datasource to call worker to querry stored raw data for entity classification

app/routers/datasources.py

@@ -109,18 +109,17 @@ async def create_source_json(
         bootstrap(orgId, payload.data)
         print(f"[api/json] ✅ Raw data stored for org: {orgId}")
          #  Pass RAW PAYLOAD to worker (not just ID)
-        timestamp = int(datetime.now().timestamp())
         task = {
-            "id": f"detect_entity:{org_id}:{source_id}:{timestamp}",  # Unique ID
-            "function": "detect_entity",  # 🎯 Must match registry
+            "id": f"detect_entity:{org_id}:{source_id}:{int(datetime.now().timestamp())}",
+            "function": "detect_entity",
             "args": {
                 "org_id": org_id,
-                "source_id": source_id,
-                "raw_data": payload.data,  # Actual data rows
-                "filename": f"{source_id}.json"
+                "source_id": source_id
+                # No raw_data - worker queries DB
             }
         }
         redis.lpush("python:task_queue", json.dumps(task))
+        print(f"[api/json] 🚀 Task queued: {task['id']}")
        
         df, industry, confidence = canonify_df(org_id, source_id)
         # 3. 🎯 Prepare preview for real-time broadcast

app/tasks/worker.py

@@ -22,25 +22,30 @@ signal.signal(signal.SIGTERM, shutdown)
 
 # ── NEW: Entity Detection Handler ───────────────────────────────────────────────
 def process_detect_entity(org_id: str, **args):
-    """
-    🎯 CRITICAL PATH: Receives raw data from Vercel, detects entity, stores in Redis
-    **DO NOT MODIFY without architect approval**
-    """
+    """🎯 Queries DuckDB for raw data instead of receiving payload"""
     source_id = args["source_id"]
-    raw_data = args["raw_data"]
-    filename = args.get("filename", f"{source_id}.json")
     
-    print(f"🔵 [{org_id}] Entity detection starting for {source_id}")
+    print(f"🔵 [{org_id}] Entity detection STARTED for {source_id}")
     
     try:
-        # 1. Convert raw payload to DataFrame (this is why we pass data directly)
-        df = pd.DataFrame(raw_data)
-        print(f"   📊 DataFrame created: {len(df)} rows × {len(df.columns)} cols")
+        # 1. ✅ Query DuckDB for latest raw rows (the ones just uploaded)
+        conn = get_duckdb(org_id)
+        rows = conn.execute("""
+            SELECT row_data FROM main.raw_rows 
+            WHERE row_data IS NOT NULL 
+            ORDER BY _ingested_at DESC LIMIT 500
+        """).fetchall()
         
-        # 2. Run hybrid detection (rule-based, ~10ms, zero LLM cost)
-        entity_type, confidence, _ = hybrid_detect_entity_type(
-            org_id, df, filename
-        )
+        if not rows:
+            raise RuntimeError(f"No raw data found for {source_id}")
+        
+        # 2. Parse into DataFrame
+        parsed = [json.loads(r[0]) for r in rows if r[0]]
+        df = pd.DataFrame(parsed)
+        print(f"   📊 DataFrame: {len(df)} rows × {len(df.columns)} cols")
+        
+        # 3. Detect entity (rest unchanged)
+        entity_type, confidence, _ = hybrid_detect_entity_type(org_id, df, f"{source_id}.json")
         print(f"   ✅ Detected: {entity_type} ({confidence:.2%})")
         
         # 3. Store in Redis for mapper to poll (HF endpoint is waiting for this)