# -*- coding: utf-8 -*-

"""Additional utility functions to use for installable applications."""

from __future__ import annotations

__all__ = ['install_launch', 'install_dmg']

import stat

import webbrowser
import collections.abc
import contextlib
import hashlib
import io
import os
import posixpath
import shutil
import subprocess  # nosec: B404
import sys
import tempfile
import typing
import urllib.parse

import requests

from anaconda_navigator.config import preferences
from anaconda_navigator.static.content import DATA_PATH
from anaconda_navigator.utils.logs import logger

if typing.TYPE_CHECKING:
    from .. import base


def safe_domain(domain: str) -> bool:
    """Check if installer at :code:`domain` is safe to use."""
    if preferences.ALLOW_UNSAFE_INSTALLERS:
        return True
    domain = f'.{domain}'
    return any(domain.endswith(item) for item in preferences.SAFE_INSTALLER_DOMAINS)


def safe_filename(filename: str) -> str | None:
    """Check if installer at :code:`filename` is safe to use."""
    if not filename:
        return None
    filename = os.path.abspath(os.path.join(DATA_PATH, filename))

    if preferences.ALLOW_UNSAFE_INSTALLERS:
        return filename

    try:
        if os.path.samefile(os.path.commonpath([DATA_PATH, filename]), DATA_PATH):
            return filename
    except ValueError:
        return None

    return None


@contextlib.contextmanager
def retrieve(url: str) -> collections.abc.Iterator[str]:
    """
    Download installer.

    Must be used as a context manager.

    :param url: URI of the installer (might be a path to local file).
    :return: Path to a downloaded file.
    """
    filename: str | None
    parsed: urllib.parse.SplitResult = urllib.parse.urlsplit(url)
    if (parsed.scheme in ('', 'file')) and (filename := safe_filename(parsed.path)):
        yield filename
    elif safe_domain(parsed.hostname or ''):
        response: requests.Response
        with requests.get(url, stream=True, timeout=30.0, verify=True) as response:
            response.raise_for_status()

            descriptor: int
            stream: io.BufferedWriter
            descriptor, filename = tempfile.mkstemp(suffix=posixpath.splitext(parsed.path)[1])
            with os.fdopen(descriptor, 'wb') as stream:
                chunk: bytes
                for chunk in response.iter_content(chunk_size=4_194_304):
                    stream.write(chunk)

        try:
            yield filename
        finally:
            try:
                os.remove(filename)
            except OSError:
                pass
    else:
        raise ValueError('source is unsafe')


@contextlib.contextmanager
def retrieve_and_validate(url: str, **kwargs: str | None) -> collections.abc.Iterator[str]:
    """Same as :func:`~retrieve`, but also checks for resulting files hash sums."""
    filename: str
    with retrieve(url) as filename:
        query: list[tuple[hashlib._Hash, str]] = [
            (hashlib.new(name), value)
            for name, value in kwargs.items()
            if value
        ]
        if query:
            expectation: str
            hasher: hashlib._Hash
            stream: io.BufferedReader
            with open(filename, 'rb') as stream:
                buffer: bytes
                while buffer := stream.read(4_294_967_296):
                    for hasher, _ in query:
                        hasher.update(buffer)
            for hasher, expectation in query:
                if hasher.hexdigest() != expectation:
                    raise ValueError(
                        f'installer hash missmatch for {os.path.basename(filename)}: '
                        f'got {hasher.hexdigest()!r} instead of {expectation!r}'
                    )
        yield filename


def install_launch(  # pylint: disable=too-many-arguments
        app: 'base.BaseInstallableApp',  # pylint: disable=unused-argument
        *,
        installer_url: str,
        fallback_url: str,
        flags: collections.abc.Sequence[str] = (),
        md5: str | None = None,
        sha256: str | None = None,
) -> None:
    """
    Install application by launching an installer.

    :param installer_url: URL to installer.
    :param fallback_url: URL to open in browser if installer fails.
    :param flags: Extra command line flags to pass to the installer.
    :param md5: Expected MD5 hash of the installer file.
    :param sha256: Expected SHA256 hash of the installer file.
    """
    try:
        with retrieve_and_validate(installer_url, md5=md5, sha256=sha256) as filename:
            if sys.platform in ('linux', 'darwin'):
                info: os.stat_result = os.stat(filename)
                mode: int = info.st_mode | stat.S_IXUSR | stat.S_IXGRP
                if mode != info.st_mode:
                    os.chmod(filename, mode)
            subprocess.check_call([filename, *flags])  # nosec: B603
    except Exception:  # pylint: disable=broad-exception-caught
        webbrowser.open(fallback_url)


def install_dmg(  # pylint: disable=too-many-arguments
        app: 'base.BaseInstallableApp',  # pylint: disable=unused-argument
        *,
        installer_url: str,
        fallback_url: str,
        md5: str | None = None,
        reinstall: bool = False,
        sha256: str | None = None,
        system: bool = True,
) -> None:
    """
    Install application from a DMG installer.

    .. warning::

        Works only on macOS.

    :param installer_url: URL to installer.
    :param fallback_url: URL to open in browser if installer fails.
    :param md5: Expected MD5 hash of the installer file.
    :param reinstall: Reinstall application if it is already installed.
    :param sha256: Expected SHA256 hash of the installer file.
    :param system: Install application system-wide instead of "only for user".
    """
    try:
        import dmglib  # pylint: disable=import-error,import-outside-toplevel

        filename: str
        with retrieve_and_validate(installer_url, md5=md5, sha256=sha256) as filename:
            if dmglib.dmg_already_attached(filename):
                dmglib.dmg_detach_already_attached(filename)

            image: dmglib.DiskImage = dmglib.DiskImage(filename)
            try:
                if image.has_license_agreement():
                    logger.debug('DMG %r requires user to accept license agreement manually', installer_url)
                    raise ValueError('installer requires user to accept license agreement')

                if system:
                    target: str = '/Applications'
                else:
                    target = os.path.join(os.environ['HOME'], 'Applications')

                applications: list[tuple[str, str]] = [
                    (name, os.path.join(root, name))
                    for root in image.attach()
                    for name in os.listdir(root)
                    if name.endswith('.app')
                ]
                if not applications:
                    logger.debug('DMG %r does not contain any applications', installer_url)
                    raise ValueError('Installer does not contain any applications')

                app_name: str
                app_root: str
                for app_name, app_root in applications:
                    app_target: str = os.path.join(target, app_name)
                    if os.path.exists(app_target):
                        if reinstall:
                            shutil.rmtree(app_target)
                        else:
                            continue
                    shutil.copytree(app_root, app_target, symlinks=True)
            finally:
                image.detach()
    except Exception:  # pylint: disable=broad-exception-caught
        webbrowser.open(fallback_url)