utils.py

# Copyright (C) 2014 Anaconda, Inc
# SPDX-License-Identifier: BSD-3-Clause
from __future__ import annotations

import contextlib
import fnmatch
import hashlib
import json
import logging
import logging.config
import mmap
import os
import re
import shutil
import stat
import subprocess
import sys
import tarfile
import tempfile
import time
import urllib.parse as urlparse
import urllib.request as urllib
from collections import OrderedDict, defaultdict
from functools import lru_cache
from glob import glob
from io import StringIO
from itertools import filterfalse
from json.decoder import JSONDecodeError
from locale import getpreferredencoding
from os import walk
from os.path import (
    abspath,
    dirname,
    expanduser,
    expandvars,
    getmtime,
    getsize,
    isdir,
    isfile,
    islink,
    join,
)
from pathlib import Path
from threading import Thread
from typing import TYPE_CHECKING, Iterable, overload

import conda_package_handling.api
import filelock
import libarchive
import yaml
from conda.base.constants import (
    CONDA_PACKAGE_EXTENSION_V1,  # noqa: F401
    CONDA_PACKAGE_EXTENSION_V2,  # noqa: F401
    CONDA_PACKAGE_EXTENSIONS,
    KNOWN_SUBDIRS,
)
from conda.base.context import context
from conda.common.path import win_path_to_unix
from conda.exceptions import CondaHTTPError
from conda.gateways.connection.download import download
from conda.gateways.disk.create import TemporaryDirectory
from conda.gateways.disk.read import compute_sum
from conda.models.channel import Channel
from conda.models.match_spec import MatchSpec
from conda.models.records import PackageRecord
from conda.models.version import VersionOrder
from conda.utils import unix_path_to_win

from .exceptions import BuildLockError

if TYPE_CHECKING:
    from typing import Mapping, TypeVar

    from .metadata import MetaData

    T = TypeVar("T")
    K = TypeVar("K")
    V = TypeVar("V")

on_win = sys.platform == "win32"
on_mac = sys.platform == "darwin"
on_linux = sys.platform == "linux"

codec = getpreferredencoding() or "utf-8"
root_script_dir = os.path.join(context.root_prefix, "Scripts" if on_win else "bin")
mmap_MAP_PRIVATE = 0 if on_win else mmap.MAP_PRIVATE
mmap_PROT_READ = 0 if on_win else mmap.PROT_READ
mmap_PROT_WRITE = 0 if on_win else mmap.PROT_WRITE

DEFAULT_SUBDIRS = set(KNOWN_SUBDIRS)

RUN_EXPORTS_TYPES = {
    "weak",
    "strong",
    "noarch",
    "weak_constrains",
    "strong_constrains",
}

PY_TMPL = r"""
# -*- coding: utf-8 -*-
import re
import sys

from %(module)s import %(import_name)s

if __name__ == '__main__':
    sys.argv[0] = re.sub(r'(-script\.pyw?|\.exe)?$', '', sys.argv[0])
    sys.exit(%(func)s())
"""

# filenames accepted as recipe meta files
VALID_METAS = ("meta.yaml", "meta.yml", "conda.yaml", "conda.yml")


@lru_cache(maxsize=None)
def stat_file(path):
    return os.stat(path)


def directory_size_slow(path):
    total_size = 0
    seen = set()

    for root, _, files in walk(path):
        for f in files:
            try:
                stat = stat_file(os.path.join(root, f))
            except OSError:
                continue

            if stat.st_ino in seen:
                continue

            seen.add(stat.st_ino)

            total_size += stat.st_size
    return total_size


def directory_size(path):
    try:
        if on_win:
            command = 'dir /s "{}"'  # Windows path can have spaces
            out = subprocess.check_output(command.format(path), shell=True)
        else:
            command = "du -s {}"
            out = subprocess.check_output(
                command.format(path).split(), stderr=subprocess.PIPE
            )

        if hasattr(out, "decode"):
            try:
                out = out.decode(errors="ignore")
            # This isn't important anyway so give up. Don't try search on bytes.
            except (UnicodeDecodeError, IndexError):
                if on_win:
                    return 0
                else:
                    pass
        if on_win:
            # Windows can give long output, we need only 2nd to last line
            out = out.strip().rsplit("\r\n", 2)[-2]
            pattern = r"\s([\d\W]+).+"  # Language and punctuation neutral
            out = re.search(pattern, out.strip()).group(1).strip()
            out = out.replace(",", "").replace(".", "").replace(" ", "")
        else:
            out = out.split()[0]
    except subprocess.CalledProcessError:
        out = directory_size_slow(path)

    try:
        return int(out)  # size in bytes
    except ValueError:
        return 0


class DummyPsutilProcess:
    def children(self, *args, **kwargs):
        return []


def _setup_rewrite_pipe(env):
    """Rewrite values of env variables back to $ENV in stdout

    Takes output on the pipe and finds any env value
    and rewrites it as the env key

    Useful for replacing "~/conda/conda-bld/pkg_<date>/_h_place..." with "$PREFIX"

    Returns an FD to be passed to Popen(stdout=...)
    """
    # replacements is the env dict reversed,
    # ordered by the length of the value so that longer replacements
    # always occur first in case of common prefixes
    replacements = OrderedDict()
    for k, v in sorted(env.items(), key=lambda kv: len(kv[1]), reverse=True):
        replacements[v] = k

    r_fd, w_fd = os.pipe()
    r = os.fdopen(r_fd, "rt")
    if on_win:
        replacement_t = "%{}%"
    else:
        replacement_t = "${}"

    def rewriter():
        while True:
            try:
                line = r.readline()
                if not line:
                    # reading done
                    r.close()
                    os.close(w_fd)
                    return
                for s, key in replacements.items():
                    line = line.replace(s, replacement_t.format(key))
                sys.stdout.write(line)
            except UnicodeDecodeError:
                try:
                    txt = os.read(r, 10000)
                    sys.stdout.write(txt or "")
                except TypeError:
                    pass

    t = Thread(target=rewriter)
    t.daemon = True
    t.start()

    return w_fd


class PopenWrapper:
    # Small wrapper around subprocess.Popen to allow memory usage monitoring
    # copied from ProtoCI, https://github.com/ContinuumIO/ProtoCI/blob/59159bc2c9f991fbfa5e398b6bb066d7417583ec/protoci/build2.py#L20  # NOQA

    def __init__(self, *args, **kwargs):
        self.elapsed = None
        self.rss = 0
        self.vms = 0
        self.returncode = None
        self.disk = 0
        self.processes = 1

        self.out, self.err = self._execute(*args, **kwargs)

    def _execute(self, *args, **kwargs):
        try:
            import psutil

            psutil_exceptions = (
                psutil.NoSuchProcess,
                psutil.AccessDenied,
                psutil.NoSuchProcess,
            )
        except ImportError as e:
            psutil = None
            psutil_exceptions = (OSError, ValueError)
            log = get_logger(__name__)
            log.warning(f"psutil import failed.  Error was {e}")
            log.warning(
                "only disk usage and time statistics will be available.  Install psutil to "
                "get CPU time and memory usage statistics."
            )

        # The polling interval (in seconds)
        time_int = kwargs.pop("time_int", 2)

        disk_usage_dir = kwargs.get("cwd", sys.prefix)

        # Create a process of this (the parent) process
        parent = psutil.Process(os.getpid()) if psutil else DummyPsutilProcess()

        cpu_usage = defaultdict(dict)

        # Using the convenience Popen class provided by psutil
        start_time = time.time()
        _popen = (
            psutil.Popen(*args, **kwargs)
            if psutil
            else subprocess.Popen(*args, **kwargs)
        )
        try:
            while self.returncode is None:
                # We need to get all of the children of our process since our
                # process spawns other processes.  Collect all of the child
                # processes

                rss = 0
                vms = 0
                processes = 0
                # We use the parent process to get mem usage of all spawned processes
                for child in parent.children(recursive=True):
                    child_cpu_usage = cpu_usage.get(child.pid, {})
                    try:
                        mem = child.memory_info()
                        rss += mem.rss
                        vms += mem.rss
                        # listing child times are only available on linux, so we don't use them.
                        #    we are instead looping over children and getting each individually.
                        #    https://psutil.readthedocs.io/en/latest/#psutil.Process.cpu_times
                        cpu_stats = child.cpu_times()
                        child_cpu_usage["sys"] = cpu_stats.system
                        child_cpu_usage["user"] = cpu_stats.user
                        cpu_usage[child.pid] = child_cpu_usage
                    except psutil_exceptions:
                        # process already died.  Just ignore it.
                        continue
                    processes += 1

                # Sum the memory usage of all the children together (2D columnwise sum)
                self.rss = max(rss, self.rss)
                self.vms = max(vms, self.vms)
                self.cpu_sys = sum(child["sys"] for child in cpu_usage.values())
                self.cpu_user = sum(child["user"] for child in cpu_usage.values())
                self.processes = max(processes, self.processes)

                # Get disk usage
                self.disk = max(directory_size(disk_usage_dir), self.disk)

                time.sleep(time_int)
                self.elapsed = time.time() - start_time
                self.returncode = _popen.poll()

        except KeyboardInterrupt:
            _popen.kill()
            raise

        self.disk = max(directory_size(disk_usage_dir), self.disk)
        self.elapsed = time.time() - start_time
        return _popen.stdout, _popen.stderr

    def __repr__(self):
        return str(
            {
                "elapsed": self.elapsed,
                "rss": self.rss,
                "vms": self.vms,
                "disk": self.disk,
                "processes": self.processes,
                "cpu_user": self.cpu_user,
                "cpu_sys": self.cpu_sys,
                "returncode": self.returncode,
            }
        )


