198 lines
6.8 KiB
Python
198 lines
6.8 KiB
Python
"""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'<a[^>]*href="([^"#]+)"', re.IGNORECASE)
|
|
_TITLE_RE = re.compile(r"<title[^>]*>(.*?)</title>", 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)),
|
|
}
|