Update main.py

main.py

@@ -1,29 +1,30 @@
 # -*- coding: utf-8 -*-
 """
-HF Space - main.py de substitution pour tests Qdrant / indexation minimale (robuste + auto-refresh)
+Remote Indexer (HF Space) — Qdrant + embeddings (HF ou dummy)
+Version: worker-thread (pas d'asyncio.create_task dans l'UI), robust logging.
 
 Endpoints:
-- GET  /                      → redirige vers UI_PATH (défaut: /ui)
-- GET  /ui (UI_PATH)          → UI Gradio
-- GET  /health                → healthcheck
-- GET  /api                   → infos service
-- GET  /debug/env             → aperçu config (sans secrets)
-- POST /wipe?project_id=XXX   → supprime la collection Qdrant
-- POST /index                 → lance un job d'indexation
-- GET  /status/{job_id}       → état + logs du job
-- GET  /collections/{proj}/count → count points dans Qdrant
-- POST /query                 → recherche sémantique
+- GET  /            → redirige vers UI_PATH (défaut: /ui)
+- GET  /ui          → UI Gradio
+- GET  /health      → healthcheck
+- GET  /api         → infos service
+- GET  /debug/env   → aperçu config (sans secrets)
+- POST /wipe?project_id=XXX
+- POST /index
+- GET  /status/{job_id}
+- GET  /collections/{project_id}/count
+- POST /query
 
 ENV:
-- QDRANT_URL, QDRANT_API_KEY (requis pour upsert)
-- COLLECTION_PREFIX (défaut "proj_")
-- EMB_PROVIDER ("hf" par défaut, "dummy" sinon)
-- HF_EMBED_MODEL (défaut "BAAI/bge-m3")
-- HUGGINGFACEHUB_API_TOKEN (si EMB_PROVIDER=hf)
-- EMB_FALLBACK_TO_DUMMY (true/false) → si vrai, bascule dummy si HF échoue
-- LOG_LEVEL (défaut DEBUG)
-- PORT (fourni par HF, défaut 7860)
-- UI_PATH (défaut "/ui")
+- QDRANT_URL, QDRANT_API_KEY      (requis pour upsert)
+- COLLECTION_PREFIX               (défaut "proj_")
+- EMB_PROVIDER                    ("hf" | "dummy"; défaut "hf")
+- HF_EMBED_MODEL                  (défaut "BAAI/bge-m3")
+- HUGGINGFACEHUB_API_TOKEN        (si EMB_PROVIDER=hf)
+- EMB_FALLBACK_TO_DUMMY           (true/false) → bascule dummy si HF échoue
+- LOG_LEVEL                       (défaut DEBUG)
+- UI_PATH                         (défaut "/ui")
+- PORT                            (défaut 7860)
 """
 
 from __future__ import annotations
@@ -32,6 +33,7 @@ import time
 import uuid
 import hashlib
 import logging
+import threading
 import asyncio
 from typing import List, Dict, Any, Optional, Tuple
 
@@ -45,7 +47,7 @@ from fastapi.responses import RedirectResponse
 import gradio as gr
 
 # ------------------------------------------------------------------------------
