def assert_columns(df: pd.DataFrame, required: list[str], name: str):
missing = [c for c in required if c not in df.columns]
if missing:
raise ValueError(f"[{name}] Faltan columnas requeridas: {missing}")
# Esquema mínimo
assert_columns(clientes, ["cliente_id", "fecha_alta"], "clientes")
assert_columns(productos, ["sku", "nombre", "categoria"], "productos")
assert_columns(sucursales,["sucursal_id", "nombre", "zona"], "sucursales")
assert_columns(ped_pos, ["order_id","fecha","cliente_id","sucursal_id","canal"], "pedidos_pos")
assert_columns(ped_app, ["order_id","fecha","cliente_id","sucursal_id","canal"], "pedidos_app")
assert_columns(lin_pos, ["order_id","line_num","sku","cantidad","precio_unitario"], "lineas_pos")
assert_columns(lin_app, ["order_id","line_num","sku","cantidad","precio_unitario"], "lineas_app")
# Tipos: fechas
for df, col, name in [
(clientes, "fecha_alta", "clientes"),
(ped_pos, "fecha", "pedidos_pos"),
(ped_app, "fecha", "pedidos_app"),
]:
df[col] = pd.to_datetime(df[col], errors="coerce")
if df[col].isna().any():
bad = df[df[col].isna()].head(5)
raise ValueError(f"[{name}] Fechas inválidas en '{col}'. Ejemplos:\n{bad}")
# No nulos en llaves operacionales
def assert_not_null(df: pd.DataFrame, cols: list[str], name: str):
for c in cols:
if df[c].isna().any():
raise ValueError(f"[{name}] Hay nulos en la columna clave: {c}")
assert_not_null(ped_pos, ["order_id","cliente_id","sucursal_id"], "pedidos_pos")
assert_not_null(ped_app, ["order_id","cliente_id","sucursal_id"], "pedidos_app")
assert_not_null(lin_pos, ["order_id","sku"], "lineas_pos")
assert_not_null(lin_app, ["order_id","sku"], "lineas_app")
# Dominios controlados
canales_validos = {"POS","APP"}
if not set(ped_pos["canal"].unique()).issubset(canales_validos):
raise ValueError(f"[pedidos_pos] Canal fuera de dominio: {set(ped_pos['canal'].unique()) - canales_validos}")
if not set(ped_app["canal"].unique()).issubset(canales_validos):
raise ValueError(f"[pedidos_app] Canal fuera de dominio: {set(ped_app['canal'].unique()) - canales_validos}")
# Rangos numéricos
for df, name in [(lin_pos,"lineas_pos"), (lin_app,"lineas_app")]:
if (df["cantidad"] <= 0).any():
raise ValueError(f"[{name}] Se encontraron cantidades <= 0")
if (df["precio_unitario"] < 0).any():
raise ValueError(f"[{name}] Se encontraron precios negativos")
print("Validaciones mínimas: OK")