From 5a5a450972f6227f7828d6d7b7ef2ef7e6e86e40 Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Fri, 31 Jul 2026 10:06:34 +0100 Subject: [PATCH] fix: readiness legge columns/rules dal run record, inspect mostra verdict MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Problema: 2 check di review_readiness (clean_columns_naming, validation_rules_coverage) erano sempre null perché la validazione buttava via columns/rules e summary() non esponeva validation. - core/validation.py: build_validation_summary salva il blocco summary completo (stats+columns+rules) nel run record - domain/readiness.py: summary() espone layers.*.validation; review_readiness legge da summary con fallback su legacy - cli/inspect/summary_ops.py: inspect (default) mostra il verdict readiness con check — comando unico fatti+giudizio - cli/cmd_run.py: review_readiness usa cfg.base_dir/dataset.yml (config=None in auto-detect rompeva il run finale) Verificato: dipendenti-pubblici run 2010, readiness needs-review 7/8 (coverage 13% reale, non più null). 1235 test passano. --- toolkit/cli/cmd_run.py | 2 +- toolkit/cli/inspect/summary_ops.py | 32 ++++++++++++++++ toolkit/core/validation.py | 9 ++++- toolkit/domain/readiness.py | 59 ++++++++++++++++++++++++------ 4 files changed, 87 insertions(+), 15 deletions(-) diff --git a/toolkit/cli/cmd_run.py b/toolkit/cli/cmd_run.py index f22dbc6..f57f786 100644 --- a/toolkit/cli/cmd_run.py +++ b/toolkit/cli/cmd_run.py @@ -1006,7 +1006,7 @@ def _run_pipeline( from toolkit.domain.readiness import review_readiness as _review_readiness - readiness = _review_readiness(config, year or None) + readiness = _review_readiness(str(cfg.base_dir / "dataset.yml"), year or None) results["steps"][str(year)]["readiness"] = readiness.get("readiness") results["steps"][str(year)]["checks"] = readiness.get("check_count", 0) results["steps"][str(year)]["checks_ok"] = readiness.get("ok_count", 0) diff --git a/toolkit/cli/inspect/summary_ops.py b/toolkit/cli/inspect/summary_ops.py index b7befc0..dd1f636 100644 --- a/toolkit/cli/inspect/summary_ops.py +++ b/toolkit/cli/inspect/summary_ops.py @@ -146,6 +146,29 @@ def _print_layer_profiles(dataset: str, year: int, layers: dict[str, Any]) -> No ) +def _print_readiness(readiness: dict[str, Any]) -> None: + """Stampa verdict readiness + check (se disponibili).""" + verdict = readiness.get("readiness") + if not verdict: + return + icon = {"ready": "✅", "needs-review": "⚠️", "incomplete": "🔴"}.get(verdict, "·") + typer.echo("") + typer.echo( + f"readiness: {icon} {verdict} " + f"({readiness.get('ok_count', 0)}/{readiness.get('check_count', 0)} ok, " + f"{readiness.get('fail_count', 0)} fail)" + ) + for check in readiness.get("checks") or []: + ok = check.get("ok") + c_icon = "✅" if ok else ("🔴" if ok is False else "·") + detail = check.get("detail") + if isinstance(detail, list): + detail = f"{len(detail)} output" + elif isinstance(detail, str) and len(detail) > 70: + detail = detail[:70] + "..." + typer.echo(f" {c_icon} {check.get('check', '?')}: {detail or ''}") + + def summary( config: str | None = typer.Option(None, "--config", "-c", help="Path or slug to dataset.yml"), year: int | None = typer.Option(None, "--year", "-y", help="Dataset year (default: first)"), @@ -299,6 +322,15 @@ def summary( _print_layer_profiles(ds_name, yr, layers) + if not as_json: + try: + from toolkit.domain.readiness import review_readiness as _review_readiness + + readiness = _review_readiness(str(cfg.base_dir / "dataset.yml"), yr) + _print_readiness(readiness) + except (FileNotFoundError, ValueError, OSError): + pass + if record.get("status") == "FAILED" and record.get("error"): typer.echo("") typer.echo(f"error: {record.get('error')}") diff --git a/toolkit/core/validation.py b/toolkit/core/validation.py index 33ec40f..ac43323 100644 --- a/toolkit/core/validation.py +++ b/toolkit/core/validation.py @@ -104,8 +104,13 @@ def build_validation_summary(result: ValidationResult) -> dict[str, Any]: }, ], } - if "stats" in result.summary: - out["stats"] = result.summary["stats"] + # Summary completo della validazione (stats + columns + rules) come blocco + # unico. Sostituisce il vecchio campo appiattito ``stats`` (mantenuto come + # fallback per i run record gia' scritti prima di questa migrazione). + if result.summary: + out["summary"] = result.summary + if "stats" in result.summary: + out["stats"] = result.summary["stats"] if result.sections: out["sections"] = {} for key, section in result.sections.items(): diff --git a/toolkit/domain/readiness.py b/toolkit/domain/readiness.py index c4cd69d..415c0e7 100644 --- a/toolkit/domain/readiness.py +++ b/toolkit/domain/readiness.py @@ -124,15 +124,26 @@ def summary(config_path: str | None = None, year: int | None = None) -> dict[str if latest_run_path and not _exists(latest_run_path): warnings.append("latest_run_record_missing") - # Estrai layer run status dal run record (se presente) + # Estrai layer run status + validation completa dal run record (se presente) layer_run_statuses: dict[str, dict[str, Any]] = {} + layer_validations: dict[str, dict[str, Any]] = {} if latest_run_record: for layer_name in ("raw", "clean", "mart"): layer_info = (latest_run_record.get("layers") or {}).get(layer_name, {}) layer_val = (latest_run_record.get("validations") or {}).get(layer_name, {}) metrics = layer_info.get("metrics") or {} + # Summary completo (stats+columns+rules) con fallback su stats + # appiattito per run record scritti prima della migrazione. + layer_summary = ( + (layer_val.get("summary") or {}) + if isinstance(layer_val.get("summary"), dict) + else {} + ) stats = ( - (layer_val.get("stats") or {}) if isinstance(layer_val.get("stats"), dict) else {} + (layer_summary.get("stats") or layer_val.get("stats") or {}) + if isinstance(layer_summary.get("stats"), dict) + or isinstance(layer_val.get("stats"), dict) + else {} ) layer_run_statuses[layer_name] = { "status": layer_info.get("status", "PENDING"), @@ -150,6 +161,10 @@ def summary(config_path: str | None = None, year: int | None = None) -> dict[str "paqa_score": stats.get("paqa_score"), "row_drop_pct": stats.get("row_drop_pct"), } + # Espone la validation completa del run record per i consumer + # (review_readiness legge columns/rules/row_count da qui). + if layer_val: + layer_validations[layer_name] = layer_val return { "dataset": paths.get("dataset"), @@ -171,6 +186,7 @@ def summary(config_path: str | None = None, year: int | None = None) -> dict[str "skip_suggested": (paths.get("raw_hints") or {}).get("skip"), "raw_warnings": (paths.get("raw_hints") or {}).get("warnings", []), "run_status": layer_run_statuses.get("raw"), + "validation": layer_validations.get("raw"), }, "clean": { "dir": str(clean_dir), @@ -179,6 +195,7 @@ def summary(config_path: str | None = None, year: int | None = None) -> dict[str "output_exists": _exists(clean_paths.get("output")), "metadata_exists": _exists(clean_paths.get("metadata")), "run_status": layer_run_statuses.get("clean"), + "validation": layer_validations.get("clean"), }, "mart": { "dir": str(mart_dir), @@ -189,6 +206,7 @@ def summary(config_path: str | None = None, year: int | None = None) -> dict[str "missing_outputs": missing_mart_outputs, "metadata_exists": _exists(mart_paths.get("metadata")), "run_status": layer_run_statuses.get("mart"), + "validation": layer_validations.get("mart"), }, }, "run": { @@ -231,6 +249,14 @@ def review_readiness(config_path: str | None = None, year: int | None = None) -> checks: list[dict[str, Any]] = [] + # ── Helper: legge campi dal blocco summary con fallback su top-level ── + def _val_field(validation: dict[str, Any], name: str) -> Any: + """Campo da validation.summary (nuovo) o validation top-level (legacy).""" + summary_block = validation.get("summary") or {} + if isinstance(summary_block, dict) and name in summary_block: + return summary_block.get(name) + return validation.get(name) + # --- Config check --- checks.append( { @@ -262,7 +288,7 @@ def review_readiness(config_path: str | None = None, year: int | None = None) -> # --- Clean layer --- clean = s.get("layers", {}).get("clean", {}) clean_val = clean.get("validation") or {} - clean_rows = clean_val.get("row_count") + clean_rows = _val_field(clean_val, "row_count") # Fallback: se validation non disponibile, leggi dal parquet diretto if clean_rows is None: clean_path_str = clean.get("output") @@ -287,7 +313,7 @@ def review_readiness(config_path: str | None = None, year: int | None = None) -> mart_checks: list[dict[str, Any]] = [] for output_name in mart_outputs: o_path = Path(output_name) - rows = mart_val.get("row_count") if o_path.exists() else None + rows = _val_field(mart_val, "row_count") if o_path.exists() else None # Fallback: leggi dal parquet se validation non disponibile if rows is None and o_path.exists(): rows = parquet_row_count(o_path) @@ -333,7 +359,7 @@ def review_readiness(config_path: str | None = None, year: int | None = None) -> ) # --- A. Clean column naming (snake_case check) --- - clean_cols = clean_val.get("columns") or clean.get("columns") + clean_cols = _val_field(clean_val, "columns") or clean.get("columns") if clean_cols: _bad_naming: list[str] = [] for col in clean_cols: @@ -356,7 +382,7 @@ def review_readiness(config_path: str | None = None, year: int | None = None) -> ) # --- B. Validation rules coverage --- - rules_obj = clean_val.get("rules") or clean.get("rules") or {} + rules_obj = _val_field(clean_val, "rules") or clean.get("rules") or {} if clean_cols: covered_cols: set[str] = set() for rule_name, rule_vals in rules_obj.items(): @@ -443,14 +469,23 @@ def review_readiness(config_path: str | None = None, year: int | None = None) -> } # Transition stats from clean validation - raw_row_count = clean_val.get("raw_row_count") - clean_row_count = clean_val.get("clean_row_count") + # (nuovo: summary.stats.* — legacy: campi top-level della validation) + raw_row_count = _val_field(clean_val, "raw_row_count") or ( + (clean_val.get("summary") or {}).get("stats") or {} + ).get("raw_rows") + clean_row_count = _val_field(clean_val, "clean_row_count") or ( + (clean_val.get("summary") or {}).get("stats") or {} + ).get("clean_rows") col_drop = None row_drop_pct = None if raw_row_count is not None and clean_row_count is not None and raw_row_count > 0: row_drop_pct = round((raw_row_count - clean_row_count) / raw_row_count * 100, 1) - raw_col_count = raw_val.get("col_count") - clean_col_count = clean_val.get("col_count") + raw_col_count = _val_field(raw_val, "col_count") or ( + (raw_val.get("summary") or {}).get("stats") or {} + ).get("raw_cols") + clean_col_count = _val_field(clean_val, "col_count") or ( + (clean_val.get("summary") or {}).get("stats") or {} + ).get("clean_cols") if raw_col_count is not None and clean_col_count is not None: col_drop = raw_col_count - clean_col_count @@ -484,8 +519,8 @@ def review_readiness(config_path: str | None = None, year: int | None = None) -> "validation_msgs": clean_msgs, "output": clean.get("output"), "row_count": clean_rows, - "columns": clean_val.get("columns") or clean.get("columns"), - "rules": clean_val.get("rules") or clean.get("rules"), + "columns": _val_field(clean_val, "columns") or clean.get("columns"), + "rules": _val_field(clean_val, "rules") or clean.get("rules"), "transition": { "raw_row_count": raw_row_count, "clean_row_count": clean_row_count,