def _func_defaulting_env_to_os_environ(func, *popenargs, **kwargs):
    if "env" not in kwargs:
        kwargs = kwargs.copy()
        env_copy = os.environ.copy()
        kwargs.update({"env": env_copy})
    kwargs["env"] = {str(key): str(value) for key, value in kwargs["env"].items()}
    _args = []
    if "stdin" not in kwargs:
        kwargs["stdin"] = subprocess.PIPE
    for arg in popenargs:
        # arguments to subprocess need to be bytestrings
        if sys.version_info.major < 3 and hasattr(arg, "encode"):
            arg = arg.encode(codec)
        elif sys.version_info.major >= 3 and hasattr(arg, "decode"):
            arg = arg.decode(codec)
        _args.append(str(arg))

    stats = kwargs.get("stats")
    if "stats" in kwargs:
        del kwargs["stats"]

    rewrite_stdout_env = kwargs.pop("rewrite_stdout_env", None)
    if rewrite_stdout_env:
        kwargs["stdout"] = _setup_rewrite_pipe(rewrite_stdout_env)

    out = None
    if stats is not None:
        proc = PopenWrapper(_args, **kwargs)
        if func == "output":
            out = proc.out.read()

        if proc.returncode != 0:
            raise subprocess.CalledProcessError(proc.returncode, _args)

        stats.update(
            {
                "elapsed": proc.elapsed,
                "disk": proc.disk,
                "processes": proc.processes,
                "cpu_user": proc.cpu_user,
                "cpu_sys": proc.cpu_sys,
                "rss": proc.rss,
                "vms": proc.vms,
            }
        )
    else:
        if func == "call":
            subprocess.check_call(_args, **kwargs)
        else:
            if "stdout" in kwargs:
                del kwargs["stdout"]
            out = subprocess.check_output(_args, **kwargs)
    return out


def check_call_env(popenargs, **kwargs):
    return _func_defaulting_env_to_os_environ("call", *popenargs, **kwargs)


def check_output_env(popenargs, **kwargs):
    return _func_defaulting_env_to_os_environ(
        "output", stdout=subprocess.PIPE, *popenargs, **kwargs
    ).rstrip()


def bytes2human(n):
    # http://code.activestate.com/recipes/578019
    # >>> bytes2human(10000)
    # '9.8K'
    # >>> bytes2human(100001221)
    # '95.4M'
    symbols = ("K", "M", "G", "T", "P", "E", "Z", "Y")
    prefix = {}
    for i, s in enumerate(symbols):
        prefix[s] = 1 << (i + 1) * 10
    for s in reversed(symbols):
        if n >= prefix[s]:
            value = float(n) / prefix[s]
            return f"{value:.1f}{s}"
    return f"{n}B"


def seconds2human(s):
    m, s = divmod(s, 60)
    h, m = divmod(int(m), 60)
    return f"{h:d}:{m:02d}:{s:04.1f}"


def get_recipe_abspath(recipe):
    """resolve recipe dir as absolute path.  If recipe is a tarball rather than a folder,
    extract it and return the extracted directory.

    Returns the absolute path, and a boolean flag that is true if a tarball has been extracted
    and needs cleanup.
    """
    if isfile(recipe):
        if recipe.lower().endswith(decompressible_exts) or recipe.lower().endswith(
            CONDA_PACKAGE_EXTENSIONS
        ):
            recipe_dir = tempfile.mkdtemp()
            if recipe.lower().endswith(CONDA_PACKAGE_EXTENSIONS):
                import conda_package_handling.api

                conda_package_handling.api.extract(recipe, recipe_dir)
            else:
                tar_xf(recipe, recipe_dir)
            # At some stage the old build system started to tar up recipes.
            recipe_tarfile = os.path.join(recipe_dir, "info", "recipe.tar")
            if isfile(recipe_tarfile):
                tar_xf(recipe_tarfile, os.path.join(recipe_dir, "info"))
            need_cleanup = True
        else:
            print(f"Ignoring non-recipe: {recipe}")
            return (None, None)
    else:
        recipe_dir = abspath(os.path.join(os.getcwd(), recipe))
        need_cleanup = False
    if not os.path.exists(recipe_dir):
        raise ValueError(f"Package or recipe at path {recipe_dir} does not exist")
    return recipe_dir, need_cleanup


@contextlib.contextmanager
def try_acquire_locks(locks, timeout):
    """Try to acquire all locks.

    If any lock can't be immediately acquired, free all locks.
    If the timeout is reached withou acquiring all locks, free all locks and raise.

    http://stackoverflow.com/questions/9814008/multiple-mutex-locking-strategies-and-why-libraries-dont-use-address-comparison
    """
    t = time.time()
    while time.time() - t < timeout:
        # Continuously try to acquire all locks.
        # By passing a short timeout to each individual lock, we give other
        # processes that might be trying to acquire the same locks (and may
        # already hold some of them) a chance to the remaining locks - and
        # hopefully subsequently release them.
        try:
            for lock in locks:
                lock.acquire(timeout=0.1)
        except filelock.Timeout:
            # If we failed to acquire a lock, it is important to release all
            # locks we may have already acquired, to avoid wedging multiple
            # processes that try to acquire the same set of locks.
            # That is, we want to avoid a situation where processes 1 and 2 try
            # to acquire locks A and B, and proc 1 holds lock A while proc 2
            # holds lock B.
            for lock in locks:
                lock.release()
        else:
            break
    else:
        # If we reach this point, we weren't able to acquire all locks within
        # the specified timeout. We shouldn't be holding any locks anymore at
        # this point, so we just raise an exception.
        raise BuildLockError("Failed to acquire all locks")

    try:
        yield
    finally:
        for lock in locks:
            lock.release()


# with each of these, we are copying less metadata.  This seems to be necessary
#   to cope with some shared filesystems with some virtual machine setups.
#  See https://github.com/conda/conda-build/issues/1426
def _copy_with_shell_fallback(src, dst):
    is_copied = False
    for func in (shutil.copy2, shutil.copy, shutil.copyfile):
        try:
            func(src, dst)
            is_copied = True
            break
        except (OSError, PermissionError):
            continue
    if not is_copied:
        try:
            subprocess.check_call(
                f"cp -a {src} {dst}",
                shell=True,
                stderr=subprocess.PIPE,
                stdout=subprocess.PIPE,
            )
        except subprocess.CalledProcessError as e:
            if not os.path.isfile(dst):
                raise OSError(f"Failed to copy {src} to {dst}.  Error was: {e}")


def get_prefix_replacement_paths(src, dst):
    ssplit = src.split(os.path.sep)
    dsplit = dst.split(os.path.sep)
    while ssplit and ssplit[-1] == dsplit[-1]:
        del ssplit[-1]
        del dsplit[-1]
    return os.path.join(*ssplit), os.path.join(*dsplit)


