import importlib
import logging
import os
import subprocess
import urllib
from importlib.abc import Loader
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

import arn.s3
import boto3
import git

from conductor import core
from conductor.config import Project


def get_project_root() -> str:
    cwd = os.getcwd()
    try:
        repo = git.Repo(cwd, search_parent_directories=True)
        return repo.git.rev_parse("--show-toplevel")
    except git.exc.InvalidGitRepositoryError:  # type: ignore
        # Fall back to the current workdir if we are unable to resolve a git repo.
        return os.getcwd()


def get_project_files() -> List[str]:
    cwd = os.getcwd()
    repo = git.Repo(cwd, search_parent_directories=True)
    root = repo.git.rev_parse("--show-toplevel")
    tracked = git.cmd.Git(root).ls_files().split("\n")
    deleted = git.cmd.Git(root).ls_files(deleted=True).split("\n")
    untracked = (
        git.cmd.Git(root).ls_files(others=True, exclude_standard=True).split("\n")
    )
    return list(filter(lambda f: len(f) > 0 and f not in deleted, tracked + untracked))


def get_airflow_env(name: str):
    client = boto3.client("mwaa")
    resp = client.get_environment(Name=name)
    env = resp["Environment"]
    return env


def exec_sync(cmd: str) -> None:
    logging.info(cmd)
    p = subprocess.Popen(cmd, shell=True, stderr=subprocess.STDOUT)
    errcode = p.wait()
    if errcode != 0:
        logging.error("failed to run cmd: ", cmd)
        exit(errcode)


def load_project(project_root: str):
    spec = importlib.util.spec_from_file_location(
        "project_config", f"{project_root}/project_config.py"
    )
    project_config = importlib.util.module_from_spec(spec)
    assert isinstance(spec.loader, Loader)
    spec.loader.exec_module(project_config)

    project: Project = project_config.project  # type: ignore
    return project


def load_conductor_instances(project_root: str):
    conductor_instances = []
    instance_names = set()
    dags_path = Path(project_root, "dags")
    for file_path in dags_path.iterdir():
        # Only process .py files.
        if file_path.suffix != ".py":
            continue
        module_name = file_path.stem
        spec = importlib.util.spec_from_file_location(module_name, file_path)
        module = importlib.util.module_from_spec(spec)
        assert isinstance(spec.loader, Loader)
        spec.loader.exec_module(module)
        for item_name in dir(module):
            item = getattr(module, item_name)
            if type(item) == core.Conductor:
                if item.name in instance_names:
                    raise Exception(f"Multiple conductor instances named {item.name}")
                conductor_instances.append(item)
                instance_names.add(item.name)
    return conductor_instances


# input parameter config: AirflowConfig <- conductor_cdk
# https://git.xarth.tv/ml/conductor-cdk/blob/583a964c760ef3eed06bc902d5dd4909cb79acc5/lib/config.ts#L35
def get_dags_s3_path(config) -> Tuple[Optional[str], Optional[str]]:
    if config.dags_s3_path is not None:
        s3_url = urllib.parse.urlparse(config.dags_s3_path)
        return s3_url.netloc, s3_url.path.lstrip("/").rstrip("/")
    elif config.mwaa_environment_name is not None:
        airflow_env = get_airflow_env(config.mwaa_environment_name)
        bucket_name = arn.s3.BucketArn(airflow_env["SourceBucketArn"]).name
        dag_s3_path = str(Path(airflow_env["DagS3Path"]))
        return bucket_name, dag_s3_path
    return None, None


# See https://stackoverflow.com/questions/3232943/update-value-of-a-nested-dictionary-of-varying-depth/30655448#30655448
def deep_update(source: Dict[str, Any], overrides: Dict[str, Any]):
    """
    Update a nested dictionary
    Modify ``source`` in place.
    """
    for key, value in overrides.items():
        if isinstance(value, dict) and value:
            returned = deep_update(source.get(key, {}), value)
            source[key] = returned
        else:
            source[key] = overrides[key]
    return source
