spb/trouve-ka Public

1# Trouve-KA — dépôts PostgreSQL2# Author: Simon-Pierre Boucher3# Contact: contact@spboucher.ai45"""Dépôts PostgreSQL de Trouve-KA.67Le frontier vit dans Postgres : les workers réclament des lots d'URLs via8FOR UPDATE SKIP LOCKED, ce qui permet N workers concurrents sans coordination9externe (CLAUDE.md §13 : pas de complexité distribuée prématurée).10"""1112from datetime import UTC, datetime, timedelta13from typing import Any1415import asyncpg161718def _dsn(url: str) -> str:19    # asyncpg accepte postgresql:// mais pas postgresql+asyncpg://20    return url.replace("postgresql+asyncpg://", "postgresql://")212223class Database:24    def __init__(self, database_url: str, *, pool_min: int = 2, pool_max: int = 10):25        self._url = _dsn(database_url)26        self._pool_min = pool_min27        self._pool_max = pool_max28        self.pool: asyncpg.Pool | None = None2930    async def connect(self) -> None:31        self.pool = await asyncpg.create_pool(32            self._url, min_size=self._pool_min, max_size=self._pool_max, command_timeout=3033        )3435    async def close(self) -> None:36        if self.pool:37            await self.pool.close()38            self.pool = None3940    # ------------------------------------------------------------------ domaines4142    async def upsert_domain(self, domain: str, *, is_seed: bool = False) -> int:43        row = await self.pool.fetchrow(44            """45            INSERT INTO domains (domain, is_seed) VALUES ($1, $2)46            ON CONFLICT (domain) DO UPDATE SET is_seed = domains.is_seed OR EXCLUDED.is_seed47            RETURNING id48            """,49            domain,50            is_seed,51        )52        return row["id"]5354    async def get_domain(self, domain: str) -> asyncpg.Record | None:55        return await self.pool.fetchrow("SELECT * FROM domains WHERE domain = $1", domain)5657    async def get_domain_by_id(self, domain_id: int) -> asyncpg.Record | None:58        return await self.pool.fetchrow("SELECT * FROM domains WHERE id = $1", domain_id)5960    async def save_robots(61        self, domain_id: int, body: str | None, status: str, crawl_delay_ms: int | None62    ) -> None:63        await self.pool.execute(64            """65            UPDATE domains SET robots_body = $2, robots_status = $3,66                   robots_fetched_at = now(), crawl_delay_ms = $467            WHERE id = $168            """,69            domain_id,70            body,71            status,72            crawl_delay_ms,73        )7475    async def block_domain(self, domain: str) -> None:76        await self.pool.execute(77            "UPDATE domains SET blocked = TRUE WHERE domain = $1", domain78        )79        await self.pool.execute(80            """81            UPDATE frontier_items SET status = 'blocked'82            WHERE status IN ('pending', 'in_progress')83              AND url_id IN (SELECT u.id FROM urls u JOIN domains d ON d.id = u.domain_id84                             WHERE d.domain = $1)85            """,86            domain,87        )8889    async def update_domain_after_page(90        self, domain_id: int, page_quebec_score: float, language: str | None91    ) -> None:92        """Met à jour le score Québec du domaine (moyenne mobile) et ses stats de langue."""93        lang = language or "unknown"94        await self.pool.execute(95            """96            UPDATE domains SET97                page_count = page_count + 1,98                last_crawled_at = now(),99                quebec_score = (quebec_score * page_count + $2) / (page_count + 1),100                language_stats = jsonb_set(101                    language_stats, ARRAY[$3],102                    (COALESCE(language_stats->>$3, '0')::int + 1)::text::jsonb)103            WHERE id = $1104            """,105            domain_id,106            page_quebec_score,107            lang,108        )109110    async def record_domain_links(self, from_domain_id: int, to_domain_counts: dict[int, int]) -> None:111        if not to_domain_counts:112            return113        await self.pool.executemany(114            """115            INSERT INTO domain_links (from_domain_id, to_domain_id, link_count)116            VALUES ($1, $2, $3)117            ON CONFLICT (from_domain_id, to_domain_id)118            DO UPDATE SET link_count = domain_links.link_count + EXCLUDED.link_count,119                          updated_at = now()120            """,121            [(from_domain_id, to_id, n) for to_id, n in to_domain_counts.items() if to_id != from_domain_id],122        )123124    # ------------------------------------------------------------------ frontier125126    async def enqueue_url(127        self,128        url: str,129        domain: str,130        *,131        priority: float = 0.5,132        depth: int = 0,133        source_url_id: int | None = None,134        is_seed: bool = False,135        max_urls_per_domain: int = 5000,136    ) -> int | None:137        """Ajoute une URL au frontier si inconnue. Retourne url_id si ajoutée, None sinon."""138        async with self.pool.acquire() as conn:139            async with conn.transaction():140                domain_id = await conn.fetchval(141                    """142                    INSERT INTO domains (domain, is_seed) VALUES ($1, $2)143                    ON CONFLICT (domain) DO UPDATE SET is_seed = domains.is_seed OR EXCLUDED.is_seed144                    RETURNING id145                    """,146                    domain,147                    is_seed,148                )149                dom = await conn.fetchrow(150                    "SELECT blocked, page_count FROM domains WHERE id = $1", domain_id151                )152                if dom["blocked"]:153                    return None154                if not is_seed and dom["page_count"] >= max_urls_per_domain:155                    return None156                url_id = await conn.fetchval(157                    """158                    INSERT INTO urls (url, domain_id) VALUES ($1, $2)159                    ON CONFLICT (url) DO NOTHING RETURNING id160                    """,161                    url,162                    domain_id,163                )164                if url_id is None:165                    return None  # URL déjà connue : pas de doublon dans le frontier166                await conn.execute(167                    """168                    INSERT INTO frontier_items (url_id, priority, depth, source_url_id)169                    VALUES ($1, $2, $3, $4) ON CONFLICT (url_id) DO NOTHING170                    """,171                    url_id,172                    priority,173                    depth,174                    source_url_id,175                )176                return url_id177178    async def claim_batch(self, worker_id: str, limit: int = 20) -> list[asyncpg.Record]:179        """Réclame un lot d'URLs prêtes à crawler (SKIP LOCKED, multi-workers sûr)."""180        return await self.pool.fetch(181            """182            WITH claimed AS (183                SELECT f.id FROM frontier_items f184                WHERE f.status = 'pending' AND f.next_crawl_at <= now()185                ORDER BY f.priority DESC, f.next_crawl_at186                LIMIT $2187                FOR UPDATE SKIP LOCKED188            )189            UPDATE frontier_items f SET status = 'in_progress', locked_by = $1, locked_at = now()190            FROM claimed191            WHERE f.id = claimed.id192            RETURNING f.id, f.url_id, f.priority, f.depth, f.retries,193                      f.last_crawled_at, f.next_crawl_at,194                      (SELECT url FROM urls WHERE id = f.url_id) AS url,195                      (SELECT domain_id FROM urls WHERE id = f.url_id) AS domain_id196            """,197            worker_id,198            limit,199        )200201    async def release_item(202        self,203        url_id: int,204        *,205        status: str,206        error_code: str | None = None,207        next_crawl_at: datetime | None = None,208        priority: float | None = None,209        increment_retries: bool = False,210    ) -> None:211        await self.pool.execute(212            """213            UPDATE frontier_items SET214                status = $2,215                error_code = $3,216                last_crawled_at = now(),217                next_crawl_at = COALESCE($4, next_crawl_at),218                priority = COALESCE($5, priority),219                retries = retries + CASE WHEN $6 THEN 1 ELSE 0 END,220                locked_by = NULL, locked_at = NULL221            WHERE url_id = $1222            """,223            url_id,224            status,225            error_code,226            next_crawl_at,227            priority,228            increment_retries,229        )230231    async def reset_stale_items(self, older_than_minutes: int = 30) -> int:232        """Relance les items in_progress abandonnés (worker mort). Retourne le nombre."""233        result = await self.pool.execute(234            """235            UPDATE frontier_items SET status = 'pending', locked_by = NULL, locked_at = NULL236            WHERE status = 'in_progress' AND locked_at < now() - ($1 || ' minutes')::interval237            """,238            str(older_than_minutes),239        )240        return int(result.split()[-1])241242    async def requeue_url(self, url: str) -> bool:243        result = await self.pool.execute(244            """245            UPDATE frontier_items SET status = 'pending', next_crawl_at = now(), retries = 0246            WHERE url_id = (SELECT id FROM urls WHERE url = $1)247            """,248            url,249        )250        return result.endswith("1")251252    async def requeue_domain(self, domain: str) -> int:253        result = await self.pool.execute(254            """255            UPDATE frontier_items SET status = 'pending', next_crawl_at = now()256            WHERE status IN ('done', 'failed')257              AND url_id IN (SELECT u.id FROM urls u JOIN domains d ON d.id = u.domain_id258                             WHERE d.domain = $1)259            """,260            domain,261        )262        return int(result.split()[-1])263264    # ------------------------------------------------------------------ crawl / documents265266    async def record_attempt(267        self,268        url_id: int,269        *,270        status_code: int | None,271        error_code: str,272        outcome: str,273        content_hash: str | None = None,274        num_bytes: int | None = None,275        duration_ms: int | None = None,276        redirect_url: str | None = None,277        title: str | None = None,278        quebec_score: float | None = None,279    ) -> None:280        await self.pool.execute(281            """282            INSERT INTO crawl_attempts283                (url_id, status_code, error_code, outcome, content_hash, bytes,284                 duration_ms, redirect_url, title, quebec_score)285            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)286            """,287            url_id,288            status_code,289            error_code,290            outcome,291            content_hash,292            num_bytes,293            duration_ms,294            redirect_url,295            title,296            quebec_score,297        )298299    async def get_document(self, url_id: int) -> asyncpg.Record | None:300        return await self.pool.fetchrow("SELECT * FROM documents WHERE url_id = $1", url_id)301302    async def find_duplicate(self, chash: str, url_id: int) -> asyncpg.Record | None:303        """Doublon exact : même hash de contenu sur une autre URL."""304        return await self.pool.fetchrow(305            "SELECT * FROM documents WHERE content_hash = $1 AND url_id != $2 LIMIT 1",306            chash,307            url_id,308        )309310    async def upsert_document(311        self,312        url_id: int,313        *,314        content_hash: str,315        etag: str | None,316        last_modified: str | None,317        title: str,318        description: str,319        language: str | None,320        page_quebec_score: float,321        published_at: datetime | None,322        changed: bool,323    ) -> int:324        row = await self.pool.fetchrow(325            """326            INSERT INTO documents327                (url_id, content_hash, etag, last_modified, title, description, language,328                 page_quebec_score, published_at, last_changed_at)329            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, now())330            ON CONFLICT (url_id) DO UPDATE SET331                content_hash = EXCLUDED.content_hash,332                etag = EXCLUDED.etag,333                last_modified = EXCLUDED.last_modified,334                title = EXCLUDED.title,335                description = EXCLUDED.description,336                language = EXCLUDED.language,337                page_quebec_score = EXCLUDED.page_quebec_score,338                published_at = EXCLUDED.published_at,339                last_indexed_at = now(),340                last_changed_at = CASE WHEN $10 THEN now() ELSE documents.last_changed_at END341            RETURNING id342            """,343            url_id,344            content_hash,345            etag,346            last_modified,347            title,348            description,349            language,350            page_quebec_score,351            published_at,352            changed,353        )354        return row["id"]355356    async def set_canonical(self, url_id: int, canonical_url: str) -> None:357        await self.pool.execute(358            "UPDATE urls SET canonical_url = $2 WHERE id = $1", url_id, canonical_url359        )360361    # ------------------------------------------------------------------ analytics / statut362363    async def record_search_query(364        self, query: str, language: str | None, total: int, took_ms: int365    ) -> None:366        await self.pool.execute(367            """368            INSERT INTO search_queries (query, language, results_total, took_ms, zero_result)369            VALUES ($1, $2, $3, $4, $3 = 0)370            """,371            query[:500],372            language,373            total,374            took_ms,375        )376377    async def add_submission(self, url: str) -> None:378        await self.pool.execute("INSERT INTO submissions (url) VALUES ($1)", url)379380    async def status_snapshot(self) -> dict[str, Any]:381        since = datetime.now(UTC) - timedelta(hours=1)382        frontier = {383            r["status"]: r["n"]384            for r in await self.pool.fetch(385                "SELECT status, count(*)::int AS n FROM frontier_items GROUP BY status"386            )387        }388        row = await self.pool.fetchrow(389            """390            SELECT391              (SELECT count(*)::int FROM documents) AS docs,392              (SELECT count(*)::int FROM domains WHERE page_count > 0) AS domains,393              (SELECT count(*)::int FROM documents WHERE last_indexed_at >= $1) AS indexed_1h,394              (SELECT count(*)::int FROM crawl_attempts WHERE fetched_at >= $1) AS fetched_1h,395              (SELECT count(*)::int FROM crawl_attempts396                 WHERE fetched_at >= $1 AND outcome = 'error') AS errors_1h397            """,398            since,399        )400        return {401            "pages_indexed": row["docs"],402            "domains_count": row["domains"],403            "indexed_last_hour": row["indexed_1h"],404            "fetched_last_hour": row["fetched_1h"],405            "errors_last_hour": row["errors_1h"],406            "frontier_pending": frontier.get("pending", 0),407            "frontier_in_progress": frontier.get("in_progress", 0),408            "frontier": frontier,409        }410411    async def admin_overview(self) -> dict[str, Any]:412        since = datetime.now(UTC) - timedelta(hours=1)413        base = await self.status_snapshot()414        http_status = {415            str(r["status_code"]): r["n"]416            for r in await self.pool.fetch(417                """418                SELECT status_code, count(*)::int AS n FROM crawl_attempts419                WHERE fetched_at >= $1 AND status_code IS NOT NULL420                GROUP BY status_code ORDER BY n DESC421                """,422                since,423            )424        }425        outcomes = {426            r["outcome"]: r["n"]427            for r in await self.pool.fetch(428                "SELECT outcome, count(*)::int AS n FROM crawl_attempts WHERE fetched_at >= $1 GROUP BY outcome",429                since,430            )431        }432        top_domains = [433            dict(r)434            for r in await self.pool.fetch(435                """436                SELECT domain, page_count AS pages, round(quebec_score::numeric, 3)::float AS quebec_score437                FROM domains WHERE page_count > 0438                ORDER BY page_count DESC LIMIT 20439                """440            )441        ]442        recent_errors = [443            {"url": r["url"], "error_code": r["error_code"], "at": r["fetched_at"].isoformat()}444            for r in await self.pool.fetch(445                """446                SELECT u.url, a.error_code, a.fetched_at447                FROM crawl_attempts a JOIN urls u ON u.id = a.url_id448                WHERE a.outcome = 'error' ORDER BY a.fetched_at DESC LIMIT 20449                """450            )451        ]452        latency = await self.pool.fetchrow(453            """454            SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY duration_ms) AS p50,455                   percentile_cont(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95456            FROM crawl_attempts WHERE fetched_at >= $1 AND duration_ms IS NOT NULL457            """,458            since,459        )460        return {461            "frontier": {462                "pending": base["frontier"].get("pending", 0),463                "in_progress": base["frontier"].get("in_progress", 0),464                "done": base["frontier"].get("done", 0),465                "failed": base["frontier"].get("failed", 0),466                "blocked": base["frontier"].get("blocked", 0),467            },468            "rates": {469                "fetched_1h": base["fetched_last_hour"],470                "indexed_1h": outcomes.get("indexed", 0),471                "errors_1h": outcomes.get("error", 0),472                "robots_blocked_1h": outcomes.get("robots_blocked", 0),473                "duplicates_1h": outcomes.get("duplicate", 0),474                "parsed_1h": outcomes.get("indexed", 0) + outcomes.get("not_quebec", 0),475            },476            "http_status": http_status,477            "top_domains": top_domains,478            "recent_errors": recent_errors,479            "latency": {480                "fetch_p50_ms": round(latency["p50"]) if latency and latency["p50"] else None,481                "fetch_p95_ms": round(latency["p95"]) if latency and latency["p95"] else None,482            },483        }484485    async def recent_events(self, limit: int = 50) -> list[dict[str, Any]]:486        rows = await self.pool.fetch(487            """488            SELECT a.fetched_at, u.url, a.status_code, a.outcome, a.quebec_score, a.title489            FROM crawl_attempts a JOIN urls u ON u.id = a.url_id490            ORDER BY a.fetched_at DESC LIMIT $1491            """,492            min(limit, 200),493        )494        return [495            {496                "at": r["fetched_at"].isoformat(),497                "url": r["url"],498                "status": r["status_code"],499                "outcome": r["outcome"],500                "quebec_score": r["quebec_score"],501                "title": r["title"],502            }503            for r in rows504        ]505506    async def frontier_inspect(507        self, domain: str | None = None, status: str | None = None, limit: int = 100508    ) -> list[dict[str, Any]]:509        rows = await self.pool.fetch(510            """511            SELECT u.url, f.priority, f.depth, f.status, f.retries, f.next_crawl_at, f.error_code512            FROM frontier_items f JOIN urls u ON u.id = f.url_id513            JOIN domains d ON d.id = u.domain_id514            WHERE ($1::text IS NULL OR d.domain = $1)515              AND ($2::text IS NULL OR f.status = $2)516            ORDER BY f.priority DESC LIMIT $3517            """,518            domain,519            status,520            min(limit, 500),521        )522        return [dict(r) | {"next_crawl_at": r["next_crawl_at"].isoformat()} for r in rows]523