"""
dag_vendas_diarias — pipeline noturno de e-commerce
====================================================

Implementação de referência da simulação em orquestrador.html
(https://murilolympius.com.br/orquestrador). Loja, conexões e caminhos
são fictícios; os jobs Spark citados ficam em jobs/*.py.

Toda madrugada, às 00:30, o DAG junta pedidos, pagamentos, marketplaces,
anúncios, catálogo e estoque do dia anterior e publica o painel de vendas
que a diretoria abre às 08:00.

O que está aqui de propósito:
  * pools separados por tipo de carga (ingestão, Spark, entrega);
  * sensores em modo reschedule — esperar arquivo não ocupa vaga;
  * retries com backoff exponencial para falhas transitórias (429, timeout);
  * AirflowFailException para o que retry não resolve (quebra de contrato);
  * SLA no painel + alerta preventivo quando a previsão passa das 08:00;
  * estoque desacoplado do painel: atraso no ERP só segura a reposição;
  * todas as escritas sobrescrevem a partição do dia → idempotente,
    então catchup/backfill pode reprocessar qualquer noite sem duplicar.
"""
from __future__ import annotations

from datetime import datetime, timedelta

import pendulum
from airflow import DAG
from airflow.exceptions import AirflowFailException
from airflow.models import Variable
from airflow.operators.python import PythonOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.providers.sftp.sensors.sftp import SFTPSensor
from airflow.utils.task_group import TaskGroup

TZ = pendulum.timezone("America/Sao_Paulo")
BRONZE = "s3a://lake/bronze"
SPARK_CONF = {
    "spark.sql.adaptive.enabled": "true",                  # AQE ajusta as partições de shuffle
    "spark.sql.autoBroadcastJoinThreshold": "64MB",        # dimensões pequenas viram broadcast join
    "spark.sql.sources.partitionOverwriteMode": "dynamic",  # overwrite só da partição do dia
}


# ---------------------------------------------------------------------------
# Callbacks de alerta
# ---------------------------------------------------------------------------
def alerta_plantao(context):
    """Chamado em qualquer falha definitiva: abre incidente para o plantão."""
    ti = context["task_instance"]
    notificar(
        canal="#dados-plantao",
        texto=f"{ti.dag_id}.{ti.task_id} falhou na noite {context['ds']} "
              f"(tentativa {ti.try_number}). Log: {ti.log_url}",
    )


def sla_perdido(dag, task_list, blocking_task_list, slas, blocking_tis):
    notificar(canal="#diretoria-dados", texto=f"Painel de vendas atrasado: {task_list}")


def notificar(canal: str, texto: str) -> None:
    ...  # Slack / PagerDuty — fora do escopo deste exemplo


# ---------------------------------------------------------------------------
# Ingestão
# ---------------------------------------------------------------------------
class RateLimited(Exception):
    """HTTP 429: transitório, o retry com backoff resolve."""


def extrair_pedidos_incremental(ds: str, **_):
    """Lê só o que mudou desde a última execução (watermark em updated_at)."""
    from jobs.common import spark_session

    spark = spark_session("ing_pedidos")
    watermark = Variable.get("pedidos_watermark", default_var=f"{ds} 00:00:00")
    df = spark.read.jdbc(
        url=Variable.get("loja_jdbc_url"),
        table=f"(SELECT * FROM order_items WHERE updated_at > '{watermark}') q",
        properties={"fetchsize": "10000"},
    )
    df.write.mode("overwrite").parquet(f"{BRONZE}/pedidos/dt={ds}")
    Variable.set("pedidos_watermark", df.agg({"updated_at": "max"}).first()[0].isoformat())


def extrair_marketplaces(ds: str, **_):
    """API paginada com limite de requisições."""
    from jobs.common import marketplace_session, paginate, sink

    session = marketplace_session()
    with sink(f"{BRONZE}/marketplace/dt={ds}") as out:
        for page in paginate(f"/v2/orders?date={ds}", size=500):
            resp = session.get(page, timeout=120)
            if resp.status_code == 429:
                raise RateLimited(resp.headers.get("Retry-After", "?"))
            resp.raise_for_status()
            out.write(resp.json()["items"])


def extrair_anuncios(ds: str, **_):
    from jobs.common import ads_client, save

    report = ads_client().get_report(date=ds, level="campaign", fields=["cost", "clicks", "impressions"])
    save(report, f"{BRONZE}/ads/dt={ds}")


def extrair_catalogo(ds: str, prev_ds: str, **_):
    from jobs.common import pim_client, save

    save(pim_client().export(updated_since=prev_ds), f"{BRONZE}/catalogo/dt={ds}")


def carregar_arquivo(origem: str):
    def _carregar(ds: str, ds_nodash: str, **_):
        from jobs.common import load_csv

        load_csv(f"/{origem}/{ds_nodash}/", f"{BRONZE}/{origem}/dt={ds}")
    return _carregar


