# /etc/crontab on the vm-procesos-nocturnos VM
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="AlpinaShop Cloud Composer"

SA_COMP="[email protected]"

for ROLE 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="$ROLE"
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

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

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

"""
proceso_nocturno.py -- AlpinaShop's nightly data process.

Flow:
  export Cloud SQL -> wait for file -> load BigQuery -> Dataflow
  -> aggregations (parallel) -> refresh view -> check quality

Deployment: copy to 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

DATA_PROJECT = "alpinashop-datos"
PROD_PROJECT = "alpinashop-prod"
DATASET = "alpinashop_analitica"
BUCKET = "alpinashop-datalake"
REGION = "europe-west1"
TIME_ZONE = pendulum.timezone("Europe/Madrid")

# ---------------------------------------------------------------------------
# DEFAULT ARGUMENTS: they apply to EVERY task in the DAG.
# Defining them here avoids repeating them in each operator.
# ---------------------------------------------------------------------------
default_arguments = {
    "owner": "equipo-datos",
    "depends_on_past": False,      # one run does not wait for the previous one
    "email": ["[email protected]"],
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 3,                  # 3 retries on failure
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,   # 5, 10, 20 min: do not hammer the source
    "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(hours=2),   # no task runs forever
    "sla": timedelta(hours=3),     # warn if the DAG does not finish within 3 h
}

with DAG(
    dag_id="alpinashop_proceso_nocturno",
    description="Nightly process: Cloud SQL -> BigQuery -> Dataflow -> aggregates",
    default_args=default_arguments,
    # Every day at 02:00 Madrid time.
    # Airflow works internally in UTC; the time zone avoids the one-hour
    # drift at the daylight saving changeovers.
    schedule="0 2 * * *",
    start_date=datetime(2026, 3, 1, tzinfo=TIME_ZONE),
    catchup=False,             # see section 10
    max_active_runs=1,         # never two overlapping nights
    tags=["alpinashop", "produccion", "datos"],
    doc_md=__doc__,            # the docstring is shown in the interface
) as dag:

    start = EmptyOperator(task_id="inicio")

    # -----------------------------------------------------------------------
    # 1) EXPORT Cloud SQL to Cloud Storage
    #
    # {{ ds }} is a Jinja template: Airflow replaces it with the run's
    # logical date (yyyy-MM-dd). It is THE piece that makes the DAG
    # reproducible: when reprocessing 12 March, {{ ds }} is 2026-03-12
    # and the output file and the query point at that day, not at today.
    # -----------------------------------------------------------------------
    export_orders = CloudSQLExportInstanceOperator(
        task_id="exportar_pedidos_cloudsql",
        project_id=PROD_PROJECT,
        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: wait for the file to genuinely exist.
    #
    # This task is the one that solves the cron problem. The export is
    # asynchronous: the previous operator launches it, but the file may
    # take a while. Instead of "let us wait 30 minutes and cross our
    # fingers", it is CHECKED.
    # -----------------------------------------------------------------------
    wait_for_file = GCSObjectExistenceSensor(
        task_id="esperar_fichero_exportado",
        bucket=BUCKET,
        object="exportaciones/{{ ds_nodash }}/pedidos.csv",
        # 'reschedule': frees the worker between checks instead of
        # tying it up waiting. Essential for long waits.
        mode="reschedule",
        poke_interval=60,          # check every minute
        timeout=60 * 60,           # give up after an hour
    )

    # -----------------------------------------------------------------------
    # 3) LOAD into BigQuery
    # -----------------------------------------------------------------------
    load_orders = GCSToBigQueryOperator(
        task_id="cargar_pedidos_bigquery",
        bucket=BUCKET,
        source_objects=["exportaciones/{{ ds_nodash }}/pedidos.csv"],
        destination_project_dataset_table=f"{DATA_PROJECT}.{DATASET}.pedidos_staging",
        source_format="CSV",
        skip_leading_rows=0,
        field_delimiter=",",
        null_marker="\\N",
        # WRITE_TRUNCATE on the staging table: it is replaced every night.
        # This makes the task IDEMPOTENT: re-running it duplicates nothing.
        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 from staging into the final table: idempotent by design
    # -----------------------------------------------------------------------
    consolidate_orders = BigQueryInsertJobOperator(
        task_id="consolidar_pedidos",
        location=REGION,
        configuration={
            "query": {
                "query": f"""
                    MERGE `{DATA_PROJECT}.{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 `{DATA_PROJECT}.{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: flex template created in 04-02
    # -----------------------------------------------------------------------
    launch_dataflow = DataflowStartFlexTemplateOperator(
        task_id="lanzar_pipeline_dataflow",
        location=REGION,
        project_id=DATA_PROJECT,
        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",
                    },
                },
            }
        },
        # The operator WAITS for the job to finish before marking the task
        # as complete. Without this, the DAG would carry on with half-done data.
        wait_until_finished=True,
    )

    # -----------------------------------------------------------------------
    # 6) AGGREGATIONS IN PARALLEL, grouped so the graph stays readable
    # -----------------------------------------------------------------------
    with TaskGroup(group_id="agregaciones") as aggregation_group:

        aggregate_sales = BigQueryInsertJobOperator(
            task_id="ventas_por_categoria",
            location=REGION,
            configuration={
                "query": {
                    "query": f"""
                        CREATE OR REPLACE TABLE
                          `{DATA_PROJECT}.{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 `{DATA_PROJECT}.{DATASET}.lineas_pedido` AS l
                        JOIN `{DATA_PROJECT}.{DATASET}.productos`     AS pr
                          USING (sku)
                        WHERE l.fecha_pedido >= DATE_SUB(DATE '{{{{ ds }}}}',
                                                         INTERVAL 400 DAY)
                        GROUP BY dia, pr.categoria
                    """,
                    "useLegacySql": False,
                }
            },
        )

        aggregate_visits = BigQueryInsertJobOperator(
            task_id="embudo_conversion",
            location=REGION,
            configuration={
                "query": {
                    "query": f"""
                        CREATE OR REPLACE TABLE
                          `{DATA_PROJECT}.{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 `{DATA_PROJECT}.{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) REFRESH the materialised view that feeds the dashboard
    # -----------------------------------------------------------------------
    refresh_view = BigQueryInsertJobOperator(
        task_id="refrescar_vista_negocio",
        location=REGION,
        configuration={
            "query": {
                "query": f"""
                    CALL BQ.REFRESH_MATERIALIZED_VIEW(
                      '{DATA_PROJECT}.{DATASET}.mv_ventas_diarias_sku')
                """,
                "useLegacySql": False,
            }
        },
    )

    # -----------------------------------------------------------------------
    # 8) QUALITY CHECK: the last line of defence before the report.
    #
    # If this task fails, the DAG is marked as failed and the alert
    # fires. An alert at 05:00 is preferable to a management committee
    # staring at false figures at 09:00.
    # -----------------------------------------------------------------------
    check_quality = 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 `{DATA_PROJECT}.{DATASET}.agg_ventas_categoria_dia`
              WHERE dia = DATE '{{{{ ds }}}}'
              GROUP BY dia
            )
        """,
    )

    def _notify_success(**context):
        """Informational callback: in production it would post to Slack or Chat."""
        print(f"Nightly process completed for {context['ds']}")

    end = PythonOperator(
        task_id="fin",
        python_callable=_notify_success,
        trigger_rule="all_success",
    )

    # -----------------------------------------------------------------------
    # DEPENDENCIES: the >> operator means "and then".
    # This is the complete declaration of the graph, in four lines.
    # -----------------------------------------------------------------------
    (
        start
        >> export_orders
        >> wait_for_file
        >> load_orders
        >> consolidate_orders
        >> launch_dataflow
        >> aggregation_group
        >> refresh_view
        >> check_quality
        >> end
    )

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

basket_analysis = DataprocCreateBatchOperator(
    task_id="analisis_cesta_mensual",
    region=REGION,
    project_id=DATA_PROJECT,
    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 _count_rows(**context):
    from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook
    hook = BigQueryHook(location=REGION, use_legacy_sql=False)
    rows = hook.get_first(
        f"""SELECT COUNT(*) FROM `{DATA_PROJECT}.{DATASET}.pedidos_staging`"""
    )[0]
    # Whatever the function returns is stored automatically as an XCom
    return int(rows)


def _validate_volume(**context):
    rows = context["ti"].xcom_pull(task_ids="contar_filas_cargadas")
    if rows == 0:
        raise ValueError(f"No orders loaded for {context['ds']}")
    if rows > 5000:
        raise ValueError(f"{rows} orders: anomalous volume, check for duplicates")
    print(f"Volume looks right: {rows} orders")


count_rows = PythonOperator(task_id="contar_filas_cargadas",
                            python_callable=_count_rows)
validate_volume = PythonOperator(task_id="validar_volumen",
                                 python_callable=_validate_volume)

from airflow.models import Variable

threshold = 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 after a
a << b                    # a after b
a >> [b, c] >> d          # b and c in parallel, both after a; d after both
[a, b] >> c               # c waits for a and b to finish

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

delete_cluster = DataprocDeleteClusterOperator(
    task_id="borrar_cluster",
    cluster_name="cluster-efimero-{{ ds_nodash }}",
    region=REGION,
    project_id=DATA_PROJECT,
    # CRITICAL: the cluster is deleted WHATEVER HAPPENS.
    # With all_success, a job failure would leave the cluster switched on
    # and billing indefinitely.
    trigger_rule="all_done",
)

notify_failure = PythonOperator(
    task_id="avisar_fallo",
    python_callable=_notify_slack,
    trigger_rule="one_failed",
)
[load_orders, launch_dataflow, refresh_view] >> notify_failure

# WRONG: not reproducible
"query": "SELECT ... WHERE fecha = CURRENT_DATE()"

# RIGHT: 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 -- AlpinaShop's nightly process with Workflows
main:
  params: [entrada]
  steps:
    - inicializar:
        assign:
          - proyecto: "alpinashop-datos"
          - proyecto_prod: "alpinashop-prod"
          - region: "europe-west1"
          - dataset: "alpinashop_analitica"
          - bucket: "alpinashop-datalake"
          # If no date is passed, yesterday is used. This allows reprocessing
          # by invoking the flow with {"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) Export Cloud SQL. The API returns a LONG-RUNNING operation:
    #    you have to wait for it, not assume it is done.
    # -----------------------------------------------------------------
    - 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) Check that the file exists. The equivalent of the 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) Load into 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) Launch Dataflow with the flex template
    # -----------------------------------------------------------------
    - 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) Aggregations IN PARALLEL with the 'parallel' branch
    # -----------------------------------------------------------------
    - 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) Quality check: if it fails, an explicit error is raised
    # -----------------------------------------------------------------
    - 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: ${"QUALITY: no aggregated sales for " + fecha}

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

