#!/usr/bin/env python3
"""
rajabets_search.py — Rajabets Search API (HF parquet remote read).

Reads via DuckDB httpfs directly from HF's auto-converted parquet.
No download. Range requests only.

Endpoints:
  GET /search?q=<phone|email|username|name>
  GET /search?field=phone&q=123&mode=exact
  POST /search/parallel
  GET /warm
  GET /health
"""

import asyncio
import json
import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from typing import Any

import duckdb
import gradio as gr
import httpx
from fastapi import FastAPI, HTTPException, Query, Response
from fastapi.responses import ORJSONResponse
from pydantic import BaseModel

# ── Config ──────────────────────────────────────────────────────────────────
HF_DATASET = os.environ.get("RAJABETS_DATASET", "CutehackX/Rajabets").strip("/")
HF_CONFIG  = os.environ.get("RAJABETS_CONFIG", "default")
HF_SPLIT   = os.environ.get("RAJABETS_SPLIT", "train")

HF_PARQUET_BASE = (
    f"https://huggingface.co/datasets/{HF_DATASET}"
    f"/resolve/refs%2Fconvert%2Fparquet/{HF_CONFIG}/{HF_SPLIT}"
)
HF_PARQUET_GLOB = os.environ.get("RAJABETS_PARQUET", f"{HF_PARQUET_BASE}/*.parquet")

PARALLELISM      = int(os.environ.get("RAJABETS_PARALLEL", str(max(4, (os.cpu_count() or 2) * 2))))
THREADS_PER_CONN = int(os.environ.get("RAJABETS_THREADS_PER_CONN", "4"))
QUERY_CACHE_SIZE = int(os.environ.get("RAJABETS_QUERY_CACHE", "4096"))
QUERY_CACHE_TTL  = int(os.environ.get("RAJABETS_QUERY_TTL", "300"))
HTTP_PORT        = int(os.environ.get("PORT", "7860"))

SEARCH_FIELDS = ["id", "email", "username", "first_name", "last_name", "phone", "country", "city"]
TEXT_FIELDS   = ["email", "username", "first_name", "last_name", "country", "city"]
NUM_FIELDS    = ["phone", "id"]

HF_TOKEN = os.environ.get("HF_TOKEN", "").strip()


# ── Parquet URL discovery ───────────────────────────────────────────────────
_parquet_urls: list[str] = []
_urls_lock = threading.Lock()
_urls_ready = threading.Event()


def _discover_parquet_urls() -> list[str]:
    if _urls_ready.is_set():
        return _parquet_urls
    with _urls_lock:
        if _urls_ready.is_set():
            return _parquet_urls

        urls: list[str] = []

        # 1) HF parquet API
        api = f"https://huggingface.co/api/datasets/{HF_DATASET}/parquet"
        headers = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
        try:
            r = httpx.get(api, headers=headers, timeout=20, follow_redirects=True)
            if r.status_code == 200:
                tree = r.json()
                cfg = tree.get(HF_CONFIG, {})
                if isinstance(cfg, dict):
                    split = cfg.get(HF_SPLIT, [])
                    if isinstance(split, list):
                        for entry in split:
                            if isinstance(entry, dict):
                                u = entry.get("url") or entry.get("path")
                                if u:
                                    urls.append(u)
                            elif isinstance(entry, str):
                                urls.append(entry)
                print(f"[parquet] api returned {len(urls)} urls")
            else:
                print(f"[parquet] api status {r.status_code}")
        except Exception as e:
            print(f"[parquet] api discovery failed: {e}")

        # 2) huggingface_hub fallback
        if not urls:
            try:
                from huggingface_hub import list_repo_files, hf_hub_url
                rev = "refs/convert/parquet"
                files = list_repo_files(
                    HF_DATASET, repo_type="dataset",
                    revision=rev, token=HF_TOKEN or None,
                )
                prefix = f"{HF_CONFIG}/{HF_SPLIT}/"
                for f in files:
                    if f.startswith(prefix) and f.endswith(".parquet"):
                        urls.append(hf_hub_url(
                            HF_DATASET, f, repo_type="dataset", revision=rev,
                        ))
                print(f"[parquet] hub listed {len(urls)} files")
            except Exception as e:
                print(f"[parquet] hub fallback failed: {e}")

        # 3) last resort — glob
        if not urls:
            urls = [HF_PARQUET_GLOB]
            print(f"[parquet] fallback to glob: {HF_PARQUET_GLOB}")

        _parquet_urls.extend(urls)
        _urls_ready.set()
        print(f"[parquet] {len(urls)} file(s) resolved")
        for u in urls[:5]:
            print(f"    {u}")
        if len(urls) > 5:
            print(f"    ... and {len(urls)-5} more")
        return _parquet_urls


