# Copyright 1999-2020 Gentoo Authors
# Distributed under the terms of the GNU General Public License v2

import copy
import io
import sys

import portage
from portage import os
from portage import _encodings
from portage import _unicode_encode
from portage import _unicode_decode
from portage.checksum import _hash_filter
from portage.elog.messages import eerror
from portage.package.ebuild.fetch import (
    _check_distfile,
    _drop_privs_userfetch,
    _want_userfetch,
    fetch,
)
from portage.util._async.AsyncTaskFuture import AsyncTaskFuture
from portage.util._async.ForkProcess import ForkProcess
from portage.util._pty import _create_pty_or_pipe
from _emerge.CompositeTask import CompositeTask


class EbuildFetcher(CompositeTask):

    __slots__ = (
        "config_pool",
        "ebuild_path",
        "fetchonly",
        "fetchall",
        "logfile",
        "pkg",
        "prefetch",
        "_fetcher_proc",
    )

    def __init__(self, **kwargs):
        CompositeTask.__init__(self, **kwargs)
        self._fetcher_proc = _EbuildFetcherProcess(**kwargs)

    def async_already_fetched(self, settings):
        """
        Returns True if all files already exist locally and have correct
        digests, otherwise return False. When returning True, appropriate
        digest checking messages are produced for display and/or logging.
        When returning False, no messages are produced, since we assume
        that a fetcher process will later be executed in order to produce
        such messages. This will raise InvalidDependString if SRC_URI is
        invalid.
        """
        return self._fetcher_proc.async_already_fetched(settings)

    def _start(self):
        self._start_task(
            AsyncTaskFuture(future=self._fetcher_proc._async_uri_map()),
            self._start_fetch,
        )

    def _start_fetch(self, uri_map_task):
        self._assert_current(uri_map_task)
        if uri_map_task.cancelled:
            self._default_final_exit(uri_map_task)
            return

        try:
            uri_map = uri_map_task.future.result()
        except portage.exception.InvalidDependString as e:
            msg_lines = []
            msg = "Fetch failed for '%s' due to invalid SRC_URI: %s" % (self.pkg.cpv, e)
            msg_lines.append(msg)
            self._fetcher_proc._eerror(msg_lines)
            self._current_task = None
            self.returncode = 1
            self._async_wait()
            return

        # First get the SRC_URI metadata (it's not cached in self.pkg.metadata
        # because some packages have an extremely large SRC_URI value).
        self._start_task(
            AsyncTaskFuture(
                future=self.pkg.root_config.trees["porttree"].dbapi.async_aux_get(
                    self.pkg.cpv, ["SRC_URI"], myrepo=self.pkg.repo, loop=self.scheduler
                )
            ),
            self._start_with_metadata,
        )

    def _start_with_metadata(self, aux_get_task):
        self._assert_current(aux_get_task)
        if aux_get_task.cancelled:
            self._default_final_exit(aux_get_task)
            return

        (self._fetcher_proc.src_uri,) = aux_get_task.future.result()
        self._start_task(self._fetcher_proc, self._default_final_exit)


