# ══════════════════════════════════════════════════════════════════
#  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 io
import csv
import json
import asyncio
import html
import re
import time
import uuid
import threading
import traceback
import logging
import urllib.parse
from collections import deque, Counter
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
# ─────────────────────────────────────────────
VERSION        = "1.4.0"

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("/")

ADMIN_TOKEN       = os.environ.get("ADMIN_TOKEN", "").strip()
ADMIN_SESSION_TTL = 3600
ADMIN_PROMPT_TTL  = 180

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 — HiTeck + Flipkart + 2 APIs
# ─────────────────────────────────────────────
HITECK_URLS = [
    f"https://huggingface.co/datasets/HiTeckGroup/HiTeckgroop/resolve/main/idx_phone.{i}.parquet"
    for i in range(7)
]
PARQUET_URLS = HITECK_URLS

# ── Flipkart dataset (single parquet, different schema) ──
# Columns: CustomerName, CustomerPhone, CustomerEmail, CustomerAddress,
#          CustomerCity, CustomerState, CustomerPincode
FLIPKART_URLS = [
    "https://huggingface.co/datasets/Hfzune/Flipkart/resolve/main/flipkart.parquet"
]
ALL_PARQUET_URLS = PARQUET_URLS + FLIPKART_URLS

API_850CR   = "https://tfqdeadlo-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
LIVE_WINDOW     = 300
HISTORY_LEN     = 10
BATCH_MAX       = 5
BATCH_TIMEOUT   = 60
BROADCAST_DELAY = 0.05

PHONE_REGEX = re.compile(r"^\d{10}$")

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...")
        try:
            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;")
            c.execute("SET memory_limit='12GB';")
            urls = ", ".join(f"'{u}'" for u in ALL_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"⚠️ Metadata warning: {e}")
            _con = c
            _con_ready.set()
            return _con
        except Exception as e:
            log.error(f"❌ DuckDB init failed: {e!r}")
            raise


def warm_phone_cache() -> None:
    global _warm_thread_started
    if _warmed.is_set():
        return
    _warm_thread_started = True
    log.info(f"🔥 Warm-up starting ({len(ALL_PARQUET_URLS)} files)...")
    t0 = time.time()
    loaded = skipped = 0
    try:
        con = get_con()
        for url in ALL_PARQUET_URLS:
            name = url.rsplit("/", 1)[-1]
            try:
                con.execute(f"SELECT * FROM read_parquet('{url}') LIMIT 1").fetchall()
                loaded += 1
                log.info(f"  ✅ warmed: {name}")
            except Exception as e:
                skipped += 1
                log.warning(f"  ⏭️ skipped {name}: {str(e)[:120]}")
        elapsed = time.time() - t0
        _warmed.set()
        log.info(f"✅ Warm-up complete in {elapsed:.1f}s ({loaded} loaded, {skipped} skipped)")
    except Exception as e:
        log.error(f"⚠️ Warm-up fatal: {e!r}")
        _warm_thread_started = False


def _flipkart_clean(rec: dict) -> dict:
    """URL-decode percent-encoded Flipkart text fields for display."""
    for k in ("name", "address", "district", "state", "town"):
        v = rec.get(k)
        if isinstance(v, str) and "%" in v:
            try:
                rec[k] = urllib.parse.unquote(v)
            except Exception:
                pass
    return rec


def run_search_safe(phone: str, limit: int = FETCH_LIMIT, timeout_secs: int = QUERY_TIMEOUT) -> list:
    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()

        results: list = []
        try:
            # ── HiTeck parquets ──
            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]
                results.extend(dict(zip(cols, r)) for r in cur.fetchall())
            except duckdb.InterruptException:
                raise TimeoutError(f"Search timed out after {timeout_secs}s")
            except Exception as e:
                log.warning(f"⚠️ HiTeck query failed for {phone}: {e!r}")

            # ── Flipkart parquet (schema-mapped) ──
            try:
                fk_urls = ", ".join(f"'{u}'" for u in FLIPKART_URLS)
                fk_sql = f"""
                    SELECT
                        CustomerName      AS name,
                        NULL              AS fathersName,
                        CustomerPhone     AS phoneNumber,
                        NULL              AS aadharNumber,
                        NULL              AS otherNumber,
                        CustomerAddress   AS address,
                        CustomerCity      AS district,
                        CustomerPincode   AS pincode,
                        CustomerState     AS state,
                        CustomerCity      AS town,
                        'Flipkart'        AS source
                    FROM read_parquet([{fk_urls}], union_by_name=true)
                    WHERE RIGHT(
                        REPLACE(REPLACE(REPLACE(
                            CAST(CustomerPhone AS VARCHAR),
                        ' ', ''), '-', ''), '+', ''),
                        10
                    ) = ?
                    LIMIT {int(limit)}
                """
                cur = con.execute(fk_sql, [phone])
                cols = [d[0] for d in cur.description]
                for r in cur.fetchall():
                    results.append(_flipkart_clean(dict(zip(cols, r))))
            except duckdb.InterruptException:
                raise TimeoutError(f"Search timed out after {timeout_secs}s")
            except Exception as e:
                log.warning(f"⚠️ Flipkart query failed for {phone}: {e!r}")

            return results
        except duckdb.InterruptException:
            raise TimeoutError(f"Search timed out after {timeout_secs}s")
        finally:
            stopped.set()

# ─────────────────────────────────────────────
# API FALLBACK
# ─────────────────────────────────────────────
async def _q_cr850(client: httpx.AsyncClient, number: str) -> list:
    """850CR API — sauravsingh2111-850crfastsearch.hf.space"""
    try:
        r = await client.get(f"{API_850CR}/{number}", timeout=40.0)
        if r.status_code != 200:
            log.warning(f"850CR HTTP {r.status_code} for {number}")
            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 (httpx.TimeoutException, httpx.ConnectError) as e:
        log.warning(f"850CR network error for {number}: {e}")
        return []
    except Exception as e:
        log.warning(f"850CR API failed for {number}: {e!r}")
        return []


async def _q_inddata(client: httpx.AsyncClient, number: str) -> list:
    """
    IndData API — tfqdeadlo-inddataapi.hf.space/search?mobile=<num>
    Response shape:
      {"status":"success","data":{mobile,name,fname,address,alt,circle,id,email}}
      OR {"status":"error","message":"..."} if not found
    """
    try:
        r = await client.get(API_INDDATA, params={"mobile": number}, timeout=30.0)
        if r.status_code != 200:
            log.warning(f"IndData HTTP {r.status_code} for {number}")
            return []
        data = r.json()
        if data.get("status") != "success":
            return []
        item = data.get("data")
        if not isinstance(item, dict):
            return []
        return [{
            "name": item.get("name"),
            "fathersName": item.get("fname"),
            "phoneNumber": number,
            "aadharNumber": None,
            "otherNumber": item.get("alt"),
            "address": item.get("address"),
            "district": None,
            "pincode": None,
            "state": None,
            "town": item.get("circle"),
            "source": "IndData 1.4B",
        }]
    except (httpx.TimeoutException, httpx.ConnectError) as e:
        log.warning(f"IndData network error for {number}: {e}")
        return []
    except Exception as e:
        log.warning(f"IndData API failed for {number}: {e!r}")
        return []


