From df7bd7c6f9743093ee788bc43a2596553c8a21aa Mon Sep 17 00:00:00 2001
From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com>
Date: Thu, 25 Jun 2026 18:28:14 +0100
Subject: [PATCH 1/6] feat: centralizza estrazione e raggruppamento link HTML
in link_extractor
Crea toolkit/scout/link_extractor.py come contratto centrale per
estrazione e raggruppamento di link dati da pagine HTML.
- DataLink e LinkGroup come dataclass con metadati (formato, prefisso, anni)
- extract_data_links() unifica le precedenti implementazioni
- group_links() per raggruppamento intelligente
- extract_candidate_links() mantenuto come bridge backward compat
- http.py: rimosso _AnchorParser, extract_candidate_links redirect
- mcp/scout_ops.py: response arricchito con data_links[] e groups[]
File: +link_extractor.py, __init__.py, http.py, scout_ops.py
---
toolkit/mcp/scout_ops.py | 53 +++++-
toolkit/scout/__init__.py | 18 +-
toolkit/scout/http.py | 46 +----
toolkit/scout/link_extractor.py | 301 ++++++++++++++++++++++++++++++++
4 files changed, 369 insertions(+), 49 deletions(-)
create mode 100644 toolkit/scout/link_extractor.py
diff --git a/toolkit/mcp/scout_ops.py b/toolkit/mcp/scout_ops.py
index 2d4c33bb..3715608d 100644
--- a/toolkit/mcp/scout_ops.py
+++ b/toolkit/mcp/scout_ops.py
@@ -10,7 +10,6 @@
from toolkit.plugins.sparql import SparqlSource
from toolkit.scout.http import (
- extract_candidate_links,
fetch_ckan_package,
fetch_html_body,
list_sdmx_dataflows as _list_sdmx_dataflows,
@@ -18,6 +17,10 @@
search_ckan_datasets as _search_ckan_datasets,
)
from toolkit.scout.infer import infer_topics
+from toolkit.scout.link_extractor import (
+ extract_data_links,
+ group_links,
+)
from toolkit.scout.probe import probe_url, probe_url_routed
@@ -144,11 +147,18 @@ def mcp_ckan_package_show(
def mcp_html_extract_links(url: str, timeout: int = 20) -> dict[str, Any]:
"""Estrae link a file dati (CSV, JSON, XLSX, ZIP, XML) da una pagina HTML.
- 1. Scrive ``fetch_html_body()`` per scaricare la pagina.
- 2. ``extract_candidate_links()`` per estrarre i link ai dati.
+ 1. Scarica la pagina via ``fetch_html_body()``.
+ 2. Usa ``extract_data_links()`` per estrarre link con metadati (formato,
+ prefisso, anni).
+ 3. Usa ``group_links()`` per raggruppare i link in dataset.
Returns:
- Dict con url, total, formats, links, is_reachable.
+ Dict con:
+ - ``url``, ``is_reachable``, ``http_status``
+ - ``links`` (list[str], backward compat) e ``total``
+ - ``formats`` (dict[str, int], backward compat, lowercase estensione)
+ - ``data_links`` (list[dict] con url, format, title, prefix, years, page_url)
+ - ``groups`` (list[dict] con group_id, prefix, count, year_range, formats)
"""
try:
body = fetch_html_body(url, timeout=timeout)
@@ -160,14 +170,21 @@ def mcp_html_extract_links(url: str, timeout: int = 20) -> dict[str, Any]:
"links": [],
"total": 0,
"formats": {},
+ "data_links": [],
+ "groups": [],
}
html_text = body.get("html_text", "")
- links = extract_candidate_links(url, html_text)
+ data_links = extract_data_links(url, html_text)
+ groups = group_links(data_links)
+
+ # Backward compat: lista di URL semplici
+ links = [dl.url for dl in data_links]
+ # Backward compat: conteggio per estensione (lowercase)
formats: dict[str, int] = {}
- for link in links:
- ext = link.rsplit(".", 1)[-1].lower() if "." in link else "unknown"
+ for dl in data_links:
+ ext = dl.url.rsplit(".", 1)[-1].lower() if "." in dl.url else "unknown"
formats[ext] = formats.get(ext, 0) + 1
return {
@@ -177,6 +194,28 @@ def mcp_html_extract_links(url: str, timeout: int = 20) -> dict[str, Any]:
"total": len(links),
"links": links,
"formats": formats,
+ "data_links": [
+ {
+ "url": dl.url,
+ "format": dl.format,
+ "title": dl.title,
+ "prefix": dl.prefix,
+ "years": dl.years,
+ "page_url": dl.page_url or url,
+ }
+ for dl in data_links
+ ],
+ "groups": [
+ {
+ "group_id": g.group_id,
+ "prefix": g.prefix,
+ "count": g.count,
+ "year_range": g.year_range,
+ "formats": sorted(g.formats),
+ "title": g.title,
+ }
+ for g in groups
+ ],
}
diff --git a/toolkit/scout/__init__.py b/toolkit/scout/__init__.py
index 1768dd63..80eb07ba 100644
--- a/toolkit/scout/__init__.py
+++ b/toolkit/scout/__init__.py
@@ -7,10 +7,11 @@
- Notebook e script interni
Moduli:
- http → HTTP transport: probe, fetch, format detection, CKAN/SDMX
- sparql → SPARQL scout: named graph discovery, schema inference
- infer → Inferenze pure: anni, granularità, topic
- probe → Orchestrazione: probe_url(), probe_url_routed()
+ http → HTTP transport: probe, fetch, format detection, CKAN/SDMX
+ link_extractor → Estrazione e raggruppamento link dati da HTML
+ sparql → SPARQL scout: named graph discovery, schema inference
+ infer → Inferenze pure: anni, granularità, topic
+ probe → Orchestrazione: probe_url(), probe_url_routed()
Scaffold non e' piu' in scout. Usa toolkit.scaffold.full e toolkit.scaffold.sources.
"""
@@ -34,13 +35,20 @@
resolve_preview_kind,
detect_ckan_in_html,
extract_ckan_dataset_id,
- extract_candidate_links,
DEFAULT_TIMEOUT,
DEFAULT_USER_AGENT,
CANDIDATE_EXTENSIONS,
EXTENDED_EXTENSIONS,
)
+from toolkit.scout.link_extractor import ( # noqa: F401
+ DataLink,
+ LinkGroup,
+ extract_data_links,
+ group_links,
+ extract_candidate_links,
+)
+
from toolkit.scout.infer import ( # noqa: F401
infer_years,
suggest_years,
diff --git a/toolkit/scout/http.py b/toolkit/scout/http.py
index 400729f4..d6af7f09 100644
--- a/toolkit/scout/http.py
+++ b/toolkit/scout/http.py
@@ -11,9 +11,8 @@
import logging
import re
import time
-from html.parser import HTMLParser
from typing import Any
-from urllib.parse import parse_qs, urljoin, urlparse
+from urllib.parse import parse_qs, urlparse
from lab_connectors.http import CircuitOpenError, HttpClient
@@ -56,27 +55,6 @@
_YEAR_RE = re.compile(r"(? da HTML per estrarre link candidati a dati."""
-
- def __init__(self) -> None:
- super().__init__()
- self.hrefs: list[str] = []
-
- def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
- if tag.lower() != "a":
- return
- for key, value in attrs:
- if key.lower() == "href" and value:
- self.hrefs.append(value)
- return
-
-
# ---------------------------------------------------------------------------
# HTTP client factory
# ---------------------------------------------------------------------------
@@ -201,6 +179,8 @@ def _filename_from_content_disposition(value: str | None) -> str | None:
# ---------------------------------------------------------------------------
# Candidate links extraction from HTML (pure, no network)
+# Reindirizzato a link_extractor come contratto centrale.
+# Mantenuto per backward compatibility.
# ---------------------------------------------------------------------------
@@ -208,21 +188,13 @@ def extract_candidate_links(base_url: str, html_text: str) -> list[str]:
"""Estrae link a file dati (CSV/XLSX/etc.) da una pagina HTML.
Returns lista di URL assoluti, deduplicati, ordinati per apparizione.
+
+ Nota: reindirizzato a ``toolkit.scout.link_extractor`` come contratto
+ centrale. Questa funzione è mantenuta per backward compatibility.
"""
- parser = _AnchorParser()
- parser.feed(html_text)
- links: list[str] = []
- seen: set[str] = set()
- for href in parser.hrefs:
- lowered = href.lower()
- if not any(ext in lowered for ext in CANDIDATE_EXTENSIONS):
- continue
- absolute = urljoin(base_url, href)
- if absolute in seen:
- continue
- seen.add(absolute)
- links.append(absolute)
- return links
+ from toolkit.scout.link_extractor import extract_candidate_links as _extract
+
+ return _extract(base_url, html_text)
# ---------------------------------------------------------------------------
diff --git a/toolkit/scout/link_extractor.py b/toolkit/scout/link_extractor.py
new file mode 100644
index 00000000..f3c95cce
--- /dev/null
+++ b/toolkit/scout/link_extractor.py
@@ -0,0 +1,301 @@
+"""Estrattore e raggruppatore di link dati da pagine HTML.
+
+Contratto centrale per l'estrazione di link a file dati da pagine HTML.
+Unifica le precedenti implementazioni in:
+- ``toolkit.scout.http.extract_candidate_links`` (toolkit)
+- ``collectors.html._extract_data_links`` (source-observatory)
+
+Consumer:
+ - ``toolkit.scout.http`` → redirect a questo modulo
+ - ``toolkit.scaffold.sources`` → raggruppamento per scaffold
+ - ``toolkit.mcp.scout_ops`` → MCP tool
+ - ``source-observatory/collectors/html.py`` → inventory HTML
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from html.parser import HTMLParser
+from urllib.parse import urljoin
+
+# ---------------------------------------------------------------------------
+# Costanti
+# ---------------------------------------------------------------------------
+
+# Estensioni di file dati riconosciute. Ordinate per lunghezza decrescente
+# per matchare .xlsx prima di .xls e .geojson prima di .json.
+DATA_EXTENSIONS = {
+ ".csv",
+ ".json",
+ ".xlsx",
+ ".xls",
+ ".ods",
+ ".zip",
+ ".xml",
+ ".parquet",
+ ".geojson",
+}
+
+# Pattern per estrarre prefisso e anni dal filename
+_PREFIX_RE = re.compile(r"^([A-Z][A-Z0-9]{1,7})")
+_YEAR_RE = re.compile(r"(20[012]\d)")
+
+# ---------------------------------------------------------------------------
+# Dataclass
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class DataLink:
+ """Un link a un file dati estratto da HTML.
+
+ Attributes:
+ url: URL assoluto del file.
+ format: Formato in maiuscolo (CSV, XLSX, ZIP, ...).
+ title: Titolo estratto da aria-label o title attribute.
+ prefix: Prefisso categorico estratto dal filename.
+ years: Anni (20xx) trovati nel filename.
+ page_url: URL della pagina HTML da cui è stato estratto.
+ """
+
+ url: str
+ format: str
+ title: str = ""
+ prefix: str = ""
+ years: list[int] = field(default_factory=list)
+ page_url: str = ""
+
+
+@dataclass
+class LinkGroup:
+ """Gruppo di link che appartengono allo stesso dataset.
+
+ Attributes:
+ group_id: Identificatore univoco del gruppo (es. prefisso).
+ title: Titolo descrittivo del gruppo.
+ prefix: Prefisso comune.
+ year_range: Anni minimo e massimo coperti.
+ formats: Formati disponibili.
+ links: Lista dei DataLink del gruppo.
+ count: Numero di link nel gruppo.
+ """
+
+ group_id: str
+ title: str
+ prefix: str
+ year_range: list[int]
+ formats: set[str]
+ links: list[DataLink]
+ count: int = 0
+
+
+# ---------------------------------------------------------------------------
+# Parsing HTML
+# ---------------------------------------------------------------------------
+
+
+def _extract_filename(url: str) -> str:
+ """Estrae il filename (senza estensione) da un URL."""
+ from urllib.parse import unquote, urlparse
+
+ path = urlparse(url).path
+ # Rimuovi query string
+ name = path.rsplit("/", 1)[-1] if "/" in path else path
+ name = unquote(name)
+ # Rimuovi estensione
+ dot = name.rfind(".")
+ if dot > 0:
+ name = name[:dot]
+ return name
+
+
+def _extract_prefix(filename: str) -> str:
+ """Estrae un prefisso categorico dal filename.
+
+ - Se contiene underscore, prende la prima parte.
+ - Altrimenti matcha una run di maiuscole/digits all'inizio.
+ - Fallback: primi 6 caratteri.
+ """
+ if "_" in filename:
+ return filename.split("_")[0]
+ m = _PREFIX_RE.match(filename)
+ if m:
+ return m.group(1)
+ return filename[:6]
+
+
+def _extract_years(filename: str) -> list[int]:
+ """Estrae tutti gli anni (20xx) presenti nel filename."""
+ return [int(y) for y in _YEAR_RE.findall(filename)]
+
+
+def _resolve_format(url: str) -> str | None:
+ """Determina il formato dall'estensione del file nell'URL."""
+ lower = url.lower()
+ # Ordina per lunghezza decrescente: .xlsx prima di .xls, .geojson prima di .json
+ for ext in sorted(DATA_EXTENSIONS, key=len, reverse=True):
+ # Cerca l'estensione nel path (non nella query string)
+ path = url.split("?")[0].lower()
+ if path.endswith(ext):
+ return ext.lstrip(".").upper()
+ # fallback: cerca estensione in tutto l'URL
+ check = lower.rstrip("/")
+ if check.endswith(ext):
+ return ext.lstrip(".").upper()
+ return None
+
+
+# ---------------------------------------------------------------------------
+# Parser HTML
+# ---------------------------------------------------------------------------
+
+
+class _DataLinkParser(HTMLParser):
+ """Parser HTML che estrae link a file dati."""
+
+ def __init__(self, base_url: str) -> None:
+ super().__init__()
+ self.base_url = base_url
+ self.links: list[DataLink] = []
+ self._seen: set[str] = set()
+
+ def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+ if tag.lower() not in ("a", "area"):
+ return
+ attrs_dict = dict(attrs)
+ href = attrs_dict.get("href", "") or attrs_dict.get("xlink:href", "")
+ if not href:
+ return
+ # Salta ancore, mailto, tel, javascript
+ if href.startswith(("#", "mailto:", "tel:", "javascript:")):
+ return
+
+ full_url = urljoin(self.base_url, href)
+ fmt = _resolve_format(full_url)
+ if fmt is None:
+ return
+
+ # Deduplica per URL
+ if full_url in self._seen:
+ return
+ self._seen.add(full_url)
+
+ title = (attrs_dict.get("aria-label") or attrs_dict.get("title") or "").strip()
+ filename = _extract_filename(full_url)
+ prefix = _extract_prefix(filename)
+ years = _extract_years(filename)
+
+ self.links.append(
+ DataLink(
+ url=full_url,
+ format=fmt,
+ title=title,
+ prefix=prefix,
+ years=years,
+ )
+ )
+
+
+# ---------------------------------------------------------------------------
+# Funzioni pubbliche
+# ---------------------------------------------------------------------------
+
+
+def extract_data_links(
+ base_url: str,
+ html: str,
+ *,
+ data_extensions: set[str] | None = None,
+) -> list[DataLink]:
+ """Estrae link a file dati da HTML.
+
+ Args:
+ base_url: URL di base per risolvere link relativi.
+ html: Testo HTML da parsare.
+ data_extensions: Estensioni da riconoscere (default: DATA_EXTENSIONS).
+
+ Returns:
+ Lista di DataLink, deduplicati per URL, ordinati per apparizione.
+ """
+ # Salva e ripristina le estensioni globali se parametro fornito
+ global DATA_EXTENSIONS
+ if data_extensions is not None:
+ _original_exts = DATA_EXTENSIONS
+ DATA_EXTENSIONS = data_extensions
+
+ try:
+ parser = _DataLinkParser(base_url)
+ parser.feed(html)
+ return parser.links
+ except Exception:
+ return []
+ finally:
+ if data_extensions is not None:
+ DATA_EXTENSIONS = _original_exts
+
+
+def group_links(links: list[DataLink]) -> list[LinkGroup]:
+ """Raggruppa link in dataset basandosi su prefisso e pattern URL.
+
+ Strategia:
+ 1. Raggruppa per ``prefix`` (quando presente).
+ 2. All'interno dello stesso prefix, se gli URL differiscono solo per
+ anno/data/versione, appartengono allo stesso dataset.
+ 3. Link senza prefix vanno in un gruppo ``other``.
+
+ Args:
+ links: Lista di DataLink da raggruppare.
+
+ Returns:
+ Lista di LinkGroup ordinata per dimensione discendente.
+ """
+ # Fase 1: raggruppa per prefisso
+ by_prefix: dict[str, list[DataLink]] = {}
+ for link in links:
+ p = link.prefix or "_other"
+ by_prefix.setdefault(p, []).append(link)
+
+ groups: list[LinkGroup] = []
+
+ for prefix, group_links in by_prefix.items():
+ years: set[int] = set()
+ formats: set[str] = set()
+ for link in group_links:
+ years.update(link.years)
+ formats.add(link.format)
+
+ group_id = prefix if prefix != "_other" else "other"
+ year_list = sorted(years)
+ title = group_links[0].title or prefix
+
+ groups.append(
+ LinkGroup(
+ group_id=group_id,
+ title=title,
+ prefix=prefix if prefix != "_other" else "",
+ year_range=year_list,
+ formats=formats,
+ links=group_links,
+ count=len(group_links),
+ )
+ )
+
+ # Ordina per count decrescente
+ groups.sort(key=lambda g: g.count, reverse=True)
+ return groups
+
+
+# ---------------------------------------------------------------------------
+# Backward compatibility: bridge da extract_candidate_links a DataLink
+# ---------------------------------------------------------------------------
+
+
+def extract_candidate_links(base_url: str, html_text: str) -> list[str]:
+ """Versione legacy: ritorna solo URL (nessun metadato).
+
+ Mantenuta per backward compatibility con codice che importa
+ ``extract_candidate_links`` da ``toolkit.scout.http``.
+ """
+ links = extract_data_links(base_url, html_text)
+ return [link.url for link in links]
From 653bc4b29ddb526d5a07fc696e3ca0563361cf8d Mon Sep 17 00:00:00 2001
From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com>
Date: Thu, 25 Jun 2026 18:37:10 +0100
Subject: [PATCH 2/6] =?UTF-8?q?feat:=20probe=5Fhtml=5Fportal=20=E2=80=94?=
=?UTF-8?q?=20probe=20leggero=20su=20portali=20HTML?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Aggiunge probe a costo fisso per scoprire struttura di un portale HTML:
- GET /robots.txt estrae Sitemap: URLs
- Prova path canonici (/sitemap.xml, /sitemap_index.xml)
- Scarica e parse sitemap lista pagine
- Cerca feed RSS e pattern JSON:API nella homepage
- Conta link interni
Risultati su portali reali:
- opencivitas: 575 pagine dalla sitemap
- giustizia_statistiche: 123 pagine dalla sitemap
- aifa, dait: robots.txt trovato ma nessuna sitemap
File: probe.py (+probe_html_portal, +PortalProfile), __init__.py
---
toolkit/scout/__init__.py | 2 +
toolkit/scout/link_extractor.py | 4 +-
toolkit/scout/probe.py | 204 +++++++++++++++++++++++++++++++-
3 files changed, 207 insertions(+), 3 deletions(-)
diff --git a/toolkit/scout/__init__.py b/toolkit/scout/__init__.py
index 80eb07ba..bbbd5e8a 100644
--- a/toolkit/scout/__init__.py
+++ b/toolkit/scout/__init__.py
@@ -58,6 +58,8 @@
)
from toolkit.scout.probe import ( # noqa: F401
+ PortalProfile,
+ probe_html_portal,
probe_url,
probe_url_routed,
)
diff --git a/toolkit/scout/link_extractor.py b/toolkit/scout/link_extractor.py
index f3c95cce..4d50c8f1 100644
--- a/toolkit/scout/link_extractor.py
+++ b/toolkit/scout/link_extractor.py
@@ -17,7 +17,7 @@
import re
from dataclasses import dataclass, field
from html.parser import HTMLParser
-from urllib.parse import urljoin
+from urllib.parse import urljoin, urlparse
# ---------------------------------------------------------------------------
# Costanti
@@ -97,7 +97,7 @@ class LinkGroup:
def _extract_filename(url: str) -> str:
"""Estrae il filename (senza estensione) da un URL."""
- from urllib.parse import unquote, urlparse
+ from urllib.parse import unquote
path = urlparse(url).path
# Rimuovi query string
diff --git a/toolkit/scout/probe.py b/toolkit/scout/probe.py
index 0aeb3792..6d52ad14 100644
--- a/toolkit/scout/probe.py
+++ b/toolkit/scout/probe.py
@@ -12,9 +12,13 @@
from __future__ import annotations
+import re
+import xml.etree.ElementTree as ET
+from dataclasses import dataclass, field
from typing import Any
-from urllib.parse import urlparse
+from urllib.parse import urljoin, urlparse
+from lab_connectors.http import HttpClient
from toolkit.scout.http import (
DEFAULT_TIMEOUT,
DEFAULT_USER_AGENT,
@@ -388,3 +392,201 @@ def _route_sparql(final_url: str, result: dict[str, Any], *, timeout: int) -> di
result["candidate_links"] = []
result["sdmx_info"] = None
return result
+
+
+# ---------------------------------------------------------------------------
+# Portal profile — probe leggero su portali HTML
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class PortalProfile:
+ """Profilo leggero di un portale HTML, ottenuto via probe senza crawl.
+
+ Attributes:
+ base_url: URL base del portale.
+ has_robots_txt: ``/robots.txt`` raggiungibile.
+ sitemap_urls: URLs delle sitemap trovate in robots.txt.
+ sitemap_pages: URLs delle pagine estratte dalla sitemap.
+ rss_feeds: Feed RSS trovati nell'HTML della homepage.
+ has_json_api: Pattern JSON:API rilevato (es. Drupal).
+ homepage_links: Numero di link interni nella homepage.
+ """
+
+ base_url: str
+ has_robots_txt: bool = False
+ sitemap_urls: list[str] = field(default_factory=list)
+ sitemap_pages: list[str] = field(default_factory=list)
+ rss_feeds: list[dict[str, str]] = field(default_factory=list)
+ has_json_api: bool = False
+ homepage_links: int = 0
+
+
+def _fetch_robots_sitemaps(base_url: str, *, timeout: int = 10) -> tuple[bool, list[str]]:
+ """Probe ``/robots.txt`` e estrae direttive ``Sitemap:``.
+
+ Returns:
+ (raggiungibile, lista URL sitemap)
+ """
+ robots_url = urljoin(base_url + "/", "robots.txt")
+ client = HttpClient(timeout=timeout)
+ result = client.get(robots_url)
+ if not result.is_ok or result.response is None or result.response.status_code >= 400:
+ return False, []
+
+ sitemaps: list[str] = []
+ for line in result.response.text.splitlines():
+ line = line.strip()
+ if line.lower().startswith("sitemap:"):
+ url = line.split(":", 1)[1].strip()
+ if url:
+ sitemaps.append(url)
+ return True, sitemaps
+
+
+def _fetch_and_parse_sitemap(sitemap_url: str, *, timeout: int = 10) -> list[str]:
+ """Fetch e parse una sitemap XML, ritorna lista di URL delle pagine."""
+ client = HttpClient(timeout=timeout)
+ result = client.get(sitemap_url)
+ if not result.is_ok or result.response is None or result.response.status_code >= 400:
+ return []
+ try:
+ root = ET.fromstring(result.response.text)
+ # Namespace sitemap standard
+ ns = {"sm": "http://www.sitemaps.org/schemas/sitemap/0.9"}
+ urls: list[str] = []
+ for url_elem in root.findall(".//sm:url", ns):
+ loc = url_elem.find("sm:loc", ns)
+ if loc is not None and loc.text:
+ urls.append(loc.text.strip())
+ # Fallback: prova senza namespace
+ if not urls:
+ for url_elem in root.findall(".//url"):
+ loc = url_elem.find("loc")
+ if loc is not None and loc.text:
+ urls.append(loc.text.strip())
+ return urls
+ except Exception:
+ return []
+
+
+def _scan_html_for_rss(html: str) -> list[dict[str, str]]:
+ """Cerca link a feed RSS/Atom nell'HTML della homepage."""
+ feeds: list[dict[str, str]] = []
+ # Pattern:
+ pattern = re.compile(
+ r']*?rel=["\']alternate["\'][^>]*?'
+ r'type=["\']application/(?:rss|atom)\+xml["\'][^>]*?'
+ r'href=["\']([^"\']+)["\'][^>]*?'
+ r'(?:title=["\']([^"\']*)["\'])?',
+ re.IGNORECASE | re.DOTALL,
+ )
+ for match in pattern.finditer(html):
+ feeds.append({"url": match.group(1), "title": match.group(2) or ""})
+ return feeds
+
+
+def _scan_html_for_jsonapi(html: str, base_url: str) -> bool:
+ """Rileva pattern JSON:API (Drupal) nell'HTML o nell'URL."""
+ # Drupal 9+ espone link nel
+ if "jsonapi" in html.lower():
+ return True
+ # Verifica /jsonapi/ nell'URL base
+ parsed = urlparse(base_url)
+ test_url = f"{parsed.scheme}://{parsed.netloc}/jsonapi"
+ client = HttpClient(timeout=5)
+ result = client.get(test_url)
+ if result.is_ok and result.response and result.response.status_code < 400:
+ content_type = (result.response.headers.get("Content-Type") or "").lower()
+ if "json" in content_type or "api" in content_type:
+ return True
+ return False
+
+
+def probe_html_portal(
+ base_url: str,
+ *,
+ timeout: int = 10,
+ fetch_sitemap: bool = True,
+ fetch_homepage: bool = True,
+) -> PortalProfile:
+ """Probe leggero su un portale HTML per scoprire struttura e pagine.
+
+ Esegue una serie di probe a costo fisso (nessun crawl):
+ 1. ``/robots.txt`` → estrae URL delle sitemap
+ 2. Sitemap XML → lista pagine del portale
+ 3. Homepage → cerca feed RSS e pattern JSON:API (Drupal)
+
+ Args:
+ base_url: URL base del portale (es. ``https://dati.istruzione.it``).
+ timeout: Timeout HTTP per ogni chiamata.
+ fetch_sitemap: Se True, scarica e parse le sitemap trovate.
+ fetch_homepage: Se True, scarica la homepage per RSS/API detection.
+
+ Returns:
+ PortalProfile con tutto ciò che è stato scoperto.
+ """
+ profile = PortalProfile(base_url=base_url)
+
+ # 1. robots.txt → sitemap
+ has_robots, sitemap_urls = _fetch_robots_sitemaps(base_url, timeout=timeout)
+ profile.has_robots_txt = has_robots
+
+ # 2. Prova anche sitemap nei path canonici (anche se robots.txt non le elenca)
+ _common_sitemap_paths = ["/sitemap.xml", "/sitemap_index.xml"]
+ for sm_path in _common_sitemap_paths:
+ sm_url = urljoin(base_url.rstrip("/") + "/", sm_path.lstrip("/"))
+ if sm_url not in sitemap_urls:
+ pages = _fetch_and_parse_sitemap(sm_url, timeout=timeout)
+ if pages:
+ sitemap_urls.append(sm_url)
+ profile.sitemap_urls = sitemap_urls
+ profile.sitemap_pages.extend(pages)
+
+ # 3. Sitemap da robots.txt → page URLs
+ if fetch_sitemap and sitemap_urls:
+ for sm_url in sitemap_urls[:3]: # max 3 sitemap
+ if sm_url not in _common_sitemap_paths or not profile.sitemap_pages:
+ pages = _fetch_and_parse_sitemap(sm_url, timeout=timeout)
+ for p in pages:
+ if p not in profile.sitemap_pages:
+ profile.sitemap_pages.append(p)
+
+ # 3. Homepage → RSS + JSON:API + link interni
+ if fetch_homepage:
+ try:
+ body = fetch_html_body(base_url, timeout=timeout)
+ html = body.get("html_text", "")
+ profile.rss_feeds = _scan_html_for_rss(html)
+ profile.has_json_api = _scan_html_for_jsonapi(html, base_url)
+
+ # Conta link interni (stesso dominio)
+ from html.parser import HTMLParser
+
+ class _InternalLinkCounter(HTMLParser):
+ def __init__(self):
+ super().__init__()
+ self.count = 0
+ self._domain = urlparse(base_url).netloc.lower()
+
+ def handle_starttag(self, tag, attrs):
+ if tag.lower() != "a":
+ return
+ for key, value in attrs:
+ if key.lower() == "href" and value:
+ href = value.strip()
+ if href and not href.startswith(
+ ("#", "mailto:", "tel:", "javascript:")
+ ):
+ full = urljoin(base_url, href)
+ if self._domain in urlparse(full).netloc.lower():
+ self.count += 1
+ return
+
+ parser = _InternalLinkCounter()
+ parser.feed(html)
+ profile.homepage_links = parser.count
+ except Exception:
+ pass
+
+ return profile
From edde83acae63ff8c2c0ea04c80a3157c8a72c2d2 Mon Sep 17 00:00:00 2001
From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com>
Date: Thu, 25 Jun 2026 18:42:56 +0100
Subject: [PATCH 3/6] feat: espone fetch_sitemap_pages come funzione pubblica
- _fetch_and_parse_sitemap rinominato a fetch_sitemap_pages pubblico
- Export in __init__.py
- Uso in probe_html_portal aggiornato
File: probe.py, __init__.py
---
toolkit/scout/__init__.py | 1 +
toolkit/scout/probe.py | 16 +++++++++++++---
2 files changed, 14 insertions(+), 3 deletions(-)
diff --git a/toolkit/scout/__init__.py b/toolkit/scout/__init__.py
index bbbd5e8a..ef70323e 100644
--- a/toolkit/scout/__init__.py
+++ b/toolkit/scout/__init__.py
@@ -59,6 +59,7 @@
from toolkit.scout.probe import ( # noqa: F401
PortalProfile,
+ fetch_sitemap_pages,
probe_html_portal,
probe_url,
probe_url_routed,
diff --git a/toolkit/scout/probe.py b/toolkit/scout/probe.py
index 6d52ad14..b362b809 100644
--- a/toolkit/scout/probe.py
+++ b/toolkit/scout/probe.py
@@ -444,7 +444,17 @@ def _fetch_robots_sitemaps(base_url: str, *, timeout: int = 10) -> tuple[bool, l
return True, sitemaps
-def _fetch_and_parse_sitemap(sitemap_url: str, *, timeout: int = 10) -> list[str]:
+def fetch_sitemap_pages(sitemap_url: str, *, timeout: int = 10) -> list[str]:
+ """Fetch e parse una sitemap XML, ritorna lista di URL delle pagine.
+
+ Args:
+ sitemap_url: URL della sitemap (es. ``https://example.gov.it/sitemap.xml``).
+ timeout: Timeout HTTP.
+
+ Returns:
+ Lista di URL delle pagine trovate nella sitemap.
+ Lista vuota se sitemap non raggiungibile o malformata.
+ """
"""Fetch e parse una sitemap XML, ritorna lista di URL delle pagine."""
client = HttpClient(timeout=timeout)
result = client.get(sitemap_url)
@@ -537,7 +547,7 @@ def probe_html_portal(
for sm_path in _common_sitemap_paths:
sm_url = urljoin(base_url.rstrip("/") + "/", sm_path.lstrip("/"))
if sm_url not in sitemap_urls:
- pages = _fetch_and_parse_sitemap(sm_url, timeout=timeout)
+ pages = fetch_sitemap_pages(sm_url, timeout=timeout)
if pages:
sitemap_urls.append(sm_url)
profile.sitemap_urls = sitemap_urls
@@ -547,7 +557,7 @@ def probe_html_portal(
if fetch_sitemap and sitemap_urls:
for sm_url in sitemap_urls[:3]: # max 3 sitemap
if sm_url not in _common_sitemap_paths or not profile.sitemap_pages:
- pages = _fetch_and_parse_sitemap(sm_url, timeout=timeout)
+ pages = fetch_sitemap_pages(sm_url, timeout=timeout)
for p in pages:
if p not in profile.sitemap_pages:
profile.sitemap_pages.append(p)
From b7a66f47da36d065e8f4b0db440dc96891197c66 Mon Sep 17 00:00:00 2001
From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com>
Date: Thu, 25 Jun 2026 18:46:06 +0100
Subject: [PATCH 4/6] fix: probe_html_portal cerca sitemap anche su domain root
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- Se base_url è un path (es. /opendata/...), prova anche la root del
dominio per robots.txt e sitemap
- Aggiunto /opendata/sitemap.xml ai path canonici
- Dedup degli URL trovati
---
toolkit/scout/probe.py | 51 +++++++++++++++++++++++++++++-------------
1 file changed, 36 insertions(+), 15 deletions(-)
diff --git a/toolkit/scout/probe.py b/toolkit/scout/probe.py
index b362b809..f023cfea 100644
--- a/toolkit/scout/probe.py
+++ b/toolkit/scout/probe.py
@@ -538,31 +538,52 @@ def probe_html_portal(
"""
profile = PortalProfile(base_url=base_url)
- # 1. robots.txt → sitemap
- has_robots, sitemap_urls = _fetch_robots_sitemaps(base_url, timeout=timeout)
- profile.has_robots_txt = has_robots
-
- # 2. Prova anche sitemap nei path canonici (anche se robots.txt non le elenca)
- _common_sitemap_paths = ["/sitemap.xml", "/sitemap_index.xml"]
- for sm_path in _common_sitemap_paths:
- sm_url = urljoin(base_url.rstrip("/") + "/", sm_path.lstrip("/"))
- if sm_url not in sitemap_urls:
+ # Determina la root del dominio (per probe sitemap/robots)
+ parsed = urlparse(base_url)
+ domain_root = f"{parsed.scheme}://{parsed.netloc}"
+ # Path verso cui fare probe: prima il base_url, poi la root del dominio
+ _probe_targets = list(dict.fromkeys([base_url.rstrip("/"), domain_root]))
+
+ # 1. robots.txt → sitemap (prova sia base_url che domain root)
+ sitemap_urls: list[str] = []
+ for target in _probe_targets:
+ has_robots, urls = _fetch_robots_sitemaps(target, timeout=timeout)
+ if has_robots:
+ profile.has_robots_txt = True
+ for u in urls:
+ if u not in sitemap_urls:
+ sitemap_urls.append(u)
+
+ # 2. Prova sitemap nei path canonici (per ogni target)
+ _common_sitemap_paths = [
+ "/sitemap.xml",
+ "/sitemap_index.xml",
+ "/opendata/sitemap.xml",
+ ]
+ for target in _probe_targets:
+ for sm_path in _common_sitemap_paths:
+ sm_url = urljoin(target + "/", sm_path.lstrip("/"))
+ if sm_url in sitemap_urls:
+ continue
pages = fetch_sitemap_pages(sm_url, timeout=timeout)
if pages:
sitemap_urls.append(sm_url)
- profile.sitemap_urls = sitemap_urls
profile.sitemap_pages.extend(pages)
- # 3. Sitemap da robots.txt → page URLs
- if fetch_sitemap and sitemap_urls:
- for sm_url in sitemap_urls[:3]: # max 3 sitemap
- if sm_url not in _common_sitemap_paths or not profile.sitemap_pages:
+ profile.sitemap_urls = list(dict.fromkeys(sitemap_urls)) # dedup
+
+ # 3. Fetch sitemap non ancora parse (quelle da robots.txt)
+ if fetch_sitemap:
+ for sm_url in sitemap_urls:
+ if sm_url not in _common_sitemap_paths:
pages = fetch_sitemap_pages(sm_url, timeout=timeout)
for p in pages:
if p not in profile.sitemap_pages:
profile.sitemap_pages.append(p)
- # 3. Homepage → RSS + JSON:API + link interni
+ profile.sitemap_pages = list(dict.fromkeys(profile.sitemap_pages)) # dedup
+
+ # 4. Homepage → RSS + JSON:API + link interni
if fetch_homepage:
try:
body = fetch_html_body(base_url, timeout=timeout)
From 7072771169f04d09f189d396bcd1baaf695ead98 Mon Sep 17 00:00:00 2001
From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com>
Date: Thu, 25 Jun 2026 18:56:29 +0100
Subject: [PATCH 5/6] fix: sitemap index support e MCP formats senza query
string
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- fetch_sitemap_pages ora gestisce sitemap index XML
() con fetch ricorsivo
- MCP formats calcolato dal path URL (urlparse) anziché URL grezzo
---
toolkit/mcp/scout_ops.py | 7 +++--
toolkit/scout/probe.py | 62 ++++++++++++++++++++++++++++++++--------
2 files changed, 55 insertions(+), 14 deletions(-)
diff --git a/toolkit/mcp/scout_ops.py b/toolkit/mcp/scout_ops.py
index 3715608d..bac4b661 100644
--- a/toolkit/mcp/scout_ops.py
+++ b/toolkit/mcp/scout_ops.py
@@ -181,10 +181,13 @@ def mcp_html_extract_links(url: str, timeout: int = 20) -> dict[str, Any]:
# Backward compat: lista di URL semplici
links = [dl.url for dl in data_links]
- # Backward compat: conteggio per estensione (lowercase)
+ # Backward compat: conteggio per estensione (lowercase, dal path senza query)
+ from urllib.parse import urlparse
+
formats: dict[str, int] = {}
for dl in data_links:
- ext = dl.url.rsplit(".", 1)[-1].lower() if "." in dl.url else "unknown"
+ path = urlparse(dl.url).path
+ ext = path.rsplit(".", 1)[-1].lower() if "." in path else "unknown"
formats[ext] = formats.get(ext, 0) + 1
return {
diff --git a/toolkit/scout/probe.py b/toolkit/scout/probe.py
index f023cfea..6f0b2af7 100644
--- a/toolkit/scout/probe.py
+++ b/toolkit/scout/probe.py
@@ -447,6 +447,10 @@ def _fetch_robots_sitemaps(base_url: str, *, timeout: int = 10) -> tuple[bool, l
def fetch_sitemap_pages(sitemap_url: str, *, timeout: int = 10) -> list[str]:
"""Fetch e parse una sitemap XML, ritorna lista di URL delle pagine.
+ Supporta sia sitemap standard (````) che sitemap index
+ (````). In caso di sitemap index, segue
+ ricorsivamente le sotto-sitemap (max 10).
+
Args:
sitemap_url: URL della sitemap (es. ``https://example.gov.it/sitemap.xml``).
timeout: Timeout HTTP.
@@ -455,27 +459,61 @@ def fetch_sitemap_pages(sitemap_url: str, *, timeout: int = 10) -> list[str]:
Lista di URL delle pagine trovate nella sitemap.
Lista vuota se sitemap non raggiungibile o malformata.
"""
- """Fetch e parse una sitemap XML, ritorna lista di URL delle pagine."""
+ return _fetch_sitemap_recursive(sitemap_url, timeout=timeout, depth=0)
+
+
+def _fetch_sitemap_recursive(
+ sitemap_url: str, *, timeout: int, depth: int, max_depth: int = 10
+) -> list[str]:
+ """Fetch e parse ricorsivo di sitemap, con protezione profondità."""
+ if depth >= max_depth:
+ return []
+
client = HttpClient(timeout=timeout)
result = client.get(sitemap_url)
if not result.is_ok or result.response is None or result.response.status_code >= 400:
return []
+
try:
root = ET.fromstring(result.response.text)
- # Namespace sitemap standard
ns = {"sm": "http://www.sitemaps.org/schemas/sitemap/0.9"}
- urls: list[str] = []
- for url_elem in root.findall(".//sm:url", ns):
- loc = url_elem.find("sm:loc", ns)
- if loc is not None and loc.text:
- urls.append(loc.text.strip())
- # Fallback: prova senza namespace
- if not urls:
- for url_elem in root.findall(".//url"):
- loc = url_elem.find("loc")
+
+ # Rileva se è sitemap index ( root)
+ is_index = root.tag.endswith("sitemapindex")
+
+ if is_index:
+ # Sitemap index: contiene ...
+ sub_sitemaps: list[str] = []
+ for sm_elem in root.findall(".//sm:sitemap", ns):
+ loc = sm_elem.find("sm:loc", ns)
+ if loc is not None and loc.text:
+ sub_sitemaps.append(loc.text.strip())
+ # Fallback senza namespace
+ if not sub_sitemaps:
+ for sm_elem in root.findall(".//sitemap"):
+ loc = sm_elem.find("loc")
+ if loc is not None and loc.text:
+ sub_sitemaps.append(loc.text.strip())
+
+ # Segue ricorsivamente ogni sotto-sitemap
+ all_urls: list[str] = []
+ for sub_url in sub_sitemaps[:50]: # max 50 sotto-sitemap
+ all_urls.extend(_fetch_sitemap_recursive(sub_url, timeout=timeout, depth=depth + 1))
+ return all_urls
+ else:
+ # Sitemap standard: contiene ...
+ urls: list[str] = []
+ for url_elem in root.findall(".//sm:url", ns):
+ loc = url_elem.find("sm:loc", ns)
if loc is not None and loc.text:
urls.append(loc.text.strip())
- return urls
+ # Fallback senza namespace
+ if not urls:
+ for url_elem in root.findall(".//url"):
+ loc = url_elem.find("loc")
+ if loc is not None and loc.text:
+ urls.append(loc.text.strip())
+ return urls
except Exception:
return []
From 002274a70ee54fdbf8522ad9c4bd0edcaa21de90 Mon Sep 17 00:00:00 2001
From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com>
Date: Thu, 25 Jun 2026 19:11:41 +0100
Subject: [PATCH 6/6] test: copertura per link_extractor, sitemap index, MCP
formats (14 test)
---
tests/test_link_extractor.py | 126 +++++++++++++++++++++++++++++++++++
tests/test_probe_sitemap.py | 126 +++++++++++++++++++++++++++++++++++
2 files changed, 252 insertions(+)
create mode 100644 tests/test_link_extractor.py
create mode 100644 tests/test_probe_sitemap.py
diff --git a/tests/test_link_extractor.py b/tests/test_link_extractor.py
new file mode 100644
index 00000000..e8d7197f
--- /dev/null
+++ b/tests/test_link_extractor.py
@@ -0,0 +1,126 @@
+"""Tests per link_extractor — contratto pubblico: extract_data_links, group_links.
+
+pure_unit: extract_data_links, group_links
+"""
+
+import pytest
+from toolkit.scout.link_extractor import (
+ DataLink,
+ extract_data_links,
+ group_links,
+)
+
+
+class TestExtractDataLinks:
+ """pure_unit: estrazione link dati da HTML."""
+
+ @pytest.mark.pure_unit
+ def test_basic_formats(self):
+ """Estrae CSV, XLSX, JSON, ZIP, ignora PDF e HTML."""
+ html = """
+ CSV
+ XLSX
+ JSON
+ ZIP
+ HTML
+ PDF
+ """
+ links = extract_data_links("https://ex.it", html)
+ assert len(links) == 4
+ assert {lnk.format for lnk in links} == {"CSV", "XLSX", "JSON", "ZIP"}
+
+ @pytest.mark.pure_unit
+ def test_skips_anchors_and_protocols(self):
+ """Ignora ancore, mailto, tel, javascript."""
+ html = """
+ X
+ X
+ X
+ X
+ """
+ assert extract_data_links("https://ex.it", html) == []
+
+ @pytest.mark.pure_unit
+ def test_resolves_relative_and_title(self):
+ """URL relativi risolti, aria-label come title."""
+ links = extract_data_links(
+ "https://ex.it/dir/",
+ 'x',
+ )
+ assert links[0].url == "https://ex.it/dir/data.csv"
+ assert links[0].title == "Report"
+
+ @pytest.mark.pure_unit
+ def test_prefix_and_years(self):
+ """Prefisso e anni estratti dal filename."""
+ links = extract_data_links(
+ "https://ex.it",
+ 'x',
+ )
+ assert links[0].prefix == "REG"
+ assert links[0].years == [2025]
+
+
+class TestGroupLinks:
+ """pure_unit: raggruppamento link per prefisso."""
+
+ @pytest.mark.pure_unit
+ def test_single_group(self):
+ """Stesso prefisso → un gruppo con anni e formati."""
+ links = [
+ DataLink(url="https://ex.it/REG_2024.csv", format="CSV", prefix="REG", years=[2024]),
+ DataLink(url="https://ex.it/REG_2025.csv", format="CSV", prefix="REG", years=[2025]),
+ ]
+ groups = group_links(links)
+ assert len(groups) == 1
+ assert groups[0].group_id == "REG"
+ assert groups[0].count == 2
+ assert groups[0].year_range == [2024, 2025]
+
+ @pytest.mark.pure_unit
+ def test_multiple_groups(self):
+ """Prefissi diversi → gruppi separati."""
+ links = [
+ DataLink(url="https://ex.it/REG_2024.csv", format="CSV", prefix="REG"),
+ DataLink(url="https://ex.it/CLA_2024.csv", format="CSV", prefix="CLA"),
+ ]
+ groups = group_links(links)
+ assert len(groups) == 2
+ assert {g.group_id for g in groups} == {"REG", "CLA"}
+
+ @pytest.mark.pure_unit
+ def test_no_prefix_becomes_other(self):
+ """Link senza prefisso → gruppo other."""
+ groups = group_links(
+ [
+ DataLink(url="https://ex.it/data.csv", format="CSV"),
+ ]
+ )
+ assert groups[0].group_id == "other"
+
+ @pytest.mark.pure_unit
+ def test_empty(self):
+ """Lista vuota → lista vuota."""
+ assert group_links([]) == []
+
+
+class TestMcpFormatsQueryString:
+ """pure_unit: MCP formats legacy usa path, non URL grezzo."""
+
+ @pytest.mark.pure_unit
+ def test_formats_from_path(self, monkeypatch):
+ """URL con query string → formato senza query."""
+ from toolkit.mcp.scout_ops import mcp_html_extract_links
+
+ html = """
+ CSV
+ ZIP
+ """
+
+ def mock_fetch(url, **kw):
+ return {"html_text": html, "status_code": 200}
+
+ monkeypatch.setattr("toolkit.mcp.scout_ops.fetch_html_body", mock_fetch)
+ result = mcp_html_extract_links("https://ex.it/")
+ assert result["formats"] == {"csv": 1, "zip": 1}
+ assert result["data_links"][0]["format"] == "CSV"
diff --git a/tests/test_probe_sitemap.py b/tests/test_probe_sitemap.py
new file mode 100644
index 00000000..81c9368c
--- /dev/null
+++ b/tests/test_probe_sitemap.py
@@ -0,0 +1,126 @@
+"""Tests per probe — fetch_sitemap_pages con sitemap index.
+
+pure_unit: fetch_sitemap_pages parsing XML (mock HTTP)
+"""
+
+from unittest.mock import patch, MagicMock
+
+import pytest
+from toolkit.scout.probe import fetch_sitemap_pages
+
+
+# ---------------------------------------------------------------------------
+# pure_unit — fetch_sitemap_pages
+# ---------------------------------------------------------------------------
+
+
+class TestFetchSitemapPages:
+ """pure_unit: parsing sitemap XML."""
+
+ @pytest.mark.pure_unit
+ def test_standard_sitemap(self):
+ """Sitemap standard ."""
+ xml = """
+
+ https://ex.it/page1
+ https://ex.it/page2
+"""
+ pages = self._mock_fetch(xml)
+ assert pages == ["https://ex.it/page1", "https://ex.it/page2"]
+
+ @pytest.mark.pure_unit
+ def test_sitemap_index(self):
+ """Sitemap index con fetch ricorsivo."""
+ index_xml = """
+
+ https://ex.it/sub1.xml
+ https://ex.it/sub2.xml
+"""
+ sub1_xml = """
+
+ https://ex.it/page1
+"""
+ sub2_xml = """
+
+ https://ex.it/page2
+ https://ex.it/page3
+"""
+
+ pages = self._mock_fetch_index(
+ index_xml,
+ {
+ "https://ex.it/sub1.xml": sub1_xml,
+ "https://ex.it/sub2.xml": sub2_xml,
+ },
+ )
+ assert pages == ["https://ex.it/page1", "https://ex.it/page2", "https://ex.it/page3"]
+
+ @pytest.mark.pure_unit
+ def test_sitemap_index_no_namespace(self):
+ """Sitemap index senza namespace XML."""
+ xml = """
+
+ https://ex.it/sub.xml
+"""
+ sub_xml = """
+
+ https://ex.it/page
+"""
+ pages = self._mock_fetch_index(xml, {"https://ex.it/sub.xml": sub_xml})
+ assert pages == ["https://ex.it/page"]
+
+ @pytest.mark.pure_unit
+ def test_unreachable_sitemap(self):
+ """Sitemap non raggiungibile → lista vuota."""
+ with patch("toolkit.scout.probe.HttpClient") as MockClient:
+ instance = MockClient.return_value
+ instance.get.return_value = MagicMock(is_ok=False)
+ pages = fetch_sitemap_pages("https://ex.it/sitemap.xml", timeout=5)
+ assert pages == []
+
+ @pytest.mark.pure_unit
+ def test_malformed_xml(self):
+ """XML malformato → lista vuota."""
+ with patch("toolkit.scout.probe.HttpClient") as MockClient:
+ instance = MockClient.return_value
+ mock_resp = MagicMock()
+ mock_resp.status_code = 200
+ mock_resp.text = "not xml"
+ instance.get.return_value = MagicMock(
+ is_ok=True, response=mock_resp, err=None, ssl_fallback_used=False
+ )
+ pages = fetch_sitemap_pages("https://ex.it/sitemap.xml", timeout=5)
+ assert pages == []
+
+ # -- helpers --
+
+ def _mock_fetch(self, xml: str) -> list[str]:
+ """Mock HttpClient per sitemap standard."""
+ with patch("toolkit.scout.probe.HttpClient") as MockClient:
+ instance = MockClient.return_value
+ mock_resp = MagicMock()
+ mock_resp.status_code = 200
+ mock_resp.text = xml
+ instance.get.return_value = MagicMock(
+ is_ok=True, response=mock_resp, err=None, ssl_fallback_used=False
+ )
+ return fetch_sitemap_pages("https://ex.it/sitemap.xml", timeout=5)
+
+ def _mock_fetch_index(self, index_xml: str, subs: dict[str, str]) -> list[str]:
+ """Mock HttpClient per sitemap index con sotto-sitemap."""
+
+ def mock_get(url):
+ mock_resp = MagicMock()
+ mock_resp.status_code = 200
+ if url == "https://ex.it/sitemap.xml":
+ mock_resp.text = index_xml
+ elif url in subs:
+ mock_resp.text = subs[url]
+ else:
+ mock_resp.text = ""
+ return MagicMock(is_ok=True, response=mock_resp, err=None, ssl_fallback_used=False)
+
+ with patch("toolkit.scout.probe.HttpClient") as MockClient:
+ instance = MockClient.return_value
+ instance.get.side_effect = mock_get
+ return fetch_sitemap_pages("https://ex.it/sitemap.xml", timeout=5)