class _EbuildFetcherProcess(ForkProcess):

    __slots__ = (
        "config_pool",
        "ebuild_path",
        "fetchonly",
        "fetchall",
        "pkg",
        "prefetch",
        "src_uri",
        "_digests",
        "_manifest",
        "_settings",
        "_uri_map",
    )

    def async_already_fetched(self, settings):
        result = self.scheduler.create_future()

        def uri_map_done(uri_map_future):
            if uri_map_future.cancelled():
                result.cancel()
                return

            if uri_map_future.exception() is not None or result.cancelled():
                if not result.cancelled():
                    result.set_exception(uri_map_future.exception())
                return

            uri_map = uri_map_future.result()
            if uri_map:
                result.set_result(self._check_already_fetched(settings, uri_map))
            else:
                result.set_result(True)

        uri_map_future = self._async_uri_map()
        result.add_done_callback(
            lambda result: uri_map_future.cancel() if result.cancelled() else None
        )
        uri_map_future.add_done_callback(uri_map_done)
        return result

    def _check_already_fetched(self, settings, uri_map):
        digests = self._get_digests()
        distdir = settings["DISTDIR"]
        allow_missing = self._get_manifest().allow_missing

        for filename in uri_map:
            # Use stat rather than lstat since fetch() creates
            # symlinks when PORTAGE_RO_DISTDIRS is used.
            try:
                st = os.stat(os.path.join(distdir, filename))
            except OSError:
                return False
            if st.st_size == 0:
                return False
            expected_size = digests.get(filename, {}).get("size")
            if expected_size is None:
                continue
            if st.st_size != expected_size:
                return False

        hash_filter = _hash_filter(settings.get("PORTAGE_CHECKSUM_FILTER", ""))
        if hash_filter.transparent:
            hash_filter = None
        stdout_orig = sys.stdout
        stderr_orig = sys.stderr
        global_havecolor = portage.output.havecolor
        out = io.StringIO()
        eout = portage.output.EOutput()
        eout.quiet = settings.get("PORTAGE_QUIET") == "1"
        success = True
        try:
            sys.stdout = out
            sys.stderr = out
            if portage.output.havecolor:
                portage.output.havecolor = not self.background

            for filename in uri_map:
                mydigests = digests.get(filename)
                if mydigests is None:
                    if not allow_missing:
                        success = False
                        break
                    continue
                ok, st = _check_distfile(
                    os.path.join(distdir, filename),
                    mydigests,
                    eout,
                    show_errors=False,
                    hash_filter=hash_filter,
                )
                if not ok:
                    success = False
                    break
        except portage.exception.FileNotFound:
            # A file disappeared unexpectedly.
            return False
        finally:
            sys.stdout = stdout_orig
            sys.stderr = stderr_orig
            portage.output.havecolor = global_havecolor

        if success:
            # When returning unsuccessfully, no messages are produced, since
            # we assume that a fetcher process will later be executed in order
            # to produce such messages.
            msg = out.getvalue()
            if msg:
                self.scheduler.output(msg, log_path=self.logfile)

        return success

    def _start(self):

        root_config = self.pkg.root_config
        portdb = root_config.trees["porttree"].dbapi
        ebuild_path = self._get_ebuild_path()
        # This is initialized by an earlier _async_uri_map call.
        uri_map = self._uri_map

        if not uri_map:
            # Nothing to fetch.
            self.returncode = os.EX_OK
            self._async_wait()
            return

        settings = self.config_pool.allocate()
        settings.setcpv(self.pkg)
        settings.configdict["pkg"]["SRC_URI"] = self.src_uri
        portage.doebuild_environment(ebuild_path, "fetch", settings=settings, db=portdb)

        if self.prefetch and self._prefetch_size_ok(uri_map, settings, ebuild_path):
            self.config_pool.deallocate(settings)
            self.returncode = os.EX_OK
            self._async_wait()
            return

        nocolor = settings.get("NOCOLOR")

        if self.prefetch:
            settings["PORTAGE_PARALLEL_FETCHONLY"] = "1"

        if self.background:
            nocolor = "true"

        if nocolor is not None:
            settings["NOCOLOR"] = nocolor

        self._settings = settings
        self.log_filter_file = settings.get("PORTAGE_LOG_FILTER_FILE_CMD")
        ForkProcess._start(self)

        # Free settings now since it's no longer needed in
        # this process (the subprocess has a private copy).
        self.config_pool.deallocate(settings)
        settings = None
        self._settings = None

    def _run(self):
        # Force consistent color output, in case we are capturing fetch
        # output through a normal pipe due to unavailability of ptys.
        portage.output.havecolor = self._settings.get("NOCOLOR") not in ("yes", "true")

        # For userfetch, drop privileges for the entire fetch call, in
        # order to handle DISTDIR on NFS with root_squash for bug 601252.
        if _want_userfetch(self._settings):
            _drop_privs_userfetch(self._settings)

        rval = 1
        allow_missing = (
            self._get_manifest().allow_missing or "digest" in self._settings.features
        )
        if fetch(
            self._uri_map,
            self._settings,
            fetchonly=self.fetchonly,
            digests=copy.deepcopy(self._get_digests()),
            allow_missing_digests=allow_missing,
        ):
            rval = os.EX_OK
        return rval

    def _get_ebuild_path(self):
        if self.ebuild_path is not None:
            return self.ebuild_path
        portdb = self.pkg.root_config.trees["porttree"].dbapi
        self.ebuild_path = portdb.findname(self.pkg.cpv, myrepo=self.pkg.repo)
        if self.ebuild_path is None:
            raise AssertionError("ebuild not found for '%s'" % self.pkg.cpv)
        return self.ebuild_path

    def _get_manifest(self):
        if self._manifest is None:
            pkgdir = os.path.dirname(self._get_ebuild_path())
            self._manifest = (
                self.pkg.root_config.settings.repositories.get_repo_for_location(
                    os.path.dirname(os.path.dirname(pkgdir))
                ).load_manifest(pkgdir, None)
            )
        return self._manifest

    def _get_digests(self):
        if self._digests is None:
            self._digests = self._get_manifest().getTypeDigests("DIST")
        return self._digests

    def _async_uri_map(self):
        """
        This calls the portdbapi.async_fetch_map method and returns the
        resulting Future (may contain InvalidDependString exception).
        """
        if self._uri_map is not None:
            result = self.scheduler.create_future()
            result.set_result(self._uri_map)
            return result

        pkgdir = os.path.dirname(self._get_ebuild_path())
        mytree = os.path.dirname(os.path.dirname(pkgdir))
        use = None
        if not self.fetchall:
            use = self.pkg.use.enabled
        portdb = self.pkg.root_config.trees["porttree"].dbapi

        def cache_result(result):
            try:
                self._uri_map = result.result()
            except Exception:
                # The caller handles this when it retrieves the result.
                pass

        result = portdb.async_fetch_map(
            self.pkg.cpv, useflags=use, mytree=mytree, loop=self.scheduler
        )
        result.add_done_callback(cache_result)
        return result

    def _prefetch_size_ok(self, uri_map, settings, ebuild_path):
        distdir = settings["DISTDIR"]

        sizes = {}
        for filename in uri_map:
            # Use stat rather than lstat since portage.fetch() creates
            # symlinks when PORTAGE_RO_DISTDIRS is used.
            try:
                st = os.stat(os.path.join(distdir, filename))
            except OSError:
                return False
            if st.st_size == 0:
                return False
            sizes[filename] = st.st_size

        digests = self._get_digests()
        for filename, actual_size in sizes.items():
            size = digests.get(filename, {}).get("size")
            if size is None:
                continue
            if size != actual_size:
                return False

        # All files are present and sizes are ok. In this case the normal
        # fetch code will be skipped, so we need to generate equivalent
        # output here.
        if self.logfile is not None:
            f = io.open(
                _unicode_encode(
                    self.logfile, encoding=_encodings["fs"], errors="strict"
                ),
                mode="a",
                encoding=_encodings["content"],
                errors="backslashreplace",
            )
            for filename in uri_map:
                f.write(
                    _unicode_decode(
                        (" * %s size ;-) ..." % filename).ljust(73) + "[ ok ]\n"
                    )
                )
            f.close()

        return True

    def _pipe(self, fd_pipes):
        """When appropriate, use a pty so that fetcher progress bars,
        like wget has, will work properly."""
        if self.background or not sys.stdout.isatty():
            # When the output only goes to a log file,
            # there's no point in creating a pty.
            return os.pipe()
        stdout_pipe = None
        if not self.background:
            stdout_pipe = fd_pipes.get(1)
        got_pty, master_fd, slave_fd = _create_pty_or_pipe(copy_term_size=stdout_pipe)
        return (master_fd, slave_fd)

    def _eerror(self, lines):
        out = io.StringIO()
        for line in lines:
            eerror(line, phase="unpack", key=self.pkg.cpv, out=out)
        msg = out.getvalue()
        if msg:
            self.scheduler.output(msg, log_path=self.logfile)

    def _proc_join_done(self, proc, future):
        """
        Extend _proc_join_done to emit an eerror message for fetch failure.
        """
        if not self.prefetch and not future.cancelled() and proc.exitcode != os.EX_OK:
            msg_lines = []
            msg = "Fetch failed for '%s'" % (self.pkg.cpv,)
            if self.logfile is not None:
                msg += ", Log file:"
            msg_lines.append(msg)
            if self.logfile is not None:
                msg_lines.append(" '%s'" % (self.logfile,))
            self._eerror(msg_lines)
        super(_EbuildFetcherProcess, self)._proc_join_done(proc, future)