industyr detection last refactor

app/mapper.py

@@ -156,11 +156,93 @@ def _fallback_detection(org_id: str, source_id: str) -> dict:
     
     return entity_info
     #poll for industry from redis
-# app/mapper.py
+def poll_for_industry(org_id: str, source_id: str, timeout: int = 10) -> dict:
+    """
+    🎯 Polls Redis for industry detection result (user-facing dashboard label).
+    Capped at 2 Redis calls (immediate + after 5s sleep).
+    In-memory cache prevents re-polling the same source.
+    
+    Returns:
+        dict: {"industry": str, "confidence": float}
+    """
+    cache_key = (org_id, source_id)
+    
+    # 1. Check cache FIRST
+    if cache_key in _INDUSTRY_CACHE:
+        print(f"[poll_industry] 💾 CACHE HIT: {cache_key}")
+        return _INDUSTRY_CACHE[cache_key]
+    
+    industry_key = f"industry:{org_id}:{source_id}"
+    print(f"[poll_industry] ⏳ Polling for key: {industry_key}")
+    
+    # 2. First attempt (immediate)
+    data = redis.get(industry_key)
+    if data:
+        industry_info = json.loads(data)
+        _INDUSTRY_CACHE[cache_key] = industry_info
+        print(f"[poll_industry] ✅ SUCCESS on first attempt: {industry_info['industry']}")
+        return industry_info
+    
+    # 3. Sleep 5 seconds (gives worker time)
+    print(f"[poll_industry] 🔄 First check failed, sleeping 5s...")
+    time.sleep(5.0)
+    
+    # 4. Second attempt (final)
+    data = redis.get(industry_key)
+    if data:
+        industry_info = json.loads(data)
+        _INDUSTRY_CACHE[cache_key] = industry_info
+        print(f"[poll_industry] ✅ SUCCESS on second attempt: {industry_info['industry']}")
+        return industry_info
+    
+    # 5. Emergency fallback (worker is dead)
+    print(f"[poll_industry] ⚠️ Both attempts failed - using direct detection")
+    return _fallback_industry_detection(org_id, source_id)
+    #fallback industry detection
+def _fallback_industry_detection(org_id: str, source_id: str) -> dict:
+    """
+    Emergency: Run industry detection directly from DuckDB data.
+    Uses the actual hybrid detector module you have.
+    Writes result to Redis and cache for recovery.
+    """
+    print(f"[fallback_industry] 🚨 Running fallback for {org_id}/{source_id}")
+    
+    try:
+        conn = get_conn(org_id)
+        rows = conn.execute("""
+            SELECT row_data
+            FROM main.raw_rows
+            WHERE row_data IS NOT NULL
+            USING SAMPLE 100
+        """).fetchall()
+        
+        if not rows:
+            print(f"[fallback_industry] ❌ No data found, returning UNKNOWN")
+            industry_info = {"industry": "UNKNOWN", "confidence": 0.0}
+        else:
+            parsed = [json.loads(r[0]) for r in rows if r[0]]
+            df = pd.DataFrame(parsed)
+            
+            # ✅ CORRECT: Import from your actual module
+            from app.hybrid_industry_detector import hybrid_detect_industry_type
+            
+            # Call it (note: it returns 3 values: industry, confidence, is_confident)
+            industry, confidence, _ = hybrid_detect_industry_type(org_id, df, source_id)
+            industry_info = {"industry": industry, "confidence": confidence}
+            print(f"[fallback_industry] ✅ Direct detection: {industry} ({confidence:.2%})")
+        
+        # Cache and write to Redis
+        cache_key = (org_id, source_id)
+        _INDUSTRY_CACHE[cache_key] = industry_info
+        redis.setex(f"industry:{org_id}:{source_id}", 3600, json.dumps(industry_info))
+        
+        return industry_info
+        
+    except Exception as e:
+        print(f"[fallback_industry] ❌ Failed: {e}")
+        return {"industry": "UNKNOWN", "confidence": 0.0}
+
 
-# ✅ Add this at top
-_ENTITY_CACHE = {}
-_INDUSTRY_CACHE = {}  # NEW
 
 def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd.DataFrame, str, float]:
     """
@@ -205,8 +287,7 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
     
     # Load dynamic aliases before mapping
     load_dynamic_aliases()
-    
-    # 1️⃣ FETCH RAW AUDIT DATA
+   
     # 1️⃣ FETCH RAW AUDIT DATA
     with get_conn(org_id) as conn:
         ensure_raw_table(conn)

app/tasks/worker.py

@@ -11,7 +11,7 @@ from app.redis_client import redis
 from app.service.ai_service import ai_service
 from app.deps import get_duckdb
 from app.hybrid_entity_detector import hybrid_detect_entity_type  
-from app.industry_detector import detect_industry_type as rule_based_detect
+from app.utils.detect_industry import detect_industry as rule_based_detect
 
 # ── Graceful Shutdown ──────────────────────────────────────────────────────────
 def shutdown(signum, frame):