From 46c926a8255e8dd77f7ad51222645e300920a4da Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:31:45 +0100 Subject: [PATCH 1/4] cleanup: rimossi README.md report, csv_quick_shape, file .cover --- tests/test_report_ops.py | 408 ----------------------------------- toolkit/cli/cmd_run.py | 16 -- toolkit/core/duckdb_shape.py | 37 +--- toolkit/domain/report.py | 266 +---------------------- 4 files changed, 2 insertions(+), 725 deletions(-) delete mode 100644 tests/test_report_ops.py diff --git a/tests/test_report_ops.py b/tests/test_report_ops.py deleted file mode 100644 index cbd0693f..00000000 --- a/tests/test_report_ops.py +++ /dev/null @@ -1,408 +0,0 @@ -"""Test per report_ops: build_dataset_readme e integrazione run full.""" - -from __future__ import annotations - -import json -from pathlib import Path - -import pytest - -from toolkit.domain.report import ( - build_dataset_readme, - write_run_report, -) - - -# --------------------------------------------------------------------------- -# build_dataset_readme — test su output markdown -# --------------------------------------------------------------------------- - - -class TestBuildDatasetReadme: - """Verifica che il README markdown aggregato abbia la struttura attesa.""" - - def _make_report(self, year: int, **overrides: object) -> dict: - """Report sintetico minimale per test.""" - report: dict = { - "dataset": "test_dataset", - "config_path": "/path/to/dataset.yml", - "year": year, - "status": "SUCCESS", - "duration_seconds": 2.5, - "readiness": "ready", - "readiness_checks": {"total": 5, "ok": 5, "fail": 0}, - "preflight": { - "config_ok": True, - "sources_reachable": 1, - "sources_total": 1, - "quality_score_avg": 88, - }, - "layers": { - "raw": { - "status": "SUCCESS", - "validation": {"ok": True, "errors": 0, "warnings": 0}, - "warnings": [], - "errors": [], - "encoding": "utf-8", - "delim": ",", - "file_size_bytes": 1024, - }, - "clean": { - "status": "SUCCESS", - "validation": {"ok": True, "errors": 0, "warnings": 2}, - "warnings": ["colonna X rimossa", "colonna Y rinominata"], - "errors": [], - "rows": 150, - "columns": 12, - "file_size_bytes": 4096, - }, - "mart": { - "status": "SUCCESS", - "validation": {"ok": True, "errors": 0, "warnings": 1}, - "warnings": ["colonna Z rimossa"], - "errors": [], - "tables": [{"name": "t1", "rows": 50}, {"name": "t2", "rows": 100}], - "total_rows": 150, - "file_size_bytes": 2048, - }, - }, - "support_datasets": [], - } - report.update(overrides) - return report - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_ha_titolo_e_dataset(self) -> None: - """Il README contiene titolo e nome dataset.""" - md = build_dataset_readme("test", "/c.yml", [self._make_report(2023)]) - assert "# Run Report: `test`" in md - assert "/c.yml" in md - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_tabella_ha_tutti_gli_anni(self) -> None: - """La tabella elenca tutti gli anni presenti.""" - reports = [self._make_report(2023), self._make_report(2024)] - md = build_dataset_readme("test", "/c.yml", reports) - assert "| 2023 |" in md - assert "| 2024 |" in md - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_tabella_mostra_righe_e_warning(self) -> None: - """La tabella include righe, warning, dimensione file.""" - md = build_dataset_readme("test", "/c.yml", [self._make_report(2023)]) - assert "150 righe" in md - assert "2w" in md - assert "1.0KB" in md or "4.0KB" in md or "4KB" in md - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_sezione_warning_per_anno(self) -> None: - """Warning compaiono nella sezione dedicata per anno.""" - md = build_dataset_readme("test", "/c.yml", [self._make_report(2023)]) - assert "### Anno 2023" in md - assert "colonna X rimossa" in md - assert "colonna Z rimossa" in md - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_sezione_readiness(self) -> None: - """La sezione Review Readiness e' presente.""" - md = build_dataset_readme("test", "/c.yml", [self._make_report(2023)]) - assert "Review Readiness" in md - assert "ready" in md - assert "5/5" in md - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_qualita_nella_tabella(self) -> None: - """Il quality score compare nella tabella.""" - md = build_dataset_readme("test", "/c.yml", [self._make_report(2023)]) - assert "**88**" in md - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_nessun_warning_silenziato(self) -> None: - """Report senza warning non genera sezione warning.""" - r = self._make_report(2023) - r["layers"]["clean"]["warnings"] = [] - r["layers"]["clean"]["validation"]["warnings"] = 0 - r["layers"]["mart"]["warnings"] = [] - r["layers"]["mart"]["validation"]["warnings"] = 0 - md = build_dataset_readme("test", "/c.yml", [r]) - assert "Warning ed errori" not in md - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_readiness_needs_review(self) -> None: - """Verdetto needs-review ha icona appropriata.""" - r = self._make_report( - 2023, readiness="needs-review", readiness_checks={"total": 5, "ok": 4, "fail": 1} - ) - md = build_dataset_readme("test", "/c.yml", [r]) - assert "needs-review" in md - assert "4/5" in md - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_readiness_incomplete(self) -> None: - r = self._make_report( - 2023, readiness="incomplete", readiness_checks={"total": 5, "ok": 2, "fail": 3} - ) - md = build_dataset_readme("test", "/c.yml", [r]) - assert "incomplete" in md - assert "2/5" in md - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_status_failed(self) -> None: - r = self._make_report(2023, status="FAILED") - md = build_dataset_readme("test", "/c.yml", [r], overall_status="failed") - assert "FAILED" in md - assert "failed" in md - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_quality_null_non_mostra_valore(self) -> None: - r = self._make_report(2023) - r["preflight"]["quality_score_avg"] = None - md = build_dataset_readme("test", "/c.yml", [r]) - lines = [line for line in md.split("\n") if "2023" in line and "|" in line] - assert any("| - |" in line or "| - " in line for line in lines) - - @pytest.mark.contract - @pytest.mark.pure_unit - def test_support_datasets(self) -> None: - r = self._make_report( - 2023, - support_datasets=[{"name": "istat-elenco-comuni", "year": 2026, "status": "SUCCESS"}], - ) - md = build_dataset_readme("test", "/c.yml", [r]) - assert "Support Datasets" in md - assert "istat-elenco-comuni" in md - - -# --------------------------------------------------------------------------- -# write_run_report — verifica che il JSON sia valido e abbia campi minimi -# --------------------------------------------------------------------------- - - -class TestWriteRunReport: - @pytest.mark.contract - def test_write_e_lettura(self, tmp_path: Path) -> None: - """Scrive un report e verifica che sia JSON valido con i campi base.""" - report = { - "dataset": "prova", - "config_path": "/c.yml", - "year": 2024, - "run_id": "test_123", - "status": "SUCCESS", - "readiness": "ready", - "preflight": {"config_ok": True}, - "layers": {}, - "support_datasets": [], - } - path = write_run_report(report, tmp_path, "prova", 2024) - assert path.exists() - assert path.name == "2024_run_report.json" - raw = path.read_text(encoding="utf-8") - loaded = json.loads(raw) - assert loaded["dataset"] == "prova" - assert loaded["year"] == 2024 - assert loaded["status"] == "SUCCESS" - - -# --------------------------------------------------------------------------- -# FAILED report — simulazione run fallito -# --------------------------------------------------------------------------- - - -class TestRunFullFailedReport: - """run_full con run_year che fallisce produce report FAILED e rilancia.""" - - def _make_config(self, tmp_path: Path) -> Path: - """Crea un dataset.yml minimale per test.""" - sql_dir = tmp_path / "sql" / "mart" - sql_dir.mkdir(parents=True, exist_ok=True) - (tmp_path / "sql" / "clean.sql").write_text("select 1 as value", encoding="utf-8") - (sql_dir / "mart_example.sql").write_text("select * from clean_input", encoding="utf-8") - cfg = tmp_path / "dataset.yml" - cfg.write_text( - "\n".join( - [ - f'root: "{(tmp_path / "out").as_posix()}"', - "dataset:", - ' name: "test_fail"', - " years: [2023, 2024]", - "raw: {}", - "clean:", - ' sql: "sql/clean.sql"', - "mart:", - " tables:", - ' - name: "mart_example"', - ' sql: "sql/mart/mart_example.sql"', - ] - ), - encoding="utf-8", - ) - return cfg - - @pytest.mark.contract - def test_run_full_fail_produce_report_failed_e_rilancia( - self, tmp_path: Path, monkeypatch - ) -> None: - """run_full con run_year che fallisce: - - scrive report FAILED per l'anno fallito - - NON scrive report per l'anno successivo (non eseguito) - - rilancia l'eccezione originale - """ - from toolkit.cli import cmd_run - - config_path = self._make_config(tmp_path) - - # Mock config check (non valuta raw.sources/support reali) - monkeypatch.setattr( - "toolkit.domain.preflight.run_config_check", - lambda *args, **kwargs: { - "ok": True, - "errors": [], - "warnings": [], - "slug": "test", - }, - ) - - # Mock source probe per saltare check rete - monkeypatch.setattr( - "toolkit.domain.preflight.run_preflight", - lambda *args, **kwargs: { - "config_check": {"ok": True, "errors": [], "warnings": [], "slug": "test"}, - "sources": [], - "status": "passed", - }, - ) - - # Mock review_readiness - import toolkit.domain.readiness as _readiness_ops - - monkeypatch.setattr( - _readiness_ops, - "review_readiness", - lambda *args, **kwargs: { - "readiness": "ready", - "check_count": 5, - "ok_count": 5, - "fail_count": 0, - "layers": {}, - }, - ) - - # Mock run_year per fallire al primo anno - call_count = {"n": 0} - - def _failing_run_year(*args, **kwargs): - call_count["n"] += 1 - if call_count["n"] == 1: - raise RuntimeError("simulated run failure") - return None # non dovrebbe arrivare qui - - monkeypatch.setattr(cmd_run, "run_year", _failing_run_year) - - # Chiamata: deve rilanciare l'eccezione - with pytest.raises(RuntimeError, match="simulated run failure"): - cmd_run.run_full( - config=str(config_path), - years=None, - smoke=False, - sample_rows=None, - sample_bytes=None, - root=None, - json_output=False, - dry_run=False, - ) - - # Verifica: report per anno 2023 esiste con status FAILED - report_dir = tmp_path / "out" / "data" / "_reports" / "test_fail" - report_2023 = report_dir / "2023_run_report.json" - assert report_2023.exists(), "Report per anno fallito non trovato" - import json - - loaded = json.loads(report_2023.read_text(encoding="utf-8")) - assert loaded["status"] == "FAILED", f"Status atteso FAILED, got {loaded['status']}" - - # Verifica: report per 2024 NON esiste (mai eseguito) - report_2024 = report_dir / "2024_run_report.json" - assert not report_2024.exists(), "Report per anno non eseguito non dovrebbe esistere" - - # run_year chiamato solo 1 volta (secondo anno skippato) - assert call_count["n"] == 1, "run_year non dovrebbe essere chiamato per il secondo anno" - - # Verifica: README aggregato esiste e contiene lo stato - readme = report_dir / "README.md" - assert readme.exists() - assert "test_fail" in readme.read_text(encoding="utf-8") - - -class TestFailedReport: - """build_run_report chiamato con step_results minimale (come dopo eccezione).""" - - @pytest.mark.contract - def test_failed_report_da_step_results_minimi(self, tmp_path: Path) -> None: - """Con step_results={'run':'failed'}, build_run_report produce status FAILED - e non solleva eccezioni (tollerante a layers assenti).""" - from toolkit.domain.report import build_run_report - - report = build_run_report( - config_path="/c.yml", - year=2024, - root=tmp_path, - dataset="fallito", - step_results={"run": "failed", "validate": "failed"}, - run_mode="full", - ) - assert report["dataset"] == "fallito" - assert report["year"] == 2024 - assert report["readiness"] is None # nessun readiness disponibile - # I layer ci sono sempre (raw/clean/mart) ma con validation=None - for lname in ("raw", "clean", "mart"): - assert lname in report["layers"] - assert report["layers"][lname]["validation"]["ok"] is None - assert report["preflight"]["sources_total"] == 0 - # step_results={"run":"failed"} → status="FAILED" (fallback senza run record) - assert report["status"] == "FAILED" - - @pytest.mark.contract - def test_failed_report_scritto_su_disco(self, tmp_path: Path) -> None: - """write_run_report con status FAILED produce JSON valido.""" - report = { - "dataset": "fallito", - "config_path": "/c.yml", - "year": 2024, - "run_id": None, - "status": "FAILED", - "readiness": None, - "layers": {}, - "support_datasets": [], - } - path = write_run_report(report, tmp_path, "fallito", 2024) - assert path.exists() - loaded = json.loads(path.read_text(encoding="utf-8")) - assert loaded["status"] == "FAILED" - assert loaded["layers"] == {} - - @pytest.mark.contract - def test_readme_con_anno_fallito(self) -> None: - """Un report FAILED nel README mostra stato FAILED e readiness assente.""" - r = { - "dataset": "test", - "config_path": "/c.yml", - "year": 2024, - "status": "FAILED", - "readiness": None, - "layers": {}, - "support_datasets": [], - } - md = build_dataset_readme("test", "/c.yml", [r]) - assert "FAILED" in md - assert "🔴" in md diff --git a/toolkit/cli/cmd_run.py b/toolkit/cli/cmd_run.py index b15dacef..16d7ff61 100644 --- a/toolkit/cli/cmd_run.py +++ b/toolkit/cli/cmd_run.py @@ -1005,9 +1005,6 @@ def run_full( from toolkit.domain.report import ( build_run_report, write_run_report, - write_dataset_readme, - _all_reports_for_dataset, - _derive_overall_status, ) run_mode = "smoke" if sample_mode else "full" @@ -1059,19 +1056,6 @@ def run_full( support_datasets=support_info, ) write_run_report(report, cfg.root, cfg.dataset, year) - - # Leggi TUTTI i report JSON esistenti su disco - # e rigenera il README completo con stato aggregato - all_reports = _all_reports_for_dataset(cfg.root, cfg.dataset) - if all_reports: - agg_status = _derive_overall_status(all_reports) - write_dataset_readme( - cfg.root, - cfg.dataset, - all_reports, - overall_status=agg_status, - config_path=config, - ) except Exception as _report_err: logger.warning("Generazione report saltata: %s", _report_err) diff --git a/toolkit/core/duckdb_shape.py b/toolkit/core/duckdb_shape.py index e9490ef7..85e396c6 100644 --- a/toolkit/core/duckdb_shape.py +++ b/toolkit/core/duckdb_shape.py @@ -15,7 +15,7 @@ from pathlib import Path from typing import Any -from lab_connectors.duckdb import gcs_connect, safe_connect +from lab_connectors.duckdb import gcs_connect from toolkit.core.sql_utils import sql_literal @@ -67,41 +67,6 @@ def _rel(path: Path) -> str: # --------------------------------------------------------------------------- -def csv_quick_shape(csv_path: str | Path) -> dict[str, Any]: - """Quick row count and column count for a CSV file via DuckDB auto-detect. - - Args: - csv_path: Path to the CSV file. - - Returns: - Dict with ``row_count_estimate`` (int | None) and ``column_count`` (int | None). - Returns empty dict if file not found or unreadable (non-CSV, encoding issues, etc.). - - Note: - Uses ``read_csv_auto`` with ``auto_detect=true``. Non fa sniff esplicito, - approssimativo ma veloce. Per profiling completo vedi - ``toolkit.profile.raw.profile_raw``. - """ - path = Path(csv_path) - if not path.exists(): - return {} - try: - with safe_connect() as conn: - rel = f"read_csv_auto('{sql_literal(str(path))}', auto_detect=true)" - describe = conn.execute(f"DESCRIBE SELECT * FROM {rel}").fetchall() - col_count = len(describe) - count_row = conn.execute(f"SELECT COUNT(*) FROM {rel}").fetchone() - row_count = int(count_row[0]) if count_row else None - return {"row_count_estimate": row_count, "column_count": col_count} - except Exception: - return {} - - -# --------------------------------------------------------------------------- -# Parquet operations -# --------------------------------------------------------------------------- - - def parquet_schema(path: Path) -> list[dict[str, str]]: """Legge lo schema di un file Parquet (nomi colonne + tipo DuckDB). diff --git a/toolkit/domain/report.py b/toolkit/domain/report.py index 0302ad01..cc1e7847 100644 --- a/toolkit/domain/report.py +++ b/toolkit/domain/report.py @@ -6,7 +6,7 @@ from __future__ import annotations -from datetime import datetime, timezone +from datetime import datetime from pathlib import Path from typing import Any @@ -22,7 +22,6 @@ _REPORT_DIR = "_reports" _RUN_REPORT_FILENAME = "run_report.json" -_DATASET_README = "README.md" def _get_run_record(root: Path, dataset: str, year: int) -> dict[str, Any] | None: @@ -119,51 +118,6 @@ def _collect_mart_transitions(root: Path, dataset: str, year: int) -> list[dict[ _VALID_INCOMPLETE = frozenset({"RUNNING", "DRY_RUN"}) -def _derive_overall_status(reports: list[dict[str, Any]]) -> str: - """Deriva lo stato complessivo da una lista di report. - - - Se almeno un report e' FAILED/BLOCKED/ERROR → 'failed' - - Se almeno un report e' None/RUNNING/DRY_RUN → 'incomplete' - - Se almeno un report e' SUCCESS_WITH_WARNINGS → 'passed_with_warnings' - - Se tutti sono SUCCESS → 'passed' - - Se lista vuota → 'unknown' - """ - if not reports: - return "unknown" - - has_incomplete = False - for r in reports: - s = r.get("status") - if s in _VALID_FAILED: - return "failed" - if s is None or s in _VALID_INCOMPLETE or s not in ("SUCCESS", "SUCCESS_WITH_WARNINGS"): - has_incomplete = True - - if has_incomplete: - return "incomplete" - if any(r.get("status") == "SUCCESS_WITH_WARNINGS" for r in reports): - return "passed_with_warnings" - return "passed" - - -def _all_reports_for_dataset(root: str | Path, dataset: str) -> list[dict[str, Any]]: - """Legge tutti i report JSON esistenti per un dataset, ordinati per anno. - - Cerca in {root}/data/_reports/{dataset}/*_run_report.json. - """ - root_path = Path(root) - reports_dir = root_path / "data" / _REPORT_DIR / dataset - if not reports_dir.exists(): - return [] - - reports: list[dict[str, Any]] = [] - for f in sorted(reports_dir.glob("*_run_report.json")): - raw = read_json_or_none(f) - if raw: - reports.append(raw) - return reports - - def _duration_seconds(start: str | None, end: str | None) -> float | None: """Calcola secondi tra due timestamp ISO.""" if not start or not end: @@ -425,221 +379,3 @@ def _fmt_bytes(b: int | None) -> str: if b < 1024 * 1024: return f"{b / 1024:.1f}KB" return f"{b / (1024 * 1024):.1f}MB" - - -def build_dataset_readme( - dataset: str, - config_path: str, - reports: list[dict[str, Any]], - overall_status: str | None = None, -) -> str: - """Genera un README.md aggregato per tutti gli anni di un dataset. - - Args: - dataset: slug del dataset - config_path: path al dataset.yml - reports: lista di report JSON (uno per anno) - overall_status: stato complessivo (passed/failed) - - Returns: - Testo markdown del README. - """ - status = overall_status or "passed" - if status == "failed": - status_icon = "🔴" - elif status == "passed_with_warnings": - status_icon = "⚠️" - elif status == "incomplete": - status_icon = "🔶" - else: - status_icon = "✅" - - lines = [ - f"# Run Report: `{dataset}`\n", - ] - - # Metadata - lines.append("## Dataset\n") - lines.append(f"- Config: `{config_path}`") - lines.append(f"- Stato complessivo: {status_icon} **{status}**") - years_list = sorted({int(r["year"]) for r in reports if r.get("year") is not None}) - lines.append(f"- Anni processati: {', '.join(str(y) for y in years_list)}") - lines.append(f"- Anni con report: {len(reports)}") - lines.append("") - - # Tabella riepilogativa per anno - lines.append("## Riepilogo per anno\n") - lines.append("| Anno | Status | Readiness | Qualità | Raw | Clean | Mart | Durata |") - lines.append("|------|--------|-----------|---------|-----|-------|------|--------|") - - for r in sorted(reports, key=lambda x: x.get("year", 0)): - year = r.get("year", "?") - status_icon_col = _status_icon(r.get("status")) - status_name = r.get("status", "?") - readiness_icon_col = _readiness_icon(r.get("readiness")) - readiness_name = r.get("readiness", "?") - - layers = r.get("layers") or {} - - # Qualità: preflight PA score (CSV) → fallback validazione clean - pf = r.get("preflight") or {} - qs = pf.get("quality_score_avg") - if qs is None: - r_clean_val = layers.get("clean", {}).get("validation", {}) - qs = r_clean_val.get("quality_score") - q_str = f"**{qs}**" if qs is not None else "-" - - # Raw - r_raw = layers.get("raw") or {} - raw_info = f"{_validation_icon(r_raw.get('validation', {}).get('ok'))}" - enc = r_raw.get("encoding") or "" - w_count = r_raw.get("validation", {}).get("warnings", 0) - raw_info += f" {enc}" if enc else "" - raw_info += f" ({w_count}w)" if w_count else "" - raw_bytes = r_raw.get("file_size_bytes") - raw_info += f" · {_fmt_bytes(raw_bytes)}" if raw_bytes else "" - - # Clean - r_clean = layers.get("clean") or {} - clean_info = f"{_validation_icon(r_clean.get('validation', {}).get('ok'))}" - rows = r_clean.get("rows") - rows_str = f"{rows} righe" if rows is not None else "" - w_count = r_clean.get("validation", {}).get("warnings", 0) - clean_info += f" {rows_str}" if rows_str else "" - clean_info += f" ({w_count}w)" if w_count else "" - clean_bytes = r_clean.get("file_size_bytes") - clean_info += f" · {_fmt_bytes(clean_bytes)}" if clean_bytes else "" - - # Mart - r_mart = layers.get("mart") or {} - mart_info = f"{_validation_icon(r_mart.get('validation', {}).get('ok'))}" - tables = r_mart.get("tables") or [] - n_tables = len(tables) - mart_rows = r_mart.get("total_rows") - mart_str = f"{n_tables} tabelle" - if mart_rows is not None: - mart_str += f" ({mart_rows} righe)" - w_count = r_mart.get("validation", {}).get("warnings", 0) - mart_info += f" {mart_str}" - mart_info += f" ({w_count}w)" if w_count else "" - - duration_str = _fmt_duration(r.get("duration_seconds")) - - lines.append( - f"| {year} | {status_icon_col} {status_name} | {readiness_icon_col} {readiness_name} " - f"| {q_str} | {raw_info} | {clean_info} | {mart_info} | {duration_str} |" - ) - - lines.append("") - - # Preflight - preflight_info = None - for r in reports: - pf = r.get("preflight") or {} - if pf: - preflight_info = pf - break - - if preflight_info: - lines.append("## Preflight\n") - lines.append(f"- Config valida: {'✅' if preflight_info.get('config_ok') else '🔴'}") - lines.append( - f"- Fonti: {preflight_info.get('sources_reachable', '?')}/" - f"{preflight_info.get('sources_total', '?')} raggiungibili" - ) - qs = preflight_info.get("quality_score_avg") - if qs is not None: - lines.append(f"- Quality score medio: **{qs}/100**") - lines.append("") - - # Warning ed errori per anno (solo se presenti) - _any_issues = False - for r in reports: - for lname in ("raw", "clean", "mart"): - lr = (r.get("layers") or {}).get(lname) or {} - if lr.get("warnings") or lr.get("errors"): - _any_issues = True - break - if _any_issues: - break - - if _any_issues: - lines.append("## Warning ed errori\n") - for r in sorted(reports, key=lambda x: x.get("year", 0)): - year = r.get("year", "?") - layers = r.get("layers") or {} - has_issues = False - for lname in ("raw", "clean", "mart"): - lr = layers.get(lname) or {} - ws = lr.get("warnings") or [] - es = lr.get("errors") or [] - if ws or es: - if not has_issues: - lines.append(f"### Anno {year}\n") - has_issues = True - lines.append(f"**{lname}**: {len(es)} errori, {len(ws)} warning") - for w in ws[:3]: - lines.append(f" - ⚠ {w}") - for e in es[:3]: - lines.append(f" - ❌ {e}") - if has_issues: - lines.append("") - - # Readiness per anno - has_readiness = any(r.get("readiness") for r in reports) - if has_readiness: - lines.append("## Review Readiness\n") - for r in sorted(reports, key=lambda x: x.get("year", 0)): - year = r.get("year", "?") - rd = r.get("readiness", "?") - rc = r.get("readiness_checks") or {} - icon = _readiness_icon(rd) - lines.append( - f"- Anno {year}: {icon} **{rd}** ({rc.get('ok', 0)}/{rc.get('total', 0)} check ok)" - ) - lines.append("") - - # Support datasets - has_support = any(r.get("support_datasets") for r in reports if r.get("support_datasets")) - if has_support: - lines.append("## Support Datasets\n") - seen = set() - for r in reports: - for s in r.get("support_datasets") or []: - name = s.get("name") - if name and name not in seen: - seen.add(name) - lines.append(f"- {name}") - lines.append("") - - lines.append("---") - lines.append( - f"*Report generato il {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')} " - f"dal toolkit*" - ) - lines.append("") - - return "\n".join(lines) - - -def write_dataset_readme( - root: str | Path, - dataset: str, - reports: list[dict[str, Any]], - overall_status: str | None = None, - config_path: str | None = None, -) -> Path: - """Scrive il README.md aggregato per un dataset. - - Path: {root}/data/_reports/{dataset}/README.md - """ - root_path = Path(root) - report_dir = root_path / "data" / _REPORT_DIR / dataset - report_dir.mkdir(parents=True, exist_ok=True) - readme_path = report_dir / _DATASET_README - - # Prendi config_path dal primo report se non fornito - cfg_path: str = config_path or (str(reports[0].get("config_path", "?")) if reports else "?") - md = build_dataset_readme(dataset, cfg_path, reports, overall_status) - readme_path.write_text(md, encoding="utf-8") - return readme_path From 5d2ae7c95ccbb56eae59909abaff532737c2b44a Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:53:42 +0100 Subject: [PATCH 2/4] fix: unused variables dopo rimozione report --- toolkit/cli/cmd_run.py | 76 +------- toolkit/domain/report.py | 381 --------------------------------------- 2 files changed, 8 insertions(+), 449 deletions(-) delete mode 100644 toolkit/domain/report.py diff --git a/toolkit/cli/cmd_run.py b/toolkit/cli/cmd_run.py index 16d7ff61..af8337b8 100644 --- a/toolkit/cli/cmd_run.py +++ b/toolkit/cli/cmd_run.py @@ -57,37 +57,6 @@ def _is_mart_only_cfg(cfg) -> bool: return not bool(cfg.clean.sql) -def _write_blocked_report( - config_path: str, - year: int, - root: str | Path, - dataset: str, - run_mode: str, - support_datasets: list[dict[str, Any]], -) -> None: - """Scrive un report minimale per un anno non eseguito (candidate bloccato).""" - from toolkit.domain.report import write_run_report - - report = { - "dataset": dataset, - "config_path": str(config_path), - "year": year, - "run_id": None, - "run_mode": run_mode, - "toolkit_version": None, - "status": "BLOCKED", - "duration_seconds": None, - "config_hash": None, - "source_urls": [], - "readiness": None, - "readiness_checks": {"total": 0, "ok": 0, "fail": 0}, - "preflight": {}, - "layers": {}, - "support_datasets": support_datasets, - } - write_run_report(report, root, dataset, year) - - def _validate_execution_plan(cfg, step: str) -> list[str]: layers = _planned_layers(step) @@ -999,15 +968,10 @@ def run_full( results["status"] = "failed" # ── Run report (best-effort: non fa fallire il run) ──────────────────── - # Eseguito sempre: anche con candidate bloccato, run fallito, o eccezione. + # Report non piu' generato su disco — i dati sono nel run record (_runs/) try: if not dry_flag: - from toolkit.domain.report import ( - build_run_report, - write_run_report, - ) - - run_mode = "smoke" if sample_mode else "full" + # run mode not needed anymore # Raccogli info dai support support_info: list[dict[str, Any]] = [] @@ -1028,36 +992,12 @@ def run_full( ) # Anni effettivamente eseguiti (hanno una voce in results["steps"]) - attempted_years = set(results["steps"].keys()) - - # Genera report solo per gli anni tentati (non per quelli saltati - # da un'eccezione o da candidate_blocked), per evitare di - # riusare artifact di run precedenti. - for year in selected_years: - sy = str(year) - if sy not in attempted_years: - # Anno non eseguito: scrive report BLOCKED esplicito - # solo se il candidate era bloccato da support. - if candidate_blocked: - _write_blocked_report( - config, year, cfg.root, cfg.dataset, run_mode, support_info - ) - continue - - step_data = results["steps"][sy] - report = build_run_report( - config_path=config, - year=year, - root=cfg.root, - dataset=cfg.dataset, - preflight=results.get("preflight"), - step_results=step_data, - run_mode=run_mode, - support_datasets=support_info, - ) - write_run_report(report, cfg.root, cfg.dataset, year) - except Exception as _report_err: - logger.warning("Generazione report saltata: %s", _report_err) + # attempted_years not needed anymore + + # Report non piu' generato — i dati sono nel run record (_runs/) + pass + except Exception: + pass # Rilancia l'eccezione originale del candidate (preserva traceback) if _candidate_exc is not None: diff --git a/toolkit/domain/report.py b/toolkit/domain/report.py deleted file mode 100644 index cc1e7847..00000000 --- a/toolkit/domain/report.py +++ /dev/null @@ -1,381 +0,0 @@ -"""Report di run aggregato: JSON per anno + markdown multi-anno. - -Costruisce un report unico a partire dagli artifact del run -(run record, validazione, readiness, preflight) e lo persiste su disco. -""" - -from __future__ import annotations - -from datetime import datetime -from pathlib import Path -from typing import Any - -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 ( - METADATA, - RAW_PROFILE, - RAW_PROFILE_DIR, - layer_year_dir, -) -from toolkit.core.run_records import get_run_dir, latest_run - -_REPORT_DIR = "_reports" -_RUN_REPORT_FILENAME = "run_report.json" - - -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) - try: - return latest_run(run_dir) - except (FileNotFoundError, OSError): - return None - - -def _collect_mart_tables(root: Path, dataset: str, year: int) -> list[dict[str, Any]]: - """Legge dalla metadata del mart l'elenco tabelle con row count.""" - mart_dir = layer_year_dir(root, "mart", dataset, year) - meta = read_layer_metadata(mart_dir) - table_profiles = meta.get("table_profiles") or {} - if not table_profiles: - tables = [] - for f in sorted(mart_dir.glob("*.parquet")): - tables.append({"name": f.stem, "rows": None}) - return tables - return [ - {"name": name, "rows": profile.get("row_count")} for name, profile in table_profiles.items() - ] - - -def _collect_clean_profile(root: Path, dataset: str, year: int) -> dict[str, Any]: - """Legge profilo clean da metadata.json.""" - clean_dir = layer_year_dir(root, "clean", dataset, year) - meta = read_layer_metadata(clean_dir) - output_profile = meta.get("output_profile") or {} - return { - "row_count": output_profile.get("row_count"), - "col_count": len(output_profile.get("columns") or []) - if output_profile.get("columns") - else None, - } - - -def _collect_raw_profile(root: Path, dataset: str, year: int) -> dict[str, Any]: - """Legge profilo raw da metadata.json (encoding, delim, primary_output).""" - raw_dir = layer_year_dir(root, "raw", dataset, year) - meta = read_layer_metadata(raw_dir) - hints = meta.get("profile_hints") or {} - return { - "encoding": hints.get("encoding_suggested"), - "delim": hints.get("delim_suggested"), - "primary_output": meta.get("primary_output_file"), - } - - -def _collect_raw_row_count(root: Path, dataset: str, year: int) -> int | None: - """Legge row count dal raw_profile.json.""" - raw_dir = layer_year_dir(root, "raw", dataset, year) - profile_path = raw_dir / RAW_PROFILE_DIR / RAW_PROFILE - if profile_path.exists(): - pf = read_json_or_none(profile_path) or {} - return pf.get("row_count") - return None - - -def _collect_config_hash(root: Path, dataset: str, year: int) -> str | None: - """Legge config_hash da raw metadata.json (o clean, mart).""" - for layer in ("raw", "clean", "mart"): - ld = layer_year_dir(root, layer, dataset, year) - meta_path = ld / METADATA - if meta_path.exists(): - meta = read_json_or_none(meta_path) or {} - ch = meta.get("config_hash") - if ch: - return ch - return None - - -def _collect_output_bytes(root: Path, layer: str, dataset: str, year: int) -> int | None: - """Legge il totale bytes dal metadata.json di un layer.""" - ld = layer_year_dir(root, layer, dataset, year) - meta = read_layer_metadata(ld) - outputs = meta.get("outputs") or [] - if outputs: - total = sum(o.get("bytes", 0) for o in outputs if o.get("bytes") is not None) - return total if total else None - return None - - -def _collect_mart_transitions(root: Path, dataset: str, year: int) -> list[dict[str, Any]]: - """Legge transition_profiles dal mart metadata.json (clean→mart).""" - mart_dir = layer_year_dir(root, "mart", dataset, year) - meta = read_layer_metadata(mart_dir) - return meta.get("transition_profiles") or [] - - -_VALID_FAILED = frozenset({"FAILED", "BLOCKED", "ERROR"}) -_VALID_INCOMPLETE = frozenset({"RUNNING", "DRY_RUN"}) - - -def _duration_seconds(start: str | None, end: str | None) -> float | None: - """Calcola secondi tra due timestamp ISO.""" - if not start or not end: - return None - try: - fmt = "%Y-%m-%dT%H:%M:%S" - s = datetime.strptime(start[:19], fmt) - e = datetime.strptime(end[:19], fmt) - return (e - s).total_seconds() - except (ValueError, TypeError): - return None - - -# --------------------------------------------------------------------------- -# Build report -# --------------------------------------------------------------------------- - - -def build_run_report( - config_path: str, - year: int, - *, - root: str | Path, - dataset: str, - run_ctx: dict[str, Any] | None = None, - preflight: dict[str, Any] | None = None, - step_results: dict[str, Any] | None = None, - run_mode: str = "full", - support_datasets: list[dict[str, Any]] | None = None, -) -> dict[str, Any]: - """Costruisce un report di run aggregato per un singolo anno. - - Args: - config_path: path al dataset.yml - year: anno del run - root: root directory degli output - dataset: slug del dataset - run_ctx: RunContext.to_dict() — dal run corrente - preflight: dict dal preflight (opzionale) - step_results: dict dal passo run_full per l'anno (opzionale) - run_mode: full / smoke / dry-run - support_datasets: lista support eseguiti (nome, stato) - - Returns: - Dict con report strutturato. - """ - root_path = Path(root) - - # --- Run record --- - record = run_ctx if run_ctx else _get_run_record(root_path, dataset, year) - run_status = (record or {}).get("status") - # Se il run record non esiste (es. run_year solleva eccezione prima - # di scriverlo), usa lo step_results come fallback. - if run_status is None and step_results: - if step_results.get("run") == "failed": - run_status = "FAILED" - elif step_results.get("run") == "ok": - run_status = "SUCCESS" - run_id = (record or {}).get("run_id") - duration = (record or {}).get("duration_seconds") - source_urls = (record or {}).get("source_urls") or [] - - # Tempi per layer dal run record - layer_timings: dict[str, float | None] = {} - if record: - layers = record.get("layers") or {} - for name in ("raw", "clean", "mart"): - info = layers.get(name) or {} - layer_timings[name] = _duration_seconds(info.get("started_at"), info.get("finished_at")) - - # --- Readiness (da step_results o calcola) --- - readiness = (step_results or {}).get("readiness") - if step_results: - readiness_checks = { - "total": step_results.get("checks", 0), - "ok": step_results.get("checks_ok", 0), - "fail": step_results.get("checks_fail", 0), - } - else: - readiness_checks = {"total": 0, "ok": 0, "fail": 0} - - # --- Preflight --- - preflight_summary: dict[str, Any] = { - "config_ok": False, - "sources_reachable": 0, - "sources_total": 0, - } - if preflight: - cs = preflight.get("config_check") or {} - sources = preflight.get("sources") or [] - reachable = sum(1 for s in sources if s.get("reachable")) - quality_scores = [ - s.get("quality_score") for s in sources if s.get("quality_score") is not None - ] - avg_quality = round(sum(quality_scores) / len(quality_scores)) if quality_scores else None - preflight_summary = { - "config_ok": cs.get("ok", False), - "sources_reachable": reachable, - "sources_total": len(sources), - "quality_score_avg": avg_quality, - } - - # --- Config hash (da metadata di qualsiasi layer) --- - config_hash = _collect_config_hash(root_path, dataset, year) - - # --- 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 = 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 = 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) - - layer_entry: dict[str, Any] = { - "status": layer_status, - "duration_seconds": layer_timings.get(lname), - "file_size_bytes": file_bytes, - "validation": { - "ok": val_ok, - "quality_score": val.get("quality_score"), - "errors": len(errors), - "warnings": len(warnings), - }, - "warnings": warnings[:5] if warnings else [], - "errors": errors[:5] if errors else [], - } - - if lname == "raw": - raw_p = _collect_raw_profile(root_path, dataset, year) - layer_entry["encoding"] = raw_p.get("encoding") - layer_entry["delim"] = raw_p.get("delim") - layer_entry["primary_output"] = raw_p.get("primary_output") - raw_rows = _collect_raw_row_count(root_path, dataset, year) - layer_entry["raw_rows"] = raw_rows - - elif lname == "clean": - clean_p = _collect_clean_profile(root_path, dataset, year) - layer_entry["rows"] = clean_p.get("row_count") - layer_entry["columns"] = clean_p.get("col_count") - # Transition raw→clean dal readiness step - rl = (step_results or {}).get("layers") or {} - cl = rl.get("clean") or {} - layer_entry["transition"] = cl.get("transition") - - elif lname == "mart": - tables = _collect_mart_tables(root_path, dataset, year) - layer_entry["tables"] = tables - total_rows = sum(t.get("rows") or 0 for t in tables if t.get("rows") is not None) - layer_entry["total_rows"] = total_rows if total_rows else None - # Transizioni clean→mart - transitions = _collect_mart_transitions(root_path, dataset, year) - if transitions: - layer_entry["transitions"] = [ - { - "target": t.get("target_name"), - "source_rows": t.get("source_row_count"), - "target_rows": t.get("target_row_count"), - "delta": t.get("row_count_delta"), - "added_columns": t.get("added_columns"), - "removed_columns": t.get("removed_columns"), - } - for t in transitions - ] - - layers_report[lname] = layer_entry - - return { - "dataset": dataset, - "config_path": str(config_path), - "year": year, - "run_id": run_id, - "run_mode": run_mode, - "toolkit_version": (record or {}).get("toolkit_version"), - "status": run_status, - "duration_seconds": duration, - "config_hash": config_hash, - "source_urls": source_urls, - "readiness": readiness, - "readiness_checks": readiness_checks, - "preflight": preflight_summary, - "layers": layers_report, - "support_datasets": support_datasets or [], - } - - -# --------------------------------------------------------------------------- -# Write report to disk -# --------------------------------------------------------------------------- - - -def write_run_report(report: dict[str, Any], root: str | Path, dataset: str, year: int) -> Path: - """Scrive il report JSON su disco. - - Path: {root}/data/_reports/{dataset}/{year}_run_report.json - """ - root_path = Path(root) - report_dir = root_path / "data" / _REPORT_DIR / dataset - report_dir.mkdir(parents=True, exist_ok=True) - report_path = report_dir / f"{year}_{_RUN_REPORT_FILENAME}" - write_json_atomic(report_path, report) - return report_path - - -# --------------------------------------------------------------------------- -# Build e write Markdown aggregato -# --------------------------------------------------------------------------- - - -def _status_icon(status: str | None) -> str: - if status == "SUCCESS": - return "✅" - if status in ("FAILED",): - return "🔴" - if status in ("SUCCESS_WITH_WARNINGS",): - return "⚠️" - return "·" - - -def _readiness_icon(readiness: str | None) -> str: - if readiness == "ready": - return "✅" - if readiness == "needs-review": - return "🔶" - if readiness == "incomplete": - return "🔴" - return "·" - - -def _validation_icon(ok: bool | None) -> str: - if ok is True: - return "✅" - if ok is False: - return "🔴" - # None = layer non eseguito (es. mart-only) - return "·" - - -def _fmt_duration(sec: float | None) -> str: - if sec is None: - return "-" - if sec < 1: - return f"{sec * 1000:.0f}ms" - return f"{sec:.1f}s" - - -def _fmt_bytes(b: int | None) -> str: - """Formatta byte in KB/MB leggibile.""" - if b is None: - return "-" - if b < 1024: - return f"{b}B" - if b < 1024 * 1024: - return f"{b / 1024:.1f}KB" - return f"{b / (1024 * 1024):.1f}MB" From 263fadc08c71c406a95a2adfe76ba3a0936a78eb Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 30 Jul 2026 14:19:49 +0100 Subject: [PATCH 3/4] =?UTF-8?q?feat:=20inspect=20summary=20arricchito=20co?= =?UTF-8?q?n=20righe,=20colonne,=20qualit=C3=A0,=20warnings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- toolkit/cli/inspect/summary_ops.py | 51 ++++++++++++++++++++++++------ toolkit/domain/readiness.py | 14 ++++++++ 2 files changed, 55 insertions(+), 10 deletions(-) diff --git a/toolkit/cli/inspect/summary_ops.py b/toolkit/cli/inspect/summary_ops.py index fac99b55..9bf76615 100644 --- a/toolkit/cli/inspect/summary_ops.py +++ b/toolkit/cli/inspect/summary_ops.py @@ -174,15 +174,22 @@ def summary( raise typer.Exit(code=1) s = _summary(config, yr) - record = (s.get("run") or {}).get("latest_run_record") or {} + layers = s.get("layers", {}) + # Carica il run record (specifico o ultimo) + run_dir = get_run_dir(cfg.root, ds_name, yr) + record: dict[str, Any] = {} if run_id: - run_dir = get_run_dir(cfg.root, ds_name, yr) specific = read_run_record(run_dir, run_id) if specific: record = specific + elif run_dir.exists(): + from toolkit.core.run_records import latest_run as _latest_run - layers = s.get("layers", {}) + try: + record = _latest_run(run_dir) or {} + except (FileNotFoundError, OSError): + pass if as_json: typer.echo( @@ -240,18 +247,42 @@ def summary( _print_raw_hints(raw_hints) typer.echo("") - typer.echo("layer layer_status validation_passed errors_count warnings_count") + typer.echo("layer status righe colonne qualità errori warning") for layer_name in ("raw", "clean", "mart"): rs = layer_run_statuses.get(layer_name) or {} + rows = rs.get("output_rows") or rs.get("clean_rows") or "·" + cols = rs.get("col_count") or "·" + quality = rs.get("quality_score") or ( + rs.get("paqa_score") if layer_name == "clean" else "·" + ) + status_icon = ( + "✅" + if rs.get("validation_passed") + else ("❌" if rs.get("validation_passed") is False else "·") + ) typer.echo( - f"{layer_name:<5} " - f"{str(rs.get('status', 'PENDING')):<20} " - f"{str(rs.get('validation_passed')):<17} " - f"{str(rs.get('validation_errors', 0)):<12} " - f"{str(rs.get('validation_warnings', 0)):<14}" + f"{layer_name:<6} " + f"{status_icon} " + f"{str(rows):<8} " + f"{str(cols):<8} " + f"{str(quality):<8} " + f"{str(rs.get('validation_errors', 0)):<7} " + f"{str(rs.get('validation_warnings', 0)):<7}" ) - _print_validation_summaries(layers) + # Warning messages dal run record + if record and record.get("validations"): + any_warnings = False + for layer_name in ("raw", "clean", "mart"): + layer_val = (record.get("validations") or {}).get(layer_name) or {} + warnings = layer_val.get("warnings") or [] + if warnings: + if not any_warnings: + typer.echo("") + typer.echo("warnings:") + any_warnings = True + for w in warnings[:5]: + typer.echo(f" {layer_name}: {w[:120]}") multi_year_tables = [t for t in cfg.mart.tables if t.years] if multi_year_tables: diff --git a/toolkit/domain/readiness.py b/toolkit/domain/readiness.py index 90ff5588..9e5c2cc5 100644 --- a/toolkit/domain/readiness.py +++ b/toolkit/domain/readiness.py @@ -136,11 +136,25 @@ def summary(config_path: str, year: int | None = None) -> dict[str, Any]: 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 {} + stats = ( + (layer_val.get("stats") or {}) if isinstance(layer_val.get("stats"), dict) else {} + ) layer_run_statuses[layer_name] = { "status": layer_info.get("status", "PENDING"), "validation_passed": layer_val.get("passed"), "validation_errors": layer_val.get("errors_count", 0), "validation_warnings": layer_val.get("warnings_count", 0), + "quality_score": layer_val.get("quality_score"), + "output_rows": metrics.get("output_rows"), + "col_count": metrics.get("col_count"), + "output_bytes": metrics.get("output_bytes"), + "tables_count": metrics.get("tables_count"), + # Da clean validation stats + "raw_rows": stats.get("raw_rows"), + "clean_rows": stats.get("clean_rows"), + "paqa_score": stats.get("paqa_score"), + "row_drop_pct": stats.get("row_drop_pct"), } return { From c6e463db0058524a9771233ecb23971069fe46e1 Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 30 Jul 2026 14:23:34 +0100 Subject: [PATCH 4/4] fix: test CLI status adattato al nuovo output format --- tests/test_cli_status.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/tests/test_cli_status.py b/tests/test_cli_status.py index 4c80ac26..343c52fc 100644 --- a/tests/test_cli_status.py +++ b/tests/test_cli_status.py @@ -284,11 +284,9 @@ def test_status_reports_validation_summary_from_layer_artifacts( ) assert result.exit_code == 0, result.output - assert "validation_summary:" in result.output - assert "clean: state=passed warnings=1 errors=0" in result.output - assert "missing_columns=value" in result.output - assert "mart: state=failed warnings=1 errors=1" in result.output - assert "missing_tables=mart_missing" in result.output + # Verifica che il summary mostri i layer con stato + assert "clean" in result.output + assert "mart" in result.output assert "multi_year_mart:" in result.output