# /etc/crontab de la VM vm-procesos-nocturnos
0  2 * * * /opt/scripts/exportar_cloudsql.sh
30 2 * * * /opt/scripts/cargar_bigquery.sh
0  3 * * * /opt/scripts/lanzar_dataflow.sh
0  4 * * * /opt/scripts/agregaciones.sh
30 4 * * * /opt/scripts/refrescar_vista.sh

gcloud config set project alpinashop-datos
gcloud services enable composer.googleapis.com

gcloud iam service-accounts create sa-composer \
  --display-name="Cloud Composer de AlpinaShop"

SA_COMP="[email protected]"

for ROL in roles/composer.worker roles/bigquery.dataEditor roles/bigquery.jobUser \
           roles/dataflow.developer roles/storage.objectAdmin \
           roles/cloudsql.viewer roles/dataproc.editor; do
  gcloud projects add-iam-policy-binding alpinashop-datos \
    --member="serviceAccount:${SA_COMP}" --role="$ROL"
done

gcloud composer environments create alpinashop-composer \
  --location=europe-west1 \
  --image-version=composer-3-airflow-2.10.5 \
  --service-account="$SA_COMP" \
  --network=alpinashop-vpc \
  --subnetwork=sn-datos-euw1 \
  --enable-private-environment \
  --environment-size=small \
  --labels=entorno=produccion,equipo=datos,centro-coste=analitica

BUCKET_DAGS=$(gcloud composer environments describe alpinashop-composer \
  --location=europe-west1 --format="value(config.dagGcsPrefix)")

gcloud storage cp dags/proceso_nocturno.py "${BUCKET_DAGS}/"

"""
proceso_nocturno.py -- Proceso nocturno de datos de AlpinaShop.

Flujo:
  exportar Cloud SQL -> esperar fichero -> cargar BigQuery -> Dataflow
  -> agregaciones (paralelas) -> refrescar vista -> comprobar calidad

Despliegue: copiar a gs://<bucket-composer>/dags/
"""
from datetime import datetime, timedelta

import pendulum
from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.operators.bigquery import (
    BigQueryCheckOperator,
    BigQueryInsertJobOperator,
)
from airflow.providers.google.cloud.operators.cloud_sql import (
    CloudSQLExportInstanceOperator,
)
from airflow.providers.google.cloud.operators.dataflow import (
    DataflowStartFlexTemplateOperator,
)
from airflow.providers.google.cloud.sensors.gcs import GCSObjectExistenceSensor
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import (
    GCSToBigQueryOperator,
)
from airflow.utils.task_group import TaskGroup

PROYECTO_DATOS = "alpinashop-datos"
PROYECTO_PROD = "alpinashop-prod"
DATASET = "alpinashop_analitica"
BUCKET = "alpinashop-datalake"
REGION = "europe-west1"
ZONA_HORARIA = pendulum.timezone("Europe/Madrid")

# ---------------------------------------------------------------------------
# ARGUMENTOS POR DEFECTO: se aplican a TODAS las tareas del DAG.
# Definirlos aqui evita repetirlos en cada operador.
# ---------------------------------------------------------------------------
argumentos_por_defecto = {
    "owner": "equipo-datos",
    "depends_on_past": False,      # una ejecucion no espera a la anterior
    "email": ["[email protected]"],
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 3,                  # 3 reintentos ante fallo
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,   # 5, 10, 20 min: no machacar el origen
    "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(hours=2),   # ninguna tarea eterna
    "sla": timedelta(hours=3),     # aviso si el DAG no acaba en 3 h
}