def copy_into(
    src, dst, timeout=900, symlinks=False, lock=None, locking=True, clobber=False
):
    """Copy all the files and directories in src to the directory dst"""
    log = get_logger(__name__)
    if symlinks and islink(src):
        try:
            os.makedirs(os.path.dirname(dst))
        except OSError:
            pass
        if os.path.lexists(dst):
            os.remove(dst)
        src_base, dst_base = get_prefix_replacement_paths(src, dst)
        src_target = os.readlink(src)
        src_replaced = src_target.replace(src_base, dst_base)
        os.symlink(src_replaced, dst)
        try:
            st = os.lstat(src)
            mode = stat.S_IMODE(st.st_mode)
            os.lchmod(dst, mode)
        except:
            pass  # lchmod not available
    elif isdir(src):
        merge_tree(
            src,
            dst,
            symlinks,
            timeout=timeout,
            lock=lock,
            locking=locking,
            clobber=clobber,
        )

    else:
        if isdir(dst):
            dst_fn = os.path.join(dst, os.path.basename(src))
        else:
            dst_fn = dst

        if os.path.isabs(src):
            src_folder = os.path.dirname(src)
        else:
            if os.path.sep in dst_fn:
                src_folder = os.path.dirname(dst_fn)
                if not os.path.isdir(src_folder):
                    os.makedirs(src_folder)
            else:
                src_folder = os.getcwd()

        if os.path.islink(src) and not os.path.exists(os.path.realpath(src)):
            log.warning("path %s is a broken symlink - ignoring copy", src)
            return

        if not lock and locking:
            lock = get_lock(src_folder, timeout=timeout)
        locks = [lock] if locking else []
        with try_acquire_locks(locks, timeout):
            # if intermediate folders not not exist create them
            dst_folder = os.path.dirname(dst)
            if dst_folder and not os.path.exists(dst_folder):
                try:
                    os.makedirs(dst_folder)
                except OSError:
                    pass
            try:
                _copy_with_shell_fallback(src, dst_fn)
            except shutil.Error:
                log.debug(
                    "skipping %s - already exists in %s", os.path.basename(src), dst
                )


def move_with_fallback(src, dst):
    try:
        shutil.move(src, dst)
    except PermissionError:
        try:
            copy_into(src, dst)
            os.unlink(src)
        except PermissionError:
            log = get_logger(__name__)
            log.debug(
                f"Failed to copy/remove path from {src} to {dst} due to permission error"
            )


# http://stackoverflow.com/a/22331852/1170370
def copytree(src, dst, symlinks=False, ignore=None, dry_run=False):
    if not os.path.exists(dst):
        os.makedirs(dst)
        shutil.copystat(src, dst)
    lst = os.listdir(src)
    if ignore:
        excl = ignore(src, lst)
        lst = [x for x in lst if x not in excl]

    # do not copy lock files
    if ".conda_lock" in lst:
        lst.remove(".conda_lock")

    dst_lst = [os.path.join(dst, item) for item in lst]

    if not dry_run:
        for idx, item in enumerate(lst):
            s = os.path.join(src, item)
            d = dst_lst[idx]
            if symlinks and os.path.islink(s):
                if os.path.lexists(d):
                    os.remove(d)
                os.symlink(os.readlink(s), d)
                try:
                    st = os.lstat(s)
                    mode = stat.S_IMODE(st.st_mode)
                    os.lchmod(d, mode)
                except:
                    pass  # lchmod not available
            elif os.path.isdir(s):
                copytree(s, d, symlinks, ignore)
            else:
                _copy_with_shell_fallback(s, d)

    return dst_lst


def merge_tree(
    src, dst, symlinks=False, timeout=900, lock=None, locking=True, clobber=False
):
    """
    Merge src into dst recursively by copying all files from src into dst.
    Return a list of all files copied.

    Like copytree(src, dst), but raises an error if merging the two trees
    would overwrite any files.
    """
    dst = os.path.normpath(os.path.normcase(dst))
    src = os.path.normpath(os.path.normcase(src))
    assert not dst.startswith(src), (
        "Can't merge/copy source into subdirectory of itself.  "
        "Please create separate spaces for these things.\n"
        f"  src: {src}\n"
        f"  dst: {dst}"
    )

    new_files = copytree(src, dst, symlinks=symlinks, dry_run=True)
    existing = [f for f in new_files if isfile(f)]

    if existing and not clobber:
        raise OSError(f"Can't merge {src} into {dst}: file exists: {existing[0]}")

    locks = []
    if locking:
        if not lock:
            lock = get_lock(src, timeout=timeout)
        locks = [lock]
    with try_acquire_locks(locks, timeout):
        copytree(src, dst, symlinks=symlinks)


# purpose here is that we want *one* lock per location on disk.  It can be locked or unlocked
#    at any time, but the lock within this process should all be tied to the same tracking
#    mechanism.
_lock_folders = (
    os.path.join(context.root_prefix, "locks"),
    os.path.expanduser(os.path.join("~", ".conda_build_locks")),
)


def get_lock(folder, timeout=900):
    fl = None
    try:
        location = os.path.abspath(os.path.normpath(folder))
    except OSError:
        location = folder
    b_location = location
    if hasattr(b_location, "encode"):
        b_location = b_location.encode()

    # Hash the entire filename to avoid collisions.
    lock_filename = hashlib.sha256(b_location).hexdigest()

    if hasattr(lock_filename, "decode"):
        lock_filename = lock_filename.decode()
    for locks_dir in _lock_folders:
        try:
            if not os.path.isdir(locks_dir):
                os.makedirs(locks_dir)
            lock_file = os.path.join(locks_dir, lock_filename)
            with open(lock_file, "w") as f:
                f.write("")
            fl = filelock.FileLock(lock_file, timeout)
            break
        except OSError:
            continue
    else:
        raise RuntimeError(
            "Could not write locks folder to either system location ({})"
            "or user location ({}).  Aborting.".format(*_lock_folders)
        )
    return fl


def get_conda_operation_locks(locking=True, bldpkgs_dirs=None, timeout=900):
    locks = []
    bldpkgs_dirs = ensure_list(bldpkgs_dirs)
    # locks enabled by default
    if locking:
        for folder in (*context.pkgs_dirs[:1], *bldpkgs_dirs):
            if not os.path.isdir(folder):
                os.makedirs(folder)
            lock = get_lock(folder, timeout=timeout)
            locks.append(lock)
        # lock used to generally indicate a conda operation occurring
        locks.append(get_lock("conda-operation", timeout=timeout))
    return locks


# This is the lowest common denominator of the formats supported by our libarchive/python-libarchive-c
# packages across all platforms
decompressible_exts = (
    ".7z",
    ".tar",
    ".tar.bz2",
    ".tar.gz",
    ".tar.lzma",
    ".tar.xz",
    ".tar.z",
    ".tar.zst",
    ".tgz",
    ".whl",
    ".zip",
    ".rpm",
    ".deb",
)


def _tar_xf_fallback(tarball, dir_path, mode="r:*"):
    from .os_utils.external import find_executable

    if tarball.lower().endswith(".tar.z"):
        uncompress = find_executable("uncompress")
        if not uncompress:
            uncompress = find_executable("gunzip")
        if not uncompress:
            sys.exit(
                """\
uncompress (or gunzip) is required to unarchive .z source files.
"""
            )
        check_call_env([uncompress, "-f", tarball])
        tarball = tarball[:-2]

    t = tarfile.open(tarball, mode)
    members = t.getmembers()
    for i, member in enumerate(members, 0):
        if os.path.isabs(member.name):
            member.name = os.path.relpath(member.name, "/")
        cwd = os.path.realpath(os.getcwd())
        if not os.path.realpath(member.name).startswith(cwd):
            member.name = member.name.replace("../", "")
        if not os.path.realpath(member.name).startswith(cwd):
            sys.exit("tarball contains unsafe path: " + member.name + " cwd is: " + cwd)
        members[i] = member

    t.extractall(path=dir_path)
    t.close()


def tar_xf_file(tarball, entries):
    entries = ensure_list(entries)
    if not os.path.isabs(tarball):
        tarball = os.path.join(os.getcwd(), tarball)
    result = None
    n_found = 0
    with libarchive.file_reader(tarball) as archive:
        for entry in archive:
            if entry.name in entries:
                n_found += 1
                for block in entry.get_blocks():
                    if result is None:
                        result = bytes(block)
                    else:
                        result += block
                break
    if n_found != len(entries):
        raise KeyError()
    return result


def tar_xf_getnames(tarball):
    if not os.path.isabs(tarball):
        tarball = os.path.join(os.getcwd(), tarball)
    result = []
    with libarchive.file_reader(tarball) as archive:
        for entry in archive:
            result.append(entry.name)
    return result


