main.py

# -*- coding: utf-8 -*-
from __future__ import annotations

import os
import io
import json
import time
import tarfile
import logging
import hashlib
from typing import Dict, Any, List, Tuple, Optional

import numpy as np
import faiss
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel

import gradio as gr

# =============================================================================
# LOGGING
# =============================================================================
LOG = logging.getLogger("remote-indexer-space")
if not LOG.handlers:
    h = logging.StreamHandler()
    h.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
    LOG.addHandler(h)
LOG.setLevel(logging.INFO)

# =============================================================================
# CONFIG (via ENV)
# =============================================================================
PORT = int(os.getenv("PORT", "7860"))
DATA_ROOT = os.getenv("DATA_ROOT", "/tmp/data")  # persistant dans le conteneur Space
os.makedirs(DATA_ROOT, exist_ok=True)

# Provider d'embeddings:
#   - "dummy" : vecteurs aléatoires déterministes (très rapide)
#   - "st"    : Sentence-Transformers (CPU-friendly, simple)
#   - "hf"    : Transformers (AutoModel/AutoTokenizer, pooling manuel)
EMB_PROVIDER = os.getenv("EMB_PROVIDER", "dummy").strip().lower()

# Modèle embeddings (utilisé si provider != "dummy")
# Reco rapide et multilingue (FR ok) : paraphrase-multilingual-MiniLM-L12-v2 (dim=384)
EMB_MODEL = os.getenv("EMB_MODEL", "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2").strip()

# Batch d'encodage
EMB_BATCH = int(os.getenv("EMB_BATCH", "32"))

# Dimension par défaut (dummy) — pour st/hf on lit depuis le modèle
EMB_DIM = int(os.getenv("EMB_DIM", "128"))

# Cache global lazy
_ST_MODEL = None
_HF_TOKENIZER = None
_HF_MODEL = None

# =============================================================================
# JOB STATE
# =============================================================================
class JobState(BaseModel):
    job_id: str
    project_id: str
    stage: str = "pending"          # pending -> chunking -> embedding -> indexing -> done/failed
    total_files: int = 0
    total_chunks: int = 0
    embedded: int = 0
    indexed: int = 0
    errors: List[str] = []
    messages: List[str] = []
    started_at: float = time.time()
    finished_at: Optional[float] = None

JOBS: Dict[str, JobState] = {}

def _now() -> str:
    return time.strftime("%H:%M:%S")

def _proj_dirs(project_id: str) -> Tuple[str, str, str]:
    base = os.path.join(DATA_ROOT, project_id)
    ds_dir = os.path.join(base, "dataset")
    fx_dir = os.path.join(base, "faiss")
    os.makedirs(ds_dir, exist_ok=True)
    os.makedirs(fx_dir, exist_ok=True)
    return base, ds_dir, fx_dir

def _add_msg(st: JobState, msg: str):
    st.messages.append(f"[{_now()}] {msg}")
    LOG.info("[%s] %s", st.job_id, msg)

def _set_stage(st: JobState, stage: str):
    st.stage = stage
    _add_msg(st, f"stage={stage}")

# =============================================================================
# UTILS
# =============================================================================
def _chunk_text(text: str, size: int = 200, overlap: int = 20) -> List[str]:
    text = (text or "").replace("\r\n", "\n")
    tokens = list(text)
    if size <= 0:
        return [text] if text else []
    if overlap < 0:
        overlap = 0
    chunks = []
    i = 0
    while i < len(tokens):
        j = min(i + size, len(tokens))
        chunk = "".join(tokens[i:j]).strip()
        if chunk:
            chunks.append(chunk)
        if j == len(tokens):
            break
        i = j - overlap if (j - overlap) > i else j
    return chunks

def _l2_normalize(x: np.ndarray) -> np.ndarray:
    n = np.linalg.norm(x, axis=1, keepdims=True) + 1e-12
    return x / n

# ----------------------- PROVIDER: DUMMY --------------------------------------
def _emb_dummy(texts: List[str], dim: int = EMB_DIM) -> np.ndarray:
    vecs = np.zeros((len(texts), dim), dtype="float32")
    for i, t in enumerate(texts):
        h = hashlib.sha1((t or "").encode("utf-8")).digest()
        rng = np.random.default_rng(int.from_bytes(h[:8], "little", signed=False))
        v = rng.standard_normal(dim).astype("float32")
        vecs[i] = v / (np.linalg.norm(v) + 1e-9)
    return vecs

# ----------------- PROVIDER: Sentence-Transformers ----------------------------
def _get_st_model():
    global _ST_MODEL
    if _ST_MODEL is None:
        from sentence_transformers import SentenceTransformer
        _ST_MODEL = SentenceTransformer(EMB_MODEL)
        LOG.info(f"[st] modèle chargé: {EMB_MODEL}")
    return _ST_MODEL

