spb/trouve-ka

1# Trouve-KA — migration de l'index de recherche vers le mapping kNN2# Author: Simon-Pierre Boucher3# Contact: contact@spboucher.ai45"""Migre l'index vers le mapping v2 (champ embedding kNN).67`index.knn` est un réglage statique OpenSearch : impossible de l'activer sur un8index existant. Procédure (idempotente) :9101. crawl en pause (les workers respectent trouveka:crawler:paused);112. création de <index>-v2 avec le mapping kNN;123. _reindex serveur <index> → <index>-v2;134. suppression de l'ancien index, alias <index> → <index>-v214   (le code continue de parler à <index> sans changement);155. reprise du crawl.1617Usage : python -m scripts.migrate-search-index.migrate  (ou via docker compose run)18"""1920import asyncio2122from opensearchpy import AsyncOpenSearch2324from trouveka.config import get_settings25from trouveka.queue import Coordination26from trouveka.search_core import INDEX_SETTINGS272829async def main() -> None:30    s = get_settings()31    client = AsyncOpenSearch(hosts=[s.search_url], timeout=600)32    coord = Coordination(s.redis_url)33    index, v2 = s.search_index, f"{s.search_index}-v2"3435    try:36        if await client.indices.exists_alias(name=index):37            print(f"✓ {index} est déjà un alias — migration déjà faite")38            return39        if not await client.indices.exists(index=index):40            await client.indices.create(index=v2, body=INDEX_SETTINGS)41            await client.indices.put_alias(index=v2, name=index)42            print(f"✓ installation neuve : {v2} créé avec alias {index}")43            return4445        print("→ pause du crawl")46        await coord.pause_crawler()47        await asyncio.sleep(5)  # laisser les indexations en vol se terminer4849        if not await client.indices.exists(index=v2):50            await client.indices.create(index=v2, body=INDEX_SETTINGS)51            print(f"→ {v2} créé")5253        print(f"→ _reindex {index} → {v2} (serveur, peut prendre quelques minutes)")54        result = await client.reindex(55            body={"source": {"index": index}, "dest": {"index": v2}},56            params={"wait_for_completion": "true", "refresh": "true"},57        )58        print(f"→ réindexé : {result.get('total')} docs, {result.get('took')} ms")5960        old_count = (await client.count(index=index))["count"]61        new_count = (await client.count(index=v2))["count"]62        print(f"→ comptes : ancien={old_count} nouveau={new_count}")63        if new_count < old_count * 0.99:64            raise RuntimeError("perte de documents détectée, on ne bascule pas")6566        await client.indices.delete(index=index)67        await client.indices.put_alias(index=v2, name=index)68        print(f"✓ alias {index} → {v2} en place")69    finally:70        await coord.resume_crawler()71        print("→ crawl repris")72        await coord.close()73        await client.close()747576if __name__ == "__main__":77    asyncio.run(main())78