00_raw_to_bronze_duckdb_polars

import sys
#!{sys.executable} -m pip install duckdb polars
#!{sys.executable} -m pip install watermark -q -U

import warnings
warnings.filterwarnings("ignore")

from __future__ import annotations

import os
import glob
import time
import uuid
from datetime import datetime
from pathlib import Path

import duckdb
import polars as pl

# Versões dos pacotes usados neste jupyter notebook
%reload_ext watermark
%watermark -a "RobertoSSoares-LfLngLrnng"

def ensure_dir(path: str | Path) -> Path:
    """
    Garante que um diretório exista.
    """
    path = Path(path)
    path.mkdir(parents=True, exist_ok=True)
    return path


def bytes_to_mb(num_bytes: int) -> float:
    """
    Converte bytes para megabytes.
    """
    return round(num_bytes / (1024 * 1024), 4)


def safe_file_size_mb(path: str | Path) -> float | None:
    """
    Retorna o tamanho do arquivo em MB, ou None se o arquivo não existir.
    """
    path = Path(path)
    if path.exists():
        return bytes_to_mb(path.stat().st_size)
    return None


def normalize_output_name(file_path: str | Path) -> str:
    """
    Gera o nome do arquivo parquet de saída a partir do CSV de entrada.
    """
    return Path(file_path).stem + ".parquet"


def now_str() -> str:
    """
    Timestamp amigável para auditoria.
    """
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# Função principal (VERSÃO POLARS)
def raw_to_bronze_duckdb(
    raw_dir: str = "../data/00-raw",
    bronze_dir: str = "../data/01-bronze",
    audit_dir: str = "../data/99-audit",
    glob_pattern: str = "*.csv",
    sample_size: int = -1,
    all_varchar: bool = False,
    ignore_errors: bool = False,
    filename: bool = True,
    union_by_name: bool = False,
    compression: str = "zstd",
    overwrite: bool = True,
    verbose: bool = True,
) -> pl.DataFrame:
    """
    Ingestão RAW -> BRONZE usando DuckDB + auditoria em Polars.
    """

    run_id = datetime.now().strftime("%Y%m%d_%H%M%S") + "_" + uuid.uuid4().hex[:8]
    ingestion_ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    raw_path = Path(raw_dir)
    bronze_path = ensure_dir(bronze_dir)
    audit_path = ensure_dir(audit_dir)

    csv_files = sorted(glob.glob(str(raw_path / glob_pattern)))

    if not csv_files:
        raise FileNotFoundError(f"Nenhum CSV encontrado em {raw_dir}")

    if verbose:
        print("=" * 90)
        print("🚀 RAW → BRONZE (DuckDB + Polars)")
        print(f"Run ID: {run_id}")
        print("=" * 90)

    audit_rows = []
    start_all = time.time()

    con = duckdb.connect()

    for file_path in csv_files:
        start_file = time.time()

        source_path = Path(file_path)
        target_path = bronze_path / (source_path.stem + ".parquet")

        if verbose:
            print(f"📦 {source_path.name}")

        try:
            csv_size_mb = safe_file_size_mb(source_path)

            read_csv_relation = f"""
                read_csv_auto(
                    '{str(source_path).replace("\\", "/")}',
                    sample_size={sample_size},
                    all_varchar={str(all_varchar).lower()},
                    ignore_errors={str(ignore_errors).lower()}
                )"""

            if target_path.exists() and overwrite:
                target_path.unlink()

            # COPY → Parquet
            con.execute(f"""
                COPY (
                    SELECT *
                    FROM {read_csv_relation}
                )
                TO '{str(target_path).replace("\\", "/")}'
                (FORMAT PARQUET, COMPRESSION '{compression}');
            """)

            # Contagens
            csv_rows = con.execute(
                f"SELECT COUNT(*) FROM {read_csv_relation}"
            ).fetchone()[0]

            parquet_rows = con.execute(
                f"SELECT COUNT(*) FROM read_parquet('{str(target_path).replace("\\", "/")}')"
            ).fetchone()[0]

            column_count = con.execute(f"""
                SELECT COUNT(*) FROM (
                    DESCRIBE SELECT * FROM read_parquet('{str(target_path).replace("\\", "/")}')
                )
            """).fetchone()[0]

            parquet_size_mb = safe_file_size_mb(target_path)

            sanity = (
                target_path.exists()
                and csv_rows == parquet_rows
                and column_count > 0
            )

            compression_ratio = None
            if csv_size_mb and parquet_size_mb:
                compression_ratio = round(csv_size_mb / parquet_size_mb, 4)

            audit_rows.append({
                "run_id": run_id,
                "file": source_path.name,
                "status": "SUCCESS" if sanity else "WARNING",
                "csv_rows": csv_rows,
                "parquet_rows": parquet_rows,
                "columns": column_count,
                "csv_mb": csv_size_mb,
                "parquet_mb": parquet_size_mb,
                "compression_ratio": compression_ratio,
                "sanity": sanity,
                "elapsed": round(time.time() - start_file, 4),
            })

        except Exception as e:
            audit_rows.append({
                "run_id": run_id,
                "file": source_path.name,
                "status": "ERROR",
                "error": str(e),
                "elapsed": round(time.time() - start_file, 4),
            })

            if verbose:
                print(f"❌ {e}")

    con.close()

    # 🔥 Aqui entra o Polars
    audit_df = pl.DataFrame(audit_rows)

    # Persistência
    audit_df.write_csv(audit_path / "raw_to_bronze_audit.csv")
    audit_df.write_parquet(audit_path / "raw_to_bronze_audit.parquet")

    if verbose:
        print("=" * 90)
        print("🏁 FINALIZADO")
    
        if audit_df.height > 0 and "status" in audit_df.columns:
            print("Resumo por status:")
            print(
                audit_df
                .group_by("status")
                .len()
                .rename({"len": "qtd_arquivos"})
                .sort("status")
            )
        else:
            print("Nenhum registro de auditoria foi gerado.")
    
        print("=" * 90)

    return audit_df

