from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any

import pendulum
import yaml
from airflow.sdk import Asset


CONFIG_PATH = Path(__file__).resolve().parents[1] / "sources" / "local-dev.yaml"


@dataclass(frozen=True)
class HelloAssetSmokeConfig:
    producer_dag_id: str
    consumer_dag_id: str
    owner: str
    start_date: pendulum.DateTime
    tags: list[str]
    producer_schedule: str | None
    consumer_schedule: list[Asset]
    raw_asset: Asset
    consumed_asset: Asset
    raw_asset_uri: str
    consumed_asset_uri: str


def load_hello_asset_smoke_config(
    config_path: Path = CONFIG_PATH,
) -> HelloAssetSmokeConfig:
    source_config = yaml.safe_load(config_path.read_text()) or {}
    hello_assets = source_config["hello_assets"]
    schedule_config = source_config["schedules"]["hello_assets"]

    raw_asset = _build_asset(hello_assets["raw_asset"])
    consumed_asset = _build_asset(hello_assets["consumed_asset"])
    raw_asset_uri = hello_assets["raw_asset"]["uri"]

    return HelloAssetSmokeConfig(
        producer_dag_id=hello_assets["producer_dag_id"],
        consumer_dag_id=hello_assets["consumer_dag_id"],
        owner=hello_assets["owner"],
        start_date=pendulum.parse(hello_assets["start_date"]),
        tags=list(hello_assets["tags"]),
        producer_schedule=_producer_schedule(schedule_config["producer"]),
        consumer_schedule=_asset_schedule(schedule_config["consumer"], raw_asset, raw_asset_uri),
        raw_asset=raw_asset,
        consumed_asset=consumed_asset,
        raw_asset_uri=raw_asset_uri,
        consumed_asset_uri=hello_assets["consumed_asset"]["uri"],
    )


def _build_asset(asset_config: dict[str, Any]) -> Asset:
    return Asset(
        uri=asset_config["uri"],
        name=asset_config["name"],
        extra=dict(asset_config.get("extra", {})),
    )


def _producer_schedule(schedule_value: str) -> str | None:
    if schedule_value == "manual":
        return None
    return schedule_value


def _asset_schedule(schedule_value: str, asset: Asset, asset_uri: str) -> list[Asset]:
    expected = f"asset:{asset_uri}"
    if schedule_value != expected:
        raise ValueError(
            f"Expected hello asset consumer schedule {expected!r}, got {schedule_value!r}"
        )
    return [asset]