added detect industry

app/hybrid_industry_detector.py

@@ -0,0 +1,28 @@
+# app/hybrid_industry_detector.py
+import logging
+import pandas as pd
+from typing import Tuple, Dict
+from app.utils.detect_industry import detect_industry as rule_based_detect
+from app.service.ai_service import ai_service
+
+logger = logging.getLogger(__name__)
+
+def hybrid_detect_industry_type(org_id: str, df: pd.DataFrame, filename: str = "") -> Tuple[str, float, bool]:
+    """
+    Detects BUSINESS VERTICAL (SUPERMARKET/MANUFACTURING/PHARMA/RETAIL/WHOLESALE/HEALTHCARE)
+    
+    Returns: (industry, confidence, is_confident)
+    """
+    # 1. Rule-based detection from utils (<10ms, zero LLM cost)
+    industry, confidence = rule_based_detect(df)
+    industry = industry.upper()  # Normalize
+    
+    logger.info(f"[hybrid_industry] RULE-BASED ONLY: {industry} ({confidence:.2f})")
+    
+    # 2. [FUTURE] LLM fallback if confidence < 0.75
+    # if confidence < 0.75:
+    #     logger.info(f"[hybrid_industry] → LLM fallback needed")
+    #     # ... LLM logic here ...
+    
+    # 3. Always return as confident (rule-based is authoritative)
+    return industry, confidence, True

app/mapper.py

@@ -82,7 +82,7 @@ def save_dynamic_aliases() -> None:
         json.dump(CANONICAL, f, indent=2)
 # ✅ Module-level cache: (org_id, source_id) -> entity_info
 _ENTITY_CACHE = {}
-
+_INDUSTRY_CACHE = {}  # NEW
 def poll_for_entity(org_id: str, source_id: str, timeout: int = 30) -> dict:
     """
     🎯 Capped at 2 Redis calls (immediate + after 5s sleep).
@@ -155,7 +155,33 @@ def _fallback_detection(org_id: str, source_id: str) -> dict:
     redis.setex(f"entity:{org_id}:{source_id}", 3600, json.dumps(entity_info))
     
     return entity_info
+    #poll for industry from redis
+# app/mapper.py
+
+# ✅ Add this at top
+_ENTITY_CACHE = {}
+_INDUSTRY_CACHE = {}  # NEW
 
+def poll_for_industry(org_id: str, source_id: str, timeout: int = 10) -> dict:
+    """Returns INDUSTRY for user dashboard"""
+    cache_key = (org_id, source_id)
+    
+    # Check cache FIRST
+    if cache_key in _INDUSTRY_CACHE:
+        print(f"[poll_industry] 💾 CACHE HIT: {cache_key}")
+        return _INDUSTRY_CACHE[cache_key]
+    
+    key = f"industry:{org_id}:{source_id}"
+    
+    for attempt in range(2):
+        data = redis.get(key)
+        if data:
+            info = json.loads(data)
+            _INDUSTRY_CACHE[cache_key] = info  # ✅ Cache it
+            return info
+        time.sleep(5.0)
+    
+    return {"industry": "UNKNOWN", "confidence": 0.0}
 # ----------  Main Canonify Function (ENTERPRISE-GRADE) ---------- #
 def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd.DataFrame, str, float]:
     """
@@ -273,12 +299,16 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
     except Exception as e:
         print(f"[canonify] Type conversion warning (non-critical): {e}")
 
-    # ✅ REPLACE WITH THIS (poll from Redis):
+    # Get ENTITY (internal)
     entity_info = poll_for_entity(org_id, source_id)
     entity_type = entity_info["entity_type"]
-    confidence = entity_info["confidence"]
-    industry = entity_type
-    print(f"[canonify] 🎯 Entity from Redis: {entity_type} ({confidence:.2%})") 
+    entity_confidence = entity_info.get("confidence", 0.0)  # Not used for frontend
+    
+    # Get INDUSTRY (user-facing)
+    industry_info = poll_for_industry(org_id, source_id)
+    industry = industry_info["industry"]
+    industry_confidence = industry_info["confidence"]
+    print(f"[canonify] Industry: {industry} ({industry_confidence:.2%}), Entity: {entity_type}")
 
     # 7) Dynamic schema evolution
     os.makedirs("./db", exist_ok=True)
