fix(pipeline): fallback DuckDB sniff, estrazione anni, circuit breaker condiviso

[Gabrymi93]

Sintesi

Tre fix della pipeline di validazione SO (scripts/pipeline/run_pipeline.py + scripts/collectors/), che migliorano copertura, dati e velocità del validated.parquet.

Contesto collegato

Nessuna issue dedicata — emersi da diagnosi pipeline (attrito SO→DI): 268 CSV "reachable senza schema", 0 gruppi con anni su fonti grosse (INPS), run completo lento su host giù.

Dipendenza: PR lab-connectors#73 (fix verify duplicato nel fallback SSL HEAD). Senza di essa, ministero_salute torna a perdere 38 CSV. Da mergiare prima o insieme.

Cosa cambia

• Nuova fonte o modifica registro (sources_registry.yaml)
• Source-check o inventory-triage
• Modifica script (radar, inventory, source-check, MCP)
• Modifica funnel o criteri di osservazione
• Workflow CI (radar.yml, observatory.yml)
• Skills o MCP tools
• Documentazione
• Altro

Dettaglio fix

1. Fallback DuckDB nello sniff CSV (_validate_base.py)

Il parser standard (csv.Sniffer + DictReader) fallisce su SDMX-CSV e CSV con newline nei campi non quotati → gruppi "reachable ma senza schema". DuckDB li gestisce. Risultato reale (run completo): +43 gruppi profilati (INPS 37, ministero_interno 4, pagopa 1, unioncamere 1).

2. Estrazione range anni (_extract_year_range + propagazione)

Popola dataset_group_year_min/max con 3 strategie: anni da nomi colonna → valori via DuckDB tipizzato → filename. Include fix del NaN (il merge pandas produce NaN, non None: il check is None non scattava e gli anni restavano vuoti). Risultato reale (run completo): gruppi con anni 101 → 2.751 (INPS 0 → 1.764, openga 401, unioncamere 142).

3. Circuit breaker condiviso (run_pipeline.py + firme validate)

Un solo HttpClient(circuit_threshold=3, timeout=5) per tutta la run, passato a probe_reachability/sniff_csv_schema/validate_tabular_group (e ai validatori ckan/sdmx/sparql, che accettano client — sdmx/sparql lo ignorano). Host giù (lavoro_opendata, dati.anticorruzione.it, bdap-opendata) bloccati dopo 3 errori invece di ~11s per gruppo. Run completo: ~15min+ → 8min. Nota: timeout=5 corretto — il default 60s del client faceva bruciare ~114s per fallimento prima dell'apertura del circuito.

Verifica

pytest tests/
ruff check .
mypy scripts/ so_mcp/

• pytest tests/pipeline/test_validate_utils.py passa (25 test, 5 nuovi: TestExtractYearRange + fallback DuckDB)
• ruff check scripts/ passa
• mypy scripts/collectors/_validate_base.py scripts/pipeline/run_pipeline.py passa
• Run completo reale: run_pipeline --workers 8 → 11.129 gruppi in 484,8s. BEFORE vs AFTER: csv profilati 2.914→2.957, con_anni 101→2.751, ministero_salute 2→40 profilati
• Circuit breaker verificato in produzione: host giù bloccati dopo 3 errori (log Circuit open for host), centinaia di gruppi skippati in ms
• Perimetro stretto: una PR = i fix della validazione (6 file)

Note per chi revisiona

• anac 47→0 nel run locale NON è regressione: dati.anticorruzione.it oggi risponde InvalidBodyLengthError (probe diretto confermato). Era vivo al 27/07. Drift di fonte, non dei fix.
• Non coperti da questa PR (documentati come follow-up): opencoesione 27 ZIP dichiarati CSV (serve apertura archivio), INPS ~34 dfb_st_* a riga unica (reshape non banale, dati fermi 2015-2022 → condizionato ad aggiornamento fonte).
• Il validated.parquet rigenerato vive in /tmp/opencode/validated_full.parquet (non versionato nel repo) — la CI lo rigenererà al merge.
• lab-connectors è stato reinstallato in editable nel venv locale per testare il fix tests: aggiungere test per resource_monitor.py #73 insieme (allinea lo standard workspace: tutti gli altri repo sono editable).