def tar_xf(tarball, dir_path):
    flags = (
        libarchive.extract.EXTRACT_TIME
        | libarchive.extract.EXTRACT_PERM
        | libarchive.extract.EXTRACT_SECURE_NODOTDOT
        | libarchive.extract.EXTRACT_SECURE_SYMLINKS
        | libarchive.extract.EXTRACT_SECURE_NOABSOLUTEPATHS
    )
    if not os.path.isabs(tarball):
        tarball = os.path.join(os.getcwd(), tarball)
    try:
        with tmp_chdir(os.path.realpath(dir_path)):
            libarchive.extract_file(tarball, flags)
    except libarchive.exception.ArchiveError:
        # try again, maybe we are on Windows and the archive contains symlinks
        # https://github.com/conda/conda-build/issues/3351
        # https://github.com/libarchive/libarchive/pull/1030
        if tarball.lower().endswith(
            (".tar", ".tar.gz", ".tgz", ".tar.bz2", ".tar.z", ".tar.xz")
        ):
            _tar_xf_fallback(tarball, dir_path)
        else:
            raise


def file_info(path):
    return {
        "size": getsize(path),
        "md5": compute_sum(path, "md5"),
        "sha256": compute_sum(path, "sha256"),
        "mtime": getmtime(path),
    }


def comma_join(items: Iterable[str], conjunction: str = "and") -> str:
    """
    Like ', '.join(items) but with and

    Examples:

    >>> comma_join(['a'])
    'a'
    >>> comma_join(['a', 'b'])
    'a and b'
    >>> comma_join(['a', 'b', 'c'])
    'a, b, and c'
    """
    items = tuple(items)
    if len(items) <= 2:
        return f"{items[0]} {conjunction} {items[1]}"
    return f"{', '.join(items[:-1])}, {conjunction} {items[-1]}"


def safe_print_unicode(*args, **kwargs):
    """
    prints unicode strings to stdout using configurable `errors` handler for
    encoding errors

    :param args: unicode strings to print to stdout
    :param sep: separator (defaults to ' ')
    :param end: ending character (defaults to '\n')
    :param errors: error handler for encoding errors (defaults to 'replace')
    """
    sep = kwargs.pop("sep", " ")
    end = kwargs.pop("end", "\n")
    errors = kwargs.pop("errors", "replace")
    func = sys.stdout.buffer.write
    line = sep.join(args) + end
    encoding = sys.stdout.encoding or "utf8"
    func(line.encode(encoding, errors))


def rec_glob(path, patterns, ignores=None):
    """
    Recursively searches path for filename patterns.

    :param path: path within to search for files
    :param patterns: list of filename patterns to search for
    :param ignore: list of directory patterns to ignore in search
    :return: list of paths in path satisfying patterns/ignore
    """
    patterns = ensure_list(patterns)
    ignores = ensure_list(ignores)

    for path, dirs, files in walk(path):
        # remove directories to ignore
        for ignore in ignores:
            for d in fnmatch.filter(dirs, ignore):
                dirs.remove(d)

        # return filepaths that match a pattern
        for pattern in patterns:
            for f in fnmatch.filter(files, pattern):
                yield os.path.join(path, f)


def convert_unix_path_to_win(path):
    from .os_utils.external import find_executable

    if find_executable("cygpath"):
        cmd = f"cygpath -w {path}"
        path = subprocess.getoutput(cmd)

    else:
        path = unix_path_to_win(path)
    return path


def convert_win_path_to_unix(path):
    from .os_utils.external import find_executable

    if find_executable("cygpath"):
        cmd = f"cygpath -u {path}"
        path = subprocess.getoutput(cmd)

    else:
        path = win_path_to_unix(path)
    return path


# Used for translating local paths into url (file://) paths
#   http://stackoverflow.com/a/14298190/1170370
def path2url(path):
    return urlparse.urljoin("file:", urllib.pathname2url(path))


def get_stdlib_dir(prefix, py_ver):
    if on_win:
        lib_dir = os.path.join(prefix, "Lib")
    else:
        lib_dir = os.path.join(prefix, "lib")
        python_folder = glob(os.path.join(lib_dir, "python?.*"), recursive=True)
        python_folder = sorted(filterfalse(islink, python_folder))
        if python_folder:
            lib_dir = os.path.join(lib_dir, python_folder[0])
        else:
            lib_dir = os.path.join(lib_dir, f"python{py_ver}")
    return lib_dir


def get_site_packages(prefix, py_ver):
    return os.path.join(get_stdlib_dir(prefix, py_ver), "site-packages")


def get_build_folders(croot: str | os.PathLike | Path) -> list[str]:
    # remember, glob is not a regex.
    return glob(os.path.join(croot, "*" + "[0-9]" * 10 + "*"), recursive=True)


def prepend_bin_path(env, prefix, prepend_prefix=False):
    env["PATH"] = join(prefix, "bin") + os.pathsep + env["PATH"]
    if on_win:
        env["PATH"] = (
            join(prefix, "Library", "mingw-w64", "bin")
            + os.pathsep
            + join(prefix, "Library", "usr", "bin")
            + os.pathsep
            + join(prefix, "Library", "bin")
            + os.pathsep
            + join(prefix, "Scripts")
            + os.pathsep
            + env["PATH"]
        )
        prepend_prefix = True  # windows has Python in the prefix.  Use it.
    if prepend_prefix:
        env["PATH"] = prefix + os.pathsep + env["PATH"]
    return env


# not currently used.  Leaving in because it may be useful for when we do things
#   like load setup.py data, and we need the modules from some prefix other than
#   the root prefix, which is what conda-build runs from.
@contextlib.contextmanager
def sys_path_prepended(prefix):
    path_backup = sys.path[:]
    if on_win:
        sys.path.insert(1, os.path.join(prefix, "lib", "site-packages"))
    else:
        lib_dir = os.path.join(prefix, "lib")
        python_dir = glob(os.path.join(lib_dir, r"python[0-9\.]*"), recursive=True)
        if python_dir:
            python_dir = python_dir[0]
            sys.path.insert(1, os.path.join(python_dir, "site-packages"))
    try:
        yield
    finally:
        sys.path = path_backup


@contextlib.contextmanager
def path_prepended(prefix, prepend_prefix=True):
    # FIXME: Unclear why prepend_prefix=True for all platforms.
    old_path = os.environ["PATH"]
    os.environ["PATH"] = prepend_bin_path(os.environ.copy(), prefix, prepend_prefix)[
        "PATH"
    ]
    try:
        yield
    finally:
        os.environ["PATH"] = old_path


bin_dirname = "Scripts" if on_win else "bin"

entry_pat = re.compile(r"\s*([\w\-\.]+)\s*=\s*([\w.]+):([\w.]+)\s*$")


def iter_entry_points(items):
    for item in items:
        m = entry_pat.match(item)
        if m is None:
            sys.exit(f"Error cound not match entry point: {item!r}")
        yield m.groups()


def create_entry_point(path, module, func, config):
    import_name = func.split(".")[0]
    pyscript = PY_TMPL % {"module": module, "func": func, "import_name": import_name}
    if on_win:
        with open(path + "-script.py", "w") as fo:
            if os.path.isfile(os.path.join(config.host_prefix, "python_d.exe")):
                fo.write("#!python_d\n")
            fo.write(pyscript)
            copy_into(
                join(dirname(__file__), f"cli-{str(config.host_arch)}.exe"),
                path + ".exe",
                config.timeout,
            )
    else:
        if os.path.islink(path):
            os.remove(path)
        with open(path, "w") as fo:
            if not config.noarch:
                fo.write(f"#!{config.host_python}\n")
            fo.write(pyscript)
        os.chmod(path, 0o775)


def create_entry_points(items, config):
    if not items:
        return
    bin_dir = join(config.host_prefix, bin_dirname)
    if not isdir(bin_dir):
        os.mkdir(bin_dir)
    for cmd, module, func in iter_entry_points(items):
        create_entry_point(join(bin_dir, cmd), module, func, config)


# Return all files in dir, and all its subdirectories, ending in pattern
def get_ext_files(start_path, pattern):
    for root, _, files in walk(start_path):
        for f in files:
            if f.endswith(pattern):
                yield os.path.join(root, f)


_posix_exes_cache = {}


def convert_path_for_cygwin_or_msys2(exe, path):
    "If exe is a Cygwin or MSYS2 executable then filters it through `cygpath -u`"
    if not on_win:
        return path
    if exe not in _posix_exes_cache:
        with open(exe, "rb") as exe_file:
            exe_binary = exe_file.read()
            msys2_cygwin = re.findall(b"(cygwin1.dll|msys-2.0.dll)", exe_binary)
            _posix_exes_cache[exe] = True if msys2_cygwin else False
    if _posix_exes_cache[exe]:
        try:
            path = (
                check_output_env(["cygpath", "-u", path])
                .splitlines()[0]
                .decode(getpreferredencoding())
            )
        except OSError:
            log = get_logger(__name__)
            log.debug(
                "cygpath executable not found.  Passing native path.  This is OK for msys2."
            )
    return path