async def search_apis(number: str) -> list:
    """Call both APIs in parallel, ignore individual failures."""
    log.info(f"🔄 API search for {number}")
    try:
        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,
            )
    except Exception as e:
        log.error(f"❌ search_apis fatal: {e!r}")
        return []

    merged = []
    for res in results:
        if isinstance(res, list):
            merged.extend(res)
        elif isinstance(res, Exception):
            log.warning(f"API task raised: {res!r}")
    return merged

# ─────────────────────────────────────────────
# DEDUP & 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:
        if not isinstance(rec, dict):
            continue
        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:
            existing = result[idx]
            ef = sum(1 for v in existing.values() if v)
            nf = sum(1 for v in rec.values() if v)
            if nf > ef:
                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 _plain_record(idx: int, row: dict) -> str:
    lines = [f"#{idx}"]
    for key, _emoji, label in FIELD_LABELS:
        v = _tv(row.get(key), TRUNC.get(key, 60))
        if v:
            lines.append(f"{label}: {v}")
    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, has_export=True):
    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}"})
    rows = [nav]
    extras = [{"text": "📋 Copy page", "callback_data": f"c:{search_id}:{page}"}]
    if has_export:
        extras.append({"text": "📥 Export", "callback_data": f"e:{search_id}"})
    rows.append(extras)
    return {"inline_keyboard": rows}

# ─────────────────────────────────────────────
# ACTIVITY TRACKING
# ─────────────────────────────────────────────
ACTIVITY_LOG    = deque(maxlen=10000)
ACTIVE_USERS    = {}
UNIQUE_USERS    = {}
PHONE_COUNTS    = Counter()
PEAK_CONCURRENT = 0

_activity_lock  = threading.Lock()
_admin_lock     = threading.Lock()
_admin_sessions = {}
_admin_states   = {}

BANNED_USERS: set = set()
BAN_REASONS: dict = {}
USER_HISTORY: dict = {}

MAINTENANCE_MODE   = False
MAINTENANCE_REASON = ""
_state_lock = threading.Lock()


def _short_user(user_id: int, user: dict | None = None) -> str:
    info = UNIQUE_USERS.get(user_id) or {}
    username = (user or {}).get("username") or info.get("username")
    first = (user or {}).get("first_name") or info.get("first_name")
    if username: return f"@{username} ({user_id})"
    if first:    return f"{first} ({user_id})"
    return f"({user_id})"


def _log_activity(user: dict, action: str, input_val: str, reply_type: str,
                  summary: str = "", elapsed_ms: int = 0, source: str = ""):
    if not user or "id" not in user:
        return
    now = time.time()
    user_id = user["id"]
    with _activity_lock:
        info = UNIQUE_USERS.setdefault(user_id, {
            "username": None, "first_name": None, "last_name": None,
            "count": 0, "first_seen": now, "last_seen": now,
        })
        if user.get("username"):   info["username"]   = user["username"]
        if user.get("first_name"): info["first_name"] = user["first_name"]
        if user.get("last_name"):  info["last_name"]  = user["last_name"]
        info["count"] += 1
        info["last_seen"] = now
        ACTIVE_USERS[user_id] = now

        if action == "search" and isinstance(input_val, str) and input_val.isdigit() and len(input_val) == 10:
            PHONE_COUNTS[input_val] += 1

        ACTIVITY_LOG.append({
            "ts": now, "user_id": user_id,
            "username": user.get("username"), "first_name": user.get("first_name"),
            "action": action, "input": (input_val or "")[:80],
            "reply_type": reply_type, "summary": (summary or "")[:150],
            "elapsed_ms": int(elapsed_ms), "source": source,
        })


def _get_live_users():
    cutoff = time.time() - LIVE_WINDOW
    with _activity_lock:
        live = [(uid, ts) for uid, ts in ACTIVE_USERS.items() if ts >= cutoff]
    live.sort(key=lambda x: -x[1])
    return live


def _get_stats():
    now = time.time()
    today_start = now - (now % 86400)
    with _activity_lock:
        today_users = set(); today_searches = 0; total_searches = 0
        for e in ACTIVITY_LOG:
            if e["action"] == "search":
                total_searches += 1
                if e["ts"] >= today_start:
                    today_searches += 1; today_users.add(e["user_id"])
        return {
            "unique_users": len(UNIQUE_USERS),
            "live_users": sum(1 for ts in ACTIVE_USERS.values() if ts >= now - LIVE_WINDOW),
            "today_users": len(today_users),
            "today_searches": today_searches,
            "total_searches": total_searches,
            "peak_concurrent": PEAK_CONCURRENT,
            "total_log_entries": len(ACTIVITY_LOG),
            "banned_count": len(BANNED_USERS),
            "maintenance": MAINTENANCE_MODE,
        }


def _add_to_history(user_id: int, phone: str, count: int, elapsed: float, source: str):
    if user_id is None: return
    with _activity_lock:
        h = USER_HISTORY.setdefault(user_id, deque(maxlen=HISTORY_LEN))
        h.appendleft({"phone": phone, "ts": time.time(), "count": count,
                      "elapsed": elapsed, "source": source})

# ─────────────────────────────────────────────
# BAN / MAINTENANCE HELPERS
# ─────────────────────────────────────────────
def _is_banned(user_id: int) -> bool:
    with _state_lock:
        return user_id in BANNED_USERS


def _ban_user(user_id: int, reason: str = ""):
    with _state_lock:
        BANNED_USERS.add(user_id)
        BAN_REASONS[user_id] = reason


def _unban_user(user_id: int) -> bool:
    with _state_lock:
        if user_id in BANNED_USERS:
            BANNED_USERS.discard(user_id)
            BAN_REASONS.pop(user_id, None)
            return True
        return False


def _get_maintenance():
    with _state_lock:
        return MAINTENANCE_MODE, MAINTENANCE_REASON


def _set_maintenance(on: bool, reason: str = ""):
    global MAINTENANCE_MODE, MAINTENANCE_REASON
    with _state_lock:
        MAINTENANCE_MODE = on
        MAINTENANCE_REASON = reason if on else ""

# ─────────────────────────────────────────────
# ADMIN AUTH & STATE
# ─────────────────────────────────────────────
def _is_admin_session(user_id: int) -> bool:
    if user_id is None: return False
    with _admin_lock:
        exp = _admin_sessions.get(user_id)
        if exp is None: return False
        if time.time() > exp:
            _admin_sessions.pop(user_id, None); return False
        return True


def _grant_admin(user_id: int):
    with _admin_lock:
        _admin_sessions[user_id] = time.time() + ADMIN_SESSION_TTL


def _revoke_admin(user_id: int):
    with _admin_lock:
        _admin_sessions.pop(user_id, None)


def _admin_state_get(user_id: int):
    with _admin_lock:
        entry = _admin_states.get(user_id)
        if entry is None: return None
        state, set_at = entry
        if time.time() - set_at > ADMIN_PROMPT_TTL:
            _admin_states.pop(user_id, None); return None
        return state


def _admin_state_set(user_id: int, state: str):
    with _admin_lock:
        _admin_states[user_id] = (state, time.time())


def _admin_state_clear(user_id: int):
    with _admin_lock:
        _admin_states.pop(user_id, None)

# ─────────────────────────────────────────────
# ADMIN RENDERING (Telegram-side only)
# ─────────────────────────────────────────────
def _fmt_ts(ts: float) -> str:
    return time.strftime("%H:%M:%S", time.localtime(ts))