@@ -313,4 +343,4 @@ def canonify_df(org_id: str, source_id: str, hours_window: int = 24) -> tuple[pd
     duck.close()
     print(f"[canonify] ✅ Pipeline complete for {org_id}")
     
-    return df, industry, confidence
+    return df, industry, industry_confidence 

app/routers/datasources.py

@@ -109,17 +109,13 @@ async def create_source_json(
         bootstrap(orgId, payload.data)
         print(f"[api/json] ✅ Raw data stored for org: {orgId}")
        
-        task = {
-            "id": f"detect_entity:{org_id}:{source_id}:{int(datetime.now().timestamp())}",
-            "function": "detect_entity",
-            "args": {
-                "org_id": org_id,
-                "source_id": source_id
-                # No raw_data - worker queries DB
-            }
+        industry_task = {
+            "id": f"detect_industry:{org_id}:{source_id}:{int(datetime.now().timestamp())}",
+            "function": "detect_industry",
+             "args": {"org_id": org_id, "source_id": source_id}
         }
-        redis.lpush("python:task_queue", json.dumps(task))
-        print(f"[api/json] 🚀 Task queued: {task['id']}")
+        redis.lpush("python:task_queue", json.dumps(industry_task))
+        #  Entity will be auto-queued by process_detect_industry()
        
         df, industry, confidence = canonify_df(org_id, source_id)
         # 3. 🎯 Prepare preview for real-time broadcast

app/tasks/worker.py

@@ -10,7 +10,8 @@ import pandas as pd  # ✅ Required for entity detection
 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  # ✅ New import
+from app.hybrid_entity_detector import hybrid_detect_entity_type  
+from app.industry_detector import detect_industry_type as rule_based_detect
 
 # ── Graceful Shutdown ──────────────────────────────────────────────────────────
 def shutdown(signum, frame):
@@ -20,6 +21,13 @@ def shutdown(signum, frame):
 signal.signal(signal.SIGINT, shutdown)
 signal.signal(signal.SIGTERM, shutdown)
 
+TASK_HANDLERS = {
+    "detect_industry": process_detect_industry,  
+    "detect_entity": process_detect_entity,      
+    "precompute_kpis": process_precompute_kpis,
+    # ...
+}
+
 # ── NEW: Entity Detection Handler ───────────────────────────────────────────────
 def process_detect_entity(org_id: str, **args):
     """🎯 Queries DuckDB for raw data instead of receiving payload"""
@@ -88,6 +96,53 @@ def process_detect_entity(org_id: str, **args):
         # CRITICAL: Re-raise so process_task logs it properly
         raise RuntimeError(f"Entity detection failed for {source_id}: {str(e)}")
 
+def process_detect_industry(org_id: str, **args):
+    """
+    🎯 DETECTS INDUSTRY (business vertical) only.
+    DOES NOT touch entity detection.
+    """
+    source_id = args["source_id"]
+    
+    print(f"🔴🔴🔴 [WORKER] INDUSTRY detection for {org_id}/{source_id}")
+    
+    try:
+        # Query raw data
+        conn = get_duckdb(org_id)
+        rows = conn.execute("SELECT row_data FROM main.raw_rows LIMIT 100").fetchall()
+        
+        if not rows:
+            raise RuntimeError("No raw data")
+        
+        parsed = [json.loads(r[0]) for r in rows if r[0]]
+        df = pd.DataFrame(parsed)
+        
+        # ✅ Use NEW detector (decoupled from entity)
+        industry, confidence, _ = hybrid_detect_industry_type(org_id, df, source_id)
+        
+        # Write to dedicated Redis key
+        redis.setex(f"industry:{org_id}:{source_id}", 3600, json.dumps({
+            "industry": industry,
+            "confidence": confidence
+        }))
+        
+        print(f"✅ [WORKER] INDUSTRY written: {industry} ({confidence:.2%})")
+        
+        # Auto-queue entity detection (separate task, independent)
+        entity_task = {
+            "id": f"detect_entity:{org_id}:{source_id}:{int(time.time())}",
+            "function": "detect_entity",
+            "args": {"org_id": org_id, "source_id": source_id}
+        }
+        redis.lpush("python:task_queue", json.dumps(entity_task))
+        
+    except Exception as e:
+        print(f"❌ [WORKER] Industry detection CRASHED: {e}")
+        redis.setex(f"industry:{org_id}:{source_id}", 3600, json.dumps({
+            "industry": "UNKNOWN",
+            "confidence": 0.0
+        }))
+        raise
+
 # ── Legacy Handlers (Keep for backward compatibility) ────────────────────────
 def canonify_df_with_entity(org_id: str, filename: str, hours_window: int = 24):
     """⚠️ DEPRECATED: Remove once all ingestion uses detect_entity worker"""