diff --git a/docs/runbook.md b/docs/runbook.md index c3010a7e..6d4639ab 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -63,6 +63,42 @@ Modello v0: - problemi di connessione/HTTP vanno letti in `radar_summary.json`, non in `catalog_signals.json` - **Issue automatiche**: in caso di variazioni rilevanti nel catalogo, `observatory.yml` crea o aggiorna una issue `catalog-alert` su GitHub +### Catalog per-repo e `pattern_probes` + +Ogni repo dati che dichiara `[tool.observatory]` nel proprio `pyproject.toml` +genera il catalog in-repo (`registry/catalog.json`) con `so-catalog` — engine +condiviso, zero secrets (vedi `catalog-reusable.yml`). + +Per le fonti **senza catalogo enumerabile ma con URL deterministici** si +dichiarano sentinelle in `pattern_probes`: se una passa da non disponibile ad +available, il verdict riceve il trigger `new_content_available`. + +```toml +[tool.observatory] +source_id = "mef_finanze" +protocol = "html" +base_url = "https://www1.finanze.gov.it/finanze/analisi_stat/" + +[[tool.observatory.pattern_probes]] +name = "iva-regionale-2026" +url = "https://www1.finanze.gov.it/finanze/analisi_stat/public/index.php?tree=2026CIVATOT0201&export=3" +min_bytes = 200 +``` + +- `min_bytes` (opzionale): sotto quella soglia una risposta 2xx **non** conta + come disponibile. Serve per le fonti che rispondono 200 con corpo vuoto + quando il dato non esiste ancora (es. MEF `analisi_stat`): senza soglia il + trigger "nuovo contenuto" scatterebbe a ogni run, per sempre. + +**Query-string negli artifact**: l'URL che va al probe è quello intero, query +inclusa (è l'identità della risorsa per molte fonti), quindi la query finisce +nei parquet tracciati — `data/pipeline/validated.parquet` in questo repo e +`out/so/validated.parquet` nei repo dati. Oggi i parametri osservati sono +`format`, `d`, `t`: nessuna credenziale. **Prima di onboardare una fonte con +URL firmati/tokenizzati** (`?token=`, `?sig=`, `?api_key=`, …) va aggiunta la +redazione dei parametri sensibili (#436), altrimenti il valore viene persistito +in chiaro nel parquet committato. + ## Catalog inventory ```bash diff --git a/scripts/catalog/config.py b/scripts/catalog/config.py index eb6ef3c3..ecc32362 100644 --- a/scripts/catalog/config.py +++ b/scripts/catalog/config.py @@ -64,6 +64,24 @@ def _extract_tool_table(data: dict[str, Any], section: str) -> dict[str, Any]: return dict(block) +def _normalize_pattern_probe(entry: dict[str, Any], where: str) -> dict[str, Any]: + """Normalizza una entry di ``pattern_probes``. + + ``min_bytes`` (opzionale) deve essere un intero >= 0: sotto quella soglia + una risposta 2xx non conta come "nuovo contenuto disponibile". + """ + probe = dict(entry) + raw = probe.get("min_bytes") + if raw is None: + return probe + if isinstance(raw, bool) or not isinstance(raw, int) or raw < 0: + raise ObservatoryConfigError( + f"[tool.observatory]{where}: pattern_probes.min_bytes must be " + f"an integer >= 0, got {raw!r}" + ) + return probe + + def validate_observatory_config( obs: dict[str, Any], *, @@ -133,7 +151,9 @@ def validate_observatory_config( if isinstance(obs.get("catalog_baseline"), dict): cfg["catalog_baseline"] = dict(obs["catalog_baseline"]) if isinstance(obs.get("pattern_probes"), list): - cfg["pattern_probes"] = [dict(p) for p in obs["pattern_probes"] if isinstance(p, dict)] + cfg["pattern_probes"] = [ + _normalize_pattern_probe(p, where) for p in obs["pattern_probes"] if isinstance(p, dict) + ] # Pass-through: blocchi specifici di protocollo e campi ignoti. known = set(REQUIRED_KEYS) | { diff --git a/scripts/catalog/pattern_probes.py b/scripts/catalog/pattern_probes.py index b55cb03d..4d077d23 100644 --- a/scripts/catalog/pattern_probes.py +++ b/scripts/catalog/pattern_probes.py @@ -20,19 +20,32 @@ TIMEOUT_SECONDS = 30 -def probe_pattern(url: str) -> dict[str, Any]: - """Probe singolo URL. Non alza eccezioni: errore → available=False.""" +def probe_pattern(url: str, min_bytes: int = 0) -> dict[str, Any]: + """Probe singolo URL. Non alza eccezioni: errore → available=False. + + Args: + url: URL sentinella da verificare. + min_bytes: se > 0, una risposta 2xx piu' corta di N byte non conta + come disponibile. Serve per le fonti che rispondono 200 con corpo + vuoto quando la risorsa non esiste ancora (es. MEF analisi_stat: + 200 + 0 byte per l'anno non pubblicato). + """ client = HttpClient(timeout=TIMEOUT_SECONDS, user_agent=USER_AGENT) result = client.get(url, allow_redirects=True, stream=True) if result.is_ok and result.response is not None: resp = result.response - available = 200 <= resp.status_code < 300 - return { + is_2xx = 200 <= resp.status_code < 300 + length = len(resp.content) if resp.content is not None else 0 + below_min_bytes = is_2xx and min_bytes > 0 and length < min_bytes + outcome: dict[str, Any] = { "url": url, - "available": available, + "available": is_2xx and not below_min_bytes, "http_code": str(resp.status_code), - "content_length": len(resp.content) if available else None, + "content_length": length if is_2xx else None, } + if below_min_bytes: + outcome["below_min_bytes"] = True + return outcome return { "url": url, "available": False, @@ -44,20 +57,33 @@ def probe_pattern(url: str) -> dict[str, Any]: def run_pattern_probes(probes: list[dict[str, Any]]) -> list[dict[str, Any]]: """Esegue tutti i pattern_probes del config. - Ogni entry: {"name": "...", "url": "..."}. Il campo name e' opzionale. + Ogni entry: {"name": "...", "url": "...", "min_bytes": }. + Il campo name e' opzionale. """ results: list[dict[str, Any]] = [] for entry in probes: url = str(entry.get("url") or "") if not url: continue - outcome = probe_pattern(url) + outcome = probe_pattern(url, min_bytes=_entry_min_bytes(entry)) if entry.get("name"): outcome["name"] = str(entry["name"]) results.append(outcome) return results +def _entry_min_bytes(entry: dict[str, Any]) -> int: + """``min_bytes`` dell'entry come int >= 0 (0 = nessuna soglia).""" + raw = entry.get("min_bytes") + if raw is None: + return 0 + try: + value = int(raw) + except (TypeError, ValueError): + return 0 + return max(0, value) + + def compute_new_content_trigger( current: list[dict[str, Any]], previous_catalog: dict[str, Any] | None, diff --git a/scripts/collectors/_validate_base.py b/scripts/collectors/_validate_base.py index bc45280e..275efc6c 100644 --- a/scripts/collectors/_validate_base.py +++ b/scripts/collectors/_validate_base.py @@ -398,11 +398,16 @@ def validate_tabular_group( url_raw = best.get("distribution_url") or best.get("url") or "" if not isinstance(url_raw, str): url_raw = "" - url = url_raw.split("?")[0] + # L'URL va usato INTERO nel probe: molte fonti servono il dato via + # query-string e il path nudo risponde altro (es. MEF analisi_stat: + # ``index.php?tree=2025CIVATOT0201&export=3`` → CSV, ``index.php`` → HTML + # del portale). Il path senza query serve solo a indovinare il formato. + url = url_raw + url_path = url_raw.split("?")[0] fmt = best.get("format") or "" if not isinstance(fmt, str): fmt = "" - is_csv = "csv" in fmt.lower() or url.lower().endswith(".csv") + is_csv = "csv" in fmt.lower() or url_path.lower().endswith(".csv") result: dict[str, Any] = { "dataset_group": best.get("dataset_group", "unknown"), diff --git a/tests/pipeline/test_run_pipeline.py b/tests/pipeline/test_run_pipeline.py index c9137577..b2790e4e 100644 --- a/tests/pipeline/test_run_pipeline.py +++ b/tests/pipeline/test_run_pipeline.py @@ -21,7 +21,7 @@ "https://www1.finanze.gov.it/finanze/analisi_stat/public/v_4_0_0/" "contenuti/REG_tipo_reddito_2025.csv?d=1615465800" ) -MEF_URL_NO_QUERY = MEF_URL.split("?")[0] +# L'URL con query-string è il dato: il validate lo usa INTERO (regressione #433) def _make_group_df(suffix: str = "") -> pd.DataFrame: @@ -44,9 +44,7 @@ def test_validate_one_with_shared_client(monkeypatch, fake_http): from scripts.pipeline import run_pipeline csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n2021,Lombardia,2000\n" - fake_http.responses[MEF_URL_NO_QUERY] = HttpResult( - response=fake_response(200, csv_body), err=None - ) + fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, csv_body), err=None) # il validate crea client con HttpClient(...) — patchiamo la factory # (importata da lab_connectors.http sia in _validate_base sia in run_pipeline) monkeypatch.setattr("lab_connectors.http.HttpClient", lambda **kw: fake_http) @@ -65,9 +63,7 @@ def test_validate_one_without_client_creates_own(monkeypatch, fake_http): from scripts.pipeline import run_pipeline csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n" - fake_http.responses[MEF_URL_NO_QUERY] = HttpResult( - response=fake_response(200, csv_body), err=None - ) + fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, csv_body), err=None) # _validate_base importa HttpClient a livello modulo: patchiamo il nome # bindato nel modulo consumer (il path lab_connectors.http non basta quando # il pacchetto e' installato, non editable) @@ -84,9 +80,7 @@ def test_run_validate_uses_shared_client(monkeypatch, fake_http): from scripts.pipeline import run_pipeline csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n" - fake_http.responses[MEF_URL_NO_QUERY] = HttpResult( - response=fake_response(200, csv_body), err=None - ) + fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, csv_body), err=None) monkeypatch.setattr("lab_connectors.http.HttpClient", lambda **kw: fake_http) # due gruppi distinti per verificare che il client sia condiviso diff --git a/tests/pipeline/test_validate_utils.py b/tests/pipeline/test_validate_utils.py index ee793aa6..65c23329 100644 --- a/tests/pipeline/test_validate_utils.py +++ b/tests/pipeline/test_validate_utils.py @@ -38,8 +38,7 @@ "https://www1.finanze.gov.it/finanze/analisi_stat/public/v_4_0_0/" "contenuti/REG_tipo_reddito_2025.csv?d=1615465800" ) -# URL senza query: validate_tabular_group fa url.split('?')[0] prima del probe -MEF_URL_NO_QUERY = MEF_URL.split("?")[0] +# L'URL con query-string è il dato: il validate lo usa INTERO (regressione #433) # ── Unit: format scoring ────────────────────────────────────────────────────── @@ -228,15 +227,13 @@ def test_validate_mef_irpef_group(self, monkeypatch, fake_http): "2024,Lazio,AAA000000000000X,1000\n" "2023,Lombardia,BBB000000000000Y,2000\n" ) - fake_http.responses[MEF_URL_NO_QUERY] = HttpResult( - response=fake_response(200, csv_body), err=None - ) + fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, csv_body), err=None) monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http) items = [ { "dataset_group": "mef_irpef/statistiche/reg-tipo-reddito", "source_id": "mef_irpef", - "distribution_url": MEF_URL_NO_QUERY, + "distribution_url": MEF_URL, "format": "csv", "year_signal": 2025, }, @@ -249,15 +246,13 @@ def test_validate_mef_irpef_group(self, monkeypatch, fake_http): def test_validate_csv_with_sniff(self, monkeypatch, fake_http): """Validazione con sniff: anno estratto dalla colonna ANNO.""" csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n2021,Lombardia,2000\n" - fake_http.responses[MEF_URL_NO_QUERY] = HttpResult( - response=fake_response(200, csv_body), err=None - ) + fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, csv_body), err=None) monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http) items = [ { "dataset_group": "mef_irpef/statistiche/reg-tipo-reddito", "source_id": "mef_irpef", - "distribution_url": MEF_URL_NO_QUERY, + "distribution_url": MEF_URL, "format": "csv", "year_signal": 2025, } @@ -267,6 +262,49 @@ def test_validate_csv_with_sniff(self, monkeypatch, fake_http): assert result.get("dataset_group_year_min") == 2020 assert result.get("dataset_group_year_max") == 2021 + def test_validate_keeps_query_string_in_probe(self, monkeypatch, fake_http): + """Regressione #433: l'URL con query-string è il dato, non va troncato. + + Con un URL tipo ``index.php?tree=...&export=3`` il path nudo risponde + HTML: il validatore deve interrogare l'URL intero. + """ + csv_body = "ANNO,REGIONE,IMPORTO\n2024,Lazio,1000\n" + fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, csv_body), err=None) + monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http) + items = [ + { + "dataset_group": "mef_finanze/iva/tot-regione", + "source_id": "mef_finanze", + "distribution_url": MEF_URL, + "format": "csv", + } + ] + result = validate_group(items) + assert result["url"] == MEF_URL + assert result["reachable"] is True + assert len(result.get("columns", [])) > 0 + + def test_validate_query_only_url_is_csv_by_format(self, monkeypatch, fake_http): + """Senza estensione nel path, è il campo ``format`` a decidere il probe.""" + url = ( + "https://www1.finanze.gov.it/finanze/analisi_stat/public/" + "index.php?tree=2025CIVATOT0201&export=3" + ) + csv_body = "Regione;Numero contribuenti IVA\nPiemonte;278397\n" + fake_http.responses[url] = HttpResult(response=fake_response(200, csv_body), err=None) + monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http) + items = [ + { + "dataset_group": "mef_finanze/iva/tot-regione", + "source_id": "mef_finanze", + "distribution_url": url, + "format": "CSV", + } + ] + result = validate_group(items) + assert result["reachable"] is True + assert result["url"] == url + def test_validate_no_url_returns_error(self): items = [{"dataset_group": "test/no-url", "source_id": "test", "format": "csv"}] result = validate_group(items) @@ -294,9 +332,7 @@ def test_validate_nan_year_propagated(self, monkeypatch, fake_http): """Il merge pandas produce NaN (non None) per gli anni: il validatore deve trattarlo come 'assente' e usare lo sniff.""" csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n2021,Lombardia,2000\n" - fake_http.responses[MEF_URL_NO_QUERY] = HttpResult( - response=fake_response(200, csv_body), err=None - ) + fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, csv_body), err=None) monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http) items = [ { @@ -317,9 +353,7 @@ def test_validate_nan_year_propagated(self, monkeypatch, fake_http): def test_validate_with_existing_years_keeps_them(self, monkeypatch, fake_http): """Se il gruppo fornisce gia' anni validi, lo sniff non li sovrascrive.""" csv_body = "ANNO,REGIONE,IMPORTO\n2020,Lazio,1000\n" - fake_http.responses[MEF_URL_NO_QUERY] = HttpResult( - response=fake_response(200, csv_body), err=None - ) + fake_http.responses[MEF_URL] = HttpResult(response=fake_response(200, csv_body), err=None) monkeypatch.setattr("scripts.collectors._validate_base.HttpClient", lambda **kw: fake_http) items = [ { diff --git a/tests/test_catalog_config.py b/tests/test_catalog_config.py index cf0d2dd4..be1105cf 100644 --- a/tests/test_catalog_config.py +++ b/tests/test_catalog_config.py @@ -118,6 +118,48 @@ def test_passthrough_protocol_blocks(self): assert cfg["catalog_baseline"] == {"value": 100, "method": "count"} assert cfg["sparql"] == {"query": "SELECT * WHERE {?s ?p ?o}"} + @pytest.mark.pure_unit + def test_pattern_probes_min_bytes_kept(self): + cfg = validate_observatory_config( + { + "source_id": "mef_finanze", + "protocol": "html", + "base_url": "https://www1.finanze.gov.it/finanze/analisi_stat/", + "pattern_probes": [ + { + "name": "iva-2026", + "url": "https://x/y?tree=2026CIVATOT0201", + "min_bytes": 200, + } + ], + } + ) + assert cfg["pattern_probes"][0]["min_bytes"] == 200 + + @pytest.mark.pure_unit + def test_pattern_probes_min_bytes_invalid(self): + with pytest.raises(ObservatoryConfigError, match="min_bytes"): + validate_observatory_config( + { + "source_id": "mef_finanze", + "protocol": "html", + "base_url": "https://x/", + "pattern_probes": [{"url": "https://x/y", "min_bytes": -1}], + } + ) + + @pytest.mark.pure_unit + def test_pattern_probes_min_bytes_not_int(self): + with pytest.raises(ObservatoryConfigError, match="min_bytes"): + validate_observatory_config( + { + "source_id": "mef_finanze", + "protocol": "html", + "base_url": "https://x/", + "pattern_probes": [{"url": "https://x/y", "min_bytes": "200"}], + } + ) + # ── load_observatory_config (file-level) ───────────────────────────────────── diff --git a/tests/test_catalog_pattern_probes.py b/tests/test_catalog_pattern_probes.py index c65256b5..aac333a6 100644 --- a/tests/test_catalog_pattern_probes.py +++ b/tests/test_catalog_pattern_probes.py @@ -56,6 +56,57 @@ def get(self, url, **kw): assert r["http_code"] == "404" assert r["content_length"] is None + @pytest.mark.pure_unit + def test_empty_200_below_min_bytes_is_unavailable(self, monkeypatch): + """Regressione #433: MEF risponde 200 + 0 byte per l'anno mancante. + + Con ``min_bytes`` la risposta vuota non conta come disponibile: + altrimenti il trigger "nuovo contenuto" scatta per sempre. + """ + + class FakeResp: + status_code = 200 + content = b"" + + class FakeResult: + is_ok = True + response = FakeResp() + err = None + + class FakeClient: + def __init__(self, **kw): ... + def get(self, url, **kw): + return FakeResult() + + monkeypatch.setattr("scripts.catalog.pattern_probes.HttpClient", FakeClient) + r = probe_pattern("https://ex.it/2026.csv", min_bytes=200) + assert r["available"] is False + assert r["http_code"] == "200" + assert r["content_length"] == 0 + assert r["below_min_bytes"] is True + + @pytest.mark.pure_unit + def test_above_min_bytes_is_available(self, monkeypatch): + class FakeResp: + status_code = 200 + content = b"x" * 500 + + class FakeResult: + is_ok = True + response = FakeResp() + err = None + + class FakeClient: + def __init__(self, **kw): ... + def get(self, url, **kw): + return FakeResult() + + monkeypatch.setattr("scripts.catalog.pattern_probes.HttpClient", FakeClient) + r = probe_pattern("https://ex.it/2026.csv", min_bytes=200) + assert r["available"] is True + assert r["content_length"] == 500 + assert "below_min_bytes" not in r + @pytest.mark.pure_unit def test_error_is_unavailable(self, monkeypatch): class FakeResult: @@ -79,7 +130,7 @@ class TestRunPatternProbes: def test_runs_all_and_names(self, monkeypatch): monkeypatch.setattr( "scripts.catalog.pattern_probes.probe_pattern", - lambda url: {"url": url, "available": True, "http_code": "200"}, + lambda url, min_bytes=0: {"url": url, "available": True, "http_code": "200"}, ) results = run_pattern_probes( [ @@ -92,6 +143,25 @@ def test_runs_all_and_names(self, monkeypatch): assert results[0]["name"] == "fsc-2026" assert "name" not in results[1] + @pytest.mark.pure_unit + def test_min_bytes_passed_to_probe(self, monkeypatch): + """``min_bytes`` dell'entry arriva al probe (sentinelle MEF: 200 + 0 byte).""" + seen: list[int] = [] + + def fake_probe(url, min_bytes=0): + seen.append(min_bytes) + return {"url": url, "available": True, "http_code": "200"} + + monkeypatch.setattr("scripts.catalog.pattern_probes.probe_pattern", fake_probe) + run_pattern_probes( + [ + {"url": "https://ex.it/2026.csv", "min_bytes": 200}, + {"url": "https://ex.it/2025.csv"}, + {"url": "https://ex.it/2024.csv", "min_bytes": "boh"}, + ] + ) + assert seen == [200, 0, 0] + class TestComputeNewContentTrigger: @pytest.mark.pure_unit