def _emb_st(texts: List[str]) -> np.ndarray:
    model = _get_st_model()
    vecs = model.encode(
        texts,
        batch_size=max(1, EMB_BATCH),
        convert_to_numpy=True,
        normalize_embeddings=True,
        show_progress_bar=False,
    ).astype("float32")
    return vecs

def _st_dim() -> int:
    model = _get_st_model()
    try:
        return int(model.get_sentence_embedding_dimension())
    except Exception:
        # fallback : encode une phrase et lit la shape
        v = model.encode(["dimension probe"], convert_to_numpy=True)
        return int(v.shape[1])

# ----------------------- PROVIDER: Transformers (HF) --------------------------
def _get_hf_model():
    global _HF_TOKENIZER, _HF_MODEL
    if _HF_MODEL is None or _HF_TOKENIZER is None:
        from transformers import AutoTokenizer, AutoModel
        _HF_TOKENIZER = AutoTokenizer.from_pretrained(EMB_MODEL)
        _HF_MODEL = AutoModel.from_pretrained(EMB_MODEL)
        _HF_MODEL.eval()
        LOG.info(f"[hf] modèle chargé: {EMB_MODEL}")
    return _HF_TOKENIZER, _HF_MODEL

def _mean_pool(last_hidden_state: "np.ndarray", attention_mask: "np.ndarray") -> "np.ndarray":
    # mean pooling masquée
    mask = attention_mask[..., None].astype(last_hidden_state.dtype)  # (b, t, 1)
    summed = (last_hidden_state * mask).sum(axis=1)                   # (b, h)
    counts = mask.sum(axis=1).clip(min=1e-9)                          # (b, 1)
    return summed / counts

def _emb_hf(texts: List[str]) -> np.ndarray:
    import torch
    tok, mod = _get_hf_model()
    all_vecs = []
    bs = max(1, EMB_BATCH)
    with torch.no_grad():
        for i in range(0, len(texts), bs):
            batch = texts[i:i+bs]
            enc = tok(batch, padding=True, truncation=True, return_tensors="pt")
            out = mod(**enc)
            last = out.last_hidden_state  # (b, t, h)
            pooled = _mean_pool(last.numpy(), enc["attention_mask"].numpy())  # numpy
            all_vecs.append(pooled.astype("float32"))
    vecs = np.concatenate(all_vecs, axis=0)
    return _l2_normalize(vecs)

def _hf_dim() -> int:
    # essaie de lire hidden_size
    try:
        _, mod = _get_hf_model()
        return int(getattr(mod.config, "hidden_size", 768))
    except Exception:
        return 768

# ---------------------------- DATASET / FAISS ---------------------------------
def _save_dataset(ds_dir: str, rows: List[Dict[str, Any]]):
    os.makedirs(ds_dir, exist_ok=True)
    data_path = os.path.join(ds_dir, "data.jsonl")
    with open(data_path, "w", encoding="utf-8") as f:
        for r in rows:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    meta = {"format": "jsonl", "columns": ["path", "text", "chunk_id"], "count": len(rows)}
    with open(os.path.join(ds_dir, "meta.json"), "w", encoding="utf-8") as f:
        json.dump(meta, f, ensure_ascii=False, indent=2)

def _load_dataset(ds_dir: str) -> List[Dict[str, Any]]:
    data_path = os.path.join(ds_dir, "data.jsonl")
    if not os.path.isfile(data_path):
        return []
    out = []
    with open(data_path, "r", encoding="utf-8") as f:
        for line in f:
            try:
                out.append(json.loads(line))
            except Exception:
                continue
    return out

def _save_faiss(fx_dir: str, xb: np.ndarray, meta: Dict[str, Any]):
    os.makedirs(fx_dir, exist_ok=True)
    idx_path = os.path.join(fx_dir, "emb.faiss")
    index = faiss.IndexFlatIP(xb.shape[1])  # cosine ~ inner product si normalisé
    index.add(xb)
    faiss.write_index(index, idx_path)
    with open(os.path.join(fx_dir, "meta.json"), "w", encoding="utf-8") as f:
        json.dump(meta, f, ensure_ascii=False, indent=2)

def _load_faiss(fx_dir: str) -> faiss.Index:
    idx_path = os.path.join(fx_dir, "emb.faiss")
    if not os.path.isfile(idx_path):
        raise FileNotFoundError(f"FAISS index introuvable: {idx_path}")
    return faiss.read_index(idx_path)