with DAG(
    dag_id="alpinashop_proceso_nocturno",
    description="Proceso nocturno: Cloud SQL -> BigQuery -> Dataflow -> agregados",
    default_args=argumentos_por_defecto,
    # Todos los dias a las 02:00 hora de Madrid.
    # Airflow trabaja internamente en UTC; la zona horaria evita el desfase
    # de una hora en los cambios de horario de verano.
    schedule="0 2 * * *",
    start_date=datetime(2026, 3, 1, tzinfo=ZONA_HORARIA),
    catchup=False,             # ver apartado 10
    max_active_runs=1,         # nunca dos noches solapadas
    tags=["alpinashop", "produccion", "datos"],
    doc_md=__doc__,            # la docstring se ve en la interfaz
) as dag:

    inicio = EmptyOperator(task_id="inicio")

    # -----------------------------------------------------------------------
    # 1) EXPORTAR Cloud SQL a Cloud Storage
    #
    # {{ ds }} es una plantilla Jinja: Airflow la sustituye por la fecha
    # logica de la ejecucion (yyyy-MM-dd). Es LA pieza que hace el DAG
    # reproducible: al reprocesar el 12 de marzo, {{ ds }} vale 2026-03-12
    # y el fichero de salida y la consulta apuntan a ese dia, no a hoy.
    # -----------------------------------------------------------------------
    exportar_pedidos = CloudSQLExportInstanceOperator(
        task_id="exportar_pedidos_cloudsql",
        project_id=PROYECTO_PROD,
        instance="alpinashop-pedidos-replica-informes",
        body={
            "exportContext": {
                "fileType": "CSV",
                "uri": f"gs://{BUCKET}/exportaciones/{{{{ ds_nodash }}}}/pedidos.csv",
                "databases": ["tienda"],
                "csvExportOptions": {
                    "selectQuery": (
                        "SELECT pedido_id, creado_en, cliente_id, canal, estado, "
                        "pais, ciudad, codigo_postal, metodo_pago, "
                        "subtotal, descuento, iva, total "
                        "FROM pedidos "
                        "WHERE DATE(creado_en) = '{{ ds }}'"
                    )
                },
            }
        },
    )

    # -----------------------------------------------------------------------
    # 2) SENSOR: esperar a que el fichero exista de verdad.
    #
    # Esta tarea es la que resuelve el problema del cron. La exportacion
    # es asincrona: el operador anterior la lanza, pero el fichero puede
    # tardar. En vez de "esperamos 30 minutos y cruzamos los dedos",
    # se COMPRUEBA.
    # -----------------------------------------------------------------------
    esperar_fichero = GCSObjectExistenceSensor(
        task_id="esperar_fichero_exportado",
        bucket=BUCKET,
        object="exportaciones/{{ ds_nodash }}/pedidos.csv",
        # 'reschedule': libera el trabajador entre comprobaciones en vez de
        # ocuparlo esperando. Imprescindible en esperas largas.
        mode="reschedule",
        poke_interval=60,          # comprobar cada minuto
        timeout=60 * 60,           # rendirse a la hora
    )

    # -----------------------------------------------------------------------
    # 3) CARGAR en BigQuery
    # -----------------------------------------------------------------------
    cargar_pedidos = GCSToBigQueryOperator(
        task_id="cargar_pedidos_bigquery",
        bucket=BUCKET,
        source_objects=["exportaciones/{{ ds_nodash }}/pedidos.csv"],
        destination_project_dataset_table=f"{PROYECTO_DATOS}.{DATASET}.pedidos_staging",
        source_format="CSV",
        skip_leading_rows=0,
        field_delimiter=",",
        null_marker="\\N",
        # WRITE_TRUNCATE en la tabla de staging: cada noche se reemplaza.
        # Esto hace la tarea IDEMPOTENTE: reejecutarla no duplica nada.
        write_disposition="WRITE_TRUNCATE",
        create_disposition="CREATE_IF_NEEDED",
        autodetect=False,
        schema_fields=[
            {"name": "pedido_id", "type": "STRING", "mode": "REQUIRED"},
            {"name": "creado_en", "type": "TIMESTAMP", "mode": "REQUIRED"},
            {"name": "cliente_id", "type": "STRING"},
            {"name": "canal", "type": "STRING"},
            {"name": "estado", "type": "STRING"},
            {"name": "pais", "type": "STRING"},
            {"name": "ciudad", "type": "STRING"},
            {"name": "codigo_postal", "type": "STRING"},
            {"name": "metodo_pago", "type": "STRING"},
            {"name": "subtotal", "type": "NUMERIC"},
            {"name": "descuento", "type": "NUMERIC"},
            {"name": "iva", "type": "NUMERIC"},
            {"name": "total", "type": "NUMERIC"},
        ],
        location=REGION,
    )

    # -----------------------------------------------------------------------
    # 4) MERGE del staging a la tabla final: idempotente por diseno
    # -----------------------------------------------------------------------
    consolidar_pedidos = BigQueryInsertJobOperator(
        task_id="consolidar_pedidos",
        location=REGION,
        configuration={
            "query": {
                "query": f"""
                    MERGE `{PROYECTO_DATOS}.{DATASET}.pedidos` AS destino
                    USING (
                      SELECT
                        pedido_id,
                        DATE(creado_en) AS fecha_pedido,
                        creado_en       AS momento_pedido,
                        cliente_id, canal, estado,
                        STRUCT(pais, NULL AS provincia, ciudad, codigo_postal,
                               NULL AS metodo, CAST(NULL AS NUMERIC) AS coste) AS envio,
                        metodo_pago, NULL AS cupon,
                        subtotal, descuento, iva, total AS total_pedido
                      FROM `{PROYECTO_DATOS}.{DATASET}.pedidos_staging`
                    ) AS origen
                    ON destino.pedido_id = origen.pedido_id
                       AND destino.fecha_pedido = origen.fecha_pedido
                    WHEN MATCHED THEN UPDATE SET
                      estado = origen.estado,
                      total_pedido = origen.total_pedido
                    WHEN NOT MATCHED THEN INSERT ROW
                """,
                "useLegacySql": False,
            }
        },
    )

    # -----------------------------------------------------------------------
    # 5) DATAFLOW: plantilla flexible creada en 04-02
    # -----------------------------------------------------------------------
    lanzar_dataflow = DataflowStartFlexTemplateOperator(
        task_id="lanzar_pipeline_dataflow",
        location=REGION,
        project_id=PROYECTO_DATOS,
        body={
            "launchParameter": {
                "jobName": "pedidos-lote-{{ ds_nodash }}",
                "containerSpecGcsPath":
                    f"gs://alpinashop-dataflow/plantillas/pedidos-lote.json",
                "parameters": {"fecha": "{{ ds }}"},
                "environment": {
                    "serviceAccountEmail":
                        "[email protected]",
                    "subnetwork":
                        f"regions/{REGION}/subnetworks/sn-datos-euw1",
                    "ipConfiguration": "WORKER_IP_PRIVATE",
                    "maxWorkers": 10,
                    "additionalUserLabels": {
                        "entorno": "produccion", "equipo": "datos",
                    },
                },
            }
        },
        # El operador ESPERA a que el job termine antes de dar la tarea
        # por completada. Sin esto, el DAG seguiria con datos a medias.
        wait_until_finished=True,
    )

    # -----------------------------------------------------------------------
    # 6) AGREGACIONES EN PARALELO, agrupadas para que el grafo sea legible
    # -----------------------------------------------------------------------
    with TaskGroup(group_id="agregaciones") as grupo_agregaciones:

        agregar_ventas = BigQueryInsertJobOperator(
            task_id="ventas_por_categoria",
            location=REGION,
            configuration={
                "query": {
                    "query": f"""
                        CREATE OR REPLACE TABLE
                          `{PROYECTO_DATOS}.{DATASET}.agg_ventas_categoria_dia`
                        PARTITION BY dia AS
                        SELECT
                          l.fecha_pedido                 AS dia,
                          pr.categoria,
                          COUNT(DISTINCT l.pedido_id)    AS pedidos,
                          SUM(l.cantidad)                AS unidades,
                          ROUND(SUM(l.importe_linea), 2) AS ventas_eur
                        FROM `{PROYECTO_DATOS}.{DATASET}.lineas_pedido` AS l
                        JOIN `{PROYECTO_DATOS}.{DATASET}.productos`     AS pr
                          USING (sku)
                        WHERE l.fecha_pedido >= DATE_SUB(DATE '{{{{ ds }}}}',
                                                         INTERVAL 400 DAY)
                        GROUP BY dia, pr.categoria
                    """,
                    "useLegacySql": False,
                }
            },
        )

        agregar_visitas = BigQueryInsertJobOperator(
            task_id="embudo_conversion",
            location=REGION,
            configuration={
                "query": {
                    "query": f"""
                        CREATE OR REPLACE TABLE
                          `{PROYECTO_DATOS}.{DATASET}.agg_embudo_dia`
                        PARTITION BY dia AS
                        WITH marcas AS (
                          SELECT
                            fecha AS dia, sesion_id,
                            LOGICAL_OR(e.tipo = 'ver_producto')   AS vio,
                            LOGICAL_OR(e.tipo = 'anadir_carrito') AS anadio,
                            LOGICAL_OR(e.tipo = 'iniciar_pago')   AS pago,
                            LOGICAL_OR(e.tipo = 'compra')         AS compro
                          FROM `{PROYECTO_DATOS}.{DATASET}.visitas`,
                               UNNEST(eventos) AS e
                          WHERE fecha BETWEEN DATE_SUB(DATE '{{{{ ds }}}}',
                                                       INTERVAL 90 DAY)
                                          AND DATE '{{{{ ds }}}}'
                          GROUP BY dia, sesion_id
                        )
                        SELECT
                          dia,
                          COUNT(*)             AS sesiones,
                          COUNTIF(vio)         AS vieron_producto,
                          COUNTIF(anadio)      AS anadieron_carrito,
                          COUNTIF(pago)        AS iniciaron_pago,
                          COUNTIF(compro)      AS compraron
                        FROM marcas
                        GROUP BY dia
                    """,
                    "useLegacySql": False,
                }
            },
        )

    # -----------------------------------------------------------------------
    # 7) REFRESCAR la vista materializada que consume el cuadro de mando
    # -----------------------------------------------------------------------
    refrescar_vista = BigQueryInsertJobOperator(
        task_id="refrescar_vista_negocio",
        location=REGION,
        configuration={
            "query": {
                "query": f"""
                    CALL BQ.REFRESH_MATERIALIZED_VIEW(
                      '{PROYECTO_DATOS}.{DATASET}.mv_ventas_diarias_sku')
                """,
                "useLegacySql": False,
            }
        },
    )

    # -----------------------------------------------------------------------
    # 8) COMPROBACION DE CALIDAD: la ultima defensa antes del informe.
    #
    # Si esta tarea falla, el DAG queda marcado como fallido y salta la
    # alerta. Es preferible una alerta a las 05:00 que un comite de
    # direccion mirando cifras falsas a las 09:00.
    # -----------------------------------------------------------------------
    comprobar_calidad = BigQueryCheckOperator(
        task_id="comprobar_calidad_datos",
        location=REGION,
        use_legacy_sql=False,
        sql=f"""
            SELECT
              COUNTIF(pedidos_dia = 0)  = 0 AND
              COUNTIF(ventas_dia < 0)   = 0 AND
              COUNTIF(pedidos_dia > 500) = 0
            FROM (
              SELECT
                dia,
                SUM(pedidos)    AS pedidos_dia,
                SUM(ventas_eur) AS ventas_dia
              FROM `{PROYECTO_DATOS}.{DATASET}.agg_ventas_categoria_dia`
              WHERE dia = DATE '{{{{ ds }}}}'
              GROUP BY dia
            )
        """,
    )

    def _avisar_exito(**contexto):
        """Callback informativo: en produccion enviaria a Slack o Chat."""
        print(f"Proceso nocturno completado para {contexto['ds']}")

    fin = PythonOperator(
        task_id="fin",
        python_callable=_avisar_exito,
        trigger_rule="all_success",
    )

    # -----------------------------------------------------------------------
    # DEPENDENCIAS: el operador >> significa "y despues".
    # Esta es la declaracion completa del grafo, en cuatro lineas.
    # -----------------------------------------------------------------------
    (
        inicio
        >> exportar_pedidos
        >> esperar_fichero
        >> cargar_pedidos
        >> consolidar_pedidos
        >> lanzar_dataflow
        >> grupo_agregaciones
        >> refrescar_vista
        >> comprobar_calidad
        >> fin
    )