-# Configuration & logs
+# Config & logs
 # ------------------------------------------------------------------------------
 LOG_LEVEL = os.getenv("LOG_LEVEL", "DEBUG").upper()
 logging.basicConfig(
@@ -58,26 +60,20 @@ QDRANT_URL = os.getenv("QDRANT_URL", "").rstrip("/")
 QDRANT_API_KEY = os.getenv("QDRANT_API_KEY", "")
 COLLECTION_PREFIX = os.getenv("COLLECTION_PREFIX", "proj_").strip() or "proj_"
 
-EMB_PROVIDER = os.getenv("EMB_PROVIDER", "hf").lower()  # "hf" | "dummy"
+EMB_PROVIDER = os.getenv("EMB_PROVIDER", "hf").lower()
 HF_EMBED_MODEL = os.getenv("HF_EMBED_MODEL", "BAAI/bge-m3")
 HF_TOKEN = os.getenv("HUGGINGFACEHUB_API_TOKEN", "")
-
 EMB_FALLBACK_TO_DUMMY = os.getenv("EMB_FALLBACK_TO_DUMMY", "false").lower() in ("1","true","yes","on")
 
-UI_PATH = os.getenv("UI_PATH", "/ui")  # UI montée ici par défaut
+UI_PATH = os.getenv("UI_PATH", "/ui")
 
 if not QDRANT_URL or not QDRANT_API_KEY:
     LOG.warning("QDRANT_URL / QDRANT_API_KEY non fournis : l'upsert échouera.")
-
-if EMB_PROVIDER == "hf" and not HF_TOKEN:
-    LOG.warning(
-        "EMB_PROVIDER=hf sans HUGGINGFACEHUB_API_TOKEN. "
-        "→ soit définis le token, soit mets EMB_PROVIDER=dummy, "
-        "soit active EMB_FALLBACK_TO_DUMMY=true."
-    )
+if EMB_PROVIDER == "hf" and not HF_TOKEN and not EMB_FALLBACK_TO_DUMMY:
+    LOG.warning("EMB_PROVIDER=hf sans HUGGINGFACEHUB_API_TOKEN (pas de fallback) → préférer EMB_PROVIDER=dummy ou EMB_FALLBACK_TO_DUMMY=true.")
 
 # ------------------------------------------------------------------------------
-# Schémas Pydantic
+# Models
 # ------------------------------------------------------------------------------
 class FileItem(BaseModel):
     path: str
@@ -96,9 +92,6 @@ class QueryRequest(BaseModel):
     text: str
     top_k: int = Field(5, ge=1, le=100)
 
-# ------------------------------------------------------------------------------
-# Job store (en mémoire)
-# ------------------------------------------------------------------------------
 class JobState(BaseModel):
     job_id: str
     project_id: str
@@ -121,7 +114,7 @@ class JobState(BaseModel):
 JOBS: Dict[str, JobState] = {}
 
 # ------------------------------------------------------------------------------
-# Utilitaires
+# Utils
 # ------------------------------------------------------------------------------
 def hash8(s: str) -> str:
     return hashlib.sha256(s.encode("utf-8")).hexdigest()[:16]
@@ -134,7 +127,6 @@ def l2_normalize(vec: List[float]) -> List[float]:
     return arr.astype(np.float32).tolist()
 
 def flatten_any(x: Any) -> List[float]:
-    """Aplatis potentiels [[...]] ou [[[...]]] en 1D."""
     if isinstance(x, (list, tuple)):
         if len(x) > 0 and isinstance(x[0], (list, tuple)):
             return flatten_any(x[0])
@@ -142,7 +134,6 @@ def flatten_any(x: Any) -> List[float]:
     raise ValueError("Embedding vector mal formé")
 
 def chunk_text(text: str, chunk_size: int, overlap: int) -> List[Tuple[int, int, str]]:
-    """Retourne [(start, end, chunk)] et ignore les fragments < 30 chars."""
     text = text or ""
     if not text.strip():
         return []
@@ -160,10 +151,9 @@ def chunk_text(text: str, chunk_size: int, overlap: int) -> List[Tuple[int, int,
     return res
 
 # ------------------------------------------------------------------------------
-# Qdrant helpers
+# Qdrant
 # ------------------------------------------------------------------------------
 async def ensure_collection(client: httpx.AsyncClient, coll: str, vector_size: int) -> None:
-    """Crée la collection Qdrant (distance=Cosine), ou la recrée si dim mismatch."""
     url = f"{QDRANT_URL}/collections/{coll}"
     r = await client.get(url, headers={"api-key": QDRANT_API_KEY}, timeout=20)
     recreate = False
@@ -201,24 +191,15 @@ async def qdrant_count(client: httpx.AsyncClient, coll: str) -> int:
 
 async def qdrant_search(client: httpx.AsyncClient, coll: str, vector: List[float], limit: int = 5) -> Dict[str, Any]:
     url = f"{QDRANT_URL}/collections/{coll}/points/search"
-    r = await client.post(
-        url,
-        headers={"api-key": QDRANT_API_KEY},
-        json={"vector": vector, "limit": limit, "with_payload": True},
-        timeout=30,
-    )
+    r = await client.post(url, headers={"api-key": QDRANT_API_KEY}, json={"vector": vector, "limit": limit, "with_payload": True}, timeout=30)
     if r.status_code != 200:
         raise HTTPException(status_code=500, detail=f"Qdrant search échoué: {r.text}")
     return r.json()
 
 # ------------------------------------------------------------------------------
-# Embeddings (HF Inference ou dummy)
+# Embeddings
 # ------------------------------------------------------------------------------
 def _maybe_prefix_for_model(texts: List[str], model_name: str) -> List[str]:
-    """
-    E5 attend en pratique des préfixes 'query: ' (ou 'passage: ' / 'document: ').
-    On préfixe automatiquement si le modèle contient 'e5'.
-    """
     m = (model_name or "").lower()
     if "e5" in m:
         return [("query: " + t) for t in texts]
@@ -270,7 +251,7 @@ async def embed_texts(client: httpx.AsyncClient, texts: List[str]) -> List[List[
     return embed_dummy(texts, dim=128)
 
 # ------------------------------------------------------------------------------
-# Pipeline d'indexation (robuste)
+# Core: run_index_job (async) + worker thread wrapper
 # ------------------------------------------------------------------------------
 async def run_index_job(job: JobState, req: IndexRequest) -> None:
     try:
@@ -282,12 +263,6 @@ async def run_index_job(job: JobState, req: IndexRequest) -> None:
             f"provider={EMB_PROVIDER} model={HF_EMBED_MODEL}"
         )
 
-        # Dédup global par hash du texte de fichier
-        file_hashes = [hash8(f.text) for f in req.files]
-        uniq = len(set(file_hashes))
-        if uniq != len(file_hashes):
-            job.log(f"Attention: {len(file_hashes)-uniq} fichier(s) ont un texte identique (hash dupliqué).")
-
         # Chunking
         records: List[Dict[str, Any]] = []
         for f in req.files:
@@ -301,7 +276,6 @@ async def run_index_job(job: JobState, req: IndexRequest) -> None:
                 records.append({"payload": payload, "raw": ch})
         job.total_chunks = len(records)
         job.log(f"Total chunks = {job.total_chunks}")
-
         if job.total_chunks == 0:
             job.stage = "failed"
             job.errors.append("Aucun chunk à indexer.")
@@ -314,11 +288,12 @@ async def run_index_job(job: JobState, req: IndexRequest) -> None:
             vec_dim = len(warmup_vec)
             job.log(f"Warmup embeddings dim={vec_dim}")
 
-            # Collection Qdrant
+            # Collection
             coll = f"{COLLECTION_PREFIX}{req.project_id}"
             await ensure_collection(client, coll, vector_size=vec_dim)
             job.log(f"Collection prête: {coll} (dim={vec_dim})")
 
+            # Upsert
             job.stage = "upserting"
             batch_points: List[Dict[str, Any]] = []
 
@@ -360,8 +335,29 @@ async def run_index_job(job: JobState, req: IndexRequest) -> None:
         job.finished_at = time.time()
         job.log(f"❌ Exception: {e}")
 
+def _run_job_in_thread(job: JobState, req: IndexRequest) -> None:
+    """Exécute l'async run_index_job dans un thread dédié avec son propre event loop."""
+    def _runner():
+        try:
+            asyncio.run(run_index_job(job, req))
+        except Exception as e:
+            job.stage = "failed"
+            job.errors.append(str(e))
+            job.finished_at = time.time()
+            job.log(f"❌ Thread exception: {e}")
+    t = threading.Thread(target=_runner, daemon=True)
+    t.start()
+
+def create_and_start_job(req: IndexRequest) -> JobState:
+    job_id = uuid.uuid4().hex[:12]
+    job = JobState(job_id=job_id, project_id=req.project_id)
+    JOBS[job_id] = job
+    job.log(f"Job {job_id} créé pour project {req.project_id}")
+    _run_job_in_thread(job, req)
+    return job
+
 # ------------------------------------------------------------------------------
-# FastAPI app + endpoints
+# FastAPI app
 # ------------------------------------------------------------------------------
 fastapi_app = FastAPI(title="Remote Indexer - Minimal Test Space")
 fastapi_app.add_middleware(
@@ -378,13 +374,11 @@ async def health():
 @fastapi_app.get("/api")
 async def api_info():
     return {
-        "ok": True,
-        "service": "remote-indexer-min",
+        "ok": True, "service": "remote-indexer-min",
         "qdrant": bool(QDRANT_URL),
-        "emb_provider": EMB_PROVIDER,
-        "hf_model": HF_EMBED_MODEL,
-        "ui_path": UI_PATH,
+        "emb_provider": EMB_PROVIDER, "hf_model": HF_EMBED_MODEL,
         "fallback_to_dummy": EMB_FALLBACK_TO_DUMMY,
+        "ui_path": UI_PATH,
     }
 
 @fastapi_app.get("/debug/env")
@@ -399,7 +393,6 @@ async def debug_env():
         "collection_prefix": COLLECTION_PREFIX,
     }
 
-# Redirige "/" → UI_PATH (ex.: /ui).
 @fastapi_app.get("/")
 async def root_redirect():
     return RedirectResponse(url=UI_PATH, status_code=307)
@@ -419,12 +412,8 @@ async def wipe(project_id: str = Query(..., min_length=1)):
 async def index(req: IndexRequest):
     if not QDRANT_URL or not QDRANT_API_KEY:
         raise HTTPException(status_code=400, detail="QDRANT_URL / QDRANT_API_KEY requis")
-    job_id = uuid.uuid4().hex[:12]
-    job = JobState(job_id=job_id, project_id=req.project_id)
-    JOBS[job_id] = job
-    asyncio.create_task(run_index_job(job, req))
-    job.log(f"Job {job_id} créé pour project {req.project_id}")
-    return {"job_id": job_id, "project_id": req.project_id}
+    job = create_and_start_job(req)
+    return {"job_id": job.job_id, "project_id": job.project_id}
 
 @fastapi_app.get("/status/{job_id}")
 async def status(job_id: str):
@@ -453,7 +442,7 @@ async def query(req: QueryRequest):
     return data
 
 # ------------------------------------------------------------------------------
-# Gradio UI (avec Status + Auto-refresh)
+# Gradio UI (avec auto-refresh)
 # ------------------------------------------------------------------------------
 def _default_two_docs() -> List[Dict[str, str]]:
     a = "Alpha bravo charlie delta echo foxtrot golf hotel india. " * 3
@@ -462,7 +451,7 @@ def _default_two_docs() -> List[Dict[str, str]]:
 
 async def ui_wipe(project: str):
     try:
-        resp = await wipe(project)  # appelle la route interne
+        resp = await wipe(project)
         return f"✅ Wipe ok — collection {resp['collection']} supprimée."
     except Exception as e:
         LOG.exception("wipe UI error")
@@ -479,10 +468,8 @@ async def ui_index_sample(project: str, chunk_size: int, overlap: int, batch_siz
         store_text=store_text,
     )
     try:
-        data = await index(req)
-        job_id = data["job_id"]
-        # On retourne ET le message ET le job_id pour remplir le champ
-        return f"🚀 Job lancé: {job_id}", job_id
+        job = create_and_start_job(req)  # ← lance dans le thread dédié
+        return f"🚀 Job lancé: {job.job_id}", job.job_id
     except ValidationError as ve:
         return f"❌ Payload invalide: {ve}", ""
     except Exception as e:
@@ -530,7 +517,7 @@ async def ui_query(project: str, text: str, topk: int):
         LOG.exception("query UI error")
         return f"❌ Query erreur: {e}"
 
-with gr.Blocks(title="Remote Indexer - Minimal Test", analytics_enabled=False) as ui:
+with gr.Blocks(title="Remote Indexer — Tests sans console", analytics_enabled=False) as ui:
     gr.Markdown("## 🔬 Remote Indexer — Tests sans console\n"
                 "Wipe → Index 2 docs → Status → Count → Query\n"
                 f"- **Embeddings**: `{EMB_PROVIDER}` (model: `{HF_EMBED_MODEL}`)\n"
@@ -561,21 +548,17 @@ with gr.Blocks(title="Remote Indexer - Minimal Test", analytics_enabled=False) a
         query_btn = gr.Button("🔎 Query")
     query_out = gr.Textbox(lines=10, label="Résultats Query", interactive=False)
 
-    # Liens UI
     wipe_btn.click(ui_wipe, inputs=[project_tb], outputs=[out_log])
-    # index renvoie (message, job_id)
     index_btn.click(ui_index_sample, inputs=[project_tb, chunk_size, overlap, batch_size, store_text], outputs=[out_log, jobid_tb])
     count_btn.click(ui_count, inputs=[project_tb], outputs=[out_log])
 
-    # Status bouton manuel
     status_btn.click(ui_status, inputs=[jobid_tb], outputs=[out_log])
-
-    # Auto-refresh avec Timer (toutes les 2s si coché)
     timer = gr.Timer(2.0, active=False)
     timer.tick(ui_status, inputs=[jobid_tb], outputs=[out_log])
     auto_chk.change(lambda x: gr.update(active=x), inputs=auto_chk, outputs=timer)
 
-# Monte l'UI Gradio sur la FastAPI au chemin UI_PATH
+# Monte l'UI Gradio
+fastapi_app.mount("/static", gr.routes.App.get_blocks().static_files)  # pour servir les assets si nécessaire
 app = gr.mount_gradio_app(fastapi_app, ui, path=UI_PATH)
 
 if __name__ == "__main__":