diff --git a/pyproject.toml b/pyproject.toml index 60f26a92..5edb5bd4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,7 +31,7 @@ dependencies = [ "pyarrow>=24.0.0", "jsonschema>=4.26.0", "ddgs>=9.14.4", - "dataciviclab-toolkit @ git+https://github.com/dataciviclab/toolkit.git", + "lab-connectors[duckdb,gcs,mcp] @ git+https://github.com/dataciviclab/lab-connectors.git", ] [project.optional-dependencies] diff --git a/scripts/catalog/build.py b/scripts/catalog/build.py index c48b4a1d..e2772645 100644 --- a/scripts/catalog/build.py +++ b/scripts/catalog/build.py @@ -177,7 +177,12 @@ def build_catalog( else: print(" → validate") merged = run_merge(inventory_df) - validated_rows_raw = run_validate(merged, max_groups=max_groups, workers=workers) + validated_rows_raw = run_validate( + merged, + max_groups=max_groups, + workers=workers, + user_agent=cfg.get("user_agent"), + ) validated_rows = [r for r in validated_rows_raw if r is not None] if validated_rows: pd.DataFrame(validated_rows).to_parquet(layout.validated_parquet, index=False) diff --git a/scripts/catalog/schemas/catalog.schema.json b/scripts/catalog/schemas/catalog.schema.json index ee0aadef..0c5bbcda 100644 --- a/scripts/catalog/schemas/catalog.schema.json +++ b/scripts/catalog/schemas/catalog.schema.json @@ -151,7 +151,7 @@ "properties": { "kind": { "type": "string", - "enum": ["dataflow", "ckan_resource", "url", "script_type"] + "enum": ["dataflow", "ckan_resource", "ckan_dataset", "url", "script_type"] }, "id": { "type": "string" }, "agency": { "type": "string" } diff --git a/scripts/catalog/upstream.py b/scripts/catalog/upstream.py index a5a2b12c..4311ac91 100644 --- a/scripts/catalog/upstream.py +++ b/scripts/catalog/upstream.py @@ -34,8 +34,11 @@ def extract_upstream_from_source(source: dict[str, Any]) -> list[dict[str, str]] if args.get("agency"): ref["agency"] = str(args["agency"]) refs.append(ref) - elif stype == "ckan" and args.get("resource_id"): - refs.append({"kind": "ckan_resource", "id": str(args["resource_id"])}) + elif stype == "ckan": + if args.get("resource_id"): + refs.append({"kind": "ckan_resource", "id": str(args["resource_id"])}) + elif args.get("dataset_id"): + refs.append({"kind": "ckan_dataset", "id": str(args["dataset_id"])}) elif stype in ("http_file", "http_post_file") and args.get("url"): refs.append({"kind": "url", "id": str(args["url"])}) elif stype == "script": diff --git a/scripts/collectors/_validate_base.py b/scripts/collectors/_validate_base.py index 1ddc94d0..311e7d5b 100644 --- a/scripts/collectors/_validate_base.py +++ b/scripts/collectors/_validate_base.py @@ -138,19 +138,19 @@ def _extract_year_range( """Estrae range anni (min, max) da un CSV profilato. Ordine di tentativo: - 1. anni nei NOMI colonna (es. ``anno_2020``) — toolkit._extract_years_from_columns + 1. anni nei NOMI colonna (es. ``anno_2020``) — infer_years_from_columns 2. valori in una colonna-anno (es. colonna ``ANNO`` con valori 2018..2024): il sample dello sniff standard contiene stringhe, quindi serve DuckDB sul raw per avere valori tipizzati e range su tutto il sample - 3. anni nel filename (es. ``beneficiari_2007-2013.zip``) — toolkit._extract_years + 3. anni nel filename (es. ``beneficiari_2007-2013.zip``) — extract_years Ritorna (None, None) se non trova nulla. """ - from toolkit.profile.preview import _extract_years_from_columns - from toolkit.scout.link_extractor import _extract_years as _filename_years + from lab_connectors.discovery.infer import infer_years_from_columns + from lab_connectors.discovery.links import extract_years as _filename_years # 1. anni da nomi colonna - y_cols = _extract_years_from_columns(columns) + y_cols = infer_years_from_columns(columns) if y_cols != (None, None): return y_cols diff --git a/scripts/collectors/ckan.py b/scripts/collectors/ckan.py index 2d46140a..26feaec8 100644 --- a/scripts/collectors/ckan.py +++ b/scripts/collectors/ckan.py @@ -288,7 +288,8 @@ def collect_ckan_inventory_via_search( from lab_connectors.http import HttpClient if client is None: - client = HttpClient(timeout=60, user_agent=USER_AGENT) + ua = source_cfg.get("user_agent") or USER_AGENT + client = HttpClient(timeout=60, user_agent=ua) endpoint = ckan_action_endpoint(source_cfg["base_url"], "package_search") page_size = 1000 @@ -386,7 +387,8 @@ def collect_ckan_inventory_via_package_list( from lab_connectors.http import HttpClient if client is None: - client = HttpClient(timeout=60, user_agent=USER_AGENT) + ua = source_cfg.get("user_agent") or USER_AGENT + client = HttpClient(timeout=60, user_agent=ua) endpoint = ckan_action_endpoint(source_cfg["base_url"], "package_list") payload = ckan_get_json(endpoint, client=client) if not payload.get("success"): @@ -545,7 +547,8 @@ def collect( from lab_connectors.http import HttpClient if client is None: - client = HttpClient(timeout=60, user_agent=USER_AGENT) + ua = source_cfg.get("user_agent") or USER_AGENT + client = HttpClient(timeout=60, user_agent=ua) inv = inventory_cfg(source_cfg) rows, warning = _ckan_standard_path( source_id=source_id, diff --git a/scripts/collectors/html.py b/scripts/collectors/html.py index 4ed8b8c8..a2bd8b17 100644 --- a/scripts/collectors/html.py +++ b/scripts/collectors/html.py @@ -12,7 +12,7 @@ Non fa full crawl (191 pagine). Campiona per stimare. -Il motore di estrazione link e raggruppamento è in ``toolkit.scout.link_extractor``. +Il motore di estrazione link è in ``lab_connectors.discovery.links``. Questo modulo è solo orchestratore: fetch pagine, costruisce righe SO, calcola summary. Uso: @@ -28,10 +28,12 @@ from collections import Counter from typing import Any +from lab_connectors.discovery.classify import resolve_preview_kind +from lab_connectors.discovery.links import DataLink, extract_data_links, group_links from lab_connectors.http import HttpClient -from toolkit.scout.http import probe_url_headers, resolve_preview_kind -from toolkit.scout.link_extractor import DataLink, extract_data_links, group_links -from toolkit.scout.probe import fetch_sitemap_pages, probe_html_portal +from lab_connectors.http.probe import probe_url_headers + +from scripts.portal import fetch_sitemap_pages, probe_html_portal from .base import CollectorResult @@ -64,26 +66,25 @@ def _extract_page_meta(html: str) -> dict[str, str]: def _extract_prefix(filename: str) -> str: - """Delega a toolkit.scout.link_extractor.""" - from toolkit.scout.link_extractor import _extract_prefix as _tk_prefix + """Delega a lab_connectors.discovery.links.""" + from lab_connectors.discovery.links import extract_prefix - return _tk_prefix(filename) + return extract_prefix(filename) def _extract_years(filename: str) -> list[int]: - """Delega a toolkit.scout.link_extractor.""" - from toolkit.scout.link_extractor import _extract_years as _tk_years + """Delega a lab_connectors.discovery.links.""" + from lab_connectors.discovery.links import extract_years - return _tk_years(filename) + return extract_years(filename) def _guess_topic(url: str, topic_hint: str | None) -> str: return topic_hint or "unknown" -# ─── Sitemap Helper (delega a toolkit) ──────────────────────────────────────── -# La funzione fetch_sitemap_pages() è in toolkit.scout.probe — la usiamo -# direttamente invece di duplicare la logica XML qui. +# ─── Sitemap Helper ─────────────────────────────────────────────────────────── +# fetch_sitemap_pages() è in scripts.portal (SO-specifico). # ─── Shared row building & stats ───────────────────────────────────────────── diff --git a/scripts/pipeline/run_pipeline.py b/scripts/pipeline/run_pipeline.py index 35184460..1b3adeb1 100644 --- a/scripts/pipeline/run_pipeline.py +++ b/scripts/pipeline/run_pipeline.py @@ -96,7 +96,10 @@ def _validate_one(group_df: pd.DataFrame, client=None) -> dict: def run_validate( - df: pd.DataFrame, max_groups: int | None = None, workers: int = 1 + df: pd.DataFrame, + max_groups: int | None = None, + workers: int = 1, + user_agent: str | None = None, ) -> list[dict | None]: """For each dataset_group, pick best URL, probe, sniff schema. @@ -124,7 +127,7 @@ def run_validate( # farebbe bruciare ~114s per fallimento prima che il breaker apra. from lab_connectors.http import HttpClient - http_client = HttpClient(timeout=5, circuit_threshold=3) + http_client = HttpClient(timeout=5, circuit_threshold=3, user_agent=user_agent) if workers <= 1: # Sequenziale diff --git a/scripts/portal.py b/scripts/portal.py new file mode 100644 index 00000000..e16409d9 --- /dev/null +++ b/scripts/portal.py @@ -0,0 +1,265 @@ +"""Portal profiling — probe leggero su portali HTML. + +Tutto SO-specifico: usato solo dal collector HTML in modalità +discovery=auto (1 fonte: opencivitas). Nessun altro consumer nel Lab. +""" + +from __future__ import annotations + +import re +import xml.etree.ElementTree as ET +from dataclasses import dataclass, field +from html.parser import HTMLParser +from urllib.parse import urljoin, urlparse + +from lab_connectors.http import HttpClient + + +@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_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 depth, max 50 sub-sitemaps). + + 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. + """ + 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) + ns = {"sm": "http://www.sitemaps.org/schemas/sitemap/0.9"} + + is_index = root.tag.endswith("sitemapindex") + + if is_index: + 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()) + 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()) + + all_urls: list[str] = [] + for sub_url in sub_sitemaps[:50]: + all_urls.extend(_fetch_sitemap_recursive(sub_url, timeout=timeout, depth=depth + 1)) + return all_urls + else: + 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()) + 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 = 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, *, timeout: int = 5) -> bool: + """Rileva pattern JSON:API (Drupal) nell'HTML o nell'URL.""" + if "jsonapi" in html.lower(): + return True + parsed = urlparse(base_url) + test_url = f"{parsed.scheme}://{parsed.netloc}/jsonapi" + client = HttpClient(timeout=timeout) + 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 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. + 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) + + parsed = urlparse(base_url) + domain_root = f"{parsed.scheme}://{parsed.netloc}" + _probe_targets = list(dict.fromkeys([base_url.rstrip("/"), domain_root])) + + # 1. robots.txt → sitemap + 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. Sitemap nei path canonici + _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_pages.extend(pages) + + profile.sitemap_urls = list(dict.fromkeys(sitemap_urls)) + + # 3. Fetch sitemap da robots.txt non ancora parse + 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) + + profile.sitemap_pages = list(dict.fromkeys(profile.sitemap_pages)) + + # 4. Homepage → RSS + JSON:API + link interni + if fetch_homepage: + try: + client = HttpClient(timeout=timeout) + result = client.get(base_url) + if result.is_ok and result.response is not None: + html = result.response.text + profile.rss_feeds = _scan_html_for_rss(html) + profile.has_json_api = _scan_html_for_jsonapi(html, base_url) + + class _InternalLinkCounter(HTMLParser): + def __init__(self): + super().__init__() + self.count = 0 + self._domain = parsed.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 diff --git a/scripts/radar_check.py b/scripts/radar_check.py index fbeff703..1c02f24e 100644 --- a/scripts/radar_check.py +++ b/scripts/radar_check.py @@ -9,9 +9,9 @@ from typing import Any, Literal import requests +from lab_connectors.discovery.classify import is_sdmx_url from lab_connectors.http import HttpClient, HttpFallbackError from lab_connectors.http.types import ResponseLike -from toolkit.scout.http import is_sdmx_url from scripts._constants import ( RADAR_HISTORY_PATH, diff --git a/tests/test_catalog_build.py b/tests/test_catalog_build.py index 1728311d..34bdab92 100644 --- a/tests/test_catalog_build.py +++ b/tests/test_catalog_build.py @@ -122,7 +122,7 @@ def fake_merge(df, dry_run: bool = False): df["dataset_group"] = df["item_name"].map(lambda n: f"demo_src/{n}") return df - def fake_validate(df, max_groups=None, workers: int = 1): + def fake_validate(df, max_groups=None, workers: int = 1, user_agent=None): return [dict(r) for r in _VALIDATED] monkeypatch.setattr("scripts.catalog.build.probe_url", fake_probe) diff --git a/tests/test_catalog_upstream.py b/tests/test_catalog_upstream.py index 1fd28519..6375bde4 100644 --- a/tests/test_catalog_upstream.py +++ b/tests/test_catalog_upstream.py @@ -120,3 +120,10 @@ def test_scans_dataset_dirs_by_name(self, tmp_path: Path): @pytest.mark.pure_unit def test_missing_dirs_ok(self, tmp_path: Path): assert load_upstream_map(tmp_path) == {} + + @pytest.mark.pure_unit + def test_ckan_dataset_id(self): + refs = extract_upstream_from_source( + {"type": "ckan", "args": {"dataset_id": "cig-{year}", "portal_url": "https://ex.it"}} + ) + assert refs == [{"kind": "ckan_dataset", "id": "cig-{year}"}] diff --git a/tests/test_html_collector_pure.py b/tests/test_html_collector_pure.py index 85e8e37b..3693d3e8 100644 --- a/tests/test_html_collector_pure.py +++ b/tests/test_html_collector_pure.py @@ -1,11 +1,11 @@ """Test delle funzioni pure per collector HTML (nessun HTTP). -Il motore di estrazione link è in ``toolkit.scout.link_extractor``. +Il motore di estrazione link è in ``lab_connectors.discovery.links``. Qui testiamo l'orchestrazione SO e le deleghe. """ import pytest -from toolkit.scout.link_extractor import DataLink +from lab_connectors.discovery.links import DataLink from scripts.collectors.html import ( _build_row, @@ -198,13 +198,13 @@ def _dl(url: str, fmt: str = "CSV") -> DataLink: from urllib.parse import urlparse filename = urlparse(url).path.rsplit("/", 1)[-1].rsplit(".", 1)[0] - from toolkit.scout.link_extractor import _extract_prefix, _extract_years + from lab_connectors.discovery.links import extract_prefix, extract_years return DataLink( url=url, format=fmt, - prefix=_extract_prefix(filename), - years=_extract_years(filename), + prefix=extract_prefix(filename), + years=extract_years(filename), ) diff --git a/tests/test_integration_toolkit_scout.py b/tests/test_integration_toolkit_scout.py index c1ab4128..bd204fca 100644 --- a/tests/test_integration_toolkit_scout.py +++ b/tests/test_integration_toolkit_scout.py @@ -1,12 +1,12 @@ -"""Test integrazione SO ↔ toolkit.scout. +"""Test integrazione SO ↔ lab_connectors.discovery. -Verifica che le funzioni condivise da toolkit abbiano i contratti attesi da SO. +Verifica che le funzioni condivise da LC abbiano i contratti attesi da SO. Nessuna dipendenza di rete — solo funzioni pure. """ import pytest -from toolkit.scout.http import resolve_preview_kind -from toolkit.scout.infer import infer_granularity, infer_years +from lab_connectors.discovery.classify import resolve_preview_kind +from lab_connectors.discovery.infer import infer_granularity, infer_years pytestmark = pytest.mark.contract