from airflow.providers.google.cloud.operators.dataproc import (
    DataprocCreateBatchOperator,
)

analisis_cesta = DataprocCreateBatchOperator(
    task_id="analisis_cesta_mensual",
    region=REGION,
    project_id=PROYECTO_DATOS,
    batch_id="cesta-{{ ds_nodash }}",
    batch={
        "pyspark_batch": {
            "main_python_file_uri": f"gs://{BUCKET}/jobs/cesta_media.py",
            "args": [
                "--fecha-desde={{ macros.ds_add(ds, -30) }}",
                "--fecha-hasta={{ ds }}",
            ],
        },
        "runtime_config": {"version": "2.2"},
        "environment_config": {
            "execution_config": {
                "service_account":
                    "[email protected]",
                "subnetwork_uri": "sn-datos-euw1",
            }
        },
    },
)

def _contar_filas(**contexto):
    from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook
    hook = BigQueryHook(location=REGION, use_legacy_sql=False)
    filas = hook.get_first(
        f"""SELECT COUNT(*) FROM `{PROYECTO_DATOS}.{DATASET}.pedidos_staging`"""
    )[0]
    # Lo devuelto por la funcion se guarda automaticamente como XCom
    return int(filas)


def _validar_volumen(**contexto):
    filas = contexto["ti"].xcom_pull(task_ids="contar_filas_cargadas")
    if filas == 0:
        raise ValueError(f"Ningun pedido cargado el {contexto['ds']}")
    if filas > 5000:
        raise ValueError(f"{filas} pedidos: volumen anomalo, revisar duplicados")
    print(f"Volumen correcto: {filas} pedidos")


