Pipeline ETL — NutriTrack SaaS

import polars as pl
from pathlib import Path

DATA = Path("data/nutritrack")

# Lazy scan — no lee nada todavia
lf_subs = pl.scan_csv(
    DATA / "subscriptions.csv",
    try_parse_dates=True
)

lf_events = pl.scan_parquet(
    DATA / "usage_events.parquet"
)

lf_plans = pl.scan_csv(
    DATA / "plans.csv"
)

# Polars sabe cuantas filas SIN leer
print(lf_events.collect_schema())

# MRR actual + variacion mensual por plan
mrr_df = (
    lf_subs
    .filter(pl.col("status") == "active")
    .join(lf_plans, on="plan_id", how="left")
    .group_by("plan_name", "month")
    .agg(
        pl.len().alias("n_clientes"),
        pl.col("price_eur").sum().alias("mrr_eur"),
    )
    .with_columns(
        # Variacion % vs mes anterior (window)
        mrr_prev=pl.col("mrr_eur")
            .shift(1)
            .over("plan_name"),
    )
    .with_columns(
        delta_pct=(
            (pl.col("mrr_eur") - pl.col("mrr_prev"))
            / pl.col("mrr_prev") * 100
        ).round(2)
    )
    .collect()  # ejecuta todo en paralelo
)

# Clientes con uso < 5 sesiones/mes = riesgo
usage_per_client = (
    lf_events
    .filter(
        pl.col("event_month") == "2026-05"
    )
    .group_by("company_id")
    .agg(
        pl.len().alias("sessions")
    )
    .collect()
)

# Clasificar riesgo con pl.when
at_risk = usage_per_client.with_columns(
    riesgo=pl.when(
        pl.col("sessions") < 3
    ).then(pl.lit("alto"))
    .when(
        pl.col("sessions") < 5
    ).then(pl.lit("medio"))
    .otherwise(pl.lit("bajo"))
).filter(
    pl.col("riesgo") != "bajo"
)

# Reporte final consolidado para BI
report = (
    lf_subs
    .join(
        lf_plans, on="plan_id", how="left"
    )
    .join(
        usage_per_client.lazy(),
        on="company_id", how="left"
    )
    .select(
        "company_id", "company_name",
        "plan_name", "price_eur",
        "sessions", "status",
        pl.col("start_date").dt.strftime("%Y-%m")
            .alias("mes_inicio")
    )
    .collect(engine="streaming")
)
# Parquet comprimido ~8x vs CSV
report.write_parquet(
    "output/nutritrack_report.parquet",
    compression="zstd"
)