spb/ora-ka Public

1# -----------------------------------------------------------------------------2# Ora-Ka — Index sémantique (embeddings OpenAI) sur les 5 univers3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4#5# Construction : python run.py index          (incrémental — n'embedde que le neuf)6# Requête      : semantic_top(q)              (repli silencieux si index/clé absents)7# -----------------------------------------------------------------------------8from __future__ import annotations910import json11import os12import sqlite313import sys14import time15from pathlib import Path1617import numpy as np18import requests1920ROOT = Path(__file__).resolve().parent.parent21APPS_DIR = ROOT / "apps"22INDEX_DIR = ROOT / "data" / "semantic"2324MODEL = "text-embedding-3-small"25DIMS = 25626BATCH = 8002728DBS = {29    "immo": APPS_DIR / "immo" / "data" / "immoka.db",30    "lou": APPS_DIR / "lou" / "data" / "louka.db",31    "fabri": APPS_DIR / "fabri" / "data" / "fabrika.db",32    "auto": APPS_DIR / "auto" / "data" / "autoka.db",33    "food": APPS_DIR / "food" / "data" / "foodka.db",34}353637def _ro(db: Path) -> sqlite3.Connection:38    con = sqlite3.connect(f"file:{db}?mode=ro", uri=True, timeout=10)39    con.row_factory = sqlite3.Row40    return con414243def _clean(*parts) -> str:44    return " · ".join(str(p).strip() for p in parts if p not in (None, "", 0))[:400]454647# ---------------------------------------------------------------------------48# Corpus : (uid, texte) par univers — le texte décrit l'inscription en clair49# ---------------------------------------------------------------------------5051def _corpus_immo() -> dict[str, str]:52    con = _ro(DBS["immo"])53    try:54        rows = con.execute(55            "SELECT uid, title, address, sector, city, region, property_type,"56            " bedrooms, bathrooms, price FROM listings"57            " WHERE active=1 AND dup_hidden=0 AND price IS NOT NULL").fetchall()58    finally:59        con.close()60    out = {}61    for r in rows:62        ch = f"{int(r['bedrooms'])} chambres" if r["bedrooms"] else None63        sdb = f"{int(r['bathrooms'])} salles de bain" if r["bathrooms"] else None64        px = f"{int(r['price'])} $" if r["price"] else None65        out[r["uid"]] = _clean(f"{r['property_type'] or 'Propriété'} à vendre",66                               r["title"], r["address"], r["sector"], r["city"],67                               r["region"], ch, sdb, px)68    return out697071def _corpus_lou() -> dict[str, str]:72    con = _ro(DBS["lou"])73    try:74        rows = con.execute(75            "SELECT uid, title, address, sector, city, unit_type, price, furnished"76            " FROM listings WHERE active=1").fetchall()77    finally:78        con.close()79    out = {}80    for r in rows:81        px = f"{int(r['price'])} $ par mois" if r["price"] else None82        out[r["uid"]] = _clean(f"{r['unit_type'] or 'Logement'} à louer, appartement",83                               r["title"], r["address"], r["sector"], r["city"], px,84                               "meublé" if r["furnished"] else None)85    return out868788def _corpus_fabri() -> dict[str, str]:89    con = _ro(DBS["fabri"])90    try:91        rows = con.execute(92            "SELECT p.uid, p.title, p.category, p.vendor, p.product_type,"93            " s.name AS store_name, s.region FROM products p"94            " LEFT JOIN stores s ON s.id=p.store_id WHERE p.active=1").fetchall()95    finally:96        con.close()97    return {r["uid"]: _clean("Produit québécois", r["title"], r["product_type"],98                             r["category"], r["vendor"], r["store_name"], r["region"])99            for r in rows}100101102def _corpus_auto() -> dict[str, str]:103    con = _ro(DBS["auto"])104    try:105        rows = con.execute(106            "SELECT uid, kind, title, make, model, trim, year, body_type, fuel,"107            " transmission, mileage_km, city, region, price FROM vehicles"108            " WHERE active=1").fetchall()109    finally:110        con.close()111    out = {}112    for r in rows:113        km = f"{int(r['mileage_km'])} km" if r["mileage_km"] else None114        px = f"{int(r['price'])} $" if r["price"] else None115        kind = {"auto": "Voiture d'occasion", "moto": "Moto",116                "scooter": "Scooter"}.get(r["kind"], "Véhicule")117        out[r["uid"]] = _clean(kind, r["year"], r["make"], r["model"], r["trim"],118                               r["body_type"], r["fuel"], r["transmission"], km,119                               r["city"], r["region"], px)120    return out121122123def _corpus_food() -> dict[str, str]:124    con = _ro(DBS["food"])125    try:126        rows = con.execute(127            "SELECT uid, name, brand, category, category_raw, size_label, source"128            " FROM products WHERE active=1").fetchall()129    finally:130        con.close()131    return {r["uid"]: _clean("Épicerie", r["name"], r["brand"], r["category"],132                             r["category_raw"], r["size_label"],133                             (r["source"] or "").replace("_", " "))134            for r in rows}135136137_CORPUS = {138    "immo": _corpus_immo,139    "lou": _corpus_lou,140    "fabri": _corpus_fabri,141    "auto": _corpus_auto,142    "food": _corpus_food,143}144145146# ---------------------------------------------------------------------------147# OpenAI148# ---------------------------------------------------------------------------149150def _embed(texts: list[str], api_key: str, retries: int = 4) -> np.ndarray:151    for attempt in range(retries):152        try:153            resp = requests.post(154                "https://api.openai.com/v1/embeddings",155                headers={"Authorization": f"Bearer {api_key}"},156                json={"model": MODEL, "input": texts, "dimensions": DIMS},157                timeout=120,158            )159            if resp.status_code == 429:160                time.sleep(5 * (attempt + 1))161                continue162            resp.raise_for_status()163            data = resp.json()["data"]164            vecs = np.array([d["embedding"] for d in data], dtype=np.float32)165            norms = np.linalg.norm(vecs, axis=1, keepdims=True)166            norms[norms == 0] = 1.0167            return vecs / norms168        except requests.RequestException:169            if attempt == retries - 1:170                raise171            time.sleep(3 * (attempt + 1))172    raise RuntimeError("embeddings: échec après retries")173174175# ---------------------------------------------------------------------------176# Construction incrémentale de l'index177# ---------------------------------------------------------------------------178179def build_index(apps: list[str] | None = None) -> None:180    api_key = os.environ.get("OPENAI_API_KEY")181    if not api_key:182        print("[ora-ka] OPENAI_API_KEY manquant (.env) — index sémantique impossible",183              file=sys.stderr)184        sys.exit(1)185    INDEX_DIR.mkdir(parents=True, exist_ok=True)186    for app in apps or list(_CORPUS):187        if not DBS[app].exists():188            print(f"[index] {app}: base absente, sauté")189            continue190        t0 = time.time()191        corpus = _CORPUS[app]()192        path = INDEX_DIR / f"{app}.npz"193        old_uids: list[str] = []194        old_vecs = None195        if path.exists():196            z = np.load(path, allow_pickle=False)197            old_uids = [str(u) for u in z["uids"]]198            old_vecs = z["vecs"]199        keep_mask = [u in corpus for u in old_uids]200        kept_uids = [u for u, k in zip(old_uids, keep_mask) if k]201        kept_vecs = old_vecs[np.array(keep_mask, dtype=bool)] if old_vecs is not None and old_uids else None202        known = set(kept_uids)203        new_uids = [u for u in corpus if u not in known]204        print(f"[index] {app}: {len(corpus)} items — {len(kept_uids)} déjà indexés,"205              f" {len(new_uids)} à embedder")206        new_chunks = []207        for i in range(0, len(new_uids), BATCH):208            chunk = new_uids[i:i + BATCH]209            vecs = _embed([corpus[u] for u in chunk], api_key)210            new_chunks.append(vecs.astype(np.float16))211            done = i + len(chunk)212            if done % (BATCH * 10) < BATCH or done == len(new_uids):213                print(f"[index] {app}: {done}/{len(new_uids)} embeddés"214                      f" ({time.time() - t0:.0f}s)", flush=True)215        parts = []216        if kept_vecs is not None and len(kept_uids):217            parts.append(kept_vecs.astype(np.float16))218        parts.extend(new_chunks)219        all_uids = kept_uids + new_uids220        if not all_uids:221            print(f"[index] {app}: rien à indexer")222            continue223        all_vecs = np.concatenate(parts, axis=0)224        np.savez_compressed(path, uids=np.array(all_uids), vecs=all_vecs)225        meta = {"model": MODEL, "dims": DIMS, "count": len(all_uids),226                "built_at": time.time()}227        (INDEX_DIR / f"{app}.meta.json").write_text(json.dumps(meta))228        print(f"[index] {app}: OK — {len(all_uids)} vecteurs,"229              f" {path.stat().st_size / 1e6:.0f} Mo, {time.time() - t0:.0f}s")230231232# ---------------------------------------------------------------------------233# Requête234# ---------------------------------------------------------------------------235236_loaded: dict[str, tuple[list[str], np.ndarray]] = {}237_loaded_at: dict[str, float] = {}238_uid_idx: dict[str, dict[str, int]] = {}239_qcache: dict[str, np.ndarray] = {}240241242def _load(app: str):243    path = INDEX_DIR / f"{app}.npz"244    if not path.exists():245        return None246    mtime = path.stat().st_mtime247    if app not in _loaded or _loaded_at.get(app) != mtime:248        z = np.load(path, allow_pickle=False)249        uids = [str(u) for u in z["uids"]]250        _loaded[app] = (uids, z["vecs"].astype(np.float32))251        _uid_idx[app] = {u: i for i, u in enumerate(uids)}252        _loaded_at[app] = mtime253    return _loaded[app]254255256def query_vector(q: str) -> np.ndarray | None:257    """Vecteur normalisé de la requête (cache LRU simple), None si indisponible."""258    api_key = os.environ.get("OPENAI_API_KEY")259    if not api_key:260        return None261    key = q.strip().lower()262    qv = _qcache.get(key)263    if qv is None:264        try:265            qv = _embed([q], api_key)[0]266        except Exception:267            return None268        if len(_qcache) > 512:269            _qcache.clear()270        _qcache[key] = qv271    return qv272273274def sims_for(app: str, uids: list[str], qv: np.ndarray) -> dict[str, float]:275    """Similarité requête↔item pour une liste d'uid (lookup direct dans l'index)."""276    if _load(app) is None or not uids:277        return {}278    idx_map = _uid_idx[app]279    pairs = [(u, idx_map[u]) for u in uids if u in idx_map]280    if not pairs:281        return {}282    arr = np.array([i for _, i in pairs])283    sims = _loaded[app][1][arr] @ qv284    return {u: float(s) for (u, _), s in zip(pairs, sims)}285286287def available() -> bool:288    return bool(os.environ.get("OPENAI_API_KEY")) and any(289        (INDEX_DIR / f"{a}.npz").exists() for a in _CORPUS)290291292def semantic_top(q: str, apps: list[str] | None = None, per_app: int = 30293                 ) -> dict[str, list[tuple[str, float]]]:294    """Retourne {app: [(uid, score), ...]} trié par similarité décroissante."""295    qv = query_vector(q)296    if qv is None:297        return {}298    out: dict[str, list[tuple[str, float]]] = {}299    for app in apps or list(_CORPUS):300        loaded = _load(app)301        if not loaded:302            continue303        uids, vecs = loaded304        sims = vecs @ qv305        k = min(per_app, len(uids))306        if k <= 0:307            continue308        idx = np.argpartition(-sims, k - 1)[:k]309        idx = idx[np.argsort(-sims[idx])]310        out[app] = [(uids[i], float(sims[i])) for i in idx]311    return out312