contar_filas = PythonOperator(task_id="contar_filas_cargadas",
                              python_callable=_contar_filas)
validar_volumen = PythonOperator(task_id="validar_volumen",
                                 python_callable=_validar_volumen)

from airflow.models import Variable

umbral = int(Variable.get("alpinashop_umbral_pedidos_dia", default_var="500"))

gcloud composer environments run alpinashop-composer \
  --location=europe-west1 variables -- set alpinashop_umbral_pedidos_dia 500

a >> b                    # b despues de a
a << b                    # a despues de b
a >> [b, c] >> d          # b y c en paralelo, ambos tras a; d tras ambos
[a, b] >> c               # c espera a que terminen a y b

from airflow.providers.google.cloud.operators.dataproc import (
    DataprocDeleteClusterOperator,
)

borrar_cluster = DataprocDeleteClusterOperator(
    task_id="borrar_cluster",
    cluster_name="cluster-efimero-{{ ds_nodash }}",
    region=REGION,
    project_id=PROYECTO_DATOS,
    # CRITICO: se borra el cluster PASE LO QUE PASE.
    # Con all_success, un fallo del job dejaria el cluster encendido
    # y facturando indefinidamente.
    trigger_rule="all_done",
)

avisar_fallo = PythonOperator(
    task_id="avisar_fallo",
    python_callable=_notificar_a_slack,
    trigger_rule="one_failed",
)
[cargar_pedidos, lanzar_dataflow, refrescar_vista] >> avisar_fallo

# MAL: no reproducible
"query": "SELECT ... WHERE fecha = CURRENT_DATE()"

# BIEN: reproducible
"query": "SELECT ... WHERE fecha = DATE '{{ ds }}'"

gcloud composer environments run alpinashop-composer \
  --location=europe-west1 dags backfill -- \
  --start-date 2026-03-10 --end-date 2026-03-15 \
  --reset-dagruns \
  alpinashop_proceso_nocturno

gcloud composer environments run alpinashop-composer \
  --location=europe-west1 tasks clear -- \
  --task-regex "refrescar_vista_negocio" \
  --start-date 2026-03-14 --end-date 2026-03-14 --yes \
  alpinashop_proceso_nocturno