# ── DuckDB pool ─────────────────────────────────────────────────────────────
_conns: list[duckdb.DuckDBPyConnection] = []
_conns_lock = threading.Lock()
_thread_local = threading.local()
pool = ThreadPoolExecutor(max_workers=PARALLELISM, thread_name_prefix="duck")


def _new_conn() -> duckdb.DuckDBPyConnection:
    urls = _discover_parquet_urls()
    con = duckdb.connect()
    con.execute("SET home_directory='/tmp'")
    con.execute("SET extension_directory='/tmp/duckdb_extensions'")
    con.execute("INSTALL parquet; LOAD parquet;")
    con.execute("INSTALL httpfs; LOAD httpfs;")

    if HF_TOKEN:
        con.execute(
            f"SET http_extra_headers = MAP{{'Authorization': 'Bearer {HF_TOKEN}'}}"
        )

    if len(urls) == 1 and "*" in urls[0]:
        source = f"'{urls[0]}'"
    else:
        lst = ", ".join(f"'{u}'" for u in urls)
        source = f"[{lst}]"

    con.execute(f"""
        CREATE OR REPLACE VIEW people AS
        SELECT * FROM read_parquet({source})
    """)
    con.execute(f"SET threads = {THREADS_PER_CONN}")
    con.execute("SET enable_progress_bar = false")
    con.execute("SET http_keep_alive = true")
    return con


def _thread_id() -> int:
    tid = getattr(_thread_local, "id", None)
    if tid is None:
        with _conns_lock:
            tid = len(_conns)
            _thread_local.id = tid
    return tid


def _get_conn() -> duckdb.DuckDBPyConnection:
    ident = _thread_id()
    with _conns_lock:
        while len(_conns) <= ident:
            _conns.append(_new_conn())
    return _conns[ident]


def _warm_pool() -> None:
    _discover_parquet_urls()
    for _ in range(min(PARALLELISM, 8)):
        with _conns_lock:
            _conns.append(_new_conn())
    try:
        con = _conns[0]
        n = con.execute("SELECT COUNT(*) FROM people").fetchone()[0]
        print(f"[pool] warm ok — {n} rows")
    except Exception as e:
        print(f"[pool] warm failed: {e}")