# Execução
audit_df = raw_to_bronze_duckdb(
    raw_dir="../data/00-raw",
    bronze_dir="../data/01-bronze",
    audit_dir="../data/99-audit",
    glob_pattern="*.csv",
    sample_size=-1,      # inferência mais confiável
    all_varchar=False,   # mantém inferência automática
    ignore_errors=False, # mais rigoroso
    compression="zstd",
    overwrite=True,
    verbose=True,
)

audit_df

# Schema Registry (LAZY)

def build_schema_registry(bronze_dir="../data/01-bronze", audit_dir="../data/99-audit"):
    rows = []

    for file in Path(bronze_dir).glob("*.parquet"):
        schema = pl.scan_parquet(str(file)).collect_schema()

        for i, (col, dtype) in enumerate(schema.items(), start=1):
            rows.append({
                "table": file.stem,
                "column": col,
                "dtype": str(dtype),
                "order": i
            })

    df = pl.DataFrame(rows)

    df.write_parquet(Path(audit_dir) / "schema_registry.parquet")
    df.write_csv(Path(audit_dir) / "schema_registry.csv")

    return df

schema_df = build_schema_registry()
display(schema_df.head(5))
print(" ")
display(schema_df.sample(5))

# Profiling de Colunas
def profile_bronze(bronze_dir="../data/01-bronze", out_dir="../data/90-profiling"):
    rows = []

    for file in Path(bronze_dir).glob("*.parquet"):
        df = pl.read_parquet(file)

        for col in df.columns:
            s = df[col]

            rows.append({
                "table": file.stem,
                "column": col,
                "dtype": str(s.dtype),
                "rows": df.height,
                "nulls": s.null_count(),
                "null_pct": s.null_count() / df.height * 100 if df.height else 0,
                "unique": s.n_unique(),
            })

    df = pl.DataFrame(rows)
    ensure_dir(out_dir)

    df.write_parquet(Path(out_dir) / "column_profile.parquet")
    df.write_csv(Path(out_dir) / "column_profile.csv")

    return df


profile_df = profile_bronze()
display(profile_df.head(5))
print()
display(profile_df.sample(5))

# Null Analysis
null_df = (profile_df.sort("null_pct", descending=True))
null_df.head(10)

# Testes Globais

assert audit_df.filter(pl.col("status") == "ERROR").height == 0
assert profile_df.height > 0
assert schema_df.height > 0

print("✅ Pipeline RAW → BRONZE validado com sucesso")

%reload_ext watermark 
%watermark -a "Roberto-SSoares-LfLngLrnng" -d -t -u --iversions -v -m -h

#!uv pip install nbconvert -U -q
!jupyter nbconvert 03_raw_to_bronze_duckdb_polars.ipynb --to html --template _my-template-html-v07.tpl