Spaces:

Hfzune
/

Zinx

Running on Zero

Zinx / app.py
Create app.py
06d96e5 verified 15 days ago
74.5 kB
# ══════════════════════════════════════════════════════════════════
#  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
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, HTMLResponse
from starlette.routing import Route as StarletteRoute

# ─────────────────────────────────────────────
# CONFIG
# ─────────────────────────────────────────────
VERSION        = "1.2.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 only (ICMR removed)
# ─────────────────────────────────────────────
HITECK_URLS = [
    f"https://huggingface.co/datasets/Hfzune/HiTeckgroop/resolve/main/idx_phone.{i}.parquet"
    for i in range(7)
]
PARQUET_URLS = HITECK_URLS

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
LIVE_WINDOW     = 300
HISTORY_LEN     = 10
BATCH_MAX       = 5
BATCH_TIMEOUT   = 60
BROADCAST_DELAY = 0.05  # 20 msgs/sec

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


def warm_phone_cache() -> None:
    global _warm_thread_started
    if _warmed.is_set():
        return
    _warm_thread_started = True

    log.info(f"🔥 Light warm-up starting ({len(PARQUET_URLS)} HiTeck files)...")
    t0 = time.time()
    loaded = 0
    skipped = 0

    try:
        con = get_con()
        for url in 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)")
        if loaded == 0:
            log.error("⚠️ NO files warmed — falling back to APIs only.")
    except Exception as e:
        log.error(f"⚠️ Warm-up fatal: {e}")
        _warm_thread_started = False


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()
        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
# ─────────────────────────────────────────────
async def _q_cr850(client, 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):
    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:
    log.info(f"🔄 API fallback 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

# ─────────────────────────────────────────────
# 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:
        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:
    """Plain-text version for copy button."""
    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 = []
    extras.append({"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   = {}

# ── User management ──
BANNED_USERS: set = set()
BAN_REASONS: dict = {}
USER_HISTORY: dict = {}    # user_id -> deque of {phone, ts, count, elapsed, source}

# ── System flags ──
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
# ─────────────────────────────────────────────
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}
    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)
    kw = {k: v for k, v in kw.items() if v is not None}
    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}")


async def send_document(chat_id: int, filename: str, content: bytes, caption: str = ""):
    """Send a document via multipart/form-data."""
    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 — "Did you mean?"
# ─────────────────────────────────────────────
def normalize_phone(text: str) -> tuple[Optional[str], Optional[str]]:
    """
    Returns (10-digit phone, hint_message) or (None, error).
    Accepts: 9876543210, +91 9876543210, 09876543210, 919876543210,
             98765-43210, (98765) 43210, etc.
    """
    raw = text.strip()
    digits = "".join(ch for ch in raw if ch.isdigit())
    if not digits:
        return None, "not a valid number"

    # Strip leading 0 (local format)
    if len(digits) == 11 and digits.startswith("0"):
        digits = digits[1:]

    # Strip 91 country code
    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)

# ─────────────────────────────────────────────
# BATCH SEARCH
# ─────────────────────────────────────────────
async def _do_single_search(phone: str) -> tuple[list, float, str]:
    """Returns (results, elapsed, source). Uses DuckDB if warmed, else API fallback."""
    loop = asyncio.get_running_loop()
    t0 = time.time()
    source = "duckdb"
    if not _warmed.is_set():
        results = await search_apis(phone)
        source = "api_fallback"
    else:
        results = await asyncio.wait_for(
            loop.run_in_executor(None, run_search_safe, phone, FETCH_LIMIT),
            timeout=QUERY_TIMEOUT + 10,
        )
    elapsed = time.time() - t0
    return dedup_records(results), elapsed, source