def _tar_dir_to_bytes(dir_path: str) -> bytes:
    bio = io.BytesIO()
    with tarfile.open(fileobj=bio, mode="w:gz") as tar:
        tar.add(dir_path, arcname=os.path.basename(dir_path))
    bio.seek(0)
    return bio.read()

# =============================================================================
# FASTAPI
# =============================================================================
fastapi_app = FastAPI(title="remote-indexer", version="2.0.0")
fastapi_app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"],
)

class FileItem(BaseModel):
    path: str
    text: str

class IndexRequest(BaseModel):
    project_id: str
    files: List[FileItem]
    chunk_size: int = 200
    overlap: int = 20
    batch_size: int = 32
    store_text: bool = True

@fastapi_app.get("/health")
def health():
    info = {
        "ok": True,
        "service": "remote-indexer",
        "provider": EMB_PROVIDER,
        "model": EMB_MODEL if EMB_PROVIDER != "dummy" else None
    }
    return info

@fastapi_app.get("/")
def root_redirect():
    return {"ok": True, "service": "remote-indexer", "ui": "/ui"}

@fastapi_app.post("/index")
def index(req: IndexRequest):
    job_id = hashlib.sha1(f"{req.project_id}{time.time()}".encode()).hexdigest()[:12]
    st = JobState(job_id=job_id, project_id=req.project_id, stage="pending", messages=[])
    JOBS[job_id] = st
    _add_msg(st, f"Job {job_id} créé pour project {req.project_id}")
    _add_msg(st, f"Index start project={req.project_id} files={len(req.files)} chunk_size={req.chunk_size} overlap={req.overlap} batch_size={req.batch_size} store_text={req.store_text} provider={EMB_PROVIDER} model={EMB_MODEL if EMB_PROVIDER!='dummy' else '-'}")
    try:
        base, ds_dir, fx_dir = _proj_dirs(req.project_id)

        # 1) Chunking
        _set_stage(st, "chunking")
        rows: List[Dict[str, Any]] = []
        st.total_files = len(req.files)
        for it in req.files:
            txt = it.text or ""
            chunks = _chunk_text(txt, size=req.chunk_size, overlap=req.overlap)
            _add_msg(st, f"{it.path}: len(text)={len(txt)} chunks={len(chunks)}")
            for ci, ck in enumerate(chunks):
                rows.append({"path": it.path, "text": ck, "chunk_id": ci})
        st.total_chunks = len(rows)
        _add_msg(st, f"Total chunks = {st.total_chunks}")

        # 2) Embedding
        _set_stage(st, "embedding")
        if EMB_PROVIDER == "dummy":
            xb = _emb_dummy([r["text"] for r in rows], dim=EMB_DIM)
            dim = xb.shape[1]
        elif EMB_PROVIDER == "st":
            xb = _emb_st([r["text"] for r in rows])
            dim = xb.shape[1]
        else:  # "hf"
            xb = _emb_hf([r["text"] for r in rows])
            dim = xb.shape[1]

        st.embedded = xb.shape[0]
        _add_msg(st, f"Embeddings {st.embedded}/{st.total_chunks}")
        _add_msg(st, f"Embeddings dim={dim}")

        # 3) Sauvegarde dataset (texte)
        _save_dataset(ds_dir, rows)
        _add_msg(st, f"Dataset (sans index) sauvegardé dans {ds_dir}")

        # 4) FAISS
        _set_stage(st, "indexing")
        faiss_meta = {
            "dim": int(dim),
            "count": int(xb.shape[0]),
            "provider": EMB_PROVIDER,
            "model": EMB_MODEL if EMB_PROVIDER != "dummy" else None
        }
        _save_faiss(fx_dir, xb, meta=faiss_meta)
        st.indexed = int(xb.shape[0])
        _add_msg(st, f"FAISS écrit sur {os.path.join(fx_dir, 'emb.faiss')}")
        _add_msg(st, f"OK — dataset+index prêts (projet={req.project_id})")

        _set_stage(st, "done")
        st.finished_at = time.time()
        return {"job_id": job_id}
    except Exception as e:
        LOG.exception("index failed")
        st.errors.append(str(e))
        _add_msg(st, f"❌ Exception: {e}")
        st.stage = "failed"
        st.finished_at = time.time()
        raise HTTPException(status_code=500, detail=str(e))

@fastapi_app.get("/status/{job_id}")
def status(job_id: str):
    st = JOBS.get(job_id)
    if not st:
        raise HTTPException(status_code=404, detail="job inconnu")
    return JSONResponse(st.model_dump())

class SearchRequest(BaseModel):
    project_id: str
    query: str
    k: int = 5