# proceso-nocturno.yaml -- Proceso nocturno de AlpinaShop con Workflows
main:
  params: [entrada]
  steps:
    - inicializar:
        assign:
          - proyecto: "alpinashop-datos"
          - proyecto_prod: "alpinashop-prod"
          - region: "europe-west1"
          - dataset: "alpinashop_analitica"
          - bucket: "alpinashop-datalake"
          # Si no se pasa fecha, se usa ayer. Permite reprocesar
          # invocando el flujo con {"fecha": "2026-03-12"}.
          - fecha: ${default(map.get(entrada, "fecha"), text.substring(time.format(sys.now() - 86400), 0, 10))}
          - fecha_compacta: ${text.replace_all(fecha, "-", "")}

    # -----------------------------------------------------------------
    # 1) Exportar Cloud SQL. La API devuelve una OPERACION de larga
    #    duracion: hay que esperarla, no darla por hecha.
    # -----------------------------------------------------------------
    - exportar_cloudsql:
        call: googleapis.sqladmin.v1.instances.export
        args:
          project: ${proyecto_prod}
          instance: "alpinashop-pedidos-replica-informes"
          body:
            exportContext:
              fileType: "CSV"
              uri: ${"gs://" + bucket + "/exportaciones/" + fecha_compacta + "/pedidos.csv"}
              databases: ["tienda"]
              csvExportOptions:
                selectQuery: ${"SELECT pedido_id, creado_en, cliente_id, canal, estado, pais, ciudad, codigo_postal, metodo_pago, subtotal, descuento, iva, total FROM pedidos WHERE DATE(creado_en) = '" + fecha + "'"}
        result: operacion_export

    - esperar_export:
        call: sys.sleep
        args:
          seconds: 30

    # -----------------------------------------------------------------
    # 2) Comprobar que el fichero existe. El equivalente al sensor.
    # -----------------------------------------------------------------
    - comprobar_fichero:
        try:
          call: googleapis.storage.v1.objects.get
          args:
            bucket: ${bucket}
            object: ${"exportaciones%2F" + fecha_compacta + "%2Fpedidos.csv"}
          result: info_fichero
        retry:
          predicate: ${http.default_retry_predicate}
          max_retries: 20
          backoff:
            initial_delay: 30
            max_delay: 120
            multiplier: 1.5

    # -----------------------------------------------------------------
    # 3) Cargar en BigQuery
    # -----------------------------------------------------------------
    - cargar_bigquery:
        call: googleapis.bigquery.v2.jobs.insert
        args:
          projectId: ${proyecto}
          body:
            configuration:
              load:
                sourceUris:
                  - ${"gs://" + bucket + "/exportaciones/" + fecha_compacta + "/pedidos.csv"}
                destinationTable:
                  projectId: ${proyecto}
                  datasetId: ${dataset}
                  tableId: "pedidos_staging"
                sourceFormat: "CSV"
                writeDisposition: "WRITE_TRUNCATE"
                autodetect: true
        result: job_carga

    - esperar_carga:
        call: espera_job_bigquery
        args:
          proyecto: ${proyecto}
          job_id: ${job_carga.jobReference.jobId}
        result: estado_carga

    # -----------------------------------------------------------------
    # 4) Lanzar Dataflow con la plantilla flexible
    # -----------------------------------------------------------------
    - lanzar_dataflow:
        call: http.post
        args:
          url: ${"https://dataflow.googleapis.com/v1b3/projects/" + proyecto + "/locations/" + region + "/flexTemplates:launch"}
          auth:
            type: OAuth2
          body:
            launchParameter:
              jobName: ${"pedidos-lote-" + fecha_compacta}
              containerSpecGcsPath: "gs://alpinashop-dataflow/plantillas/pedidos-lote.json"
              parameters:
                fecha: ${fecha}
              environment:
                serviceAccountEmail: "[email protected]"
                subnetwork: ${"regions/" + region + "/subnetworks/sn-datos-euw1"}
                ipConfiguration: "WORKER_IP_PRIVATE"
        result: job_dataflow

    # -----------------------------------------------------------------
    # 5) Agregaciones EN PARALELO con la rama 'parallel'
    # -----------------------------------------------------------------
    - agregaciones:
        parallel:
          branches:
            - ventas:
                steps:
                  - consulta_ventas:
                      call: ejecuta_sql
                      args:
                        proyecto: ${proyecto}
                        sql: ${"CREATE OR REPLACE TABLE `" + proyecto + "." + dataset + ".agg_ventas_categoria_dia` PARTITION BY dia AS SELECT l.fecha_pedido AS dia, pr.categoria, COUNT(DISTINCT l.pedido_id) AS pedidos, SUM(l.cantidad) AS unidades, ROUND(SUM(l.importe_linea),2) AS ventas_eur FROM `" + proyecto + "." + dataset + ".lineas_pedido` l JOIN `" + proyecto + "." + dataset + ".productos` pr USING (sku) WHERE l.fecha_pedido >= DATE_SUB(DATE '" + fecha + "', INTERVAL 400 DAY) GROUP BY dia, pr.categoria"}
            - embudo:
                steps:
                  - consulta_embudo:
                      call: ejecuta_sql
                      args:
                        proyecto: ${proyecto}
                        sql: ${"CREATE OR REPLACE TABLE `" + proyecto + "." + dataset + ".agg_embudo_dia` PARTITION BY dia AS SELECT fecha AS dia, COUNT(DISTINCT sesion_id) AS sesiones FROM `" + proyecto + "." + dataset + ".visitas` WHERE fecha = DATE '" + fecha + "' GROUP BY dia"}

    # -----------------------------------------------------------------
    # 6) Comprobacion de calidad: si falla, se lanza un error explicito
    # -----------------------------------------------------------------
    - comprobar_calidad:
        call: ejecuta_sql
        args:
          proyecto: ${proyecto}
          sql: ${"SELECT COUNT(*) AS n FROM `" + proyecto + "." + dataset + ".agg_ventas_categoria_dia` WHERE dia = DATE '" + fecha + "'"}
        result: resultado_calidad

    - evaluar_calidad:
        switch:
          - condition: ${int(resultado_calidad.rows[0].f[0].v) == 0}
            raise: ${"CALIDAD: no hay ventas agregadas para " + fecha}

    - devolver:
        return:
          fecha: ${fecha}
          estado: "OK"
          job_dataflow: ${job_dataflow.body.job.id}

# ---------------------------------------------------------------------
# SUBFLUJOS reutilizables
# ---------------------------------------------------------------------
ejecuta_sql:
  params: [proyecto, sql]
  steps:
    - lanzar:
        call: googleapis.bigquery.v2.jobs.query
        args:
          projectId: ${proyecto}
          body:
            query: ${sql}
            useLegacySql: false
            location: "europe-west1"
            timeoutMs: 300000
        result: r
    - devolver:
        return: ${r}

espera_job_bigquery:
  params: [proyecto, job_id]
  steps:
    - consultar:
        call: googleapis.bigquery.v2.jobs.get
        args:
          projectId: ${proyecto}
          jobId: ${job_id}
        result: estado
    - evaluar:
        switch:
          - condition: ${estado.status.state == "DONE"}
            next: comprobar_error
        next: esperar
    - esperar:
        call: sys.sleep
        args:
          seconds: 15
        next: consultar
    - comprobar_error:
        switch:
          - condition: ${"errorResult" in estado.status}
            raise: ${estado.status.errorResult.message}
    - devolver:
        return: ${estado}

gcloud iam service-accounts create sa-workflows-nocturno \
  --display-name="Orquestacion nocturna con Workflows"

SA_WF="[email protected]"

for ROL in roles/bigquery.dataEditor roles/bigquery.jobUser \
           roles/dataflow.developer roles/storage.objectAdmin \
           roles/cloudsql.editor roles/logging.logWriter; do
  gcloud projects add-iam-policy-binding alpinashop-datos \
    --member="serviceAccount:${SA_WF}" --role="$ROL"
done

gcloud workflows deploy alpinashop-proceso-nocturno \
  --source=proceso-nocturno.yaml \
  --location=europe-west1 \
  --service-account="$SA_WF" \
  --labels=entorno=produccion,equipo=datos,centro-coste=analitica

# Ejecucion manual con fecha explicita (reproceso)
gcloud workflows run alpinashop-proceso-nocturno \
  --location=europe-west1 \
  --data='{"fecha":"2026-03-12"}'

# Ver el resultado
gcloud workflows executions list alpinashop-proceso-nocturno \
  --location=europe-west1 --limit=5 \
  --format="table(name.basename(), state, startTime, endTime)"

