spb/trouve-ka Public

1# Trouve-KA — worker de crawl (pipeline complet)2# Author: Simon-Pierre Boucher3# Contact: contact@spboucher.ai45"""Worker de crawl : réclame des URLs au frontier et exécute le pipeline inline6fetch → parse → score Québec → indexation IMMÉDIATE (principe cardinal §0.3).78Chaque page traitée avec succès est cherchable en quelques secondes.9L'enrichissement (étapes 2-3) part dans Redis Streams et ne bloque jamais.10N workers peuvent tourner en parallèle (frontier SKIP LOCKED + locks Redis par hôte).11"""1213import asyncio14import os15import random16import signal17import uuid18from datetime import UTC, datetime, timedelta1920import httpx2122from trouveka.classifier import score_page23from trouveka.config import get_settings24from trouveka.database import Database25from trouveka.frontier import compute_priority, next_recrawl_delay, retry_delay26from trouveka.indexer import build_search_document27from trouveka.logging import get_logger28from trouveka.parser import looks_like_garbage, parse_html29from trouveka.queue import Coordination30from trouveka.search_core import SearchCore31from trouveka.shared import canonicalize_url, content_hash, extract_domain32from trouveka.types import ErrorCode, Outcome3334from .fetcher import Fetcher, scheme_host35from .robots import RobotsCache36from .traps import looks_like_trap3738log = get_logger("crawler.worker")3940TRANSIENT_ERRORS = {ErrorCode.TIMEOUT, ErrorCode.HTTP_5XX, ErrorCode.CONNECTION, ErrorCode.DNS}41MAX_RETRIES = 3424344class CrawlerWorker:45    def __init__(self) -> None:46        self.s = get_settings()47        self.worker_id = f"crawler-{os.getpid()}-{uuid.uuid4().hex[:6]}"48        self.db = Database(self.s.database_url, pool_min=self.s.pg_pool_min, pool_max=self.s.pg_pool_max)49        self.coord = Coordination(self.s.redis_url)50        self.search = SearchCore(self.s.search_url, self.s.search_index)51        self.client = httpx.AsyncClient(52            limits=httpx.Limits(53                max_connections=self.s.max_global_concurrency,54                max_keepalive_connections=self.s.max_global_concurrency,55            ),56            http2=False,57        )58        self.fetcher = Fetcher(self.client, self.s)59        self.robots: RobotsCache | None = None60        self.stop_event = asyncio.Event()61        self._domain_cache: dict[int, dict] = {}62        # Cache domaine→row pour la découverte : évite un SELECT par lien sortant63        # (une page peut avoir 300 liens, souvent vers les mêmes domaines).64        self._domain_by_name: dict[str, tuple[object, datetime]] = {}6566    # ------------------------------------------------------------------ cycle de vie6768    async def start(self) -> None:69        await self.db.connect()70        self.robots = RobotsCache(self.db, self.client, self.s.crawler_user_agent)71        await self.search.ensure_index()72        log.info("worker démarré", extra={"ctx": {"worker_id": self.worker_id}})7374        loop = asyncio.get_running_loop()75        for sig in (signal.SIGINT, signal.SIGTERM):76            loop.add_signal_handler(sig, self.stop_event.set)7778        # Pool continu : on réclame de nouvelles URLs dès qu'un slot se libère.79        # (Un gather par lot créait du head-of-line blocking : une page lente80        # bloquait tout le lot avant la réclamation suivante.)81        in_flight: set[asyncio.Task] = set()82        idle_sleep = 1.08384        async def run_one(item) -> None:85            try:86                await self.process_item(item)87            except Exception:88                log.exception(89                    "échec inattendu du pipeline",90                    extra={"ctx": {"url": item["url"], "worker_id": self.worker_id}},91                )92                await self.db.release_item(93                    item["url_id"], status="failed", error_code="parse_failed"94                )9596        while not self.stop_event.is_set():97            if await self.coord.is_paused():98                await asyncio.sleep(3)99                continue100            free = self.s.max_global_concurrency - len(in_flight)101            if free <= 0:102                await asyncio.wait(in_flight, return_when=asyncio.FIRST_COMPLETED)103                continue104            batch = await self.db.claim_batch(self.worker_id, limit=free)105            if not batch:106                if in_flight:107                    await asyncio.wait(108                        in_flight, return_when=asyncio.FIRST_COMPLETED, timeout=idle_sleep109                    )110                else:111                    await asyncio.sleep(idle_sleep)112                idle_sleep = min(idle_sleep * 1.5, 15)113                continue114            idle_sleep = 1.0115            for item in batch:116                task = asyncio.create_task(run_one(item))117                in_flight.add(task)118                task.add_done_callback(in_flight.discard)119120        if in_flight:121            await asyncio.gather(*in_flight, return_exceptions=True)122        await self.shutdown()123124    async def shutdown(self) -> None:125        await self.client.aclose()126        await self.search.close()127        await self.coord.close()128        await self.db.close()129        log.info("worker arrêté", extra={"ctx": {"worker_id": self.worker_id}})130131    # ------------------------------------------------------------------ helpers132133    async def _domain(self, domain_id: int) -> dict:134        cached = self._domain_cache.get(domain_id)135        if cached and (datetime.now(UTC) - cached["_at"]).total_seconds() < 60:136            return cached137        row = await self.db.get_domain_by_id(domain_id)138        entry = dict(row) | {"_at": datetime.now(UTC)}139        self._domain_cache[domain_id] = entry140        if len(self._domain_cache) > 5000:141            self._domain_cache.pop(next(iter(self._domain_cache)))142        return entry143144    async def _defer(self, url_id: int, seconds: float) -> None:145        """Repousse un item sans compter d'erreur (politesse : trop tôt pour cet hôte).146147        Jitter aléatoire pour désynchroniser les grappes d'URLs d'un même hôte148        (sinon elles reviennent toutes en tête de file au même instant et churnent).149        """150        await self.db.release_item(151            url_id,152            status="pending",153            next_crawl_at=datetime.now(UTC) + timedelta(seconds=seconds + random.uniform(0.2, 3.0)),154        )155156    async def _domain_row_by_name(self, domain: str):157        """Row domaine (ou None si inconnu), avec cache TTL 60 s."""158        cached = self._domain_by_name.get(domain)159        now = datetime.now(UTC)160        if cached and (now - cached[1]).total_seconds() < 60:161            return cached[0]162        row = await self.db.get_domain(domain)163        self._domain_by_name[domain] = (row, now)164        if len(self._domain_by_name) > 20_000:165            self._domain_by_name.pop(next(iter(self._domain_by_name)))166        return row167168    def _previous_delay_hours(self, item) -> float | None:169        if item["last_crawled_at"] and item["next_crawl_at"]:170            delta = (item["next_crawl_at"] - item["last_crawled_at"]).total_seconds() / 3600171            return max(delta, 0.1)172        return None173174    # ------------------------------------------------------------------ pipeline175176    async def process_item(self, item) -> None:177        url: str = item["url"]178        url_id: int = item["url_id"]179        domain_row = await self._domain(item["domain_id"])180        domain: str = domain_row["domain"]181182        if domain_row["blocked"]:183            await self.db.release_item(url_id, status="blocked")184            return185186        host = scheme_host(url)187188        # robots.txt d'abord (le fetch de robots ne compte pas dans la politesse)189        allowed, robots_delay = await self.robots.allowed(item["domain_id"], url, host)190        if not allowed:191            await self.db.record_attempt(192                url_id, status_code=None, error_code=ErrorCode.ROBOTS_DENIED,193                outcome=Outcome.ROBOTS_BLOCKED,194            )195            await self.db.release_item(url_id, status="done", error_code=ErrorCode.ROBOTS_DENIED)196            return197198        # Politesse par hôte, tous workers confondus199        delay = max(robots_delay or 0, self.s.default_host_delay)200        if not await self.coord.acquire_host_slot(domain, delay):201            await self._defer(url_id, delay + 0.5)202            return203204        # Cache HTTP conditionnel205        existing = await self.db.get_document(url_id)206        result = await self.fetcher.fetch(207            url,208            etag=existing["etag"] if existing else None,209            last_modified=existing["last_modified"] if existing else None,210        )211212        # --- 304 : inchangé213        if result.status_code == 304:214            await self._finish_unchanged(item, existing)215            return216217        # --- erreurs218        if result.error_code not in (ErrorCode.OK, ErrorCode.ROBOTS_DENIED) or result.body is None:219            await self._finish_error(item, result)220            return221222        # --- redirection vers une autre URL canonique : suivre la cible223        final = canonicalize_url(result.final_url) or result.final_url224        if final != url:225            final_domain = extract_domain(final)226            if final_domain and not looks_like_trap(final):227                await self.db.enqueue_url(228                    final, final_domain, priority=item["priority"], depth=item["depth"],229                    source_url_id=url_id, max_urls_per_domain=self.s.max_urls_per_domain,230                )231            await self.db.set_canonical(url_id, final)232            await self.db.record_attempt(233                url_id, status_code=result.status_code, error_code=ErrorCode.OK,234                outcome=Outcome.REDIRECT, redirect_url=final, duration_ms=result.duration_ms,235            )236            await self.db.release_item(url_id, status="done")237            return238239        # --- parse240        try:241            page = parse_html(242                url, result.body, max_links=self.s.max_links_per_page, charset=result.charset243            )244        except Exception:245            await self.db.record_attempt(246                url_id, status_code=result.status_code, error_code=ErrorCode.PARSE_FAILED,247                outcome=Outcome.ERROR, duration_ms=result.duration_ms,248            )249            await self.db.release_item(url_id, status="failed", error_code=ErrorCode.PARSE_FAILED)250            return251252        # --- garde anti-charabia : du binaire/mal décodé ne doit JAMAIS être indexé253        if looks_like_garbage(page.title) or looks_like_garbage(page.body):254            await self.db.record_attempt(255                url_id, status_code=result.status_code, error_code=ErrorCode.PARSE_FAILED,256                outcome=Outcome.ERROR, duration_ms=result.duration_ms, title=page.title[:80],257            )258            await self.db.release_item(url_id, status="failed", error_code=ErrorCode.PARSE_FAILED)259            return260261        # --- score Québec (immédiat, déterministe)262        signals = score_page(page, domain)263264        # --- détection de changement265        chash = content_hash(page.title, page.body)266        if existing and existing["content_hash"] == chash:267            await self._finish_unchanged(item, existing, status_code=result.status_code)268            await self._discover_links(item, page, signals)  # les liens peuvent avoir changé de priorité269            return270271        # --- doublon exact sur une autre URL272        duplicate = await self.db.find_duplicate(chash, url_id)273        if duplicate:274            await self.db.record_attempt(275                url_id, status_code=result.status_code, error_code=ErrorCode.DUPLICATE,276                outcome=Outcome.DUPLICATE, content_hash=chash, duration_ms=result.duration_ms,277                title=page.title, quebec_score=signals.score,278            )279            await self.db.release_item(280                url_id, status="done",281                next_crawl_at=datetime.now(UTC) + timedelta(hours=self.s.max_recrawl_hours),282            )283            return284285        indexable = (286            not page.noindex287            and result.error_code != ErrorCode.ROBOTS_DENIED  # X-Robots-Tag: noindex288            and signals.score >= self.s.min_quebec_score_to_index289            and len(page.body) >= self.s.min_body_length290        )291292        if indexable:293            # ------- INDEXATION IMMÉDIATE : la page est cherchable en ~1 s -------294            canonical_target = page.canonical_url or url295            doc = build_search_document(296                page, signals,297                domain=domain,298                domain_quebec_score=float(domain_row["quebec_score"]),299                authority_score=float(domain_row["authority_score"]),300            )301            await self.search.index_document(doc)302            await self.db.upsert_document(303                url_id,304                content_hash=chash,305                etag=result.etag,306                last_modified=result.last_modified,307                title=page.title,308                description=page.description,309                language=page.language,310                page_quebec_score=signals.score,311                published_at=page.published_at,312                changed=existing is not None,313            )314            if page.canonical_url and page.canonical_url != url:315                await self.db.set_canonical(url_id, canonical_target)316            # Enrichissement asynchrone (jamais bloquant)317            await self.coord.enqueue_enrichment(318                {"url": canonical_target, "url_id": url_id, "domain_id": item["domain_id"], "stage": 2}319            )320            outcome = Outcome.INDEXED321        else:322            outcome = Outcome.NOT_QUEBEC if signals.score < self.s.min_quebec_score_to_index else Outcome.UNCHANGED323324        await self.db.update_domain_after_page(item["domain_id"], signals.score, page.language)325        self._domain_cache.pop(item["domain_id"], None)326327        await self.db.record_attempt(328            url_id, status_code=result.status_code, error_code=ErrorCode.OK,329            outcome=outcome, content_hash=chash, num_bytes=len(result.body),330            duration_ms=result.duration_ms, title=page.title, quebec_score=signals.score,331        )332        recrawl = next_recrawl_delay(333            changed=True, previous_delay_hours=self._previous_delay_hours(item),334            min_hours=self.s.min_recrawl_hours, max_hours=self.s.max_recrawl_hours,335            default_hours=self.s.default_recrawl_hours,336        )337        await self.db.release_item(338            url_id, status="pending", next_crawl_at=datetime.now(UTC) + recrawl,339        )340341        await self._discover_links(item, page, signals)342        log.info(343            "page traitée",344            extra={"ctx": {345                "worker_id": self.worker_id, "url_id": url_id, "domain_id": item["domain_id"],346                "url": url, "outcome": str(outcome), "quebec_score": signals.score,347                "links": len(page.links), "ms": result.duration_ms,348            }},349        )350351    # ------------------------------------------------------------------ issues terminales352353    async def _finish_unchanged(self, item, existing, status_code: int | None = 304) -> None:354        recrawl = next_recrawl_delay(355            changed=False, previous_delay_hours=self._previous_delay_hours(item),356            min_hours=self.s.min_recrawl_hours, max_hours=self.s.max_recrawl_hours,357            default_hours=self.s.default_recrawl_hours,358        )359        await self.db.record_attempt(360            item["url_id"], status_code=status_code, error_code=ErrorCode.OK,361            outcome=Outcome.UNCHANGED,362            content_hash=existing["content_hash"] if existing else None,363        )364        await self.db.release_item(365            item["url_id"], status="pending",366            next_crawl_at=datetime.now(UTC) + recrawl,367        )368369    async def _finish_error(self, item, result) -> None:370        url_id = item["url_id"]371        await self.db.record_attempt(372            url_id, status_code=result.status_code, error_code=result.error_code,373            outcome=Outcome.ERROR, duration_ms=result.duration_ms,374        )375        transient = result.error_code in TRANSIENT_ERRORS376        if transient and item["retries"] < MAX_RETRIES:377            await self.db.release_item(378                url_id, status="pending", error_code=result.error_code,379                next_crawl_at=datetime.now(UTC) + retry_delay(item["retries"]),380                increment_retries=True,381            )382        else:383            await self.db.release_item(url_id, status="failed", error_code=result.error_code)384385    # ------------------------------------------------------------------ découverte386387    async def _discover_links(self, item, page, signals) -> None:388        """Boucle de découverte (§8) : liens sortants → scoring → frontier."""389        if item["depth"] >= self.s.max_crawl_depth:390            return391        domain_row = await self._domain(item["domain_id"])392        source_domain = domain_row["domain"]393        source_quebec = max(float(domain_row["quebec_score"]), signals.score)394395        outlink_domains: dict[int, int] = {}396        enqueued = 0397        for link in page.links:398            if link.nofollow or looks_like_trap(399                link.url,400                max_query_params=self.s.max_query_params,401                max_path_segments=self.s.max_path_segments,402            ):403                continue404            target_domain = extract_domain(link.url)405            if not target_domain:406                continue407            same_domain = target_domain == source_domain408            target_row = await self._domain_row_by_name(target_domain)409            is_new = target_row is None410            target_quebec = float(target_row["quebec_score"]) if target_row else 0.0411            # Un domaine découvert depuis une page québécoise hérite d'un a priori Québec412            effective_quebec = target_quebec if not is_new else source_quebec * 0.7413            if same_domain:414                effective_quebec = max(effective_quebec, source_quebec)415416            priority = compute_priority(417                domain_quebec_score=effective_quebec,418                authority_score=float(target_row["authority_score"]) if target_row else 0.0,419                link_signal=0.5 if not same_domain else 0.2,420                is_new_domain=is_new,421                depth=item["depth"] + 1,422            )423            # Économie de crawl : ignorer les cibles au signal Québec quasi nul424            if priority < 0.1:425                continue426            url_id = await self.db.enqueue_url(427                link.url, target_domain,428                priority=priority, depth=item["depth"] + 1, source_url_id=item["url_id"],429                max_urls_per_domain=self.s.max_urls_per_domain,430            )431            if url_id:432                enqueued += 1433            if not same_domain:434                target_id = (target_row["id"] if target_row435                             else (await self.db.upsert_domain(target_domain)))436                outlink_domains[target_id] = outlink_domains.get(target_id, 0) + 1437438        if outlink_domains:439            await self.db.record_domain_links(item["domain_id"], outlink_domains)440441442def main() -> None:443    asyncio.run(CrawlerWorker().start())444445446if __name__ == "__main__":447    main()448