Create app.py

app.py

@@ -0,0 +1,689 @@
+# ══════════════════════════════════════════════════════════════════
+#  CRITICAL SHIM — MUST BE THE FIRST EXECUTABLE CODE
+# ══════════════════════════════════════════════════════════════════
+import huggingface_hub
+
+if not hasattr(huggingface_hub, "HfFolder"):
+    class _HfFolder:
+        @staticmethod
+        def get_token():
+            try: return huggingface_hub.get_token()
+            except Exception: return None
+        @staticmethod
+        def save_token(token):
+            try: huggingface_hub.login(token)
+            except Exception: pass
+        @staticmethod
+        def delete_token():
+            try: huggingface_hub.logout()
+            except Exception: pass
+    huggingface_hub.HfFolder = _HfFolder
+
+# ── Now safe to import gradio ─────────────────────────────────────
+import os
+import asyncio
+import html
+import re
+import time
+import uuid
+import threading
+import traceback
+import logging
+from typing import Optional
+
+import duckdb
+import httpx
+import gradio as gr
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from starlette.routing import Route as StarletteRoute
+
+# ─────────────────────────────────────────────
+# CONFIG
+# ─────────────────────────────────────────────
+BOT_TOKEN      = os.environ.get("BOT_TOKEN", "").strip()
+HF_TOKEN       = os.environ.get("HF_TOKEN", "").strip()
+WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "1221").strip()
+SPACE_URL      = os.environ.get("SPACE_URL", "").strip().rstrip("/")
+TELEGRAM_PROXY = os.environ.get("TELEGRAM_PROXY", "").strip().rstrip("/")
+
+TELEGRAM_API = (
+    f"{TELEGRAM_PROXY}/bot{BOT_TOKEN}" if TELEGRAM_PROXY
+    else f"https://api.telegram.org/bot{BOT_TOKEN}"
+)
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+log = logging.getLogger("bot")
+
+# ─────────────────────────────────────────────
+# DATASETS & API FALLBACKS
+# ─────────────────────────────────────────────
+# 1. ICMR Dataset
+ICMR_URLS = [
+    "https://huggingface.co/datasets/Kzr0xx/Icmr-and-hitek/resolve/main/part1.parquet",
+    "https://huggingface.co/datasets/Kzr0xx/Icmr-and-hitek/resolve/main/part2a.parquet",
+    "https://huggingface.co/datasets/Kzr0xx/Icmr-and-hitek/resolve/main/part2b_new.parquet",
+]
+
+# 2. HiTeck Dataset (Only streaming the 7 phone index files, ignoring Aadhaar)
+HITECK_URLS = [
+    f"https://huggingface.co/datasets/Hfzune/HiTeckgroop/resolve/main/idx_phone.{i}.parquet"
+    for i in range(7)
+]
+
+# Combined list for DuckDB
+PARQUET_URLS = ICMR_URLS + HITECK_URLS
+
+# API Fallback Endpoints (Used when DuckDB is warming up)
+API_850CR = "https://sauravsingh2111-850crfastsearch.hf.space/search"
+API_INDDATA = "https://tfqdeadlo-inddataapi.hf.space/search"
+
+PAGE_SIZE     = 4
+FETCH_LIMIT   = 40
+QUERY_TIMEOUT = 90
+USER_COOLDOWN = 5
+CACHE_TTL     = 300
+
+# Only 10-digit mobile numbers are searched
+PHONE_REGEX = re.compile(r"^\d{10}$")
+
+# Truncation per column (keeps pages under Telegram's 4096-char limit)
+TRUNC = {
+    "name": 80, "fathersName": 80, "address": 200,
+    "phoneNumber": 40, "aadharNumber": 40, "otherNumber": 40,
+    "district": 40, "pincode": 15, "state": 40, "town": 40, "source": 30,
+}
+
+# ─────────────────────────────────────────────
+# DUCKDB ENGINE
+# ─────────────────────────────────────────────
+_con = None
+_con_lock = threading.Lock()
+_con_ready = threading.Event()
+_query_lock = threading.Lock()
+_warmed = threading.Event()
+_warm_thread_started = False
+
+def get_con() -> duckdb.DuckDBPyConnection:
+    global _con
+    with _con_lock:
+        if _con is not None:
+            return _con
+        log.info("🔧 Initializing DuckDB...")
+        c = duckdb.connect(":memory:")
+        c.execute("INSTALL httpfs; LOAD httpfs;")
+        if HF_TOKEN:
+            c.execute(f"CREATE SECRET hf (TYPE HUGGINGFACE, TOKEN '{HF_TOKEN}');")
+        c.execute("SET enable_object_cache=true;")
+        c.execute("SET preserve_insertion_order=false;")
+        c.execute("SET threads=6;")  # Increased threads for 10 files
+        c.execute("SET memory_limit='12GB';")
+        urls = ", ".join(f"'{u}'" for u in PARQUET_URLS)
+        try:
+            c.execute(f"SELECT COUNT(*) FROM read_parquet([{urls}], union_by_name=true) WHERE 1=0")
+            log.info("✅ DuckDB connection ready")
+        except Exception as e:
+            log.warning(f"⚠️ Warm-up metadata warning: {e}")
+        _con = c
+        _con_ready.set()
+        return _con
+
+def warm_phone_cache() -> None:
+    """Forces DuckDB to read the entire phoneNumber column across all 10 files."""
+    global _warm_thread_started
+    if _warmed.is_set():
+        return
+    _warm_thread_started = True
+    try:
+        log.info("🔥 Warm-up starting (may take 3-10 minutes for 200GB+)...")
+        t0 = time.time()
+        con = get_con()
+        urls = ", ".join(f"'{u}'" for u in PARQUET_URLS)
+        with _query_lock:
+            con.execute(f"""
+                SELECT SUM(HASH(CAST(phoneNumber AS VARCHAR)))
+                FROM read_parquet([{urls}], union_by_name=true)
+                WHERE phoneNumber IS NOT NULL
+            """).fetchall()
+        elapsed = time.time() - t0
+        _warmed.set()
+        log.info(f"✅ Warm-up complete in {elapsed:.1f}s")
+    except Exception as e:
+        log.error(f"⚠️ Warm-up failed: {e}")
+        _warm_thread_started = False # Allow retry
+
+def run_search_safe(phone: str, limit: int = FETCH_LIMIT, timeout_secs: int = QUERY_TIMEOUT) -> list:
+    """Search by 10-digit phone in DuckDB. Cancel-on-timeout via con.interrupt()."""
+    con = get_con()
+    with _query_lock:
+        stopped = threading.Event()
+        def watchdog():
+            if not stopped.wait(timeout_secs):
+                log.warning(f"⏱️ Timeout {timeout_secs}s — interrupting DuckDB")
+                try: con.interrupt()
+                except Exception as e: log.warning(f"⚠️ interrupt() failed: {e}")
+        wd = threading.Thread(target=watchdog, daemon=True)
+        wd.start()
+        try:
+            urls = ", ".join(f"'{u}'" for u in PARQUET_URLS)
+            sql = f"""
+                SELECT name, fathersName, phoneNumber, aadharNumber, otherNumber,
+                       address, district, pincode, state, town, source
+                FROM read_parquet([{urls}], union_by_name=true)
+                WHERE CAST(phoneNumber AS VARCHAR) = ?
+                LIMIT {int(limit)}
+            """
+            cur = con.execute(sql, [phone])
+            cols = [d[0] for d in cur.description]
+            return [dict(zip(cols, r)) for r in cur.fetchall()]
+        except duckdb.InterruptException:
+            raise TimeoutError(f"Search timed out after {timeout_secs}s")
+        finally:
+            stopped.set()
+
+# ─────────────────────────────────────────────
+# API FALLBACK ENGINE
+# ─────────────────────────────────────────────
+async def _q_cr850(client, number):
+    """850CR Fast — GET /search/{number}"""
+    try:
+        r = await client.get(f"{API_850CR}/{number}", timeout=40.0)
+        if r.status_code != 200:
+            return []
+        data = r.json()
+        if not data.get("success") or not data.get("results"):
+            return []
+        out = []
+        for item in data["results"]:
+            addr = item.get("address") or {}
+            full = addr.get("full_address", "") if isinstance(addr, dict) else ""
+            if full: full = full.lstrip(", ").strip()
+            out.append({
+                "name": item.get("name"),
+                "fathersName": item.get("father_name"),
+                "phoneNumber": number,
+                "aadharNumber": None,
+                "otherNumber": item.get("alternate_number"),
+                "address": full,
+                "district": None,
+                "pincode": None,
+                "state": item.get("state"),
+                "town": None,
+                "source": "850CR Fast"
+            })
+        return out
+    except Exception as e:
+        log.warning(f"850CR API failed: {e}")
+        return []
+
+async def _q_inddata(client, number):
+    """IndData 1.4B — GET /search?mobile={number}"""
+    try:
+        r = await client.get(API_INDDATA, params={"mobile": number}, timeout=30.0)
+        if r.status_code != 200:
+            return []
+        data = r.json()
+        rows = data.get("data") if isinstance(data, dict) else None
+        if not isinstance(rows, list):
+            return []
+        out = []
+        for item in rows:
+            out.append({
+                "name": item.get("name"),
+                "fathersName": item.get("fname"),
+                "phoneNumber": number,
+                "aadharNumber": None,
+                "otherNumber": None,
+                "address": item.get("address"),
+                "district": None,
+                "pincode": None,
+                "state": None,
+                "town": None,
+                "source": "IndData 1.4B"
+            })
+        return out
+    except Exception as e:
+        log.warning(f"IndData API failed: {e}")
+        return []
+
+async def search_apis(number: str) -> list:
+    """Fallback to APIs when DuckDB is not warmed up."""
+    log.info(f"🔄 DuckDB not warmed. Falling back to APIs for {number}")
+    async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client:
+        results = await asyncio.gather(
+            _q_cr850(client, number),
+            _q_inddata(client, number),
+            return_exceptions=True
+        )
+    merged = []
+    for res in results:
+        if isinstance(res, list):
+            merged.extend(res)
+    return merged
+
+# ─────────────────────────────────────────────
+# DEDUPLICATION & FORMATTING
+# ─────────────────────────────────────────────
+def _norm(s) -> str:
+    if not s: return ""
+    return re.sub(r"\s+", " ", str(s).strip().lower())
+
+def _dedup_key(rec: dict):
+    name = _norm(rec.get("name"))
+    father = _norm(rec.get("fathersName"))
+    addr = _norm(rec.get("address"))
+    addr_n = re.sub(r"[^a-z0-9]+", " ", addr).strip()
+
+    if name and addr_n: return ("na", name, addr_n)
+    if name and father: return ("nf", name, father)
+    if addr_n: return ("a", addr_n)
+    return None
+
+def dedup_records(records: list) -> list:
+    seen = {}
+    result = []
+    for rec in records:
+        key = _dedup_key(rec)
+        if key is None:
+            result.append(rec)
+            continue
+        idx = seen.get(key)
+        if idx is None:
+            seen[key] = len(result)
+            result.append(rec)
+        else:
+            # Keep the one with more filled fields
+            existing = result[idx]
+            existing_fields = sum(1 for v in existing.values() if v)
+            new_fields = sum(1 for v in rec.values() if v)
+            if new_fields > existing_fields:
+                result[idx] = rec
+    return result
+
+def _tv(v, n: int) -> str:
+    if v is None: return ""
+    s = str(v).strip()
+    if s.lower() in ("", "null", "none", "nan"): return ""
+    return s if len(s) <= n else s[: n - 1] + "…"
+
+FIELD_LABELS = [
+    ("name",        "👤", "Name"),
+    ("fathersName", "👨", "Father"),
+    ("phoneNumber", "📞", "Phone"),
+    ("otherNumber", "📱", "Other"),
+    ("aadharNumber","🆔", "Aadhaar"),
+    ("address",     "🏠", "Address"),
+    ("district",    "🏙️", "District"),
+    ("pincode",     "📮", "Pincode"),
+    ("state",       "🗺️", "State"),
+    ("town",        "🌆", "Town"),
+    ("source",      "📂", "Source"),
+]
+
+def format_result(idx: int, row: dict) -> str:
+    lines = [f"<b>━━━ #{idx} ━━━</b>"]
+    for key, emoji, label in FIELD_LABELS:
+        v = _tv(row.get(key), TRUNC.get(key, 60))
+        if v:
+            lines.append(f"{emoji} <b>{label}:</b> <code>{html.escape(v)}</code>")
+    return "\n".join(lines)
+
+def render_page(results, page, total_pages, phone, elapsed, source_note=""):
+    start = page * PAGE_SIZE
+    chunk = results[start:start + PAGE_SIZE]
+    head = (
+        f"📱 <b>Phone:</b> <code>{html.escape(phone)}</code>\n"
+        f"📄 <b>Page {page+1}/{total_pages}</b> · "
+        f"{len(results)} result(s) · ⏱️ {elapsed:.1f}s"
+    )
+    if source_note:
+        head += f"\n⚠️ <i>{source_note}</i>"
+    body = "\n\n".join(format_result(start + i + 1, r) for i, r in enumerate(chunk))
+    return f"{head}\n\n{body}"
+
+def page_keyboard(search_id, page, total_pages):
+    nav = []
+    if page > 0:
+        nav.append({"text": "◀ Prev", "callback_data": f"p:{search_id}:{page-1}"})
+    nav.append({"text": f"{page+1} / {total_pages}", "callback_data": "noop"})
+    if page < total_pages - 1:
+        nav.append({"text": "Next ▶", "callback_data": f"p:{search_id}:{page+1}"})
+    return {"inline_keyboard": [nav]}
+
+# ─────────────────────────────────────────────
+# TELEGRAM API HELPERS
+# ─────────────────────────────────────────────
+async def tg(method: str, payload: dict, timeout: float = 30) -> dict:
+    try:
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            r = await client.post(f"{TELEGRAM_API}/{method}", json=payload)
+            return r.json()
+    except Exception as e:
+        log.error(f"❌ tg({method}) error: {e!r}")
+        return {"ok": False, "error": str(e)}
+
+async def send(chat_id, text, **kw):
+    kw.setdefault("parse_mode", "HTML")
+    kw.setdefault("disable_web_page_preview", True)
+    res = await tg("sendMessage", {"chat_id": chat_id, "text": text, **kw})
+    if res.get("ok"): return res["result"]["message_id"]
+    log.warning(f"⚠️ sendMessage failed: {res}")
+    return None
+
+async def edit(chat_id, message_id, text, **kw):
+    kw.setdefault("parse_mode", "HTML")
+    kw.setdefault("disable_web_page_preview", True)
+    res = await tg("editMessageText", {"chat_id": chat_id, "message_id": message_id, "text": text, **kw})
+    if not res.get("ok") and "message is not modified" not in str(res):
+        log.warning(f"⚠️ editMessageText failed: {res}")
+
+# ─────────────────────────────────────────────
+# STATE MANAGEMENT
+# ─────────────────────────────────────────────
+SEARCH_CACHE = {}
+USER_LAST = {}
+STATS = {"queries": 0, "timeouts": 0, "errors": 0, "started": time.time()}
+
+def cache_gc():
+    while True:
+        time.sleep(300)
+        now = time.time()
+        dead = [k for k, v in SEARCH_CACHE.items() if now - v["ts"] > CACHE_TTL]
+        for k in dead: SEARCH_CACHE.pop(k, None)
+
+# ─────────────────────────────────────────────
+# HANDLERS
+# ─────────────────────────────────────────────
+WELCOME = (
+    "👋 <b>Phone Lookup Bot</b>\n\n"
+    "Send me any <b>10-digit mobile number</b> and I'll return every "
+    "record we have on file.\n\n"
+    "Example: <code>9555179248</code>\n\n"
+    "<i>Commands:</i>\n"
+    "• /start — show this message\n"
+    "• /status — bot health\n"
+)
+
+async def handle_message(msg: dict):
+    try:
+        chat_id = msg["chat"]["id"]
+        text = (msg.get("text") or "").strip()
+        log.info(f"💬 Message from {chat_id}: {text[:50]!r}")
+        if not text: return
+
+        # ── /status ─────────────────────────────
+        if text == "/status":
+            uptime = (time.time() - STATS["started"]) / 60
+            await send(chat_id, (
+                f"<b>📊 Bot Status</b>\n\n"
+                f"⏱️ Uptime: <code>{uptime:.1f} min</code>\n"
+                f"🔌 DuckDB: <code>{_con_ready.is_set()}</code>\n"
+                f"🔥 Cache warmed: <code>{_warmed.is_set()}</code>\n"
+                f"🔍 Queries: <code>{STATS['queries']}</code>\n"
+                f"⏰ Timeouts: <code>{STATS['timeouts']}</code>\n"
+                f"❌ Errors: <code>{STATS['errors']}</code>\n"
+                f"💾 Cached searches: <code>{len(SEARCH_CACHE)}</code>"
+            ))
+            return
+
+        # ── /start ──────────────────────────────
+        if text in ("/start", "/help"):
+            await send(chat_id, WELCOME)
+            if not _warmed.is_set():
+                if not _warm_thread_started:
+                    threading.Thread(target=warm_phone_cache, daemon=True).start()
+            return
+
+        # ── Validate 10-digit phone ─────────────
+        cleaned = "".join(ch for ch in text if ch.isdigit())
+        if not PHONE_REGEX.match(cleaned):
+            await send(chat_id, "❌ <b>Invalid input</b>\n\nSend exactly a <b>10-digit mobile number</b>.\nExample: <code>9555179248</code>")
+            return
+
+        phone = cleaned
+
+        # ── Cooldown ────────────────────────────
+        now = time.time()
+        if now - USER_LAST.get(chat_id, 0) < USER_COOLDOWN:
+            wait = USER_COOLDOWN - (now - USER_LAST.get(chat_id, 0))
+            await send(chat_id, f"⏳ Wait {wait:.0f}s before next search.")
+            return
+        USER_LAST[chat_id] = now
+
+        # ── Ensure warm-up is running ───────────
+        if not _warmed.is_set() and not _warm_thread_started:
+            threading.Thread(target=warm_phone_cache, daemon=True).start()
+
+        # ── Placeholder ─────────────────────────
+        placeholder = await send(
+            chat_id,
+            f"🔍 <b>Searching…</b>\n\n📱 <code>{html.escape(phone)}</code>\n\n"
+            f"<i>First query after restart may take up to {QUERY_TIMEOUT}s. Repeats are fast.</i>"
+        )
+        if placeholder is None: return
+
+        # ── Run Search (DuckDB with API Fallback) ─
+        loop = asyncio.get_running_loop()
+        t0 = time.time()
+        source_note = ""
+        
+        try:
+            if not _warmed.is_set():
+                # Fallback to APIs if warming
+                results = await search_apis(phone)
+                source_note = "DuckDB warming in background. Showing API fallback results."
+            else:
+                # Use DuckDB
+                results = await asyncio.wait_for(
+                    loop.run_in_executor(None, run_search_safe, phone, FETCH_LIMIT),
+                    timeout=QUERY_TIMEOUT + 10,
+                )
+        except asyncio.TimeoutError:
+            STATS["timeouts"] += 1
+            await edit(chat_id, placeholder, f"⏱️ <b>Timed out after {QUERY_TIMEOUT}s</b>\n\nTry the same number again — cache is warming.")
+            return
+        except TimeoutError:
+            STATS["timeouts"] += 1
+            await edit(chat_id, placeholder, f"⏱️ <b>Timed out after {QUERY_TIMEOUT}s</b>\n\nTry the same number again.")
+            return
+        except Exception as e:
+            STATS["errors"] += 1
+            log.error(f"❌ search error: {e!r}\n{traceback.format_exc()}")
+            await edit(chat_id, placeholder, f"❌ <b>Error</b>\n<code>{html.escape(str(e)[:200])}</code>")
+            return
+
+        elapsed = time.time() - t0
+        STATS["queries"] += 1
+        log.info(f"✅ Search returned {len(results)} rows in {elapsed:.1f}s")
+
+        # ── Deduplicate ──────────────────────────
+        results = dedup_records(results)
+        log.info(f"🧹 Deduplicated to {len(results)} rows")
+
+        # ── No results ──────────────────────────
+        if not results:
+            await edit(chat_id, placeholder, f"❌ <b>No record found</b>\n\n📱 <code>{html.escape(phone)}</code>\n⏱️ {elapsed:.1f}s")
+            return
+
+        # ── Paginate ────────────────────────────
+        search_id = uuid.uuid4().hex[:10]
+        total_pages = (len(results) + PAGE_SIZE - 1) // PAGE_SIZE
+        SEARCH_CACHE[search_id] = {
+            "ts": time.time(),
+            "results": results,
+            "query": phone,
+            "elapsed": elapsed,
+            "source_note": source_note
+        }
+        await edit(
+            chat_id, placeholder,
+            render_page(results, 0, total_pages, phone, elapsed, source_note),
+            reply_markup=page_keyboard(search_id, 0, total_pages),
+        )
+    except Exception as e:
+        log.error(f"❌ handle_message fatal: {e!r}\n{traceback.format_exc()}")
+
+async def handle_callback(cb: dict):
+    try:
+        data = cb.get("data", "")
+        cb_id = cb["id"]
+        msg = cb.get("message") or {}
+        chat_id = msg.get("chat", {}).get("id")
+        message_id = msg.get("message_id")
+        
+        if not chat_id or not message_id:
+            await tg("answerCallbackQuery", {"callback_query_id": cb_id})
+            return
+        if data == "noop" or not data.startswith("p:"):
+            await tg("answerCallbackQuery", {"callback_query_id": cb_id})
+            return
+
+        try:
+            _, search_id, page_str = data.split(":")
+            page = int(page_str)
+        except Exception:
+            await tg("answerCallbackQuery", {"callback_query_id": cb_id})
+            return
+
+        entry = SEARCH_CACHE.get(search_id)
+        if not entry:
+            await tg("answerCallbackQuery", {"callback_query_id": cb_id, "text": "Session expired — search again.", "show_alert": True})
+            return
+
+        results = entry["results"]
+        total_pages = (len(results) + PAGE_SIZE - 1) // PAGE_SIZE
+        page = max(0, min(page, total_pages - 1))
+
+        await edit(
+            chat_id, message_id,
+            render_page(results, page, total_pages, entry["query"], entry["elapsed"], entry.get("source_note", "")),
+            reply_markup=page_keyboard(search_id, page, total_pages),
+        )
+        await tg("answerCallbackQuery", {"callback_query_id": cb_id})
+    except Exception as e:
+        log.error(f"❌ handle_callback fatal: {e!r}\n{traceback.format_exc()}")
+
+# ─────────────────────────────────────────────
+# WEBHOOK & HEALTH
+# ─────────────────────────────────────────────
+async def set_webhook():
+    if not SPACE_URL: return
+    url = f"{SPACE_URL}/webhook"
+    payload = {
+        "url": url,
+        "secret_token": WEBHOOK_SECRET,
+        "drop_pending_updates": True,
+        "allowed_updates": ["message", "callback_query"],
+    }
+    for attempt in range(1, 4):
+        try:
+            async with httpx.AsyncClient(timeout=30) as client:
+                r = await client.post(f"{TELEGRAM_API}/setWebhook", json=payload)
+                data = r.json()
+                log.info(f"🔗 setWebhook → {data}")
+                if data.get("ok"): return
+        except Exception as e:
+            log.warning(f"⚠️ setWebhook attempt {attempt}: {e}")
+        await asyncio.sleep(3)
+
+async def _health_endpoint(request: Request):
+    return JSONResponse({
+        "status": "ok",
+        "duckdb_ready": _con_ready.is_set(),
+        "warmed": _warmed.is_set(),
+        "cached_searches": len(SEARCH_CACHE),
+        "bot_token_set": bool(BOT_TOKEN),
+        "stats": STATS,
+    })
+
+async def webhook_handler(request: Request):
+    if WEBHOOK_SECRET:
+        got = request.headers.get("x-telegram-bot-api-secret-token", "")
+        if got != WEBHOOK_SECRET:
+            return JSONResponse({"ok": False}, status_code=403)
+    try:
+        update = await request.json()
+    except Exception:
+        return JSONResponse({"ok": True})
+    
+    if "message" in update:
+        asyncio.create_task(handle_message(update["message"]))
+    elif "callback_query" in update:
+        asyncio.create_task(handle_callback(update["callback_query"]))
+    return JSONResponse({"ok": True})
+
+# ─────────────────────────────────────────────
+# GRADIO UI & LAUNCH
+# ─────────────────────────────────────────────
+with gr.Blocks(title="Phone Lookup Bot") as demo:
+    gr.Markdown("# 📞 Phone Lookup Bot")
+    gr.Markdown(
+        "Telegram bot that streams `Kzr0xx/Icmr-and-hitek` and `Hfzune/HiTeckgroop` "
+        "from HuggingFace on demand, with API fallback during warm-up."
+    )
+    gr.Markdown("### Status")
+    status_box = gr.JSON(label="Health")
+    demo.load(
+        fn=lambda: {
+            "status": "ok",
+            "duckdb_ready": _con_ready.is_set(),
+            "warmed": _warmed.is_set(),
+            "cached_searches": len(SEARCH_CACHE),
+        },
+        outputs=status_box,
+    )
+    try:
+        import spaces
+        @spaces.GPU(duration=10)
+        def _zerogpu_register():
+            return "gpu-ready"
+        _zerogpu_btn = gr.Button("init-gpu", visible=False)
+        _zerogpu_btn.click(fn=_zerogpu_register, inputs=None, outputs=None)
+        log.info("✅ @spaces.GPU registered")
+    except Exception as e:
+        log.warning(f"⚠️ ZeroGPU registration: {e}")
+
+# ─────────────────────────────────────────────
+# STARTUP THREADS
+# ─────────────────────────────────────────────
+threading.Thread(target=cache_gc, daemon=True).start()
+threading.Thread(target=get_con, daemon=True).start()
+threading.Thread(target=warm_phone_cache, daemon=True).start()
+
+# ─────────────────────────────────────────────
+# LAUNCH & ROUTE INJECTION
+# ─────────────────────────────────────────────
+demo.launch(
+    server_name="0.0.0.0",
+    server_port=7860,
+    share=False,
+    ssr_mode=False,
+    prevent_thread_lock=True,
+)
+
+log.info("🔎 Launch returned — injecting routes")
+live_app = demo.app
+
+live_app.router.routes.insert(0, StarletteRoute("/webhook", webhook_handler, methods=["POST"]))
+live_app.router.routes.insert(0, StarletteRoute("/health", _health_endpoint, methods=["GET"]))
+
+try:
+    live_app.build_middleware_stack()
+    log.info("✅ Middleware stack rebuilt")
+except Exception as e:
+    log.warning(f"⚠️ Stack rebuild: {e}")
+
+log.info("📋 First 5 routes:")
+for route in live_app.router.routes[:5]:
+    path = getattr(route, "path", "?")
+    methods = getattr(route, "methods", None)
+    log.info(f"   {sorted(methods) if methods else '?'} {path}")
+
+def _register_webhook_later():
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    loop.run_until_complete(set_webhook())
+
+threading.Thread(target=_register_webhook_later, daemon=True).start()
+
+demo.block_thread()