spb/company-atlas

1"""Sitemap connector (spec §9.2): `urlset` and `sitemapindex` (plain or .gz), `lastmod`, bounded by `settings.discovery_max_sitemap_urls`.2The observable change is the *set of URLs* (new / gone since the previous snapshot — the pipeline diffs `extracted.urls`) plus `lastmod`3moves; every URL is classified with `urls.classify_url` and emitted as `DiscoveredUrl` so the discovery feedback loop can add sensors.4Child sitemaps of an index are fetched breadth-first (bounded `MAX_CHILD_SITEMAPS`)."""5from __future__ import annotations67import gzip8import hashlib9import logging10import re11from collections.abc import Mapping12from typing import Any13from urllib.parse import urlparse1415from companyatlas.config import settings16from companyatlas.connectors._util import merged_result17from companyatlas.fetch import Fetcher, FetchResult18from companyatlas.sdk.connector import Connector, ConnectorContext, ConnectorMeta, register19from companyatlas.sdk.models import Block, DiscoveredUrl, Extraction20from companyatlas.sdk.normalize import simhash21from companyatlas.taxonomy import FetchMode, Surface22from companyatlas.urls import canonicalize_url, classify_url, is_static_asset, looks_like_trap2324log = logging.getLogger(__name__)2526MAX_CHILD_SITEMAPS = 2527LOC_RE = re.compile(r"<loc>\s*(?:<!\[CDATA\[)?\s*([^<\]\s]+)\s*(?:\]\]>)?\s*</loc>", re.IGNORECASE)28ENTRY_RE = re.compile(r"<(url|sitemap)>(.*?)</\1>", re.IGNORECASE | re.DOTALL)29LASTMOD_RE = re.compile(r"<lastmod>\s*([^<\s]+)\s*</lastmod>", re.IGNORECASE)30# Sitemap children whose names suggest low-value bulk content (products/tags/images) are visited last.31LOW_VALUE_CHILD_RE = re.compile(r"(image|video|tag|category|author|product|shop|store|collection|attachment|page-\d+|post-\d+)", re.IGNORECASE)32HIGH_VALUE_CHILD_RE = re.compile(r"(page|pages|static|main|site|news|press|blog|careers?|jobs?|misc|general)", re.IGNORECASE)333435def _decode(content: bytes) -> str:36    if content[:2] == b"\x1f\x8b":37        try:38            content = gzip.decompress(content)39        except (OSError, EOFError):40            return ""41    return content.decode("utf-8", errors="replace")424344def parse_sitemap(text: str) -> tuple[list[tuple[str, str | None]], list[tuple[str, str | None]]]:45    """→ (page_urls[(loc, lastmod)], child_sitemaps[(loc, lastmod)])."""46    pages: list[tuple[str, str | None]] = []47    children: list[tuple[str, str | None]] = []48    for m in ENTRY_RE.finditer(text):49        body = m.group(2)50        loc = LOC_RE.search(body)51        if not loc:52            continue53        lm = LASTMOD_RE.search(body)54        item = (loc.group(1).strip(), lm.group(1).strip() if lm else None)55        (children if m.group(1).lower() == "sitemap" else pages).append(item)56    if not pages and not children:      # tolerate sitemaps without <url> wrappers / text sitemaps57        for m in LOC_RE.finditer(text):58            pages.append((m.group(1).strip(), None))59        if not pages:60            for line in text.splitlines():61                line = line.strip()62                if line.startswith(("http://", "https://")):63                    pages.append((line, None))64    return pages, children656667@register68class SitemapConnector(Connector):69    meta = ConnectorMeta(connector_id="sitemap-v1", name="XML sitemap", version="1", category=Surface.SITEMAP, fetch_mode=FetchMode.SITEMAP,70                         supports_discovery=True, default_interval_s=24 * 3600, url_pattern=r"sitemap[^/]*\.xml(\.gz)?$|/sitemap_index\.xml$|/sitemaps?/",71                         priority=40, accept="application/xml,text/xml,*/*;q=0.5", description="urlset / sitemapindex with lastmod, bounded")7273    async def fetch(self, ctx: ConnectorContext, sensor: Mapping[str, Any], fetcher: Fetcher) -> FetchResult:74        first = await fetcher.get(str(sensor["url"]), etag=sensor.get("etag"), last_modified=sensor.get("last_modified"), accept=self.meta.accept)75        pages, children = parse_sitemap(_decode(first.content))76        if not children:77            return first78        limit = settings.discovery_max_sitemap_urls79        ordered = sorted(children, key=lambda c: (0 if HIGH_VALUE_CHILD_RE.search(c[0]) else (2 if LOW_VALUE_CHILD_RE.search(c[0]) else 1), c[0]))80        merged: list[dict[str, Any]] = [{"loc": u, "lastmod": lm} for u, lm in pages]81        visited: list[str] = []82        for child_url, _lm in ordered[:MAX_CHILD_SITEMAPS]:83            if len(merged) >= limit:84                break85            try:86                res = await fetcher.get(child_url, accept=self.meta.accept)87            except Exception as exc:  # noqa: BLE001 - one broken child must not fail the whole sitemap88                log.info("child sitemap skipped", extra={"child": child_url, "error": str(exc)[:200]})89                continue90            p, _c = parse_sitemap(_decode(res.content))91            merged.extend({"loc": u, "lastmod": lm} for u, lm in p[: max(0, limit - len(merged))])92            visited.append(child_url)93        payload = {"index": first.final_url, "children": [c[0] for c in ordered], "visited": visited, "urls": merged[:limit]}94        return merged_result(first, payload, pages=1 + len(visited))9596    def extract(self, sensor: Mapping[str, Any], result: FetchResult) -> Extraction:97        limit = settings.discovery_max_sitemap_urls98        if result.headers.get("x-companyatlas-pages"):99            data = result.json()100            entries = [(e.get("loc"), e.get("lastmod")) for e in data.get("urls", []) if e.get("loc")]101            children = list(data.get("children", []))102        else:103            entries, kids = parse_sitemap(_decode(result.content))104            children = [c[0] for c in kids]105        entries = entries[:limit]106        canonical_domain = str((sensor.get("config") or {}).get("canonical_domain") or urlparse(str(sensor["url"])).hostname or "")107        seen: set[str] = set()108        blocks: list[Block] = []109        discovered: list[DiscoveredUrl] = []110        by_surface: dict[str, int] = {}111        urls_out: list[dict[str, Any]] = []112        for loc, lastmod in entries:113            if not loc or is_static_asset(loc):114                continue115            canon = canonicalize_url(loc)116            if canon in seen:117                continue118            seen.add(canon)119            surface, conf = classify_url(loc, canonical_domain=canonical_domain)120            by_surface[str(surface)] = by_surface.get(str(surface), 0) + 1121            urls_out.append({"url": canon, "lastmod": lastmod, "surface": str(surface), "confidence": conf})122            key = f"url:{hashlib.blake2b(canon.encode('utf-8'), digest_size=8).hexdigest()}"123            blocks.append(Block(key=key, kind="list", text=canon, path="Sitemap", hash=hashlib.sha256(canon.encode()).hexdigest()[:16],124                                simhash=simhash(canon), weight=0.6, order=len(blocks), attrs={"lastmod": lastmod}))125            if conf >= settings.discovery_min_confidence and surface not in (Surface.OTHER, Surface.SITEMAP, Surface.HOMEPAGE) and not looks_like_trap(loc):126                discovered.append(DiscoveredUrl(url=loc, surface=surface, confidence=conf, method="sitemap"))127        text = "\n".join(u["url"] for u in urls_out)128        meta = {"url_count": len(urls_out), "child_sitemaps": children[:50], "by_surface": by_surface, "truncated": len(entries) >= limit,129                "urls": urls_out, "structured": True}130        return Extraction(text=text, blocks=blocks, title=f"Sitemap — {len(urls_out)} URLs", meta=meta, discovered=discovered[:500])131132133__all__ = ["SitemapConnector", "parse_sitemap"]134