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/ 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..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) @@ -304,17 +269,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/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"], diff --git a/toolkit/raw/run.py b/toolkit/raw/run.py index 95e341e4..9224d452 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( @@ -253,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, 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)