feat: pipeline merge+validate — nuovo funnel SO (−6.400 righe nette)#412
Open
Gabrymi93 wants to merge 28 commits into
Open
feat: pipeline merge+validate — nuovo funnel SO (−6.400 righe nette)#412Gabrymi93 wants to merge 28 commits into
Gabrymi93 wants to merge 28 commits into
Conversation
- _merge_utils.py: normalize_title_for_merge() con 8 strategie: strip_years, strip_temporal_stopwords, strip_territory_prefix, strip_territory_prefix_pattern, strip_variation_suffix, strip_territory_suffix, strip_leading_year_prefix - compute_dataset_group() con disambiguazione via item_id stem per titoli generici (es. MEF IRPEF: 147→11 gruppi) - add_dataset_group_columns() per applicazione su dataframe - 72 test passanti con titoli reali da ACI, OpenBDAP, Unioncamere, MIM, INPS, Ministero Interno, MEF IRPEF, AIFA, OpenGA Impatto su inventory reale: 10.897 item → 7.567 gruppi (1.4x) Best: MEF IRPEF 13.4x, MIM 13.4x, MIMIT RNA 8.6x, ACI 7x
- _validate_utils.py: pick_best_url() sceglie CSV>altro e anno recente - probe_reachability(): HTTP HEAD su URL - sniff_csv_schema(): Range GET + csv.DictReader per colonne - validate_group(): end-to-end per gruppo dataset - 20 test, 80% coverage su _validate_utils - Sostituisce: bulk_source_check + source_check_fetch + source_check_analyze
- sniff_csv_schema usava requests.get() direttamente, saltando il fallback SSL che HttpClient fornisce - AIFA e altre fonti con certificati SSL non validi ora sniffano correttamente (12 colonne da AIFA)
Root cause: _distribution_url prendeva resources[0].url senza considerare il formato. Su ANAC, la prima risorsa e' spesso TTL (formato peggiore), ma _resource_format segnava 'csv,json,ttl' dando l'impressione che l'URL servisse CSV. Fix: _best_resource_url() itera tutte le risorse e sceglie quella col formato piu' pipeline-friendly: CSV > JSON > XML > XLSX > ZIP > TTL. _resource_format (tutti i formati disponibili) rimane invariato. Impatto ANAC: 29/70 URL gia' puntavano a CSV (invariato). I restanti 41 passano da TTL/JSON a CSV.
SDMX ha caratteristiche diverse da CKAN/HTML: - Non supporta HEAD (405) ne Range (416) - Endpoint lentissimo (13MB XML, minuti per rispondere) - Il collector ha gia verificato la fonte in fase di inventory Il validatore SDMX: - Non fa probe HTTP attivo (inaffidabile per SDMX) - Considera un dataflow raggiungibile se ha api_base_url + distribution_url - Rileva dimensioni dal titolo (sesso, eta, territorio...) - Segnala anomalie (item senza titolo, senza URL)
La logica di validazione ora vive nei collector, non in pipeline/. Ogni collector puo' definire validate_items() con la logica specifica per il suo protocollo.
- validate_tabular_group: se il formato non contiene CSV, salta completamente HEAD e sniff (prima faceva HEAD comunque) - Per CSV: sniff leggero interno (encoding, delim, colonne) - deep=True opzionale per usare toolkit.profile.preview (piu' lento ma con tipi colonna e quality score) - Impatto: ANAC passa da 48 probe HTTP a 6 (solo i gruppi CSV)
- run_validate: supporta --workers N per probe paralleli - 4 workers: 100 gruppi in 27s (3.7/s) vs 100s sequenziale (3.7x) - 8 workers non migliora (rete e' il collo di bottiglia, non CPU) - Full run stimato: ~35 minuti per 11.387 gruppi (weekly batch ok)
Rimosso (sicuro - nessuna dipendenza CI): source_check_analyze.py, source_check_fetch.py, pipeline/_validate_utils.py, pipeline/validate_sdmx.py. Aggiornati test. -1351 righe
-5348 righe nette, 264 test passano
Rimosso: - build_catalog_signals.py (catalog signals obsoleti) - catalog_diff.py (diff segnali) - bulk_source_check.py (sostituito da run_pipeline.py) - publish_source_check_summary.py (legacy) - build_issue_body.py (alert issue non più generato) - data/catalog/ (artefatti obsoleti) - data/catalog_inventory/generated/source_check_results.parquet Nuovo flusso CI weekly: 1. build_catalog_inventory.py 2. run_pipeline.py --workers 4 (merge + validate) 3. build_source_reports.py 4. Upload GCS (validated.parquet + inventory snapshot) 5. Artifact + radar summary
Il worker thread non trovava so_mcp perche' sys.path conteneva scripts/ ma non il repo root. Aggiunge entrambi.
- Bug: readiness_score calcolato PRIMA dello sniff → num_columns sempre 0 Fix: spostato dopo sniff (score 10 raggiungibile, 126 item a 10) - Esteso a 0-10 con 8 componenti - SPARQL validator: COUNT query endpoint + reachability - Cablato in VALIDATORS (era # DA FARE)
- Sniff fallito (num_columns=0 dopo aver provato a parsare CSV): -3 - Content-type HTTP non CSV (es. application/vnd.ms-excel): -1 - Score 6 prima: 274 falsi CSV (XLSX spacciati per CSV) - Score dopo: ridistribuiti su 2-3 (onesti)
Packaging: - scripts reso package installabile (__init__.py, pyproject.toml) - 0 sys.path.insert (erano 4 in script + 1 in test) - 5 entry point CLI (so-run-pipeline, so-build-reports, so-radar-check, so-sync-datasets) - import con scripts. prefix invece di sys.path hacks Docs: - README, CONTRIBUTING, architecture.md aggiornati - validated_schema.md (sostituisce source_check_results_schema.md) - CI: rm catalog_signals.json inutile rimosso (loop cleanup-rigenera) Test MCP: - 17 test skip riscritti per validated.parquet - intake_score/paqa_score/intake_candidate → readiness_score/reachable - inventory_status/inventory_diff su nuovo output schema - 278 passed, 2 skip (SPARQL obsoleti) year_signal: - merge usa year_signal come fallback per year_min/year_max
Gabrymi93
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Sintesi
Sostituisce il vecchio flusso source-check (bulk_source_check, source_check_analyze, source_check_fetch, catalog_diff) con una nuova pipeline merge → validate. Risultato: −6.400 righe nette (+3.400 / −9.700), readiness_score 0-10, merge/dedup logico per dataset.
Cosa cambia
Dettaglio
Pipeline merge + validate (nuovo)
dataset_group), slug coerentireadiness_score0-10# DA FAREresiduo)Rimosso
bulk_source_check.py(1283 righe),source_check_analyze.py(546),source_check_fetch.py(390)catalog_diff.py(151),build_catalog_signals.py(362),run_source.py(371)_validate_utils.py(duplicato di_validate_base.py)JOIN_KEY_PATTERNSe funzioni associate (150 righe morte)_FORMAT_PRIORITYtriplicato → centralizzato in_constants.pycompute_dataset_groupduplicato → unificato in_merge_utils.pyPackaging
scripts/reso package installabile (prima non era incluso in pyproject.toml)sys.path.insert(erano 4 in script + 1 in test)source-observatory-mcp,so-run-pipeline,so-build-reports,so-radar-check,so-sync-datasetsMCP
so_source_checklegge davalidated.parquet(con fallback al vecchio path)min_paqa_scorerimosso dalla signature (parametro ignorato con backward compat)inventory_status()einventory_diff()riscritte per nuovo report schemaTest
intake_score→readiness_score,_CHECK_PARQUET→_VALIDATED_PARQUET)Checklist
pytest tests/passa (278 passed, 2 skip)ruff check .passadocs/architecture.mdaggiornatoNote per chi revisiona
discover_named_graphs,_collect_named_graphs). Vanno riscritti per la nuova implementazione SPARQL.report_version: 2: campiintake_candidates→tot_reachable/tot_csv_count. Verificare che nessun consumer downstream usi i vecchi campi.