gcloud iam service-accounts create sa-scheduler-nocturno \
  --display-name="Disparador del proceso nocturno"

SA_SCH="[email protected]"

gcloud projects add-iam-policy-binding alpinashop-datos \
  --member="serviceAccount:${SA_SCH}" --role="roles/workflows.invoker"

gcloud scheduler jobs create http disparar-proceso-nocturno \
  --location=europe-west1 \
  --schedule="0 2 * * *" \
  --time-zone="Europe/Madrid" \
  --uri="https://workflowexecutions.googleapis.com/v1/projects/alpinashop-datos/locations/europe-west1/workflows/alpinashop-proceso-nocturno/executions" \
  --http-method=POST \
  --oauth-service-account-email="$SA_SCH" \
  --message-body='{"argument":"{}"}' \
  --max-retry-attempts=3 \
  --min-backoff=60s \
  --max-backoff=600s \
  --attempt-deadline=60s

"""alpinashop_analisis_mensual.py -- Analisis de cesta y margen mensual."""
from datetime import datetime, timedelta

import pendulum
from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import (
    BigQueryCheckOperator, BigQueryInsertJobOperator,
)
from airflow.providers.google.cloud.operators.dataproc import (
    DataprocCreateBatchOperator,
)
from airflow.providers.google.cloud.operators.pubsub import (
    PubSubPublishMessageOperator,
)
from airflow.utils.task_group import TaskGroup

PROYECTO = "alpinashop-datos"
DATASET = "alpinashop_analitica"
BUCKET = "alpinashop-datalake"
REGION = "europe-west1"
TZ = pendulum.timezone("Europe/Madrid")

argumentos = {
    "owner": "equipo-datos",
    "email": ["[email protected]"],
    "email_on_failure": True,
    "retries": 2,
    "retry_delay": timedelta(minutes=10),
    "retry_exponential_backoff": True,
    "execution_timeout": timedelta(hours=3),
    "sla": timedelta(hours=4),
}

with DAG(
    dag_id="alpinashop_analisis_mensual",
    default_args=argumentos,
    schedule="0 4 1 * *",                     # dia 1 de cada mes a las 04:00
    start_date=datetime(2026, 3, 1, tzinfo=TZ),
    catchup=False,
    max_active_runs=1,
    tags=["alpinashop", "mensual", "datos"],
) as dag:

    # ds del dia 1 -> el mes anterior va del dia 1 anterior al ultimo dia
    PRIMER_DIA_MES_ANT = "{{ macros.ds_format(macros.ds_add(ds, -1), '%Y-%m-%d', '%Y-%m-01') }}"
    ULTIMO_DIA_MES_ANT = "{{ macros.ds_add(ds, -1) }}"

    # 1) Comprobar que hay datos del mes anterior
    comprobar_datos = BigQueryCheckOperator(
        task_id="comprobar_datos_mes_anterior",
        location=REGION,
        use_legacy_sql=False,
        sql=f"""
            SELECT COUNT(*) > 0
            FROM `{PROYECTO}.{DATASET}.lineas_pedido`
            WHERE fecha_pedido BETWEEN DATE '{PRIMER_DIA_MES_ANT}'
                                   AND DATE '{ULTIMO_DIA_MES_ANT}'
        """,
    )

    # 2) Analisis de cesta en Dataproc Serverless
    analisis_cesta = DataprocCreateBatchOperator(
        task_id="analisis_cesta",
        region=REGION,
        project_id=PROYECTO,
        batch_id="cesta-{{ ds_nodash }}",
        batch={
            "pyspark_batch": {
                "main_python_file_uri": f"gs://{BUCKET}/jobs/cesta_media.py",
                "args": [
                    f"--fecha-desde={PRIMER_DIA_MES_ANT}",
                    f"--fecha-hasta={ULTIMO_DIA_MES_ANT}",
                ],
            },
            "runtime_config": {"version": "2.2"},
            "environment_config": {
                "execution_config": {
                    "service_account":
                        "[email protected]",
                    "subnetwork_uri": "sn-datos-euw1",
                }
            },
        },
    )

    # 3) Dos consultas en paralelo
    with TaskGroup(group_id="informes_mensuales") as informes:

        margen_categoria = BigQueryInsertJobOperator(
            task_id="margen_por_categoria",
            location=REGION,
            configuration={"query": {"useLegacySql": False, "query": f"""
                CREATE OR REPLACE TABLE `{PROYECTO}.{DATASET}.agg_margen_mes` AS
                SELECT
                  DATE '{PRIMER_DIA_MES_ANT}'                          AS mes,
                  pr.categoria,
                  SUM(l.cantidad)                                      AS unidades,
                  ROUND(SUM(l.importe_linea), 2)                       AS ventas_eur,
                  ROUND(SUM(l.cantidad * c.coste_medio_eur), 2)        AS coste_eur,
                  ROUND(SUM(l.importe_linea)
                        - SUM(l.cantidad * c.coste_medio_eur), 2)      AS margen_eur
                FROM `{PROYECTO}.{DATASET}.lineas_pedido`               AS l
                JOIN `{PROYECTO}.{DATASET}.productos`                   AS pr USING (sku)
                JOIN `{PROYECTO}.{DATASET}.costes_producto_erp`         AS c  USING (sku)
                WHERE l.fecha_pedido BETWEEN DATE '{PRIMER_DIA_MES_ANT}'
                                         AND DATE '{ULTIMO_DIA_MES_ANT}'
                GROUP BY mes, pr.categoria
            """}},
        )

        sin_ventas = BigQueryInsertJobOperator(
            task_id="productos_sin_ventas",
            location=REGION,
            configuration={"query": {"useLegacySql": False, "query": f"""
                CREATE OR REPLACE TABLE `{PROYECTO}.{DATASET}.agg_sin_ventas_mes` AS
                SELECT
                  DATE '{PRIMER_DIA_MES_ANT}' AS mes,
                  pr.sku, pr.nombre, pr.categoria, pr.precio_catalogo
                FROM `{PROYECTO}.{DATASET}.productos` AS pr
                WHERE pr.activo = TRUE
                  AND NOT EXISTS (
                    SELECT 1 FROM `{PROYECTO}.{DATASET}.lineas_pedido` AS l
                    WHERE l.sku = pr.sku
                      AND l.fecha_pedido BETWEEN DATE '{PRIMER_DIA_MES_ANT}'
                                             AND DATE '{ULTIMO_DIA_MES_ANT}'
                  )
            """}},
        )

    # 4) Refrescar la vista materializada
    refrescar = BigQueryInsertJobOperator(
        task_id="refrescar_vista",
        location=REGION,
        configuration={"query": {"useLegacySql": False, "query": f"""
            CALL BQ.REFRESH_MATERIALIZED_VIEW(
              '{PROYECTO}.{DATASET}.mv_ventas_diarias_sku')
        """}},
    )

    # 5) Avisar de que el informe esta listo
    avisar = PubSubPublishMessageOperator(
        task_id="avisar_informe_listo",
        project_id=PROYECTO,
        topic="informes-listos",
        messages=[{
            "data": b'{"informe":"mensual","estado":"completado"}',
            "attributes": {
                "tipo_evento": "informe_listo",
                "periodo": PRIMER_DIA_MES_ANT,
            },
        }],
    )

    comprobar_datos >> analisis_cesta >> informes >> refrescar >> avisar

