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}] Missing required columns: {missing}")
# Minimal schema assertions
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")
# Date type casting and validation
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}] Invalid dates in '{col}'. Examples:\n{bad}")
# Non-null operational key enforcement
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}] Null values found in key column: {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")
# Controlled domain validation
valid_channels = {"POS","APP"}
if not set(ped_pos["canal"].unique()).issubset(valid_channels):
raise ValueError(f"[pedidos_pos] Channel out of domain: {set(ped_pos['canal'].unique()) - valid_channels}")
if not set(ped_app["canal"].unique()).issubset(valid_channels):
raise ValueError(f"[pedidos_app] Channel out of domain: {set(ped_app['canal'].unique()) - valid_channels}")
# Numeric range assertions
for df, name in [(lin_pos,"lineas_pos"), (lin_app,"lineas_app")]:
if (df["cantidad"] <= 0).any():
raise ValueError(f"[{name}] Quantities <= 0 detected")
if (df["precio_unitario"] < 0).any():
raise ValueError(f"[{name}] Negative unit prices detected")
print("Minimum quality checks: OK")