From a8a531c8afb5ce3a732d4b11fe3611f715d62240 Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:39:54 +0100 Subject: [PATCH 1/6] =?UTF-8?q?Fase=201:=20report=20unificato=20=E2=80=94?= =?UTF-8?q?=20validazione=20in=20memoria,=20merge=20snellito?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/helpers_assert_paths.py | 12 ++------ tests/test_pipeline_integration.py | 2 -- tests/test_validate_layers.py | 45 ++++++++++++++---------------- toolkit/clean/run.py | 2 -- toolkit/clean/validate.py | 16 +---------- toolkit/cli/cmd_run.py | 5 ++++ toolkit/core/metadata.py | 17 +---------- toolkit/core/validation.py | 7 +++++ toolkit/domain/report.py | 11 ++++---- toolkit/mart/run.py | 2 -- toolkit/mart/validate.py | 16 ++--------- toolkit/raw/run.py | 27 ++---------------- toolkit/raw/validate.py | 17 ++--------- 13 files changed, 51 insertions(+), 128 deletions(-) diff --git a/tests/helpers_assert_paths.py b/tests/helpers_assert_paths.py index 48363ed4..bd068fef 100644 --- a/tests/helpers_assert_paths.py +++ b/tests/helpers_assert_paths.py @@ -160,9 +160,9 @@ def assert_golden_path_artifacts( """Verifica la struttura completa degli artifact dopo un golden path. Controlla che per ogni anno esistano: - - RAW dir con raw_validation.json + - RAW dir con metadata.json - CLEAN dir con almeno un parquet e metadata.json - - MART dir con ogni tabella dichiarata e mart_validation.json + - MART dir con ogni tabella dichiarata Args: mart_tables: Lista di nomi tabelle mart (senza estensione). @@ -171,10 +171,7 @@ def assert_golden_path_artifacts( for year in years: # RAW - raw_dir = assert_raw_dir(root, dataset, year) - assert (raw_dir / "raw_validation.json").exists(), ( - f"raw_validation.json mancante in {raw_dir}" - ) + assert_raw_dir(root, dataset, year) # CLEAN clean_dir = assert_clean_dir(root, dataset, year) @@ -189,6 +186,3 @@ def assert_golden_path_artifacts( assert (mart_dir / f"{table}.parquet").exists(), ( f"MART table {table}.parquet mancante in {mart_dir}" ) - assert (mart_dir / "_validate" / "mart_validation.json").exists(), ( - f"mart_validation.json mancante in {mart_dir}" - ) diff --git a/tests/test_pipeline_integration.py b/tests/test_pipeline_integration.py index e96cb5bd..bda664d9 100644 --- a/tests/test_pipeline_integration.py +++ b/tests/test_pipeline_integration.py @@ -80,8 +80,6 @@ def test_artifacts_policy_minimal_behaves_like_standard(project_example: Path, m assert (raw_dir / "metadata.json").exists() assert (clean_dir / "metadata.json").exists() assert (mart_dir / "metadata.json").exists() - assert (mart_dir / "metadata.json").exists() - assert (mart_dir / "_validate" / "mart_validation.json").exists() def test_artifacts_policy_standard_keeps_expected_artifacts( diff --git a/tests/test_validate_layers.py b/tests/test_validate_layers.py index 092e6914..075a426c 100644 --- a/tests/test_validate_layers.py +++ b/tests/test_validate_layers.py @@ -260,21 +260,22 @@ def test_run_mart_validation_merges_transition_warnings_into_report(tmp_path: Pa assert summary["passed"] is True assert summary["warnings_count"] == 2 - report = json.loads( - (mart_dir / "_validate" / "mart_validation.json").read_text(encoding="utf-8") - ) - assert len(report["warnings"]) == 2 - assert any("row drop 30.0%" in warning for warning in report["warnings"]) - assert any("columns removed from clean" in warning for warning in report["warnings"]) - assert report["transition"]["profiles_count"] == 1 - assert report["transition"]["warnings_count"] == 2 - assert report["transition"]["config"] == { + warnings = summary.get("warnings") or [] + assert len(warnings) == 2 + assert any("row drop 30.0%" in warning for warning in warnings) + assert any("columns removed from clean" in warning for warning in warnings) + sections = summary.get("sections") or {} + transition = sections.get("transition") or {} + assert transition.get("profiles_count") == 1 + assert transition.get("warnings_count") == 2 + assert transition.get("config") == { "max_row_drop_pct": 20.0, "warn_removed_columns": True, "fail_on_row_drop_exceeded": False, } - assert any(item["kind"] == "row_drop_pct" for item in report["transition"]["warnings"]) - assert any(item["kind"] == "removed_columns" for item in report["transition"]["warnings"]) + transition_warnings = transition.get("warnings") or [] + assert any(item["kind"] == "row_drop_pct" for item in transition_warnings) + assert any(item["kind"] == "removed_columns" for item in transition_warnings) @pytest.mark.policy @@ -392,12 +393,10 @@ def test_run_clean_validation_raw_probe_source_legacy_autodetect(tmp_path: Path) # With no profile, the probe must fall back to legacy autodetect assert result["stats"].get("raw_probe_source") == "legacy_autodetect" - # Warning must mention the fallback reason - # build_validation_summary only includes the raw stats, not the full result.warnings - # So we check the warning was emitted by inspecting via the ValidationResult if accessible, - # or by verifying that the fallback path was taken (raw_probe_source = legacy_autodetect) - warning_texts = " ".join(_read_warnings_from_validation_report(clean_dir)) - assert "falling back to read_csv(auto_detect=true)" in warning_texts + # Warning must mention the fallback reason (from return value, not disk) + warnings = result.get("warnings") or [] + warning_text = " ".join(warnings) + assert "falling back to read_csv(auto_detect=true)" in warning_text @pytest.mark.policy @@ -445,7 +444,7 @@ def test_run_clean_validation_raw_probe_source_unavailable_when_no_raw_file( def _read_warnings_from_validation_report(clean_dir: Path) -> list[str]: - """Read warnings from the written validation JSON.""" + """Read warnings from the written validation JSON (legacy, file may not exist).""" report_path = clean_dir / "_validate" / "clean_validation.json" if report_path.exists(): data = json.loads(report_path.read_text(encoding="utf-8")) @@ -687,12 +686,10 @@ def test_run_mart_validation_transition_errors_set_ok_false(tmp_path: Path): ) assert summary["errors_count"] >= 1, "Expected at least 1 transition error" - # Verifica che il JSON di validazione su disco contenga l'errore - report = json.loads( - (mart_dir / "_validate" / "mart_validation.json").read_text(encoding="utf-8") - ) - assert len(report.get("errors", [])) >= 1 - assert any("row drop" in e for e in report["errors"]) + # Verifica che il summary contenga l'errore + errors = summary.get("errors") or [] + assert len(errors) >= 1 + assert any("row drop" in e for e in errors) # --------------------------------------------------------------------------- diff --git a/toolkit/clean/run.py b/toolkit/clean/run.py index 92161160..0f35c864 100644 --- a/toolkit/clean/run.py +++ b/toolkit/clean/run.py @@ -16,7 +16,6 @@ write_metadata, ) from toolkit.core.paths import ( - CLEAN_VALIDATION, layer_year_dir, resolve_root, resolve_sql_path, @@ -296,7 +295,6 @@ def run_clean( merge_layer_manifest( out_dir, metadata_path=metadata_path.name, - validation_path=CLEAN_VALIDATION, outputs=outputs, ) logger.info(f"CLEAN -> {output_path}") diff --git a/toolkit/clean/validate.py b/toolkit/clean/validate.py index 532e1507..6d2c11b7 100644 --- a/toolkit/clean/validate.py +++ b/toolkit/clean/validate.py @@ -20,9 +20,7 @@ from toolkit.clean._helpers import _input_files_from_clean_metadata, _profile_raw_input from toolkit.core.config import CleanValidationSpec, RangeRuleConfig, TransitionConfig from toolkit.core.layer_profile import compare_layer_profiles -from toolkit.core.metadata import merge_layer_manifest from toolkit.core.paths import ( - CLEAN_VALIDATION, METADATA, RAW_PROFILE, layer_year_dir, @@ -33,7 +31,6 @@ build_validation_summary, check_transitions, required_columns_check, - write_validation_json, ) from toolkit.quality.pa_csv_quality import assess_quality @@ -593,16 +590,5 @@ def _to_snake(n: str) -> str: sections=merged_sections, ) - report = write_validation_json(Path(out_dir) / CLEAN_VALIDATION, result) - metadata = read_json_or_none(out_dir / METADATA) or {} - merge_layer_manifest( - out_dir, - metadata_path=METADATA, - validation_path="_validate/clean_validation.json", - outputs=metadata.get("outputs", []), - ok=result.ok, - errors_count=len(result.errors), - warnings_count=len(result.warnings), - ) - logger.info(f"VALIDATE CLEAN -> {report} (ok={result.ok})") + logger.info(f"VALIDATE CLEAN -> (ok={result.ok})") return build_validation_summary(result) diff --git a/toolkit/cli/cmd_run.py b/toolkit/cli/cmd_run.py index b1e9a2d8..b15dacef 100644 --- a/toolkit/cli/cmd_run.py +++ b/toolkit/cli/cmd_run.py @@ -315,6 +315,8 @@ def run_year( run_has_validation_warnings = False sample_mode = sample_rows is not None or sample_bytes is not None + validations: dict[str, dict[str, Any]] = {} + def _execute_layer(layer_name: str, target, *args, **kwargs) -> bool: """Esegue un layer e restituisce True se ok, False se fallito. @@ -326,6 +328,7 @@ def _execute_layer(layer_name: str, target, *args, **kwargs) -> bool: warn_only: gli errori di validazione diventano warning, la pipeline continua """ nonlocal run_has_validation_warnings + nonlocal validations layer_logger = bind_logger(base_logger, layer=layer_name) context.start_layer(layer_name) @@ -340,6 +343,7 @@ def _execute_layer(layer_name: str, target, *args, **kwargs) -> bool: validation_kwargs["sample_mode"] = sample_mode summary = _validation_runner(layer_name)(cfg, year, layer_logger, **validation_kwargs) context.set_validation(layer_name, summary) + validations[layer_name] = summary if not summary.get("passed", False): message = f"{layer_name.upper()} validation failed" if validation_mode == "strict" and fail_on_error: @@ -962,6 +966,7 @@ def run_full( results["steps"][str(year)] = { "run": "ok", "validate": "passed" if all_passed else "failed", + "validations": ctx.validations, } if not all_passed and fail_on_error_flag: results["status"] = "failed" diff --git a/toolkit/core/metadata.py b/toolkit/core/metadata.py index 8ca065a8..26945fa6 100644 --- a/toolkit/core/metadata.py +++ b/toolkit/core/metadata.py @@ -84,7 +84,7 @@ def read_layer_metadata(layer_dir: Path) -> dict[str, Any]: # --------------------------------------------------------------------------- -# merge_layer_manifest — merge validation fields into metadata.json +# merge_layer_manifest — merge output file info into metadata.json # --------------------------------------------------------------------------- @@ -92,28 +92,13 @@ def merge_layer_manifest( folder: Path, *, metadata_path: str = METADATA, - validation_path: str | None = None, outputs: list[dict[str, Any]] | None = None, - ok: bool | None = None, - errors_count: int | None = None, - warnings_count: int | None = None, primary_output_file: str | None = None, sources: list[Any] | None = None, ) -> Path: meta = _read_metadata(folder, metadata_path) or {} if outputs is not None: meta["outputs"] = outputs - if validation_path is not None: - meta["validation"] = validation_path - if ok is not None or errors_count is not None or warnings_count is not None: - summary: dict[str, Any] = {} - if ok is not None: - summary["ok"] = ok - if errors_count is not None: - summary["errors_count"] = errors_count - if warnings_count is not None: - summary["warnings_count"] = warnings_count - meta["summary"] = summary if primary_output_file and not meta.get("primary_output_file"): meta["primary_output_file"] = primary_output_file if sources and not meta.get("sources"): diff --git a/toolkit/core/validation.py b/toolkit/core/validation.py index b5c8d0bf..f2b2ede7 100644 --- a/toolkit/core/validation.py +++ b/toolkit/core/validation.py @@ -106,6 +106,13 @@ def build_validation_summary(result: ValidationResult) -> dict[str, Any]: } if "stats" in result.summary: out["stats"] = result.summary["stats"] + if result.sections: + out["sections"] = {} + for key, section in result.sections.items(): + if isinstance(section, dict): + out["sections"][key] = {k: v for k, v in section.items() if k != "warning_messages"} + else: + out["sections"][key] = section return out diff --git a/toolkit/domain/report.py b/toolkit/domain/report.py index 435cf218..46ff8adb 100644 --- a/toolkit/domain/report.py +++ b/toolkit/domain/report.py @@ -304,17 +304,18 @@ def build_run_report( # --- Config hash (da metadata di qualsiasi layer) --- config_hash = _collect_config_hash(root_path, dataset, year) - # --- Validation per layer --- + # --- Validation per layer (da memoria, non da disco) --- + step_val = (step_results or {}).get("validations") or {} layers_report: dict[str, Any] = {} for lname in ("raw", "clean", "mart"): - val = _read_validation(root_path, lname, dataset, year) - warnings = _get_warnings(val) - errors = _get_errors(val) + val = step_val.get(lname, {}) + warnings = val.get("warnings", []) + errors = val.get("errors", []) file_bytes = _collect_output_bytes(root_path, lname, dataset, year) layer_status = ((record or {}).get("layers") or {}).get(lname, {}).get("status") # Distingue layer non presente (es. mart-only: raw/clean assenti) da layer fallito - val_ok: bool | None = val.get("ok") + val_ok = val.get("passed") if val_ok is None and not val and layer_status is None: val_ok = None # layer non eseguito (non è un fallimento) diff --git a/toolkit/mart/run.py b/toolkit/mart/run.py index 5347735e..33105b16 100644 --- a/toolkit/mart/run.py +++ b/toolkit/mart/run.py @@ -22,7 +22,6 @@ ) from toolkit.core.multi_year_source import bind_multi_year_view, collect_multi_year_files from toolkit.core.paths import ( - MART_VALIDATION, layer_dataset_dir, layer_year_dir, resolve_root, @@ -498,7 +497,6 @@ def run_mart( merge_layer_manifest( mart_dir, metadata_path=metadata_path.name, - validation_path=MART_VALIDATION, outputs=outputs, ) total_bytes = sum(p.stat().st_size for p in written if p.exists()) diff --git a/toolkit/mart/validate.py b/toolkit/mart/validate.py index c9711937..7b63219a 100644 --- a/toolkit/mart/validate.py +++ b/toolkit/mart/validate.py @@ -13,15 +13,13 @@ ) from toolkit.core.io import read_json_or_none from toolkit.core.config import MartTableRuleConfig, MartValidationSpec -from toolkit.core.metadata import merge_layer_manifest -from toolkit.core.paths import MART_VALIDATION, METADATA, layer_year_dir, to_root_relative +from toolkit.core.paths import METADATA, layer_year_dir, to_root_relative from toolkit.core.sql_utils import sql_path from toolkit.core.validation import ( ValidationResult, build_validation_summary, check_transitions, required_columns_check, - write_validation_json, ) @@ -240,15 +238,5 @@ def run_mart_validation(cfg, year: int, logger, *, sample_mode: bool = False) -> sections={"transition": transition_report}, ) - report = write_validation_json(Path(mart_dir) / MART_VALIDATION, result) - merge_layer_manifest( - mart_dir, - metadata_path=METADATA, - validation_path="_validate/mart_validation.json", - outputs=metadata.get("outputs", []), - ok=result.ok, - errors_count=len(result.errors), - warnings_count=len(result.warnings), - ) - logger.info(f"VALIDATE MART -> {report} (ok={result.ok})") + logger.info(f"VALIDATE MART -> (ok={result.ok})") return build_validation_summary(result) diff --git a/toolkit/raw/run.py b/toolkit/raw/run.py index 95e341e4..8a84a5ae 100644 --- a/toolkit/raw/run.py +++ b/toolkit/raw/run.py @@ -11,9 +11,8 @@ sha256_bytes, write_metadata, ) -from toolkit.core.paths import RAW_VALIDATION, RAW_PROFILE, layer_year_dir, to_root_relative +from toolkit.core.paths import RAW_PROFILE, layer_year_dir, to_root_relative from toolkit.core.registry import register_builtin_plugins -from toolkit.core.validation import write_validation_json from toolkit.profile.raw import ( sniff_source_file, profile_raw, @@ -29,7 +28,6 @@ _resolve_output_path, ) from toolkit.raw.extractors import get_extractor -from toolkit.raw.validate import validate_raw_output def run_raw( @@ -210,18 +208,12 @@ def run_raw( metadata_payload["source_id"] = source_id write_metadata(out_dir, metadata_payload) - # --- QA RAW --- - result = validate_raw_output(out_dir, files_written) - vpath = write_validation_json(out_dir / RAW_VALIDATION, result) + # Track output files in metadata (validation is done by run_raw_validation) merge_layer_manifest( out_dir, - validation_path=vpath.name, outputs=[ {"file": f["file"], "sha256": f["sha256"], "bytes": f["bytes"]} for f in files_written ], - ok=result.ok, - errors_count=len(result.errors), - warnings_count=len(result.warnings), primary_output_file=primary_output_file, sources=[ {"name": entry["name"], "output_file": entry["output_file"]} @@ -229,21 +221,6 @@ def run_raw( ], ) - if result.warnings: - logger.warning( - f"RAW QA warnings ({dataset} {year}): {len(result.warnings)} -> {vpath.name}" - ) - for w in result.warnings[:10]: - logger.warning(f" - {w}") - - if not result.ok: - logger.error(f"RAW QA FAILED ({dataset} {year}) -> {vpath}") - for e in result.errors[:20]: - logger.error(f" - {e}") - raise RuntimeError(f"RAW validation failed for {dataset} {year}. See {vpath}") - else: - logger.info(f"RAW QA OK ({dataset} {year}) -> {vpath.name}") - output_bytes = sum(f.get("bytes", 0) for f in files_written) if files_written else None source_urls = list( dict.fromkeys( diff --git a/toolkit/raw/validate.py b/toolkit/raw/validate.py index c5ca299d..16765d5e 100644 --- a/toolkit/raw/validate.py +++ b/toolkit/raw/validate.py @@ -4,12 +4,10 @@ from typing import Any from toolkit.core.io import read_json_or_none -from toolkit.core.metadata import merge_layer_manifest -from toolkit.core.paths import METADATA, RAW_VALIDATION, layer_year_dir +from toolkit.core.paths import METADATA, layer_year_dir from toolkit.core.validation import ( ValidationResult, build_validation_summary, - write_validation_json, ) TEXT_EXT = {".csv", ".txt", ".tsv", ".json", ".xml", ".html"} @@ -121,21 +119,12 @@ def run_raw_validation(root: str | None, dataset: str, year: int, logger) -> dic **({"delim": hints["delim_suggested"]} if hints.get("delim_suggested") else {}), **result.summary, } - result = result.__class__( + result = ValidationResult( ok=result.ok, errors=result.errors, warnings=result.warnings, summary=enriched_summary, sections=result.sections, ) - report = write_validation_json(out_dir / RAW_VALIDATION, result) - merge_layer_manifest( - out_dir, - validation_path=report.name, - outputs=metadata.get("outputs", []), - ok=result.ok, - errors_count=len(result.errors), - warnings_count=len(result.warnings), - ) - logger.info(f"VALIDATE RAW -> {report} (ok={result.ok})") + logger.info(f"VALIDATE RAW -> (ok={result.ok})") return build_validation_summary(result) From ef8efd1906d2c56835229d9a68e5ab9a66d9101d Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:42:50 +0100 Subject: [PATCH 2/6] Passo 5-6: pulizia constanti validation e dead code - Rimosse _read_validation(), _get_warnings(), _get_errors() da report.py - Puliti import validation costanti non piu' necessari --- datasets/popolazione-istat-comunale/README.md | 62 ++++++++++++++++++ .../popolazione-istat-comunale/dataset.yml | 65 +++++++++++++++++++ datasets/popolazione-istat-comunale/notes.md | 38 +++++++++++ .../popolazione-istat-comunale/sql/clean.sql | 32 +++++++++ .../popolazione-istat-comunale/sql/mart.sql | 11 ++++ .../sql/mart_by_eta.sql | 13 ++++ .../sql/mart_by_provincia.sql | 48 ++++++++++++++ .../sql/mart_by_regione.sql | 45 +++++++++++++ .../sql/mart_indicatori.sql | 58 +++++++++++++++++ toolkit/domain/report.py | 35 ---------- 10 files changed, 372 insertions(+), 35 deletions(-) create mode 100644 datasets/popolazione-istat-comunale/README.md create mode 100644 datasets/popolazione-istat-comunale/dataset.yml create mode 100644 datasets/popolazione-istat-comunale/notes.md create mode 100644 datasets/popolazione-istat-comunale/sql/clean.sql create mode 100644 datasets/popolazione-istat-comunale/sql/mart.sql create mode 100644 datasets/popolazione-istat-comunale/sql/mart_by_eta.sql create mode 100644 datasets/popolazione-istat-comunale/sql/mart_by_provincia.sql create mode 100644 datasets/popolazione-istat-comunale/sql/mart_by_regione.sql create mode 100644 datasets/popolazione-istat-comunale/sql/mart_indicatori.sql diff --git a/datasets/popolazione-istat-comunale/README.md b/datasets/popolazione-istat-comunale/README.md new file mode 100644 index 00000000..f92a16ea --- /dev/null +++ b/datasets/popolazione-istat-comunale/README.md @@ -0,0 +1,62 @@ +# Popolazione ISTAT comunale + +## Tipo + +**Support dataset** — base infrastrutturale per join, coverage check, indicatori pro capite. + +## Fonte + +ISTAT — demo.posas `POSAS_{year}_it_Comuni.zip` +URL: `https://demo.istat.it/data/posas/POSAS_{year}_it_Comuni.zip` +Formato: ZIP → CSV, `;` delim, `utf-8` con BOM. Skip 1 riga (header doppio). + +## Schema + +**Clean**: 22 colonne — rinomine a snake_case, più colonna calcolata `fascia_eta`. Escluse le righe con ETA=999 (totali pre-aggregati — somma ridondante delle età 0-100). Utilizza le macro standard del toolkit (`cast_int`, `normalize_string`). + +**Mart `popolazione_by_comune`**: `[anno_riferimento, codice_comune, comune, popolazione_residente, popolazione_maschile, popolazione_femminile]` — un record per comune (SUM GROUP BY). + +**Mart `popolazione_by_eta`**: `[anno_riferimento, codice_comune, comune, eta, popolazione_residente, popolazione_maschile, popolazione_femminile]` — un record per comune per classe di età (0-100). + +**Mart `popolazione_by_regione`**: `[anno_riferimento, regione, popolazione_residente, popolazione_maschile, popolazione_femminile, numero_comuni, superficie_km2, densita_ab_km2]` — aggregazione regionale con densità. JOIN con `comuni-master` per codice ISTAT e denominazione. + +**Mart `popolazione_by_provincia`**: `[anno_riferimento, sigla_provincia, provincia, regione, popolazione_residente, popolazione_maschile, popolazione_femminile, numero_comuni, superficie_km2, densita_ab_km2]` — aggregazione provinciale con densità. + +**Mart `popolazione_indicatori`**: `[anno_riferimento, codice_comune, comune, popolazione_totale, pop_0_14, pop_15_64, pop_65_plus, pop_under_18, pop_75_plus, indice_vecchiaia, rapporto_dipendenza, pct_under_18, pct_over_65, pct_over_75]` — indicatori demografici per comune. + +**Hierarchy `h_fascia`**: `[codice_comune, comune, fascia_eta]` + 17 metriche demografiche — un record per comune per fascia d'età (0-14, 15-29, 30-44, 45-59, 60-74, 75+). Generato automaticamente dalla mart hierarchy del toolkit. + +Chiave di join: `codice_comune` + +## Copertura + +| Anno | Comuni | Popolazione | +|---|---|---| +| 2019 | 7.954 | 59.816.673 | +| 2020 | 7.914 | 59.641.488 | +| 2021 | 7.903 | 59.236.213 | +| 2022 | 7.904 | 59.030.133 | +| 2023 | 7.901 | 58.997.201 | +| 2024 | 7.900 | 58.971.230 | +| 2025 | 7.896 | 58.943.464 | + +*La variazione 7.954 → 7.896 riflette fusioni comunali nel periodo — non un errore. Vedi notes.md §Cautele.* + +## QC + +- Clean filtra ETA=999 (totali pre-aggregati — somma ridondante) ✅ +- Maschi + Femmine = Popolazione su tutti gli anni ✅ +- Nessun comune senza nome o con pop=0 ✅ +- by_comune vs sum(by_eta): differenza = 0 ✅ +- hierarchy h_fascia vs mart by_comune: ratio 1.0000 su popolazione_residente ✅ + +## Uso + +- join con dataset comunali (es. IRPEF) +- controlli coverage territoriale +- indicatori pro capite +- detector anomalie territoriali + +## Stato + +`runnable` — tutti e 7 gli anni con raw, clean, mart verificati. diff --git a/datasets/popolazione-istat-comunale/dataset.yml b/datasets/popolazione-istat-comunale/dataset.yml new file mode 100644 index 00000000..d291eee1 --- /dev/null +++ b/datasets/popolazione-istat-comunale/dataset.yml @@ -0,0 +1,65 @@ +root: "../.." +schema_version: 1 + +dataset: + name: "popolazione_istat_comunale" + source_id: "istat_sdmx" + years: [2019, 2020, 2021, 2022, 2023, 2024, 2025] + +raw: + extractor: + type: "unzip_first_csv" + sources: + - name: "istat_posas_http_zip" + type: "http_file" + args: + url: "https://demo.istat.it/data/posas/POSAS_{year}_it_Comuni.zip" + filename: "POSAS_{year}_it_Comuni.zip" + primary: true + +clean: + sql: "sql/clean.sql" + read: + source: config_only + mode: latest + header: true + skip: 1 + delim: ";" + encoding: "utf-8" + trim_whitespace: true + required_columns: + - "codice_comune" + - "comune" + - "eta" + - "popolazione_residente" + validate: + not_null: + - "codice_comune" + - "comune" + - "eta" + min_rows: 1000 + +mart: + tables: + - name: "popolazione_by_comune" + sql: "sql/mart.sql" + - name: "popolazione_by_eta" + sql: "sql/mart_by_eta.sql" + - name: "popolazione_by_regione" + sql: "sql/mart_by_regione.sql" + - name: "popolazione_by_provincia" + sql: "sql/mart_by_provincia.sql" + - name: "popolazione_indicatori" + sql: "sql/mart_indicatori.sql" + hierarchy: + axis: categorico + levels: + - level: fascia_eta + table: h_fascia + grain: [codice_comune, comune, fascia_eta] + exclude_metrics: [anno, eta] + required_tables: ["popolazione_by_comune", "popolazione_by_eta"] + validate: {} + +validation: + fail_on_error: true diff --git a/datasets/popolazione-istat-comunale/notes.md b/datasets/popolazione-istat-comunale/notes.md new file mode 100644 index 00000000..55b32eb1 --- /dev/null +++ b/datasets/popolazione-istat-comunale/notes.md @@ -0,0 +1,38 @@ +# Notes — popolazione-istat-comunale-2019-2025 + +## Support dataset + +Base infrastrutturale per join e coverage check. Non è un filone narrativo autonomo. + +## Fonte + +- ISTAT POSAS `POSAS_{year}_it_Comuni.zip` +- URL: `https://demo.istat.it/data/posas/POSAS_{year}_it_Comuni.zip` +- Formato: ZIP → CSV, `;` delim, `utf-8` con BOM, skip 1 riga +- Chiave: `codice_comune` + +## Run completati (2026-04-27) + +Tutti e 7 gli anni: raw ✅ clean ✅ mart ✅ + +2020-2023 erano PENDING (clean/mart non partiti per timeout). Rilanciati oggi — nessun problema di schema. + +## Schema + +- Clean: 22 colonne, colonna calcolata `fascia_eta`, filtrato ETA=999 (totali ridondanti). Migrato alle macro toolkit standard (cast_int, normalize_string) il 2026-07-25. +- `popolazione_by_comune`: 1 riga per comune, SUM GROUP BY +- `popolazione_by_eta`: 1 riga per comune per classe di età (ETA 0-100) +- `popolazione_by_regione`: aggregazione regionale con densità (JOIN comuni-master per codice + denominazione) +- `popolazione_by_provincia`: aggregazione provinciale con densità (JOIN comuni-master per codice + denominazione) +- `popolazione_indicatori`: indicatori demografici per comune (indice vecchiaia, rapporto dipendenza, % under_18, % over_65, % over_75) +- `h_fascia` (hierarchy): 1 riga per comune per fascia d'età, 17 metriche — generato automaticamente dal toolkit + +## Cautele + +- `comune` è descrittivo — usare `codice_comune` come chiave +- Variazioni territoriali / fusioni di comuni nel periodo possono creare mismatches nei join +- Non assumere schema perfettamente stabile senza verifica su ogni anno + +## IRPEF join test + +Con IRPEF 2023: ~7892 match su ~7897 comuni (99,9%). I pochi non-match confermano il dataset come detector di anomalie territoriali. diff --git a/datasets/popolazione-istat-comunale/sql/clean.sql b/datasets/popolazione-istat-comunale/sql/clean.sql new file mode 100644 index 00000000..3d471609 --- /dev/null +++ b/datasets/popolazione-istat-comunale/sql/clean.sql @@ -0,0 +1,32 @@ +select + {year}::INTEGER as anno, + normalize_string("Codice comune") as codice_comune, + normalize_string("Comune") as comune, + cast_int("Età") as eta, + case + when cast_int("Età") between 0 and 14 then '0-14' + when cast_int("Età") between 15 and 29 then '15-29' + when cast_int("Età") between 30 and 44 then '30-44' + when cast_int("Età") between 45 and 59 then '45-59' + when cast_int("Età") between 60 and 74 then '60-74' + when cast_int("Età") >= 75 then '75+' + end as fascia_eta, + cast_int("Celibi") as celibi, + cast_int("Coniugati") as coniugati, + cast_int("Divorziati") as divorziati, + cast_int("Vedovi") as vedovi, + cast_int("Uniti civilmente") as uniti_civilmente_maschi, + cast_int("Maschi già in unione civile (per scioglimento unione)") as maschi_gia_unione_civile_scioglimento, + cast_int("Maschi già in unione civile (per decesso del partner)") as maschi_gia_unione_civile_decesso, + cast_int("Totale maschi") as totale_maschi, + cast_int("Nubili") as nubili, + cast_int("Coniugate") as coniugate, + cast_int("Divorziate") as divorziate, + cast_int("Vedove") as vedove, + cast_int("Unite civilmente") as unite_civilmente_femmine, + cast_int("Femmine già in unione civile (per scioglimento unione)") as femmine_gia_unione_civile_scioglimento, + cast_int("Femmine già in unione civile (per decesso del partner)") as femmine_gia_unione_civile_decesso, + cast_int("Totale femmine") as totale_femmine, + cast_int("Totale") as popolazione_residente +from raw_input +where cast_int("Età") <> 999 diff --git a/datasets/popolazione-istat-comunale/sql/mart.sql b/datasets/popolazione-istat-comunale/sql/mart.sql new file mode 100644 index 00000000..bc22d053 --- /dev/null +++ b/datasets/popolazione-istat-comunale/sql/mart.sql @@ -0,0 +1,11 @@ +select + {year} as anno_riferimento, + codice_comune, + comune, + sum(popolazione_residente) as popolazione_residente, + sum(totale_maschi) as popolazione_maschile, + sum(totale_femmine) as popolazione_femminile +from clean_input +where codice_comune is not null + and comune is not null +group by codice_comune, comune diff --git a/datasets/popolazione-istat-comunale/sql/mart_by_eta.sql b/datasets/popolazione-istat-comunale/sql/mart_by_eta.sql new file mode 100644 index 00000000..869dc762 --- /dev/null +++ b/datasets/popolazione-istat-comunale/sql/mart_by_eta.sql @@ -0,0 +1,13 @@ +select + {year} as anno_riferimento, + codice_comune, + comune, + eta, + popolazione_residente, + totale_maschi as popolazione_maschile, + totale_femmine as popolazione_femminile +from clean_input +where codice_comune is not null + and comune is not null + and eta is not null + and eta <> 999 diff --git a/datasets/popolazione-istat-comunale/sql/mart_by_provincia.sql b/datasets/popolazione-istat-comunale/sql/mart_by_provincia.sql new file mode 100644 index 00000000..903f14a2 --- /dev/null +++ b/datasets/popolazione-istat-comunale/sql/mart_by_provincia.sql @@ -0,0 +1,48 @@ +with +comuni_registry as ( + select codice_istat, sigla_provincia, provincia, regione, denominazione, superficie_km2 + from read_parquet('https://storage.googleapis.com/dataciviclab-clean/comuni_master/2026/comuni_master_2026_clean.parquet') +), +provincia_map as ( + select distinct left(codice_istat, 3) as codice_provincia, regione + from comuni_registry + where codice_istat is not null +), +per_comune as ( + select + codice_comune, + comune, + sum(popolazione_residente) as pop_residente, + sum(totale_maschi) as pop_maschile, + sum(totale_femmine) as pop_femminile + from clean_input + where codice_comune is not null + group by codice_comune, comune +), +per_comune_arricchito as ( + select + pc.*, + -- Match: 1) per codice ISTAT, 2) per denominazione, 3) fallback + coalesce(r1.sigla_provincia, r2.sigla_provincia, left(pc.codice_comune, 3)) as sigla_provincia, + coalesce(r1.provincia, r2.provincia, left(pc.codice_comune, 3)) as provincia, + coalesce(r1.regione, r2.regione, pr.regione, 'Altro') as regione, + coalesce(r1.superficie_km2, r2.superficie_km2, 0.0) as superficie_km2 + from per_comune pc + left join comuni_registry r1 on pc.codice_comune = r1.codice_istat + left join comuni_registry r2 on lower(trim(pc.comune)) = lower(trim(r2.denominazione)) + left join provincia_map pr on left(pc.codice_comune, 3) = pr.codice_provincia +) +select + {year} as anno_riferimento, + sigla_provincia, + provincia, + regione, + sum(pop_residente) as popolazione_residente, + sum(pop_maschile) as popolazione_maschile, + sum(pop_femminile) as popolazione_femminile, + count(*) as numero_comuni, + round(sum(superficie_km2), 2) as superficie_km2, + round(sum(pop_residente)::double / nullif(sum(superficie_km2), 0), 1) as densita_ab_km2 +from per_comune_arricchito +group by sigla_provincia, provincia, regione +order by sigla_provincia diff --git a/datasets/popolazione-istat-comunale/sql/mart_by_regione.sql b/datasets/popolazione-istat-comunale/sql/mart_by_regione.sql new file mode 100644 index 00000000..84ba1836 --- /dev/null +++ b/datasets/popolazione-istat-comunale/sql/mart_by_regione.sql @@ -0,0 +1,45 @@ +with +comuni_registry as ( + select codice_istat, regione, denominazione, superficie_km2 + from read_parquet('https://storage.googleapis.com/dataciviclab-clean/comuni_master/2026/comuni_master_2026_clean.parquet') +), +-- Mappa provincia (prime 3 cifre) -> regione, per fallback su comuni non in registry +provincia_map as ( + select distinct left(codice_istat, 3) as codice_provincia, regione + from comuni_registry + where codice_istat is not null +), +per_comune as ( + select + codice_comune, + comune, + sum(popolazione_residente) as pop_residente, + sum(totale_maschi) as pop_maschile, + sum(totale_femmine) as pop_femminile + from clean_input + where codice_comune is not null + group by codice_comune, comune +), +per_comune_arricchito as ( + select + pc.*, + -- Match: 1) per codice ISTAT, 2) per denominazione, 3) per provincia + coalesce(r1.regione, r2.regione, pr.regione, 'Altro') as regione, + coalesce(r1.superficie_km2, r2.superficie_km2, 0.0) as superficie_km2 + from per_comune pc + left join comuni_registry r1 on pc.codice_comune = r1.codice_istat + left join comuni_registry r2 on lower(trim(pc.comune)) = lower(trim(r2.denominazione)) + left join provincia_map pr on left(pc.codice_comune, 3) = pr.codice_provincia +) +select + {year} as anno_riferimento, + regione, + sum(pop_residente) as popolazione_residente, + sum(pop_maschile) as popolazione_maschile, + sum(pop_femminile) as popolazione_femminile, + count(*) as numero_comuni, + round(sum(superficie_km2), 2) as superficie_km2, + round(sum(pop_residente)::double / nullif(sum(superficie_km2), 0), 1) as densita_ab_km2 +from per_comune_arricchito +group by regione +order by regione diff --git a/datasets/popolazione-istat-comunale/sql/mart_indicatori.sql b/datasets/popolazione-istat-comunale/sql/mart_indicatori.sql new file mode 100644 index 00000000..af9509a5 --- /dev/null +++ b/datasets/popolazione-istat-comunale/sql/mart_indicatori.sql @@ -0,0 +1,58 @@ +with +base as ( + select + codice_comune, + comune, + eta, + popolazione_residente + from clean_input + where codice_comune is not null + and comune is not null + and eta is not null + and eta <> 999 +), +fasce as ( + select + codice_comune, + comune, + sum(case when eta between 0 and 14 then popolazione_residente else 0 end) as pop_0_14, + sum(case when eta between 15 and 64 then popolazione_residente else 0 end) as pop_15_64, + sum(case when eta >= 65 then popolazione_residente else 0 end) as pop_65_plus, + sum(case when eta < 18 then popolazione_residente else 0 end) as pop_under_18, + sum(case when eta >= 75 then popolazione_residente else 0 end) as pop_75_plus, + sum(popolazione_residente) as popolazione_totale + from base + group by codice_comune, comune +) +select + {year} as anno_riferimento, + codice_comune, + comune, + popolazione_totale, + pop_0_14, + pop_15_64, + pop_65_plus, + pop_under_18, + pop_75_plus, + -- Indice di vecchiaia: pop 65+ / pop 0-14 * 100 + case when pop_0_14 > 0 + then round((pop_65_plus::double / pop_0_14::double) * 100, 1) + end as indice_vecchiaia, + -- Rapporto di dipendenza: (pop 0-14 + pop 65+) / pop 15-64 * 100 + case when pop_15_64 > 0 + then round(((pop_0_14 + pop_65_plus)::double / pop_15_64::double) * 100, 1) + end as rapporto_dipendenza, + -- Percentuale under 18 + case when popolazione_totale > 0 + then round((pop_under_18::double / popolazione_totale::double) * 100, 1) + end as pct_under_18, + -- Percentuale over 65 + case when popolazione_totale > 0 + then round((pop_65_plus::double / popolazione_totale::double) * 100, 1) + end as pct_over_65, + -- Percentuale over 75 + case when popolazione_totale > 0 + then round((pop_75_plus::double / popolazione_totale::double) * 100, 1) + end as pct_over_75 +from fasce +where popolazione_totale > 0 diff --git a/toolkit/domain/report.py b/toolkit/domain/report.py index 46ff8adb..0302ad01 100644 --- a/toolkit/domain/report.py +++ b/toolkit/domain/report.py @@ -13,12 +13,9 @@ from toolkit.core.io import read_json_or_none, write_json_atomic from toolkit.core.metadata import read_layer_metadata from toolkit.core.paths import ( - CLEAN_VALIDATION, - MART_VALIDATION, METADATA, RAW_PROFILE, RAW_PROFILE_DIR, - RAW_VALIDATION, layer_year_dir, ) from toolkit.core.run_records import get_run_dir, latest_run @@ -28,38 +25,6 @@ _DATASET_README = "README.md" -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - - -def _read_validation(root: Path, layer: str, dataset: str, year: int) -> dict[str, Any]: - """Legge il validation JSON di un layer, restituendo dict vuoto se non presente.""" - val_map = { - "raw": RAW_VALIDATION, - "clean": CLEAN_VALIDATION, - "mart": MART_VALIDATION, - } - val_name = val_map.get(layer) - if not val_name: - return {} - val_path = layer_year_dir(root, layer, dataset, year) / val_name - if val_path.exists(): - return read_json_or_none(val_path) or {} - val_path2 = layer_year_dir(root, layer, dataset, year) / "_validate" / val_name - if val_path2.exists(): - return read_json_or_none(val_path2) or {} - return {} - - -def _get_warnings(validation: dict[str, Any]) -> list[str]: - return validation.get("warnings", []) - - -def _get_errors(validation: dict[str, Any]) -> list[str]: - return validation.get("errors", []) - - def _get_run_record(root: Path, dataset: str, year: int) -> dict[str, Any] | None: """Legge il run record più recente per un dataset/anno.""" run_dir = get_run_dir(root, dataset, year) From c2d8dce978507668d080a6025973e84754005c44 Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:42:57 +0100 Subject: [PATCH 3/6] chore: rimuovi datasets/ dal tracking --- datasets/popolazione-istat-comunale/README.md | 62 ------------------ .../popolazione-istat-comunale/dataset.yml | 65 ------------------- datasets/popolazione-istat-comunale/notes.md | 38 ----------- .../popolazione-istat-comunale/sql/clean.sql | 32 --------- .../popolazione-istat-comunale/sql/mart.sql | 11 ---- .../sql/mart_by_eta.sql | 13 ---- .../sql/mart_by_provincia.sql | 48 -------------- .../sql/mart_by_regione.sql | 45 ------------- .../sql/mart_indicatori.sql | 58 ----------------- 9 files changed, 372 deletions(-) delete mode 100644 datasets/popolazione-istat-comunale/README.md delete mode 100644 datasets/popolazione-istat-comunale/dataset.yml delete mode 100644 datasets/popolazione-istat-comunale/notes.md delete mode 100644 datasets/popolazione-istat-comunale/sql/clean.sql delete mode 100644 datasets/popolazione-istat-comunale/sql/mart.sql delete mode 100644 datasets/popolazione-istat-comunale/sql/mart_by_eta.sql delete mode 100644 datasets/popolazione-istat-comunale/sql/mart_by_provincia.sql delete mode 100644 datasets/popolazione-istat-comunale/sql/mart_by_regione.sql delete mode 100644 datasets/popolazione-istat-comunale/sql/mart_indicatori.sql diff --git a/datasets/popolazione-istat-comunale/README.md b/datasets/popolazione-istat-comunale/README.md deleted file mode 100644 index f92a16ea..00000000 --- a/datasets/popolazione-istat-comunale/README.md +++ /dev/null @@ -1,62 +0,0 @@ -# Popolazione ISTAT comunale - -## Tipo - -**Support dataset** — base infrastrutturale per join, coverage check, indicatori pro capite. - -## Fonte - -ISTAT — demo.posas `POSAS_{year}_it_Comuni.zip` -URL: `https://demo.istat.it/data/posas/POSAS_{year}_it_Comuni.zip` -Formato: ZIP → CSV, `;` delim, `utf-8` con BOM. Skip 1 riga (header doppio). - -## Schema - -**Clean**: 22 colonne — rinomine a snake_case, più colonna calcolata `fascia_eta`. Escluse le righe con ETA=999 (totali pre-aggregati — somma ridondante delle età 0-100). Utilizza le macro standard del toolkit (`cast_int`, `normalize_string`). - -**Mart `popolazione_by_comune`**: `[anno_riferimento, codice_comune, comune, popolazione_residente, popolazione_maschile, popolazione_femminile]` — un record per comune (SUM GROUP BY). - -**Mart `popolazione_by_eta`**: `[anno_riferimento, codice_comune, comune, eta, popolazione_residente, popolazione_maschile, popolazione_femminile]` — un record per comune per classe di età (0-100). - -**Mart `popolazione_by_regione`**: `[anno_riferimento, regione, popolazione_residente, popolazione_maschile, popolazione_femminile, numero_comuni, superficie_km2, densita_ab_km2]` — aggregazione regionale con densità. JOIN con `comuni-master` per codice ISTAT e denominazione. - -**Mart `popolazione_by_provincia`**: `[anno_riferimento, sigla_provincia, provincia, regione, popolazione_residente, popolazione_maschile, popolazione_femminile, numero_comuni, superficie_km2, densita_ab_km2]` — aggregazione provinciale con densità. - -**Mart `popolazione_indicatori`**: `[anno_riferimento, codice_comune, comune, popolazione_totale, pop_0_14, pop_15_64, pop_65_plus, pop_under_18, pop_75_plus, indice_vecchiaia, rapporto_dipendenza, pct_under_18, pct_over_65, pct_over_75]` — indicatori demografici per comune. - -**Hierarchy `h_fascia`**: `[codice_comune, comune, fascia_eta]` + 17 metriche demografiche — un record per comune per fascia d'età (0-14, 15-29, 30-44, 45-59, 60-74, 75+). Generato automaticamente dalla mart hierarchy del toolkit. - -Chiave di join: `codice_comune` - -## Copertura - -| Anno | Comuni | Popolazione | -|---|---|---| -| 2019 | 7.954 | 59.816.673 | -| 2020 | 7.914 | 59.641.488 | -| 2021 | 7.903 | 59.236.213 | -| 2022 | 7.904 | 59.030.133 | -| 2023 | 7.901 | 58.997.201 | -| 2024 | 7.900 | 58.971.230 | -| 2025 | 7.896 | 58.943.464 | - -*La variazione 7.954 → 7.896 riflette fusioni comunali nel periodo — non un errore. Vedi notes.md §Cautele.* - -## QC - -- Clean filtra ETA=999 (totali pre-aggregati — somma ridondante) ✅ -- Maschi + Femmine = Popolazione su tutti gli anni ✅ -- Nessun comune senza nome o con pop=0 ✅ -- by_comune vs sum(by_eta): differenza = 0 ✅ -- hierarchy h_fascia vs mart by_comune: ratio 1.0000 su popolazione_residente ✅ - -## Uso - -- join con dataset comunali (es. IRPEF) -- controlli coverage territoriale -- indicatori pro capite -- detector anomalie territoriali - -## Stato - -`runnable` — tutti e 7 gli anni con raw, clean, mart verificati. diff --git a/datasets/popolazione-istat-comunale/dataset.yml b/datasets/popolazione-istat-comunale/dataset.yml deleted file mode 100644 index d291eee1..00000000 --- a/datasets/popolazione-istat-comunale/dataset.yml +++ /dev/null @@ -1,65 +0,0 @@ -root: "../.." -schema_version: 1 - -dataset: - name: "popolazione_istat_comunale" - source_id: "istat_sdmx" - years: [2019, 2020, 2021, 2022, 2023, 2024, 2025] - -raw: - extractor: - type: "unzip_first_csv" - sources: - - name: "istat_posas_http_zip" - type: "http_file" - args: - url: "https://demo.istat.it/data/posas/POSAS_{year}_it_Comuni.zip" - filename: "POSAS_{year}_it_Comuni.zip" - primary: true - -clean: - sql: "sql/clean.sql" - read: - source: config_only - mode: latest - header: true - skip: 1 - delim: ";" - encoding: "utf-8" - trim_whitespace: true - required_columns: - - "codice_comune" - - "comune" - - "eta" - - "popolazione_residente" - validate: - not_null: - - "codice_comune" - - "comune" - - "eta" - min_rows: 1000 - -mart: - tables: - - name: "popolazione_by_comune" - sql: "sql/mart.sql" - - name: "popolazione_by_eta" - sql: "sql/mart_by_eta.sql" - - name: "popolazione_by_regione" - sql: "sql/mart_by_regione.sql" - - name: "popolazione_by_provincia" - sql: "sql/mart_by_provincia.sql" - - name: "popolazione_indicatori" - sql: "sql/mart_indicatori.sql" - hierarchy: - axis: categorico - levels: - - level: fascia_eta - table: h_fascia - grain: [codice_comune, comune, fascia_eta] - exclude_metrics: [anno, eta] - required_tables: ["popolazione_by_comune", "popolazione_by_eta"] - validate: {} - -validation: - fail_on_error: true diff --git a/datasets/popolazione-istat-comunale/notes.md b/datasets/popolazione-istat-comunale/notes.md deleted file mode 100644 index 55b32eb1..00000000 --- a/datasets/popolazione-istat-comunale/notes.md +++ /dev/null @@ -1,38 +0,0 @@ -# Notes — popolazione-istat-comunale-2019-2025 - -## Support dataset - -Base infrastrutturale per join e coverage check. Non è un filone narrativo autonomo. - -## Fonte - -- ISTAT POSAS `POSAS_{year}_it_Comuni.zip` -- URL: `https://demo.istat.it/data/posas/POSAS_{year}_it_Comuni.zip` -- Formato: ZIP → CSV, `;` delim, `utf-8` con BOM, skip 1 riga -- Chiave: `codice_comune` - -## Run completati (2026-04-27) - -Tutti e 7 gli anni: raw ✅ clean ✅ mart ✅ - -2020-2023 erano PENDING (clean/mart non partiti per timeout). Rilanciati oggi — nessun problema di schema. - -## Schema - -- Clean: 22 colonne, colonna calcolata `fascia_eta`, filtrato ETA=999 (totali ridondanti). Migrato alle macro toolkit standard (cast_int, normalize_string) il 2026-07-25. -- `popolazione_by_comune`: 1 riga per comune, SUM GROUP BY -- `popolazione_by_eta`: 1 riga per comune per classe di età (ETA 0-100) -- `popolazione_by_regione`: aggregazione regionale con densità (JOIN comuni-master per codice + denominazione) -- `popolazione_by_provincia`: aggregazione provinciale con densità (JOIN comuni-master per codice + denominazione) -- `popolazione_indicatori`: indicatori demografici per comune (indice vecchiaia, rapporto dipendenza, % under_18, % over_65, % over_75) -- `h_fascia` (hierarchy): 1 riga per comune per fascia d'età, 17 metriche — generato automaticamente dal toolkit - -## Cautele - -- `comune` è descrittivo — usare `codice_comune` come chiave -- Variazioni territoriali / fusioni di comuni nel periodo possono creare mismatches nei join -- Non assumere schema perfettamente stabile senza verifica su ogni anno - -## IRPEF join test - -Con IRPEF 2023: ~7892 match su ~7897 comuni (99,9%). I pochi non-match confermano il dataset come detector di anomalie territoriali. diff --git a/datasets/popolazione-istat-comunale/sql/clean.sql b/datasets/popolazione-istat-comunale/sql/clean.sql deleted file mode 100644 index 3d471609..00000000 --- a/datasets/popolazione-istat-comunale/sql/clean.sql +++ /dev/null @@ -1,32 +0,0 @@ -select - {year}::INTEGER as anno, - normalize_string("Codice comune") as codice_comune, - normalize_string("Comune") as comune, - cast_int("Età") as eta, - case - when cast_int("Età") between 0 and 14 then '0-14' - when cast_int("Età") between 15 and 29 then '15-29' - when cast_int("Età") between 30 and 44 then '30-44' - when cast_int("Età") between 45 and 59 then '45-59' - when cast_int("Età") between 60 and 74 then '60-74' - when cast_int("Età") >= 75 then '75+' - end as fascia_eta, - cast_int("Celibi") as celibi, - cast_int("Coniugati") as coniugati, - cast_int("Divorziati") as divorziati, - cast_int("Vedovi") as vedovi, - cast_int("Uniti civilmente") as uniti_civilmente_maschi, - cast_int("Maschi già in unione civile (per scioglimento unione)") as maschi_gia_unione_civile_scioglimento, - cast_int("Maschi già in unione civile (per decesso del partner)") as maschi_gia_unione_civile_decesso, - cast_int("Totale maschi") as totale_maschi, - cast_int("Nubili") as nubili, - cast_int("Coniugate") as coniugate, - cast_int("Divorziate") as divorziate, - cast_int("Vedove") as vedove, - cast_int("Unite civilmente") as unite_civilmente_femmine, - cast_int("Femmine già in unione civile (per scioglimento unione)") as femmine_gia_unione_civile_scioglimento, - cast_int("Femmine già in unione civile (per decesso del partner)") as femmine_gia_unione_civile_decesso, - cast_int("Totale femmine") as totale_femmine, - cast_int("Totale") as popolazione_residente -from raw_input -where cast_int("Età") <> 999 diff --git a/datasets/popolazione-istat-comunale/sql/mart.sql b/datasets/popolazione-istat-comunale/sql/mart.sql deleted file mode 100644 index bc22d053..00000000 --- a/datasets/popolazione-istat-comunale/sql/mart.sql +++ /dev/null @@ -1,11 +0,0 @@ -select - {year} as anno_riferimento, - codice_comune, - comune, - sum(popolazione_residente) as popolazione_residente, - sum(totale_maschi) as popolazione_maschile, - sum(totale_femmine) as popolazione_femminile -from clean_input -where codice_comune is not null - and comune is not null -group by codice_comune, comune diff --git a/datasets/popolazione-istat-comunale/sql/mart_by_eta.sql b/datasets/popolazione-istat-comunale/sql/mart_by_eta.sql deleted file mode 100644 index 869dc762..00000000 --- a/datasets/popolazione-istat-comunale/sql/mart_by_eta.sql +++ /dev/null @@ -1,13 +0,0 @@ -select - {year} as anno_riferimento, - codice_comune, - comune, - eta, - popolazione_residente, - totale_maschi as popolazione_maschile, - totale_femmine as popolazione_femminile -from clean_input -where codice_comune is not null - and comune is not null - and eta is not null - and eta <> 999 diff --git a/datasets/popolazione-istat-comunale/sql/mart_by_provincia.sql b/datasets/popolazione-istat-comunale/sql/mart_by_provincia.sql deleted file mode 100644 index 903f14a2..00000000 --- a/datasets/popolazione-istat-comunale/sql/mart_by_provincia.sql +++ /dev/null @@ -1,48 +0,0 @@ -with -comuni_registry as ( - select codice_istat, sigla_provincia, provincia, regione, denominazione, superficie_km2 - from read_parquet('https://storage.googleapis.com/dataciviclab-clean/comuni_master/2026/comuni_master_2026_clean.parquet') -), -provincia_map as ( - select distinct left(codice_istat, 3) as codice_provincia, regione - from comuni_registry - where codice_istat is not null -), -per_comune as ( - select - codice_comune, - comune, - sum(popolazione_residente) as pop_residente, - sum(totale_maschi) as pop_maschile, - sum(totale_femmine) as pop_femminile - from clean_input - where codice_comune is not null - group by codice_comune, comune -), -per_comune_arricchito as ( - select - pc.*, - -- Match: 1) per codice ISTAT, 2) per denominazione, 3) fallback - coalesce(r1.sigla_provincia, r2.sigla_provincia, left(pc.codice_comune, 3)) as sigla_provincia, - coalesce(r1.provincia, r2.provincia, left(pc.codice_comune, 3)) as provincia, - coalesce(r1.regione, r2.regione, pr.regione, 'Altro') as regione, - coalesce(r1.superficie_km2, r2.superficie_km2, 0.0) as superficie_km2 - from per_comune pc - left join comuni_registry r1 on pc.codice_comune = r1.codice_istat - left join comuni_registry r2 on lower(trim(pc.comune)) = lower(trim(r2.denominazione)) - left join provincia_map pr on left(pc.codice_comune, 3) = pr.codice_provincia -) -select - {year} as anno_riferimento, - sigla_provincia, - provincia, - regione, - sum(pop_residente) as popolazione_residente, - sum(pop_maschile) as popolazione_maschile, - sum(pop_femminile) as popolazione_femminile, - count(*) as numero_comuni, - round(sum(superficie_km2), 2) as superficie_km2, - round(sum(pop_residente)::double / nullif(sum(superficie_km2), 0), 1) as densita_ab_km2 -from per_comune_arricchito -group by sigla_provincia, provincia, regione -order by sigla_provincia diff --git a/datasets/popolazione-istat-comunale/sql/mart_by_regione.sql b/datasets/popolazione-istat-comunale/sql/mart_by_regione.sql deleted file mode 100644 index 84ba1836..00000000 --- a/datasets/popolazione-istat-comunale/sql/mart_by_regione.sql +++ /dev/null @@ -1,45 +0,0 @@ -with -comuni_registry as ( - select codice_istat, regione, denominazione, superficie_km2 - from read_parquet('https://storage.googleapis.com/dataciviclab-clean/comuni_master/2026/comuni_master_2026_clean.parquet') -), --- Mappa provincia (prime 3 cifre) -> regione, per fallback su comuni non in registry -provincia_map as ( - select distinct left(codice_istat, 3) as codice_provincia, regione - from comuni_registry - where codice_istat is not null -), -per_comune as ( - select - codice_comune, - comune, - sum(popolazione_residente) as pop_residente, - sum(totale_maschi) as pop_maschile, - sum(totale_femmine) as pop_femminile - from clean_input - where codice_comune is not null - group by codice_comune, comune -), -per_comune_arricchito as ( - select - pc.*, - -- Match: 1) per codice ISTAT, 2) per denominazione, 3) per provincia - coalesce(r1.regione, r2.regione, pr.regione, 'Altro') as regione, - coalesce(r1.superficie_km2, r2.superficie_km2, 0.0) as superficie_km2 - from per_comune pc - left join comuni_registry r1 on pc.codice_comune = r1.codice_istat - left join comuni_registry r2 on lower(trim(pc.comune)) = lower(trim(r2.denominazione)) - left join provincia_map pr on left(pc.codice_comune, 3) = pr.codice_provincia -) -select - {year} as anno_riferimento, - regione, - sum(pop_residente) as popolazione_residente, - sum(pop_maschile) as popolazione_maschile, - sum(pop_femminile) as popolazione_femminile, - count(*) as numero_comuni, - round(sum(superficie_km2), 2) as superficie_km2, - round(sum(pop_residente)::double / nullif(sum(superficie_km2), 0), 1) as densita_ab_km2 -from per_comune_arricchito -group by regione -order by regione diff --git a/datasets/popolazione-istat-comunale/sql/mart_indicatori.sql b/datasets/popolazione-istat-comunale/sql/mart_indicatori.sql deleted file mode 100644 index af9509a5..00000000 --- a/datasets/popolazione-istat-comunale/sql/mart_indicatori.sql +++ /dev/null @@ -1,58 +0,0 @@ -with -base as ( - select - codice_comune, - comune, - eta, - popolazione_residente - from clean_input - where codice_comune is not null - and comune is not null - and eta is not null - and eta <> 999 -), -fasce as ( - select - codice_comune, - comune, - sum(case when eta between 0 and 14 then popolazione_residente else 0 end) as pop_0_14, - sum(case when eta between 15 and 64 then popolazione_residente else 0 end) as pop_15_64, - sum(case when eta >= 65 then popolazione_residente else 0 end) as pop_65_plus, - sum(case when eta < 18 then popolazione_residente else 0 end) as pop_under_18, - sum(case when eta >= 75 then popolazione_residente else 0 end) as pop_75_plus, - sum(popolazione_residente) as popolazione_totale - from base - group by codice_comune, comune -) -select - {year} as anno_riferimento, - codice_comune, - comune, - popolazione_totale, - pop_0_14, - pop_15_64, - pop_65_plus, - pop_under_18, - pop_75_plus, - -- Indice di vecchiaia: pop 65+ / pop 0-14 * 100 - case when pop_0_14 > 0 - then round((pop_65_plus::double / pop_0_14::double) * 100, 1) - end as indice_vecchiaia, - -- Rapporto di dipendenza: (pop 0-14 + pop 65+) / pop 15-64 * 100 - case when pop_15_64 > 0 - then round(((pop_0_14 + pop_65_plus)::double / pop_15_64::double) * 100, 1) - end as rapporto_dipendenza, - -- Percentuale under 18 - case when popolazione_totale > 0 - then round((pop_under_18::double / popolazione_totale::double) * 100, 1) - end as pct_under_18, - -- Percentuale over 65 - case when popolazione_totale > 0 - then round((pop_65_plus::double / popolazione_totale::double) * 100, 1) - end as pct_over_65, - -- Percentuale over 75 - case when popolazione_totale > 0 - then round((pop_75_plus::double / popolazione_totale::double) * 100, 1) - end as pct_over_75 -from fasce -where popolazione_totale > 0 From c6c4bb608286e8c2288bf7266f85c370e0f2c243 Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:43:58 +0100 Subject: [PATCH 4/6] chore: gitignore data/ e datasets/ (sperimentazioni vecchie) --- .gitignore | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.gitignore b/.gitignore index 50fc1325..8fe97cf5 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,5 @@ tmp/ generated/ /sql/ *.py,cover +data/ +datasets/ From d9e458f43faab4e32690fe48835e6cf376a9a422 Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:01:10 +0100 Subject: [PATCH 5/6] fix: row_count in raw_profile.json e run record MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Aggiunto campo row_count a RawProfile - Calcolato in profile_raw() tramite DuckDB COUNT(*) - Già correttamente letto da clean validation e run record (raw_rows=40521, raw_cols=23 nel run reale) --- toolkit/profile/raw.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/toolkit/profile/raw.py b/toolkit/profile/raw.py index 9b1627ad..682de390 100644 --- a/toolkit/profile/raw.py +++ b/toolkit/profile/raw.py @@ -597,6 +597,7 @@ class RawProfile: header_line: Optional[str] columns_raw: List[str] columns_norm: List[str] + row_count: Optional[int] missingness_top: List[Dict[str, Any]] sample_rows: List[Dict[str, Any]] @@ -672,6 +673,7 @@ def profile_raw( header_line=None, columns_raw=runtime_result["columns_raw"], columns_norm=runtime_result["columns_norm"], + row_count=runtime_result.get("row_count"), missingness_top=runtime_result["missingness_top"], sample_rows=runtime_result["sample_rows"], mapping_suggestions=runtime_result["mapping_suggestions"], @@ -692,6 +694,7 @@ def profile_raw( header_line=None, columns_raw=[], columns_norm=[], + row_count=None, missingness_top=[], sample_rows=[], mapping_suggestions={}, @@ -711,6 +714,22 @@ def profile_raw( # Phase 3: DuckDB runtime profiling runtime_result = profile_with_read_cfg(file0, sniff_hints, effective_read_cfg) + # Count total rows using DuckDB (riusa read_csv options from profiling) + row_count: int | None = None + try: + from lab_connectors.duckdb import safe_connect + from toolkit.core.csv_read import csv_read_option_strings + + with safe_connect() as _con: + _read_opts = ", ".join(csv_read_option_strings(effective_read_cfg)) + csv_path = str(file0.resolve()) + _con.execute( + f"CREATE OR REPLACE VIEW _profile_count AS SELECT * FROM read_csv('{csv_path}', {_read_opts})" + ) + row_count = int(_con.execute("SELECT COUNT(*) FROM _profile_count").fetchone()[0]) + except Exception: + pass + # Phase 4: extract raw string values for date-typed columns # (DuckDB converts dates to Timestamp, losing the original format). # Uses the same effective_read_cfg as DuckDB (encoding, delim, header, @@ -734,6 +753,7 @@ def profile_raw( header_line=header_line, columns_raw=runtime_result["columns_raw"], columns_norm=runtime_result["columns_norm"], + row_count=row_count, missingness_top=runtime_result["missingness_top"], sample_rows=runtime_result["sample_rows"], mapping_suggestions=runtime_result["mapping_suggestions"], From 2222cd5391bebff4777919f04b8e8f2f3349d930 Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:05:17 +0100 Subject: [PATCH 6/6] fix: sostituito csv_quick_shape con lettura da raw_profile --- toolkit/raw/run.py | 26 ++++++++++++-------------- 1 file changed, 12 insertions(+), 14 deletions(-) diff --git a/toolkit/raw/run.py b/toolkit/raw/run.py index 8a84a5ae..9224d452 100644 --- a/toolkit/raw/run.py +++ b/toolkit/raw/run.py @@ -230,22 +230,20 @@ def run_raw( ) ) - # Calcola righe/colonne del primary output (riusa csv_quick_shape da toolit.core) + # Legge righe/colonne dal raw_profile appena scritto (se disponibile) output_rows = None col_count = None - if primary_output_path.exists() and primary_output_path.suffix.lower() in { - ".csv", - ".tsv", - ".txt", - }: - try: - from toolkit.core.duckdb_shape import csv_quick_shape - - shape = csv_quick_shape(str(primary_output_path)) - output_rows = shape.get("row_count_estimate") - col_count = shape.get("column_count") - except Exception: - pass + try: + profile_dir = out_dir / "_profile" + profile_path = profile_dir / RAW_PROFILE + if profile_path.exists(): + from toolkit.core.io import read_json_or_none as _read_json + + prof = _read_json(profile_path) or {} + output_rows = prof.get("row_count") + col_count = len(prof.get("columns_raw", [])) if prof.get("columns_raw") else None + except Exception: + pass return { "output_bytes": output_bytes,