Validar datos con Python y Duckdb

import pandas as pd
import duckdb
import json

def validate_csv_with_duckdb(csv_path: str, validations_path: str):
    print("\n--- INICIANDO VALIDACIÓN DE DATOS ---")
    try:
        df = pd.read_csv(csv_path)
        print(f"CSV '{csv_path}' cargado exitosamente.")
    except FileNotFoundError:
        print(f"ERROR: Archivo CSV no encontrado en '{csv_path}'")
        return
    except Exception as e:
        print(f"ERROR al cargar CSV: {e}")
        return

    try:
        with open(validations_path, 'r', encoding='utf-8') as f:
            validations = json.load(f)
        print(f"Cargadas {len(validations)} validaciones desde '{validations_path}'.")
    except FileNotFoundError:
        print(f"ERROR: Archivo de validaciones no encontrado en '{validations_path}'")
        return
    except json.JSONDecodeError:
        print(f"ERROR: El archivo '{validations_path}' no es un JSON válido.")
        return

    # Conectar DuckDB y registrar el DataFrame
    con = duckdb.connect(database=':memory:', read_only=False)
    con.register('data_to_validate', df) # Registramos el DF con el nombre que usamos en las validacione JSON

    failed_validations = []

    # Ejecutar cada validación y recolectar fallos
    for validation in validations:
        name = validation.get('name', 'Validación sin nombre')
        query_sql = validation.get('query_sql')

        if not query_sql:
            print(f"ADVERTENCIA: Validación '{name}' sin query SQL. Saltando.")
            continue

        try:
            # Ejecutar la query. Si devuelve filas, la validación falló.
            failed_records = con.execute(query_sql).fetchdf()

            if not failed_records.empty:
                failed_validations.append(name)
                print(f"  FALLÓ: '{name}' - Se encontraron {len(failed_records)} problemas.")
        except Exception as e:
            print(f"ERROR al ejecutar query para '{name}': {e}")
            failed_validations.append(f"{name} (ERROR SQL)")
            # No queremos que un error en una query detenga todo, solo lo reportamos.

    con.close() # Cerrar la conexión a DuckDB

    # Resultado final
    print("\n--- REPORTE FINAL ---")
    if not failed_validations:
        print("\033[92m✅ ¡Todas las validaciones pasaron correctamente! Los datos están limpios.\033[0m") # Verde
    else:
        print("\033[91m❌ Se encontraron problemas de validación en los siguientes puntos:\033[0m") # Rojo
        for validation_name in failed_validations:
            print(f"  - {validation_name}")
        print("\n\033[93mPor favor, revisa los datos y las validaciones fallidas.\033[0m") # Amarillo

    print("\n--- VALIDACIÓN FINALIZADA ---")

if __name__ == "__main__":
    CSV_FILE = '/content/sample_data/data.csv'
    VALIDATIONS_FILE = '/content/sample_data/validations.json'

    validate_csv_with_duckdb(CSV_FILE, VALIDATIONS_FILE)