# informe-mensual.yaml
main:
  params: [entrada]
  steps:
    - inicializar:
        assign:
          - proyecto: "alpinashop-datos"
          - dataset: "alpinashop_analitica"
          - region: "europe-west1"
          - bucket: "alpinashop-datalake"
          - mes: ${default(map.get(entrada, "mes"), text.substring(time.format(sys.now() - 2592000), 0, 7))}
          - primer_dia: ${mes + "-01"}
          - mes_compacto: ${text.replace_all(mes, "-", "")}

    - calcular_ultimo_dia:
        call: ejecuta_sql
        args:
          proyecto: ${proyecto}
          sql: ${"SELECT CAST(LAST_DAY(DATE '" + primer_dia + "') AS STRING) AS ultimo"}
        result: r_ultimo

    - asignar_ultimo:
        assign:
          - ultimo_dia: ${r_ultimo.rows[0].f[0].v}

    # 1) Comprobar que hay datos
    - comprobar_datos:
        call: ejecuta_sql
        args:
          proyecto: ${proyecto}
          sql: ${"SELECT COUNT(*) AS n FROM `" + proyecto + "." + dataset + ".lineas_pedido` WHERE fecha_pedido BETWEEN DATE '" + primer_dia + "' AND DATE '" + ultimo_dia + "'"}
        result: r_datos

    - evaluar_datos:
        switch:
          - condition: ${int(r_datos.rows[0].f[0].v) == 0}
            raise: ${"Sin lineas de pedido para el mes " + mes}

    # 2) Dataproc Serverless: operacion de larga duracion
    - lanzar_cesta:
        call: http.post
        args:
          url: ${"https://dataproc.googleapis.com/v1/projects/" + proyecto + "/locations/" + region + "/batches?batchId=cesta-" + mes_compacto}
          auth:
            type: OAuth2
          body:
            pysparkBatch:
              mainPythonFileUri: ${"gs://" + bucket + "/jobs/cesta_media.py"}
              args:
                - ${"--fecha-desde=" + primer_dia}
                - ${"--fecha-hasta=" + ultimo_dia}
            runtimeConfig:
              version: "2.2"
            environmentConfig:
              executionConfig:
                serviceAccount: "[email protected]"
                subnetworkUri: "sn-datos-euw1"
        result: r_batch

    - esperar_cesta:
        call: espera_batch
        args:
          proyecto: ${proyecto}
          region: ${region}
          batch_id: ${"cesta-" + mes_compacto}

    # 3) Dos consultas en paralelo
    - informes:
        parallel:
          branches:
            - margen:
                steps:
                  - q_margen:
                      call: ejecuta_sql
                      args:
                        proyecto: ${proyecto}
                        sql: ${"CREATE OR REPLACE TABLE `" + proyecto + "." + dataset + ".agg_margen_mes` AS SELECT DATE '" + primer_dia + "' AS mes, pr.categoria, SUM(l.cantidad) AS unidades, ROUND(SUM(l.importe_linea),2) AS ventas_eur, ROUND(SUM(l.cantidad*c.coste_medio_eur),2) AS coste_eur FROM `" + proyecto + "." + dataset + ".lineas_pedido` l JOIN `" + proyecto + "." + dataset + ".productos` pr USING (sku) JOIN `" + proyecto + "." + dataset + ".costes_producto_erp` c USING (sku) WHERE l.fecha_pedido BETWEEN DATE '" + primer_dia + "' AND DATE '" + ultimo_dia + "' GROUP BY mes, pr.categoria"}
            - sin_ventas:
                steps:
                  - q_sin_ventas:
                      call: ejecuta_sql
                      args:
                        proyecto: ${proyecto}
                        sql: ${"CREATE OR REPLACE TABLE `" + proyecto + "." + dataset + ".agg_sin_ventas_mes` AS SELECT DATE '" + primer_dia + "' AS mes, pr.sku, pr.nombre, pr.categoria FROM `" + proyecto + "." + dataset + ".productos` pr WHERE pr.activo = TRUE AND NOT EXISTS (SELECT 1 FROM `" + proyecto + "." + dataset + ".lineas_pedido` l WHERE l.sku = pr.sku AND l.fecha_pedido BETWEEN DATE '" + primer_dia + "' AND DATE '" + ultimo_dia + "')"}

    - devolver:
        return:
          mes: ${mes}
          estado: "OK"

