"""Daily metrics rollup — extract, transform, load, on a schedule.

Pulls yesterday's raw events from an internal HTTP API, rolls them up per
channel, and posts the summary to a reporting endpoint. Runs at 03:00 UTC.

Written for **Airflow 3.x**, which exposes the TaskFlow decorators from
``airflow.sdk``. On Airflow 2.4-2.x, change the import to::

    from airflow.decorators import dag, task

Dependencies: ``requests`` and the standard library. No provider packages, so
this loads on a stock Airflow install.

Before you enable it
--------------------
1. Set three Airflow Variables (Admin -> Variables), or replace the
   ``Variable.get`` calls with your own config source:
   ``metrics_source_url``, ``metrics_sink_url``, ``metrics_api_token``.
2. Drop this file in your ``dags/`` folder and wait for the scheduler to parse
   it (or run ``airflow dags reserialize``).
3. Trigger it once manually with a past logical date before unpausing.

The token is read from a Variable rather than hard-coded; move it to a
Connection or your secrets backend if you have one.

MIT licensed. Review it before you point production data at it.
"""

from __future__ import annotations

from collections import defaultdict
from datetime import timedelta

import pendulum
import requests
from airflow.sdk import Variable, dag, task

REQUEST_TIMEOUT_SECONDS = 30


@dag(
    dag_id="daily_metrics_rollup",
    description="Roll up yesterday's events per channel and publish the summary.",
    schedule="0 3 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    max_active_runs=1,
    tags=["codebrewerz", "reporting"],
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
    },
)
def daily_metrics_rollup():
    @task
    def extract(data_interval_start=None) -> list[dict]:
        """Fetch the raw events for the interval this run covers."""
        day = pendulum.instance(data_interval_start).to_date_string()
        response = requests.get(
            Variable.get("metrics_source_url"),
            params={"date": day},
            headers={"Authorization": f"Bearer {Variable.get('metrics_api_token')}"},
            timeout=REQUEST_TIMEOUT_SECONDS,
        )
        response.raise_for_status()

        events = response.json()["events"]
        print(f"{day}: fetched {len(events)} events")
        return events

    @task
    def transform(events: list[dict]) -> dict:
        """Group by channel. Anything without a channel lands in 'unknown'."""
        totals: dict[str, dict[str, float]] = defaultdict(
            lambda: {"count": 0, "value": 0.0}
        )

        for event in events:
            channel = event.get("channel") or "unknown"
            totals[channel]["count"] += 1
            totals[channel]["value"] += float(event.get("value") or 0)

        summary = {
            channel: {"count": t["count"], "value": round(t["value"], 2)}
            for channel, t in sorted(
                totals.items(), key=lambda kv: kv[1]["value"], reverse=True
            )
        }
        print(f"channels: {list(summary)}")
        return summary

    @task
    def load(summary: dict, data_interval_start=None) -> str:
        """Publish the summary. Raises on a non-2xx so the run goes red."""
        day = pendulum.instance(data_interval_start).to_date_string()
        response = requests.post(
            Variable.get("metrics_sink_url"),
            json={"date": day, "channels": summary},
            headers={"Authorization": f"Bearer {Variable.get('metrics_api_token')}"},
            timeout=REQUEST_TIMEOUT_SECONDS,
        )
        response.raise_for_status()

        print(f"published rollup for {day}")
        return day

    load(transform(extract()))


daily_metrics_rollup()