def admin_menu_kb():
    m_on, _ = _get_maintenance()
    maint_label = "🟢 Maintenance: OFF" if not m_on else "🔴 Maintenance: ON"
    return {
        "inline_keyboard": [
            [{"text": "🔴 Live Users",   "callback_data": "adm:live"},
             {"text": "📈 Stats",        "callback_data": "adm:stats"}],
            [{"text": "📜 Activity Log", "callback_data": "adm:log:0"},
             {"text": "👥 All Users",    "callback_data": "adm:users:0"}],
            [{"text": "🔍 Top Searches", "callback_data": "adm:top"},
             {"text": "🚫 Banned",       "callback_data": "adm:banned"}],
            [{"text": "🚫 Ban User",     "callback_data": "adm:ban"},
             {"text": "✅ Unban User",   "callback_data": "adm:unban"}],
            [{"text": "📢 Broadcast",    "callback_data": "adm:broadcast"},
             {"text": "📥 Export Log",   "callback_data": "adm:export"}],
            [{"text": "🔎 Log by Phone", "callback_data": "adm:logphone"},
             {"text": maint_label,       "callback_data": "adm:maint"}],
            [{"text": "🔄 Refresh",      "callback_data": "adm:menu"}],
        ]
    }


def render_admin_menu() -> str:
    m_on, reason = _get_maintenance()
    m_line = f"🔴 Maintenance ON — {html.escape(reason)}" if m_on else "🟢 Operational"
    return (
        f"🛡️ <b>Admin Panel</b> — v{VERSION}\n"
        "━━━━━━━━━━━━━━━━━━\n"
        f"{m_line}\n"
        f"👥 Users: {len(UNIQUE_USERS)} · 🚫 Banned: {len(BANNED_USERS)}\n\n"
        "Tap a button or use commands:\n"
        "• /live /stats /log /users /top\n"
        "• /ban &lt;id&gt; [reason] · /unban &lt;id&gt;\n"
        "• /broadcast &lt;msg&gt;\n"
        "• /export · /logphone &lt;num&gt;\n"
        "• /maintenance on|off [reason]\n"
        "• /logout"
    )


def render_live() -> str:
    live = _get_live_users()
    if not live:
        return f"🔴 <b>Live Users</b>\n\nNo one active in the last {LIVE_WINDOW // 60} minutes."
    lines = [f"🔴 <b>Live Users</b> (last {LIVE_WINDOW // 60} min): <b>{len(live)}</b>\n"]
    for uid, ts in live[:25]:
        lines.append(f"• {_short_user(uid)} — {_fmt_ts(ts)}")
    return "\n".join(lines)


def render_stats() -> str:
    s = _get_stats()
    uptime = (time.time() - STATS["started"]) / 60
    return (
        "📈 <b>Bot Statistics</b>\n"
        "━━━━━━━━━━━━━━━━━━\n"
        f"⏱️ Uptime: <code>{uptime:.1f} min</code>\n"
        f"👥 Unique users: <code>{s['unique_users']}</code>\n"
        f"🔴 Live now: <code>{s['live_users']}</code>\n"
        f"📅 Active today: <code>{s['today_users']}</code>\n"
        f"🔍 Searches today: <code>{s['today_searches']}</code>\n"
        f"🔍 Total searches: <code>{s['total_searches']}</code>\n"
        f"📊 Peak concurrent: <code>{s['peak_concurrent']}</code>\n"
        f"🚫 Banned users: <code>{s['banned_count']}</code>\n"
        f"💾 Cached queries: <code>{len(SEARCH_CACHE)}</code>\n"
        f"📜 Log entries: <code>{s['total_log_entries']}</code>\n"
        f"⏰ Timeouts: <code>{STATS['timeouts']}</code>\n"
        f"❌ Errors: <code>{STATS['errors']}</code>"
    )


def render_log(entries, page=0, page_size=8):
    total = len(entries)
    total_pages = max(1, (total + page_size - 1) // page_size)
    page = max(0, min(page, total_pages - 1))
    chunk = entries[page * page_size:(page + 1) * page_size]
    if not chunk:
        return "📜 <b>Activity Log</b>\n\nNo activity yet.", None

    lines = [f"📜 <b>Activity Log</b> — page {page + 1}/{total_pages} ({total} entries)\n"]
    for e in chunk:
        uname = e.get("username"); first = e.get("first_name")
        name_str = f"@{uname}" if uname else (first or "(no name)")
        uid = e["user_id"]
        reply = e.get("summary") or e.get("reply_type") or "-"
        elapsed = e.get("elapsed_ms", 0); source = e.get("source") or ""
        block  = f"<code>{_fmt_ts(e['ts'])}</code> · <b>{html.escape(name_str)}</b> · <code>{uid}</code>\n"
        block += f"  📥 <code>{html.escape(e['input'] or '-')}</code>\n"
        reply_line = f"  📤 {html.escape(reply)}"
        if elapsed: reply_line += f" · {elapsed}ms"
        if source:  reply_line += f" · {source}"
        block += reply_line
        lines.append(block)

    text = "\n\n".join(lines)
    nav = []
    if page > 0:
        nav.append({"text": "◀ Newer", "callback_data": f"adm:log:{page - 1}"})
    nav.append({"text": f"{page + 1}/{total_pages}", "callback_data": "adm:noop"})
    if page < total_pages - 1:
        nav.append({"text": "Older ▶", "callback_data": f"adm:log:{page + 1}"})
    kb = {"inline_keyboard": [nav]} if len(nav) > 1 else None
    return text, kb


def render_users(page=0, page_size=8):
    users = sorted(UNIQUE_USERS.items(), key=lambda kv: -kv[1]["last_seen"])
    total = len(users)
    total_pages = max(1, (total + page_size - 1) // page_size)
    page = max(0, min(page, total_pages - 1))
    chunk = users[page * page_size:(page + 1) * page_size]
    if not chunk:
        return "👥 <b>All Users</b>\n\nNo users yet.", None

    lines = [f"👥 <b>All Users</b> — page {page + 1}/{total_pages} ({total} total)\n"]
    for uid, info in chunk:
        username = info.get("username"); first = info.get("first_name") or ""
        name_str = f"@{username}" if username else (first or "(no name)")
        ban_mark = " 🚫" if _is_banned(uid) else ""
        lines.append(
            f"<b>{html.escape(name_str)}</b>{ban_mark} · <code>{uid}</code>\n"
            f"  Requests: {info['count']} · Last: {_fmt_ts(info['last_seen'])}"
        )
    text = "\n\n".join(lines)
    nav = []
    if page > 0:
        nav.append({"text": "◀ Prev", "callback_data": f"adm:users:{page - 1}"})
    nav.append({"text": f"{page + 1}/{total_pages}", "callback_data": "adm:noop"})
    if page < total_pages - 1:
        nav.append({"text": "Next ▶", "callback_data": f"adm:users:{page + 1}"})
    kb = {"inline_keyboard": [nav]} if len(nav) > 1 else None
    return text, kb


def render_top() -> str:
    if not PHONE_COUNTS:
        return "🔍 <b>Top Searched Numbers</b>\n\nNo searches yet."
    lines = ["🔍 <b>Top Searched Numbers</b>\n"]
    for i, (num, cnt) in enumerate(PHONE_COUNTS.most_common(10), 1):
        lines.append(f"{i}. <code>{num}</code> — {cnt} time(s)")
    return "\n".join(lines)


def render_banned() -> str:
    with _state_lock:
        if not BANNED_USERS:
            return "🚫 <b>Banned Users</b>\n\nNo banned users."
        lines = [f"🚫 <b>Banned Users</b> ({len(BANNED_USERS)})\n"]
        for uid in list(BANNED_USERS)[:30]:
            reason = BAN_REASONS.get(uid) or "—"
            lines.append(f"• {_short_user(uid)} — {html.escape(reason)}")
    return "\n".join(lines)

# ─────────────────────────────────────────────
# TELEGRAM 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)
    kw = {k: v for k, v in kw.items() if v is not None}
    try:
        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}")
    except Exception as e:
        log.error(f"❌ send error: {e!r}")
    return None


async def edit(chat_id, message_id, text, **kw):
    kw.setdefault("parse_mode", "HTML")
    kw.setdefault("disable_web_page_preview", True)
    kw = {k: v for k, v in kw.items() if v is not None}
    try:
        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}")
    except Exception as e:
        log.error(f"❌ edit error: {e!r}")


