refactor(decoupling): rimuovi dipendenza toolkit → lab_connectors.discovery

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]

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)

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" }

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":

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
 

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,

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 ─────────────────────────────────────────────

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