def get_skip_message(m: MetaData) -> str:
    return (
        f"Skipped: {m.name()} from {m.path} defines build/skip for this configuration "
        f"({({k: m.config.variant[k] for k in m.get_used_vars()})})."
    )


def package_has_file(package_path, file_path, refresh_mode="modified"):
    # This version does nothing to the package cache.
    with TemporaryDirectory() as td:
        if file_path.startswith("info"):
            conda_package_handling.api.extract(
                package_path, dest_dir=td, components="info"
            )
        else:
            conda_package_handling.api.extract(
                package_path, dest_dir=td, components=file_path
            )
        resolved_file_path = os.path.join(td, file_path)
        if os.path.exists(resolved_file_path):
            # TODO :: Remove this text-mode load. Files are binary.
            try:
                with open(resolved_file_path) as f:
                    content = f.read()
            except UnicodeDecodeError:
                with open(resolved_file_path, "rb") as f:
                    content = f.read()
        else:
            content = False
        return content


def ensure_list(arg: T | Iterable[T] | None, include_dict: bool = True) -> list[T]:
    """
    Ensure the object is a list. If not return it in a list.

    :param arg: Object to ensure is a list
    :type arg: any
    :param include_dict: Whether to treat `dict` as a `list`
    :type include_dict: bool, optional
    :return: `arg` as a `list`
    :rtype: list
    """
    if arg is None:
        return []
    elif islist(arg, include_dict=include_dict):
        return list(arg)
    else:
        return [arg]


def islist(
    arg: T | Iterable[T],
    uniform: bool = False,
    include_dict: bool = True,
) -> bool:
    """
    Check whether `arg` is a `list`. Optionally determine whether the list elements
    are all uniform.

    When checking for generic uniformity (`uniform=True`) we check to see if all
    elements are of the first element's type (`type(arg[0]) == type(arg[1])`). For
    any other kinds of uniformity checks are desired provide a uniformity function:

    .. code-block:: pycon
        # uniformity function checking if elements are str and not empty
        >>> truthy_str = lambda e: isinstance(e, str) and e
        >>> islist(["foo", "bar"], uniform=truthy_str)
        True
        >>> islist(["", "bar"], uniform=truthy_str)
        False
        >>> islist([0, "bar"], uniform=truthy_str)
        False

    .. note::
        Testing for uniformity will consume generators.

    :param arg: Object to ensure is a `list`
    :type arg: any
    :param uniform: Whether to check for uniform or uniformity function
    :type uniform: bool, function, optional
    :param include_dict: Whether to treat `dict` as a `list`
    :type include_dict: bool, optional
    :return: Whether `arg` is a `list`
    :rtype: bool
    """
    if isinstance(arg, str) or not isinstance(arg, Iterable):
        # str and non-iterables are not lists
        return False
    elif not include_dict and isinstance(arg, dict):
        # do not treat dict as a list
        return False
    elif not uniform:
        # short circuit for non-uniformity
        return True

    # NOTE: not checking for Falsy arg since arg may be a generator
    # WARNING: if uniform != False and arg is a generator then arg will be consumed

    if uniform is True:
        arg = iter(arg)
        try:
            etype = type(next(arg))
        except StopIteration:
            # StopIteration: list is empty, an empty list is still uniform
            return True
        # check for explicit type match, do not allow the ambiguity of isinstance
        uniform = lambda e: type(e) == etype  # noqa: E721

    try:
        return all(uniform(e) for e in arg)
    except (ValueError, TypeError):
        # ValueError, TypeError: uniform function failed
        return False


@contextlib.contextmanager
def tmp_chdir(dest):
    curdir = os.getcwd()
    try:
        os.chdir(dest)
        yield
    finally:
        os.chdir(curdir)


def expand_globs(
    path_list: str | os.PathLike | Path | Iterable[str | os.PathLike | Path],
    root_dir: str | os.PathLike | Path,
) -> list[str]:
    files = []
    for path in ensure_list(path_list):
        path = str(path)
        if not os.path.isabs(path):
            path = os.path.join(root_dir, path)
        if os.path.isfile(path):
            files.append(path)
        elif os.path.islink(path):
            files.append(path)
        elif os.path.isdir(path):
            for root, dirnames, fs in walk(path):
                files.extend(os.path.join(root, f) for f in fs)
                for folder in dirnames:
                    if os.path.islink(os.path.join(root, folder)):
                        files.append(os.path.join(root, folder))
        else:
            # File compared to the globs use / as separator independently of the os
            glob_files = glob(path, recursive=True)
            if not glob_files:
                log = get_logger(__name__)
                log.error(f"Glob {path} did not match in root_dir {root_dir}")
            # https://docs.python.org/3/library/glob.html#glob.glob states that
            # "whether or not the results are sorted depends on the file system".
            # Avoid this potential ambiguity by sorting. (see #4185)
            files.extend(sorted(glob_files))
    prefix_path_re = re.compile("^" + re.escape(f"{root_dir}{os.path.sep}"))
    return [prefix_path_re.sub("", f, 1) for f in files]


def find_recipe(path: str) -> str:
    """recurse through a folder, locating valid meta files (see VALID_METAS).  Raises error if more than one is found.

    Returns full path to meta file to be built.

    If we have a base level meta file and other supplemental (nested) ones, use the base level.
    """
    # if initial path is absolute then any path we find (via rec_glob)
    # will also be absolute
    if not os.path.isabs(path):
        path = os.path.normpath(os.path.join(os.getcwd(), path))

    if os.path.isfile(path):
        if os.path.basename(path) in VALID_METAS:
            return path
        raise OSError(
            "{} is not a valid meta file ({})".format(path, ", ".join(VALID_METAS))
        )

    results = list(rec_glob(path, VALID_METAS, ignores=(".AppleDouble",)))

    if not results:
        raise OSError(
            "No meta files ({}) found in {}".format(", ".join(VALID_METAS), path)
        )

    if len(results) == 1:
        return results[0]

    # got multiple valid meta files
    # check if a meta file is defined on the base level in which case use that one

    metas = [m for m in VALID_METAS if os.path.isfile(os.path.join(path, m))]
    if len(metas) == 1:
        get_logger(__name__).warning(
            "Multiple meta files found. "
            f"The {metas[0]} file in the base directory ({path}) "
            "will be used."
        )
        return os.path.join(path, metas[0])

    raise OSError(
        "More than one meta files ({}) found in {}".format(", ".join(VALID_METAS), path)
    )


class LoggingContext:
    default_loggers = [
        "conda",
        "binstar",
        "install",
        "conda.install",
        "fetch",
        "conda.instructions",
        "fetch.progress",
        "print",
        "progress",
        "dotupdate",
        "stdoutlog",
        "requests",
        "conda.core.package_cache_data",
        "conda.plan",
        "conda.gateways.disk.delete",
        "conda_build",
        "conda_build.index",
        "conda_build.noarch_python",
        "urllib3.connectionpool",
        "conda_index",
        "conda_index.index",
        "conda_index.index.convert_cache",
    ]

    def __init__(self, level=logging.WARN, handler=None, close=True, loggers=None):
        self.level = level
        self.old_levels = {}
        self.handler = handler
        self.close = close
        self.quiet = context.quiet
        if not loggers:
            self.loggers = LoggingContext.default_loggers
        else:
            self.loggers = loggers

    def __enter__(self):
        for logger in self.loggers:
            if isinstance(logger, str):
                log = logging.getLogger(logger)
            self.old_levels[logger] = log.level
            log.setLevel(
                self.level
                if ("install" not in logger or self.level < logging.INFO)
                else self.level + 10
            )
        if self.handler:
            self.logger.addHandler(self.handler)

        context.quiet = True

    def __exit__(self, et, ev, tb):
        for logger, level in self.old_levels.items():
            logging.getLogger(logger).setLevel(level)
        if self.handler:
            self.logger.removeHandler(self.handler)
        if self.handler and self.close:
            self.handler.close()

        context.quiet = self.quiet

        # implicit return of None => don't swallow exceptions


def get_installed_packages(path):
    """
    Scan all json files in 'path' and return a dictionary with their contents.
    Files are assumed to be in 'index.json' format.
    """
    installed = dict()
    for filename in glob(os.path.join(path, "conda-meta", "*.json"), recursive=True):
        with open(filename) as file:
            data = json.load(file)
            installed[data["name"]] = data
    return installed