async def send_document(chat_id: int, filename: str, content: bytes, caption: str = ""):
    try:
        files = {"document": (filename, content, "application/octet-stream")}
        data = {"chat_id": str(chat_id)}
        if caption:
            data["caption"] = caption
        async with httpx.AsyncClient(timeout=60) as client:
            r = await client.post(f"{TELEGRAM_API}/sendDocument", data=data, files=files)
            res = r.json()
            if not res.get("ok"):
                log.warning(f"⚠️ sendDocument failed: {res}")
            return res
    except Exception as e:
        log.error(f"❌ sendDocument error: {e!r}")
        return {"ok": False}

# ─────────────────────────────────────────────
# INPUT NORMALIZATION
# ─────────────────────────────────────────────
def normalize_phone(text: str) -> tuple[Optional[str], Optional[str]]:
    raw = text.strip()
    digits = "".join(ch for ch in raw if ch.isdigit())
    if not digits:
        return None, "not a valid number"
    if len(digits) == 11 and digits.startswith("0"):
        digits = digits[1:]
    if len(digits) == 12 and digits.startswith("91"):
        digits = digits[2:]
    elif len(digits) == 11 and digits.startswith("91"):
        digits = digits[1:]
    if len(digits) == 10:
        if PHONE_REGEX.match(digits):
            return digits, None
        return None, "not a valid 10-digit number"
    return None, f"expected 10 digits, got {len(digits)}"

# ─────────────────────────────────────────────
# STATE
# ─────────────────────────────────────────────
SEARCH_CACHE = {}
USER_LAST = {}
STATS = {"queries": 0, "timeouts": 0, "errors": 0, "started": time.time()}


def cache_gc():
    global PEAK_CONCURRENT
    while True:
        time.sleep(60)
        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)
        with _activity_lock:
            cutoff = now - LIVE_WINDOW
            for uid in list(ACTIVE_USERS.keys()):
                if ACTIVE_USERS[uid] < cutoff:
                    del ACTIVE_USERS[uid]
            if len(ACTIVE_USERS) > PEAK_CONCURRENT:
                PEAK_CONCURRENT = len(ACTIVE_USERS)

# ─────────────────────────────────────────────
# CORE SEARCH — PROGRESSIVE FLOW
# ─────────────────────────────────────────────
async def _db_search_async(phone: str, loop) -> list:
    """Blocking DuckDB search wrapped for asyncio, with graceful failure."""
    if not _con_ready.is_set():
        return []
    try:
        return await asyncio.wait_for(
            loop.run_in_executor(None, run_search_safe, phone, FETCH_LIMIT),
            timeout=QUERY_TIMEOUT + 10,
        )
    except (asyncio.TimeoutError, TimeoutError):
        STATS["timeouts"] += 1
        log.warning(f"⏱️ DB timeout for {phone}")
        return []
    except Exception as e:
        STATS["errors"] += 1
        log.warning(f"⚠️ DB error for {phone}: {e!r}")
        return []


