# Trouve-KA — worker de crawl (pipeline complet) # Author: Simon-Pierre Boucher # Contact: contact@spboucher.ai """Worker de crawl : réclame des URLs au frontier et exécute le pipeline inline fetch → parse → score Québec → indexation IMMÉDIATE (principe cardinal §0.3). Chaque page traitée avec succès est cherchable en quelques secondes. L'enrichissement (étapes 2-3) part dans Redis Streams et ne bloque jamais. N workers peuvent tourner en parallèle (frontier SKIP LOCKED + locks Redis par hôte). """ import asyncio import os import random import signal import uuid from datetime import UTC, datetime, timedelta import httpx from trouveka.classifier import score_page from trouveka.config import get_settings from trouveka.database import Database from trouveka.frontier import compute_priority, next_recrawl_delay, retry_delay from trouveka.indexer import build_search_document from trouveka.logging import get_logger from trouveka.parser import looks_like_garbage, parse_html from trouveka.queue import Coordination from trouveka.search_core import SearchCore from trouveka.shared import canonicalize_url, content_hash, extract_domain from trouveka.types import ErrorCode, Outcome from .fetcher import Fetcher, scheme_host from .robots import RobotsCache from .traps import looks_like_trap log = get_logger("crawler.worker") TRANSIENT_ERRORS = {ErrorCode.TIMEOUT, ErrorCode.HTTP_5XX, ErrorCode.CONNECTION, ErrorCode.DNS} MAX_RETRIES = 3 class CrawlerWorker: def __init__(self) -> None: self.s = get_settings() self.worker_id = f"crawler-{os.getpid()}-{uuid.uuid4().hex[:6]}" self.db = Database(self.s.database_url, pool_min=self.s.pg_pool_min, pool_max=self.s.pg_pool_max) self.coord = Coordination(self.s.redis_url) self.search = SearchCore(self.s.search_url, self.s.search_index) self.client = httpx.AsyncClient( limits=httpx.Limits( max_connections=self.s.max_global_concurrency, max_keepalive_connections=self.s.max_global_concurrency, ), http2=False, ) self.fetcher = Fetcher(self.client, self.s) self.robots: RobotsCache | None = None self.stop_event = asyncio.Event() self._domain_cache: dict[int, dict] = {} # Cache domaine→row pour la découverte : évite un SELECT par lien sortant # (une page peut avoir 300 liens, souvent vers les mêmes domaines). self._domain_by_name: dict[str, tuple[object, datetime]] = {} # ------------------------------------------------------------------ cycle de vie async def start(self) -> None: await self.db.connect() self.robots = RobotsCache(self.db, self.client, self.s.crawler_user_agent) await self.search.ensure_index() log.info("worker démarré", extra={"ctx": {"worker_id": self.worker_id}}) loop = asyncio.get_running_loop() for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, self.stop_event.set) # Pool continu : on réclame de nouvelles URLs dès qu'un slot se libère. # (Un gather par lot créait du head-of-line blocking : une page lente # bloquait tout le lot avant la réclamation suivante.) in_flight: set[asyncio.Task] = set() idle_sleep = 1.0 async def run_one(item) -> None: try: await self.process_item(item) except Exception: log.exception( "échec inattendu du pipeline", extra={"ctx": {"url": item["url"], "worker_id": self.worker_id}}, ) await self.db.release_item( item["url_id"], status="failed", error_code="parse_failed" ) while not self.stop_event.is_set(): if await self.coord.is_paused(): await asyncio.sleep(3) continue free = self.s.max_global_concurrency - len(in_flight) if free <= 0: await asyncio.wait(in_flight, return_when=asyncio.FIRST_COMPLETED) continue batch = await self.db.claim_batch(self.worker_id, limit=free) if not batch: if in_flight: await asyncio.wait( in_flight, return_when=asyncio.FIRST_COMPLETED, timeout=idle_sleep ) else: await asyncio.sleep(idle_sleep) idle_sleep = min(idle_sleep * 1.5, 15) continue idle_sleep = 1.0 for item in batch: task = asyncio.create_task(run_one(item)) in_flight.add(task) task.add_done_callback(in_flight.discard) if in_flight: await asyncio.gather(*in_flight, return_exceptions=True) await self.shutdown() async def shutdown(self) -> None: await self.client.aclose() await self.search.close() await self.coord.close() await self.db.close() log.info("worker arrêté", extra={"ctx": {"worker_id": self.worker_id}}) # ------------------------------------------------------------------ helpers async def _domain(self, domain_id: int) -> dict: cached = self._domain_cache.get(domain_id) if cached and (datetime.now(UTC) - cached["_at"]).total_seconds() < 60: return cached row = await self.db.get_domain_by_id(domain_id) entry = dict(row) | {"_at": datetime.now(UTC)} self._domain_cache[domain_id] = entry if len(self._domain_cache) > 5000: self._domain_cache.pop(next(iter(self._domain_cache))) return entry async def _defer(self, url_id: int, seconds: float) -> None: """Repousse un item sans compter d'erreur (politesse : trop tôt pour cet hôte). Jitter aléatoire pour désynchroniser les grappes d'URLs d'un même hôte (sinon elles reviennent toutes en tête de file au même instant et churnent). """ await self.db.release_item( url_id, status="pending", next_crawl_at=datetime.now(UTC) + timedelta(seconds=seconds + random.uniform(0.2, 3.0)), ) async def _domain_row_by_name(self, domain: str): """Row domaine (ou None si inconnu), avec cache TTL 60 s.""" cached = self._domain_by_name.get(domain) now = datetime.now(UTC) if cached and (now - cached[1]).total_seconds() < 60: return cached[0] row = await self.db.get_domain(domain) self._domain_by_name[domain] = (row, now) if len(self._domain_by_name) > 20_000: self._domain_by_name.pop(next(iter(self._domain_by_name))) return row def _previous_delay_hours(self, item) -> float | None: if item["last_crawled_at"] and item["next_crawl_at"]: delta = (item["next_crawl_at"] - item["last_crawled_at"]).total_seconds() / 3600 return max(delta, 0.1) return None # ------------------------------------------------------------------ pipeline async def process_item(self, item) -> None: url: str = item["url"] url_id: int = item["url_id"] domain_row = await self._domain(item["domain_id"]) domain: str = domain_row["domain"] if domain_row["blocked"]: await self.db.release_item(url_id, status="blocked") return host = scheme_host(url) # robots.txt d'abord (le fetch de robots ne compte pas dans la politesse) allowed, robots_delay = await self.robots.allowed(item["domain_id"], url, host) if not allowed: await self.db.record_attempt( url_id, status_code=None, error_code=ErrorCode.ROBOTS_DENIED, outcome=Outcome.ROBOTS_BLOCKED, ) await self.db.release_item(url_id, status="done", error_code=ErrorCode.ROBOTS_DENIED) return # Politesse par hôte, tous workers confondus delay = max(robots_delay or 0, self.s.default_host_delay) if not await self.coord.acquire_host_slot(domain, delay): await self._defer(url_id, delay + 0.5) return # Cache HTTP conditionnel existing = await self.db.get_document(url_id) result = await self.fetcher.fetch( url, etag=existing["etag"] if existing else None, last_modified=existing["last_modified"] if existing else None, ) # --- 304 : inchangé if result.status_code == 304: await self._finish_unchanged(item, existing) return # --- erreurs if result.error_code not in (ErrorCode.OK, ErrorCode.ROBOTS_DENIED) or result.body is None: await self._finish_error(item, result) return # --- redirection vers une autre URL canonique : suivre la cible final = canonicalize_url(result.final_url) or result.final_url if final != url: final_domain = extract_domain(final) if final_domain and not looks_like_trap(final): await self.db.enqueue_url( final, final_domain, priority=item["priority"], depth=item["depth"], source_url_id=url_id, max_urls_per_domain=self.s.max_urls_per_domain, ) await self.db.set_canonical(url_id, final) await self.db.record_attempt( url_id, status_code=result.status_code, error_code=ErrorCode.OK, outcome=Outcome.REDIRECT, redirect_url=final, duration_ms=result.duration_ms, ) await self.db.release_item(url_id, status="done") return # --- parse try: page = parse_html( url, result.body, max_links=self.s.max_links_per_page, charset=result.charset ) except Exception: await self.db.record_attempt( url_id, status_code=result.status_code, error_code=ErrorCode.PARSE_FAILED, outcome=Outcome.ERROR, duration_ms=result.duration_ms, ) await self.db.release_item(url_id, status="failed", error_code=ErrorCode.PARSE_FAILED) return # --- garde anti-charabia : du binaire/mal décodé ne doit JAMAIS être indexé if looks_like_garbage(page.title) or looks_like_garbage(page.body): await self.db.record_attempt( url_id, status_code=result.status_code, error_code=ErrorCode.PARSE_FAILED, outcome=Outcome.ERROR, duration_ms=result.duration_ms, title=page.title[:80], ) await self.db.release_item(url_id, status="failed", error_code=ErrorCode.PARSE_FAILED) return # --- score Québec (immédiat, déterministe) signals = score_page(page, domain) # --- détection de changement chash = content_hash(page.title, page.body) if existing and existing["content_hash"] == chash: await self._finish_unchanged(item, existing, status_code=result.status_code) await self._discover_links(item, page, signals) # les liens peuvent avoir changé de priorité return # --- doublon exact sur une autre URL duplicate = await self.db.find_duplicate(chash, url_id) if duplicate: await self.db.record_attempt( url_id, status_code=result.status_code, error_code=ErrorCode.DUPLICATE, outcome=Outcome.DUPLICATE, content_hash=chash, duration_ms=result.duration_ms, title=page.title, quebec_score=signals.score, ) await self.db.release_item( url_id, status="done", next_crawl_at=datetime.now(UTC) + timedelta(hours=self.s.max_recrawl_hours), ) return indexable = ( not page.noindex and result.error_code != ErrorCode.ROBOTS_DENIED # X-Robots-Tag: noindex and signals.score >= self.s.min_quebec_score_to_index and len(page.body) >= self.s.min_body_length ) if indexable: # ------- INDEXATION IMMÉDIATE : la page est cherchable en ~1 s ------- canonical_target = page.canonical_url or url doc = build_search_document( page, signals, domain=domain, domain_quebec_score=float(domain_row["quebec_score"]), authority_score=float(domain_row["authority_score"]), ) await self.search.index_document(doc) await self.db.upsert_document( url_id, content_hash=chash, etag=result.etag, last_modified=result.last_modified, title=page.title, description=page.description, language=page.language, page_quebec_score=signals.score, published_at=page.published_at, changed=existing is not None, ) if page.canonical_url and page.canonical_url != url: await self.db.set_canonical(url_id, canonical_target) # Enrichissement asynchrone (jamais bloquant) await self.coord.enqueue_enrichment( {"url": canonical_target, "url_id": url_id, "domain_id": item["domain_id"], "stage": 2} ) outcome = Outcome.INDEXED else: outcome = Outcome.NOT_QUEBEC if signals.score < self.s.min_quebec_score_to_index else Outcome.UNCHANGED await self.db.update_domain_after_page(item["domain_id"], signals.score, page.language) self._domain_cache.pop(item["domain_id"], None) await self.db.record_attempt( url_id, status_code=result.status_code, error_code=ErrorCode.OK, outcome=outcome, content_hash=chash, num_bytes=len(result.body), duration_ms=result.duration_ms, title=page.title, quebec_score=signals.score, ) recrawl = next_recrawl_delay( changed=True, previous_delay_hours=self._previous_delay_hours(item), min_hours=self.s.min_recrawl_hours, max_hours=self.s.max_recrawl_hours, default_hours=self.s.default_recrawl_hours, ) await self.db.release_item( url_id, status="pending", next_crawl_at=datetime.now(UTC) + recrawl, ) await self._discover_links(item, page, signals) log.info( "page traitée", extra={"ctx": { "worker_id": self.worker_id, "url_id": url_id, "domain_id": item["domain_id"], "url": url, "outcome": str(outcome), "quebec_score": signals.score, "links": len(page.links), "ms": result.duration_ms, }}, ) # ------------------------------------------------------------------ issues terminales async def _finish_unchanged(self, item, existing, status_code: int | None = 304) -> None: recrawl = next_recrawl_delay( changed=False, previous_delay_hours=self._previous_delay_hours(item), min_hours=self.s.min_recrawl_hours, max_hours=self.s.max_recrawl_hours, default_hours=self.s.default_recrawl_hours, ) await self.db.record_attempt( item["url_id"], status_code=status_code, error_code=ErrorCode.OK, outcome=Outcome.UNCHANGED, content_hash=existing["content_hash"] if existing else None, ) await self.db.release_item( item["url_id"], status="pending", next_crawl_at=datetime.now(UTC) + recrawl, ) async def _finish_error(self, item, result) -> None: url_id = item["url_id"] await self.db.record_attempt( url_id, status_code=result.status_code, error_code=result.error_code, outcome=Outcome.ERROR, duration_ms=result.duration_ms, ) transient = result.error_code in TRANSIENT_ERRORS if transient and item["retries"] < MAX_RETRIES: await self.db.release_item( url_id, status="pending", error_code=result.error_code, next_crawl_at=datetime.now(UTC) + retry_delay(item["retries"]), increment_retries=True, ) else: await self.db.release_item(url_id, status="failed", error_code=result.error_code) # ------------------------------------------------------------------ découverte async def _discover_links(self, item, page, signals) -> None: """Boucle de découverte (§8) : liens sortants → scoring → frontier.""" if item["depth"] >= self.s.max_crawl_depth: return domain_row = await self._domain(item["domain_id"]) source_domain = domain_row["domain"] source_quebec = max(float(domain_row["quebec_score"]), signals.score) outlink_domains: dict[int, int] = {} enqueued = 0 for link in page.links: if link.nofollow or looks_like_trap( link.url, max_query_params=self.s.max_query_params, max_path_segments=self.s.max_path_segments, ): continue target_domain = extract_domain(link.url) if not target_domain: continue same_domain = target_domain == source_domain target_row = await self._domain_row_by_name(target_domain) is_new = target_row is None target_quebec = float(target_row["quebec_score"]) if target_row else 0.0 # Un domaine découvert depuis une page québécoise hérite d'un a priori Québec effective_quebec = target_quebec if not is_new else source_quebec * 0.7 if same_domain: effective_quebec = max(effective_quebec, source_quebec) priority = compute_priority( domain_quebec_score=effective_quebec, authority_score=float(target_row["authority_score"]) if target_row else 0.0, link_signal=0.5 if not same_domain else 0.2, is_new_domain=is_new, depth=item["depth"] + 1, ) # Économie de crawl : ignorer les cibles au signal Québec quasi nul if priority < 0.1: continue url_id = await self.db.enqueue_url( link.url, target_domain, priority=priority, depth=item["depth"] + 1, source_url_id=item["url_id"], max_urls_per_domain=self.s.max_urls_per_domain, ) if url_id: enqueued += 1 if not same_domain: target_id = (target_row["id"] if target_row else (await self.db.upsert_domain(target_domain))) outlink_domains[target_id] = outlink_domains.get(target_id, 0) + 1 if outlink_domains: await self.db.record_domain_links(item["domain_id"], outlink_domains) def main() -> None: asyncio.run(CrawlerWorker().start()) if __name__ == "__main__": main()