# http://stackoverflow.com/a/10743550/1170370
@contextlib.contextmanager
def capture():
    import sys

    oldout, olderr = sys.stdout, sys.stderr
    try:
        out = [StringIO(), StringIO()]
        sys.stdout, sys.stderr = out
        yield out
    finally:
        sys.stdout, sys.stderr = oldout, olderr
        out[0] = out[0].getvalue()
        out[1] = out[1].getvalue()


# copied from conda; added in 4.3, not currently part of exported functionality
@contextlib.contextmanager
def env_var(name, value, callback=None):
    # NOTE: will likely want to call reset_context() when using this function, so pass
    #       it as callback
    name, value = str(name), str(value)
    saved_env_var = os.environ.get(name)
    try:
        os.environ[name] = value
        if callback:
            callback()
        yield
    finally:
        if saved_env_var:
            os.environ[name] = saved_env_var
        else:
            del os.environ[name]
        if callback:
            callback()


def trim_empty_keys(dict_):
    to_remove = set()
    negative_means_empty = ("final", "noarch_python", "zip_keys")
    for k, v in dict_.items():
        if hasattr(v, "keys"):
            trim_empty_keys(v)
        # empty lists and empty strings, and None are always empty.
        if v == list() or v == "" or v is None or v == dict():
            to_remove.add(k)
        # other things that evaluate as False may not be "empty" - things can be manually set to
        #     false, and we need to keep that setting.
        if not v and k in negative_means_empty:
            to_remove.add(k)
    if "zip_keys" in dict_ and not any(v for v in dict_["zip_keys"]):
        to_remove.add("zip_keys")
    for k in to_remove:
        del dict_[k]


def _increment(version, alpha_ver):
    try:
        if alpha_ver:
            suffix = "a"
        else:
            suffix = ".0a0"
        last_version = str(int(version) + 1) + suffix
    except ValueError:
        last_version = chr(ord(version) + 1)
    return last_version


def apply_pin_expressions(version, min_pin="x.x.x.x.x.x.x", max_pin="x"):
    pins = [len(p.split(".")) if p else None for p in (min_pin, max_pin)]
    parsed_version = VersionOrder(version).version[1:]
    nesting_position = None
    flat_list = []
    for idx, item in enumerate(parsed_version):
        if isinstance(item, list):
            nesting_position = idx
            flat_list.extend(item)
        else:
            flat_list.append(item)
    if max_pin and len(max_pin.split(".")) > len(flat_list):
        pins[1] = len(flat_list)
    versions = ["", ""]
    # first idx is lower bound pin; second is upper bound pin.
    #    pin value is number of places to pin.
    for p_idx, pin in enumerate(pins):
        if pin:
            # flat_list is the blown-out representation of the version
            for v_idx, v in enumerate(flat_list[:pin]):
                # upper bound pin
                if p_idx == 1 and v_idx == pin - 1:
                    # is the last place an alphabetic character?  OpenSSL, JPEG
                    alpha_ver = str(flat_list[min(pin, len(flat_list) - 1)]).isalpha()
                    v = _increment(v, alpha_ver)
                versions[p_idx] += str(v)
                if v_idx != nesting_position:
                    versions[p_idx] += "."
            if versions[p_idx][-1] == ".":
                versions[p_idx] = versions[p_idx][:-1]
    if versions[0]:
        if version.endswith(".*"):
            version_order = VersionOrder(version[:-2])
        elif version.endswith("*"):
            version_order = VersionOrder(version[:-1])
        else:
            version_order = VersionOrder(version)
        if version_order < VersionOrder(versions[0]):
            # If the minimum is greater than the version this is a pre-release build.
            # Use the version as the lower bound
            versions[0] = ">=" + version
        else:
            versions[0] = ">=" + versions[0]
    if versions[1]:
        versions[1] = "<" + versions[1]
    return ",".join([v for v in versions if v])


def filter_files(
    files_list,
    prefix,
    filter_patterns=(
        r"(.*[\\/])?\.git[\\/].*",
        r"(.*[\\/])?\.git$",
        r"(.*)?\.DS_Store.*",
        r".*\.la$",
        r"conda-meta.*",
        r".*\.conda_trash(?:_\d+)*$",
    ),
):
    """Remove things like the .git directory from the list of files to be copied"""
    for pattern in filter_patterns:
        r = re.compile(pattern)
        files_list = set(files_list) - set(filter(r.match, files_list))
    return [
        f
        for f in files_list
        if not os.path.isdir(os.path.join(prefix, f))
        or os.path.islink(os.path.join(prefix, f))
    ]


def filter_info_files(files_list, prefix):
    return filter_files(
        files_list,
        prefix,
        filter_patterns=(
            "info[\\\\/]index.json",
            "info[\\\\/]files",
            "info[\\\\/]paths.json",
            "info[\\\\/]about.json",
            "info[\\\\/]has_prefix",
            "info[\\\\/]hash_input_files",  # legacy, not used anymore
            "info[\\\\/]hash_input.json",
            "info[\\\\/]run_exports.yaml",  # legacy
            "info[\\\\/]run_exports.json",  # current
            "info[\\\\/]git",
            "info[\\\\/]recipe[\\\\/].*",
            "info[\\\\/]recipe_log.json",
            "info[\\\\/]recipe.tar",
            "info[\\\\/]test[\\\\/].*",
            "info[\\\\/]LICENSE.txt",  # legacy, some tests rely on this
            "info[\\\\/]licenses[\\\\/]*",
            "info[\\\\/]prelink_messages[\\\\/]*",
            "info[\\\\/]requires",
            "info[\\\\/]meta",
            "info[\\\\/]platform",
            "info[\\\\/]no_link",
            "info[\\\\/]link.json",
            "info[\\\\/]icon.png",
        ),
    )


def rm_rf(path: str | os.PathLike) -> None:
    from conda.core.prefix_data import delete_prefix_from_linked_data
    from conda.gateways.disk.delete import rm_rf

    rm_rf(str(path))
    delete_prefix_from_linked_data(str(path))


# https://stackoverflow.com/a/31459386/1170370
class LessThanFilter(logging.Filter):
    def __init__(self, exclusive_maximum, name=""):
        super().__init__(name)
        self.max_level = exclusive_maximum

    def filter(self, record):
        # non-zero return means we log this message
        return 1 if record.levelno < self.max_level else 0


class GreaterThanFilter(logging.Filter):
    def __init__(self, exclusive_minimum, name=""):
        super().__init__(name)
        self.min_level = exclusive_minimum

    def filter(self, record):
        # non-zero return means we log this message
        return 1 if record.levelno > self.min_level else 0


# unclutter logs - show messages only once
class DuplicateFilter(logging.Filter):
    def __init__(self):
        self.msgs = set()

    def filter(self, record):
        log = record.msg not in self.msgs
        self.msgs.add(record.msg)
        return int(log)


dedupe_filter = DuplicateFilter()
info_debug_stdout_filter = LessThanFilter(logging.WARNING)
warning_error_stderr_filter = GreaterThanFilter(logging.INFO)
level_formatter = logging.Formatter("%(levelname)s: %(message)s")

# set filelock's logger to only show warnings by default
logging.getLogger("filelock").setLevel(logging.WARN)

# quiet some of conda's less useful output
logging.getLogger("conda.core.linked_data").setLevel(logging.WARN)
logging.getLogger("conda.gateways.disk.delete").setLevel(logging.WARN)
logging.getLogger("conda.gateways.disk.test").setLevel(logging.WARN)


def reset_deduplicator():
    """Most of the time, we want the deduplication.  There are some cases (tests especially)
    where we want to be able to control the duplication."""
    global dedupe_filter
    dedupe_filter = DuplicateFilter()


def get_logger(name, level=logging.INFO, dedupe=True, add_stdout_stderr_handlers=True):
    config_file = None
    if log_config_file := context.conda_build.get("log_config_file"):
        config_file = abspath(expanduser(expandvars(log_config_file)))
    # by loading config file here, and then only adding handlers later, people
    # should be able to override conda-build's logger settings here.
    if config_file:
        with open(config_file) as f:
            config_dict = yaml.safe_load(f)
        logging.config.dictConfig(config_dict)
        level = config_dict.get("loggers", {}).get(name, {}).get("level", level)
    log = logging.getLogger(name)
    if log.level != level:
        log.setLevel(level)
    if dedupe:
        log.addFilter(dedupe_filter)

    # these are defaults.  They can be overridden by configuring a log config yaml file.
    top_pkg = name.split(".")[0]
    if top_pkg == "conda_build":
        # we don't want propagation in CLI, but we do want it in tests
        # this is a pytest limitation: https://github.com/pytest-dev/pytest/issues/3697
        logging.getLogger(top_pkg).propagate = "PYTEST_CURRENT_TEST" in os.environ
    if add_stdout_stderr_handlers and not log.handlers:
        stdout_handler = logging.StreamHandler(sys.stdout)
        stderr_handler = logging.StreamHandler(sys.stderr)
        stdout_handler.addFilter(info_debug_stdout_filter)
        stderr_handler.addFilter(warning_error_stderr_filter)
        stderr_handler.setFormatter(level_formatter)
        stdout_handler.setLevel(level)
        stderr_handler.setLevel(level)
        log.addHandler(stdout_handler)
        log.addHandler(stderr_handler)
    return log


