# /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 d 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 -- Proces nocturn de dades d AlpinaShop.

Flux:
  exportar Cloud SQL -> esperar fitxer -> carregar BigQuery -> Dataflow
  -> agregacions (paral·leles) -> refrescar vista -> comprovar qualitat

Desplegament: 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

PROJECTE_DADES = "alpinashop-datos"
PROJECTE_PROD = "alpinashop-prod"
DATASET = "alpinashop_analitica"
BUCKET = "alpinashop-datalake"
REGIO = "europe-west1"
ZONA_HORARIA = pendulum.timezone("Europe/Madrid")

# ---------------------------------------------------------------------------
# ARGUMENTS PER DEFECTE: s apliquen a TOTES les tasques del DAG.
# Definir-los aqui evita repetir-los a cada operador.
# ---------------------------------------------------------------------------
arguments_per_defecte = {
    "owner": "equipo-datos",
    "depends_on_past": False,      # una execucio no espera l anterior
    "email": ["[email protected]"],
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 3,                  # 3 reintents davant d una fallada
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,   # 5, 10, 20 min: no matxucar l origen
    "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(hours=2),   # cap tasca eterna
    "sla": timedelta(hours=3),     # avis si el DAG no acaba en 3 h
}

with DAG(
    dag_id="alpinashop_proceso_nocturno",
    description="Proces nocturn: Cloud SQL -> BigQuery -> Dataflow -> agregats",
    default_args=arguments_per_defecte,
    # Tots els dies a les 02:00 hora de Madrid.
    # Airflow treballa internament en UTC; la zona horaria evita el desfasament
    # d una hora als canvis d horari d estiu.
    schedule="0 2 * * *",
    start_date=datetime(2026, 3, 1, tzinfo=ZONA_HORARIA),
    catchup=False,             # veure apartat 10
    max_active_runs=1,         # mai dues nits encavalcades
    tags=["alpinashop", "produccion", "datos"],
    doc_md=__doc__,            # la docstring es veu a la interficie
) as dag:

    inici = EmptyOperator(task_id="inicio")

    # -----------------------------------------------------------------------
    # 1) EXPORTAR Cloud SQL a Cloud Storage
    #
    # {{ ds }} es una plantilla Jinja: Airflow la substitueix per la data
    # logica de l execucio (yyyy-MM-dd). Es LA peca que fa el DAG
    # reproduible: en reprocessar el 12 de marc, {{ ds }} val 2026-03-12
    # i el fitxer de sortida i la consulta apunten a aquell dia, no a avui.
    # -----------------------------------------------------------------------
    exportar_comandes = CloudSQLExportInstanceOperator(
        task_id="exportar_pedidos_cloudsql",
        project_id=PROJECTE_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 que el fitxer existeixi de veritat.
    #
    # Aquesta tasca es la que resol el problema del cron. L exportacio
    # es asincrona: l operador anterior la llanca, pero el fitxer pot
    # trigar. En comptes de "esperem 30 minuts i creuem els dits",
    # es COMPROVA.
    # -----------------------------------------------------------------------
    esperar_fitxer = GCSObjectExistenceSensor(
        task_id="esperar_fichero_exportado",
        bucket=BUCKET,
        object="exportaciones/{{ ds_nodash }}/pedidos.csv",
        # 'reschedule': allibera el treballador entre comprovacions en comptes
        # d ocupar-lo esperant. Imprescindible en esperes llargues.
        mode="reschedule",
        poke_interval=60,          # comprovar cada minut
        timeout=60 * 60,           # rendir-se a l hora
    )

    # -----------------------------------------------------------------------
    # 3) CARREGAR a BigQuery
    # -----------------------------------------------------------------------
    carregar_comandes = GCSToBigQueryOperator(
        task_id="cargar_pedidos_bigquery",
        bucket=BUCKET,
        source_objects=["exportaciones/{{ ds_nodash }}/pedidos.csv"],
        destination_project_dataset_table=f"{PROJECTE_DADES}.{DATASET}.pedidos_staging",
        source_format="CSV",
        skip_leading_rows=0,
        field_delimiter=",",
        null_marker="\\N",
        # WRITE_TRUNCATE a la taula de staging: cada nit es reemplaca.
        # Aixo fa la tasca IDEMPOTENT: reexecutar-la no duplica res.
        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=REGIO,
    )

    # -----------------------------------------------------------------------
    # 4) MERGE del staging a la taula final: idempotent per disseny
    # -----------------------------------------------------------------------
    consolidar_comandes = BigQueryInsertJobOperator(
        task_id="consolidar_pedidos",
        location=REGIO,
        configuration={
            "query": {
                "query": f"""
                    MERGE `{PROJECTE_DADES}.{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 `{PROJECTE_DADES}.{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 a 04-02
    # -----------------------------------------------------------------------
    llancar_dataflow = DataflowStartFlexTemplateOperator(
        task_id="lanzar_pipeline_dataflow",
        location=REGIO,
        project_id=PROJECTE_DADES,
        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/{REGIO}/subnetworks/sn-datos-euw1",
                    "ipConfiguration": "WORKER_IP_PRIVATE",
                    "maxWorkers": 10,
                    "additionalUserLabels": {
                        "entorno": "produccion", "equipo": "datos",
                    },
                },
            }
        },
        # L operador ESPERA que el job acabi abans de donar la tasca
        # per completada. Sense aixo, el DAG continuaria amb dades a mitges.
        wait_until_finished=True,
    )

    # -----------------------------------------------------------------------
    # 6) AGREGACIONS EN PARAL·LEL, agrupades perque el graf sigui llegible
    # -----------------------------------------------------------------------
    with TaskGroup(group_id="agregaciones") as grup_agregacions:

        agregar_vendes = BigQueryInsertJobOperator(
            task_id="ventas_por_categoria",
            location=REGIO,
            configuration={
                "query": {
                    "query": f"""
                        CREATE OR REPLACE TABLE
                          `{PROJECTE_DADES}.{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 `{PROJECTE_DADES}.{DATASET}.lineas_pedido` AS l
                        JOIN `{PROJECTE_DADES}.{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_visites = BigQueryInsertJobOperator(
            task_id="embudo_conversion",
            location=REGIO,
            configuration={
                "query": {
                    "query": f"""
                        CREATE OR REPLACE TABLE
                          `{PROJECTE_DADES}.{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 `{PROJECTE_DADES}.{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 materialitzada que consumeix el tauler de control
    # -----------------------------------------------------------------------
    refrescar_vista = BigQueryInsertJobOperator(
        task_id="refrescar_vista_negocio",
        location=REGIO,
        configuration={
            "query": {
                "query": f"""
                    CALL BQ.REFRESH_MATERIALIZED_VIEW(
                      '{PROJECTE_DADES}.{DATASET}.mv_ventas_diarias_sku')
                """,
                "useLegacySql": False,
            }
        },
    )

    # -----------------------------------------------------------------------
    # 8) COMPROVACIO DE QUALITAT: l ultima defensa abans de l informe.
    #
    # Si aquesta tasca falla, el DAG queda marcat com a fallit i salta
    # l alerta. Es preferible una alerta a les 05:00 que un comite de
    # direccio mirant xifres falses a les 09:00.
    # -----------------------------------------------------------------------
    comprovar_qualitat = BigQueryCheckOperator(
        task_id="comprobar_calidad_datos",
        location=REGIO,
        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 `{PROJECTE_DADES}.{DATASET}.agg_ventas_categoria_dia`
              WHERE dia = DATE '{{{{ ds }}}}'
              GROUP BY dia
            )
        """,
    )

    def _avisar_exit(**context):
        """Callback informatiu: en produccio enviaria a Slack o Chat."""
        print(f"Proces nocturn completat per a {context['ds']}")

    fi = PythonOperator(
        task_id="fin",
        python_callable=_avisar_exit,
        trigger_rule="all_success",
    )

    # -----------------------------------------------------------------------
    # DEPENDENCIES: l operador >> significa "i despres".
    # Aquesta es la declaracio completa del graf, en quatre linies.
    # -----------------------------------------------------------------------
    (
        inici
        >> exportar_comandes
        >> esperar_fitxer
        >> carregar_comandes
        >> consolidar_comandes
        >> llancar_dataflow
        >> grup_agregacions
        >> refrescar_vista
        >> comprovar_qualitat
        >> fi
    )

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

analisi_cistella = DataprocCreateBatchOperator(
    task_id="analisis_cesta_mensual",
    region=REGIO,
    project_id=PROJECTE_DADES,
    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 _comptar_files(**context):
    from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook
    hook = BigQueryHook(location=REGIO, use_legacy_sql=False)
    files = hook.get_first(
        f"""SELECT COUNT(*) FROM `{PROJECTE_DADES}.{DATASET}.pedidos_staging`"""
    )[0]
    # El que retorna la funcio es desa automaticament com a XCom
    return int(files)


def _validar_volum(**context):
    files = context["ti"].xcom_pull(task_ids="contar_filas_cargadas")
    if files == 0:
        raise ValueError(f"Cap comanda carregada el {context['ds']}")
    if files > 5000:
        raise ValueError(f"{files} comandes: volum anomal, revisar duplicats")
    print(f"Volum correcte: {files} comandes")


comptar_files = PythonOperator(task_id="contar_filas_cargadas",
                               python_callable=_comptar_files)
validar_volum = PythonOperator(task_id="validar_volumen",
                               python_callable=_validar_volum)

from airflow.models import Variable

llindar = 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 despres d a
a << b                    # a despres de b
a >> [b, c] >> d          # b i c en paral·lel, tots dos despres d a; d despres d ambdos
[a, b] >> c               # c espera que acabin a i b

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

esborrar_cluster = DataprocDeleteClusterOperator(
    task_id="borrar_cluster",
    cluster_name="cluster-efimero-{{ ds_nodash }}",
    region=REGIO,
    project_id=PROJECTE_DADES,
    # CRITIC: s esborra el cluster PASSI EL QUE PASSI.
    # Amb all_success, una fallada del job deixaria el cluster ences
    # i facturant indefinidament.
    trigger_rule="all_done",
)

avisar_fallada = PythonOperator(
    task_id="avisar_fallo",
    python_callable=_notificar_a_slack,
    trigger_rule="one_failed",
)
[carregar_comandes, llancar_dataflow, refrescar_vista] >> avisar_fallada

# MALAMENT: no reproduible
"query": "SELECT ... WHERE fecha = CURRENT_DATE()"

# BE: reproduible
"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 -- Proces nocturn d AlpinaShop amb 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 es passa data, s utilitza ahir. Permet reprocessar
          # invocant el flux amb {"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. L API retorna una OPERACIO de llarga
    #    durada: cal esperar-la, no donar-la per feta.
    # -----------------------------------------------------------------
    - 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) Comprovar que el fitxer existeix. L equivalent 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) Carregar a 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) Llancar Dataflow amb 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) Agregacions EN PARAL·LEL amb la branca '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) Comprovacio de qualitat: si falla, es llanca un error explicit
    # -----------------------------------------------------------------
    - 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: ${"QUALITAT: no hi ha vendes agregades per a " + fecha}

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

# ---------------------------------------------------------------------
# SUBFLUXOS reutilitzables
# ---------------------------------------------------------------------
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="Orquestracio nocturna amb 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

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

# Veure el resultat
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 proces nocturn"

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 -- Analisi de cistella i marge 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

PROJECTE = "alpinashop-datos"
DATASET = "alpinashop_analitica"
BUCKET = "alpinashop-datalake"
REGIO = "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 * *",                     # dia 1 de cada mes a les 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 a l ultim dia
    PRIMER_DIA_MES_ANT = "{{ macros.ds_format(macros.ds_add(ds, -1), '%Y-%m-%d', '%Y-%m-01') }}"
    ULTIM_DIA_MES_ANT = "{{ macros.ds_add(ds, -1) }}"

    # 1) Comprovar que hi ha dades del mes anterior
    comprovar_dades = BigQueryCheckOperator(
        task_id="comprobar_datos_mes_anterior",
        location=REGIO,
        use_legacy_sql=False,
        sql=f"""
            SELECT COUNT(*) > 0
            FROM `{PROJECTE}.{DATASET}.lineas_pedido`
            WHERE fecha_pedido BETWEEN DATE '{PRIMER_DIA_MES_ANT}'
                                   AND DATE '{ULTIM_DIA_MES_ANT}'
        """,
    )

    # 2) Analisi de cistella a Dataproc Serverless
    analisi_cistella = DataprocCreateBatchOperator(
        task_id="analisis_cesta",
        region=REGIO,
        project_id=PROJECTE,
        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={ULTIM_DIA_MES_ANT}",
                ],
            },
            "runtime_config": {"version": "2.2"},
            "environment_config": {
                "execution_config": {
                    "service_account":
                        "[email protected]",
                    "subnetwork_uri": "sn-datos-euw1",
                }
            },
        },
    )

    # 3) Dues consultes en paral·lel
    with TaskGroup(group_id="informes_mensuales") as informes:

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

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

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

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

    comprovar_dades >> analisi_cistella >> 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) Comprovar que hi ha dades
    - 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: ${"Sense linies de comanda per al mes " + mes}

    # 2) Dataproc Serverless: operacio de llarga durada
    - 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) Dues consultes en paral·lel
    - 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}