ejecuta_sql:
  params: [proyecto, sql]
  steps:
    - lanzar:
        call: googleapis.bigquery.v2.jobs.query
        args:
          projectId: ${proyecto}
          body:
            query: ${sql}
            useLegacySql: false
            location: "europe-west1"
            timeoutMs: 300000
        result: r
    - devolver:
        return: ${r}

# Sondeo de la operacion de larga duracion: Workflows NO puede
# esperar 20 minutos en una sola llamada HTTP (limite de 30 min,
# y ademas el batch podria tardar mas).
espera_batch:
  params: [proyecto, region, batch_id]
  steps:
    - consultar:
        call: http.get
        args:
          url: ${"https://dataproc.googleapis.com/v1/projects/" + proyecto + "/locations/" + region + "/batches/" + batch_id}
          auth:
            type: OAuth2
        result: estado
    - evaluar:
        switch:
          - condition: ${estado.body.state == "SUCCEEDED"}
            return: ${estado.body}
          - condition: ${estado.body.state == "FAILED"}
            raise: ${"Batch de Dataproc fallido: " + default(map.get(estado.body, "stateMessage"), "sin detalle")}
          - condition: ${estado.body.state == "CANCELLED"}
            raise: "Batch de Dataproc cancelado"
    - esperar:
        call: sys.sleep
        args:
          seconds: 30
        next: consultar

gcloud workflows deploy alpinashop-informe-mensual \
  --source=informe-mensual.yaml --location=europe-west1 \
  --service-account="[email protected]"

gcloud scheduler jobs create http disparar-informe-mensual \
  --location=europe-west1 \
  --schedule="0 4 1 * *" \
  --time-zone="Europe/Madrid" \
  --uri="https://workflowexecutions.googleapis.com/v1/projects/alpinashop-datos/locations/europe-west1/workflows/alpinashop-informe-mensual/executions" \
  --http-method=POST \
  --oauth-service-account-email="[email protected]" \
  --message-body='{"argument":"{}"}' \
  --max-retry-attempts=3

MERGE `alpinashop-datos.alpinashop_analitica.pedidos` AS d
USING (SELECT * FROM `alpinashop-datos.alpinashop_analitica.pedidos_staging`) AS o
ON d.pedido_id = o.pedido_id AND d.fecha_pedido = o.fecha_pedido
WHEN MATCHED THEN UPDATE SET estado = o.estado, total_pedido = o.total_pedido
WHEN NOT MATCHED THEN INSERT ROW

agregar_ventas = BigQueryInsertJobOperator(
    task_id="ventas_por_categoria",
    location=REGION,
    configuration={
        "query": {
            "useLegacySql": False,
            # Escribe SOLO la particion de la fecha logica
            "destinationTable": {
                "projectId": PROYECTO,
                "datasetId": DATASET,
                "tableId": "agg_ventas_categoria_dia${{ ds_nodash }}",
            },
            "writeDisposition": "WRITE_TRUNCATE",
            "timePartitioning": {"type": "DAY", "field": "dia"},
            "query": f"""
                SELECT
                  l.fecha_pedido                 AS dia,
                  pr.categoria,
                  COUNT(DISTINCT l.pedido_id)    AS pedidos,
                  SUM(l.cantidad)                AS unidades,
                  ROUND(SUM(l.importe_linea), 2) AS ventas_eur
                FROM `{PROYECTO}.{DATASET}.lineas_pedido` AS l
                JOIN `{PROYECTO}.{DATASET}.productos`     AS pr USING (sku)
                WHERE l.fecha_pedido = DATE '{{{{ ds }}}}'
                GROUP BY dia, pr.categoria
            """,
        }
    },
)

comprobar_calidad = BigQueryCheckOperator(
    task_id="comprobar_calidad_datos",
    location=REGION,
    use_legacy_sql=False,
    sql=f"""
        WITH dia AS (
          SELECT
            SUM(pedidos)                      AS pedidos_dia,
            SUM(ventas_eur)                   AS ventas_dia,
            COUNT(*)                          AS filas
          FROM `{PROYECTO}.{DATASET}.agg_ventas_categoria_dia`
          WHERE dia = DATE '{{{{ ds }}}}'
        ),
        control_duplicados AS (
          SELECT COUNT(*) AS duplicados FROM (
            SELECT pedido_id
            FROM `{PROYECTO}.{DATASET}.pedidos`
            WHERE fecha_pedido = DATE '{{{{ ds }}}}'
            GROUP BY pedido_id
            HAVING COUNT(*) > 1
          )
        )
        SELECT
          d.filas       > 0    AND
          d.pedidos_dia > 0    AND
          d.pedidos_dia < 500  AND
          d.ventas_dia  > 0    AND
          c.duplicados  = 0
        FROM dia AS d CROSS JOIN control_duplicados AS c
    """,
)

-- 1) Deduplicar la tabla de pedidos conservando la ultima version
CREATE OR REPLACE TABLE `alpinashop-datos.alpinashop_analitica.pedidos`
PARTITION BY fecha_pedido
CLUSTER BY estado, canal AS
SELECT * EXCEPT(rn) FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY pedido_id, fecha_pedido
                       ORDER BY momento_pedido DESC) AS rn
  FROM `alpinashop-datos.alpinashop_analitica.pedidos`
)
WHERE rn = 1;

# 2) Reprocesar las agregaciones del periodo afectado, ya con el DAG corregido
gcloud composer environments run alpinashop-composer \
  --location=europe-west1 dags backfill -- \
  --start-date 2026-02-10 --end-date 2026-03-16 \
  --reset-dagruns alpinashop_proceso_nocturno