def _equivalent(base_value, value, path):
    equivalent = value == base_value
    if isinstance(value, str) and isinstance(base_value, str):
        if not os.path.isabs(base_value):
            base_value = os.path.abspath(
                os.path.normpath(os.path.join(path, base_value))
            )
        if not os.path.isabs(value):
            value = os.path.abspath(os.path.normpath(os.path.join(path, value)))
        equivalent |= base_value == value
    return equivalent


def merge_or_update_dict(
    base, new, path="", merge=True, raise_on_clobber=False, add_missing_keys=True
):
    if base == new:
        return base
    log = get_logger(__name__)
    for key, value in new.items():
        if key in base or add_missing_keys:
            base_value = base.get(key, value)
            if hasattr(value, "keys"):
                base_value = merge_or_update_dict(
                    base_value, value, path, merge, raise_on_clobber=raise_on_clobber
                )
                base[key] = base_value
            elif hasattr(value, "__iter__") and not isinstance(value, str):
                if merge:
                    if base_value != value:
                        try:
                            base_value.extend(value)
                        except (TypeError, AttributeError):
                            base_value = value
                    try:
                        base[key] = list(base_value)
                    except TypeError:
                        base[key] = base_value
                else:
                    base[key] = value
            else:
                if (
                    base_value
                    and merge
                    and not _equivalent(base_value, value, path)
                    and raise_on_clobber
                ):
                    log.debug(
                        f"clobbering key {key} (original value {base_value}) with value {value}"
                    )
                if value is None and key in base:
                    del base[key]
                else:
                    base[key] = value
    return base


def merge_dicts_of_lists(
    dol1: Mapping[K, Iterable[V]],
    dol2: Mapping[K, Iterable[V]],
) -> dict[K, list[V]]:
    """
    From Alex Martelli: https://stackoverflow.com/a/1495821/3257826
    """
    keys = set(dol1).union(dol2)
    no = []
    return {k: dol1.get(k, no) + dol2.get(k, no) for k in keys}


def prefix_files(prefix: str | os.PathLike | Path) -> set[str]:
    """
    Returns a set of all files in prefix.
    """
    prefix = f"{os.path.abspath(prefix)}{os.path.sep}"
    prefix_files: set[str] = set()
    for root, directories, files in walk(prefix):
        # this is effectively os.path.relpath, just hacked to be faster
        relroot = root[len(prefix) :].lstrip(os.path.sep)
        # add all files
        prefix_files.update(join(relroot, file) for file in files)
        # add all symlink directories (they are "files")
        prefix_files.update(
            join(relroot, directory)
            for directory in directories
            if islink(join(root, directory))
        )
    return prefix_files


def mmap_mmap(
    fileno,
    length,
    tagname=None,
    flags=0,
    prot=mmap_PROT_READ | mmap_PROT_WRITE,
    access=None,
    offset=0,
):
    """
    Hides the differences between mmap.mmap on Windows and Unix.
    Windows has `tagname`.
    Unix does not, but makes up for it with `flags` and `prot`.
    On both, the default value for `access` is determined from how the file
    was opened so must not be passed in at all to get this default behaviour.
    """
    if on_win:
        if access:
            return mmap.mmap(
                fileno, length, tagname=tagname, access=access, offset=offset
            )
        else:
            return mmap.mmap(fileno, length, tagname=tagname)
    else:
        if access:
            return mmap.mmap(
                fileno, length, flags=flags, prot=prot, access=access, offset=offset
            )
        else:
            return mmap.mmap(fileno, length, flags=flags, prot=prot)


def remove_pycache_from_scripts(build_prefix):
    """Remove pip created pycache directory from bin or Scripts."""
    if on_win:
        scripts_path = os.path.join(build_prefix, "Scripts")
    else:
        scripts_path = os.path.join(build_prefix, "bin")

    if os.path.isdir(scripts_path):
        for entry in os.listdir(scripts_path):
            entry_path = os.path.join(scripts_path, entry)
            if os.path.isdir(entry_path) and entry.strip(os.sep) == "__pycache__":
                shutil.rmtree(entry_path)

            elif os.path.isfile(entry_path) and entry_path.endswith(".pyc"):
                os.remove(entry_path)


def sort_list_in_nested_structure(dictionary, omissions=""):
    """Recurse through a nested dictionary and sort any lists that are found.

    If the list that is found contains anything but strings, it is skipped
    as we can't compare lists containing different types. The omissions argument
    allows for certain sections of the dictionary to be omitted from sorting.
    """
    for field, value in dictionary.items():
        if isinstance(value, dict):
            for key in value.keys():
                section = dictionary[field][key]
                if isinstance(section, dict):
                    sort_list_in_nested_structure(section)
                elif (
                    isinstance(section, list)
                    and f"{field}/{key}" not in omissions
                    and all(isinstance(item, str) for item in section)
                ):
                    section.sort()

        # there's a possibility for nested lists containing dictionaries
        # in this case we recurse until we find a list to sort
        elif isinstance(value, list):
            for element in value:
                if isinstance(element, dict):
                    sort_list_in_nested_structure(element)
            try:
                value.sort()
            except TypeError:
                pass


# group one: package name
# group two: version (allows _, +, . in version)
# group three: build string - mostly not used here.  Match primarily matters
#        to specify when not to add .*

# if you are seeing mysterious unsatisfiable errors, with the package you're building being the
#    unsatisfiable part, then you probably need to update this regex.

spec_needing_star_re = re.compile(
    r"([\w\d\.\-\_]+)\s+((?<![><=])[\w\d\.\-\_]+?(?!\*))(\s+[\w\d\.\_]+)?$"
)  # NOQA
spec_ver_needing_star_re = re.compile(r"^([0-9a-zA-Z\.]+)$")


@overload
def ensure_valid_spec(spec: str, warn: bool = False) -> str: ...


@overload
def ensure_valid_spec(spec: MatchSpec, warn: bool = False) -> MatchSpec: ...


def ensure_valid_spec(spec: str | MatchSpec, warn: bool = False) -> str | MatchSpec:
    if isinstance(spec, MatchSpec):
        if (
            hasattr(spec, "version")
            and spec.version
            and (not spec.get("build", ""))
            and spec_ver_needing_star_re.match(str(spec.version))
        ):
            if str(spec.name) not in ("python", "numpy") or str(spec.version) != "x.x":
                spec = MatchSpec(
                    "{} {}".format(str(spec.name), str(spec.version) + ".*")
                )
    else:
        match = spec_needing_star_re.match(spec)
        # ignore exact pins (would be a 3rd group)
        if match and not match.group(3):
            if match.group(1) in ("python", "numpy") and match.group(2) == "x.x":
                spec = spec_needing_star_re.sub(r"\1 \2", spec)
            else:
                if "*" not in spec:
                    if match.group(1) not in ("python", "vc") and warn:
                        log = get_logger(__name__)
                        log.warning(
                            f"Adding .* to spec '{spec}' to ensure satisfiability.  Please "
                            "consider putting {{{{ var_name }}}}.* or some relational "
                            "operator (>/</>=/<=) on this spec in meta.yaml, or if req is "
                            "also a build req, using {{{{ pin_compatible() }}}} jinja2 "
                            "function instead.  See "
                            "https://conda.io/docs/user-guide/tasks/build-packages/variants.html#pinning-at-the-variant-level"
                        )
                    spec = spec_needing_star_re.sub(r"\1 \2.*", spec)
    return spec


