# Celda 1: Definir la configuración del catálogo y esquemas
catalog_name = "sesion_5"
bronze_schema = "bronze"
# ... (y así sucesivamente para silver, gold, auditoria)

# Celda 2: Crear la infraestructura en Unity Catalog
spark.sql(f"CREATE CATALOG IF NOT EXISTS {catalog_name}")
# ... (creación de esquemas)

# Celda 3: Crear el volumen para los archivos crudos
bronze_volume_path = f"/Volumes/{catalog_name}/{bronze_schema}/raw_files"
spark.sql(f"CREATE VOLUME IF NOT EXISTS {catalog_name}.{bronze_schema}.raw_files")

print(f"Por favor, cargue los archivos CSV del dataset Olist en: {bronze_volume_path}")

# Celda 2: Función de ayuda para cargar y escribir tablas
def ingest_csv_to_bronze(file_name, table_name):
    df = spark.read.option("header", "true").option("inferSchema", "true").csv(f"{bronze_volume_path}/{file_name}")
    df.write.format("delta").mode("overwrite").saveAsTable(f"{catalog_name}.{bronze_schema}.{table_name}")
    print(f"Tabla '{table_name}' creada en la capa Bronce.")

# Celda 3: Ejecutar la ingesta para cada archivo
ingest_csv_to_bronze("olist_customers_dataset.csv", "customers_bronze")
# ... (y así para el resto de archivos)

# Celda 1: Función de ayuda para validaciones de calidad
def ejecutar_validacion_calidad(query, descripcion_validacion):
  print(f"Ejecutando validación: '{descripcion_validacion}'...")
  df_resultado = spark.sql(query)
  conteo_filas_malas = df_resultado.first()[0]
  
  if conteo_filas_malas > 0:
    raise Exception(f"VALIDACIÓN FALLIDA: Se encontraron {conteo_filas_malas} filas que violan la regla.")
  else:
    print(f"VALIDACIÓN EXITOSA ✔️")

# Celda 3: Ejecutar las reglas de negocio
# REGLA 1: La llave primaria de un cliente NUNCA debe ser nula.
query_pk_nula_cliente = "SELECT COUNT(*) FROM customers_bronze_vw WHERE customer_id IS NULL"
ejecutar_validacion_calidad(query_pk_nula_cliente, "La columna 'customer_id' no debe contener nulos.")

# REGLA 3: No debe haber pedidos sin al menos un item.
query_pedidos_sin_items = """
  SELECT COUNT(o.order_id)
  FROM orders_bronze_vw o
  LEFT JOIN order_items_bronze_vw i ON o.order_id = i.order_id
  WHERE i.order_id IS NULL
"""
ejecutar_validacion_calidad(query_pedidos_sin_items, "Existen pedidos que no tienen ningún item asociado.")

-- Celda 2: Crear Dimensiones (dim_customers, dim_products)
CREATE OR REPLACE TABLE dim_products AS
SELECT
  p.product_id,
  p.product_category_name,
  t.product_category_name_english AS product_category_name_en,
  ...
FROM sesion_5.bronze.products_bronze p
LEFT JOIN sesion_5.bronze.category_translation_bronze t ON p.product_category_name = t.product_category_name;

-- Celda 3: Crear Tabla de Hechos (fact_orders) particionada
CREATE OR REPLACE TABLE fact_orders
PARTITIONED BY (order_purchase_year)
AS
SELECT ...
FROM sesion_5.bronze.orders_bronze o
JOIN sesion_5.bronze.order_items_bronze i ON o.order_id = i.order_id
...

-- Celda 2: Crear tabla de ventas mensuales por categoría
CREATE OR REPLACE TABLE monthly_sales_by_category_gold AS
SELECT
  YEAR(fo.order_purchase_date) AS anio,
  MONTH(fo.order_purchase_date) AS mes,
  dp.product_category_name_en AS categoria_producto,
  COUNT(DISTINCT fo.order_id) AS numero_pedidos,
  SUM(fo.payment_value) AS ingresos_totales
FROM silver.fact_orders fo
JOIN silver.dim_products dp ON fo.product_id = dp.product_id
WHERE fo.order_status = 'delivered'
GROUP BY anio, mes, categoria_producto;