# ── Query cache ─────────────────────────────────────────────────────────────
class _TTLCache:
    def __init__(self, maxsize: int, ttl: int):
        self._max, self._ttl = maxsize, ttl
        self._d: dict[str, tuple[float, Any]] = {}
        self._lock = threading.Lock()

    def get(self, k: str):
        now = time.time()
        with self._lock:
            v = self._d.get(k)
            if not v:
                return None
            exp, val = v
            if exp < now:
                self._d.pop(k, None)
                return None
            return val

    def set(self, k: str, val: Any):
        now = time.time()
        with self._lock:
            if len(self._d) >= self._max:
                for kk in sorted(self._d, key=lambda x: self._d[x][0])[: self._max // 10]:
                    self._d.pop(kk, None)
            self._d[k] = (now + self._ttl, val)

    def stats(self):
        with self._lock:
            return {"entries": len(self._d), "maxsize": self._max, "ttl": self._ttl}


_qcache = _TTLCache(QUERY_CACHE_SIZE, QUERY_CACHE_TTL)


# ── Search logic ────────────────────────────────────────────────────────────
def _qkey(field, value, mode, limit):
    return f"{field}|{mode}|{limit}|{value}"


def _run_field_search(field: str, value: str, mode: str, limit: int) -> dict:
    if field not in SEARCH_FIELDS:
        raise ValueError(f"unknown field: {field}")

    ck = _qkey(field, value, mode, limit)
    cached = _qcache.get(ck)
    if cached is not None:
        return cached

    v = value.replace("'", "''")

    if field in NUM_FIELDS:
        if mode == "exact":
            sql = f"SELECT * FROM people WHERE CAST({field} AS VARCHAR) = '{v}' LIMIT {limit}"
        else:
            v2 = v.replace("%", r"\%").replace("_", r"\_")
            sql = (f"SELECT * FROM people WHERE CAST({field} AS VARCHAR) "
                   f"ILIKE '%{v2}%' ESCAPE '\\' LIMIT {limit}")
    else:
        if mode == "exact":
            sql = f"SELECT * FROM people WHERE {field} = '{v}' LIMIT {limit}"
        else:
            v2 = v.replace("%", r"\%").replace("_", r"\_")
            sql = f"SELECT * FROM people WHERE {field} ILIKE '%{v2}%' ESCAPE '\\' LIMIT {limit}"

    con = _get_conn()
    rows = con.execute(sql).fetchall()
    cols = [d[0] for d in con.description]
    results = [dict(zip(cols, r)) for r in rows]
    out = {"field": field, "value": value, "mode": mode, "count": len(results), "results": results}
    _qcache.set(ck, out)
    return out


def _unified_search(q: str, limit: int = 20) -> dict:
    q = q.strip()
    if not q:
        return {"query": q, "searched_fields": [], "count": 0, "results": []}

    ck = f"u|{limit}|{q}"
    cached = _qcache.get(ck)
    if cached is not None:
        return cached

    is_num = q.isdigit() and len(q) >= 8
    all_rows, searched = [], []

    if is_num:
        r = _run_field_search("phone", q, "exact", limit)
        all_rows.extend(r["results"])
        searched.append("phone")
        if not all_rows:
            r = _run_field_search("id", q, "exact", limit)
            all_rows.extend(r["results"])
            searched.append("id")
    else:
        for f in ["email", "username", "first_name", "last_name"]:
            r = _run_field_search(f, q, "contains", limit)
            if r["results"]:
                all_rows.extend(r["results"])
                searched.append(f)
            if len(all_rows) >= limit:
                break

    seen, dedup = set(), []
    for row in all_rows:
        rid = row.get("id")
        if rid in seen:
            continue
        seen.add(rid)
        dedup.append(row)

    out = {"query": q, "searched_fields": searched,
           "count": len(dedup[:limit]), "results": dedup[:limit]}
    _qcache.set(ck, out)
    return out


# ── FastAPI ─────────────────────────────────────────────────────────────────
fastapi_app = FastAPI(title="Rajabets Search API", default_response_class=ORJSONResponse)


class BatchRequest(BaseModel):
    queries: list[dict[str, Any]]
    limit: int = 20


@fastapi_app.get("/")
def root():
    return {
        "app": "Rajabets Search API",
        "dataset": HF_DATASET,
        "mode": "hf-parquet-remote",
        "parquet_files": len(_parquet_urls) if _urls_ready.is_set() else 0,
        "parallelism": PARALLELISM,
        "fields": SEARCH_FIELDS,
        "docs": "/docs",
        "developer": "@SrcSellerOnTopBot | channel @NothingFlyXyz",
    }


@fastapi_app.get("/health")
def health():
    return {
        "status": "ok",
        "parquet_ready": _urls_ready.is_set(),
        "parquet_files": len(_parquet_urls),
        "conns": len(_conns),
        "query_cache": _qcache.stats(),
    }


@fastapi_app.get("/warm")
async def warm():
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(pool, _warm_pool)
    return {"ok": True, "parquet_files": len(_parquet_urls), "conns": len(_conns)}


@fastapi_app.get("/search")
async def search(
    q: str | None = Query(None),
    mobile: str | None = Query(None),
    field: str | None = Query(None),
    mode: str = Query("exact"),
    limit: int = Query(20, ge=1, le=200),
    pretty: bool = Query(False),
):
    q_val = (q or mobile or "").strip()
    if not q_val:
        raise HTTPException(422, "provide q or mobile")
    loop = asyncio.get_running_loop()
    if field:
        data = await loop.run_in_executor(pool, _run_field_search, field, q_val, mode, limit)
    else:
        data = await loop.run_in_executor(pool, _unified_search, q_val, limit)

    result = {"success": bool(data["count"]), **data, "number": q_val, "total": data["count"]}
    body = json.dumps(result, indent=2 if pretty else None, ensure_ascii=False)
    return Response(
        content=body,
        media_type="application/json",
        headers={"Cache-Control": "public, max-age=60"},
    )


@fastapi_app.post("/search/parallel")
async def search_parallel(req: BatchRequest):
    if not req.queries:
        raise HTTPException(400, "queries must not be empty")
    if len(req.queries) > 50:
        raise HTTPException(400, "max 50 queries per batch")
    loop = asyncio.get_running_loop()
    tasks = [
        loop.run_in_executor(
            pool, _run_field_search,
            item.get("field", "phone"),
            item.get("value", ""),
            item.get("mode", "exact"),
            int(item.get("limit", req.limit)),
        )
        for item in req.queries
    ]
    results = await asyncio.gather(*tasks)
    return {"searches": len(req.queries), "results": list(results)}


# ── Pinger ──────────────────────────────────────────────────────────────────
async def pinger():
    url = f"http://localhost:{HTTP_PORT}/health"
    async with httpx.AsyncClient(timeout=10) as client:
        while True:
            await asyncio.sleep(120)
            try:
                r = await client.get(url)
                print(f"[Pinger] {r.status_code}")
            except Exception as e:
                print(f"[Pinger] {e}")


@asynccontextmanager
async def lifespan(app):
    loop = asyncio.get_running_loop()
    loop.run_in_executor(pool, _warm_pool)
    task = asyncio.create_task(pinger())
    try:
        yield
    finally:
        task.cancel()


fastapi_app.router.lifespan_context = lifespan


# ── Gradio UI ───────────────────────────────────────────────────────────────
def format_result(row: dict) -> str:
    lines = []
    for f in SEARCH_FIELDS:
        val = row.get(f, "")
        if val not in ("", None):
            lines.append(f"**{f}:** {val}")
    return "\n\n".join(lines)


def search_ui(query: str, limit: int) -> str:
    if not query or not query.strip():
        return "⚠️ kuch toh search karo — phone, email, username, ya name daalo."
    q = query.strip()
    try:
        data = _unified_search(q, int(limit))
    except Exception as e:
        return f"❌ error: {e}"

    results = data["results"]
    searched = ", ".join(data.get("searched_fields", []))
    if not results:
        return f"🔍 **Query:** `{q}`\n**Searched:** {searched}\n\n❌ no data found."

    header = f"🔍 **Query:** `{q}`  |  **Found:** {len(results)}  |  **Searched:** {searched}\n\n---\n\n"
    parts = [f"### Result {i}\n{format_result(r)}" for i, r in enumerate(results, 1)]
    return header + "\n\n---\n\n".join(parts)


def build_ui():
    with gr.Blocks(title="Rajabets Search") as demo:
        gr.Markdown("# 🔍 Rajabets Search API", elem_classes="main-title")
        gr.Markdown(
            "Search leaked **Rajabets** user records (HF parquet remote) — phone, email, username, name",
            elem_classes="subtitle",
        )

        with gr.Row():
            with gr.Column(scale=3):
                query_input = gr.Textbox(
                    label="Search Query",
                    placeholder="phone, email, username, ya name...",
                    lines=1,
                )
            with gr.Column(scale=1):
                limit_slider = gr.Slider(
                    minimum=1, maximum=100, value=20, step=1,
                    label="Max Results",
                )

        search_btn = gr.Button("🔍 Search", variant="primary", size="lg")
        output = gr.Markdown(label="Results")

        search_btn.click(fn=search_ui, inputs=[query_input, limit_slider], outputs=output)
        query_input.submit(fn=search_ui, inputs=[query_input, limit_slider], outputs=output)

        gr.Markdown("---")
        with gr.Accordion("📡 API Info", open=False):
            gr.Markdown(f"""
**Endpoints**:
- `GET /search?q=<value>`
- `GET /search?field=phone&q=123&mode=exact`
- `GET /warm` — prime metadata
- `GET /health`
- `GET /docs`

**Dataset:** [{HF_DATASET}](https://huggingface.co/datasets/{HF_DATASET})
**Mode:** parquet remote read via DuckDB httpfs
""")

        gr.Markdown(
            "---\n<div class='footer'>👨‍💻 **Developer:** @SrcSellerOnTopBot  |  📢 **Channel:** @NothingFlyXyz</div>",
            elem_classes="footer",
        )

    return demo


# ── Mount ───────────────────────────────────────────────────────────────────
demo = build_ui()

app = gr.mount_gradio_app(
    fastapi_app,
    demo,
    path="/",
    theme=gr.themes.Soft(),
    css=""".main-title{text-align:center;margin-bottom:0}
.subtitle{text-align:center;color:#666;margin-top:0}
.footer{text-align:center;color:#888;margin-top:20px}""",
)


if __name__ == "__main__":
    if os.environ.get("LOCAL_DEV") == "1":
        import uvicorn
        uvicorn.run(app, host="0.0.0.0", port=HTTP_PORT)