spb/trouve-ka Public

1# Trouve-KA — worker d'enrichissement2# Author: Simon-Pierre Boucher3# Contact: contact@spboucher.ai45"""Worker d'enrichissement : consomme trouveka:enrich (Redis Streams).67Étape 2 (implémentée) : propage domain_quebec_score et authority_score à jour8dans le document indexé (le domaine apprend au fil du crawl, les documents déjà9indexés en profitent rétroactivement).10Étapes futures : embeddings, entités, classification thématique — même canal,11même contrat : mise à jour partielle du document, jamais bloquante.12"""1314import asyncio15import os16import signal17import uuid1819from trouveka.config import get_settings20from trouveka.database import Database21from trouveka.logging import get_logger22from trouveka.queue import Coordination23from trouveka.search_core import SearchCore2425log = get_logger("enrichment")2627GROUP = "enrichers"282930class EnrichmentWorker:31    def __init__(self) -> None:32        self.s = get_settings()33        self.consumer = f"enrich-{os.getpid()}-{uuid.uuid4().hex[:6]}"34        self.db = Database(self.s.database_url, pool_min=1, pool_max=3)35        self.coord = Coordination(self.s.redis_url)36        self.search = SearchCore(self.s.search_url, self.s.search_index)37        self.stop_event = asyncio.Event()3839    async def enrich(self, payload: dict) -> None:40        domain_row = await self.db.get_domain_by_id(int(payload["domain_id"]))41        if not domain_row:42            return43        await self.search.update_document(44            payload["url"],45            {46                "domain_quebec_score": round(float(domain_row["quebec_score"]), 4),47                "authority_score": round(float(domain_row["authority_score"]), 4),48            },49        )5051    async def start(self) -> None:52        await self.db.connect()53        loop = asyncio.get_running_loop()54        for sig in (signal.SIGINT, signal.SIGTERM):55            loop.add_signal_handler(sig, self.stop_event.set)56        log.info("worker d'enrichissement démarré", extra={"ctx": {"consumer": self.consumer}})5758        while not self.stop_event.is_set():59            try:60                messages = await self.coord.read_enrichment(GROUP, self.consumer, count=20, block_ms=5000)61                for msg_id, payload in messages:62                    try:63                        await self.enrich(payload)64                    except Exception:65                        log.exception("échec enrichissement", extra={"ctx": payload})66                    finally:67                        await self.coord.ack_enrichment(GROUP, msg_id)68            except Exception:69                log.exception("erreur boucle enrichissement (on continue)")70                await asyncio.sleep(2)7172        await self.search.close()73        await self.coord.close()74        await self.db.close()757677def main() -> None:78    asyncio.run(EnrichmentWorker().start())798081if __name__ == "__main__":82    main()83