# ---------------------------------------------------------------------
# Reusable SUBWORKFLOWS
# ---------------------------------------------------------------------
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="Nightly orchestration with Workflows"

SA_WF="[email protected]"

for ROLE 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="$ROLE"
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

# Manual run with an explicit date (reprocessing)
gcloud workflows run alpinashop-proceso-nocturno \
  --location=europe-west1 \
  --data='{"fecha":"2026-03-12"}'

# See the result
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="Nightly process trigger"

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 -- Monthly basket and margin analysis."""
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

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

arguments = {
    "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=arguments,
    schedule="0 4 1 * *",                     # 1st of each month at 04:00
    start_date=datetime(2026, 3, 1, tzinfo=TZ),
    catchup=False,
    max_active_runs=1,
    tags=["alpinashop", "mensual", "datos"],
) as dag:

    # ds of the 1st -> the previous month runs from its 1st to its last day
    FIRST_DAY_PREV_MONTH = "{{ macros.ds_format(macros.ds_add(ds, -1), '%Y-%m-%d', '%Y-%m-01') }}"
    LAST_DAY_PREV_MONTH = "{{ macros.ds_add(ds, -1) }}"

    # 1) Check that there is data from the previous month
    check_data = BigQueryCheckOperator(
        task_id="comprobar_datos_mes_anterior",
        location=REGION,
        use_legacy_sql=False,
        sql=f"""
            SELECT COUNT(*) > 0
            FROM `{PROJECT}.{DATASET}.lineas_pedido`
            WHERE fecha_pedido BETWEEN DATE '{FIRST_DAY_PREV_MONTH}'
                                   AND DATE '{LAST_DAY_PREV_MONTH}'
        """,
    )

    # 2) Basket analysis on Dataproc Serverless
    basket_analysis = DataprocCreateBatchOperator(
        task_id="analisis_cesta",
        region=REGION,
        project_id=PROJECT,
        batch_id="cesta-{{ ds_nodash }}",
        batch={
            "pyspark_batch": {
                "main_python_file_uri": f"gs://{BUCKET}/jobs/cesta_media.py",
                "args": [
                    f"--fecha-desde={FIRST_DAY_PREV_MONTH}",
                    f"--fecha-hasta={LAST_DAY_PREV_MONTH}",
                ],
            },
            "runtime_config": {"version": "2.2"},
            "environment_config": {
                "execution_config": {
                    "service_account":
                        "[email protected]",
                    "subnetwork_uri": "sn-datos-euw1",
                }
            },
        },
    )

    # 3) Two queries in parallel
    with TaskGroup(group_id="informes_mensuales") as reports:

        margin_by_category = BigQueryInsertJobOperator(
            task_id="margen_por_categoria",
            location=REGION,
            configuration={"query": {"useLegacySql": False, "query": f"""
                CREATE OR REPLACE TABLE `{PROJECT}.{DATASET}.agg_margen_mes` AS
                SELECT
                  DATE '{FIRST_DAY_PREV_MONTH}'                        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 `{PROJECT}.{DATASET}.lineas_pedido`                 AS l
                JOIN `{PROJECT}.{DATASET}.productos`                     AS pr USING (sku)
                JOIN `{PROJECT}.{DATASET}.costes_producto_erp`           AS c  USING (sku)
                WHERE l.fecha_pedido BETWEEN DATE '{FIRST_DAY_PREV_MONTH}'
                                         AND DATE '{LAST_DAY_PREV_MONTH}'
                GROUP BY mes, pr.categoria
            """}},
        )

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

    # 4) Refresh the materialised view
    refresh = BigQueryInsertJobOperator(
        task_id="refrescar_vista",
        location=REGION,
        configuration={"query": {"useLegacySql": False, "query": f"""
            CALL BQ.REFRESH_MATERIALIZED_VIEW(
              '{PROJECT}.{DATASET}.mv_ventas_diarias_sku')
        """}},
    )

    # 5) Announce that the report is ready
    notify = PubSubPublishMessageOperator(
        task_id="avisar_informe_listo",
        project_id=PROJECT,
        topic="informes-listos",
        messages=[{
            "data": b'{"informe":"mensual","estado":"completado"}',
            "attributes": {
                "tipo_evento": "informe_listo",
                "periodo": FIRST_DAY_PREV_MONTH,
            },
        }],
    )

    check_data >> basket_analysis >> reports >> refresh >> notify

# 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) Check that there is data
    - 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: ${"No order lines for month " + mes}

    # 2) Dataproc Serverless: long-running operation
    - 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) Two queries in parallel
    - 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}

# Polling the long-running operation: Workflows CANNOT wait
# 20 minutes in a single HTTP call (30 min limit, and besides
# the batch could take longer).
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: ${"Dataproc batch failed: " + default(map.get(estado.body, "stateMessage"), "no detail")}
          - condition: ${estado.body.state == "CANCELLED"}
            raise: "Dataproc batch cancelled"
    - 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

aggregate_sales = BigQueryInsertJobOperator(
    task_id="ventas_por_categoria",
    location=REGION,
    configuration={
        "query": {
            "useLegacySql": False,
            # Writes ONLY the partition for the logical date
            "destinationTable": {
                "projectId": PROJECT,
                "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 `{PROJECT}.{DATASET}.lineas_pedido` AS l
                JOIN `{PROJECT}.{DATASET}.productos`     AS pr USING (sku)
                WHERE l.fecha_pedido = DATE '{{{{ ds }}}}'
                GROUP BY dia, pr.categoria
            """,
        }
    },
)

check_quality = 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 `{PROJECT}.{DATASET}.agg_ventas_categoria_dia`
          WHERE dia = DATE '{{{{ ds }}}}'
        ),
        control_duplicados AS (
          SELECT COUNT(*) AS duplicados FROM (
            SELECT pedido_id
            FROM `{PROJECT}.{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) Deduplicate the orders table keeping the latest 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) Reprocess the aggregations for the affected period, now with the fixed DAG
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