"""Multi-page site crawl — ``crawl_site`` (phase 2 #92, WRITE + confirmation). The assistant can digest a small public site (documentation, docs portal) and store a Markdown summary inside a vault: one section per page, title, URL and readable text. The crawl is bounded and same-host only: * max 20 pages (``max_pages``), same hostname, breadth-first from the entry URL; * SSRF guard on every URL (scheme + private-address rejection), size caps; * no third-party crawler dependency (scrapy deliberately avoided — a bounded httpx BFS keeps the surface small and the runtime predictable; the task is executed as a single background-style tool run instead of a web request pipeline). Risk is WRITE: the digest is written into a vault, so the two-step confirmation applies (Apply card in the UI, propose/apply over MCP). """ from __future__ import annotations import logging import re import time from typing import Any from urllib.parse import urljoin, urlparse import httpx from backend.tools.context import ToolContext, ToolError, ToolRisk from backend.tools.registry import tool from backend.tools.schemas import CrawlSiteInput from backend.tools.web import ( USER_AGENT, _assert_public_http_url, _html_to_text, _response_text, ) logger = logging.getLogger("obsigate.tools.crawler") MAX_PAGE_BYTES = 800_000 MAX_TOTAL_BYTES = 6_000_000 MAX_TEXT_PER_PAGE = 12_000 PAGE_TIMEOUT = 10.0 _LINK_RE = re.compile(r']*href="([^"#]+)"', re.IGNORECASE) _TITLE_RE = re.compile(r"]*>(.*?)", re.IGNORECASE | re.DOTALL) def _same_host(url: str, host: str) -> bool: return (urlparse(url).hostname or "") == host def _extract_links(raw: str, base_url: str) -> list[str]: import html as html_lib links: list[str] = [] for match in _LINK_RE.finditer(raw): href = html_lib.unescape(match.group(1)).strip() if not href or href.lower().startswith(("javascript:", "mailto:", "tel:")): continue absolute = urljoin(base_url, href) if absolute.lower().endswith((".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp", ".pdf", ".zip")): continue links.append(absolute.split("#", 1)[0]) return links def _fetch_page(url: str) -> tuple[str, str]: """Fetch one page (SSRF-guarded, manual redirects) → (title, text).""" current = _assert_public_http_url(url) resp = None for _hop in range(5): resp = httpx.get( current, headers={"User-Agent": USER_AGENT, "Accept": "text/html,*/*"}, timeout=PAGE_TIMEOUT, follow_redirects=False, ) if resp.status_code in (301, 302, 303, 307, 308): location = resp.headers.get("location") or "" if not location: break current = _assert_public_http_url(str(httpx.URL(current).join(location))) continue break assert resp is not None resp.raise_for_status() ctype = (resp.headers.get("content-type") or "").lower() if "html" not in ctype and "text" not in ctype: raise ToolError( f"Type de contenu non pris en charge: {ctype.split(';')[0] or 'inconnu'}", code="unsupported_content_type", ) raw = (resp.content[:MAX_PAGE_BYTES]).decode(resp.encoding or "utf-8", errors="replace") title_match = _TITLE_RE.search(raw) import html as html_lib title = html_lib.unescape(title_match.group(1)).strip()[:300] if title_match else "" return title, _html_to_text(raw)[:MAX_TEXT_PER_PAGE] @tool( name="crawl_site", description=( "Crawl a small public site (same-host only, max 20 pages) starting at " "a URL and save a Markdown digest (title, url, readable text per page) " "into a vault. Use to capture an online documentation for offline use." ), input_model=CrawlSiteInput, risk=ToolRisk.WRITE, requires_vault=True, ) def crawl_site(ctx: ToolContext, params: CrawlSiteInput) -> dict[str, Any]: """Bounded BFS crawl; writes the digest file and returns a summary.""" from backend.services.errors import ServiceError from backend.services.mutations import save_raw_file start = _assert_public_http_url(params.url.strip()) host = urlparse(start).hostname or "" if not host: raise ToolError("URL sans hôte", code="invalid_url") queue: list[str] = [start] seen: set[str] = {start} pages: list[dict[str, Any]] = [] total_bytes = 0 failures: list[str] = [] while queue and len(pages) < params.max_pages and total_bytes < MAX_TOTAL_BYTES: url = queue.pop(0) try: title, text = _fetch_page(url) except ToolError as e: failures.append(url) logger.warning("crawl_site page failed %s: %s", url, e.code) continue except httpx.HTTPError as e: failures.append(url) logger.warning("crawl_site page failed %s: %s", url, e) continue pages.append({"url": url, "title": title, "text": text}) total_bytes += len(text) if len(pages) >= params.max_pages: break try: raw_resp = httpx.get( url, headers={"User-Agent": USER_AGENT}, timeout=PAGE_TIMEOUT, follow_redirects=False, ) raw = _response_text(raw_resp) except (httpx.HTTPError, ValueError): continue for link in _extract_links(raw, url): if len(pages) + len(queue) >= params.max_pages: break if link in seen or not _same_host(link, host): continue try: _assert_public_http_url(link) except ToolError: continue seen.add(link) queue.append(link) if not pages: raise ToolError( "Aucune page n'a pu être récupérée pour ce site.", code="crawl_failed", ) lines = [ f"# Crawl de {host}", "", f"> {len(pages)} page(s) capturée(s) depuis {start} — {time.strftime('%Y-%m-%d %H:%M')}", "", ] for page in pages: lines.append(f"## {page['title'] or page['url']}") lines.append("") lines.append(f"Source : {page['url']}") lines.append("") lines.append(page["text"]) lines.append("") digest = "\n".join(lines).encode("utf-8") try: saved = save_raw_file( params.vault, params.path, digest, overwrite=True, allow_docs=False ) except ServiceError as e: raise ToolError(e.message, code=e.code, details=e.details) from e return { "url": start, "vault": params.vault, "path": saved.get("path", params.path), "pages": len(pages), "failed": failures[:10], "size": saved.get("size", len(digest)), }