# Sondeig de l operacio de llarga durada: Workflows NO pot
# esperar 20 minuts en una sola crida HTTP (limit de 30 min,
# i a mes el batch podria trigar mes).
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 fallit: " + default(map.get(estado.body, "stateMessage"), "sense detall")}
          - condition: ${estado.body.state == "CANCELLED"}
            raise: "Batch de Dataproc cancel·lat"
    - 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_vendes = BigQueryInsertJobOperator(
    task_id="ventas_por_categoria",
    location=REGIO,
    configuration={
        "query": {
            "useLegacySql": False,
            # Escriu NOMES la particio de la data logica
            "destinationTable": {
                "projectId": PROJECTE,
                "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 `{PROJECTE}.{DATASET}.lineas_pedido` AS l
                JOIN `{PROJECTE}.{DATASET}.productos`     AS pr USING (sku)
                WHERE l.fecha_pedido = DATE '{{{{ ds }}}}'
                GROUP BY dia, pr.categoria
            """,
        }
    },
)

comprovar_qualitat = BigQueryCheckOperator(
    task_id="comprobar_calidad_datos",
    location=REGIO,
    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 `{PROJECTE}.{DATASET}.agg_ventas_categoria_dia`
          WHERE dia = DATE '{{{{ ds }}}}'
        ),
        control_duplicados AS (
          SELECT COUNT(*) AS duplicados FROM (
            SELECT pedido_id
            FROM `{PROJECTE}.{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 taula de comandes conservant l ultima versio
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) Reprocessar les agregacions del periode afectat, ja amb el DAG corregit
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