async def _show_progressive(chat_id, message_id, phone, results, elapsed, source_note=""):
    """Create a new search entry & edit the message. Returns search_id."""
    search_id = uuid.uuid4().hex[:10]
    total_pages = max(1, (len(results) + PAGE_SIZE - 1) // PAGE_SIZE)
    SEARCH_CACHE[search_id] = {
        "ts": time.time(), "results": results, "query": phone,
        "elapsed": elapsed, "source_note": source_note,
    }
    try:
        await edit(
            chat_id, message_id,
            render_page(results, 0, total_pages, phone, elapsed, source_note),
            reply_markup=page_keyboard(search_id, 0, total_pages),
        )
    except Exception as e:
        log.warning(f"⚠️ progressive render failed: {e!r}")
    return search_id


async def _do_progressive_search(phone: str, chat_id: int, placeholder_id: int, user: dict):
    """
    Progressive search:
      1. Fire both API and DuckDB in parallel.
      2. Show API results as soon as they arrive.
      3. Edit the SAME message when DuckDB (HiTeck + Flipkart) returns with more data.
    """
    t0 = time.time()
    loop = asyncio.get_running_loop()

    db_task  = asyncio.create_task(_db_search_async(phone, loop))
    api_task = asyncio.create_task(search_apis(phone))

    # ── Phase 1: API results (fast) ──
    api_results = []
    try:
        api_results = await asyncio.wait_for(api_task, timeout=75)
    except asyncio.TimeoutError:
        log.warning(f"⏱️ API timeout for {phone}")
    except Exception as e:
        log.warning(f"⚠️ API error for {phone}: {e!r}")

    api_deduped = dedup_records(api_results)
    api_elapsed = time.time() - t0
    search_id = None

    if api_deduped:
        search_id = await _show_progressive(
            chat_id, placeholder_id, phone, api_deduped, api_elapsed,
            source_note="⚡ Fast results — querying database for more…",
        )
    else:
        try:
            await edit(
                chat_id, placeholder_id,
                f"🔍 <b>Querying database…</b>\n\n"
                f"📱 <code>{html.escape(phone)}</code>\n"
                f"⏱️ {api_elapsed:.1f}s · APIs returned nothing, trying DuckDB…",
            )
        except Exception:
            pass

    # ── Phase 2: DuckDB results (HiTeck + Flipkart) ──
    db_results = []
    try:
        db_results = await db_task
    except Exception as e:
        log.warning(f"⚠️ DB task raised for {phone}: {e!r}")

    elapsed = time.time() - t0
    combined = dedup_records(api_results + db_results)
    new_from_db = max(0, len(combined) - len(api_deduped))

    if api_results and db_results:
        source = "hybrid"
    elif db_results:
        source = "duckdb"
    elif api_results:
        source = "api"
    else:
        source = "empty"

    # ── No results anywhere ──
    if not combined:
        _log_activity(user, "search", phone, "no_results",
                      "no records", int(elapsed * 1000), source)
        try:
            await edit(
                chat_id, placeholder_id,
                f"❌ <b>No record found</b>\n\n"
                f"📱 <code>{html.escape(phone)}</code>\n"
                f"⏱️ {elapsed:.1f}s",
            )
        except Exception:
            pass
        return 0, elapsed, source

    # ── Final render ──
    note = f"➕ {new_from_db} more from database" if (new_from_db > 0 and api_deduped) else ""
    total_pages = max(1, (len(combined) + PAGE_SIZE - 1) // PAGE_SIZE)

    if search_id:
        # Same message, mutated cache entry
        SEARCH_CACHE[search_id]["results"] = combined
        SEARCH_CACHE[search_id]["elapsed"] = elapsed
        SEARCH_CACHE[search_id]["source_note"] = note
        SEARCH_CACHE[search_id]["ts"] = time.time()
        try:
            await edit(
                chat_id, placeholder_id,
                render_page(combined, 0, total_pages, phone, elapsed, note),
                reply_markup=page_keyboard(search_id, 0, total_pages),
            )
        except Exception as e:
            log.warning(f"⚠️ final render failed: {e!r}")
    else:
        search_id = await _show_progressive(
            chat_id, placeholder_id, phone, combined, elapsed, note,
        )

    STATS["queries"] += 1
    _add_to_history(user.get("id"), phone, len(combined), elapsed, source)
    _log_activity(user, "search", phone, "results",
                  f"{len(combined)} result(s) (+{new_from_db} new)",
                  int(elapsed * 1000), source)
    return len(combined), elapsed, source


async def _search_quiet(phone: str) -> tuple[int, bool]:
    """For batch mode — no UI updates, returns (count, ok)."""
    loop = asyncio.get_running_loop()
    api_task = asyncio.create_task(search_apis(phone))
    db_task  = asyncio.create_task(_db_search_async(phone, loop))

    api_results, db_results = [], []
    try:
        api_results = await asyncio.wait_for(api_task, timeout=25)
    except Exception:
        pass
    try:
        db_results = await asyncio.wait_for(db_task, timeout=max(5, BATCH_TIMEOUT - 25))
    except Exception:
        pass

    combined = dedup_records(api_results + db_results)
    return len(combined), len(combined) > 0

# ─────────────────────────────────────────────
# HANDLERS
# ─────────────────────────────────────────────
WELCOME_TEMPLATE = (
    "👋 <b>Phone Lookup Bot</b> v{v}\n\n"
    "Send me any <b>10-digit mobile number</b> and I'll return every "
    "record we have on file.\n\n"
    "📌 <b>Examples</b>\n"
    "• <code>9876543210</code>\n"
    "• <code>+91 98765 43210</code>\n"
    "• <code>9876543210,9876543211</code> (batch)\n\n"
    "<i>Commands:</i>\n"
    "• /start — this message\n"
    "• /info — bot info & credits\n"
    "• /status — bot health\n"
    "• /history — your last 10 searches\n"
    "• /myid — your Telegram ID"
)


async def handle_message(msg: dict):
    try:
        chat_id = msg["chat"]["id"]
        user = msg.get("from") or {}
        user_id = user.get("id")
        text = (msg.get("text") or "").strip()
        log.info(f"💬 Message from {user_id}: {text[:50]!r}")

        if not text or user_id is None:
            return

        # ── BAN CHECK ──
        if _is_banned(user_id):
            with _state_lock:
                reason = BAN_REASONS.get(user_id) or "no reason given"
            await send(chat_id, f"🚫 <b>You are banned.</b>\n\nReason: {html.escape(reason)}")
            _log_activity(user, "banned_hit", text[:40], "blocked", reason)
            return

        # ── ADMIN TOKEN INPUT ──
        state = _admin_state_get(user_id)
        if state and _is_admin_session(user_id) is False:
            if state == "await_admin_token":
                _admin_state_clear(user_id)
                if ADMIN_TOKEN and text == ADMIN_TOKEN:
                    _grant_admin(user_id)
                    _log_activity(user, "admin_login", "***", "granted", "session started")
                    await send(chat_id, "✅ <b>Admin access granted</b>",
                               reply_markup=admin_menu_kb())
                    await send(chat_id, render_admin_menu())
                else:
                    _log_activity(user, "admin_login", "***", "denied", "invalid token")
                    await send(chat_id, "❌ <b>Invalid token.</b>\nTry /admin again.")
                return

        # ── ADMIN INPUT CAPTURE ──
        if state and _is_admin_session(user_id):
            _admin_state_clear(user_id)

            if state == "await_ban_id":
                parts = text.split(None, 1)
                target = parts[0].lstrip("@")
                reason = parts[1] if len(parts) > 1 else ""
                try:
                    tid = int(target)
                except ValueError:
                    await send(chat_id, "❗ Invalid user ID.")
                    return
                _ban_user(tid, reason)
                _log_activity(user, "admin", f"/ban {tid}", "banned", reason or "no reason")
                await send(chat_id, f"🚫 Banned <code>{tid}</code> — {html.escape(reason or 'no reason')}")
                return

            if state == "await_unban_id":
                try:
                    tid = int(text.strip())
                except ValueError:
                    await send(chat_id, "❗ Invalid user ID.")
                    return
                ok = _unban_user(tid)
                _log_activity(user, "admin", f"/unban {tid}", "unbanned")
                await send(chat_id, "✅ Unbanned." if ok else "ℹ️ User was not banned.")
                return

            if state == "await_broadcast":
                asyncio.create_task(_run_broadcast(chat_id, text))
                _log_activity(user, "admin", "/broadcast", "sending", f"{len(text)} chars")
                return

            if state == "await_logphone":
                phone, hint = normalize_phone(text)
                if not phone:
                    await send(chat_id, f"❗ {hint}")
                    return
                entries = [e for e in ACTIVITY_LOG
                           if e["action"] == "search" and e["input"] == phone]
                entries.reverse()
                if not entries:
                    await send(chat_id, f"🔎 No searches logged for <code>{phone}</code>.")
                    return
                lines = [f"🔎 <b>Searches for</b> <code>{phone}</code> ({len(entries)} total)\n"]
                for e in entries[:20]:
                    uname = e.get("username"); first = e.get("first_name")
                    nm = f"@{uname}" if uname else (first or "(anon)")
                    lines.append(f"• {_fmt_ts(e['ts'])} · {html.escape(nm)} · <code>{e['user_id']}</code>")
                await send(chat_id, "\n".join(lines))
                _log_activity(user, "admin", f"logphone {phone}", "text")
                return

            if state == "await_maint_reason":
                _set_maintenance(True, text[:100])
                _log_activity(user, "admin", "maintenance on", "state", text[:100])
                await send(chat_id, f"🔴 <b>Maintenance ON</b>\nReason: {html.escape(text[:100])}")
                return

        # ── MAINTENANCE GATE ──
        m_on, m_reason = _get_maintenance()
        if m_on and not _is_admin_session(user_id):
            if text in ("/start", "/help", "/status", "/info", "/myid"):
                pass
            else:
                await send(chat_id, f"🛠️ <b>Bot under maintenance</b>\n\n{html.escape(m_reason)}")
                _log_activity(user, "maintenance_block", text[:40], "blocked")
                return

        # ── /admin ──
        if text == "/admin":
            if not ADMIN_TOKEN:
                await send(chat_id, "⚠️ Admin panel is not configured.")
                return
            if _is_admin_session(user_id):
                _log_activity(user, "admin", "/admin", "menu")
                await send(chat_id, render_admin_menu(), reply_markup=admin_menu_kb())
            else:
                _admin_state_set(user_id, "await_admin_token")
                await send(chat_id, "🔐 <b>Admin access required</b>\n\n"
                                    "Send the admin token to continue. "
                                    "It expires in 3 minutes.")
            return

        # ── /logout ──
        if text == "/logout":
            if _is_admin_session(user_id):
                _revoke_admin(user_id)
                _log_activity(user, "admin_logout", "/logout", "revoked")
                await send(chat_id, "✅ Admin session ended.")
            else:
                await send(chat_id, "You are not logged in as admin.")
            return

        # ── ADMIN COMMANDS ──
        if text.startswith("/") and _is_admin_session(user_id):
            cmd = text.split()[0].lower()
            args = text.split()[1:]

            if cmd == "/live":
                _log_activity(user, "admin", "/live", "text")
                await send(chat_id, render_live()); return
            if cmd == "/stats":
                _log_activity(user, "admin", "/stats", "text")
                await send(chat_id, render_stats()); return
            if cmd == "/users":
                _log_activity(user, "admin", "/users", "text")
                t, kb = render_users(0); await send(chat_id, t, reply_markup=kb); return
            if cmd == "/top":
                _log_activity(user, "admin", "/top", "text")
                await send(chat_id, render_top()); return
            if cmd == "/log":
                entries = list(ACTIVITY_LOG); entries.reverse()
                label = ""
                if args:
                    try:
                        tid = int(args[0])
                        entries = [e for e in entries if e["user_id"] == tid]
                        label = f" (user {tid})"
                    except ValueError:
                        target = args[0].lstrip("@").lower()
                        entries = [e for e in entries if (e.get("username") or "").lower() == target]
                        label = f" (@{target})"
                _log_activity(user, "admin", text, "log_view")
                t, kb = render_log(entries, 0)
                if label: t = t.replace("<b>Activity Log</b>", f"<b>Activity Log{label}</b>", 1)
                await send(chat_id, t, reply_markup=kb); return
            if cmd == "/banned":
                _log_activity(user, "admin", "/banned", "text")
                await send(chat_id, render_banned()); return
            if cmd == "/ban":
                if not args:
                    _admin_state_set(user_id, "await_ban_id")
                    await send(chat_id, "Send user ID (and optional reason):")
                    return
                tid_s = args[0].lstrip("@")
                reason = " ".join(args[1:]) if len(args) > 1 else ""
                try: tid = int(tid_s)
                except ValueError:
                    await send(chat_id, "❗ Invalid user ID."); return
                _ban_user(tid, reason)
                _log_activity(user, "admin", f"/ban {tid}", "banned", reason or "no reason")
                await send(chat_id, f"🚫 Banned <code>{tid}</code> — {html.escape(reason or 'no reason')}")
                return
            if cmd == "/unban":
                if not args:
                    _admin_state_set(user_id, "await_unban_id")
                    await send(chat_id, "Send user ID to unban:")
                    return
                try: tid = int(args[0])
                except ValueError:
                    await send(chat_id, "❗ Invalid user ID."); return
                ok = _unban_user(tid)
                _log_activity(user, "admin", f"/unban {tid}", "unbanned")
                await send(chat_id, "✅ Unbanned." if ok else "ℹ️ User was not banned.")
                return
            if cmd == "/broadcast":
                if not args:
                    _admin_state_set(user_id, "await_broadcast")
                    await send(chat_id, "Send the message to broadcast to all users:")
                    return
                await send(chat_id, "ℹ️ Use /broadcast with no args, then send the message.")
                return
            if cmd == "/export":
                _log_activity(user, "admin", "/export", "file")
                await _send_log_export(chat_id); return
            if cmd == "/logphone":
                if not args:
                    _admin_state_set(user_id, "await_logphone")
                    await send(chat_id, "Send a 10-digit number to search the log:")
                    return
                phone, hint = normalize_phone(args[0])
                if not phone:
                    await send(chat_id, f"❗ {hint}"); return
                entries = [e for e in ACTIVITY_LOG
                           if e["action"] == "search" and e["input"] == phone]
                entries.reverse()
                if not entries:
                    await send(chat_id, f"🔎 No searches for <code>{phone}</code>."); return
                lines = [f"🔎 <b>Searches for</b> <code>{phone}</code> ({len(entries)})\n"]
                for e in entries[:20]:
                    uname = e.get("username"); first = e.get("first_name")
                    nm = f"@{uname}" if uname else (first or "(anon)")
                    lines.append(f"• {_fmt_ts(e['ts'])} · {html.escape(nm)} · <code>{e['user_id']}</code>")
                await send(chat_id, "\n".join(lines)); return
            if cmd == "/maintenance":
                if not args:
                    m_on, m_reason = _get_maintenance()
                    st = "ON" if m_on else "OFF"
                    await send(chat_id, f"🛠️ Maintenance: <b>{st}</b>"
                                        + (f"\nReason: {html.escape(m_reason)}" if m_on else ""))
                    return
                mode = args[0].lower()
                if mode == "on":
                    reason = " ".join(args[1:]) if len(args) > 1 else ""
                    if reason:
                        _set_maintenance(True, reason)
                        _log_activity(user, "admin", "maintenance on", "state", reason)
                        await send(chat_id, f"🔴 Maintenance ON — {html.escape(reason)}")
                    else:
                        _admin_state_set(user_id, "await_maint_reason")
                        await send(chat_id, "Send a short reason for maintenance:")
                    return
                elif mode == "off":
                    _set_maintenance(False)
                    _log_activity(user, "admin", "maintenance off", "state")
                    await send(chat_id, "🟢 Maintenance OFF"); return
                await send(chat_id, "Usage: /maintenance on|off [reason]"); return

        # ── /status ──
        if text == "/status":
            uptime = (time.time() - STATS["started"]) / 60
            m_on, _ = _get_maintenance()
            _log_activity(user, "status", "/status", "text")
            await send(chat_id, (
                f"<b>📊 Bot Status</b> — v{VERSION}\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"📂 Files: <code>{len(PARQUET_URLS)}</code> HiTeck + "
                f"<code>{len(FLIPKART_URLS)}</code> Flipkart\n"
                f"🔗 APIs: <code>850CR + IndData</code>\n"
                f"🛠️ Maintenance: <code>{m_on}</code>\n"
                f"🔍 Queries: <code>{STATS['queries']}</code>\n"
                f"⏰ Timeouts: <code>{STATS['timeouts']}</code>\n"
                f"❌ Errors: <code>{STATS['errors']}</code>"
            )); return

        # ── /info ──
        if text == "/info":
            uptime = (time.time() - STATS["started"]) / 60
            s = _get_stats()
            _log_activity(user, "info", "/info", "text")
            await send(chat_id, (
                f"ℹ️ <b>About This Bot</b>\n"
                "━━━━━━━━━━━━━━━━━━\n"
                f"🤖 Version: <code>{VERSION}</code>\n"
                f"📊 Sources: HiTeck (7) + Flipkart (1) + 2 APIs\n"
                f"👥 Unique users: <code>{s['unique_users']}</code>\n"
                f"🔍 Searches: <code>{s['total_searches']}</code>\n"
                f"⏱️ Avg latency (today): "
                f"<code>{_avg_latency_today():.1f}s</code>\n"
                f"🕒 Uptime: <code>{uptime:.0f} min</code>\n"
                "━━━━━━━━━━━━━━━━━━\n"
                "Built with ❤️ for fast, private lookups."
            )); return

        # ── /history ──
        if text == "/history":
            with _activity_lock:
                h = list(USER_HISTORY.get(user_id, []))
            _log_activity(user, "history", "/history", "text")
            if not h:
                await send(chat_id, "📭 <b>No history yet</b>\n\nSearch a number to start.")
                return
            rows = []
            for i, item in enumerate(h, 1):
                rows.append([{
                    "text": f"{i}. {item['phone']} ({item['count']} res)",
                    "callback_data": f"h:{item['phone']}"
                }])
            await send(chat_id,
                       "🕘 <b>Your last searches</b>\n\nTap to search again:",
                       reply_markup={"inline_keyboard": rows})
            return

        # ── /myid ──
        if text == "/myid":
            _log_activity(user, "myid", "/myid", "text")
            await send(chat_id, f"🆔 <b>Your Telegram ID</b>\n<code>{user_id}</code>")
            return

        # ── /start, /help ──
        if text in ("/start", "/help"):
            _log_activity(user, "start", text, "text")
            await send(chat_id, WELCOME_TEMPLATE.format(v=VERSION))
            if not _warmed.is_set() and not _warm_thread_started:
                threading.Thread(target=warm_phone_cache, daemon=True).start()
            return

        # ── BATCH SEARCH ──
        if "," in text:
            parts = [p.strip() for p in text.split(",") if p.strip()]
            if len(parts) < 2:
                await send(chat_id, "❌ <b>Batch format</b>\n\nExample: <code>9876543210,9876543211</code>")
                return
            if len(parts) > BATCH_MAX:
                await send(chat_id, f"❌ Max {BATCH_MAX} numbers per batch.")
                return
            phones = []
            for p in parts:
                ph, hint = normalize_phone(p)
                if not ph:
                    await send(chat_id, f"❌ Invalid number: <code>{html.escape(p)}</code> — {hint}")
                    return
                phones.append(ph)
            await _handle_batch(chat_id, user, phones)
            return

        # ── SINGLE SEARCH ──
        phone, hint = normalize_phone(text)
        if not phone:
            _log_activity(user, "invalid", text, "error", hint)
            await send(chat_id,
                       "❌ <b>Invalid input</b>\n\n"
                       "Send a 10-digit number.\n"
                       "Also accepts: <code>+91 98765 43210</code>, "
                       "<code>09876543210</code>, <code>9876543210,9876543211</code>")
            return

        if text.strip() != phone and not text.strip().isdigit():
            await send(chat_id, f"ℹ️ Interpreted as <code>{phone}</code>")

        # Cooldown
        now = time.time()
        if now - USER_LAST.get(user_id, 0) < USER_COOLDOWN:
            wait = USER_COOLDOWN - (now - USER_LAST.get(user_id, 0))
            _log_activity(user, "cooldown", phone, "text", f"wait {wait:.0f}s")
            await send(chat_id, f"⏳ Wait {wait:.0f}s before next search.")
            return
        USER_LAST[user_id] = now

        if not _warmed.is_set() and not _warm_thread_started:
            threading.Thread(target=warm_phone_cache, daemon=True).start()

        placeholder = await send(
            chat_id,
            f"🔍 <b>Searching…</b>\n\n📱 <code>{html.escape(phone)}</code>\n\n"
            f"<i>Fast results appear instantly; database enrich follows.</i>"
        )
        if placeholder is None:
            return

        try:
            await _do_progressive_search(phone, chat_id, placeholder, user)
        except asyncio.TimeoutError:
            STATS["timeouts"] += 1
            _log_activity(user, "search", phone, "timeout", "progressive timeout")
            await edit(chat_id, placeholder, f"⏱️ Timed out. Try again.")
        except Exception as e:
            STATS["errors"] += 1
            log.error(f"❌ progressive search error: {e!r}\n{traceback.format_exc()}")
            _log_activity(user, "search", phone, "error", str(e)[:100])
            await edit(chat_id, placeholder, f"❌ Error: <code>{html.escape(str(e)[:200])}</code>")
    except Exception as e:
        log.error(f"❌ handle_message fatal: {e!r}\n{traceback.format_exc()}")


async def _handle_batch(chat_id, user, phones):
    await send(chat_id, f"🔍 <b>Batch search</b> — {len(phones)} number(s)…")
    t0 = time.time()
    tasks = [asyncio.create_task(_search_quiet(p)) for p in phones]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    total_elapsed = time.time() - t0

    lines = [f"📊 <b>Batch results</b> ({len(phones)} numbers) — {total_elapsed:.1f}s\n"]
    for phone, res in zip(phones, results):
        if isinstance(res, Exception):
            lines.append(f"❌ <code>{phone}</code> — error: {html.escape(str(res)[:60])}")
        else:
            n, ok = res
            if ok:
                lines.append(f"✅ <code>{phone}</code> — {n} record(s)")
            else:
                lines.append(f"❌ <code>{phone}</code> — no results")

    kb_rows = [[{"text": f"View {phone}", "callback_data": f"h:{phone}"}] for phone in phones]
    _log_activity(user, "batch_search", ",".join(phones), "text",
                  f"{len(phones)} nums in {total_elapsed:.1f}s")
    await send(chat_id, "\n".join(lines),
               reply_markup={"inline_keyboard": kb_rows})


async def _send_log_export(chat_id: int):
    entries = list(ACTIVITY_LOG)
    if not entries:
        await send(chat_id, "📭 No activity to export.")
        return
    buf = io.StringIO()
    w = csv.writer(buf)
    w.writerow(["timestamp", "user_id", "username", "first_name",
                "action", "input", "reply_type", "summary", "elapsed_ms", "source"])
    for e in entries:
        w.writerow([
            time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(e["ts"])),
            e["user_id"], e.get("username") or "", e.get("first_name") or "",
            e["action"], e.get("input", ""), e.get("reply_type", ""),
            e.get("summary", ""), e.get("elapsed_ms", 0), e.get("source", ""),
        ])
    csv_bytes = buf.getvalue().encode("utf-8")
    ts = time.strftime("%Y%m%d_%H%M%S")
    await send_document(chat_id, f"activity_log_{ts}.csv", csv_bytes,
                        caption=f"📥 Activity log — {len(entries)} entries")


async def _run_broadcast(admin_chat_id: int, message: str):
    with _activity_lock:
        recipients = list(UNIQUE_USERS.keys())
    if not recipients:
        await send(admin_chat_id, "📭 No users to broadcast to.")
        return
    await send(admin_chat_id, f"📢 Broadcasting to <b>{len(recipients)}</b> users…")
    sent = failed = 0
    for uid in recipients:
        if _is_banned(uid):
            continue
        try:
            res = await tg("sendMessage", {
                "chat_id": uid,
                "text": f"📢 <b>Announcement</b>\n\n{message}",
                "parse_mode": "HTML",
                "disable_web_page_preview": True,
            }, timeout=15)
            if res.get("ok"):
                sent += 1
            else:
                failed += 1
        except Exception:
            failed += 1
        await asyncio.sleep(BROADCAST_DELAY)
    await send(admin_chat_id,
               f"✅ <b>Broadcast complete</b>\n\n"
               f"Sent: <code>{sent}</code>\n"
               f"Failed: <code>{failed}</code>")
    _log_activity({"id": 0}, "admin", "broadcast", "done",
                  f"sent={sent} failed={failed}")


def _avg_latency_today() -> float:
    now = time.time()
    today_start = now - (now % 86400)
    with _activity_lock:
        vals = [e["elapsed_ms"] for e in ACTIVITY_LOG
                if e["action"] == "search" and e["ts"] >= today_start and e["elapsed_ms"] > 0]
    if not vals:
        return 0.0
    return (sum(vals) / len(vals)) / 1000.0


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")
        user = cb.get("from") or {}
        user_id = user.get("id")

        if not chat_id or not message_id:
            await tg("answerCallbackQuery", {"callback_query_id": cb_id})
            return

        if _is_banned(user_id):
            await tg("answerCallbackQuery", {
                "callback_query_id": cb_id, "text": "🚫 You are banned.",
                "show_alert": True,
            }); return

        # ── Admin callbacks ──
        if data.startswith("adm:"):
            if not _is_admin_session(user_id):
                await tg("answerCallbackQuery", {
                    "callback_query_id": cb_id,
                    "text": "Admin session expired — send /admin again.",
                    "show_alert": True,
                }); return
            parts = data.split(":")
            action = parts[1] if len(parts) > 1 else "menu"

            if action == "menu":
                await edit(chat_id, message_id, render_admin_menu(), reply_markup=admin_menu_kb())
            elif action == "live":
                await edit(chat_id, message_id, render_live(), reply_markup=admin_menu_kb())
            elif action == "stats":
                await edit(chat_id, message_id, render_stats(), reply_markup=admin_menu_kb())
            elif action == "top":
                await edit(chat_id, message_id, render_top(), reply_markup=admin_menu_kb())
            elif action == "banned":
                await edit(chat_id, message_id, render_banned(), reply_markup=admin_menu_kb())
            elif action == "log":
                page = int(parts[2]) if len(parts) > 2 else 0
                entries = list(ACTIVITY_LOG); entries.reverse()
                t, kb = render_log(entries, page)
                await edit(chat_id, message_id, t, reply_markup=kb)
            elif action == "users":
                page = int(parts[2]) if len(parts) > 2 else 0
                t, kb = render_users(page)
                await edit(chat_id, message_id, t, reply_markup=kb)
            elif action == "ban":
                _admin_state_set(user_id, "await_ban_id")
                await edit(chat_id, message_id, "🚫 Send user ID (and optional reason):")
            elif action == "unban":
                _admin_state_set(user_id, "await_unban_id")
                await edit(chat_id, message_id, "✅ Send user ID to unban:")
            elif action == "broadcast":
                _admin_state_set(user_id, "await_broadcast")
                await edit(chat_id, message_id, "📢 Send the message to broadcast:")
            elif action == "export":
                await edit(chat_id, message_id, "📥 Generating export…",
                           reply_markup=admin_menu_kb())
                await _send_log_export(chat_id)
            elif action == "logphone":
                _admin_state_set(user_id, "await_logphone")
                await edit(chat_id, message_id, "🔎 Send a 10-digit number:")
            elif action == "maint":
                m_on, _ = _get_maintenance()
                if m_on:
                    _set_maintenance(False)
                    await edit(chat_id, message_id, "🟢 Maintenance OFF",
                               reply_markup=admin_menu_kb())
                else:
                    _admin_state_set(user_id, "await_maint_reason")
                    await edit(chat_id, message_id, "Send a short maintenance reason:")
            elif action == "noop":
                pass

            await tg("answerCallbackQuery", {"callback_query_id": cb_id})
            _log_activity(user, "admin", data, "menu")
            return

        # ── History re-run ──
        if data.startswith("h:"):
            phone = data.split(":", 1)[1]
            await tg("answerCallbackQuery", {
                "callback_query_id": cb_id,
                "text": f"Searching {phone}…",
            })
            fake_msg = {"chat": {"id": chat_id}, "from": user, "text": phone}
            await handle_message(fake_msg)
            return

        # ── Copy page ──
        if data.startswith("c:"):
            try:
                _, sid, page_s = data.split(":")
                page = int(page_s)
            except Exception:
                await tg("answerCallbackQuery", {"callback_query_id": cb_id})
                return
            entry = SEARCH_CACHE.get(sid)
            if not entry:
                await tg("answerCallbackQuery", {
                    "callback_query_id": cb_id,
                    "text": "Session expired.", "show_alert": True,
                }); return
            results = entry["results"]
            total_pages = max(1, (len(results) + PAGE_SIZE - 1) // PAGE_SIZE)
            page = max(0, min(page, total_pages - 1))
            start = page * PAGE_SIZE
            chunk = results[start:start + PAGE_SIZE]
            text = "\n\n".join(_plain_record(start + i + 1, r)
                               for i, r in enumerate(chunk))
            await send(chat_id, f"<b>📋 Copy text for page {page + 1}</b>\n\n"
                                f"<pre>{html.escape(text[:3500])}</pre>")
            await tg("answerCallbackQuery", {"callback_query_id": cb_id,
                                              "text": "Sent as a copyable message"})
            _log_activity(user, "copy", entry["query"], "text")
            return

        # ── Export ──
        if data.startswith("e:"):
            sid = data.split(":", 1)[1]
            entry = SEARCH_CACHE.get(sid)
            if not entry:
                await tg("answerCallbackQuery", {
                    "callback_query_id": cb_id,
                    "text": "Session expired.", "show_alert": True,
                }); return
            phone = entry["query"]
            results = entry["results"]
            payload = json.dumps(results, indent=2, ensure_ascii=False).encode("utf-8")
            await send_document(chat_id, f"results_{phone}.json", payload,
                                caption=f"📥 {len(results)} record(s) for {phone}")
            await tg("answerCallbackQuery", {"callback_query_id": cb_id,
                                              "text": "Sent as file"})
            _log_activity(user, "export", phone, "file", f"{len(results)} records")
            return

        # ── Pagination ──
        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 = max(1, (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})
        _log_activity(user, "paginate", entry["query"], "page",
                      f"page {page + 1}/{total_pages}")
    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):
    s = _get_stats()
    return JSONResponse({
        "status": "ok",
        "version": VERSION,
        "duckdb_ready": _con_ready.is_set(),
        "warmed": _warmed.is_set(),
        "files": len(ALL_PARQUET_URLS),
        "hiteck_files": len(PARQUET_URLS),
        "flipkart_files": len(FLIPKART_URLS),
        "cached_searches": len(SEARCH_CACHE),
        "live_users": s["live_users"],
        "unique_users": s["unique_users"],
        "banned_users": s["banned_count"],
        "maintenance": s["maintenance"],
        "apis": {"cr850": API_850CR, "inddata": API_INDDATA},
        "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(f"# 📞 Phone Lookup Bot v{VERSION}")
    gr.Markdown("Telegram bot: HiTeck + Flipkart parquet + 850CR API + IndData API (progressive results).")
    status_box = gr.JSON(label="Health")
    demo.load(
        fn=lambda: {
            "status": "ok", "version": VERSION,
            "duckdb_ready": _con_ready.is_set(),
            "warmed": _warmed.is_set(),
            "files": len(ALL_PARQUET_URLS),
            "cached_searches": len(SEARCH_CACHE),
            "live_users": _get_stats()["live_users"],
            "unique_users": _get_stats()["unique_users"],
            "maintenance": _get_stats()["maintenance"],
        },
        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
# ─────────────────────────────────────────────
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()

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"]))

# ── Admin web panel — loaded from separate file if available ──
try:
    from admin_panel import admin_dashboard  # type: ignore
    live_app.router.routes.insert(0, StarletteRoute("/admin", admin_dashboard, methods=["GET"]))
    log.info("✅ /admin web dashboard loaded from admin_panel.py")
except ImportError:
    log.info("ℹ️ admin_panel.py not present — /admin web route disabled (Telegram /admin still works)")
except Exception as e:
    log.warning(f"⚠️ admin_panel.py load failed: {e!r}")

try:
    live_app.build_middleware_stack()
    log.info("✅ Middleware stack rebuilt")
except Exception as e:
    log.warning(f"⚠️ Stack rebuild: {e}")

log.info("📋 First 6 routes:")
for route in live_app.router.routes[:6]:
    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()