Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 36 additions & 0 deletions docs/runbook.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
22 changes: 21 additions & 1 deletion scripts/catalog/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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],
*,
Expand Down Expand Up @@ -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) | {
Expand Down
42 changes: 34 additions & 8 deletions scripts/catalog/pattern_probes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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": <int, opzionale>}.
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,
Expand Down
9 changes: 7 additions & 2 deletions scripts/collectors/_validate_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"),
Expand Down
14 changes: 4 additions & 10 deletions tests/pipeline/test_run_pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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
Expand Down
66 changes: 50 additions & 16 deletions tests/pipeline/test_validate_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 ──────────────────────────────────────────────────────

Expand Down Expand Up @@ -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,
},
Expand All @@ -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,
}
Expand All @@ -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)
Expand Down Expand Up @@ -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 = [
{
Expand All @@ -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 = [
{
Expand Down
42 changes: 42 additions & 0 deletions tests/test_catalog_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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) ─────────────────────────────────────

Expand Down
Loading
Loading