"""DAG trigger utilities for the API plugin.

NOTE: Runs inside the Airflow api-server process.
Handles import resolution across Airflow 3.x versions.
"""

import logging

log = logging.getLogger(__name__)

_trigger_dag = None

try:
    from airflow.api.common.trigger_dag import trigger_dag as _trigger_dag_fn
    from airflow.models.dag import DagRunTriggeredByType

    def _trigger_dag_wrapper(dag_id, conf, run_id=None):
        return _trigger_dag_fn(
            dag_id,
            conf=conf,
            run_id=run_id,
            triggered_by=DagRunTriggeredByType.REST_API,
        )

    _trigger_dag = _trigger_dag_wrapper
except ImportError:
    pass

if _trigger_dag is None:
    try:
        from airflow.models.dagbag import DagBag
        from airflow.models.dagrun import DagRun
        from airflow.utils import timezone as af_timezone
        from airflow.utils.state import DagRunState
        from airflow.utils.types import DagRunType

        def _trigger_dag_fallback(dag_id, conf, run_id=None):
            dagbag = DagBag(read_dags_from_db=True)
            dag = dagbag.get_dag(dag_id)
            if dag is None:
                raise ValueError(f"DAG '{dag_id}' not found")
            logical_date = af_timezone.utcnow()
            run_id = run_id or DagRun.generate_run_id(DagRunType.MANUAL, logical_date)
            return dag.create_dagrun(
                run_id=run_id,
                logical_date=logical_date,
                conf=conf,
                run_type=DagRunType.MANUAL,
                state=DagRunState.QUEUED,
            )

        _trigger_dag = _trigger_dag_fallback
    except ImportError:
        log.error("Cannot import Airflow DAG trigger API — endpoints that trigger DAGs will not work")


def trigger_dag(dag_id: str, conf: dict) -> None:
    """Trigger an Airflow DAG run.

    Args:
        dag_id: The DAG identifier.
        conf: Configuration dict passed as dag_run.conf.

    Raises:
        RuntimeError: If no DAG trigger API is available.
    """
    if _trigger_dag is None:
        raise RuntimeError("Airflow DAG trigger API not available")
    _trigger_dag(dag_id, conf=conf)