import json
import warnings
from functools import cached_property
from hashlib import md5
from typing import Any
from typing import Dict
from typing import Optional
from typing import Union
from typing import cast
from urllib.parse import urljoin

import requests
from requests import PreparedRequest
from requests import Response
from requests.auth import AuthBase

from anaconda_auth import __version__ as version
from anaconda_auth.config import AnacondaAuthConfig
from anaconda_auth.exceptions import TokenExpiredError
from anaconda_auth.exceptions import TokenNotFoundError
from anaconda_auth.token import TokenInfo
from anaconda_auth.utils import get_hostname

# VersionInfo was renamed and is deprecated in semver>=3
try:
    from semver.version import Version
except ImportError:
    # In semver<3, it's called VersionInfo
    from semver import VersionInfo as Version


def login_required(response: Response, *args: Any, **kwargs: Any) -> Response:
    has_auth_header = response.request.headers.get("Authorization", False)

    if response.status_code in [401, 403]:
        try:
            error_code = response.json().get("error", {}).get("code", "")
        except requests.JSONDecodeError:
            error_code = ""

        if error_code == "auth_required":
            if has_auth_header:
                response.reason = "Your API key or login token is invalid."
            else:
                response.reason = (
                    "You must login before using this API endpoint"
                    " or provide an api_key to your client."
                )

    return response


class BearerAuth(AuthBase):
    def __init__(
        self, domain: Optional[str] = None, api_key: Optional[str] = None
    ) -> None:
        self.api_key = api_key
        if domain is None:
            domain = AnacondaAuthConfig().domain

        self._token_info = TokenInfo(domain=domain)

    def __call__(self, r: PreparedRequest) -> PreparedRequest:
        if not self.api_key:
            try:
                r.headers["Authorization"] = (
                    f"Bearer {self._token_info.get_access_token()}"
                )
            except (TokenNotFoundError, TokenExpiredError):
                pass
        else:
            r.headers["Authorization"] = f"Bearer {self.api_key}"
        return r


class BaseClient(requests.Session):
    _user_agent: str = f"anaconda-auth/{version}"
    _api_version: Optional[str] = None

    def __init__(
        self,
        base_uri: Optional[str] = None,
        domain: Optional[str] = None,
        api_key: Optional[str] = None,
        user_agent: Optional[str] = None,
        api_version: Optional[str] = None,
        ssl_verify: Optional[bool] = None,
        extra_headers: Optional[Union[str, dict]] = None,
        hash_hostname: Optional[bool] = None,
    ):
        super().__init__()

        if base_uri and domain:
            raise ValueError("Can only specify one of `domain` or `base_uri` argument")

        kwargs: Dict[str, Any] = {}
        if domain is not None:
            kwargs["domain"] = domain
        if api_key is not None:
            kwargs["api_key"] = api_key
        if ssl_verify is not None:
            kwargs["ssl_verify"] = ssl_verify
        if extra_headers is not None:
            kwargs["extra_headers"] = extra_headers
        if hash_hostname is not None:
            kwargs["hash_hostname"] = hash_hostname

        self.config = AnacondaAuthConfig(**kwargs)

        # base_url overrides domain
        self._base_uri = base_uri or f"https://{self.config.domain}"
        self.headers["User-Agent"] = user_agent or self._user_agent
        self.headers["X-Client-Hostname"] = get_hostname(hash=self.config.hash_hostname)
        self.api_version = api_version or self._api_version
        if self.api_version:
            self.headers["Api-Version"] = self.api_version

        if self.config.extra_headers is not None:
            if isinstance(self.config.extra_headers, str):
                try:
                    self.config.extra_headers = cast(
                        dict, json.loads(self.config.extra_headers)
                    )
                except json.decoder.JSONDecodeError:
                    raise ValueError(
                        f"{repr(self.config.extra_headers)} is not valid JSON."
                    )

            keys_to_add = self.config.extra_headers.keys() - self.headers.keys()
            for k in keys_to_add:
                self.headers[k] = self.config.extra_headers[k]

        self.auth = BearerAuth(domain=domain, api_key=self.config.api_key)
        self.hooks["response"].append(login_required)

    def urljoin(self, url: str) -> str:
        return urljoin(self._base_uri, url)

    def prepare_request(self, request: requests.Request) -> PreparedRequest:
        request.url = self.urljoin(str(request.url))
        return super().prepare_request(request)

    def request(
        self,
        method: Union[str, bytes],
        url: Union[str, bytes],
        *args: Any,
        **kwargs: Any,
    ) -> Response:
        joined_url = self.urljoin(str(url))

        # Ensure we don't set `verify` twice. If it is passed as a kwarg to this method,
        # that becomes the value. Otherwise, we use the value in `self.config.ssl_verify`.
        kwargs.setdefault("verify", self.config.ssl_verify)

        response = super().request(method, joined_url, *args, **kwargs)

        min_api_version_string = response.headers.get("Min-Api-Version")
        self._validate_api_version(min_api_version_string)

        return response

    @cached_property
    def account(self) -> dict:
        res = self.get("/api/account")
        res.raise_for_status()
        account = res.json()
        return account

    @property
    def name(self) -> str:
        user = self.account.get("user", {})

        first_name = user.get("first_name", "")
        last_name = user.get("last_name", "")
        if not first_name and not last_name:
            return self.email
        else:
            return f"{first_name} {last_name}".strip()

    @property
    def email(self) -> str:
        value = self.account.get("user", {}).get("email")
        if value is None:
            raise ValueError(
                "Something is wrong with your account. An email address could not be found."
            )
        else:
            return value

    @cached_property
    def avatar(self) -> Union[bytes, None]:
        hashed = md5(self.email.encode("utf-8")).hexdigest()
        res = requests.get(
            f"https://gravatar.com/avatar/{hashed}.png?size=120&d=404",
            verify=self.config.ssl_verify,
        )
        if res.ok:
            return res.content
        else:
            return None

    def _validate_api_version(self, min_api_version_string: Optional[str]) -> None:
        """Validate that the client API version against the min API version from the service."""
        if min_api_version_string is None or self.api_version is None:
            return None

        # Convert to optional Version objects
        api_version = _parse_semver_string(self.api_version)
        min_api_version = _parse_semver_string(min_api_version_string)

        if api_version is None or min_api_version is None:
            return None

        if api_version < min_api_version:
            warnings.warn(
                f"Client API version is {self.api_version}, minimum supported API version is {min_api_version_string}. "
                "You may need to update your client.",
                DeprecationWarning,
            )


def _parse_semver_string(version: str) -> Optional[Version]:
    """Parse a version string into a semver Version object, stripping off any leading zeros from the components.

    If the version string is invalid, returns None.

    """
    norm_version = ".".join(s.lstrip("0") for s in version.split("."))
    try:
        return Version.parse(norm_version)
    except ValueError:
        return None


def client_factory(
    user_agent: Optional[str], api_version: Optional[str] = None
) -> BaseClient:
    return BaseClient(user_agent=user_agent, api_version=api_version)
