spb/trouve-ka Public

1# Trouve-KA — coordination Redis (politesse, pause, enrichissement)2# Author: Simon-Pierre Boucher3# Contact: contact@spboucher.ai45"""Coordination inter-workers via Redis.67- Politesse par hôte : lock SET NX PX — un seul fetch par hôte par fenêtre,8  quel que soit le nombre de workers.9- Pause globale du crawler : simple clé drapeau.10- Enrichissement asynchrone : Redis Stream (jamais bloquant pour l'indexation).11"""1213import json14from typing import Any1516import redis.asyncio as aioredis1718PAUSE_KEY = "trouveka:crawler:paused"19HOST_LOCK_PREFIX = "trouveka:host-lock:"20ENRICH_STREAM = "trouveka:enrich"212223class Coordination:24    def __init__(self, redis_url: str):25        self.redis: aioredis.Redis = aioredis.from_url(redis_url, decode_responses=True)2627    async def close(self) -> None:28        await self.redis.aclose()2930    # ------------------------------------------------------------- politesse31    async def acquire_host_slot(self, host: str, delay_seconds: float) -> bool:32        """Réserve le droit de fetcher cet hôte. False = trop tôt, repasser plus tard."""33        px = max(int(delay_seconds * 1000), 100)34        return bool(await self.redis.set(HOST_LOCK_PREFIX + host, "1", nx=True, px=px))3536    # ------------------------------------------------------------- pause37    async def pause_crawler(self) -> None:38        await self.redis.set(PAUSE_KEY, "1")3940    async def resume_crawler(self) -> None:41        await self.redis.delete(PAUSE_KEY)4243    async def is_paused(self) -> bool:44        return await self.redis.exists(PAUSE_KEY) == 14546    # ------------------------------------------------------------- enrichissement47    async def enqueue_enrichment(self, payload: dict[str, Any]) -> None:48        await self.redis.xadd(ENRICH_STREAM, {"data": json.dumps(payload, default=str)}, maxlen=100_000)4950    async def read_enrichment(51        self, group: str, consumer: str, count: int = 10, block_ms: int = 500052    ) -> list[tuple[str, dict[str, Any]]]:53        try:54            await self.redis.xgroup_create(ENRICH_STREAM, group, id="0", mkstream=True)55        except aioredis.ResponseError as exc:56            if "BUSYGROUP" not in str(exc):57                raise58        entries = await self.redis.xreadgroup(59            group, consumer, {ENRICH_STREAM: ">"}, count=count, block=block_ms60        )61        out: list[tuple[str, dict[str, Any]]] = []62        for _stream, items in entries or []:63            for msg_id, fields in items:64                out.append((msg_id, json.loads(fields["data"])))65        return out6667    async def ack_enrichment(self, group: str, msg_id: str) -> None:68        await self.redis.xack(ENRICH_STREAM, group, msg_id)6970    async def enrich_backlog(self) -> int:71        return await self.redis.xlen(ENRICH_STREAM)72