engine.py

import os
import re
import json
import time
import logging
import threading
from typing import Dict, List, Any, Optional, Tuple
from concurrent.futures import ThreadPoolExecutor

import duckdb
from huggingface_hub import HfFileSystem
from cachetools import TTLCache

try:
    import phonenumbers
    from phonenumbers import geocoder
    HAS_PHONENUMBERS = True
except ImportError:
    HAS_PHONENUMBERS = False

logger = logging.getLogger("LookupEngine")
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

BASE_DIR = os.path.dirname(os.path.abspath(__file__))


class LookupEngine:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            with cls._lock:
                if not cls._instance:
                    cls._instance = super(LookupEngine, cls).__new__(cls)
                    cls._instance._initialized = False
        return cls._instance

    def __init__(self, partitions_path: Optional[str] = None):
        if getattr(self, "_initialized", False):
            return

        self._initialized = True
        self.db_lock = threading.Lock()
        
        # In-memory TTL Cache (100,000 queries, 24-hour expiration)
        self.cache = TTLCache(maxsize=100000, ttl=86400)
        self.cache_lock = threading.Lock()

        base_dir = os.path.dirname(os.path.abspath(__file__))

        # Helper to safely load JSON index
        def _load_index(filename: str, default: Any = None):
            p = os.path.join(base_dir, filename)
            if os.path.exists(p):
                try:
                    with open(p, "r", encoding="utf-8") as f:
                        return json.load(f)
                except Exception as e:
                    logger.error(f"Error loading {filename}: {e}")
            return default if default is not None else []

        # Load all Master Database range index routers
        self.partitions = _load_index("partitions.json", {"phone": [], "aadhar": []})
        self.tg_partitions = _load_index("tg_partitions.json", [])
        self.tc_partitions = _load_index("truecaller_index.json", [])
        self.airtel_partitions = _load_index("airtel_index.json", [])
        self.hdfc_partitions = _load_index("hdfc_index.json", [])
        self.bsnl_partitions = _load_index("bsnl_index.json", [])
        self.paytm_partitions = _load_index("paytm_index.json", [])
        self.flipkart_partitions = _load_index("flipkart_index.json", [])
        self.twitter_partitions = _load_index("twitter_index.json", [])
        self.ig_phone_partitions = _load_index("instagram_phone_index.json", [])
        self.ig_user_partitions = _load_index("instagram_username_index.json", [])
        self.ind_partitions = _load_index("ind_partitions.json", [])

        logger.info(
            f"Loaded Master Index Routers: "
            f"Truecaller({len(self.tc_partitions)}), "
            f"Airtel({len(self.airtel_partitions)}), "
            f"HDFC({len(self.hdfc_partitions)}), "
            f"BSNL({len(self.bsnl_partitions)}), "
            f"Paytm({len(self.paytm_partitions)}), "
            f"Flipkart({len(self.flipkart_partitions)}), "
            f"Telegram({len(self.tg_partitions)}), "
            f"Twitter({len(self.twitter_partitions)}), "
            f"InstagramPhone({len(self.ig_phone_partitions)}), "
            f"InstagramUser({len(self.ig_user_partitions)}), "
            f"IND({len(self.ind_partitions)}), "
            f"TelecomPhone({len(self.partitions.get('phone', []))})"
        )

        # Initialize DuckDB and HfFileSystem with HF token from environment
        hf_token = os.environ.get("HF_TOKEN")
        self.fs = HfFileSystem(token=hf_token)
        self.con = duckdb.connect()
        self.con.register_filesystem(self.fs)
        
        # Configure DuckDB runtime settings
        self.con.execute("SET threads = 4;")
        self.con.execute("SET preserve_insertion_order = false;")
        if hf_token:
            try:
                self.con.execute(f"CREATE SECRET (TYPE HUGGINGFACE, TOKEN '{hf_token}');")
            except Exception:
                pass
        logger.info("DuckDB engine initialized with HfFileSystem and secret registered.")

        # Thread pool for fast parallel queries
        self.executor = ThreadPoolExecutor(max_workers=10)

        # Background pre-warm thread
        self.is_warming = False
        self.is_warmed = False

    def warm_up(self):
        """Pre-warm DuckDB metadata cache for instant sub-second response times."""
        if self.is_warming or self.is_warmed:
            return

        def _worker():
            self.is_warming = True
            logger.info("Starting background engine pre-warming...")
            t0 = time.time()
            
            warm_targets = []
            if self.tc_partitions:
                warm_targets.append((self.tc_partitions[0]["url"], "mobile"))
            if self.airtel_partitions:
                warm_targets.append((self.airtel_partitions[0]["url"], "mobile"))
            if self.hdfc_partitions:
                warm_targets.append((self.hdfc_partitions[0]["url"], "mobile"))
            if self.tg_partitions:
                warm_targets.append((self.tg_partitions[0]["url"], "phone"))
            if self.ind_partitions:
                warm_targets.append((self.ind_partitions[0]["url"], "phone"))
            
            for url, col in warm_targets:
                try:
                    with self.db_lock:
                        self.con.execute(f"SELECT {col} FROM '{url}' LIMIT 1").fetchall()
                    logger.info(f"Pre-warmed: {url.split('/')[-1]}")
                except Exception as e:
                    logger.warning(f"Pre-warm notice for {url}: {e}")

            self.is_warming = False
            self.is_warmed = True
            logger.info(f"Pre-warming completed in {time.time()-t0:.2f}s!")

        threading.Thread(target=_worker, daemon=True).start()

    # ------------------------------------------------------------------
    # Helper Utilities
    # ------------------------------------------------------------------
    @staticmethod
    def sanitize_phone(phone: Any) -> Optional[int]:
        """Cleans and standardizes Indian phone numbers to 10-digit integer."""
        if not phone:
            return None
        digits = re.sub(r"\D", "", str(phone).strip())
        if len(digits) == 12 and digits.startswith("91"):
            digits = digits[2:]
        elif len(digits) == 11 and digits.startswith("0"):
            digits = digits[1:]
        if len(digits) == 10 and digits[0] in "6789":
            try:
                return int(digits)
            except ValueError:
                return None
        return None

    @staticmethod
    def _route_range(partitions: List[Dict[str, Any]], val: Any, min_key: str, max_key: str) -> Optional[Dict[str, Any]]:
        """Ultra-fast O(log N) binary search routing to exact partition Parquet file."""
        if not partitions or val is None:
            return None
        
        # Detect whether partition bounds are numbers or strings
        try:
            first_val = partitions[0].get(min_key)
            is_numeric = isinstance(first_val, (int, float)) or (isinstance(first_val, str) and first_val.strip().isdigit())
        except Exception:
            is_numeric = False

        if is_numeric:
            try:
                target = int(val)
                cast_fn = lambda x: int(x) if x is not None else 0
            except (ValueError, TypeError):
                return None
        else:
            target = str(val).strip().lower()
            cast_fn = lambda x: str(x).strip().lower() if x is not None else ""

        low = 0
        high = len(partitions) - 1
        while low <= high:
            mid = (low + high) // 2
            p = partitions[mid]
            try:
                p_min = cast_fn(p[min_key])
                p_max = cast_fn(p[max_key])
            except Exception:
                p_min = p[min_key]
                p_max = p[max_key]

            if p_min <= target <= p_max:
                return p
            elif target < p_min:
                high = mid - 1
            else:
                low = mid + 1
        return None

    @staticmethod
    def _clean_record(row: tuple, cols: List[str]) -> Dict[str, Any]:
        """Cleans raw row tuples into formatted dictionary."""
        rec = {}
        for col_name, val in zip(cols, row):
            if val is None:
                rec[col_name] = None
            elif isinstance(val, str):
                cleaned = val.replace("\x00", "").replace("\t", " ").strip()
                # Clean ="..." CSV formula artifacts
                if cleaned.startswith('="') and cleaned.endswith('"'):
                    cleaned = cleaned[2:-1]
                rec[col_name] = cleaned if cleaned else None
            else:
                rec[col_name] = val
        return rec

    def _execute_query(self, query: str, cols: List[str]) -> List[Dict[str, Any]]:
        """Executes a query safely with db_lock."""
        try:
            with self.db_lock:
                rows = self.con.execute(query).fetchall()
            return [self._clean_record(r, cols) for r in rows]
        except Exception as e:
            logger.error(f"DuckDB Query Error: {e} | Query: {query}")
            return []

    # ------------------------------------------------------------------
    # 1. Truecaller 285M Master Database Lookup
    # ------------------------------------------------------------------
    def query_truecaller(self, phone_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        mobile = self.sanitize_phone(phone_input)
        if not mobile:
            return {"status": "error", "message": "Invalid 10-digit Indian mobile number."}

        cache_key = f"tc:{mobile}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        partition = self._route_range(self.tc_partitions, mobile, "min_mobile", "max_mobile")
        if not partition:
            return {"status": "not_found", "database": "Truecaller", "mobile": mobile, "latency_ms": round((time.time() - t0) * 1000, 2)}

        cols = ["mobile", "name", "address", "email", "gender", "carrier"]
        q = f"SELECT {', '.join(cols)} FROM '{partition['url']}' WHERE mobile = {mobile} LIMIT 5"
        records = self._execute_query(q, cols)

        res = {
            "status": "success" if records else "not_found",
            "database": "Truecaller 285M",
            "mobile": mobile,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 2. Airtel India 73M Master Database Lookup
    # ------------------------------------------------------------------
    def query_airtel(self, phone_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        mobile = self.sanitize_phone(phone_input)
        if not mobile:
            return {"status": "error", "message": "Invalid 10-digit Indian mobile number."}

        cache_key = f"airtel:{mobile}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        partition = self._route_range(self.airtel_partitions, mobile, "min_mobile", "max_mobile")
        if not partition:
            return {"status": "not_found", "database": "Airtel", "mobile": mobile, "latency_ms": round((time.time() - t0) * 1000, 2)}

        cols = ["mobile", "name", "carrier", "address", "email", "gender", "job_title", "company", "facebook", "twitter"]
        q = f"SELECT {', '.join(cols)} FROM '{partition['url']}' WHERE mobile = {mobile} LIMIT 5"
        records = self._execute_query(q, cols)

        res = {
            "status": "success" if records else "not_found",
            "database": "Airtel India 73M",
            "mobile": mobile,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 3. HDFC Bank Master Database Lookup
    # ------------------------------------------------------------------
    def query_hdfc(self, phone_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        mobile = self.sanitize_phone(phone_input)
        if not mobile:
            return {"status": "error", "message": "Invalid 10-digit Indian mobile number."}

        cache_key = f"hdfc:{mobile}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        partition = self._route_range(self.hdfc_partitions, mobile, "min_mobile", "max_mobile")
        if not partition:
            return {"status": "not_found", "database": "HDFC Bank", "mobile": mobile, "latency_ms": round((time.time() - t0) * 1000, 2)}

        cols = [
            "mobile", "name", "dob", "city", "state", "pincode", "cibil", 
            "highmark_score", "email", "bank", "branch", "product_type", 
            "application_id", "status", "applied_amount", "approved_amount", 
            "finance_amount", "card_category", "login_date"
        ]
        q = f"SELECT {', '.join(cols)} FROM '{partition['url']}' WHERE mobile = {mobile} LIMIT 5"
        records = self._execute_query(q, cols)

        res = {
            "status": "success" if records else "not_found",
            "database": "HDFC Bank Master",
            "mobile": mobile,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 4. BSNL 19M Master Database Lookup
    # ------------------------------------------------------------------
    def query_bsnl(self, phone_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        mobile = self.sanitize_phone(phone_input)
        if not mobile:
            return {"status": "error", "message": "Invalid 10-digit Indian mobile number."}

        cache_key = f"bsnl:{mobile}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        partition = self._route_range(self.bsnl_partitions, mobile, "min_phone", "max_phone")
        if not partition:
            return {"status": "not_found", "database": "BSNL", "mobile": mobile, "latency_ms": round((time.time() - t0) * 1000, 2)}

        cols = ["phone", "name", "carrier", "address", "email"]
        q = f"SELECT {', '.join(cols)} FROM '{partition['url']}' WHERE phone = {mobile} LIMIT 5"
        records = self._execute_query(q, cols)

        res = {
            "status": "success" if records else "not_found",
            "database": "BSNL 19M Master",
            "mobile": mobile,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 5. Paytm India Master Database Lookup
    # ------------------------------------------------------------------
    def query_paytm(self, phone_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        mobile = self.sanitize_phone(phone_input)
        if not mobile:
            return {"status": "error", "message": "Invalid 10-digit Indian mobile number."}

        cache_key = f"paytm:{mobile}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        partition = self._route_range(self.paytm_partitions, mobile, "min_phone", "max_phone")
        if not partition:
            return {"status": "not_found", "database": "Paytm", "mobile": mobile, "latency_ms": round((time.time() - t0) * 1000, 2)}

        cols = ["phone", "name", "email", "city", "state", "bank", "dob", "pan"]
        q = f"SELECT {', '.join(cols)} FROM '{partition['url']}' WHERE phone = {mobile} LIMIT 5"
        records = self._execute_query(q, cols)

        res = {
            "status": "success" if records else "not_found",
            "database": "Paytm India Master",
            "mobile": mobile,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 6. Flipkart India Master Database Lookup
    # ------------------------------------------------------------------
    def query_flipkart(self, phone_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        mobile = self.sanitize_phone(phone_input)
        if not mobile:
            return {"status": "error", "message": "Invalid 10-digit Indian mobile number."}

        cache_key = f"flipkart:{mobile}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        partition = self._route_range(self.flipkart_partitions, mobile, "min_phone", "max_phone")
        if not partition:
            return {"status": "not_found", "database": "Flipkart", "mobile": mobile, "latency_ms": round((time.time() - t0) * 1000, 2)}

        cols = ["phone", "name", "email", "address", "city", "state", "pincode"]
        q = f"SELECT {', '.join(cols)} FROM '{partition['url']}' WHERE phone = {mobile} LIMIT 5"
        records = self._execute_query(q, cols)

        res = {
            "status": "success" if records else "not_found",
            "database": "Flipkart India Master",
            "mobile": mobile,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 7. Instagram Master Database Lookup (Phone or Username)
    # ------------------------------------------------------------------
    def query_instagram(self, query_input: str) -> Dict[str, Any]:
        t0 = time.time()
        clean = str(query_input).strip().lstrip("@")
        
        # Check if phone
        mobile = self.sanitize_phone(clean)
        if mobile:
            cache_key = f"ig_p:{mobile}"
            with self.cache_lock:
                if cache_key in self.cache:
                    return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

            partition = self._route_range(self.ig_phone_partitions, mobile, "min_phone", "max_phone")
            if not partition:
                return {"status": "not_found", "database": "Instagram", "query": clean, "latency_ms": round((time.time() - t0) * 1000, 2)}

            cols = ["user_id", "clean_num", "phone", "username", "name", "email", "address"]
            q = f"SELECT {', '.join(cols)} FROM '{partition['url']}' WHERE clean_num = {mobile} LIMIT 5"
            records = self._execute_query(q, cols)
            res = {
                "status": "success" if records else "not_found",
                "database": "Instagram (by Phone)",
                "query": str(mobile),
                "count": len(records),
                "records": records,
                "latency_ms": round((time.time() - t0) * 1000, 2),
                "cached": False
            }
            with self.cache_lock:
                self.cache[cache_key] = res
            return res

        # Otherwise query by username
        uname = clean.lower()
        cache_key = f"ig_u:{uname}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        # Route by username range
        target_part = None
        for p in self.ig_user_partitions:
            if str(p.get("min_username", "")).lower() <= uname <= str(p.get("max_username", "")).lower() + "\uffff":
                target_part = p
                break
        
        if not target_part and self.ig_user_partitions:
            target_part = self.ig_user_partitions[0]

        if not target_part:
            return {"status": "not_found", "database": "Instagram", "query": uname, "latency_ms": round((time.time() - t0) * 1000, 2)}

        cols = ["user_id", "username", "clean_num", "phone", "name", "email", "address"]
        q = f"SELECT {', '.join(cols)} FROM '{target_part['url']}' WHERE LOWER(username) = '{uname}' LIMIT 5"
        records = self._execute_query(q, cols)

        res = {
            "status": "success" if records else "not_found",
            "database": "Instagram (by Username)",
            "query": uname,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 8. Twitter / X Master Database Lookup (User ID or Screen Name)
    # ------------------------------------------------------------------
    def query_twitter(self, query_input: str) -> Dict[str, Any]:
        t0 = time.time()
        clean = str(query_input).strip().lstrip("@")
        cols = ["user_id", "screen_name", "name", "email", "phone", "location", "description", "followers_count", "friends_count", "created_at", "verified"]

        # Check if integer user_id
        if clean.isdigit() and len(clean) >= 5:
            uid = int(clean)
            cache_key = f"tw_id:{uid}"
            with self.cache_lock:
                if cache_key in self.cache:
                    return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

            partition = self._route_range(self.twitter_partitions, uid, "min_id", "max_id")
            if not partition:
                return {"status": "not_found", "database": "Twitter", "query": clean, "latency_ms": round((time.time() - t0) * 1000, 2)}

            q = f"SELECT {', '.join(cols)} FROM '{partition['url']}' WHERE user_id = {uid} LIMIT 1"
            records = self._execute_query(q, cols)
            res = {
                "status": "success" if records else "not_found",
                "database": "Twitter / X (by User ID)",
                "query": str(uid),
                "count": len(records),
                "records": records,
                "latency_ms": round((time.time() - t0) * 1000, 2),
                "cached": False
            }
            with self.cache_lock:
                self.cache[cache_key] = res
            return res

        # Screen Name query across 6 partitions
        sname = clean.lower()
        cache_key = f"tw_sn:{sname}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        records = []
        for p in self.twitter_partitions:
            q = f"SELECT {', '.join(cols)} FROM '{p['url']}' WHERE LOWER(screen_name) = '{sname}' LIMIT 1"
            records = self._execute_query(q, cols)
            if records:
                break

        res = {
            "status": "success" if records else "not_found",
            "database": "Twitter / X (by Screen Name)",
            "query": sname,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 9. Telegram Master DB (Existing)
    # ------------------------------------------------------------------
    def route_telegram(self, user_id: int) -> Optional[Dict[str, Any]]:
        return self._route_range(self.tg_partitions, user_id, "min_id", "max_id")

    @staticmethod
    def resolve_country(raw_phone: str) -> Tuple[str, str, str, str]:
        clean = str(raw_phone).strip().lstrip('+')
        if not clean:
            return ("", "Unknown", "", "")
        num_str = "+" + clean
        if HAS_PHONENUMBERS:
            try:
                obj = phonenumbers.parse(num_str, None)
                if phonenumbers.is_valid_number(obj) or phonenumbers.is_possible_number(obj):
                    country_name = geocoder.country_name_for_number(obj, "en") or geocoder.description_for_number(obj, "en")
                    iso_code = phonenumbers.region_code_for_number(obj) or ""
                    intl_fmt = phonenumbers.format_number(obj, phonenumbers.PhoneNumberFormat.INTERNATIONAL)
                    e164_fmt = phonenumbers.format_number(obj, phonenumbers.PhoneNumberFormat.E164)
                    return (e164_fmt, country_name or "International", iso_code, intl_fmt)
            except Exception:
                pass
        return ("+" + clean, "International", "", "+" + clean)

    def query_by_telegram(self, user_id_input: Any) -> Dict[str, Any]:
        t_start = time.time()
        try:
            clean_id = int(re.sub(r"\D", "", str(user_id_input).strip()))
        except (ValueError, TypeError):
            return {"status": "error", "message": "Invalid user_id."}

        cache_key = f"tg:{clean_id}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t_start) * 1000, 2)}

        partition = self.route_telegram(clean_id)
        if not partition:
            return {"status": "not_found", "user_id": clean_id, "message": "User ID not found.", "latency_ms": round((time.time() - t_start) * 1000, 2)}

        query = f"SELECT phone FROM '{partition['url']}' WHERE user_id = {clean_id} LIMIT 1"
        row = None
        try:
            with self.db_lock:
                row = self.con.execute(query).fetchone()
        except Exception as e:
            logger.error(f"Telegram query error: {e}")

        elapsed_ms = round((time.time() - t_start) * 1000, 2)
        if row and row[0]:
            e164, country_name, iso_code, intl_fmt = self.resolve_country(str(row[0]).strip())
            response = {
                "status": "success",
                "user_id": clean_id,
                "country": country_name,
                "num": e164,
                "phone": e164,
                "formatted_phone": intl_fmt,
                "country_code": iso_code,
                "latency_ms": elapsed_ms,
                "cached": False
            }
        else:
            response = {
                "status": "not_found",
                "user_id": clean_id,
                "message": f"User ID {clean_id} not found in Telegram Master DB.",
                "latency_ms": elapsed_ms,
                "cached": False
            }

        with self.cache_lock:
            self.cache[cache_key] = response
        return response

    # ------------------------------------------------------------------
    # 10. Indian Telecom & Aadhar DBs (Existing)
    # ------------------------------------------------------------------
    def query_by_phone(self, phone_input: str) -> Dict[str, Any]:
        t_start = time.time()
        mobile = self.sanitize_phone(phone_input)
        if not mobile:
            return {"success": False, "error": "Invalid phone number.", "records": [], "count": 0}

        phone = str(mobile)
        cache_key = f"phone:{phone}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t_start) * 1000, 2)}

        matches = []
        for p in self.partitions.get("phone", []):
            if p.get("min", "") <= phone <= p.get("max", ""):
                matches.append(p["url"])
        if not matches and self.partitions.get("phone"):
            matches.append(self.partitions["phone"][-1]["url"])

        cols = ["name", "fathersName", "phoneNumber", "aadharNumber", "otherNumber", "address", "district", "pincode", "state", "town", "source"]
        records = []
        for url in matches:
            q = f"SELECT {', '.join(cols)} FROM '{url}' WHERE phoneNumber = '{phone}'"
            records.extend(self._execute_query(q, cols))

        unique_records = []
        seen = set()
        for r in records:
            k = (r.get("name"), r.get("phoneNumber"), r.get("aadharNumber"), r.get("address"))
            if k not in seen:
                seen.add(k)
                unique_records.append(r)

        res = {
            "success": True if unique_records else False,
            "database": "Telecom Master DB",
            "query": phone,
            "formatted_query": f"+91 {phone}",
            "count": len(unique_records),
            "records": unique_records,
            "latency_ms": round((time.time() - t_start) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    def query_by_aadhar(self, aadhar_input: str) -> Dict[str, Any]:
        t_start = time.time()
        clean = re.sub(r"\D", "", str(aadhar_input).strip())
        if len(clean) != 12:
            return {"success": False, "error": "Aadhaar number must be 12 digits.", "records": [], "count": 0}

        cache_key = f"aadhar:{clean}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t_start) * 1000, 2)}

        matches = []
        for p in self.partitions.get("aadhar", []):
            if p.get("min", "") <= clean <= p.get("max", ""):
                matches.append(p["url"])

        cols = ["name", "fathersName", "phoneNumber", "aadharNumber", "otherNumber", "address", "district", "pincode", "state", "town", "source"]
        records = []
        for url in matches:
            q = f"SELECT {', '.join(cols)} FROM '{url}' WHERE aadharNumber = '{clean}'"
            records.extend(self._execute_query(q, cols))

        res = {
            "success": True if records else False,
            "database": "Aadhaar Master DB",
            "query": clean,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t_start) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 10b. Aadhaar Family & Household Linkage Database
    # ------------------------------------------------------------------
    def query_by_aadhar_family(self, query_input: str) -> Dict[str, Any]:
        """
        Aadhaar Family & Household Linkage Lookup.
        Takes 12-digit Aadhaar UID, 10-digit mobile number, or address:
        1. Identifies the primary individual's record.
        2. Discovers linked household members sharing the same address, 
           co-residents sharing same father's name, or linked family phone numbers.
        3. Returns structured family tree with relations and demographics.
        """
        t_start = time.time()
        raw = str(query_input).strip()
        clean = re.sub(r"\D", "", raw)

        cache_key = f"family:{clean or raw}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t_start) * 1000, 2)}

        family_records: List[Dict[str, Any]] = []
        seen = set()

        def _add_record(r: Dict[str, Any], default_relation: str = "Household Member"):
            if not r or not isinstance(r, dict):
                return
            key = (r.get("name"), r.get("aadharNumber"), r.get("phoneNumber"), r.get("address"))
            if key in seen:
                return
            seen.add(key)
            rec = dict(r)
            if not rec.get("relation"):
                rec["relation"] = default_relation
            family_records.append(rec)

        primary_rec: Optional[Dict[str, Any]] = None
        cols = ["name", "fathersName", "phoneNumber", "aadharNumber", "otherNumber", "address", "district", "pincode", "state", "town", "source"]

        # Case 1: 12-digit Aadhaar Number
        if len(clean) == 12:
            base_res = self.query_by_aadhar(clean)
            if base_res.get("records"):
                primary_rec = base_res["records"][0]
                for r in base_res["records"]:
                    _add_record(r, "Self (Primary Target)")

        # Case 2: 10-digit Phone Number
        elif len(clean) == 10:
            base_res = self.query_by_phone(clean)
            if base_res.get("records"):
                primary_rec = base_res["records"][0]
                for r in base_res["records"]:
                    _add_record(r, "Self (Primary Target)")
                target_aadhar = primary_rec.get("aadharNumber")
                if target_aadhar and len(re.sub(r"\D", "", str(target_aadhar))) == 12:
                    a_res = self.query_by_aadhar(str(target_aadhar))
                    for r in a_res.get("records", []):
                        _add_record(r, "Self (Aadhaar Verified)")

        # If primary record located, locate family members via linkages
        if primary_rec:
            p_addr = (primary_rec.get("address") or "").strip()
            p_father = (primary_rec.get("fathersName") or "").strip()
            p_other = (primary_rec.get("otherNumber") or "").strip()
            p_phone = (primary_rec.get("phoneNumber") or "").strip()
            p_name = (primary_rec.get("name") or "").strip()

            # Linkage 1: Check otherNumber / alternate contact in Telecom DB (Instant indexed partition routing)
            linked_phones = []
            if p_other and len(re.sub(r"\D", "", p_other)) == 10 and p_other != p_phone:
                linked_phones.append(p_other)
            if p_phone and len(re.sub(r"\D", "", p_phone)) == 10 and len(clean) == 12:
                linked_phones.append(p_phone)

            for lp in linked_phones:
                ph_res = self.query_by_phone(lp)
                for r in ph_res.get("records", []):
                    rel = "Linked Household Member"
                    if p_father and r.get("fathersName") and str(r.get("fathersName")).lower() == p_father.lower():
                        rel = "Sibling / Co-Resident"
                    elif p_name and r.get("fathersName") and p_name.lower() in str(r.get("fathersName")).lower():
                        rel = "Child / Dependant"
                    _add_record(r, rel)

            # Linkage 2: Check Truecaller for alternate family name registered on this phone
            target_phone = p_phone or (clean if len(clean) == 10 else None)
            if target_phone and len(re.sub(r"\D", "", target_phone)) == 10:
                try:
                    tc_res = self.query_truecaller(target_phone)
                    for r in tc_res.get("records", []):
                        tc_name = r.get("name")
                        if tc_name and tc_name.lower() != p_name.lower():
                            _add_record({
                                "name": tc_name,
                                "phoneNumber": target_phone,
                                "address": r.get("address") or p_addr,
                                "source": "Truecaller 285M",
                                "fathersName": p_father,
                                "aadharNumber": None,
                                "otherNumber": p_other
                            }, "Co-Registered Subscriber")
                except Exception as e:
                    logger.debug(f"Truecaller family linkage notice: {e}")

            # Linkage 3: Check Airtel Master DB for alternate family name
            if target_phone and len(re.sub(r"\D", "", target_phone)) == 10:
                try:
                    airtel_res = self.query_airtel(target_phone)
                    for r in airtel_res.get("records", []):
                        a_name = r.get("name")
                        if a_name and a_name.lower() != p_name.lower():
                            _add_record({
                                "name": a_name,
                                "phoneNumber": target_phone,
                                "address": r.get("address") or p_addr,
                                "source": "Airtel 73M",
                                "fathersName": p_father,
                                "aadharNumber": None,
                                "otherNumber": p_other
                            }, "Registered Mobile Subscriber")
                except Exception as e:
                    logger.debug(f"Airtel family linkage notice: {e}")

        res = {
            "success": True if family_records else False,
            "database": "Aadhaar Family Master DB",
            "query": clean or raw,
            "count": len(family_records),
            "records": family_records,
            "latency_ms": round((time.time() - t_start) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 10c. SBI POSP & Indian Banking Master Database
    # ------------------------------------------------------------------
    def query_sbi(self, query_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        raw_val = str(query_input).strip()
        clean_phone = self.sanitize_phone(raw_val)

        cache_key = f"sbi:{clean_phone or raw_val.lower()}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        local_path = os.path.join(BASE_DIR, "sbi_master.parquet")
        target_url = local_path if os.path.exists(local_path) else "hf://datasets/ItsXyron/SBI-Master-DB/sbi_master.parquet"

        cols = [
            "phone", "name", "address", "city", "state", "pincode", 
            "email", "aadhaar", "pan", "father_name", "bank_name", 
            "account_number", "ifsc_code", "category", "source"
        ]

        if clean_phone:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE phone = '{clean_phone}' LIMIT 10"
        elif len(raw_val) == 12 and raw_val.isdigit():
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE aadhaar = '{raw_val}' LIMIT 10"
        elif len(raw_val) == 10 and raw_val.isalnum():
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE UPPER(pan) = '{raw_val.upper()}' LIMIT 10"
        elif "@" in raw_val:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE LOWER(email) = '{raw_val.lower()}' LIMIT 10"
        else:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE LOWER(name) LIKE '%{raw_val.lower()}%' LIMIT 10"

        records = self._execute_query(q, cols)

        res = {
            "status": "success" if records else "not_found",
            "database": "SBI POSP & Indian Banking Master",
            "query": raw_val,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 10d. IND Education, Admissions & KYC Master Database
    # ------------------------------------------------------------------
    def query_ind(self, query_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        raw_val = str(query_input).strip()
        clean_phone = self.sanitize_phone(raw_val)

        cache_key = f"ind:{clean_phone or raw_val.lower()}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        cols = [
            "phone", "email", "name", "father_name", "mother_name", "dob",
            "aadhaar", "pan", "address", "city", "state", "pincode",
            "college_name", "bank_name", "account_number", "ifsc_code",
            "role", "source"
        ]

        records = []
        if clean_phone:
            phone_int = int(clean_phone)
            partition = self._route_range(self.ind_partitions, phone_int, "min_id", "max_id")
            if partition:
                q = f"SELECT {', '.join(cols)} FROM '{partition['url']}' WHERE phone = {phone_int} LIMIT 10"
                records = self._execute_query(q, cols)
        else:
            urls = [p["url"] for p in self.ind_partitions] if self.ind_partitions else []
            if urls:
                url_list_str = "[" + ", ".join([f"'{u}'" for u in urls]) + "]"
                if len(raw_val) == 12 and raw_val.isdigit():
                    q = f"SELECT {', '.join(cols)} FROM read_parquet({url_list_str}) WHERE aadhaar = '{raw_val}' LIMIT 10"
                elif len(raw_val) == 10 and raw_val.isalnum() and not raw_val.isdigit():
                    q = f"SELECT {', '.join(cols)} FROM read_parquet({url_list_str}) WHERE UPPER(pan) = '{raw_val.upper()}' LIMIT 10"
                elif "@" in raw_val:
                    q = f"SELECT {', '.join(cols)} FROM read_parquet({url_list_str}) WHERE LOWER(email) = '{raw_val.lower()}' LIMIT 10"
                else:
                    q = f"SELECT {', '.join(cols)} FROM read_parquet({url_list_str}) WHERE LOWER(name) LIKE '%{raw_val.lower()}%' LIMIT 10"
                records = self._execute_query(q, cols)

        res = {
            "status": "success" if records else "not_found",
            "database": "IND Admissions & KYC Master",
            "query": raw_val,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 10e. Muthoot Finance Gold Loans & KYC Master Database
    # ------------------------------------------------------------------
    def query_muthoot(self, query_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        raw_val = str(query_input).strip()
        clean_phone = self.sanitize_phone(raw_val)

        cache_key = f"muthoot:{clean_phone or raw_val.lower()}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        local_path = os.path.join(BASE_DIR, "muthootfinance.parquet")
        target_url = local_path if os.path.exists(local_path) else "hf://datasets/ItsXyron/SBI-Master-DB/muthootfinance.parquet"

        cols = [
            "first_name", "last_name", "fathers_name", "mothers_name", "mobile", 
            "email_id", "aadhaar_no", "pan", "educational_qualification", 
            "address_1", "address_2", "city", "state", "pincode", 
            "bank_account_number", "account_name", "bank_name", "ifsc", "bank_branch"
        ]

        if clean_phone:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE mobile = '{clean_phone}' LIMIT 10"
        elif len(raw_val) == 12 and raw_val.isdigit():
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE aadhaar_no = '{raw_val}' LIMIT 10"
        elif len(raw_val) == 10 and raw_val.isalnum() and not raw_val.isdigit():
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE UPPER(pan) = '{raw_val.upper()}' LIMIT 10"
        elif "@" in raw_val:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE LOWER(email_id) = '{raw_val.lower()}' LIMIT 10"
        else:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE LOWER(first_name) LIKE '%{raw_val.lower()}%' OR LOWER(last_name) LIKE '%{raw_val.lower()}%' LIMIT 10"

        records = self._execute_query(q, cols)
        res = {
            "status": "success" if records else "not_found",
            "database": "Muthoot Finance KYC Master",
            "query": raw_val,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 10f. OLX Auto Dealers & Classifieds Merchants Master Database
    # ------------------------------------------------------------------
    def query_olx(self, query_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        raw_val = str(query_input).strip()
        clean_phone = self.sanitize_phone(raw_val)

        cache_key = f"olx:{clean_phone or raw_val.lower()}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        local_path = os.path.join(BASE_DIR, "olx_bulk.parquet")
        target_url = local_path if os.path.exists(local_path) else "hf://datasets/ItsXyron/SBI-Master-DB/olx_bulk.parquet"

        cols = ["username", "mobile_number", "email", "address", "state", "city", "pincode", "aadhaar", "pan"]

        if clean_phone:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE mobile_number = '{clean_phone}' LIMIT 10"
        elif len(raw_val) == 12 and raw_val.isdigit():
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE aadhaar = '{raw_val}' LIMIT 10"
        elif len(raw_val) == 10 and raw_val.isalnum() and not raw_val.isdigit():
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE UPPER(pan) = '{raw_val.upper()}' LIMIT 10"
        elif "@" in raw_val:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE LOWER(email) = '{raw_val.lower()}' LIMIT 10"
        else:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE LOWER(username) LIKE '%{raw_val.lower()}%' LIMIT 10"

        records = self._execute_query(q, cols)
        res = {
            "status": "success" if records else "not_found",
            "database": "OLX India Merchant Master",
            "query": raw_val,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 10g. ICICI Private Car & Auto Insurance Master Database
    # ------------------------------------------------------------------
    def query_icici(self, query_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        raw_val = str(query_input).strip()

        cache_key = f"icici:{raw_val.lower()}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        local_path = os.path.join(BASE_DIR, "icici_pvtcar_master.parquet")
        target_url = local_path if os.path.exists(local_path) else "hf://datasets/ItsXyron/SBI-Master-DB/icici_pvtcar_master.parquet"

        cols = ["Manufacture", "VehicleModel", "VehicleSubClassDesc", "CubicCapacity", "SeatingCapacity", "FuelType", "CarCategory", "ExShowroomPrice"]
        q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE LOWER(VehicleModel) LIKE '%{raw_val.lower()}%' OR LOWER(Manufacture) LIKE '%{raw_val.lower()}%' LIMIT 15"

        records = self._execute_query(q, cols)
        res = {
            "status": "success" if records else "not_found",
            "database": "ICICI Auto Insurance Master",
            "query": raw_val,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 10h. Unified Xyron Intel Master (43K Combined Banking, Muthoot, OLX, POSP)
    # ------------------------------------------------------------------
    def query_xyron_master(self, query_input: Any) -> Dict[str, Any]:
        t0 = time.time()
        raw_val = str(query_input).strip()
        clean_phone = self.sanitize_phone(raw_val)

        cache_key = f"xyron_master:{clean_phone or raw_val.lower()}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        local_path = os.path.join(BASE_DIR, "xyron_master.parquet")
        target_url = local_path if os.path.exists(local_path) else "hf://datasets/ItsXyron/SBI-Master-DB/xyron_intel_master.parquet"

        cols = ["phone", "name", "email", "aadhaar", "pan", "address", "city", "state", "pincode", "bank_name", "account_number", "ifsc_code", "category", "source_table"]

        if clean_phone:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE phone = '{clean_phone}' OR clean_num = {clean_phone} LIMIT 10"
        elif len(raw_val) == 12 and raw_val.isdigit():
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE aadhaar = '{raw_val}' LIMIT 10"
        elif len(raw_val) == 10 and raw_val.isalnum() and not raw_val.isdigit():
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE UPPER(pan) = '{raw_val.upper()}' LIMIT 10"
        elif "@" in raw_val:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE LOWER(email) = '{raw_val.lower()}' LIMIT 10"
        else:
            q = f"SELECT {', '.join(cols)} FROM '{target_url}' WHERE LOWER(name) LIKE '%{raw_val.lower()}%' LIMIT 10"

        records = self._execute_query(q, cols)
        res = {
            "status": "success" if records else "not_found",
            "database": "Unified Xyron Intel Master",
            "query": raw_val,
            "count": len(records),
            "records": records,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }
        with self.cache_lock:
            self.cache[cache_key] = res
        return res

    # ------------------------------------------------------------------
    # 🌟 11. UNIFIED COMBINED SEARCH ACROSS ALL 14 MASTER DATABASES
    # ------------------------------------------------------------------
    def query_combined(self, query_input: str) -> Dict[str, Any]:
        """
        Executes parallel lookups across Truecaller, Airtel, HDFC, SBI, BSNL, Paytm, 
        Flipkart, Instagram, and Telecom DB for a phone number.
        Returns unified consolidated profile + granular database hits.
        """
        t0 = time.time()
        clean = str(query_input).strip()
        
        # Check if 10-digit mobile
        mobile = self.sanitize_phone(clean)
        if not mobile:
            # Fallback to Telegram or Aadhaar or Instagram username
            if clean.isdigit() and len(clean) == 12:
                return self.query_by_aadhar(clean)
            elif clean.isdigit() and 5 <= len(clean) <= 16:
                return self.query_by_telegram(clean)
            else:
                return self.query_instagram(clean)

        cache_key = f"combined:{mobile}"
        with self.cache_lock:
            if cache_key in self.cache:
                return {**self.cache[cache_key], "cached": True, "latency_ms": round((time.time() - t0) * 1000, 2)}

        # Run queries across all phone-based databases concurrently
        futures = {
            "truecaller": self.executor.submit(self.query_truecaller, mobile),
            "airtel": self.executor.submit(self.query_airtel, mobile),
            "hdfc": self.executor.submit(self.query_hdfc, mobile),
            "sbi": self.executor.submit(self.query_sbi, str(mobile)),
            "ind": self.executor.submit(self.query_ind, str(mobile)),
            "bsnl": self.executor.submit(self.query_bsnl, mobile),
            "paytm": self.executor.submit(self.query_paytm, mobile),
            "flipkart": self.executor.submit(self.query_flipkart, mobile),
            "instagram": self.executor.submit(self.query_instagram, str(mobile)),
            "telecom_db": self.executor.submit(self.query_by_phone, str(mobile)),
            "muthoot": self.executor.submit(self.query_muthoot, str(mobile)),
            "olx": self.executor.submit(self.query_olx, str(mobile)),
            "xyron_master": self.executor.submit(self.query_xyron_master, str(mobile))
        }

        results = {}
        for source, fut in futures.items():
            try:
                results[source] = fut.result(timeout=10)
            except Exception as e:
                results[source] = {"status": "error", "message": str(e)}

        # Build consolidated profile intelligence
        names = []
        emails = []
        carriers = []
        locations = []
        cibil = None
        highmark = None
        instagram_profiles = []

        # Helper to extract unique strings
        def _add_unique(lst: List[str], val: Any):
            if val and isinstance(val, str):
                s = val.strip()
                if s and s.upper() not in ["NONE", "NULL", "UNKNOWN", "0"] and s not in lst:
                    lst.append(s)

        # 1. Parse Truecaller
        for r in results.get("truecaller", {}).get("records", []):
            _add_unique(names, r.get("name"))
            _add_unique(emails, r.get("email"))
            _add_unique(carriers, r.get("carrier"))
            _add_unique(locations, r.get("address"))

        # 2. Parse Airtel
        for r in results.get("airtel", {}).get("records", []):
            _add_unique(names, r.get("name"))
            _add_unique(emails, r.get("email"))
            _add_unique(carriers, r.get("carrier"))
            _add_unique(locations, r.get("address"))

        # 3. Parse HDFC Bank
        for r in results.get("hdfc", {}).get("records", []):
            _add_unique(names, r.get("name"))
            _add_unique(emails, r.get("email"))
            loc = ", ".join(filter(None, [r.get("city"), r.get("state"), r.get("pincode")]))
            _add_unique(locations, loc)
            if r.get("cibil"):
                cibil = r.get("cibil")
            if r.get("highmark_score"):
                highmark = r.get("highmark_score")

        # 4. Parse BSNL
        for r in results.get("bsnl", {}).get("records", []):
            _add_unique(names, r.get("name"))
            _add_unique(emails, r.get("email"))
            _add_unique(carriers, r.get("carrier"))
            _add_unique(locations, r.get("address"))

        # 5. Parse Paytm
        for r in results.get("paytm", {}).get("records", []):
            _add_unique(names, r.get("name"))
            _add_unique(emails, r.get("email"))
            loc = ", ".join(filter(None, [r.get("city"), r.get("state")]))
            _add_unique(locations, loc)

        # 6. Parse Flipkart
        for r in results.get("flipkart", {}).get("records", []):
            _add_unique(names, r.get("name"))
            _add_unique(emails, r.get("email"))
            loc = ", ".join(filter(None, [r.get("address"), r.get("city"), r.get("state"), r.get("pincode")]))
            _add_unique(locations, loc)

        # 7. Parse Instagram
        for r in results.get("instagram", {}).get("records", []):
            _add_unique(names, r.get("name"))
            _add_unique(emails, r.get("email"))
            if r.get("username"):
                instagram_profiles.append(r.get("username"))

        # 8. Parse Telecom DB
        for r in results.get("telecom_db", {}).get("records", []):
            _add_unique(names, r.get("name"))
            _add_unique(locations, r.get("address"))

        # 9. Parse SBI POSP & Indian Banking DB
        for r in results.get("sbi", {}).get("records", []):
            _add_unique(names, r.get("name"))
            _add_unique(emails, r.get("email"))
            loc = ", ".join(filter(None, [r.get("address"), r.get("city"), r.get("state"), r.get("pincode")]))
            _add_unique(locations, loc)

        # 10. Parse IND Admissions & Education Master DB
        for r in results.get("ind", {}).get("records", []):
            _add_unique(names, r.get("name"))
            _add_unique(emails, r.get("email"))
            loc = ", ".join(filter(None, [r.get("address"), r.get("city"), r.get("state"), r.get("pincode")]))
            _add_unique(locations, loc)

        # 11. Parse Muthoot Finance KYC Master DB
        for r in results.get("muthoot", {}).get("records", []):
            m_name = f"{r.get('first_name') or ''} {r.get('last_name') or ''}".strip()
            _add_unique(names, m_name)
            _add_unique(emails, r.get("email_id"))
            loc = ", ".join(filter(None, [r.get("address_1"), r.get("city"), r.get("state"), r.get("pincode")]))
            _add_unique(locations, loc)

        # 12. Parse OLX India Merchants Master DB
        for r in results.get("olx", {}).get("records", []):
            _add_unique(names, r.get("username"))
            _add_unique(emails, r.get("email"))
            loc = ", ".join(filter(None, [r.get("address"), r.get("city"), r.get("state"), r.get("pincode")]))
            _add_unique(locations, loc)

        # 13. Parse Unified Xyron Intel Master DB
        for r in results.get("xyron_master", {}).get("records", []):
            _add_unique(names, r.get("name"))
            _add_unique(emails, r.get("email"))
            loc = ", ".join(filter(None, [r.get("address"), r.get("city"), r.get("state"), r.get("pincode")]))
            _add_unique(locations, loc)

        # Calculate hits count
        hits = 0
        for src, data in results.items():
            if data.get("status") == "success" or data.get("success") is True or data.get("count", 0) > 0:
                hits += 1

        response = {
            "status": "success" if hits > 0 else "not_found",
            "query": str(mobile),
            "formatted_phone": f"+91 {mobile}",
            "total_sources_hit": hits,
            "total_sources_queried": len(results),
            "consolidated_summary": {
                "names_found": names,
                "primary_name": names[0] if names else None,
                "emails_found": emails,
                "carriers_identified": carriers,
                "locations_identified": locations,
                "cibil_score": cibil,
                "highmark_score": highmark,
                "instagram_usernames": list(set(instagram_profiles))
            },
            "sources": results,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }

        with self.cache_lock:
            self.cache[cache_key] = response

        return response

    # =========================================================================
    # UPI & VPA 360° RESOLVER SUITE
    # =========================================================================
    def resolve_upi_to_num(self, vpa: str) -> Dict[str, Any]:
        """
        Resolves a UPI ID / VPA (e.g. 9826287975@paytm, 9769009011@okhdfcbank) to:
        - 10-digit Indian Mobile Number
        - PSP Application (Google Pay, PhonePe, Paytm, BHIM, etc.)
        - Acquiring/Sponsor Bank & IFSC prefix
        - Handle categorization (phone-based vs custom alias)
        """
        raw = str(vpa).strip().lower()
        if not raw:
            return {"status": "error", "message": "Empty UPI ID / VPA provided"}

        handle_suffix = ""
        prefix = raw
        if "@" in raw:
            parts = raw.split("@", 1)
            prefix = parts[0].strip()
            handle_suffix = parts[1].strip()

        # Extract 10-digit Indian phone number (starting 6, 7, 8, or 9)
        phone_match = re.search(r'(?:(?:\+?91)|0)?([6-9]\d{9})', prefix)
        resolved_phone = phone_match.group(1) if phone_match else None

        psp_meta = UPI_HANDLES_MAP.get(handle_suffix, {
            "app": "UPI PSP Gateway",
            "bank": "Scheduled Indian Bank",
            "ifsc": "UPI",
            "type": "Standard VPA"
        })

        if resolved_phone:
            return {
                "status": "success",
                "upi_id": raw,
                "resolved_phone": resolved_phone,
                "formatted_phone": f"+91 {resolved_phone}",
                "handle_type": "phone_based",
                "handle_suffix": f"@{handle_suffix}" if handle_suffix else "@upi",
                "psp_app": psp_meta.get("app"),
                "linked_bank": psp_meta.get("bank"),
                "ifsc_prefix": psp_meta.get("ifsc"),
                "account_type": psp_meta.get("type"),
                "payment_intent": f"upi://pay?pa={raw}&pn={resolved_phone}",
                "confidence": "VERY HIGH (Direct Phone VPA Extraction)"
            }

        # If prefix doesn't directly contain a 10-digit phone number (e.g. name@okhdfcbank)
        matched_phone = None
        matched_name = None

        # Check IND admissions by email prefix or name
        ind_res = self.query_ind(prefix)
        if ind_res.get("status") == "success" and ind_res.get("records"):
            rec = ind_res["records"][0]
            matched_phone = rec.get("mobile")
            matched_name = rec.get("name")

        if not matched_phone:
            # Check Truecaller by name
            tc_res = self.query_truecaller(prefix)
            if tc_res.get("status") == "success" and tc_res.get("records"):
                rec = tc_res["records"][0]
                matched_phone = rec.get("phone") or rec.get("mobile")
                matched_name = rec.get("name")

        if matched_phone:
            clean_match = "".join(c for c in str(matched_phone) if c.isdigit())[-10:]
            return {
                "status": "success",
                "upi_id": raw,
                "resolved_phone": clean_match,
                "formatted_phone": f"+91 {clean_match}",
                "handle_type": "custom_alias_cross_referenced",
                "handle_suffix": f"@{handle_suffix}" if handle_suffix else "@upi",
                "matched_name": matched_name,
                "psp_app": psp_meta.get("app"),
                "linked_bank": psp_meta.get("bank"),
                "ifsc_prefix": psp_meta.get("ifsc"),
                "payment_intent": f"upi://pay?pa={raw}&pn={clean_match}",
                "confidence": "HIGH (Cross-Referenced Identity Match)"
            }

        return {
            "status": "partial_match",
            "upi_id": raw,
            "resolved_phone": None,
            "handle_type": "custom_name_alias",
            "handle_suffix": f"@{handle_suffix}" if handle_suffix else "@upi",
            "psp_app": psp_meta.get("app"),
            "linked_bank": psp_meta.get("bank"),
            "ifsc_prefix": psp_meta.get("ifsc"),
            "message": f"Identified {psp_meta.get('app')} handle linked to {psp_meta.get('bank')}. Custom alias prefix '{prefix}' could not be resolved to a single mobile number without further KYC context."
        }

    def resolve_num_to_upi(self, phone: str) -> Dict[str, Any]:
        """
        Given a 10-digit Indian phone number, generates and prioritizes all active
        UPI VPAs across PhonePe, Google Pay, Paytm, BHIM, Amazon Pay, CRED, and WhatsApp Pay,
        cross-referenced with the user's detected banking relationships.
        """
        clean_phone = "".join(c for c in str(phone) if c.isdigit())
        if len(clean_phone) == 12 and clean_phone.startswith("91"):
            clean_phone = clean_phone[2:]
        elif len(clean_phone) == 11 and clean_phone.startswith("0"):
            clean_phone = clean_phone[1:]

        if len(clean_phone) != 10 or clean_phone[0] not in "6789":
            return {"status": "error", "message": f"Invalid Indian phone number: '{phone}'. Must be a 10-digit mobile."}

        # Cross-reference banking databases to detect bank relationships
        detected_banks = []
        sbi_hit = False
        hdfc_hit = False
        paytm_hit = False

        # Fast parallel checks across banking sources
        with ThreadPoolExecutor(max_workers=3) as executor:
            f_sbi = executor.submit(self.query_sbi, clean_phone)
            f_hdfc = executor.submit(self.query_hdfc, clean_phone)
            f_paytm = executor.submit(self.query_paytm, clean_phone)

            try:
                r_sbi = f_sbi.result()
                if r_sbi.get("status") == "success" and r_sbi.get("count", 0) > 0:
                    detected_banks.append("State Bank of India (SBI)")
                    sbi_hit = True
            except Exception:
                pass

            try:
                r_hdfc = f_hdfc.result()
                if r_hdfc.get("status") == "success" and r_hdfc.get("count", 0) > 0:
                    detected_banks.append("HDFC Bank")
                    hdfc_hit = True
            except Exception:
                pass

            try:
                r_paytm = f_paytm.result()
                if r_paytm.get("status") == "success" and r_paytm.get("count", 0) > 0:
                    detected_banks.append("Paytm Payments Bank")
                    paytm_hit = True
            except Exception:
                pass

        # Candidate handles
        # Priority order: PhonePe, Google Pay, Paytm, BHIM, Amazon Pay, WhatsApp, CRED
        candidate_handles = [
            # PhonePe (Highest UPI market share ~48%)
            ("ybl", "PhonePe", "YES Bank", "VERY HIGH (Primary PhonePe Default)"),
            ("ibl", "PhonePe", "ICICI Bank", "HIGH (Secondary PhonePe Default)"),
            ("axl", "PhonePe", "Axis Bank", "MEDIUM (Multi-Bank Failover)"),
            # Paytm
            ("paytm", "Paytm", "Paytm Payments / Partner Bank", "VERY HIGH (Confirmed Registered User)" if paytm_hit else "HIGH (Standard Paytm Default)"),
            ("ptsbi", "Paytm", "State Bank of India", "VERY HIGH (Detected SBI Account)" if sbi_hit else "MEDIUM"),
            ("pthdfc", "Paytm", "HDFC Bank", "VERY HIGH (Detected HDFC Account)" if hdfc_hit else "MEDIUM"),
            # Google Pay (~37% market share)
            ("oksbi", "Google Pay", "State Bank of India", "VERY HIGH (Detected SBI Account)" if sbi_hit else "HIGH (Standard GPay Default)"),
            ("okhdfcbank", "Google Pay", "HDFC Bank", "VERY HIGH (Detected HDFC Account)" if hdfc_hit else "HIGH (Standard GPay Default)"),
            ("okaxis", "Google Pay", "Axis Bank", "HIGH (Standard GPay Default)"),
            ("okicici", "Google Pay", "ICICI Bank", "HIGH (Standard GPay Default)"),
            # BHIM National Switch
            ("upi", "BHIM / NPCI", "NPCI Unified Platform", "HIGH (NPCI National Switch Default)"),
            # Amazon Pay
            ("apl", "Amazon Pay", "Axis Bank", "MEDIUM (Amazon Pay Registered)"),
            # WhatsApp Pay
            ("waicici", "WhatsApp Pay", "ICICI Bank", "MEDIUM (WhatsApp Payments Active)"),
            # CRED
            ("cred", "CRED", "Axis Bank", "MEDIUM (CRED Members Only)")
        ]

        upi_handles = []
        for suffix, app, bank, conf in candidate_handles:
            vpa = f"{clean_phone}@{suffix}"
            upi_handles.append({
                "vpa": vpa,
                "app": app,
                "bank": bank,
                "handle_suffix": f"@{suffix}",
                "active_confidence": conf,
                "qr_intent": f"upi://pay?pa={vpa}&pn={clean_phone}"
            })

        if sbi_hit:
            primary_vpa = f"{clean_phone}@oksbi"
        elif hdfc_hit:
            primary_vpa = f"{clean_phone}@okhdfcbank"
        elif paytm_hit:
            primary_vpa = f"{clean_phone}@paytm"
        else:
            primary_vpa = f"{clean_phone}@ybl"

        return {
            "status": "success",
            "phone": clean_phone,
            "formatted_phone": f"+91 {clean_phone}",
            "detected_banking_relationships": detected_banks,
            "total_upi_handles_generated": len(upi_handles),
            "primary_recommended_vpa": primary_vpa,
            "primary_qr_payload": f"upi://pay?pa={primary_vpa}&pn={clean_phone}",
            "upi_handles": upi_handles
        }

    def query_upi_full_info(self, query_input: str) -> Dict[str, Any]:
        """
        Deep 360° UPI KYC & OSINT Intelligence Lookup:
        - Resolves UPI ID to mobile number or normalizes 10-digit mobile.
        - Resolves acquiring bank, PSP app, and IFSC code.
        - Cross-queries all 11 Master DBs (Telecom 2.5B, Truecaller 285M, Paytm 2.95M,
          HDFC Bank 11.3M, SBI POSP 43K, IND Admissions 686K, BSNL 18.1M, Airtel 73M).
        - Returns consolidated KYC dossier: Primary Legal Name, Aliases, Primary & Alternate
          Mobile numbers, Emails, Addresses, Aadhaar & PAN (if found), Bank Accounts,
          Carrier, Circle, and all Active UPI Handles.
        """
        t0 = time.time()
        raw = str(query_input).strip()
        if not raw:
            return {"status": "error", "message": "Empty query provided"}

        cache_key = f"upi_full:{raw.lower()}"
        with self.cache_lock:
            cached = self.cache.get(cache_key)
            if cached:
                cached["cached"] = True
                return cached

        # Step 1: Resolve UPI metadata and phone
        is_vpa = "@" in raw
        vpa_meta = None
        target_phone = None

        if is_vpa:
            vpa_meta = self.resolve_upi_to_num(raw)
            target_phone = vpa_meta.get("resolved_phone")
        else:
            clean_digits = "".join(c for c in raw if c.isdigit())
            if len(clean_digits) == 12 and clean_digits.startswith("91"):
                clean_digits = clean_digits[2:]
            elif len(clean_digits) == 11 and clean_digits.startswith("0"):
                clean_digits = clean_digits[1:]
            if len(clean_digits) == 10 and clean_digits[0] in "6789":
                target_phone = clean_digits

        if not target_phone:
            return {
                "status": "partial_vpa_resolved",
                "query": raw,
                "vpa_intelligence": vpa_meta or {},
                "message": f"Resolved UPI PSP handle details for '{raw}', but phone number could not be extracted without additional KYC parameters."
            }

        # Step 2: Query full 360° master database across all 11 engines
        combined_data = self.query_combined(target_phone)

        # Step 3: Enumerate active UPI handles for this phone
        handles_info = self.resolve_num_to_upi(target_phone)

        # Step 4: Extract and synthesize KYC intelligence
        consolidated = combined_data.get("consolidated_summary", {})
        names = consolidated.get("names_found", [])
        emails = consolidated.get("emails_found", [])
        locations = consolidated.get("locations_identified", [])
        carriers = consolidated.get("carriers_identified", [])
        cibil = consolidated.get("cibil_score")
        highmark = consolidated.get("highmark_score")
        instagram = consolidated.get("instagram_usernames", [])

        # Extract PAN & Aadhaar from banking and student sources
        pan_numbers = []
        aadhaar_numbers = []
        bank_accounts = []

        sources = combined_data.get("sources", {})

        # From SBI
        for r in sources.get("sbi", {}).get("records", []):
            if r.get("pan"):
                _p = str(r["pan"]).strip().upper()
                if _p not in pan_numbers and len(_p) == 10:
                    pan_numbers.append(_p)
            if r.get("aadhaar"):
                _a = str(r["aadhaar"]).strip()
                if _a not in aadhaar_numbers:
                    aadhaar_numbers.append(_a)
            if r.get("bank"):
                _b = str(r["bank"]).strip()
                if _b not in bank_accounts:
                    bank_accounts.append(_b)

        # From HDFC
        for r in sources.get("hdfc", {}).get("records", []):
            if r.get("account_number") or r.get("acc_no"):
                acc = str(r.get("account_number") or r.get("acc_no")).strip()
                masked_acc = f"HDFC Bank (XXXX{acc[-4:]})" if len(acc) >= 4 else "HDFC Bank"
                if masked_acc not in bank_accounts:
                    bank_accounts.append(masked_acc)
            elif "HDFC Bank" not in bank_accounts:
                bank_accounts.append("HDFC Bank")

        # From IND Admissions
        for r in sources.get("ind", {}).get("records", []):
            if r.get("aadhaar"):
                _a = str(r["aadhaar"]).strip()
                if _a not in aadhaar_numbers:
                    aadhaar_numbers.append(_a)

        # From Paytm
        if sources.get("paytm", {}).get("records") or sources.get("paytm", {}).get("count", 0) > 0:
            if "Paytm Payments Bank / Wallet" not in bank_accounts:
                bank_accounts.append("Paytm Payments Bank / Wallet")

        primary_name = names[0] if names else None

        response = {
            "status": "success",
            "query": raw,
            "query_type": "vpa_address" if is_vpa else "phone_number",
            "resolved_phone": target_phone,
            "formatted_phone": f"+91 {target_phone}",
            "upi_metadata": vpa_meta if vpa_meta else {
                "primary_vpa": handles_info.get("primary_recommended_vpa"),
                "psp_app": "Auto-Discovered UPI Ecosystem",
                "linked_bank": bank_accounts[0] if bank_accounts else "Indian Scheduled Bank"
            },
            "kyc_dossier": {
                "primary_legal_name": primary_name,
                "all_names_identified": names,
                "primary_phone": target_phone,
                "formatted_phone": f"+91 {target_phone}",
                "email_addresses": emails,
                "pan_numbers": pan_numbers,
                "aadhaar_numbers": aadhaar_numbers,
                "residential_addresses": locations,
                "cibil_score": cibil,
                "highmark_score": highmark,
                "instagram_profiles": instagram
            },
            "banking_intelligence": {
                "detected_banking_institutions": handles_info.get("detected_banking_relationships", []) or bank_accounts,
                "linked_bank_accounts": bank_accounts,
                "primary_recommended_vpa": handles_info.get("primary_recommended_vpa"),
                "qr_payment_intent": handles_info.get("primary_qr_payload"),
                "total_active_upi_handles": handles_info.get("total_upi_handles_generated", 0),
                "top_active_upi_handles": handles_info.get("upi_handles", [])[:8]
            },
            "telecom_profile": {
                "carriers": carriers,
                "locations": locations
            },
            "total_databases_hit": combined_data.get("total_sources_hit", 0),
            "sources": sources,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }

        with self.cache_lock:
            self.cache[cache_key] = response

        return response

    def query_upi(self, query_input: str, mode: str = "full") -> Dict[str, Any]:
        """Unified entry point for UPI lookups."""
        raw = str(query_input).strip()
        if mode == "upi_to_num":
            return self.resolve_upi_to_num(raw)
        elif mode == "num_to_upi":
            return self.resolve_num_to_upi(raw)
        else:
            return self.query_upi_full_info(raw)

    # ------------------------------------------------------------------
    # 🌟 13. 🔥 360° DEEP INTELLIGENCE DOSSIER ENGINE (ALL INDIAN MASTER DBS)
    # ------------------------------------------------------------------
    def query_deep_dossier(self, query_input: Any) -> Dict[str, Any]:
        """
        India's Most Advanced Deep Intelligence Dossier Engine.
        Accepts any Indian identifier (Phone, Aadhaar, PAN, UPI, Vehicle, Telegram, Email, Name)
        and builds a complete cross-correlated dossier across all 16 Master databases.
        """
        t0 = time.time()
        raw = str(query_input).strip()
        if not raw:
            return {"status": "error", "message": "Empty query provided"}

        cache_key = f"deep_dossier:{raw.lower()}"
        with self.cache_lock:
            if cache_key in self.cache:
                cached = dict(self.cache[cache_key])
                cached["cached"] = True
                cached["latency_ms"] = round((time.time() - t0) * 1000, 2)
                return cached

        # Step 1: Detect Primary Target Anchor
        clean_digits = re.sub(r'\D', '', raw)
        clean_upper = raw.upper().strip()
        
        target_phone = None
        target_aadhar = None
        target_pan = None
        target_vpa = None
        target_vehicle = None
        target_tg_id = None
        target_name = None
        target_email = None

        if "@" in raw and "." in raw.split("@")[-1] and not any(h in raw.lower() for h in ["ok", "ybl", "ibl", "axl", "upi", "paytm", "apl"]):
            target_email = raw.lower()
        elif "@" in raw:
            target_vpa = raw.lower()
        elif len(clean_digits) == 12:
            target_aadhar = clean_digits
        elif len(clean_digits) == 10 and clean_digits[0] in "6789":
            target_phone = clean_digits
        elif re.match(r'^[A-Z]{5}[0-9]{4}[A-Z]$', clean_upper):
            target_pan = clean_upper
        elif re.match(r'^[A-Z]{2}[0-9]{1,2}[A-Z]{0,3}[0-9]{4}$', clean_upper):
            target_vehicle = clean_upper
        elif clean_digits and 5 <= len(clean_digits) <= 16 and not raw.startswith("+"):
            target_tg_id = int(clean_digits)
        else:
            target_name = raw

        # Anchor Resolution: If target is UPI, vehicle, Aadhaar, or Telegram, resolve to phone/owner
        resolved_vehicle_info = None
        if target_vpa:
            v_res = self.resolve_upi_to_num(target_vpa)
            if v_res.get("resolved_phone"):
                target_phone = v_res["resolved_phone"]

        if target_vehicle and not target_phone:
            try:
                from external_services import ServiceGateway
                v_res = ServiceGateway.get_veh_to_num(target_vehicle)
                if isinstance(v_res, dict) and v_res.get("phone"):
                    target_phone = v_res["phone"]
                rc_res = ServiceGateway.get_vehicle_rc(target_vehicle)
                if isinstance(rc_res, dict) and rc_res.get("status") == "success":
                    resolved_vehicle_info = rc_res
            except Exception:
                pass

        if target_aadhar and not target_phone:
            a_res = self.query_by_aadhar(target_aadhar)
            for r in a_res.get("records", []):
                if r.get("phoneNumber") and len(re.sub(r'\D', '', str(r["phoneNumber"]))) == 10:
                    target_phone = re.sub(r'\D', '', str(r["phoneNumber"]))
                    break

        if target_tg_id and not target_phone:
            tg_res = self.query_by_telegram(target_tg_id)
            if tg_res.get("num") or tg_res.get("phone"):
                raw_ph = tg_res.get("num") or tg_res.get("phone")
                clean_ph = re.sub(r'\D', '', str(raw_ph))
                if len(clean_ph) == 12 and clean_ph.startswith("91"):
                    clean_ph = clean_ph[2:]
                if len(clean_ph) == 10 and clean_ph[0] in "6789":
                    target_phone = clean_ph

        # Step 2: Concurrent Multi-Database Sweep across ALL 16 databases
        futures = {}
        if target_phone:
            futures["telecom_db"] = self.executor.submit(self.query_by_phone, target_phone)
            futures["truecaller"] = self.executor.submit(self.query_truecaller, int(target_phone))
            futures["airtel"] = self.executor.submit(self.query_airtel, int(target_phone))
            futures["bsnl"] = self.executor.submit(self.query_bsnl, int(target_phone))
            futures["hdfc"] = self.executor.submit(self.query_hdfc, int(target_phone))
            futures["sbi"] = self.executor.submit(self.query_sbi, target_phone)
            futures["paytm"] = self.executor.submit(self.query_paytm, int(target_phone))
            futures["flipkart"] = self.executor.submit(self.query_flipkart, int(target_phone))
            futures["ind"] = self.executor.submit(self.query_ind, target_phone)
            futures["instagram"] = self.executor.submit(self.query_instagram, target_phone)
            futures["muthoot"] = self.executor.submit(self.query_muthoot, target_phone)
            futures["olx"] = self.executor.submit(self.query_olx, target_phone)
            futures["xyron_master"] = self.executor.submit(self.query_xyron_master, target_phone)
            futures["family"] = self.executor.submit(self.query_by_aadhar_family, target_phone)
            futures["upi"] = self.executor.submit(self.resolve_num_to_upi, target_phone)
        elif target_aadhar:
            futures["telecom_db"] = self.executor.submit(self.query_by_aadhar, target_aadhar)
            futures["sbi"] = self.executor.submit(self.query_sbi, target_aadhar)
            futures["ind"] = self.executor.submit(self.query_ind, target_aadhar)
            futures["muthoot"] = self.executor.submit(self.query_muthoot, target_aadhar)
            futures["olx"] = self.executor.submit(self.query_olx, target_aadhar)
            futures["xyron_master"] = self.executor.submit(self.query_xyron_master, target_aadhar)
            futures["family"] = self.executor.submit(self.query_by_aadhar_family, target_aadhar)
        elif target_pan:
            futures["sbi"] = self.executor.submit(self.query_sbi, target_pan)
            futures["ind"] = self.executor.submit(self.query_ind, target_pan)
            futures["muthoot"] = self.executor.submit(self.query_muthoot, target_pan)
            futures["olx"] = self.executor.submit(self.query_olx, target_pan)
            futures["xyron_master"] = self.executor.submit(self.query_xyron_master, target_pan)
        else:
            futures["sbi"] = self.executor.submit(self.query_sbi, raw)
            futures["ind"] = self.executor.submit(self.query_ind, raw)
            futures["muthoot"] = self.executor.submit(self.query_muthoot, raw)
            futures["olx"] = self.executor.submit(self.query_olx, raw)
            futures["xyron_master"] = self.executor.submit(self.query_xyron_master, raw)
            futures["instagram"] = self.executor.submit(self.query_instagram, raw)
            futures["twitter"] = self.executor.submit(self.query_twitter, raw)
            if target_tg_id:
                futures["telegram"] = self.executor.submit(self.query_by_telegram, target_tg_id)

        raw_results = {}
        for k, fut in futures.items():
            try:
                raw_results[k] = fut.result(timeout=10)
            except Exception as e:
                raw_results[k] = {"status": "error", "message": str(e)}

        # Step 3: Synthesis of Dossier Sections
        names: List[str] = []
        father_names: List[str] = []
        mother_names: List[str] = []
        dob_list: List[str] = []
        qualifications: List[str] = []
        emails: List[str] = []
        addresses: List[str] = []
        carriers: List[str] = []
        cibil_score = None
        highmark_score = None
        pan_cards: List[str] = []
        aadhaars: List[str] = []
        bank_accounts: List[str] = []
        colleges: List[str] = []
        ig_handles: List[str] = []
        tw_handles: List[str] = []
        tg_info = None

        def _clean_add(lst: List[str], val: Any):
            if val and isinstance(val, str):
                s = val.strip().replace('="', '').replace('"', '')
                if s and s.upper() not in ["NONE", "NULL", "UNKNOWN", "0", "-", "—"] and s not in lst:
                    lst.append(s)

        # Parse Muthoot
        for r in raw_results.get("muthoot", {}).get("records", []):
            m_fn = r.get("first_name") or ""
            m_ln = r.get("last_name") or ""
            _clean_add(names, f"{m_fn} {m_ln}".strip())
            _clean_add(father_names, r.get("fathers_name"))
            _clean_add(mother_names, r.get("mothers_name"))
            _clean_add(emails, r.get("email_id"))
            _clean_add(dob_list, r.get("dob"))
            _clean_add(qualifications, r.get("educational_qualification"))
            if r.get("aadhaar_no"):
                _clean_add(aadhaars, str(r["aadhaar_no"]))
            if r.get("pan"):
                _clean_add(pan_cards, str(r["pan"]).upper())
            loc = ", ".join(filter(None, [r.get("address_1"), r.get("address_2"), r.get("city"), r.get("state"), r.get("pincode")]))
            _clean_add(addresses, loc)
            if r.get("bank_name"):
                b_acc = f"{r['bank_name']} (A/C: {r.get('bank_account_number', 'N/A')})"
                _clean_add(bank_accounts, b_acc)

        # Parse IND Admissions
        for r in raw_results.get("ind", {}).get("records", []):
            _clean_add(names, r.get("name"))
            _clean_add(father_names, r.get("father_name"))
            _clean_add(mother_names, r.get("mother_name"))
            _clean_add(emails, r.get("email"))
            _clean_add(dob_list, r.get("dob"))
            _clean_add(colleges, r.get("college_name"))
            if r.get("aadhaar"):
                _clean_add(aadhaars, str(r["aadhaar"]))
            if r.get("pan"):
                _clean_add(pan_cards, str(r["pan"]).upper())
            loc = ", ".join(filter(None, [r.get("address"), r.get("city"), r.get("state"), r.get("pincode")]))
            _clean_add(addresses, loc)
            if r.get("bank_name"):
                b_acc = f"{r['bank_name']} ({r.get('account_number', '')})"
                _clean_add(bank_accounts, b_acc)

        # Parse SBI POSP
        for r in raw_results.get("sbi", {}).get("records", []):
            _clean_add(names, r.get("name"))
            _clean_add(father_names, r.get("father_name"))
            _clean_add(emails, r.get("email"))
            if r.get("aadhaar"):
                _clean_add(aadhaars, str(r["aadhaar"]))
            if r.get("pan"):
                _clean_add(pan_cards, str(r["pan"]).upper())
            loc = ", ".join(filter(None, [r.get("address"), r.get("city"), r.get("state"), r.get("pincode")]))
            _clean_add(addresses, loc)
            if r.get("bank_name"):
                b_acc = f"{r['bank_name']} ({r.get('account_number', '')})"
                _clean_add(bank_accounts, b_acc)

        # Parse OLX Merchants
        for r in raw_results.get("olx", {}).get("records", []):
            _clean_add(names, r.get("username"))
            _clean_add(emails, r.get("email"))
            if r.get("aadhaar"):
                _clean_add(aadhaars, str(r["aadhaar"]))
            if r.get("pan"):
                _clean_add(pan_cards, str(r["pan"]).upper())
            loc = ", ".join(filter(None, [r.get("address"), r.get("city"), r.get("state"), r.get("pincode")]))
            _clean_add(addresses, loc)

        # Parse Unified Xyron Master
        for r in raw_results.get("xyron_master", {}).get("records", []):
            _clean_add(names, r.get("name"))
            _clean_add(emails, r.get("email"))
            if r.get("aadhaar"):
                _clean_add(aadhaars, str(r["aadhaar"]))
            if r.get("pan"):
                _clean_add(pan_cards, str(r["pan"]).upper())
            loc = ", ".join(filter(None, [r.get("address"), r.get("city"), r.get("state"), r.get("pincode")]))
            _clean_add(addresses, loc)
            if r.get("bank_name"):
                _clean_add(bank_accounts, r.get("bank_name"))

        # Parse Telecom DB
        for r in raw_results.get("telecom_db", {}).get("records", []):
            _clean_add(names, r.get("name"))
            _clean_add(father_names, r.get("fathersName"))
            _clean_add(addresses, r.get("address"))
            if r.get("aadharNumber"):
                _clean_add(aadhaars, str(r["aadharNumber"]))

        # Parse Truecaller
        for r in raw_results.get("truecaller", {}).get("records", []):
            _clean_add(names, r.get("name"))
            _clean_add(emails, r.get("email"))
            _clean_add(carriers, r.get("carrier"))
            _clean_add(addresses, r.get("address"))

        # Parse Airtel
        for r in raw_results.get("airtel", {}).get("records", []):
            _clean_add(names, r.get("name"))
            _clean_add(emails, r.get("email"))
            _clean_add(carriers, "Airtel India")
            _clean_add(addresses, r.get("address"))

        # Parse HDFC
        for r in raw_results.get("hdfc", {}).get("records", []):
            _clean_add(names, r.get("name"))
            _clean_add(emails, r.get("email"))
            _clean_add(dob_list, r.get("dob"))
            loc = ", ".join(filter(None, [r.get("city"), r.get("state"), r.get("pincode")]))
            _clean_add(addresses, loc)
            if r.get("cibil"):
                cibil_score = r.get("cibil")
            if r.get("highmark_score"):
                highmark_score = r.get("highmark_score")
            _clean_add(bank_accounts, "HDFC Bank")

        # Parse Paytm
        for r in raw_results.get("paytm", {}).get("records", []):
            _clean_add(names, r.get("name"))
            _clean_add(emails, r.get("email"))
            _clean_add(dob_list, r.get("dob"))
            if r.get("pan"):
                _clean_add(pan_cards, str(r["pan"]).upper())
            loc = ", ".join(filter(None, [r.get("city"), r.get("state")]))
            _clean_add(addresses, loc)
            _clean_add(bank_accounts, "Paytm Payments Bank / Wallet")

        # Parse Flipkart
        for r in raw_results.get("flipkart", {}).get("records", []):
            _clean_add(names, r.get("name"))
            _clean_add(emails, r.get("email"))
            loc = ", ".join(filter(None, [r.get("address"), r.get("city"), r.get("state"), r.get("pincode")]))
            _clean_add(addresses, loc)

        # Parse Instagram
        for r in raw_results.get("instagram", {}).get("records", []):
            _clean_add(names, r.get("name"))
            _clean_add(emails, r.get("email"))
            if r.get("username"):
                _clean_add(ig_handles, r["username"])

        # Parse Twitter
        for r in raw_results.get("twitter", {}).get("records", []):
            _clean_add(names, r.get("name"))
            if r.get("screen_name"):
                _clean_add(tw_handles, r["screen_name"])

        # Parse Telegram
        if raw_results.get("telegram", {}).get("status") == "success":
            tg_info = raw_results["telegram"]

        # Parse UPI
        upi_info = raw_results.get("upi", {})
        upi_handles = upi_info.get("upi_handles", [])[:8] if isinstance(upi_info, dict) else []
        primary_vpa = upi_info.get("primary_recommended_vpa") if isinstance(upi_info, dict) else None

        # Parse Family
        family_members = []
        for r in raw_results.get("family", {}).get("records", []):
            rel = r.get("relation", "Household Linkage")
            m_name = r.get("name")
            if m_name:
                family_members.append({
                    "name": m_name,
                    "relation": rel,
                    "phone": r.get("phoneNumber"),
                    "source": r.get("source", "Aadhaar Registry")
                })

        # Mask Aadhaar
        masked_aadhaar_list = []
        for a in aadhaars:
            clean_a = re.sub(r'\D', '', a)
            if len(clean_a) == 12:
                masked_aadhaar_list.append(f"XXXXXXXX{clean_a[-4:]}")
            else:
                masked_aadhaar_list.append(clean_a)

        # Clean PAN
        valid_pans = list(set([p for p in pan_cards if len(p) == 10]))

        # Calculate Hits and Confidence Score
        hits = 0
        for src, d in raw_results.items():
            if isinstance(d, dict) and (d.get("status") == "success" or d.get("success") is True or d.get("count", 0) > 0):
                hits += 1

        if hits >= 4:
            confidence = "98% (Extremely High - Multi-Database Consensus)"
        elif hits >= 2:
            confidence = "85% (High - Verified across multiple registries)"
        elif hits == 1:
            confidence = "65% (Single Registry Verified)"
        else:
            confidence = "Unverified Record"

        canonical_name = names[0] if names else (target_name or "Unknown Subject")

        # Resolve LPG Gas Connection details for target phone
        gas_dossier = None
        if target_phone:
            try:
                from external_services import ServiceGateway
                user_kyc_ctx = {
                    "name": canonical_name,
                    "fathersName": father_names[0] if father_names else None,
                    "phoneNumber": target_phone,
                    "aadharNumber": aadhaars[0] if aadhaars else None,
                    "address": addresses[0] if addresses else None
                }
                gas_dossier = ServiceGateway.get_lpg(target_phone, user_kyc=user_kyc_ctx)
            except Exception:
                pass

        dossier = {
            "status": "success" if hits > 0 or resolved_vehicle_info else "not_found",
            "dossier_type": "🔥 360° Master Indian Deep Intelligence Dossier",
            "owner": "@its_Xyron",
            "query": raw,
            "target_anchor": {
                "resolved_phone": target_phone,
                "resolved_aadhaar": masked_aadhaar_list[0] if masked_aadhaar_list else None,
                "resolved_pan": valid_pans[0] if valid_pans else None,
                "resolved_vpa": target_vpa or primary_vpa,
                "resolved_vehicle": target_vehicle
            },
            "kyc_verification_score": confidence,
            "canonical_identity": {
                "legal_name": canonical_name,
                "all_aliases_identified": list(set(names)),
                "father_or_husband_name": father_names[0] if father_names else None,
                "mother_name": mother_names[0] if mother_names else None,
                "date_of_birth": dob_list[0] if dob_list else None,
                "educational_qualifications": list(set(qualifications)),
                "higher_education_institutes": list(set(colleges))
            },
            "contact_and_telecom": {
                "primary_mobile": target_phone,
                "formatted_mobile": f"+91 {target_phone}" if target_phone else None,
                "identified_carriers": list(set(carriers)),
                "email_addresses": list(set(emails))
            },
            "government_and_national_identity": {
                "linked_aadhaars": list(set(masked_aadhaar_list)),
                "verified_pan_cards": valid_pans,
                "pan_entity_type": "Individual / Person" if valid_pans else None,
                "lpg_connection": {
                    "lpg_id": gas_dossier.get("lpg_id") if isinstance(gas_dossier, dict) else None,
                    "provider": gas_dossier.get("primary_provider") if isinstance(gas_dossier, dict) else None,
                    "agency": gas_dossier.get("distributor_agency") if isinstance(gas_dossier, dict) else None,
                    "subsidy_status": gas_dossier.get("subsidy_status") if isinstance(gas_dossier, dict) else None
                } if gas_dossier else None
            },
            "residential_and_delivery_footprint": {
                "primary_registered_address": addresses[0] if addresses else None,
                "all_addresses_discovered": list(set(addresses))
            },
            "financial_and_banking_footprint": {
                "detected_bank_accounts": list(set(bank_accounts)),
                "active_upi_handles": upi_handles,
                "primary_recommended_vpa": primary_vpa,
                "qr_payment_intent": upi_info.get("primary_qr_payload") if isinstance(upi_info, dict) else None,
                "cibil_credit_score": cibil_score,
                "highmark_credit_score": highmark_score
            },
            "vehicles_and_transport": {
                "vehicle_rc_record": resolved_vehicle_info
            } if resolved_vehicle_info else None,
            "household_and_family_tree": {
                "total_family_connections": len(family_members),
                "family_members": family_members
            },
            "digital_and_social_presence": {
                "telegram_profile": tg_info,
                "instagram_usernames": list(set(ig_handles)),
                "twitter_handles": list(set(tw_handles))
            },
            "intelligence_metadata": {
                "total_master_databases_queried": len(futures),
                "total_master_databases_hit": hits,
                "engine_version": "3.0.0 Enterprise Ultra",
                "latency_ms": round((time.time() - t0) * 1000, 2)
            },
            "raw_sources": raw_results,
            "latency_ms": round((time.time() - t0) * 1000, 2),
            "cached": False
        }

        with self.cache_lock:
            self.cache[cache_key] = dossier

        return dossier



# Master registry of official Indian UPI handles, PSP apps, and partner banks
UPI_HANDLES_MAP = {
    # PhonePe
    "ybl": {"app": "PhonePe", "bank": "YES Bank", "ifsc": "YESB", "type": "PSP Primary"},
    "ibl": {"app": "PhonePe", "bank": "ICICI Bank", "ifsc": "ICIC", "type": "PSP Secondary"},
    "axl": {"app": "PhonePe", "bank": "Axis Bank", "ifsc": "UTIB", "type": "PSP Tertiary"},
    # Google Pay (Tez)
    "okhdfcbank": {"app": "Google Pay", "bank": "HDFC Bank", "ifsc": "HDFC", "type": "PSP Partner"},
    "oksbi": {"app": "Google Pay", "bank": "State Bank of India", "ifsc": "SBIN", "type": "PSP Partner"},
    "okaxis": {"app": "Google Pay", "bank": "Axis Bank", "ifsc": "UTIB", "type": "PSP Partner"},
    "okicici": {"app": "Google Pay", "bank": "ICICI Bank", "ifsc": "ICIC", "type": "PSP Partner"},
    # Paytm
    "paytm": {"app": "Paytm", "bank": "Paytm Payments / Partner Bank", "ifsc": "PYTM", "type": "PSP Primary"},
    "pthdfc": {"app": "Paytm", "bank": "HDFC Bank", "ifsc": "HDFC", "type": "Partner Bank"},
    "ptaxis": {"app": "Paytm", "bank": "Axis Bank", "ifsc": "UTIB", "type": "Partner Bank"},
    "ptsbi": {"app": "Paytm", "bank": "State Bank of India", "ifsc": "SBIN", "type": "Partner Bank"},
    "pticici": {"app": "Paytm", "bank": "ICICI Bank", "ifsc": "ICIC", "type": "Partner Bank"},
    # BHIM / NPCI National Switch
    "upi": {"app": "BHIM / NPCI", "bank": "NPCI Unified Platform", "ifsc": "NPCI", "type": "National Switch"},
    # Amazon Pay
    "apl": {"app": "Amazon Pay", "bank": "Axis Bank", "ifsc": "UTIB", "type": "Co-Branded"},
    "rapl": {"app": "Amazon Pay", "bank": "RBL Bank", "ifsc": "RATN", "type": "Co-Branded"},
    # WhatsApp Pay
    "waicici": {"app": "WhatsApp Pay", "bank": "ICICI Bank", "ifsc": "ICIC", "type": "Social PSP"},
    "wahdfcbank": {"app": "WhatsApp Pay", "bank": "HDFC Bank", "ifsc": "HDFC", "type": "Social PSP"},
    "waaxis": {"app": "WhatsApp Pay", "bank": "Axis Bank", "ifsc": "UTIB", "type": "Social PSP"},
    "wasbi": {"app": "WhatsApp Pay", "bank": "State Bank of India", "ifsc": "SBIN", "type": "Social PSP"},
    # CRED
    "cred": {"app": "CRED", "bank": "Axis Bank", "ifsc": "UTIB", "type": "Fintech App"},
    "crd": {"app": "CRED", "bank": "NPCI", "ifsc": "NPCI", "type": "Fintech App"},
    # Super.money & Slice & Navi
    "superyes": {"app": "Super.money (Flipkart)", "bank": "YES Bank", "ifsc": "YESB", "type": "Fintech App"},
    "superaxis": {"app": "Super.money (Flipkart)", "bank": "Axis Bank", "ifsc": "UTIB", "type": "Fintech App"},
    "sliceaxis": {"app": "Slice", "bank": "Axis Bank", "ifsc": "UTIB", "type": "Fintech App"},
    "navi": {"app": "Navi UPI", "bank": "Karnataka Bank / Axis", "ifsc": "UTIB", "type": "Fintech App"},
    # Direct Bank Applications
    "sbi": {"app": "BHIM SBI Pay / YONO", "bank": "State Bank of India", "ifsc": "SBIN", "type": "Bank Direct"},
    "hdfcbank": {"app": "PayZapp / HDFC MobileBanking", "bank": "HDFC Bank", "ifsc": "HDFC", "type": "Bank Direct"},
    "icici": {"app": "iMobile Pay", "bank": "ICICI Bank", "ifsc": "ICIC", "type": "Bank Direct"},
    "axisbank": {"app": "Axis Mobile / Open", "bank": "Axis Bank", "ifsc": "UTIB", "type": "Bank Direct"},
    "barodampay": {"app": "bob World UPI", "bank": "Bank of Baroda", "ifsc": "BARB", "type": "Bank Direct"},
    "kotak": {"app": "Kotak 811", "bank": "Kotak Mahindra Bank", "ifsc": "KKBK", "type": "Bank Direct"},
    "pnb": {"app": "PNB ONE", "bank": "Punjab National Bank", "ifsc": "PUNB", "type": "Bank Direct"},
    "airtel": {"app": "Airtel Thanks", "bank": "Airtel Payments Bank", "ifsc": "AIRP", "type": "Payments Bank"},
    "jio": {"app": "MyJio UPI", "bank": "Jio Payments Bank", "ifsc": "JIOP", "type": "Payments Bank"},
    "postbank": {"app": "DakPay", "bank": "India Post Payments Bank", "ifsc": "IPOS", "type": "Payments Bank"},
    "federal": {"app": "FedMobile", "bank": "Federal Bank", "ifsc": "FDRL", "type": "Bank Direct"},
    "idfcfirst": {"app": "IDFC FIRST Bank", "bank": "IDFC FIRST Bank", "ifsc": "IDFB", "type": "Bank Direct"},
    "indus": {"app": "IndusMobile", "bank": "IndusInd Bank", "ifsc": "INDB", "type": "Bank Direct"},
    "unionbank": {"app": "Vyom", "bank": "Union Bank of India", "ifsc": "UBIN", "type": "Bank Direct"},
    "canarabank": {"app": "Canara ai1", "bank": "Canara Bank", "ifsc": "CNRB", "type": "Bank Direct"}
}