# ─────────────────────────────────────────────
# 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 (after login) ──
        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 (public blocked, admin allowed) ──
        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   # allow basic commands
            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 (session required) ──
        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 only)\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 files) + 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 (comma-separated) ──
        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

        # Show hint if user typed a non-10-digit but we still parsed it
        if text.strip() != phone and text.strip().isdigit() is False:
            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>First query may take up to {QUERY_TIMEOUT}s. Repeats are fast.</i>"
        )
        if placeholder is None:
            return

        t0 = time.time()
        try:
            results, elapsed, source_used = await _do_single_search(phone)
        except asyncio.TimeoutError:
            STATS["timeouts"] += 1
            _log_activity(user, "search", phone, "timeout", f"after {QUERY_TIMEOUT}s")
            await edit(chat_id, placeholder, f"⏱️ Timed out after {QUERY_TIMEOUT}s. Try again.")
            return
        except TimeoutError:
            STATS["timeouts"] += 1
            _log_activity(user, "search", phone, "timeout", f"after {QUERY_TIMEOUT}s")
            await edit(chat_id, placeholder, f"⏱️ Timed out after {QUERY_TIMEOUT}s. Try again.")
            return
        except Exception as e:
            STATS["errors"] += 1
            log.error(f"❌ 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>")
            return

        STATS["queries"] += 1
        log.info(f"✅ Search returned {len(results)} rows in {elapsed:.1f}s")

        _add_to_history(user_id, phone, len(results), elapsed, source_used)

        if not results:
            _log_activity(user, "search", phone, "no_results",
                          "no records found", int(elapsed * 1000), source_used)
            await edit(chat_id, placeholder,
                       f"❌ <b>No record found</b>\n\n"
                       f"📱 <code>{html.escape(phone)}</code>\n"
                       f"⏱️ {elapsed:.1f}s")
            return

        search_id = uuid.uuid4().hex[:10]
        total_pages = (len(results) + PAGE_SIZE - 1) // PAGE_SIZE
        source_note = "Using API fallback (DuckDB warming)" if source_used == "api_fallback" else ""
        SEARCH_CACHE[search_id] = {
            "ts": time.time(), "results": results, "query": phone,
            "elapsed": elapsed, "source_note": source_note,
        }

        _log_activity(user, "search", phone, "results",
                      f"{len(results)} result(s)", int(elapsed * 1000), source_used)

        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_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(_safe_batch_one(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 = []
    for phone in phones:
        kb_rows.append([{"text": f"View {phone}", "callback_data": f"h:{phone}"}])
    _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 _safe_batch_one(phone: str) -> tuple[int, bool]:
    try:
        results, _, _ = await asyncio.wait_for(_do_single_search(phone), timeout=BATCH_TIMEOUT)
        return len(results), len(results) > 0
    except Exception:
        return 0, False


async def _send_log_export(chat_id: int):
    """Generate CSV + JSON, send as documents."""
    entries = list(ACTIVITY_LOG)
    if not entries:
        await send(chat_id, "📭 No activity to export.")
        return

    # CSV
    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):
    """Background broadcast with rate limit. Sends a summary to admin."""
    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 = 0; 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

        # ── Search history re-run ──
        if data.startswith("h:"):
            phone = data.split(":", 1)[1]
            await tg("answerCallbackQuery", {
                "callback_query_id": cb_id,
                "text": f"Searching {phone}…",
            })
            # Reuse the single-search flow by faking a message
            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 = (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 = (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(PARQUET_URLS),
        "cached_searches": len(SEARCH_CACHE),
        "live_users": s["live_users"],
        "unique_users": s["unique_users"],
        "banned_users": s["banned_count"],
        "maintenance": s["maintenance"],
        "stats": STATS,
    })


def _esc_html(s):
    return html.escape(str(s)) if s is not None else ""


async def _admin_dashboard(request: Request):
    token = request.query_params.get("token", "")
    if not WEBHOOK_SECRET or token != WEBHOOK_SECRET:
        return HTMLResponse("<h1>403 — Invalid or missing token</h1>", status_code=403)

    s = _get_stats()
    uptime = (time.time() - STATS["started"]) / 60

    entries = list(ACTIVITY_LOG); entries.reverse()
    rows = []
    for e in entries[:200]:
        uname = e.get("username"); first = e.get("first_name")
        name_str = f"@{uname}" if uname else (first or "(no name)")
        rows.append(
            f"<tr>"
            f"<td>{_fmt_ts(e['ts'])}</td>"
            f"<td>{_esc_html(name_str)}</td>"
            f"<td><code>{e['user_id']}</code></td>"
            f"<td>{_esc_html(e['action'])}</td>"
            f"<td><code>{_esc_html(e['input'])}</code></td>"
            f"<td>{_esc_html(e.get('summary') or e.get('reply_type') or '-')}</td>"
            f"<td>{e.get('elapsed_ms', 0)}ms</td>"
            f"<td>{_esc_html(e.get('source') or '')}</td>"
            f"</tr>"
        )

    html_doc = f"""<!DOCTYPE html>
<html><head><meta charset="utf-8">
<title>Admin Dashboard</title>
<meta http-equiv="refresh" content="10">
<style>
  body {{ font-family: -apple-system, sans-serif; background: #f5f5f5; padding: 20px; }}
  h1 {{ margin: 0 0 10px 0; }}
  .stats {{ display: flex; gap: 20px; margin: 20px 0; flex-wrap: wrap; }}
  .card {{ background: white; padding: 16px 24px; border-radius: 8px; box-shadow: 0 1px 3px rgba(0,0,0,0.08); }}
  .card .v {{ font-size: 28px; font-weight: 600; color: #c96442; }}
  .card .l {{ font-size: 12px; color: #888; text-transform: uppercase; }}
  table {{ width: 100%; background: white; border-collapse: collapse; border-radius: 8px; overflow: hidden; box-shadow: 0 1px 3px rgba(0,0,0,0.08); }}
  th, td {{ padding: 10px 12px; text-align: left; border-bottom: 1px solid #eee; font-size: 13px; }}
  th {{ background: #fafafa; font-weight: 600; color: #555; }}
  tr:hover {{ background: #fafafa; }}
  code {{ background: #f0f0f0; padding: 2px 5px; border-radius: 3px; font-size: 12px; }}
</style></head>
<body>
  <h1>📞 Phone Lookup Bot — Admin v{VERSION}</h1>
  <p style="color:#888; font-size:13px;">Auto-refresh 10s. Uptime: {uptime:.1f} min</p>
  <div class="stats">
    <div class="card"><div class="v">{s['unique_users']}</div><div class="l">Unique Users</div></div>
    <div class="card"><div class="v">{s['live_users']}</div><div class="l">Live (5 min)</div></div>
    <div class="card"><div class="v">{s['today_users']}</div><div class="l">Active Today</div></div>
    <div class="card"><div class="v">{s['today_searches']}</div><div class="l">Searches Today</div></div>
    <div class="card"><div class="v">{s['total_searches']}</div><div class="l">Total Searches</div></div>
    <div class="card"><div class="v">{s['peak_concurrent']}</div><div class="l">Peak Concurrent</div></div>
    <div class="card"><div class="v">{s['banned_count']}</div><div class="l">Banned</div></div>
    <div class="card"><div class="v">{'ON' if s['maintenance'] else 'OFF'}</div><div class="l">Maintenance</div></div>
  </div>
  <h2 style="margin-top:30px;">Recent Activity (last 200)</h2>
  <table>
    <thead><tr>
      <th>Time</th><th>User</th><th>User ID</th><th>Action</th>
      <th>Input</th><th>Reply</th><th>Latency</th><th>Source</th>
    </tr></thead>
    <tbody>{''.join(rows) or '<tr><td colspan="8" style="text-align:center;color:#aaa;">No activity yet</td></tr>'}</tbody>
  </table>
</body></html>"""
    return HTMLResponse(html_doc)


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 streaming `Hfzune/HiTeckgroop` with API fallback.")
    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(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"]))
live_app.router.routes.insert(0, StarletteRoute("/admin", _admin_dashboard, 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 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()