@fastapi_app.post("/search")
def search(req: SearchRequest):
    base, ds_dir, fx_dir = _proj_dirs(req.project_id)
    rows = _load_dataset(ds_dir)
    if not rows:
        raise HTTPException(status_code=404, detail="dataset introuvable (index pas encore construit ?)")

    # Embedding de la requête avec le MÊME provider
    if EMB_PROVIDER == "dummy":
        q = _emb_dummy([req.query], dim=EMB_DIM)[0:1, :]
    elif EMB_PROVIDER == "st":
        q = _emb_st([req.query])[0:1, :]
    else:
        q = _emb_hf([req.query])[0:1, :]

    # FAISS
    index = _load_faiss(fx_dir)
    if index.d != q.shape[1]:
        raise HTTPException(status_code=500, detail=f"dim incompatibles: index.d={index.d} vs query={q.shape[1]}")
    scores, ids = index.search(q, int(max(1, req.k)))
    ids = ids[0].tolist()
    scores = scores[0].tolist()

    out = []
    for idx, sc in zip(ids, scores):
        if idx < 0 or idx >= len(rows):
            continue
        r = rows[idx]
        out.append({"path": r.get("path"), "text": r.get("text"), "score": float(sc)})
    return {"results": out}

# ----------- ARTIFACTS EXPORT -----------
@fastapi_app.get("/artifacts/{project_id}/dataset")
def download_dataset(project_id: str):
    base, ds_dir, _ = _proj_dirs(project_id)
    if not os.path.isdir(ds_dir):
        raise HTTPException(status_code=404, detail="Dataset introuvable")
    buf = _tar_dir_to_bytes(ds_dir)
    headers = {"Content-Disposition": f'attachment; filename="{project_id}_dataset.tgz"'}
    return StreamingResponse(io.BytesIO(buf), media_type="application/gzip", headers=headers)

@fastapi_app.get("/artifacts/{project_id}/faiss")
def download_faiss(project_id: str):
    base, _, fx_dir = _proj_dirs(project_id)
    if not os.path.isdir(fx_dir):
        raise HTTPException(status_code=404, detail="FAISS introuvable")
    buf = _tar_dir_to_bytes(fx_dir)
    headers = {"Content-Disposition": f'attachment; filename="{project_id}_faiss.tgz"'}
    return StreamingResponse(io.BytesIO(buf), media_type="application/gzip", headers=headers)

# =============================================================================
# GRADIO UI (facultatif)
# =============================================================================
def _ui_index(project_id: str, sample_text: str):
    files = [{"path": "sample.txt", "text": sample_text}]
    from pydantic import ValidationError
    try:
        req = IndexRequest(project_id=project_id, files=[FileItem(**f) for f in files])
    except ValidationError as e:
        return f"Erreur: {e}"
    try:
        res = index(req)
        return f"Job lancé: {res['job_id']}"
    except Exception as e:
        return f"Erreur index: {e}"

def _ui_search(project_id: str, query: str, k: int):
    try:
        res = search(SearchRequest(project_id=project_id, query=query, k=int(k)))
        return json.dumps(res, ensure_ascii=False, indent=2)
    except Exception as e:
        return f"Erreur search: {e}"

with gr.Blocks(title="Remote Indexer (FAISS)", analytics_enabled=False) as ui:
    gr.Markdown("## Remote Indexer — demo UI (API: `/index`, `/status/{job}`, `/search`, `/artifacts/...`).")
    gr.Markdown(f"**Provider**: `{EMB_PROVIDER}` — **Model**: `{EMB_MODEL if EMB_PROVIDER!='dummy' else '-'}'")
    with gr.Tab("Index"):
        pid = gr.Textbox(label="Project ID", value="DEEPWEB")
        sample = gr.Textbox(label="Texte d’exemple", value="Alpha bravo charlie delta echo foxtrot.", lines=4)
        btn = gr.Button("Lancer index (sample)")
        out = gr.Textbox(label="Résultat")
        btn.click(_ui_index, inputs=[pid, sample], outputs=[out])

    with gr.Tab("Search"):
        pid2 = gr.Textbox(label="Project ID", value="DEEPWEB")
        q = gr.Textbox(label="Query", value="alpha")
        k = gr.Slider(1, 20, value=5, step=1, label="k")
        btn2 = gr.Button("Rechercher")
        out2 = gr.Code(label="Résultats")
        btn2.click(_ui_search, inputs=[pid2, q, k], outputs=[out2])

fastapi_app = gr.mount_gradio_app(fastapi_app, ui, path="/ui")

# =============================================================================
# MAIN
# =============================================================================
if __name__ == "__main__":
    import uvicorn
    LOG.info("Démarrage Uvicorn sur 0.0.0.0:%s (UI_PATH=/ui)", PORT)
    uvicorn.run(fastapi_app, host="0.0.0.0", port=PORT)