def insert_variant_versions(requirements_dict, variant, env):
    build_deps = ensure_list(requirements_dict.get("build")) + ensure_list(
        requirements_dict.get("host")
    )
    reqs = ensure_list(requirements_dict.get(env))
    for key, val in variant.items():
        regex = re.compile(r"^({})(?:\s*$)".format(key.replace("_", "[-_]")))
        matches = [regex.match(pkg) for pkg in reqs]
        if any(matches):
            for i, x in enumerate(matches):
                if x and (env in ("build", "host") or x.group(1) in build_deps):
                    del reqs[i]
                    if not isinstance(val, str):
                        val = val[0]
                    reqs.insert(i, ensure_valid_spec(" ".join((x.group(1), val))))

    xx_re = re.compile(r"([0-9a-zA-Z\.\-\_]+)\s+x\.x")

    matches = [xx_re.match(pkg) for pkg in reqs]
    if any(matches):
        for i, x in enumerate(matches):
            if x:
                del reqs[i]
                reqs.insert(
                    i,
                    ensure_valid_spec(" ".join((x.group(1), variant.get(x.group(1))))),
                )
    if reqs:
        requirements_dict[env] = reqs


def match_peer_job(target_matchspec, other_m, this_m=None):
    """target_matchspec comes from the recipe.  target_variant is the variant from the recipe whose
    deps we are matching.  m is the peer job, which must satisfy conda and also have matching keys
    for any keys that are shared between target_variant and m.config.variant"""
    name, version, build = other_m.name(), other_m.version(), ""
    matchspec_matches = target_matchspec.match(
        PackageRecord(
            name=name,
            version=version,
            build=build,
            build_number=other_m.build_number(),
        )
    )

    variant_matches = True
    if this_m:
        other_m_used_vars = other_m.get_used_loop_vars()
        for v in this_m.get_used_loop_vars():
            if v in other_m_used_vars:
                variant_matches &= this_m.config.variant[v] == other_m.config.variant[v]
    return matchspec_matches and variant_matches


def expand_reqs(reqs_entry):
    if not hasattr(reqs_entry, "keys"):
        original = ensure_list(reqs_entry)[:]
        reqs_entry = (
            {"host": ensure_list(original), "run": ensure_list(original)}
            if original
            else {}
        )
    else:
        for sec in reqs_entry:
            reqs_entry[sec] = ensure_list(reqs_entry[sec])
    return reqs_entry


def sha256_checksum(filename, buffersize=65536):
    if islink(filename) and not isfile(filename):
        # symlink to nowhere so an empty file
        # this is the sha256 hash of an empty file
        return "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
    if not isfile(filename):
        return None
    sha256 = hashlib.sha256()
    with open(filename, "rb") as f:
        for block in iter(lambda: f.read(buffersize), b""):
            sha256.update(block)
    return sha256.hexdigest()


def write_bat_activation_text(file_handle, m):
    from .os_utils.external import find_executable

    file_handle.write(f'call "{root_script_dir}\\..\\condabin\\conda_hook.bat"\n')
    if m.is_cross:
        # HACK: we need both build and host envs "active" - i.e. on PATH,
        #     and with their activate.d scripts sourced. Conda only
        #     lets us activate one, though. This is a
        #     vile hack to trick conda into "stacking"
        #     two environments.
        #
        # Net effect: binaries come from host first, then build
        #
        # Conda 4.4 may break this by reworking the activate scripts.
        #  ^^ shouldn't be true
        # In conda 4.4, export CONDA_MAX_SHLVL=2 to stack envs to two
        #   levels deep.
        # conda 4.4 does require that a conda-meta/history file
        #   exists to identify a valid conda environment
        # conda 4.6 changes this one final time, by adding a '--stack' flag to the 'activate'
        #   command, and 'activate' does not stack environments by default without that flag
        history_file = join(m.config.host_prefix, "conda-meta", "history")
        if not isfile(history_file):
            if not isdir(dirname(history_file)):
                os.makedirs(dirname(history_file))
            open(history_file, "a").close()

        file_handle.write(
            f'call "{root_script_dir}\\..\\condabin\\conda.bat" activate "{m.config.host_prefix}"\n'
        )

    # Write build prefix activation AFTER host prefix, so that its executables come first
    file_handle.write(
        f'call "{root_script_dir}\\..\\condabin\\conda.bat" activate --stack "{m.config.build_prefix}"\n'
    )

    ccache = find_executable("ccache", m.config.build_prefix, False)
    if ccache:
        if isinstance(ccache, list):
            ccache = ccache[0]
        ccache_methods = {}
        ccache_methods["env_vars"] = False
        ccache_methods["symlinks"] = False
        ccache_methods["native"] = False
        if hasattr(m.config, "ccache_method"):
            ccache_methods[m.config.ccache_method] = True
        for method, value in ccache_methods.items():
            if value:
                if method == "env_vars":
                    file_handle.write(f'set "CC={ccache} %CC%"\n')
                    file_handle.write(f'set "CXX={ccache} %CXX%"\n')
                elif method == "symlinks":
                    dirname_ccache_ln_bin = join(m.config.build_prefix, "ccache-ln-bin")
                    file_handle.write(f"mkdir {dirname_ccache_ln_bin}\n")
                    file_handle.write(f"pushd {dirname_ccache_ln_bin}\n")
                    # If you use mklink.exe instead of mklink here it breaks as it's a builtin.
                    for ext in (".exe", ""):
                        # MSVC
                        file_handle.write(f"mklink cl{ext} {ccache}\n")
                        file_handle.write(f"mklink link{ext} {ccache}\n")
                        # GCC
                        file_handle.write(f"mklink gcc{ext} {ccache}\n")
                        file_handle.write(f"mklink g++{ext} {ccache}\n")
                        file_handle.write(f"mklink cc{ext} {ccache}\n")
                        file_handle.write(f"mklink c++{ext} {ccache}\n")
                        file_handle.write(f"mklink as{ext} {ccache}\n")
                        file_handle.write(f"mklink ar{ext} {ccache}\n")
                        file_handle.write(f"mklink nm{ext} {ccache}\n")
                        file_handle.write(f"mklink ranlib{ext} {ccache}\n")
                        file_handle.write(f"mklink gcc-ar{ext} {ccache}\n")
                        file_handle.write(f"mklink gcc-nm{ext} {ccache}\n")
                        file_handle.write(f"mklink gcc-ranlib{ext} {ccache}\n")
                    file_handle.write("popd\n")
                    file_handle.write(
                        f"set PATH={dirname_ccache_ln_bin};{os.path.dirname(ccache)};%PATH%\n"
                    )
                elif method == "native":
                    pass
                else:
                    print("ccache method {} not implemented")


channeldata_cache = {}


def download_channeldata(channel_url):
    global channeldata_cache
    if channel_url.startswith("file://") or channel_url not in channeldata_cache:
        urls = Channel.from_value(channel_url).urls()
        urls = {url.rsplit("/", 1)[0] for url in urls}
        data = {}
        for url in urls:
            with TemporaryDirectory() as td:
                tf = os.path.join(td, "channeldata.json")
                try:
                    download(url + "/channeldata.json", tf)
                    with open(tf) as f:
                        new_channeldata = json.load(f)
                except (JSONDecodeError, CondaHTTPError):
                    new_channeldata = {}
            merge_or_update_dict(data, new_channeldata)
        channeldata_cache[channel_url] = data
    else:
        data = channeldata_cache[channel_url]
    return data


def shutil_move_more_retrying(src, dest, debug_name):
    log = get_logger(__name__)
    log.info(f"Renaming {debug_name} directory '{src}' to '{dest}'")
    attempts_left = 5

    while attempts_left > 0:
        if os.path.exists(dest):
            rm_rf(dest)
        try:
            log.info(f"shutil.move({debug_name})={src}, dest={dest})")
            shutil.move(src, dest)
            if attempts_left != 5:
                log.warning(
                    f"shutil.move({debug_name}={src}, dest={dest}) succeeded on attempt number {6 - attempts_left}"
                )
            attempts_left = -1
        except:
            attempts_left = attempts_left - 1
        if attempts_left > 0:
            log.warning(
                f"Failed to rename {debug_name} directory, check with strace, struss or procmon. "
                "Will sleep for 3 seconds and try again!"
            )
            import time

            time.sleep(3)
        elif attempts_left != -1:
            log.error(
                f"Failed to rename {debug_name} directory despite sleeping and retrying."
            )


def is_conda_pkg(pkg_path: str) -> bool:
    """
    Determines whether string is pointing to a valid conda pkg
    """
    path = Path(pkg_path)

    return path.is_file() and (
        any(path.name.endswith(ext) for ext in CONDA_PACKAGE_EXTENSIONS)
    )


def package_record_to_requirement(prec: PackageRecord) -> str:
    return f"{prec.name} {prec.version} {prec.build}"