# ---------------------------------------------------------------------------
# Entrega
# ---------------------------------------------------------------------------
def publicar_painel(ds: str, **_):
    from jobs.common import refresh_dashboard

    refresh_dashboard("vendas_diarias", partition=ds)


def spark_job(task_id: str, app: str, pool: str = "spark", **kwargs) -> SparkSubmitOperator:
    return SparkSubmitOperator(
        task_id=task_id,
        application=f"jobs/{app}.py",
        application_args=["--data", "{{ ds }}"],
        conf=SPARK_CONF,
        pool=pool,
        **kwargs,
    )


# ---------------------------------------------------------------------------
# DAG
# ---------------------------------------------------------------------------
default_args = {
    "owner": "dados-ecommerce",
    "retries": 3,
    "retry_delay": timedelta(minutes=3),
    "retry_exponential_backoff": True,      # 3, 6, 12 min
    "max_retry_delay": timedelta(minutes=20),
    "execution_timeout": timedelta(hours=2),
    "on_failure_callback": alerta_plantao,
}

with DAG(
    dag_id="vendas_diarias",
    description="Pedidos, pagamentos, marketplaces, anúncios, catálogo e estoque → painel das 08:00",
    schedule="30 0 * * *",
    start_date=datetime(2026, 1, 1, tzinfo=TZ),
    catchup=True,                # backfill: cada noite é uma execução independente e idempotente
    max_active_runs=2,           # no máximo 2 noites reprocessando ao mesmo tempo
    default_args=default_args,
    sla_miss_callback=sla_perdido,
    tags=["ecommerce", "vendas", "sla-0800"],
) as dag:

    with TaskGroup("ingestao", default_args={"pool": "ingestao"}):
        pedidos = PythonOperator(task_id="pedidos", python_callable=extrair_pedidos_incremental)

        espera_pagamentos = SFTPSensor(
            task_id="arquivo_pagamentos",
            sftp_conn_id="gateway_sftp",
            path="/gateway/{{ ds_nodash }}/settlement.csv",
            mode="reschedule",
            poke_interval=300,
            timeout=6 * 3600,
        )
        pagamentos = PythonOperator(task_id="pagamentos", python_callable=carregar_arquivo("gateway"))

        marketplaces = PythonOperator(task_id="marketplaces", python_callable=extrair_marketplaces)
        anuncios = PythonOperator(task_id="anuncios", python_callable=extrair_anuncios)
        catalogo = PythonOperator(task_id="catalogo", python_callable=extrair_catalogo)

        espera_estoque = SFTPSensor(
            task_id="arquivo_estoque",
            sftp_conn_id="erp_sftp",
            path="/erp/{{ ds_nodash }}/estoque.csv",
            mode="reschedule",
            poke_interval=300,
            timeout=5 * 3600,
            soft_fail=True,      # sem arquivo até 05:30 → SKIPPED; o painel não depende dele
        )
        estoque = PythonOperator(task_id="estoque", python_callable=carregar_arquivo("erp"))

        espera_pagamentos >> pagamentos
        espera_estoque >> estoque

    # Quebra de contrato levanta AirflowFailException dentro do job: retry não adianta.
    validacao = spark_job("validacao", "validar_contratos", retries=0)
    dim_cliente = spark_job("dim_cliente", "dim_cliente_scd2")
    dim_produto = spark_job("dim_produto", "dim_produto_scd2")
    fato_vendas = spark_job("fato_vendas", "fato_vendas", priority_weight=50)
    reconciliacao = spark_job("reconciliacao", "reconciliar_pagamentos", retries=0)

    with TaskGroup("entrega"):
        painel = PythonOperator(
            task_id="painel_vendas",
            python_callable=publicar_painel,
            pool="entrega",
            sla=timedelta(hours=7, minutes=30),   # 00:30 + 7h30 = 08:00
            priority_weight=100,
        )
        roi = spark_job("roi_marketing", "roi_marketing", pool="entrega")
        reposicao = spark_job("reposicao", "reposicao_estoque", pool="entrega")

    [pedidos, pagamentos, marketplaces] >> validacao
    pedidos >> dim_cliente
    catalogo >> dim_produto
    [validacao, dim_cliente, dim_produto] >> fato_vendas
    fato_vendas >> reconciliacao >> painel
    [fato_vendas, anuncios] >> roi
    [fato_vendas, estoque] >> reposicao


# ---------------------------------------------------------------------------
# Exemplo do que roda dentro de jobs/reconciliar_pagamentos.py
# ---------------------------------------------------------------------------
def checar_reconciliacao(total_pedidos: float, total_pagamentos: float, limite: float = 0.005) -> float:
    diff = abs(total_pedidos - total_pagamentos) / total_pagamentos
    if diff > limite:
        raise AirflowFailException(f"Divergência de {diff:.2%} entre pedidos e pagamentos")
    return diff