Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-21 15:56
📄 __init__.py """Extensions to the 'distutils' for large or complex distributions""" from fnmatch import fnmatchcase import functools import os import re import _distutils_hack.override # noqa: F401 import distutils.core from distutils.errors import DistutilsOptionError from distutils.util import convert_path from ._deprecation_warning import SetuptoolsDeprecationWarning import setuptools.version from setuptools.extension import Extension from setuptools.dist import Distribution from setuptools.depends import Require from . import monkey __all__ = [ 'setup', 'Distribution', 'Command', 'Extension', 'Require', 'SetuptoolsDeprecationWarning', 'find_packages', 'find_namespace_packages', ] __version__ = setuptools.version.__version__ bootstrap_install_from = None # If we run 2to3 on .py files, should we also convert docstrings? # Default: yes; assume that we can detect doctests reliably run_2to3_on_doctests = True # Standard package names for fixer packages lib2to3_fixer_packages = ['lib2to3.fixes'] class PackageFinder: """ Generate a list of all Python packages found within a directory """ @classmethod def find(cls, where='.', exclude=(), include=('',)): """Return a list all Python packages found within directory 'where' 'where' is the root directory which will be searched for packages. It should be supplied as a "cross-platform" (i.e. URL-style) path; it will be converted to the appropriate local path syntax. 'exclude' is a sequence of package names to exclude; '' can be used as a wildcard in the names, such that 'foo.' will exclude all subpackages of 'foo' (but not 'foo' itself). 'include' is a sequence of package names to include. If it's specified, only the named packages will be included. If it's not specified, all found packages will be included. 'include' can contain shell style wildcard patterns just like 'exclude'. """ return list(cls._find_packages_iter( convert_path(where), cls._build_filter('ez_setup', '__pycache__', exclude), cls._build_filter(include))) @classmethod def _find_packages_iter(cls, where, exclude, include): """ All the packages found in 'where' that pass the 'include' filter, but not the 'exclude' filter. """ for root, dirs, files in os.walk(where, followlinks=True): # Copy dirs to iterate over it, then empty dirs. all_dirs = dirs[:] dirs[:] = [] for dir in all_dirs: full_path = os.path.join(root, dir) rel_path = os.path.relpath(full_path, where) package = rel_path.replace(os.path.sep, '.') # Skip directory trees that are not valid packages if ('.' in dir or not cls._looks_like_package(full_path)): continue # Should this package be included? if include(package) and not exclude(package): yield package # Keep searching subdirectories, as there may be more packages # down there, even if the parent was excluded. dirs.append(dir) @staticmethod def _looks_like_package(path): """Does a directory look like a package?""" return os.path.isfile(os.path.join(path, '__init__.py')) @staticmethod def _build_filter(patterns): """ Given a list of patterns, return a callable that will be true only if the input matches at least one of the patterns. """ return lambda name: any(fnmatchcase(name, pat=pat) for pat in patterns) class PEP420PackageFinder(PackageFinder): @staticmethod def _looks_like_package(path): return True find_packages = PackageFinder.find find_namespace_packages = PEP420PackageFinder.find def _install_setup_requires(attrs): # Note: do not use `setuptools.Distribution` directly, as # our PEP 517 backend patch `distutils.core.Distribution`. class MinimalDistribution(distutils.core.Distribution): """ A minimal version of a distribution for supporting the fetch_build_eggs interface. """ def __init__(self, attrs): _incl = 'dependency_links', 'setup_requires' filtered = { k: attrs[k] for k in set(_incl) & set(attrs) } distutils.core.Distribution.__init__(self, filtered) def finalize_options(self): """ Disable finalize_options to avoid building the working set. Ref #2158. """ dist = MinimalDistribution(attrs) # Honor setup.cfg's options. dist.parse_config_files(ignore_option_errors=True) if dist.setup_requires: dist.fetch_build_eggs(dist.setup_requires) def setup(attrs): # Make sure we have any requirements needed to interpret 'attrs'. _install_setup_requires(attrs) return distutils.core.setup(attrs) setup.__doc__ = distutils.core.setup.__doc__ _Command = monkey.get_unpatched(distutils.core.Command) class Command(_Command): __doc__ = _Command.__doc__ command_consumes_arguments = False def __init__(self, dist, kw): """ Construct the command for dist, updating vars(self) with any keyword parameters. """ _Command.__init__(self, dist) vars(self).update(kw) def _ensure_stringlike(self, option, what, default=None): val = getattr(self, option) if val is None: setattr(self, option, default) return default elif not isinstance(val, str): raise DistutilsOptionError("'%s' must be a %s (got `%s`)" % (option, what, val)) return val def ensure_string_list(self, option): r"""Ensure that 'option' is a list of strings. If 'option' is currently a string, we split it either on /,\s/ or /\s+/, so "foo bar baz", "foo,bar,baz", and "foo, bar baz" all become ["foo", "bar", "baz"]. """ val = getattr(self, option) if val is None: return elif isinstance(val, str): setattr(self, option, re.split(r',\s\|\s+', val)) else: if isinstance(val, list): ok = all(isinstance(v, str) for v in val) else: ok = False if not ok: raise DistutilsOptionError( "'%s' must be a list of strings (got %r)" % (option, val)) def reinitialize_command(self, command, reinit_subcommands=0, *kw): cmd = _Command.reinitialize_command(self, command, reinit_subcommands) vars(cmd).update(kw) return cmd def _find_all_simple(path): """ Find all files under 'path' """ results = ( os.path.join(base, file) for base, dirs, files in os.walk(path, followlinks=True) for file in files ) return filter(os.path.isfile, results) def findall(dir=os.curdir): """ Find all files under 'dir' and return the list of full filenames. Unless dir is '.', return full filenames with dir prepended. """ files = _find_all_simple(dir) if dir == os.curdir: make_rel = functools.partial(os.path.relpath, start=dir) files = map(make_rel, files) return list(files) class sic(str): """Treat this string as-is (https://en.wikipedia.org/wiki/Sic)""" # Apply monkey patches monkey.patch_all()	7.5 KB	2025-09-21 11:46
📁 __pycache__	-	2026-03-21 15:45
📄 _deprecation_warning.py class SetuptoolsDeprecationWarning(Warning): """ Base class for warning deprecations in ``setuptools`` This class is not derived from ``DeprecationWarning``, and as such is visible by default. """	218 B	2025-09-21 11:46
📁 _distutils	-	2026-03-21 15:45
📄 _imp.py """ Re-implementation of find_module and get_frozen_object from the deprecated imp module. """ import os import importlib.util import importlib.machinery from .py34compat import module_from_spec PY_SOURCE = 1 PY_COMPILED = 2 C_EXTENSION = 3 C_BUILTIN = 6 PY_FROZEN = 7 def find_spec(module, paths): finder = ( importlib.machinery.PathFinder().find_spec if isinstance(paths, list) else importlib.util.find_spec ) return finder(module, paths) def find_module(module, paths=None): """Just like 'imp.find_module()', but with package support""" spec = find_spec(module, paths) if spec is None: raise ImportError("Can't find %s" % module) if not spec.has_location and hasattr(spec, 'submodule_search_locations'): spec = importlib.util.spec_from_loader('__init__.py', spec.loader) kind = -1 file = None static = isinstance(spec.loader, type) if spec.origin == 'frozen' or static and issubclass( spec.loader, importlib.machinery.FrozenImporter): kind = PY_FROZEN path = None # imp compabilty suffix = mode = '' # imp compability elif spec.origin == 'built-in' or static and issubclass( spec.loader, importlib.machinery.BuiltinImporter): kind = C_BUILTIN path = None # imp compabilty suffix = mode = '' # imp compability elif spec.has_location: path = spec.origin suffix = os.path.splitext(path)[1] mode = 'r' if suffix in importlib.machinery.SOURCE_SUFFIXES else 'rb' if suffix in importlib.machinery.SOURCE_SUFFIXES: kind = PY_SOURCE elif suffix in importlib.machinery.BYTECODE_SUFFIXES: kind = PY_COMPILED elif suffix in importlib.machinery.EXTENSION_SUFFIXES: kind = C_EXTENSION if kind in {PY_SOURCE, PY_COMPILED}: file = open(path, mode) else: path = None suffix = mode = '' return file, path, (suffix, mode, kind) def get_frozen_object(module, paths=None): spec = find_spec(module, paths) if not spec: raise ImportError("Can't find %s" % module) return spec.loader.get_code(module) def get_module(module, paths, info): spec = find_spec(module, paths) if not spec: raise ImportError("Can't find %s" % module) return module_from_spec(spec)	2.33 KB	2025-09-21 11:46
📁 _vendor	-	2026-03-21 15:45
📄 archive_util.py """Utilities for extracting common archive formats""" import zipfile import tarfile import os import shutil import posixpath import contextlib from distutils.errors import DistutilsError from pkg_resources import ensure_directory __all__ = [ "unpack_archive", "unpack_zipfile", "unpack_tarfile", "default_filter", "UnrecognizedFormat", "extraction_drivers", "unpack_directory", ] class UnrecognizedFormat(DistutilsError): """Couldn't recognize the archive type""" def default_filter(src, dst): """The default progress/filter callback; returns True for all files""" return dst def unpack_archive( filename, extract_dir, progress_filter=default_filter, drivers=None): """Unpack `filename` to `extract_dir`, or raise ``UnrecognizedFormat`` `progress_filter` is a function taking two arguments: a source path internal to the archive ('/'-separated), and a filesystem path where it will be extracted. The callback must return the desired extract path (which may be the same as the one passed in), or else ``None`` to skip that file or directory. The callback can thus be used to report on the progress of the extraction, as well as to filter the items extracted or alter their extraction paths. `drivers`, if supplied, must be a non-empty sequence of functions with the same signature as this function (minus the `drivers` argument), that raise ``UnrecognizedFormat`` if they do not support extracting the designated archive type. The `drivers` are tried in sequence until one is found that does not raise an error, or until all are exhausted (in which case ``UnrecognizedFormat`` is raised). If you do not supply a sequence of drivers, the module's ``extraction_drivers`` constant will be used, which means that ``unpack_zipfile`` and ``unpack_tarfile`` will be tried, in that order. """ for driver in drivers or extraction_drivers: try: driver(filename, extract_dir, progress_filter) except UnrecognizedFormat: continue else: return else: raise UnrecognizedFormat( "Not a recognized archive type: %s" % filename ) def unpack_directory(filename, extract_dir, progress_filter=default_filter): """"Unpack" a directory, using the same interface as for archives Raises ``UnrecognizedFormat`` if `filename` is not a directory """ if not os.path.isdir(filename): raise UnrecognizedFormat("%s is not a directory" % filename) paths = { filename: ('', extract_dir), } for base, dirs, files in os.walk(filename): src, dst = paths[base] for d in dirs: paths[os.path.join(base, d)] = src + d + '/', os.path.join(dst, d) for f in files: target = os.path.join(dst, f) target = progress_filter(src + f, target) if not target: # skip non-files continue ensure_directory(target) f = os.path.join(base, f) shutil.copyfile(f, target) shutil.copystat(f, target) def unpack_zipfile(filename, extract_dir, progress_filter=default_filter): """Unpack zip `filename` to `extract_dir` Raises ``UnrecognizedFormat`` if `filename` is not a zipfile (as determined by ``zipfile.is_zipfile()``). See ``unpack_archive()`` for an explanation of the `progress_filter` argument. """ if not zipfile.is_zipfile(filename): raise UnrecognizedFormat("%s is not a zip file" % (filename,)) with zipfile.ZipFile(filename) as z: for info in z.infolist(): name = info.filename # don't extract absolute paths or ones with .. in them if name.startswith('/') or '..' in name.split('/'): continue target = os.path.join(extract_dir, name.split('/')) target = progress_filter(name, target) if not target: continue if name.endswith('/'): # directory ensure_directory(target) else: # file ensure_directory(target) data = z.read(info.filename) with open(target, 'wb') as f: f.write(data) unix_attributes = info.external_attr >> 16 if unix_attributes: os.chmod(target, unix_attributes) def _resolve_tar_file_or_dir(tar_obj, tar_member_obj): """Resolve any links and extract link targets as normal files.""" while tar_member_obj is not None and ( tar_member_obj.islnk() or tar_member_obj.issym()): linkpath = tar_member_obj.linkname if tar_member_obj.issym(): base = posixpath.dirname(tar_member_obj.name) linkpath = posixpath.join(base, linkpath) linkpath = posixpath.normpath(linkpath) tar_member_obj = tar_obj._getmember(linkpath) is_file_or_dir = ( tar_member_obj is not None and (tar_member_obj.isfile() or tar_member_obj.isdir()) ) if is_file_or_dir: return tar_member_obj raise LookupError('Got unknown file type') def _iter_open_tar(tar_obj, extract_dir, progress_filter): """Emit member-destination pairs from a tar archive.""" # don't do any chowning! tar_obj.chown = lambda args: None with contextlib.closing(tar_obj): for member in tar_obj: name = member.name # don't extract absolute paths or ones with .. in them if name.startswith('/') or '..' in name.split('/'): continue prelim_dst = os.path.join(extract_dir, *name.split('/')) try: member = _resolve_tar_file_or_dir(tar_obj, member) except LookupError: continue final_dst = progress_filter(name, prelim_dst) if not final_dst: continue if final_dst.endswith(os.sep): final_dst = final_dst[:-1] yield member, final_dst def unpack_tarfile(filename, extract_dir, progress_filter=default_filter): """Unpack tar/tar.gz/tar.bz2 `filename` to `extract_dir` Raises ``UnrecognizedFormat`` if `filename` is not a tarfile (as determined by ``tarfile.open()``). See ``unpack_archive()`` for an explanation of the `progress_filter` argument. """ try: tarobj = tarfile.open(filename) except tarfile.TarError as e: raise UnrecognizedFormat( "%s is not a compressed or uncompressed tar file" % (filename,) ) from e for member, final_dst in _iter_open_tar( tarobj, extract_dir, progress_filter, ): try: # XXX Ugh tarobj._extract_member(member, final_dst) except tarfile.ExtractError: # chown/chmod/mkfifo/mknode/makedev failed pass return True extraction_drivers = unpack_directory, unpack_zipfile, unpack_tarfile	6.91 KB	2025-09-21 11:46
📄 build_meta.py """A PEP 517 interface to setuptools Previously, when a user or a command line tool (let's call it a "frontend") needed to make a request of setuptools to take a certain action, for example, generating a list of installation requirements, the frontend would would call "setup.py egg_info" or "setup.py bdist_wheel" on the command line. PEP 517 defines a different method of interfacing with setuptools. Rather than calling "setup.py" directly, the frontend should: 1. Set the current directory to the directory with a setup.py file 2. Import this module into a safe python interpreter (one in which setuptools can potentially set global variables or crash hard). 3. Call one of the functions defined in PEP 517. What each function does is defined in PEP 517. However, here is a "casual" definition of the functions (this definition should not be relied on for bug reports or API stability): - `build_wheel`: build a wheel in the folder and return the basename - `get_requires_for_build_wheel`: get the `setup_requires` to build - `prepare_metadata_for_build_wheel`: get the `install_requires` - `build_sdist`: build an sdist in the folder and return the basename - `get_requires_for_build_sdist`: get the `setup_requires` to build Again, this is not a formal definition! Just a "taste" of the module. """ import io import os import sys import tokenize import shutil import contextlib import tempfile import setuptools import distutils from pkg_resources import parse_requirements __all__ = ['get_requires_for_build_sdist', 'get_requires_for_build_wheel', 'prepare_metadata_for_build_wheel', 'build_wheel', 'build_sdist', '__legacy__', 'SetupRequirementsError'] class SetupRequirementsError(BaseException): def __init__(self, specifiers): self.specifiers = specifiers class Distribution(setuptools.dist.Distribution): def fetch_build_eggs(self, specifiers): specifier_list = list(map(str, parse_requirements(specifiers))) raise SetupRequirementsError(specifier_list) @classmethod @contextlib.contextmanager def patch(cls): """ Replace distutils.dist.Distribution with this class for the duration of this context. """ orig = distutils.core.Distribution distutils.core.Distribution = cls try: yield finally: distutils.core.Distribution = orig @contextlib.contextmanager def no_install_setup_requires(): """Temporarily disable installing setup_requires Under PEP 517, the backend reports build dependencies to the frontend, and the frontend is responsible for ensuring they're installed. So setuptools (acting as a backend) should not try to install them. """ orig = setuptools._install_setup_requires setuptools._install_setup_requires = lambda attrs: None try: yield finally: setuptools._install_setup_requires = orig def _get_immediate_subdirectories(a_dir): return [name for name in os.listdir(a_dir) if os.path.isdir(os.path.join(a_dir, name))] def _file_with_extension(directory, extension): matching = ( f for f in os.listdir(directory) if f.endswith(extension) ) file, = matching return file def _open_setup_script(setup_script): if not os.path.exists(setup_script): # Supply a default setup.py return io.StringIO(u"from setuptools import setup; setup()") return getattr(tokenize, 'open', open)(setup_script) class _BuildMetaBackend(object): def _fix_config(self, config_settings): config_settings = config_settings or {} config_settings.setdefault('--global-option', []) return config_settings def _get_build_requires(self, config_settings, requirements): config_settings = self._fix_config(config_settings) sys.argv = sys.argv[:1] + ['egg_info'] + \ config_settings["--global-option"] try: with Distribution.patch(): self.run_setup() except SetupRequirementsError as e: requirements += e.specifiers return requirements def run_setup(self, setup_script='setup.py'): # Note that we can reuse our build directory between calls # Correctness comes first, then optimization later __file__ = setup_script __name__ = '__main__' with _open_setup_script(__file__) as f: code = f.read().replace(r'\r\n', r'\n') exec(compile(code, __file__, 'exec'), locals()) def get_requires_for_build_wheel(self, config_settings=None): config_settings = self._fix_config(config_settings) return self._get_build_requires( config_settings, requirements=['wheel']) def get_requires_for_build_sdist(self, config_settings=None): config_settings = self._fix_config(config_settings) return self._get_build_requires(config_settings, requirements=[]) def prepare_metadata_for_build_wheel(self, metadata_directory, config_settings=None): sys.argv = sys.argv[:1] + [ 'dist_info', '--egg-base', metadata_directory] with no_install_setup_requires(): self.run_setup() dist_info_directory = metadata_directory while True: dist_infos = [f for f in os.listdir(dist_info_directory) if f.endswith('.dist-info')] if ( len(dist_infos) == 0 and len(_get_immediate_subdirectories(dist_info_directory)) == 1 ): dist_info_directory = os.path.join( dist_info_directory, os.listdir(dist_info_directory)[0]) continue assert len(dist_infos) == 1 break # PEP 517 requires that the .dist-info directory be placed in the # metadata_directory. To comply, we MUST copy the directory to the root if dist_info_directory != metadata_directory: shutil.move( os.path.join(dist_info_directory, dist_infos[0]), metadata_directory) shutil.rmtree(dist_info_directory, ignore_errors=True) return dist_infos[0] def _build_with_temp_dir(self, setup_command, result_extension, result_directory, config_settings): config_settings = self._fix_config(config_settings) result_directory = os.path.abspath(result_directory) # Build in a temporary directory, then copy to the target. os.makedirs(result_directory, exist_ok=True) with tempfile.TemporaryDirectory(dir=result_directory) as tmp_dist_dir: sys.argv = (sys.argv[:1] + setup_command + ['--dist-dir', tmp_dist_dir] + config_settings["--global-option"]) with no_install_setup_requires(): self.run_setup() result_basename = _file_with_extension( tmp_dist_dir, result_extension) result_path = os.path.join(result_directory, result_basename) if os.path.exists(result_path): # os.rename will fail overwriting on non-Unix. os.remove(result_path) os.rename(os.path.join(tmp_dist_dir, result_basename), result_path) return result_basename def build_wheel(self, wheel_directory, config_settings=None, metadata_directory=None): return self._build_with_temp_dir(['bdist_wheel'], '.whl', wheel_directory, config_settings) def build_sdist(self, sdist_directory, config_settings=None): return self._build_with_temp_dir(['sdist', '--formats', 'gztar'], '.tar.gz', sdist_directory, config_settings) class _BuildMetaLegacyBackend(_BuildMetaBackend): """Compatibility backend for setuptools This is a version of setuptools.build_meta that endeavors to maintain backwards compatibility with pre-PEP 517 modes of invocation. It exists as a temporary bridge between the old packaging mechanism and the new packaging mechanism, and will eventually be removed. """ def run_setup(self, setup_script='setup.py'): # In order to maintain compatibility with scripts assuming that # the setup.py script is in a directory on the PYTHONPATH, inject # '' into sys.path. (pypa/setuptools#1642) sys_path = list(sys.path) # Save the original path script_dir = os.path.dirname(os.path.abspath(setup_script)) if script_dir not in sys.path: sys.path.insert(0, script_dir) # Some setup.py scripts (e.g. in pygame and numpy) use sys.argv[0] to # get the directory of the source code. They expect it to refer to the # setup.py script. sys_argv_0 = sys.argv[0] sys.argv[0] = setup_script try: super(_BuildMetaLegacyBackend, self).run_setup(setup_script=setup_script) finally: # While PEP 517 frontends should be calling each hook in a fresh # subprocess according to the standard (and thus it should not be # strictly necessary to restore the old sys.path), we'll restore # the original path so that the path manipulation does not persist # within the hook after run_setup is called. sys.path[:] = sys_path sys.argv[0] = sys_argv_0 # The primary backend _BACKEND = _BuildMetaBackend() get_requires_for_build_wheel = _BACKEND.get_requires_for_build_wheel get_requires_for_build_sdist = _BACKEND.get_requires_for_build_sdist prepare_metadata_for_build_wheel = _BACKEND.prepare_metadata_for_build_wheel build_wheel = _BACKEND.build_wheel build_sdist = _BACKEND.build_sdist # The legacy backend __legacy__ = _BuildMetaLegacyBackend()	9.86 KB	2025-09-21 11:46
📁 command	-	2026-03-21 15:45
📄 config.py import ast import io import os import sys import warnings import functools import importlib from collections import defaultdict from functools import partial from functools import wraps import contextlib from distutils.errors import DistutilsOptionError, DistutilsFileError from setuptools.extern.packaging.version import LegacyVersion, parse from setuptools.extern.packaging.specifiers import SpecifierSet class StaticModule: """ Attempt to load the module by the name """ def __init__(self, name): spec = importlib.util.find_spec(name) with open(spec.origin) as strm: src = strm.read() module = ast.parse(src) vars(self).update(locals()) del self.self def __getattr__(self, attr): try: return next( ast.literal_eval(statement.value) for statement in self.module.body if isinstance(statement, ast.Assign) for target in statement.targets if isinstance(target, ast.Name) and target.id == attr ) except Exception as e: raise AttributeError( "{self.name} has no attribute {attr}".format(locals()) ) from e @contextlib.contextmanager def patch_path(path): """ Add path to front of sys.path for the duration of the context. """ try: sys.path.insert(0, path) yield finally: sys.path.remove(path) def read_configuration( filepath, find_others=False, ignore_option_errors=False): """Read given configuration file and returns options from it as a dict. :param str\|unicode filepath: Path to configuration file to get options from. :param bool find_others: Whether to search for other configuration files which could be on in various places. :param bool ignore_option_errors: Whether to silently ignore options, values of which could not be resolved (e.g. due to exceptions in directives such as file:, attr:, etc.). If False exceptions are propagated as expected. :rtype: dict """ from setuptools.dist import Distribution, _Distribution filepath = os.path.abspath(filepath) if not os.path.isfile(filepath): raise DistutilsFileError( 'Configuration file %s does not exist.' % filepath) current_directory = os.getcwd() os.chdir(os.path.dirname(filepath)) try: dist = Distribution() filenames = dist.find_config_files() if find_others else [] if filepath not in filenames: filenames.append(filepath) _Distribution.parse_config_files(dist, filenames=filenames) handlers = parse_configuration( dist, dist.command_options, ignore_option_errors=ignore_option_errors) finally: os.chdir(current_directory) return configuration_to_dict(handlers) def _get_option(target_obj, key): """ Given a target object and option key, get that option from the target object, either through a get_{key} method or from an attribute directly. """ getter_name = 'get_{key}'.format(locals()) by_attribute = functools.partial(getattr, target_obj, key) getter = getattr(target_obj, getter_name, by_attribute) return getter() def configuration_to_dict(handlers): """Returns configuration data gathered by given handlers as a dict. :param list[ConfigHandler] handlers: Handlers list, usually from parse_configuration() :rtype: dict """ config_dict = defaultdict(dict) for handler in handlers: for option in handler.set_options: value = _get_option(handler.target_obj, option) config_dict[handler.section_prefix][option] = value return config_dict def parse_configuration( distribution, command_options, ignore_option_errors=False): """Performs additional parsing of configuration options for a distribution. Returns a list of used option handlers. :param Distribution distribution: :param dict command_options: :param bool ignore_option_errors: Whether to silently ignore options, values of which could not be resolved (e.g. due to exceptions in directives such as file:, attr:, etc.). If False exceptions are propagated as expected. :rtype: list """ options = ConfigOptionsHandler( distribution, command_options, ignore_option_errors) options.parse() meta = ConfigMetadataHandler( distribution.metadata, command_options, ignore_option_errors, distribution.package_dir) meta.parse() return meta, options class ConfigHandler: """Handles metadata supplied in configuration files.""" section_prefix = None """Prefix for config sections handled by this handler. Must be provided by class heirs. """ aliases = {} """Options aliases. For compatibility with various packages. E.g.: d2to1 and pbr. Note: `-` in keys is replaced with `_` by config parser. """ def __init__(self, target_obj, options, ignore_option_errors=False): sections = {} section_prefix = self.section_prefix for section_name, section_options in options.items(): if not section_name.startswith(section_prefix): continue section_name = section_name.replace(section_prefix, '').strip('.') sections[section_name] = section_options self.ignore_option_errors = ignore_option_errors self.target_obj = target_obj self.sections = sections self.set_options = [] @property def parsers(self): """Metadata item name to parser function mapping.""" raise NotImplementedError( '%s must provide .parsers property' % self.__class__.__name__) def __setitem__(self, option_name, value): unknown = tuple() target_obj = self.target_obj # Translate alias into real name. option_name = self.aliases.get(option_name, option_name) current_value = getattr(target_obj, option_name, unknown) if current_value is unknown: raise KeyError(option_name) if current_value: # Already inhabited. Skipping. return skip_option = False parser = self.parsers.get(option_name) if parser: try: value = parser(value) except Exception: skip_option = True if not self.ignore_option_errors: raise if skip_option: return setter = getattr(target_obj, 'set_%s' % option_name, None) if setter is None: setattr(target_obj, option_name, value) else: setter(value) self.set_options.append(option_name) @classmethod def _parse_list(cls, value, separator=','): """Represents value as a list. Value is split either by separator (defaults to comma) or by lines. :param value: :param separator: List items separator character. :rtype: list """ if isinstance(value, list): # _get_parser_compound case return value if '\n' in value: value = value.splitlines() else: value = value.split(separator) return [chunk.strip() for chunk in value if chunk.strip()] @classmethod def _parse_dict(cls, value): """Represents value as a dict. :param value: :rtype: dict """ separator = '=' result = {} for line in cls._parse_list(value): key, sep, val = line.partition(separator) if sep != separator: raise DistutilsOptionError( 'Unable to parse option value to dict: %s' % value) result[key.strip()] = val.strip() return result @classmethod def _parse_bool(cls, value): """Represents value as boolean. :param value: :rtype: bool """ value = value.lower() return value in ('1', 'true', 'yes') @classmethod def _exclude_files_parser(cls, key): """Returns a parser function to make sure field inputs are not files. Parses a value after getting the key so error messages are more informative. :param key: :rtype: callable """ def parser(value): exclude_directive = 'file:' if value.startswith(exclude_directive): raise ValueError( 'Only strings are accepted for the {0} field, ' 'files are not accepted'.format(key)) return value return parser @classmethod def _parse_file(cls, value): """Represents value as a string, allowing including text from nearest files using `file:` directive. Directive is sandboxed and won't reach anything outside directory with setup.py. Examples: file: README.rst, CHANGELOG.md, src/file.txt :param str value: :rtype: str """ include_directive = 'file:' if not isinstance(value, str): return value if not value.startswith(include_directive): return value spec = value[len(include_directive):] filepaths = (os.path.abspath(path.strip()) for path in spec.split(',')) return '\n'.join( cls._read_file(path) for path in filepaths if (cls._assert_local(path) or True) and os.path.isfile(path) ) @staticmethod def _assert_local(filepath): if not filepath.startswith(os.getcwd()): raise DistutilsOptionError( '`file:` directive can not access %s' % filepath) @staticmethod def _read_file(filepath): with io.open(filepath, encoding='utf-8') as f: return f.read() @classmethod def _parse_attr(cls, value, package_dir=None): """Represents value as a module attribute. Examples: attr: package.attr attr: package.module.attr :param str value: :rtype: str """ attr_directive = 'attr:' if not value.startswith(attr_directive): return value attrs_path = value.replace(attr_directive, '').strip().split('.') attr_name = attrs_path.pop() module_name = '.'.join(attrs_path) module_name = module_name or '__init__' parent_path = os.getcwd() if package_dir: if attrs_path[0] in package_dir: # A custom path was specified for the module we want to import custom_path = package_dir[attrs_path[0]] parts = custom_path.rsplit('/', 1) if len(parts) > 1: parent_path = os.path.join(os.getcwd(), parts[0]) module_name = parts[1] else: module_name = custom_path elif '' in package_dir: # A custom parent directory was specified for all root modules parent_path = os.path.join(os.getcwd(), package_dir['']) with patch_path(parent_path): try: # attempt to load value statically return getattr(StaticModule(module_name), attr_name) except Exception: # fallback to simple import module = importlib.import_module(module_name) return getattr(module, attr_name) @classmethod def _get_parser_compound(cls, parse_methods): """Returns parser function to represents value as a list. Parses a value applying given methods one after another. :param parse_methods: :rtype: callable """ def parse(value): parsed = value for method in parse_methods: parsed = method(parsed) return parsed return parse @classmethod def _parse_section_to_dict(cls, section_options, values_parser=None): """Parses section options into a dictionary. Optionally applies a given parser to values. :param dict section_options: :param callable values_parser: :rtype: dict """ value = {} values_parser = values_parser or (lambda val: val) for key, (_, val) in section_options.items(): value[key] = values_parser(val) return value def parse_section(self, section_options): """Parses configuration file section. :param dict section_options: """ for (name, (_, value)) in section_options.items(): try: self[name] = value except KeyError: pass # Keep silent for a new option may appear anytime. def parse(self): """Parses configuration file items from one or more related sections. """ for section_name, section_options in self.sections.items(): method_postfix = '' if section_name: # [section.option] variant method_postfix = '_%s' % section_name section_parser_method = getattr( self, # Dots in section names are translated into dunderscores. ('parse_section%s' % method_postfix).replace('.', '__'), None) if section_parser_method is None: raise DistutilsOptionError( 'Unsupported distribution option section: [%s.%s]' % ( self.section_prefix, section_name)) section_parser_method(section_options) def _deprecated_config_handler(self, func, msg, warning_class): """ this function will wrap around parameters that are deprecated :param msg: deprecation message :param warning_class: class of warning exception to be raised :param func: function to be wrapped around """ @wraps(func) def config_handler(args, *kwargs): warnings.warn(msg, warning_class) return func(args, kwargs) return config_handler class ConfigMetadataHandler(ConfigHandler): section_prefix = 'metadata' aliases = { 'home_page': 'url', 'summary': 'description', 'classifier': 'classifiers', 'platform': 'platforms', } strict_mode = False """We need to keep it loose, to be partially compatible with `pbr` and `d2to1` packages which also uses `metadata` section. """ def __init__(self, target_obj, options, ignore_option_errors=False, package_dir=None): super(ConfigMetadataHandler, self).__init__(target_obj, options, ignore_option_errors) self.package_dir = package_dir @property def parsers(self): """Metadata item name to parser function mapping.""" parse_list = self._parse_list parse_file = self._parse_file parse_dict = self._parse_dict exclude_files_parser = self._exclude_files_parser return { 'platforms': parse_list, 'keywords': parse_list, 'provides': parse_list, 'requires': self._deprecated_config_handler( parse_list, "The requires parameter is deprecated, please use " "install_requires for runtime dependencies.", DeprecationWarning), 'obsoletes': parse_list, 'classifiers': self._get_parser_compound(parse_file, parse_list), 'license': exclude_files_parser('license'), 'license_file': self._deprecated_config_handler( exclude_files_parser('license_file'), "The license_file parameter is deprecated, " "use license_files instead.", DeprecationWarning), 'license_files': parse_list, 'description': parse_file, 'long_description': parse_file, 'version': self._parse_version, 'project_urls': parse_dict, } def _parse_version(self, value): """Parses `version` option value. :param value: :rtype: str """ version = self._parse_file(value) if version != value: version = version.strip() # Be strict about versions loaded from file because it's easy to # accidentally include newlines and other unintended content if isinstance(parse(version), LegacyVersion): tmpl = ( 'Version loaded from {value} does not ' 'comply with PEP 440: {version}' ) raise DistutilsOptionError(tmpl.format(locals())) return version version = self._parse_attr(value, self.package_dir) if callable(version): version = version() if not isinstance(version, str): if hasattr(version, '__iter__'): version = '.'.join(map(str, version)) else: version = '%s' % version return version class ConfigOptionsHandler(ConfigHandler): section_prefix = 'options' @property def parsers(self): """Metadata item name to parser function mapping.""" parse_list = self._parse_list parse_list_semicolon = partial(self._parse_list, separator=';') parse_bool = self._parse_bool parse_dict = self._parse_dict return { 'zip_safe': parse_bool, 'use_2to3': parse_bool, 'include_package_data': parse_bool, 'package_dir': parse_dict, 'use_2to3_fixers': parse_list, 'use_2to3_exclude_fixers': parse_list, 'convert_2to3_doctests': parse_list, 'scripts': parse_list, 'eager_resources': parse_list, 'dependency_links': parse_list, 'namespace_packages': parse_list, 'install_requires': parse_list_semicolon, 'setup_requires': parse_list_semicolon, 'tests_require': parse_list_semicolon, 'packages': self._parse_packages, 'entry_points': self._parse_file, 'py_modules': parse_list, 'python_requires': SpecifierSet, } def _parse_packages(self, value): """Parses `packages` option value. :param value: :rtype: list """ find_directives = ['find:', 'find_namespace:'] trimmed_value = value.strip() if trimmed_value not in find_directives: return self._parse_list(value) findns = trimmed_value == find_directives[1] # Read function arguments from a dedicated section. find_kwargs = self.parse_section_packages__find( self.sections.get('packages.find', {})) if findns: from setuptools import find_namespace_packages as find_packages else: from setuptools import find_packages return find_packages(*find_kwargs) def parse_section_packages__find(self, section_options): """Parses `packages.find` configuration file section. To be used in conjunction with _parse_packages(). :param dict section_options: """ section_data = self._parse_section_to_dict( section_options, self._parse_list) valid_keys = ['where', 'include', 'exclude'] find_kwargs = dict( [(k, v) for k, v in section_data.items() if k in valid_keys and v]) where = find_kwargs.get('where') if where is not None: find_kwargs['where'] = where[0] # cast list to single val return find_kwargs def parse_section_entry_points(self, section_options): """Parses `entry_points` configuration file section. :param dict section_options: """ parsed = self._parse_section_to_dict(section_options, self._parse_list) self['entry_points'] = parsed def _parse_package_data(self, section_options): parsed = self._parse_section_to_dict(section_options, self._parse_list) root = parsed.get('') if root: parsed[''] = root del parsed['*'] return parsed def parse_section_package_data(self, section_options): """Parses `package_data` configuration file section. :param dict section_options: """ self['package_data'] = self._parse_package_data(section_options) def parse_section_exclude_package_data(self, section_options): """Parses `exclude_package_data` configuration file section. :param dict section_options: """ self['exclude_package_data'] = self._parse_package_data( section_options) def parse_section_extras_require(self, section_options): """Parses `extras_require` configuration file section. :param dict section_options: """ parse_list = partial(self._parse_list, separator=';') self['extras_require'] = self._parse_section_to_dict( section_options, parse_list) def parse_section_data_files(self, section_options): """Parses `data_files` configuration file section. :param dict section_options: """ parsed = self._parse_section_to_dict(section_options, self._parse_list) self['data_files'] = [(k, v) for k, v in parsed.items()]	21.23 KB	2025-09-21 11:46
📄 dep_util.py from distutils.dep_util import newer_group # yes, this is was almost entirely copy-pasted from # 'newer_pairwise()', this is just another convenience # function. def newer_pairwise_group(sources_groups, targets): """Walk both arguments in parallel, testing if each source group is newer than its corresponding target. Returns a pair of lists (sources_groups, targets) where sources is newer than target, according to the semantics of 'newer_group()'. """ if len(sources_groups) != len(targets): raise ValueError( "'sources_group' and 'targets' must be the same length") # build a pair of lists (sources_groups, targets) where source is newer n_sources = [] n_targets = [] for i in range(len(sources_groups)): if newer_group(sources_groups[i], targets[i]): n_sources.append(sources_groups[i]) n_targets.append(targets[i]) return n_sources, n_targets	949 B	2025-09-21 11:46
📄 depends.py import sys import marshal import contextlib import dis from distutils.version import StrictVersion from ._imp import find_module, PY_COMPILED, PY_FROZEN, PY_SOURCE from . import _imp __all__ = [ 'Require', 'find_module', 'get_module_constant', 'extract_constant' ] class Require: """A prerequisite to building or installing a distribution""" def __init__( self, name, requested_version, module, homepage='', attribute=None, format=None): if format is None and requested_version is not None: format = StrictVersion if format is not None: requested_version = format(requested_version) if attribute is None: attribute = '__version__' self.__dict__.update(locals()) del self.self def full_name(self): """Return full package/distribution name, w/version""" if self.requested_version is not None: return '%s-%s' % (self.name, self.requested_version) return self.name def version_ok(self, version): """Is 'version' sufficiently up-to-date?""" return self.attribute is None or self.format is None or \ str(version) != "unknown" and version >= self.requested_version def get_version(self, paths=None, default="unknown"): """Get version number of installed module, 'None', or 'default' Search 'paths' for module. If not found, return 'None'. If found, return the extracted version attribute, or 'default' if no version attribute was specified, or the value cannot be determined without importing the module. The version is formatted according to the requirement's version format (if any), unless it is 'None' or the supplied 'default'. """ if self.attribute is None: try: f, p, i = find_module(self.module, paths) if f: f.close() return default except ImportError: return None v = get_module_constant(self.module, self.attribute, default, paths) if v is not None and v is not default and self.format is not None: return self.format(v) return v def is_present(self, paths=None): """Return true if dependency is present on 'paths'""" return self.get_version(paths) is not None def is_current(self, paths=None): """Return true if dependency is present and up-to-date on 'paths'""" version = self.get_version(paths) if version is None: return False return self.version_ok(version) def maybe_close(f): @contextlib.contextmanager def empty(): yield return if not f: return empty() return contextlib.closing(f) def get_module_constant(module, symbol, default=-1, paths=None): """Find 'module' by searching 'paths', and extract 'symbol' Return 'None' if 'module' does not exist on 'paths', or it does not define 'symbol'. If the module defines 'symbol' as a constant, return the constant. Otherwise, return 'default'.""" try: f, path, (suffix, mode, kind) = info = find_module(module, paths) except ImportError: # Module doesn't exist return None with maybe_close(f): if kind == PY_COMPILED: f.read(8) # skip magic & date code = marshal.load(f) elif kind == PY_FROZEN: code = _imp.get_frozen_object(module, paths) elif kind == PY_SOURCE: code = compile(f.read(), path, 'exec') else: # Not something we can parse; we'll have to import it. :( imported = _imp.get_module(module, paths, info) return getattr(imported, symbol, None) return extract_constant(code, symbol, default) def extract_constant(code, symbol, default=-1): """Extract the constant value of 'symbol' from 'code' If the name 'symbol' is bound to a constant value by the Python code object 'code', return that value. If 'symbol' is bound to an expression, return 'default'. Otherwise, return 'None'. Return value is based on the first assignment to 'symbol'. 'symbol' must be a global, or at least a non-"fast" local in the code block. That is, only 'STORE_NAME' and 'STORE_GLOBAL' opcodes are checked, and 'symbol' must be present in 'code.co_names'. """ if symbol not in code.co_names: # name's not there, can't possibly be an assignment return None name_idx = list(code.co_names).index(symbol) STORE_NAME = 90 STORE_GLOBAL = 97 LOAD_CONST = 100 const = default for byte_code in dis.Bytecode(code): op = byte_code.opcode arg = byte_code.arg if op == LOAD_CONST: const = code.co_consts[arg] elif arg == name_idx and (op == STORE_NAME or op == STORE_GLOBAL): return const else: const = default def _update_globals(): """ Patch the globals to remove the objects not available on some platforms. XXX it'd be better to test assertions about bytecode instead. """ if not sys.platform.startswith('java') and sys.platform != 'cli': return incompatible = 'extract_constant', 'get_module_constant' for name in incompatible: del globals()[name] __all__.remove(name) _update_globals()	5.35 KB	2025-09-21 11:46
📄 dist.py # -- coding: utf-8 -- __all__ = ['Distribution'] import io import sys import re import os import warnings import numbers import distutils.log import distutils.core import distutils.cmd import distutils.dist from distutils.util import strtobool from distutils.debug import DEBUG from distutils.fancy_getopt import translate_longopt from glob import iglob import itertools from collections import defaultdict from email import message_from_file from distutils.errors import DistutilsOptionError, DistutilsSetupError from distutils.util import rfc822_escape from distutils.version import StrictVersion from setuptools.extern import packaging from setuptools.extern import ordered_set from . import SetuptoolsDeprecationWarning import setuptools from setuptools import windows_support from setuptools.monkey import get_unpatched from setuptools.config import parse_configuration import pkg_resources __import__('setuptools.extern.packaging.specifiers') __import__('setuptools.extern.packaging.version') def _get_unpatched(cls): warnings.warn("Do not call this function", DistDeprecationWarning) return get_unpatched(cls) def get_metadata_version(self): mv = getattr(self, 'metadata_version', None) if mv is None: if self.long_description_content_type or self.provides_extras: mv = StrictVersion('2.1') elif (self.maintainer is not None or self.maintainer_email is not None or getattr(self, 'python_requires', None) is not None or self.project_urls): mv = StrictVersion('1.2') elif (self.provides or self.requires or self.obsoletes or self.classifiers or self.download_url): mv = StrictVersion('1.1') else: mv = StrictVersion('1.0') self.metadata_version = mv return mv def read_pkg_file(self, file): """Reads the metadata values from a file object.""" msg = message_from_file(file) def _read_field(name): value = msg[name] if value == 'UNKNOWN': return None return value def _read_list(name): values = msg.get_all(name, None) if values == []: return None return values self.metadata_version = StrictVersion(msg['metadata-version']) self.name = _read_field('name') self.version = _read_field('version') self.description = _read_field('summary') # we are filling author only. self.author = _read_field('author') self.maintainer = None self.author_email = _read_field('author-email') self.maintainer_email = None self.url = _read_field('home-page') self.license = _read_field('license') if 'download-url' in msg: self.download_url = _read_field('download-url') else: self.download_url = None self.long_description = _read_field('description') self.description = _read_field('summary') if 'keywords' in msg: self.keywords = _read_field('keywords').split(',') self.platforms = _read_list('platform') self.classifiers = _read_list('classifier') # PEP 314 - these fields only exist in 1.1 if self.metadata_version == StrictVersion('1.1'): self.requires = _read_list('requires') self.provides = _read_list('provides') self.obsoletes = _read_list('obsoletes') else: self.requires = None self.provides = None self.obsoletes = None self.license_files = _read_list('license-file') def single_line(val): # quick and dirty validation for description pypa/setuptools#1390 if '\n' in val: # TODO after 2021-07-31: Replace with `raise ValueError("newlines not allowed")` warnings.warn("newlines not allowed and will break in the future") val = val.replace('\n', ' ') return val # Based on Python 3.5 version def write_pkg_file(self, file): # noqa: C901 # is too complex (14) # FIXME """Write the PKG-INFO format data to a file object. """ version = self.get_metadata_version() def write_field(key, value): file.write("%s: %s\n" % (key, value)) write_field('Metadata-Version', str(version)) write_field('Name', self.get_name()) write_field('Version', self.get_version()) write_field('Summary', single_line(self.get_description())) write_field('Home-page', self.get_url()) if version < StrictVersion('1.2'): write_field('Author', self.get_contact()) write_field('Author-email', self.get_contact_email()) else: optional_fields = ( ('Author', 'author'), ('Author-email', 'author_email'), ('Maintainer', 'maintainer'), ('Maintainer-email', 'maintainer_email'), ) for field, attr in optional_fields: attr_val = getattr(self, attr) if attr_val is not None: write_field(field, attr_val) write_field('License', self.get_license()) if self.download_url: write_field('Download-URL', self.download_url) for project_url in self.project_urls.items(): write_field('Project-URL', '%s, %s' % project_url) long_desc = rfc822_escape(self.get_long_description()) write_field('Description', long_desc) keywords = ','.join(self.get_keywords()) if keywords: write_field('Keywords', keywords) if version >= StrictVersion('1.2'): for platform in self.get_platforms(): write_field('Platform', platform) else: self._write_list(file, 'Platform', self.get_platforms()) self._write_list(file, 'Classifier', self.get_classifiers()) # PEP 314 self._write_list(file, 'Requires', self.get_requires()) self._write_list(file, 'Provides', self.get_provides()) self._write_list(file, 'Obsoletes', self.get_obsoletes()) # Setuptools specific for PEP 345 if hasattr(self, 'python_requires'): write_field('Requires-Python', self.python_requires) # PEP 566 if self.long_description_content_type: write_field( 'Description-Content-Type', self.long_description_content_type ) if self.provides_extras: for extra in self.provides_extras: write_field('Provides-Extra', extra) self._write_list(file, 'License-File', self.license_files or []) sequence = tuple, list def check_importable(dist, attr, value): try: ep = pkg_resources.EntryPoint.parse('x=' + value) assert not ep.extras except (TypeError, ValueError, AttributeError, AssertionError) as e: raise DistutilsSetupError( "%r must be importable 'module:attrs' string (got %r)" % (attr, value) ) from e def assert_string_list(dist, attr, value): """Verify that value is a string list""" try: # verify that value is a list or tuple to exclude unordered # or single-use iterables assert isinstance(value, (list, tuple)) # verify that elements of value are strings assert ''.join(value) != value except (TypeError, ValueError, AttributeError, AssertionError) as e: raise DistutilsSetupError( "%r must be a list of strings (got %r)" % (attr, value) ) from e def check_nsp(dist, attr, value): """Verify that namespace packages are valid""" ns_packages = value assert_string_list(dist, attr, ns_packages) for nsp in ns_packages: if not dist.has_contents_for(nsp): raise DistutilsSetupError( "Distribution contains no modules or packages for " + "namespace package %r" % nsp ) parent, sep, child = nsp.rpartition('.') if parent and parent not in ns_packages: distutils.log.warn( "WARNING: %r is declared as a package namespace, but %r" " is not: please correct this in setup.py", nsp, parent ) def check_extras(dist, attr, value): """Verify that extras_require mapping is valid""" try: list(itertools.starmap(_check_extra, value.items())) except (TypeError, ValueError, AttributeError) as e: raise DistutilsSetupError( "'extras_require' must be a dictionary whose values are " "strings or lists of strings containing valid project/version " "requirement specifiers." ) from e def _check_extra(extra, reqs): name, sep, marker = extra.partition(':') if marker and pkg_resources.invalid_marker(marker): raise DistutilsSetupError("Invalid environment marker: " + marker) list(pkg_resources.parse_requirements(reqs)) def assert_bool(dist, attr, value): """Verify that value is True, False, 0, or 1""" if bool(value) != value: tmpl = "{attr!r} must be a boolean value (got {value!r})" raise DistutilsSetupError(tmpl.format(attr=attr, value=value)) def check_requirements(dist, attr, value): """Verify that install_requires is a valid requirements list""" try: list(pkg_resources.parse_requirements(value)) if isinstance(value, (dict, set)): raise TypeError("Unordered types are not allowed") except (TypeError, ValueError) as error: tmpl = ( "{attr!r} must be a string or list of strings " "containing valid project/version requirement specifiers; {error}" ) raise DistutilsSetupError( tmpl.format(attr=attr, error=error) ) from error def check_specifier(dist, attr, value): """Verify that value is a valid version specifier""" try: packaging.specifiers.SpecifierSet(value) except packaging.specifiers.InvalidSpecifier as error: tmpl = ( "{attr!r} must be a string " "containing valid version specifiers; {error}" ) raise DistutilsSetupError( tmpl.format(attr=attr, error=error) ) from error def check_entry_points(dist, attr, value): """Verify that entry_points map is parseable""" try: pkg_resources.EntryPoint.parse_map(value) except ValueError as e: raise DistutilsSetupError(e) from e def check_test_suite(dist, attr, value): if not isinstance(value, str): raise DistutilsSetupError("test_suite must be a string") def check_package_data(dist, attr, value): """Verify that value is a dictionary of package names to glob lists""" if not isinstance(value, dict): raise DistutilsSetupError( "{!r} must be a dictionary mapping package names to lists of " "string wildcard patterns".format(attr)) for k, v in value.items(): if not isinstance(k, str): raise DistutilsSetupError( "keys of {!r} dict must be strings (got {!r})" .format(attr, k) ) assert_string_list(dist, 'values of {!r} dict'.format(attr), v) def check_packages(dist, attr, value): for pkgname in value: if not re.match(r'\w+(\.\w+)', pkgname): distutils.log.warn( "WARNING: %r not a valid package name; please use only " ".-separated package names in setup.py", pkgname ) _Distribution = get_unpatched(distutils.core.Distribution) class Distribution(_Distribution): """Distribution with support for tests and package data This is an enhanced version of 'distutils.dist.Distribution' that effectively adds the following new optional keyword arguments to 'setup()': 'install_requires' -- a string or sequence of strings specifying project versions that the distribution requires when installed, in the format used by 'pkg_resources.require()'. They will be installed automatically when the package is installed. If you wish to use packages that are not available in PyPI, or want to give your users an alternate download location, you can add a 'find_links' option to the '[easy_install]' section of your project's 'setup.cfg' file, and then setuptools will scan the listed web pages for links that satisfy the requirements. 'extras_require' -- a dictionary mapping names of optional "extras" to the additional requirement(s) that using those extras incurs. For example, this:: extras_require = dict(reST = ["docutils>=0.3", "reSTedit"]) indicates that the distribution can optionally provide an extra capability called "reST", but it can only be used if docutils and reSTedit are installed. If the user installs your package using EasyInstall and requests one of your extras, the corresponding additional requirements will be installed if needed. 'test_suite' -- the name of a test suite to run for the 'test' command. If the user runs 'python setup.py test', the package will be installed, and the named test suite will be run. The format is the same as would be used on a 'unittest.py' command line. That is, it is the dotted name of an object to import and call to generate a test suite. 'package_data' -- a dictionary mapping package names to lists of filenames or globs to use to find data files contained in the named packages. If the dictionary has filenames or globs listed under '""' (the empty string), those names will be searched for in every package, in addition to any names for the specific package. Data files found using these names/globs will be installed along with the package, in the same location as the package. Note that globs are allowed to reference the contents of non-package subdirectories, as long as you use '/' as a path separator. (Globs are automatically converted to platform-specific paths at runtime.) In addition to these new keywords, this class also has several new methods for manipulating the distribution's contents. For example, the 'include()' and 'exclude()' methods can be thought of as in-place add and subtract commands that add or remove packages, modules, extensions, and so on from the distribution. """ _DISTUTILS_UNSUPPORTED_METADATA = { 'long_description_content_type': None, 'project_urls': dict, 'provides_extras': ordered_set.OrderedSet, 'license_file': lambda: None, 'license_files': lambda: None, } _patched_dist = None def patch_missing_pkg_info(self, attrs): # Fake up a replacement for the data that would normally come from # PKG-INFO, but which might not yet be built if this is a fresh # checkout. # if not attrs or 'name' not in attrs or 'version' not in attrs: return key = pkg_resources.safe_name(str(attrs['name'])).lower() dist = pkg_resources.working_set.by_key.get(key) if dist is not None and not dist.has_metadata('PKG-INFO'): dist._version = pkg_resources.safe_version(str(attrs['version'])) self._patched_dist = dist def __init__(self, attrs=None): have_package_data = hasattr(self, "package_data") if not have_package_data: self.package_data = {} attrs = attrs or {} self.dist_files = [] # Filter-out setuptools' specific options. self.src_root = attrs.pop("src_root", None) self.patch_missing_pkg_info(attrs) self.dependency_links = attrs.pop('dependency_links', []) self.setup_requires = attrs.pop('setup_requires', []) for ep in pkg_resources.iter_entry_points('distutils.setup_keywords'): vars(self).setdefault(ep.name, None) _Distribution.__init__(self, { k: v for k, v in attrs.items() if k not in self._DISTUTILS_UNSUPPORTED_METADATA }) # Fill-in missing metadata fields not supported by distutils. # Note some fields may have been set by other tools (e.g. pbr) # above; they are taken preferrentially to setup() arguments for option, default in self._DISTUTILS_UNSUPPORTED_METADATA.items(): for source in self.metadata.__dict__, attrs: if option in source: value = source[option] break else: value = default() if default else None setattr(self.metadata, option, value) self.metadata.version = self._normalize_version( self._validate_version(self.metadata.version)) self._finalize_requires() @staticmethod def _normalize_version(version): if isinstance(version, setuptools.sic) or version is None: return version normalized = str(packaging.version.Version(version)) if version != normalized: tmpl = "Normalizing '{version}' to '{normalized}'" warnings.warn(tmpl.format(locals())) return normalized return version @staticmethod def _validate_version(version): if isinstance(version, numbers.Number): # Some people apparently take "version number" too literally :) version = str(version) if version is not None: try: packaging.version.Version(version) except (packaging.version.InvalidVersion, TypeError): warnings.warn( "The version specified (%r) is an invalid version, this " "may not work as expected with newer versions of " "setuptools, pip, and PyPI. Please see PEP 440 for more " "details." % version ) return setuptools.sic(version) return version def _finalize_requires(self): """ Set `metadata.python_requires` and fix environment markers in `install_requires` and `extras_require`. """ if getattr(self, 'python_requires', None): self.metadata.python_requires = self.python_requires if getattr(self, 'extras_require', None): for extra in self.extras_require.keys(): # Since this gets called multiple times at points where the # keys have become 'converted' extras, ensure that we are only # truly adding extras we haven't seen before here. extra = extra.split(':')[0] if extra: self.metadata.provides_extras.add(extra) self._convert_extras_requirements() self._move_install_requirements_markers() def _convert_extras_requirements(self): """ Convert requirements in `extras_require` of the form `"extra": ["barbazquux; {marker}"]` to `"extra:{marker}": ["barbazquux"]`. """ spec_ext_reqs = getattr(self, 'extras_require', None) or {} self._tmp_extras_require = defaultdict(list) for section, v in spec_ext_reqs.items(): # Do not strip empty sections. self._tmp_extras_require[section] for r in pkg_resources.parse_requirements(v): suffix = self._suffix_for(r) self._tmp_extras_require[section + suffix].append(r) @staticmethod def _suffix_for(req): """ For a requirement, return the 'extras_require' suffix for that requirement. """ return ':' + str(req.marker) if req.marker else '' def _move_install_requirements_markers(self): """ Move requirements in `install_requires` that are using environment markers `extras_require`. """ # divide the install_requires into two sets, simple ones still # handled by install_requires and more complex ones handled # by extras_require. def is_simple_req(req): return not req.marker spec_inst_reqs = getattr(self, 'install_requires', None) or () inst_reqs = list(pkg_resources.parse_requirements(spec_inst_reqs)) simple_reqs = filter(is_simple_req, inst_reqs) complex_reqs = itertools.filterfalse(is_simple_req, inst_reqs) self.install_requires = list(map(str, simple_reqs)) for r in complex_reqs: self._tmp_extras_require[':' + str(r.marker)].append(r) self.extras_require = dict( (k, [str(r) for r in map(self._clean_req, v)]) for k, v in self._tmp_extras_require.items() ) def _clean_req(self, req): """ Given a Requirement, remove environment markers and return it. """ req.marker = None return req def _finalize_license_files(self): """Compute names of all license files which should be included.""" license_files: Optional[List[str]] = self.metadata.license_files patterns: List[str] = license_files if license_files else [] license_file: Optional[str] = self.metadata.license_file if license_file and license_file not in patterns: patterns.append(license_file) if license_files is None and license_file is None: # Default patterns match the ones wheel uses # See https://wheel.readthedocs.io/en/stable/user_guide.html # -> 'Including license files in the generated wheel file' patterns = ('LICEN[CS]E', 'COPYING', 'NOTICE', 'AUTHORS') self.metadata.license_files = list( unique_everseen(self._expand_patterns(patterns))) @staticmethod def _expand_patterns(patterns): return ( path for pattern in patterns for path in iglob(pattern) if not path.endswith('~') and os.path.isfile(path) ) # FIXME: 'Distribution._parse_config_files' is too complex (14) def _parse_config_files(self, filenames=None): # noqa: C901 """ Adapted from distutils.dist.Distribution.parse_config_files, this method provides the same functionality in subtly-improved ways. """ from configparser import ConfigParser # Ignore install directory options if we have a venv ignore_options = [] if sys.prefix == sys.base_prefix else [ 'install-base', 'install-platbase', 'install-lib', 'install-platlib', 'install-purelib', 'install-headers', 'install-scripts', 'install-data', 'prefix', 'exec-prefix', 'home', 'user', 'root', ] ignore_options = frozenset(ignore_options) if filenames is None: filenames = self.find_config_files() if DEBUG: self.announce("Distribution.parse_config_files():") parser = ConfigParser() parser.optionxform = str for filename in filenames: with io.open(filename, encoding='utf-8') as reader: if DEBUG: self.announce(" reading {filename}".format(locals())) parser.read_file(reader) for section in parser.sections(): options = parser.options(section) opt_dict = self.get_option_dict(section) for opt in options: if opt == '__name__' or opt in ignore_options: continue val = parser.get(section, opt) opt = opt.replace('-', '_') opt_dict[opt] = (filename, val) # Make the ConfigParser forget everything (so we retain # the original filenames that options come from) parser.__init__() if 'global' not in self.command_options: return # If there was a "global" section in the config file, use it # to set Distribution options. for (opt, (src, val)) in self.command_options['global'].items(): alias = self.negative_opt.get(opt) if alias: val = not strtobool(val) elif opt in ('verbose', 'dry_run'): # ugh! val = strtobool(val) try: setattr(self, alias or opt, val) except ValueError as e: raise DistutilsOptionError(e) from e # FIXME: 'Distribution._set_command_options' is too complex (14) def _set_command_options(self, command_obj, option_dict=None): # noqa: C901 """ Set the options for 'command_obj' from 'option_dict'. Basically this means copying elements of a dictionary ('option_dict') to attributes of an instance ('command'). 'command_obj' must be a Command instance. If 'option_dict' is not supplied, uses the standard option dictionary for this command (from 'self.command_options'). (Adopted from distutils.dist.Distribution._set_command_options) """ command_name = command_obj.get_command_name() if option_dict is None: option_dict = self.get_option_dict(command_name) if DEBUG: self.announce(" setting options for '%s' command:" % command_name) for (option, (source, value)) in option_dict.items(): if DEBUG: self.announce(" %s = %s (from %s)" % (option, value, source)) try: bool_opts = [translate_longopt(o) for o in command_obj.boolean_options] except AttributeError: bool_opts = [] try: neg_opt = command_obj.negative_opt except AttributeError: neg_opt = {} try: is_string = isinstance(value, str) if option in neg_opt and is_string: setattr(command_obj, neg_opt[option], not strtobool(value)) elif option in bool_opts and is_string: setattr(command_obj, option, strtobool(value)) elif hasattr(command_obj, option): setattr(command_obj, option, value) else: raise DistutilsOptionError( "error in %s: command '%s' has no such option '%s'" % (source, command_name, option)) except ValueError as e: raise DistutilsOptionError(e) from e def parse_config_files(self, filenames=None, ignore_option_errors=False): """Parses configuration files from various levels and loads configuration. """ self._parse_config_files(filenames=filenames) parse_configuration(self, self.command_options, ignore_option_errors=ignore_option_errors) self._finalize_requires() self._finalize_license_files() def fetch_build_eggs(self, requires): """Resolve pre-setup requirements""" resolved_dists = pkg_resources.working_set.resolve( pkg_resources.parse_requirements(requires), installer=self.fetch_build_egg, replace_conflicting=True, ) for dist in resolved_dists: pkg_resources.working_set.add(dist, replace=True) return resolved_dists def finalize_options(self): """ Allow plugins to apply arbitrary operations to the distribution. Each hook may optionally define a 'order' to influence the order of execution. Smaller numbers go first and the default is 0. """ group = 'setuptools.finalize_distribution_options' def by_order(hook): return getattr(hook, 'order', 0) eps = map(lambda e: e.load(), pkg_resources.iter_entry_points(group)) for ep in sorted(eps, key=by_order): ep(self) def _finalize_setup_keywords(self): for ep in pkg_resources.iter_entry_points('distutils.setup_keywords'): value = getattr(self, ep.name, None) if value is not None: ep.require(installer=self.fetch_build_egg) ep.load()(self, ep.name, value) def _finalize_2to3_doctests(self): if getattr(self, 'convert_2to3_doctests', None): # XXX may convert to set here when we can rely on set being builtin self.convert_2to3_doctests = [ os.path.abspath(p) for p in self.convert_2to3_doctests ] else: self.convert_2to3_doctests = [] def get_egg_cache_dir(self): egg_cache_dir = os.path.join(os.curdir, '.eggs') if not os.path.exists(egg_cache_dir): os.mkdir(egg_cache_dir) windows_support.hide_file(egg_cache_dir) readme_txt_filename = os.path.join(egg_cache_dir, 'README.txt') with open(readme_txt_filename, 'w') as f: f.write('This directory contains eggs that were downloaded ' 'by setuptools to build, test, and run plug-ins.\n\n') f.write('This directory caches those eggs to prevent ' 'repeated downloads.\n\n') f.write('However, it is safe to delete this directory.\n\n') return egg_cache_dir def fetch_build_egg(self, req): """Fetch an egg needed for building""" from setuptools.installer import fetch_build_egg return fetch_build_egg(self, req) def get_command_class(self, command): """Pluggable version of get_command_class()""" if command in self.cmdclass: return self.cmdclass[command] eps = pkg_resources.iter_entry_points('distutils.commands', command) for ep in eps: ep.require(installer=self.fetch_build_egg) self.cmdclass[command] = cmdclass = ep.load() return cmdclass else: return _Distribution.get_command_class(self, command) def print_commands(self): for ep in pkg_resources.iter_entry_points('distutils.commands'): if ep.name not in self.cmdclass: # don't require extras as the commands won't be invoked cmdclass = ep.resolve() self.cmdclass[ep.name] = cmdclass return _Distribution.print_commands(self) def get_command_list(self): for ep in pkg_resources.iter_entry_points('distutils.commands'): if ep.name not in self.cmdclass: # don't require extras as the commands won't be invoked cmdclass = ep.resolve() self.cmdclass[ep.name] = cmdclass return _Distribution.get_command_list(self) def include(self, attrs): """Add items to distribution that are named in keyword arguments For example, 'dist.include(py_modules=["x"])' would add 'x' to the distribution's 'py_modules' attribute, if it was not already there. Currently, this method only supports inclusion for attributes that are lists or tuples. If you need to add support for adding to other attributes in this or a subclass, you can add an '_include_X' method, where 'X' is the name of the attribute. The method will be called with the value passed to 'include()'. So, 'dist.include(foo={"bar":"baz"})' will try to call 'dist._include_foo({"bar":"baz"})', which can then handle whatever special inclusion logic is needed. """ for k, v in attrs.items(): include = getattr(self, '_include_' + k, None) if include: include(v) else: self._include_misc(k, v) def exclude_package(self, package): """Remove packages, modules, and extensions in named package""" pfx = package + '.' if self.packages: self.packages = [ p for p in self.packages if p != package and not p.startswith(pfx) ] if self.py_modules: self.py_modules = [ p for p in self.py_modules if p != package and not p.startswith(pfx) ] if self.ext_modules: self.ext_modules = [ p for p in self.ext_modules if p.name != package and not p.name.startswith(pfx) ] def has_contents_for(self, package): """Return true if 'exclude_package(package)' would do something""" pfx = package + '.' for p in self.iter_distribution_names(): if p == package or p.startswith(pfx): return True def _exclude_misc(self, name, value): """Handle 'exclude()' for list/tuple attrs without a special handler""" if not isinstance(value, sequence): raise DistutilsSetupError( "%s: setting must be a list or tuple (%r)" % (name, value) ) try: old = getattr(self, name) except AttributeError as e: raise DistutilsSetupError( "%s: No such distribution setting" % name ) from e if old is not None and not isinstance(old, sequence): raise DistutilsSetupError( name + ": this setting cannot be changed via include/exclude" ) elif old: setattr(self, name, [item for item in old if item not in value]) def _include_misc(self, name, value): """Handle 'include()' for list/tuple attrs without a special handler""" if not isinstance(value, sequence): raise DistutilsSetupError( "%s: setting must be a list (%r)" % (name, value) ) try: old = getattr(self, name) except AttributeError as e: raise DistutilsSetupError( "%s: No such distribution setting" % name ) from e if old is None: setattr(self, name, value) elif not isinstance(old, sequence): raise DistutilsSetupError( name + ": this setting cannot be changed via include/exclude" ) else: new = [item for item in value if item not in old] setattr(self, name, old + new) def exclude(self, *attrs): """Remove items from distribution that are named in keyword arguments For example, 'dist.exclude(py_modules=["x"])' would remove 'x' from the distribution's 'py_modules' attribute. Excluding packages uses the 'exclude_package()' method, so all of the package's contained packages, modules, and extensions are also excluded. Currently, this method only supports exclusion from attributes that are lists or tuples. If you need to add support for excluding from other attributes in this or a subclass, you can add an '_exclude_X' method, where 'X' is the name of the attribute. The method will be called with the value passed to 'exclude()'. So, 'dist.exclude(foo={"bar":"baz"})' will try to call 'dist._exclude_foo({"bar":"baz"})', which can then handle whatever special exclusion logic is needed. """ for k, v in attrs.items(): exclude = getattr(self, '_exclude_' + k, None) if exclude: exclude(v) else: self._exclude_misc(k, v) def _exclude_packages(self, packages): if not isinstance(packages, sequence): raise DistutilsSetupError( "packages: setting must be a list or tuple (%r)" % (packages,) ) list(map(self.exclude_package, packages)) def _parse_command_opts(self, parser, args): # Remove --with-X/--without-X options when processing command args self.global_options = self.__class__.global_options self.negative_opt = self.__class__.negative_opt # First, expand any aliases command = args[0] aliases = self.get_option_dict('aliases') while command in aliases: src, alias = aliases[command] del aliases[command] # ensure each alias can expand only once! import shlex args[:1] = shlex.split(alias, True) command = args[0] nargs = _Distribution._parse_command_opts(self, parser, args) # Handle commands that want to consume all remaining arguments cmd_class = self.get_command_class(command) if getattr(cmd_class, 'command_consumes_arguments', None): self.get_option_dict(command)['args'] = ("command line", nargs) if nargs is not None: return [] return nargs def get_cmdline_options(self): """Return a '{cmd: {opt:val}}' map of all command-line options Option names are all long, but do not include the leading '--', and contain dashes rather than underscores. If the option doesn't take an argument (e.g. '--quiet'), the 'val' is 'None'. Note that options provided by config files are intentionally excluded. """ d = {} for cmd, opts in self.command_options.items(): for opt, (src, val) in opts.items(): if src != "command line": continue opt = opt.replace('_', '-') if val == 0: cmdobj = self.get_command_obj(cmd) neg_opt = self.negative_opt.copy() neg_opt.update(getattr(cmdobj, 'negative_opt', {})) for neg, pos in neg_opt.items(): if pos == opt: opt = neg val = None break else: raise AssertionError("Shouldn't be able to get here") elif val == 1: val = None d.setdefault(cmd, {})[opt] = val return d def iter_distribution_names(self): """Yield all packages, modules, and extension names in distribution""" for pkg in self.packages or (): yield pkg for module in self.py_modules or (): yield module for ext in self.ext_modules or (): if isinstance(ext, tuple): name, buildinfo = ext else: name = ext.name if name.endswith('module'): name = name[:-6] yield name def handle_display_options(self, option_order): """If there were any non-global "display-only" options (--help-commands or the metadata display options) on the command line, display the requested info and return true; else return false. """ import sys if self.help_commands: return _Distribution.handle_display_options(self, option_order) # Stdout may be StringIO (e.g. in tests) if not isinstance(sys.stdout, io.TextIOWrapper): return _Distribution.handle_display_options(self, option_order) # Don't wrap stdout if utf-8 is already the encoding. Provides # workaround for #334. if sys.stdout.encoding.lower() in ('utf-8', 'utf8'): return _Distribution.handle_display_options(self, option_order) # Print metadata in UTF-8 no matter the platform encoding = sys.stdout.encoding errors = sys.stdout.errors newline = sys.platform != 'win32' and '\n' or None line_buffering = sys.stdout.line_buffering sys.stdout = io.TextIOWrapper( sys.stdout.detach(), 'utf-8', errors, newline, line_buffering) try: return _Distribution.handle_display_options(self, option_order) finally: sys.stdout = io.TextIOWrapper( sys.stdout.detach(), encoding, errors, newline, line_buffering) class DistDeprecationWarning(SetuptoolsDeprecationWarning): """Class for warning about deprecations in dist in setuptools. Not ignored by default, unlike DeprecationWarning.""" def unique_everseen(iterable, key=None): "List unique elements, preserving order. Remember all elements ever seen." # unique_everseen('AAAABBBCCDAABBB') --> A B C D # unique_everseen('ABBCcAD', str.lower) --> A B C D seen = set() seen_add = seen.add if key is None: for element in itertools.filterfalse(seen.__contains__, iterable): seen_add(element) yield element else: for element in iterable: k = key(element) if k not in seen: seen_add(k) yield element	39.82 KB	2025-09-21 11:46
📄 errors.py """setuptools.errors Provides exceptions used by setuptools modules. """ from distutils.errors import DistutilsError class RemovedCommandError(DistutilsError, RuntimeError): """Error used for commands that have been removed in setuptools. Since ``setuptools`` is built on ``distutils``, simply removing a command from ``setuptools`` will make the behavior fall back to ``distutils``; this error is raised if a command exists in ``distutils`` but has been actively removed in ``setuptools``. """	524 B	2025-09-21 11:46
📄 extension.py import re import functools import distutils.core import distutils.errors import distutils.extension from .monkey import get_unpatched def _have_cython(): """ Return True if Cython can be imported. """ cython_impl = 'Cython.Distutils.build_ext' try: # from (cython_impl) import build_ext __import__(cython_impl, fromlist=['build_ext']).build_ext return True except Exception: pass return False # for compatibility have_pyrex = _have_cython _Extension = get_unpatched(distutils.core.Extension) class Extension(_Extension): """Extension that uses '.c' files in place of '.pyx' files""" def __init__(self, name, sources, args, kw): # The args is needed for compatibility as calls may use positional # arguments. py_limited_api may be set only via keyword. self.py_limited_api = kw.pop("py_limited_api", False) _Extension.__init__(self, name, sources, args, *kw) def _convert_pyx_sources_to_lang(self): """ Replace sources with .pyx extensions to sources with the target language extension. This mechanism allows language authors to supply pre-converted sources but to prefer the .pyx sources. """ if _have_cython(): # the build has Cython, so allow it to compile the .pyx files return lang = self.language or '' target_ext = '.cpp' if lang.lower() == 'c++' else '.c' sub = functools.partial(re.sub, '.pyx$', target_ext) self.sources = list(map(sub, self.sources)) class Library(Extension): """Just like a regular Extension, but built as a library instead"""	1.64 KB	2025-09-21 11:46
📁 extern	-	2026-03-21 15:45
📄 glob.py """ Filename globbing utility. Mostly a copy of `glob` from Python 3.5. Changes include: * `yield from` and PEP3102 `` removed. Hidden files are not ignored. """ import os import re import fnmatch __all__ = ["glob", "iglob", "escape"] def glob(pathname, recursive=False): """Return a list of paths matching a pathname pattern. The pattern may contain simple shell-style wildcards a la fnmatch. However, unlike fnmatch, filenames starting with a dot are special cases that are not matched by '' and '?' patterns. If recursive is true, the pattern '' will match any files and zero or more directories and subdirectories. """ return list(iglob(pathname, recursive=recursive)) def iglob(pathname, recursive=False): """Return an iterator which yields the paths matching a pathname pattern. The pattern may contain simple shell-style wildcards a la fnmatch. However, unlike fnmatch, filenames starting with a dot are special cases that are not matched by '' and '?' patterns. If recursive is true, the pattern '*' will match any files and zero or more directories and subdirectories. """ it = _iglob(pathname, recursive) if recursive and _isrecursive(pathname): s = next(it) # skip empty string assert not s return it def _iglob(pathname, recursive): dirname, basename = os.path.split(pathname) glob_in_dir = glob2 if recursive and _isrecursive(basename) else glob1 if not has_magic(pathname): if basename: if os.path.lexists(pathname): yield pathname else: # Patterns ending with a slash should match only directories if os.path.isdir(dirname): yield pathname return if not dirname: yield from glob_in_dir(dirname, basename) return # `os.path.split()` returns the argument itself as a dirname if it is a # drive or UNC path. Prevent an infinite recursion if a drive or UNC path # contains magic characters (i.e. r'\\?\C:'). if dirname != pathname and has_magic(dirname): dirs = _iglob(dirname, recursive) else: dirs = [dirname] if not has_magic(basename): glob_in_dir = glob0 for dirname in dirs: for name in glob_in_dir(dirname, basename): yield os.path.join(dirname, name) # These 2 helper functions non-recursively glob inside a literal directory. # They return a list of basenames. `glob1` accepts a pattern while `glob0` # takes a literal basename (so it only has to check for its existence). def glob1(dirname, pattern): if not dirname: if isinstance(pattern, bytes): dirname = os.curdir.encode('ASCII') else: dirname = os.curdir try: names = os.listdir(dirname) except OSError: return [] return fnmatch.filter(names, pattern) def glob0(dirname, basename): if not basename: # `os.path.split()` returns an empty basename for paths ending with a # directory separator. 'qx/' should match only directories. if os.path.isdir(dirname): return [basename] else: if os.path.lexists(os.path.join(dirname, basename)): return [basename] return [] # This helper function recursively yields relative pathnames inside a literal # directory. def glob2(dirname, pattern): assert _isrecursive(pattern) yield pattern[:0] for x in _rlistdir(dirname): yield x # Recursively yields relative pathnames inside a literal directory. def _rlistdir(dirname): if not dirname: if isinstance(dirname, bytes): dirname = os.curdir.encode('ASCII') else: dirname = os.curdir try: names = os.listdir(dirname) except os.error: return for x in names: yield x path = os.path.join(dirname, x) if dirname else x for y in _rlistdir(path): yield os.path.join(x, y) magic_check = re.compile('([?[])') magic_check_bytes = re.compile(b'([?[])') def has_magic(s): if isinstance(s, bytes): match = magic_check_bytes.search(s) else: match = magic_check.search(s) return match is not None def _isrecursive(pattern): if isinstance(pattern, bytes): return pattern == b'' else: return pattern == '' def escape(pathname): """Escape all special characters. """ # Escaping is done by wrapping any of "*?[" between square brackets. # Metacharacters do not work in the drive part and shouldn't be escaped. drive, pathname = os.path.splitdrive(pathname) if isinstance(pathname, bytes): pathname = magic_check_bytes.sub(br'[\1]', pathname) else: pathname = magic_check.sub(r'[\1]', pathname) return drive + pathname	4.76 KB	2025-09-21 11:46
📄 installer.py import glob import os import subprocess import sys import tempfile from distutils import log from distutils.errors import DistutilsError import pkg_resources from setuptools.wheel import Wheel def _fixup_find_links(find_links): """Ensure find-links option end-up being a list of strings.""" if isinstance(find_links, str): return find_links.split() assert isinstance(find_links, (tuple, list)) return find_links def fetch_build_egg(dist, req): # noqa: C901 # is too complex (16) # FIXME """Fetch an egg needed for building. Use pip/wheel to fetch/build a wheel.""" # Warn if wheel is not available try: pkg_resources.get_distribution('wheel') except pkg_resources.DistributionNotFound: dist.announce('WARNING: The wheel package is not available.', log.WARN) # Ignore environment markers; if supplied, it is required. req = strip_marker(req) # Take easy_install options into account, but do not override relevant # pip environment variables (like PIP_INDEX_URL or PIP_QUIET); they'll # take precedence. opts = dist.get_option_dict('easy_install') if 'allow_hosts' in opts: raise DistutilsError('the `allow-hosts` option is not supported ' 'when using pip to install requirements.') quiet = 'PIP_QUIET' not in os.environ and 'PIP_VERBOSE' not in os.environ if 'PIP_INDEX_URL' in os.environ: index_url = None elif 'index_url' in opts: index_url = opts['index_url'][1] else: index_url = None find_links = ( _fixup_find_links(opts['find_links'][1])[:] if 'find_links' in opts else [] ) if dist.dependency_links: find_links.extend(dist.dependency_links) eggs_dir = os.path.realpath(dist.get_egg_cache_dir()) environment = pkg_resources.Environment() for egg_dist in pkg_resources.find_distributions(eggs_dir): if egg_dist in req and environment.can_add(egg_dist): return egg_dist with tempfile.TemporaryDirectory() as tmpdir: cmd = [ sys.executable, '-m', 'pip', '--disable-pip-version-check', 'wheel', '--no-deps', '-w', tmpdir, ] if quiet: cmd.append('--quiet') if index_url is not None: cmd.extend(('--index-url', index_url)) for link in find_links or []: cmd.extend(('--find-links', link)) # If requirement is a PEP 508 direct URL, directly pass # the URL to pip, as `req @ url` does not work on the # command line. cmd.append(req.url or str(req)) try: subprocess.check_call(cmd) except subprocess.CalledProcessError as e: raise DistutilsError(str(e)) from e wheel = Wheel(glob.glob(os.path.join(tmpdir, '*.whl'))[0]) dist_location = os.path.join(eggs_dir, wheel.egg_name()) wheel.install_as_egg(dist_location) dist_metadata = pkg_resources.PathMetadata( dist_location, os.path.join(dist_location, 'EGG-INFO')) dist = pkg_resources.Distribution.from_filename( dist_location, metadata=dist_metadata) return dist def strip_marker(req): """ Return a new requirement without the environment marker to avoid calling pip with something like `babel; extra == "i18n"`, which would always be ignored. """ # create a copy to avoid mutating the input req = pkg_resources.Requirement.parse(str(req)) req.marker = None return req	3.48 KB	2025-09-21 11:46
📄 launch.py """ Launch the Python script on the command line after setuptools is bootstrapped via import. """ # Note that setuptools gets imported implicitly by the # invocation of this script using python -m setuptools.launch import tokenize import sys def run(): """ Run the script in sys.argv[1] as if it had been invoked naturally. """ __builtins__ script_name = sys.argv[1] namespace = dict( __file__=script_name, __name__='__main__', __doc__=None, ) sys.argv[:] = sys.argv[1:] open_ = getattr(tokenize, 'open', open) with open_(script_name) as fid: script = fid.read() norm_script = script.replace('\\r\\n', '\\n') code = compile(norm_script, script_name, 'exec') exec(code, namespace) if __name__ == '__main__': run()	812 B	2025-09-21 11:46
📄 lib2to3_ex.py """ Customized Mixin2to3 support: - adds support for converting doctests """ import warnings from distutils.util import Mixin2to3 as _Mixin2to3 from distutils import log from lib2to3.refactor import RefactoringTool, get_fixers_from_package import setuptools from ._deprecation_warning import SetuptoolsDeprecationWarning class DistutilsRefactoringTool(RefactoringTool): def log_error(self, msg, args, kw): log.error(msg, args) def log_message(self, msg, args): log.info(msg, args) def log_debug(self, msg, args): log.debug(msg, args) class Mixin2to3(_Mixin2to3): def run_2to3(self, files, doctests=False): # See of the distribution option has been set, otherwise check the # setuptools default. if self.distribution.use_2to3 is not True: return if not files: return warnings.warn( "2to3 support is deprecated. If the project still " "requires Python 2 support, please migrate to " "a single-codebase solution or employ an " "independent conversion process.", SetuptoolsDeprecationWarning) log.info("Fixing " + " ".join(files)) self.__build_fixer_names() self.__exclude_fixers() if doctests: if setuptools.run_2to3_on_doctests: r = DistutilsRefactoringTool(self.fixer_names) r.refactor(files, write=True, doctests_only=True) else: _Mixin2to3.run_2to3(self, files) def __build_fixer_names(self): if self.fixer_names: return self.fixer_names = [] for p in setuptools.lib2to3_fixer_packages: self.fixer_names.extend(get_fixers_from_package(p)) if self.distribution.use_2to3_fixers is not None: for p in self.distribution.use_2to3_fixers: self.fixer_names.extend(get_fixers_from_package(p)) def __exclude_fixers(self): excluded_fixers = getattr(self, 'exclude_fixers', []) if self.distribution.use_2to3_exclude_fixers is not None: excluded_fixers.extend(self.distribution.use_2to3_exclude_fixers) for fixer_name in excluded_fixers: if fixer_name in self.fixer_names: self.fixer_names.remove(fixer_name)	2.28 KB	2025-09-21 11:46
📄 monkey.py """ Monkey patching of distutils. """ import sys import distutils.filelist import platform import types import functools from importlib import import_module import inspect import setuptools __all__ = [] """ Everything is private. Contact the project team if you think you need this functionality. """ def _get_mro(cls): """ Returns the bases classes for cls sorted by the MRO. Works around an issue on Jython where inspect.getmro will not return all base classes if multiple classes share the same name. Instead, this function will return a tuple containing the class itself, and the contents of cls.__bases__. See https://github.com/pypa/setuptools/issues/1024. """ if platform.python_implementation() == "Jython": return (cls,) + cls.__bases__ return inspect.getmro(cls) def get_unpatched(item): lookup = ( get_unpatched_class if isinstance(item, type) else get_unpatched_function if isinstance(item, types.FunctionType) else lambda item: None ) return lookup(item) def get_unpatched_class(cls): """Protect against re-patching the distutils if reloaded Also ensures that no other distutils extension monkeypatched the distutils first. """ external_bases = ( cls for cls in _get_mro(cls) if not cls.__module__.startswith('setuptools') ) base = next(external_bases) if not base.__module__.startswith('distutils'): msg = "distutils has already been patched by %r" % cls raise AssertionError(msg) return base def patch_all(): # we can't patch distutils.cmd, alas distutils.core.Command = setuptools.Command has_issue_12885 = sys.version_info <= (3, 5, 3) if has_issue_12885: # fix findall bug in distutils (http://bugs.python.org/issue12885) distutils.filelist.findall = setuptools.findall needs_warehouse = ( sys.version_info < (2, 7, 13) or (3, 4) < sys.version_info < (3, 4, 6) or (3, 5) < sys.version_info <= (3, 5, 3) ) if needs_warehouse: warehouse = 'https://upload.pypi.org/legacy/' distutils.config.PyPIRCCommand.DEFAULT_REPOSITORY = warehouse _patch_distribution_metadata() # Install Distribution throughout the distutils for module in distutils.dist, distutils.core, distutils.cmd: module.Distribution = setuptools.dist.Distribution # Install the patched Extension distutils.core.Extension = setuptools.extension.Extension distutils.extension.Extension = setuptools.extension.Extension if 'distutils.command.build_ext' in sys.modules: sys.modules['distutils.command.build_ext'].Extension = ( setuptools.extension.Extension ) patch_for_msvc_specialized_compiler() def _patch_distribution_metadata(): """Patch write_pkg_file and read_pkg_file for higher metadata standards""" for attr in ('write_pkg_file', 'read_pkg_file', 'get_metadata_version'): new_val = getattr(setuptools.dist, attr) setattr(distutils.dist.DistributionMetadata, attr, new_val) def patch_func(replacement, target_mod, func_name): """ Patch func_name in target_mod with replacement Important - original must be resolved by name to avoid patching an already patched function. """ original = getattr(target_mod, func_name) # set the 'unpatched' attribute on the replacement to # point to the original. vars(replacement).setdefault('unpatched', original) # replace the function in the original module setattr(target_mod, func_name, replacement) def get_unpatched_function(candidate): return getattr(candidate, 'unpatched') def patch_for_msvc_specialized_compiler(): """ Patch functions in distutils to use standalone Microsoft Visual C++ compilers. """ # import late to avoid circular imports on Python < 3.5 msvc = import_module('setuptools.msvc') if platform.system() != 'Windows': # Compilers only available on Microsoft Windows return def patch_params(mod_name, func_name): """ Prepare the parameters for patch_func to patch indicated function. """ repl_prefix = 'msvc9_' if 'msvc9' in mod_name else 'msvc14_' repl_name = repl_prefix + func_name.lstrip('_') repl = getattr(msvc, repl_name) mod = import_module(mod_name) if not hasattr(mod, func_name): raise ImportError(func_name) return repl, mod, func_name # Python 2.7 to 3.4 msvc9 = functools.partial(patch_params, 'distutils.msvc9compiler') # Python 3.5+ msvc14 = functools.partial(patch_params, 'distutils._msvccompiler') try: # Patch distutils.msvc9compiler patch_func(msvc9('find_vcvarsall')) patch_func(msvc9('query_vcvarsall')) except ImportError: pass try: # Patch distutils._msvccompiler._get_vc_env patch_func(msvc14('_get_vc_env')) except ImportError: pass try: # Patch distutils._msvccompiler.gen_lib_options for Numpy patch_func(msvc14('gen_lib_options')) except ImportError: pass	5.09 KB	2025-09-21 11:46
📄 msvc.py """ Improved support for Microsoft Visual C++ compilers. Known supported compilers: -------------------------- Microsoft Visual C++ 9.0: Microsoft Visual C++ Compiler for Python 2.7 (x86, amd64) Microsoft Windows SDK 6.1 (x86, x64, ia64) Microsoft Windows SDK 7.0 (x86, x64, ia64) Microsoft Visual C++ 10.0: Microsoft Windows SDK 7.1 (x86, x64, ia64) Microsoft Visual C++ 14.X: Microsoft Visual C++ Build Tools 2015 (x86, x64, arm) Microsoft Visual Studio Build Tools 2017 (x86, x64, arm, arm64) Microsoft Visual Studio Build Tools 2019 (x86, x64, arm, arm64) This may also support compilers shipped with compatible Visual Studio versions. """ import json from io import open from os import listdir, pathsep from os.path import join, isfile, isdir, dirname import sys import contextlib import platform import itertools import subprocess import distutils.errors from setuptools.extern.packaging.version import LegacyVersion from .monkey import get_unpatched if platform.system() == 'Windows': import winreg from os import environ else: # Mock winreg and environ so the module can be imported on this platform. class winreg: HKEY_USERS = None HKEY_CURRENT_USER = None HKEY_LOCAL_MACHINE = None HKEY_CLASSES_ROOT = None environ = dict() _msvc9_suppress_errors = ( # msvc9compiler isn't available on some platforms ImportError, # msvc9compiler raises DistutilsPlatformError in some # environments. See #1118. distutils.errors.DistutilsPlatformError, ) try: from distutils.msvc9compiler import Reg except _msvc9_suppress_errors: pass def msvc9_find_vcvarsall(version): """ Patched "distutils.msvc9compiler.find_vcvarsall" to use the standalone compiler build for Python (VCForPython / Microsoft Visual C++ Compiler for Python 2.7). Fall back to original behavior when the standalone compiler is not available. Redirect the path of "vcvarsall.bat". Parameters ---------- version: float Required Microsoft Visual C++ version. Return ------ str vcvarsall.bat path """ vc_base = r'Software\%sMicrosoft\DevDiv\VCForPython\%0.1f' key = vc_base % ('', version) try: # Per-user installs register the compiler path here productdir = Reg.get_value(key, "installdir") except KeyError: try: # All-user installs on a 64-bit system register here key = vc_base % ('Wow6432Node\\', version) productdir = Reg.get_value(key, "installdir") except KeyError: productdir = None if productdir: vcvarsall = join(productdir, "vcvarsall.bat") if isfile(vcvarsall): return vcvarsall return get_unpatched(msvc9_find_vcvarsall)(version) def msvc9_query_vcvarsall(ver, arch='x86', args, kwargs): """ Patched "distutils.msvc9compiler.query_vcvarsall" for support extra Microsoft Visual C++ 9.0 and 10.0 compilers. Set environment without use of "vcvarsall.bat". Parameters ---------- ver: float Required Microsoft Visual C++ version. arch: str Target architecture. Return ------ dict environment """ # Try to get environment from vcvarsall.bat (Classical way) try: orig = get_unpatched(msvc9_query_vcvarsall) return orig(ver, arch, args, *kwargs) except distutils.errors.DistutilsPlatformError: # Pass error if Vcvarsall.bat is missing pass except ValueError: # Pass error if environment not set after executing vcvarsall.bat pass # If error, try to set environment directly try: return EnvironmentInfo(arch, ver).return_env() except distutils.errors.DistutilsPlatformError as exc: _augment_exception(exc, ver, arch) raise def _msvc14_find_vc2015(): """Python 3.8 "distutils/_msvccompiler.py" backport""" try: key = winreg.OpenKey( winreg.HKEY_LOCAL_MACHINE, r"Software\Microsoft\VisualStudio\SxS\VC7", 0, winreg.KEY_READ \| winreg.KEY_WOW64_32KEY ) except OSError: return None, None best_version = 0 best_dir = None with key: for i in itertools.count(): try: v, vc_dir, vt = winreg.EnumValue(key, i) except OSError: break if v and vt == winreg.REG_SZ and isdir(vc_dir): try: version = int(float(v)) except (ValueError, TypeError): continue if version >= 14 and version > best_version: best_version, best_dir = version, vc_dir return best_version, best_dir def _msvc14_find_vc2017(): """Python 3.8 "distutils/_msvccompiler.py" backport Returns "15, path" based on the result of invoking vswhere.exe If no install is found, returns "None, None" The version is returned to avoid unnecessarily changing the function result. It may be ignored when the path is not None. If vswhere.exe is not available, by definition, VS 2017 is not installed. """ root = environ.get("ProgramFiles(x86)") or environ.get("ProgramFiles") if not root: return None, None try: path = subprocess.check_output([ join(root, "Microsoft Visual Studio", "Installer", "vswhere.exe"), "-latest", "-prerelease", "-requires", "Microsoft.VisualStudio.Component.VC.Tools.x86.x64", "-property", "installationPath", "-products", "", ]).decode(encoding="mbcs", errors="strict").strip() except (subprocess.CalledProcessError, OSError, UnicodeDecodeError): return None, None path = join(path, "VC", "Auxiliary", "Build") if isdir(path): return 15, path return None, None PLAT_SPEC_TO_RUNTIME = { 'x86': 'x86', 'x86_amd64': 'x64', 'x86_arm': 'arm', 'x86_arm64': 'arm64' } def _msvc14_find_vcvarsall(plat_spec): """Python 3.8 "distutils/_msvccompiler.py" backport""" _, best_dir = _msvc14_find_vc2017() vcruntime = None if plat_spec in PLAT_SPEC_TO_RUNTIME: vcruntime_plat = PLAT_SPEC_TO_RUNTIME[plat_spec] else: vcruntime_plat = 'x64' if 'amd64' in plat_spec else 'x86' if best_dir: vcredist = join(best_dir, "..", "..", "redist", "MSVC", "*", vcruntime_plat, "Microsoft.VC14.CRT", "vcruntime140.dll") try: import glob vcruntime = glob.glob(vcredist, recursive=True)[-1] except (ImportError, OSError, LookupError): vcruntime = None if not best_dir: best_version, best_dir = _msvc14_find_vc2015() if best_version: vcruntime = join(best_dir, 'redist', vcruntime_plat, "Microsoft.VC140.CRT", "vcruntime140.dll") if not best_dir: return None, None vcvarsall = join(best_dir, "vcvarsall.bat") if not isfile(vcvarsall): return None, None if not vcruntime or not isfile(vcruntime): vcruntime = None return vcvarsall, vcruntime def _msvc14_get_vc_env(plat_spec): """Python 3.8 "distutils/_msvccompiler.py" backport""" if "DISTUTILS_USE_SDK" in environ: return { key.lower(): value for key, value in environ.items() } vcvarsall, vcruntime = _msvc14_find_vcvarsall(plat_spec) if not vcvarsall: raise distutils.errors.DistutilsPlatformError( "Unable to find vcvarsall.bat" ) try: out = subprocess.check_output( 'cmd /u /c "{}" {} && set'.format(vcvarsall, plat_spec), stderr=subprocess.STDOUT, ).decode('utf-16le', errors='replace') except subprocess.CalledProcessError as exc: raise distutils.errors.DistutilsPlatformError( "Error executing {}".format(exc.cmd) ) from exc env = { key.lower(): value for key, _, value in (line.partition('=') for line in out.splitlines()) if key and value } if vcruntime: env['py_vcruntime_redist'] = vcruntime return env def msvc14_get_vc_env(plat_spec): """ Patched "distutils._msvccompiler._get_vc_env" for support extra Microsoft Visual C++ 14.X compilers. Set environment without use of "vcvarsall.bat". Parameters ---------- plat_spec: str Target architecture. Return ------ dict environment """ # Always use backport from CPython 3.8 try: return _msvc14_get_vc_env(plat_spec) except distutils.errors.DistutilsPlatformError as exc: _augment_exception(exc, 14.0) raise def msvc14_gen_lib_options(args, kwargs): """ Patched "distutils._msvccompiler.gen_lib_options" for fix compatibility between "numpy.distutils" and "distutils._msvccompiler" (for Numpy < 1.11.2) """ if "numpy.distutils" in sys.modules: import numpy as np if LegacyVersion(np.__version__) < LegacyVersion('1.11.2'): return np.distutils.ccompiler.gen_lib_options(args, *kwargs) return get_unpatched(msvc14_gen_lib_options)(args, kwargs) def _augment_exception(exc, version, arch=''): """ Add details to the exception message to help guide the user as to what action will resolve it. """ # Error if MSVC++ directory not found or environment not set message = exc.args[0] if "vcvarsall" in message.lower() or "visual c" in message.lower(): # Special error message if MSVC++ not installed tmpl = 'Microsoft Visual C++ {version:0.1f} or greater is required.' message = tmpl.format(locals()) msdownload = 'www.microsoft.com/download/details.aspx?id=%d' if version == 9.0: if arch.lower().find('ia64') > -1: # For VC++ 9.0, if IA64 support is needed, redirect user # to Windows SDK 7.0. # Note: No download link available from Microsoft. message += ' Get it with "Microsoft Windows SDK 7.0"' else: # For VC++ 9.0 redirect user to Vc++ for Python 2.7 : # This redirection link is maintained by Microsoft. # Contact vspython@microsoft.com if it needs updating. message += ' Get it from http://aka.ms/vcpython27' elif version == 10.0: # For VC++ 10.0 Redirect user to Windows SDK 7.1 message += ' Get it with "Microsoft Windows SDK 7.1": ' message += msdownload % 8279 elif version >= 14.0: # For VC++ 14.X Redirect user to latest Visual C++ Build Tools message += (' Get it with "Microsoft C++ Build Tools": ' r'https://visualstudio.microsoft.com' r'/visual-cpp-build-tools/') exc.args = (message, ) class PlatformInfo: """ Current and Target Architectures information. Parameters ---------- arch: str Target architecture. """ current_cpu = environ.get('processor_architecture', '').lower() def __init__(self, arch): self.arch = arch.lower().replace('x64', 'amd64') @property def target_cpu(self): """ Return Target CPU architecture. Return ------ str Target CPU """ return self.arch[self.arch.find('_') + 1:] def target_is_x86(self): """ Return True if target CPU is x86 32 bits.. Return ------ bool CPU is x86 32 bits """ return self.target_cpu == 'x86' def current_is_x86(self): """ Return True if current CPU is x86 32 bits.. Return ------ bool CPU is x86 32 bits """ return self.current_cpu == 'x86' def current_dir(self, hidex86=False, x64=False): """ Current platform specific subfolder. Parameters ---------- hidex86: bool return '' and not '\x86' if architecture is x86. x64: bool return '\x64' and not '\amd64' if architecture is amd64. Return ------ str subfolder: '\target', or '' (see hidex86 parameter) """ return ( '' if (self.current_cpu == 'x86' and hidex86) else r'\x64' if (self.current_cpu == 'amd64' and x64) else r'\%s' % self.current_cpu ) def target_dir(self, hidex86=False, x64=False): r""" Target platform specific subfolder. Parameters ---------- hidex86: bool return '' and not '\x86' if architecture is x86. x64: bool return '\x64' and not '\amd64' if architecture is amd64. Return ------ str subfolder: '\current', or '' (see hidex86 parameter) """ return ( '' if (self.target_cpu == 'x86' and hidex86) else r'\x64' if (self.target_cpu == 'amd64' and x64) else r'\%s' % self.target_cpu ) def cross_dir(self, forcex86=False): r""" Cross platform specific subfolder. Parameters ---------- forcex86: bool Use 'x86' as current architecture even if current architecture is not x86. Return ------ str subfolder: '' if target architecture is current architecture, '\current_target' if not. """ current = 'x86' if forcex86 else self.current_cpu return ( '' if self.target_cpu == current else self.target_dir().replace('\\', '\\%s_' % current) ) class RegistryInfo: """ Microsoft Visual Studio related registry information. Parameters ---------- platform_info: PlatformInfo "PlatformInfo" instance. """ HKEYS = (winreg.HKEY_USERS, winreg.HKEY_CURRENT_USER, winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CLASSES_ROOT) def __init__(self, platform_info): self.pi = platform_info @property def visualstudio(self): """ Microsoft Visual Studio root registry key. Return ------ str Registry key """ return 'VisualStudio' @property def sxs(self): """ Microsoft Visual Studio SxS registry key. Return ------ str Registry key """ return join(self.visualstudio, 'SxS') @property def vc(self): """ Microsoft Visual C++ VC7 registry key. Return ------ str Registry key """ return join(self.sxs, 'VC7') @property def vs(self): """ Microsoft Visual Studio VS7 registry key. Return ------ str Registry key """ return join(self.sxs, 'VS7') @property def vc_for_python(self): """ Microsoft Visual C++ for Python registry key. Return ------ str Registry key """ return r'DevDiv\VCForPython' @property def microsoft_sdk(self): """ Microsoft SDK registry key. Return ------ str Registry key """ return 'Microsoft SDKs' @property def windows_sdk(self): """ Microsoft Windows/Platform SDK registry key. Return ------ str Registry key """ return join(self.microsoft_sdk, 'Windows') @property def netfx_sdk(self): """ Microsoft .NET Framework SDK registry key. Return ------ str Registry key """ return join(self.microsoft_sdk, 'NETFXSDK') @property def windows_kits_roots(self): """ Microsoft Windows Kits Roots registry key. Return ------ str Registry key """ return r'Windows Kits\Installed Roots' def microsoft(self, key, x86=False): """ Return key in Microsoft software registry. Parameters ---------- key: str Registry key path where look. x86: str Force x86 software registry. Return ------ str Registry key """ node64 = '' if self.pi.current_is_x86() or x86 else 'Wow6432Node' return join('Software', node64, 'Microsoft', key) def lookup(self, key, name): """ Look for values in registry in Microsoft software registry. Parameters ---------- key: str Registry key path where look. name: str Value name to find. Return ------ str value """ key_read = winreg.KEY_READ openkey = winreg.OpenKey closekey = winreg.CloseKey ms = self.microsoft for hkey in self.HKEYS: bkey = None try: bkey = openkey(hkey, ms(key), 0, key_read) except (OSError, IOError): if not self.pi.current_is_x86(): try: bkey = openkey(hkey, ms(key, True), 0, key_read) except (OSError, IOError): continue else: continue try: return winreg.QueryValueEx(bkey, name)[0] except (OSError, IOError): pass finally: if bkey: closekey(bkey) class SystemInfo: """ Microsoft Windows and Visual Studio related system information. Parameters ---------- registry_info: RegistryInfo "RegistryInfo" instance. vc_ver: float Required Microsoft Visual C++ version. """ # Variables and properties in this class use originals CamelCase variables # names from Microsoft source files for more easy comparison. WinDir = environ.get('WinDir', '') ProgramFiles = environ.get('ProgramFiles', '') ProgramFilesx86 = environ.get('ProgramFiles(x86)', ProgramFiles) def __init__(self, registry_info, vc_ver=None): self.ri = registry_info self.pi = self.ri.pi self.known_vs_paths = self.find_programdata_vs_vers() # Except for VS15+, VC version is aligned with VS version self.vs_ver = self.vc_ver = ( vc_ver or self._find_latest_available_vs_ver()) def _find_latest_available_vs_ver(self): """ Find the latest VC version Return ------ float version """ reg_vc_vers = self.find_reg_vs_vers() if not (reg_vc_vers or self.known_vs_paths): raise distutils.errors.DistutilsPlatformError( 'No Microsoft Visual C++ version found') vc_vers = set(reg_vc_vers) vc_vers.update(self.known_vs_paths) return sorted(vc_vers)[-1] def find_reg_vs_vers(self): """ Find Microsoft Visual Studio versions available in registry. Return ------ list of float Versions """ ms = self.ri.microsoft vckeys = (self.ri.vc, self.ri.vc_for_python, self.ri.vs) vs_vers = [] for hkey, key in itertools.product(self.ri.HKEYS, vckeys): try: bkey = winreg.OpenKey(hkey, ms(key), 0, winreg.KEY_READ) except (OSError, IOError): continue with bkey: subkeys, values, _ = winreg.QueryInfoKey(bkey) for i in range(values): with contextlib.suppress(ValueError): ver = float(winreg.EnumValue(bkey, i)[0]) if ver not in vs_vers: vs_vers.append(ver) for i in range(subkeys): with contextlib.suppress(ValueError): ver = float(winreg.EnumKey(bkey, i)) if ver not in vs_vers: vs_vers.append(ver) return sorted(vs_vers) def find_programdata_vs_vers(self): r""" Find Visual studio 2017+ versions from information in "C:\ProgramData\Microsoft\VisualStudio\Packages\_Instances". Return ------ dict float version as key, path as value. """ vs_versions = {} instances_dir = \ r'C:\ProgramData\Microsoft\VisualStudio\Packages\_Instances' try: hashed_names = listdir(instances_dir) except (OSError, IOError): # Directory not exists with all Visual Studio versions return vs_versions for name in hashed_names: try: # Get VS installation path from "state.json" file state_path = join(instances_dir, name, 'state.json') with open(state_path, 'rt', encoding='utf-8') as state_file: state = json.load(state_file) vs_path = state['installationPath'] # Raises OSError if this VS installation does not contain VC listdir(join(vs_path, r'VC\Tools\MSVC')) # Store version and path vs_versions[self._as_float_version( state['installationVersion'])] = vs_path except (OSError, IOError, KeyError): # Skip if "state.json" file is missing or bad format continue return vs_versions @staticmethod def _as_float_version(version): """ Return a string version as a simplified float version (major.minor) Parameters ---------- version: str Version. Return ------ float version """ return float('.'.join(version.split('.')[:2])) @property def VSInstallDir(self): """ Microsoft Visual Studio directory. Return ------ str path """ # Default path default = join(self.ProgramFilesx86, 'Microsoft Visual Studio %0.1f' % self.vs_ver) # Try to get path from registry, if fail use default path return self.ri.lookup(self.ri.vs, '%0.1f' % self.vs_ver) or default @property def VCInstallDir(self): """ Microsoft Visual C++ directory. Return ------ str path """ path = self._guess_vc() or self._guess_vc_legacy() if not isdir(path): msg = 'Microsoft Visual C++ directory not found' raise distutils.errors.DistutilsPlatformError(msg) return path def _guess_vc(self): """ Locate Visual C++ for VS2017+. Return ------ str path """ if self.vs_ver <= 14.0: return '' try: # First search in known VS paths vs_dir = self.known_vs_paths[self.vs_ver] except KeyError: # Else, search with path from registry vs_dir = self.VSInstallDir guess_vc = join(vs_dir, r'VC\Tools\MSVC') # Subdir with VC exact version as name try: # Update the VC version with real one instead of VS version vc_ver = listdir(guess_vc)[-1] self.vc_ver = self._as_float_version(vc_ver) return join(guess_vc, vc_ver) except (OSError, IOError, IndexError): return '' def _guess_vc_legacy(self): """ Locate Visual C++ for versions prior to 2017. Return ------ str path """ default = join(self.ProgramFilesx86, r'Microsoft Visual Studio %0.1f\VC' % self.vs_ver) # Try to get "VC++ for Python" path from registry as default path reg_path = join(self.ri.vc_for_python, '%0.1f' % self.vs_ver) python_vc = self.ri.lookup(reg_path, 'installdir') default_vc = join(python_vc, 'VC') if python_vc else default # Try to get path from registry, if fail use default path return self.ri.lookup(self.ri.vc, '%0.1f' % self.vs_ver) or default_vc @property def WindowsSdkVersion(self): """ Microsoft Windows SDK versions for specified MSVC++ version. Return ------ tuple of str versions """ if self.vs_ver <= 9.0: return '7.0', '6.1', '6.0a' elif self.vs_ver == 10.0: return '7.1', '7.0a' elif self.vs_ver == 11.0: return '8.0', '8.0a' elif self.vs_ver == 12.0: return '8.1', '8.1a' elif self.vs_ver >= 14.0: return '10.0', '8.1' @property def WindowsSdkLastVersion(self): """ Microsoft Windows SDK last version. Return ------ str version """ return self._use_last_dir_name(join(self.WindowsSdkDir, 'lib')) @property # noqa: C901 def WindowsSdkDir(self): # noqa: C901 # is too complex (12) # FIXME """ Microsoft Windows SDK directory. Return ------ str path """ sdkdir = '' for ver in self.WindowsSdkVersion: # Try to get it from registry loc = join(self.ri.windows_sdk, 'v%s' % ver) sdkdir = self.ri.lookup(loc, 'installationfolder') if sdkdir: break if not sdkdir or not isdir(sdkdir): # Try to get "VC++ for Python" version from registry path = join(self.ri.vc_for_python, '%0.1f' % self.vc_ver) install_base = self.ri.lookup(path, 'installdir') if install_base: sdkdir = join(install_base, 'WinSDK') if not sdkdir or not isdir(sdkdir): # If fail, use default new path for ver in self.WindowsSdkVersion: intver = ver[:ver.rfind('.')] path = r'Microsoft SDKs\Windows Kits\%s' % intver d = join(self.ProgramFiles, path) if isdir(d): sdkdir = d if not sdkdir or not isdir(sdkdir): # If fail, use default old path for ver in self.WindowsSdkVersion: path = r'Microsoft SDKs\Windows\v%s' % ver d = join(self.ProgramFiles, path) if isdir(d): sdkdir = d if not sdkdir: # If fail, use Platform SDK sdkdir = join(self.VCInstallDir, 'PlatformSDK') return sdkdir @property def WindowsSDKExecutablePath(self): """ Microsoft Windows SDK executable directory. Return ------ str path """ # Find WinSDK NetFx Tools registry dir name if self.vs_ver <= 11.0: netfxver = 35 arch = '' else: netfxver = 40 hidex86 = True if self.vs_ver <= 12.0 else False arch = self.pi.current_dir(x64=True, hidex86=hidex86) fx = 'WinSDK-NetFx%dTools%s' % (netfxver, arch.replace('\\', '-')) # list all possibles registry paths regpaths = [] if self.vs_ver >= 14.0: for ver in self.NetFxSdkVersion: regpaths += [join(self.ri.netfx_sdk, ver, fx)] for ver in self.WindowsSdkVersion: regpaths += [join(self.ri.windows_sdk, 'v%sA' % ver, fx)] # Return installation folder from the more recent path for path in regpaths: execpath = self.ri.lookup(path, 'installationfolder') if execpath: return execpath @property def FSharpInstallDir(self): """ Microsoft Visual F# directory. Return ------ str path """ path = join(self.ri.visualstudio, r'%0.1f\Setup\F#' % self.vs_ver) return self.ri.lookup(path, 'productdir') or '' @property def UniversalCRTSdkDir(self): """ Microsoft Universal CRT SDK directory. Return ------ str path """ # Set Kit Roots versions for specified MSVC++ version vers = ('10', '81') if self.vs_ver >= 14.0 else () # Find path of the more recent Kit for ver in vers: sdkdir = self.ri.lookup(self.ri.windows_kits_roots, 'kitsroot%s' % ver) if sdkdir: return sdkdir or '' @property def UniversalCRTSdkLastVersion(self): """ Microsoft Universal C Runtime SDK last version. Return ------ str version """ return self._use_last_dir_name(join(self.UniversalCRTSdkDir, 'lib')) @property def NetFxSdkVersion(self): """ Microsoft .NET Framework SDK versions. Return ------ tuple of str versions """ # Set FxSdk versions for specified VS version return (('4.7.2', '4.7.1', '4.7', '4.6.2', '4.6.1', '4.6', '4.5.2', '4.5.1', '4.5') if self.vs_ver >= 14.0 else ()) @property def NetFxSdkDir(self): """ Microsoft .NET Framework SDK directory. Return ------ str path """ sdkdir = '' for ver in self.NetFxSdkVersion: loc = join(self.ri.netfx_sdk, ver) sdkdir = self.ri.lookup(loc, 'kitsinstallationfolder') if sdkdir: break return sdkdir @property def FrameworkDir32(self): """ Microsoft .NET Framework 32bit directory. Return ------ str path """ # Default path guess_fw = join(self.WinDir, r'Microsoft.NET\Framework') # Try to get path from registry, if fail use default path return self.ri.lookup(self.ri.vc, 'frameworkdir32') or guess_fw @property def FrameworkDir64(self): """ Microsoft .NET Framework 64bit directory. Return ------ str path """ # Default path guess_fw = join(self.WinDir, r'Microsoft.NET\Framework64') # Try to get path from registry, if fail use default path return self.ri.lookup(self.ri.vc, 'frameworkdir64') or guess_fw @property def FrameworkVersion32(self): """ Microsoft .NET Framework 32bit versions. Return ------ tuple of str versions """ return self._find_dot_net_versions(32) @property def FrameworkVersion64(self): """ Microsoft .NET Framework 64bit versions. Return ------ tuple of str versions """ return self._find_dot_net_versions(64) def _find_dot_net_versions(self, bits): """ Find Microsoft .NET Framework versions. Parameters ---------- bits: int Platform number of bits: 32 or 64. Return ------ tuple of str versions """ # Find actual .NET version in registry reg_ver = self.ri.lookup(self.ri.vc, 'frameworkver%d' % bits) dot_net_dir = getattr(self, 'FrameworkDir%d' % bits) ver = reg_ver or self._use_last_dir_name(dot_net_dir, 'v') or '' # Set .NET versions for specified MSVC++ version if self.vs_ver >= 12.0: return ver, 'v4.0' elif self.vs_ver >= 10.0: return 'v4.0.30319' if ver.lower()[:2] != 'v4' else ver, 'v3.5' elif self.vs_ver == 9.0: return 'v3.5', 'v2.0.50727' elif self.vs_ver == 8.0: return 'v3.0', 'v2.0.50727' @staticmethod def _use_last_dir_name(path, prefix=''): """ Return name of the last dir in path or '' if no dir found. Parameters ---------- path: str Use dirs in this path prefix: str Use only dirs starting by this prefix Return ------ str name """ matching_dirs = ( dir_name for dir_name in reversed(listdir(path)) if isdir(join(path, dir_name)) and dir_name.startswith(prefix) ) return next(matching_dirs, None) or '' class EnvironmentInfo: """ Return environment variables for specified Microsoft Visual C++ version and platform : Lib, Include, Path and libpath. This function is compatible with Microsoft Visual C++ 9.0 to 14.X. Script created by analysing Microsoft environment configuration files like "vcvars[...].bat", "SetEnv.Cmd", "vcbuildtools.bat", ... Parameters ---------- arch: str Target architecture. vc_ver: float Required Microsoft Visual C++ version. If not set, autodetect the last version. vc_min_ver: float Minimum Microsoft Visual C++ version. """ # Variables and properties in this class use originals CamelCase variables # names from Microsoft source files for more easy comparison. def __init__(self, arch, vc_ver=None, vc_min_ver=0): self.pi = PlatformInfo(arch) self.ri = RegistryInfo(self.pi) self.si = SystemInfo(self.ri, vc_ver) if self.vc_ver < vc_min_ver: err = 'No suitable Microsoft Visual C++ version found' raise distutils.errors.DistutilsPlatformError(err) @property def vs_ver(self): """ Microsoft Visual Studio. Return ------ float version """ return self.si.vs_ver @property def vc_ver(self): """ Microsoft Visual C++ version. Return ------ float version """ return self.si.vc_ver @property def VSTools(self): """ Microsoft Visual Studio Tools. Return ------ list of str paths """ paths = [r'Common7\IDE', r'Common7\Tools'] if self.vs_ver >= 14.0: arch_subdir = self.pi.current_dir(hidex86=True, x64=True) paths += [r'Common7\IDE\CommonExtensions\Microsoft\TestWindow'] paths += [r'Team Tools\Performance Tools'] paths += [r'Team Tools\Performance Tools%s' % arch_subdir] return [join(self.si.VSInstallDir, path) for path in paths] @property def VCIncludes(self): """ Microsoft Visual C++ & Microsoft Foundation Class Includes. Return ------ list of str paths """ return [join(self.si.VCInstallDir, 'Include'), join(self.si.VCInstallDir, r'ATLMFC\Include')] @property def VCLibraries(self): """ Microsoft Visual C++ & Microsoft Foundation Class Libraries. Return ------ list of str paths """ if self.vs_ver >= 15.0: arch_subdir = self.pi.target_dir(x64=True) else: arch_subdir = self.pi.target_dir(hidex86=True) paths = ['Lib%s' % arch_subdir, r'ATLMFC\Lib%s' % arch_subdir] if self.vs_ver >= 14.0: paths += [r'Lib\store%s' % arch_subdir] return [join(self.si.VCInstallDir, path) for path in paths] @property def VCStoreRefs(self): """ Microsoft Visual C++ store references Libraries. Return ------ list of str paths """ if self.vs_ver < 14.0: return [] return [join(self.si.VCInstallDir, r'Lib\store\references')] @property def VCTools(self): """ Microsoft Visual C++ Tools. Return ------ list of str paths """ si = self.si tools = [join(si.VCInstallDir, 'VCPackages')] forcex86 = True if self.vs_ver <= 10.0 else False arch_subdir = self.pi.cross_dir(forcex86) if arch_subdir: tools += [join(si.VCInstallDir, 'Bin%s' % arch_subdir)] if self.vs_ver == 14.0: path = 'Bin%s' % self.pi.current_dir(hidex86=True) tools += [join(si.VCInstallDir, path)] elif self.vs_ver >= 15.0: host_dir = (r'bin\HostX86%s' if self.pi.current_is_x86() else r'bin\HostX64%s') tools += [join( si.VCInstallDir, host_dir % self.pi.target_dir(x64=True))] if self.pi.current_cpu != self.pi.target_cpu: tools += [join( si.VCInstallDir, host_dir % self.pi.current_dir(x64=True))] else: tools += [join(si.VCInstallDir, 'Bin')] return tools @property def OSLibraries(self): """ Microsoft Windows SDK Libraries. Return ------ list of str paths """ if self.vs_ver <= 10.0: arch_subdir = self.pi.target_dir(hidex86=True, x64=True) return [join(self.si.WindowsSdkDir, 'Lib%s' % arch_subdir)] else: arch_subdir = self.pi.target_dir(x64=True) lib = join(self.si.WindowsSdkDir, 'lib') libver = self._sdk_subdir return [join(lib, '%sum%s' % (libver, arch_subdir))] @property def OSIncludes(self): """ Microsoft Windows SDK Include. Return ------ list of str paths """ include = join(self.si.WindowsSdkDir, 'include') if self.vs_ver <= 10.0: return [include, join(include, 'gl')] else: if self.vs_ver >= 14.0: sdkver = self._sdk_subdir else: sdkver = '' return [join(include, '%sshared' % sdkver), join(include, '%sum' % sdkver), join(include, '%swinrt' % sdkver)] @property def OSLibpath(self): """ Microsoft Windows SDK Libraries Paths. Return ------ list of str paths """ ref = join(self.si.WindowsSdkDir, 'References') libpath = [] if self.vs_ver <= 9.0: libpath += self.OSLibraries if self.vs_ver >= 11.0: libpath += [join(ref, r'CommonConfiguration\Neutral')] if self.vs_ver >= 14.0: libpath += [ ref, join(self.si.WindowsSdkDir, 'UnionMetadata'), join( ref, 'Windows.Foundation.UniversalApiContract', '1.0.0.0'), join(ref, 'Windows.Foundation.FoundationContract', '1.0.0.0'), join( ref, 'Windows.Networking.Connectivity.WwanContract', '1.0.0.0'), join( self.si.WindowsSdkDir, 'ExtensionSDKs', 'Microsoft.VCLibs', '%0.1f' % self.vs_ver, 'References', 'CommonConfiguration', 'neutral'), ] return libpath @property def SdkTools(self): """ Microsoft Windows SDK Tools. Return ------ list of str paths """ return list(self._sdk_tools()) def _sdk_tools(self): """ Microsoft Windows SDK Tools paths generator. Return ------ generator of str paths """ if self.vs_ver < 15.0: bin_dir = 'Bin' if self.vs_ver <= 11.0 else r'Bin\x86' yield join(self.si.WindowsSdkDir, bin_dir) if not self.pi.current_is_x86(): arch_subdir = self.pi.current_dir(x64=True) path = 'Bin%s' % arch_subdir yield join(self.si.WindowsSdkDir, path) if self.vs_ver in (10.0, 11.0): if self.pi.target_is_x86(): arch_subdir = '' else: arch_subdir = self.pi.current_dir(hidex86=True, x64=True) path = r'Bin\NETFX 4.0 Tools%s' % arch_subdir yield join(self.si.WindowsSdkDir, path) elif self.vs_ver >= 15.0: path = join(self.si.WindowsSdkDir, 'Bin') arch_subdir = self.pi.current_dir(x64=True) sdkver = self.si.WindowsSdkLastVersion yield join(path, '%s%s' % (sdkver, arch_subdir)) if self.si.WindowsSDKExecutablePath: yield self.si.WindowsSDKExecutablePath @property def _sdk_subdir(self): """ Microsoft Windows SDK version subdir. Return ------ str subdir """ ucrtver = self.si.WindowsSdkLastVersion return ('%s\\' % ucrtver) if ucrtver else '' @property def SdkSetup(self): """ Microsoft Windows SDK Setup. Return ------ list of str paths """ if self.vs_ver > 9.0: return [] return [join(self.si.WindowsSdkDir, 'Setup')] @property def FxTools(self): """ Microsoft .NET Framework Tools. Return ------ list of str paths """ pi = self.pi si = self.si if self.vs_ver <= 10.0: include32 = True include64 = not pi.target_is_x86() and not pi.current_is_x86() else: include32 = pi.target_is_x86() or pi.current_is_x86() include64 = pi.current_cpu == 'amd64' or pi.target_cpu == 'amd64' tools = [] if include32: tools += [join(si.FrameworkDir32, ver) for ver in si.FrameworkVersion32] if include64: tools += [join(si.FrameworkDir64, ver) for ver in si.FrameworkVersion64] return tools @property def NetFxSDKLibraries(self): """ Microsoft .Net Framework SDK Libraries. Return ------ list of str paths """ if self.vs_ver < 14.0 or not self.si.NetFxSdkDir: return [] arch_subdir = self.pi.target_dir(x64=True) return [join(self.si.NetFxSdkDir, r'lib\um%s' % arch_subdir)] @property def NetFxSDKIncludes(self): """ Microsoft .Net Framework SDK Includes. Return ------ list of str paths """ if self.vs_ver < 14.0 or not self.si.NetFxSdkDir: return [] return [join(self.si.NetFxSdkDir, r'include\um')] @property def VsTDb(self): """ Microsoft Visual Studio Team System Database. Return ------ list of str paths """ return [join(self.si.VSInstallDir, r'VSTSDB\Deploy')] @property def MSBuild(self): """ Microsoft Build Engine. Return ------ list of str paths """ if self.vs_ver < 12.0: return [] elif self.vs_ver < 15.0: base_path = self.si.ProgramFilesx86 arch_subdir = self.pi.current_dir(hidex86=True) else: base_path = self.si.VSInstallDir arch_subdir = '' path = r'MSBuild\%0.1f\bin%s' % (self.vs_ver, arch_subdir) build = [join(base_path, path)] if self.vs_ver >= 15.0: # Add Roslyn C# & Visual Basic Compiler build += [join(base_path, path, 'Roslyn')] return build @property def HTMLHelpWorkshop(self): """ Microsoft HTML Help Workshop. Return ------ list of str paths """ if self.vs_ver < 11.0: return [] return [join(self.si.ProgramFilesx86, 'HTML Help Workshop')] @property def UCRTLibraries(self): """ Microsoft Universal C Runtime SDK Libraries. Return ------ list of str paths """ if self.vs_ver < 14.0: return [] arch_subdir = self.pi.target_dir(x64=True) lib = join(self.si.UniversalCRTSdkDir, 'lib') ucrtver = self._ucrt_subdir return [join(lib, '%sucrt%s' % (ucrtver, arch_subdir))] @property def UCRTIncludes(self): """ Microsoft Universal C Runtime SDK Include. Return ------ list of str paths """ if self.vs_ver < 14.0: return [] include = join(self.si.UniversalCRTSdkDir, 'include') return [join(include, '%sucrt' % self._ucrt_subdir)] @property def _ucrt_subdir(self): """ Microsoft Universal C Runtime SDK version subdir. Return ------ str subdir """ ucrtver = self.si.UniversalCRTSdkLastVersion return ('%s\\' % ucrtver) if ucrtver else '' @property def FSharp(self): """ Microsoft Visual F#. Return ------ list of str paths """ if 11.0 > self.vs_ver > 12.0: return [] return [self.si.FSharpInstallDir] @property def VCRuntimeRedist(self): """ Microsoft Visual C++ runtime redistributable dll. Return ------ str path """ vcruntime = 'vcruntime%d0.dll' % self.vc_ver arch_subdir = self.pi.target_dir(x64=True).strip('\\') # Installation prefixes candidates prefixes = [] tools_path = self.si.VCInstallDir redist_path = dirname(tools_path.replace(r'\Tools', r'\Redist')) if isdir(redist_path): # Redist version may not be exactly the same as tools redist_path = join(redist_path, listdir(redist_path)[-1]) prefixes += [redist_path, join(redist_path, 'onecore')] prefixes += [join(tools_path, 'redist')] # VS14 legacy path # CRT directory crt_dirs = ('Microsoft.VC%d.CRT' % (self.vc_ver * 10), # Sometime store in directory with VS version instead of VC 'Microsoft.VC%d.CRT' % (int(self.vs_ver) * 10)) # vcruntime path for prefix, crt_dir in itertools.product(prefixes, crt_dirs): path = join(prefix, arch_subdir, crt_dir, vcruntime) if isfile(path): return path def return_env(self, exists=True): """ Return environment dict. Parameters ---------- exists: bool It True, only return existing paths. Return ------ dict environment """ env = dict( include=self._build_paths('include', [self.VCIncludes, self.OSIncludes, self.UCRTIncludes, self.NetFxSDKIncludes], exists), lib=self._build_paths('lib', [self.VCLibraries, self.OSLibraries, self.FxTools, self.UCRTLibraries, self.NetFxSDKLibraries], exists), libpath=self._build_paths('libpath', [self.VCLibraries, self.FxTools, self.VCStoreRefs, self.OSLibpath], exists), path=self._build_paths('path', [self.VCTools, self.VSTools, self.VsTDb, self.SdkTools, self.SdkSetup, self.FxTools, self.MSBuild, self.HTMLHelpWorkshop, self.FSharp], exists), ) if self.vs_ver >= 14 and isfile(self.VCRuntimeRedist): env['py_vcruntime_redist'] = self.VCRuntimeRedist return env def _build_paths(self, name, spec_path_lists, exists): """ Given an environment variable name and specified paths, return a pathsep-separated string of paths containing unique, extant, directories from those paths and from the environment variable. Raise an error if no paths are resolved. Parameters ---------- name: str Environment variable name spec_path_lists: list of str Paths exists: bool It True, only return existing paths. Return ------ str Pathsep-separated paths """ # flatten spec_path_lists spec_paths = itertools.chain.from_iterable(spec_path_lists) env_paths = environ.get(name, '').split(pathsep) paths = itertools.chain(spec_paths, env_paths) extant_paths = list(filter(isdir, paths)) if exists else paths if not extant_paths: msg = "%s environment variable is empty" % name.upper() raise distutils.errors.DistutilsPlatformError(msg) unique_paths = self._unique_everseen(extant_paths) return pathsep.join(unique_paths) # from Python docs @staticmethod def _unique_everseen(iterable, key=None): """ List unique elements, preserving order. Remember all elements ever seen. _unique_everseen('AAAABBBCCDAABBB') --> A B C D _unique_everseen('ABBCcAD', str.lower) --> A B C D """ seen = set() seen_add = seen.add if key is None: for element in itertools.filterfalse(seen.__contains__, iterable): seen_add(element) yield element else: for element in iterable: k = key(element) if k not in seen: seen_add(k) yield element	49.93 KB	2025-09-21 11:46
📄 namespaces.py import os from distutils import log import itertools flatten = itertools.chain.from_iterable class Installer: nspkg_ext = '-nspkg.pth' def install_namespaces(self): nsp = self._get_all_ns_packages() if not nsp: return filename, ext = os.path.splitext(self._get_target()) filename += self.nspkg_ext self.outputs.append(filename) log.info("Installing %s", filename) lines = map(self._gen_nspkg_line, nsp) if self.dry_run: # always generate the lines, even in dry run list(lines) return with open(filename, 'wt') as f: f.writelines(lines) def uninstall_namespaces(self): filename, ext = os.path.splitext(self._get_target()) filename += self.nspkg_ext if not os.path.exists(filename): return log.info("Removing %s", filename) os.remove(filename) def _get_target(self): return self.target _nspkg_tmpl = ( "import sys, types, os", "has_mfs = sys.version_info > (3, 5)", "p = os.path.join(%(root)s, *%(pth)r)", "importlib = has_mfs and __import__('importlib.util')", "has_mfs and __import__('importlib.machinery')", ( "m = has_mfs and " "sys.modules.setdefault(%(pkg)r, " "importlib.util.module_from_spec(" "importlib.machinery.PathFinder.find_spec(%(pkg)r, " "[os.path.dirname(p)])))" ), ( "m = m or " "sys.modules.setdefault(%(pkg)r, types.ModuleType(%(pkg)r))" ), "mp = (m or []) and m.__dict__.setdefault('__path__',[])", "(p not in mp) and mp.append(p)", ) "lines for the namespace installer" _nspkg_tmpl_multi = ( 'm and setattr(sys.modules[%(parent)r], %(child)r, m)', ) "additional line(s) when a parent package is indicated" def _get_root(self): return "sys._getframe(1).f_locals['sitedir']" def _gen_nspkg_line(self, pkg): pth = tuple(pkg.split('.')) root = self._get_root() tmpl_lines = self._nspkg_tmpl parent, sep, child = pkg.rpartition('.') if parent: tmpl_lines += self._nspkg_tmpl_multi return ';'.join(tmpl_lines) % locals() + '\n' def _get_all_ns_packages(self): """Return sorted list of all package namespaces""" pkgs = self.distribution.namespace_packages or [] return sorted(flatten(map(self._pkg_names, pkgs))) @staticmethod def _pkg_names(pkg): """ Given a namespace package, yield the components of that package. >>> names = Installer._pkg_names('a.b.c') >>> set(names) == set(['a', 'a.b', 'a.b.c']) True """ parts = pkg.split('.') while parts: yield '.'.join(parts) parts.pop() class DevelopInstaller(Installer): def _get_root(self): return repr(str(self.egg_path)) def _get_target(self): return self.egg_link	3.02 KB	2025-09-21 11:46
📄 package_index.py """PyPI and direct package downloading""" import subprocess import sys import os import re import io import shutil import socket import base64 import hashlib import itertools import warnings import configparser import html import http.client import urllib.parse import urllib.request import urllib.error from functools import wraps import setuptools from pkg_resources import ( CHECKOUT_DIST, Distribution, BINARY_DIST, normalize_path, SOURCE_DIST, Environment, find_distributions, safe_name, safe_version, to_filename, Requirement, DEVELOP_DIST, EGG_DIST, ) from setuptools import ssl_support from distutils import log from distutils.errors import DistutilsError from fnmatch import translate from setuptools.wheel import Wheel EGG_FRAGMENT = re.compile(r'^egg=([-A-Za-z0-9_.+!]+)$') HREF = re.compile(r"""href\s=\s['"]?([^'"> ]+)""", re.I) PYPI_MD5 = re.compile( r'<a href="([^"#]+)">([^<]+)</a>\n\s+$<a (?:title="MD5 hash"\n\s+)' r'href="[^?]+\?:action=show_md5&digest=([0-9a-f]{32})">md5</a>$' ) URL_SCHEME = re.compile('([-+.a-z0-9]{2,}):', re.I).match EXTENSIONS = ".tar.gz .tar.bz2 .tar .zip .tgz".split() __all__ = [ 'PackageIndex', 'distros_for_url', 'parse_bdist_wininst', 'interpret_distro_name', ] _SOCKET_TIMEOUT = 15 _tmpl = "setuptools/{setuptools.__version__} Python-urllib/{py_major}" user_agent = _tmpl.format( py_major='{}.{}'.format(sys.version_info), setuptools=setuptools) def parse_requirement_arg(spec): try: return Requirement.parse(spec) except ValueError as e: raise DistutilsError( "Not a URL, existing file, or requirement spec: %r" % (spec,) ) from e def parse_bdist_wininst(name): """Return (base,pyversion) or (None,None) for possible .exe name""" lower = name.lower() base, py_ver, plat = None, None, None if lower.endswith('.exe'): if lower.endswith('.win32.exe'): base = name[:-10] plat = 'win32' elif lower.startswith('.win32-py', -16): py_ver = name[-7:-4] base = name[:-16] plat = 'win32' elif lower.endswith('.win-amd64.exe'): base = name[:-14] plat = 'win-amd64' elif lower.startswith('.win-amd64-py', -20): py_ver = name[-7:-4] base = name[:-20] plat = 'win-amd64' return base, py_ver, plat def egg_info_for_url(url): parts = urllib.parse.urlparse(url) scheme, server, path, parameters, query, fragment = parts base = urllib.parse.unquote(path.split('/')[-1]) if server == 'sourceforge.net' and base == 'download': # XXX Yuck base = urllib.parse.unquote(path.split('/')[-2]) if '#' in base: base, fragment = base.split('#', 1) return base, fragment def distros_for_url(url, metadata=None): """Yield egg or source distribution objects that might be found at a URL""" base, fragment = egg_info_for_url(url) for dist in distros_for_location(url, base, metadata): yield dist if fragment: match = EGG_FRAGMENT.match(fragment) if match: for dist in interpret_distro_name( url, match.group(1), metadata, precedence=CHECKOUT_DIST ): yield dist def distros_for_location(location, basename, metadata=None): """Yield egg or source distribution objects based on basename""" if basename.endswith('.egg.zip'): basename = basename[:-4] # strip the .zip if basename.endswith('.egg') and '-' in basename: # only one, unambiguous interpretation return [Distribution.from_location(location, basename, metadata)] if basename.endswith('.whl') and '-' in basename: wheel = Wheel(basename) if not wheel.is_compatible(): return [] return [Distribution( location=location, project_name=wheel.project_name, version=wheel.version, # Increase priority over eggs. precedence=EGG_DIST + 1, )] if basename.endswith('.exe'): win_base, py_ver, platform = parse_bdist_wininst(basename) if win_base is not None: return interpret_distro_name( location, win_base, metadata, py_ver, BINARY_DIST, platform ) # Try source distro extensions (.zip, .tgz, etc.) # for ext in EXTENSIONS: if basename.endswith(ext): basename = basename[:-len(ext)] return interpret_distro_name(location, basename, metadata) return [] # no extension matched def distros_for_filename(filename, metadata=None): """Yield possible egg or source distribution objects based on a filename""" return distros_for_location( normalize_path(filename), os.path.basename(filename), metadata ) def interpret_distro_name( location, basename, metadata, py_version=None, precedence=SOURCE_DIST, platform=None ): """Generate alternative interpretations of a source distro name Note: if `location` is a filesystem filename, you should call ``pkg_resources.normalize_path()`` on it before passing it to this routine! """ # Generate alternative interpretations of a source distro name # Because some packages are ambiguous as to name/versions split # e.g. "adns-python-1.1.0", "egenix-mx-commercial", etc. # So, we generate each possible interepretation (e.g. "adns, python-1.1.0" # "adns-python, 1.1.0", and "adns-python-1.1.0, no version"). In practice, # the spurious interpretations should be ignored, because in the event # there's also an "adns" package, the spurious "python-1.1.0" version will # compare lower than any numeric version number, and is therefore unlikely # to match a request for it. It's still a potential problem, though, and # in the long run PyPI and the distutils should go for "safe" names and # versions in distribution archive names (sdist and bdist). parts = basename.split('-') if not py_version and any(re.match(r'py\d\.\d$', p) for p in parts[2:]): # it is a bdist_dumb, not an sdist -- bail out return for p in range(1, len(parts) + 1): yield Distribution( location, metadata, '-'.join(parts[:p]), '-'.join(parts[p:]), py_version=py_version, precedence=precedence, platform=platform ) # From Python 2.7 docs def unique_everseen(iterable, key=None): "List unique elements, preserving order. Remember all elements ever seen." # unique_everseen('AAAABBBCCDAABBB') --> A B C D # unique_everseen('ABBCcAD', str.lower) --> A B C D seen = set() seen_add = seen.add if key is None: for element in itertools.filterfalse(seen.__contains__, iterable): seen_add(element) yield element else: for element in iterable: k = key(element) if k not in seen: seen_add(k) yield element def unique_values(func): """ Wrap a function returning an iterable such that the resulting iterable only ever yields unique items. """ @wraps(func) def wrapper(args, *kwargs): return unique_everseen(func(args, *kwargs)) return wrapper REL = re.compile(r"""<([^>]\srel\s{0,10}=\s{0,10}['"]?([^'" >]+)[^>])>""", re.I) # this line is here to fix emacs' cruddy broken syntax highlighting @unique_values def find_external_links(url, page): """Find rel="homepage" and rel="download" links in `page`, yielding URLs""" for match in REL.finditer(page): tag, rel = match.groups() rels = set(map(str.strip, rel.lower().split(','))) if 'homepage' in rels or 'download' in rels: for match in HREF.finditer(tag): yield urllib.parse.urljoin(url, htmldecode(match.group(1))) for tag in ("<th>Home Page", "<th>Download URL"): pos = page.find(tag) if pos != -1: match = HREF.search(page, pos) if match: yield urllib.parse.urljoin(url, htmldecode(match.group(1))) class ContentChecker: """ A null content checker that defines the interface for checking content """ def feed(self, block): """ Feed a block of data to the hash. """ return def is_valid(self): """ Check the hash. Return False if validation fails. """ return True def report(self, reporter, template): """ Call reporter with information about the checker (hash name) substituted into the template. """ return class HashChecker(ContentChecker): pattern = re.compile( r'(?P<hash_name>sha1\|sha224\|sha384\|sha256\|sha512\|md5)=' r'(?P<expected>[a-f0-9]+)' ) def __init__(self, hash_name, expected): self.hash_name = hash_name self.hash = hashlib.new(hash_name) self.expected = expected @classmethod def from_url(cls, url): "Construct a (possibly null) ContentChecker from a URL" fragment = urllib.parse.urlparse(url)[-1] if not fragment: return ContentChecker() match = cls.pattern.search(fragment) if not match: return ContentChecker() return cls(match.groupdict()) def feed(self, block): self.hash.update(block) def is_valid(self): return self.hash.hexdigest() == self.expected def report(self, reporter, template): msg = template % self.hash_name return reporter(msg) class PackageIndex(Environment): """A distribution index that scans web pages for download URLs""" def __init__( self, index_url="https://pypi.org/simple/", hosts=('',), ca_bundle=None, verify_ssl=True, args, kw ): Environment.__init__(self, args, *kw) self.index_url = index_url + "/" [:not index_url.endswith('/')] self.scanned_urls = {} self.fetched_urls = {} self.package_pages = {} self.allows = re.compile('\|'.join(map(translate, hosts))).match self.to_scan = [] use_ssl = ( verify_ssl and ssl_support.is_available and (ca_bundle or ssl_support.find_ca_bundle()) ) if use_ssl: self.opener = ssl_support.opener_for(ca_bundle) else: self.opener = urllib.request.urlopen # FIXME: 'PackageIndex.process_url' is too complex (14) def process_url(self, url, retrieve=False): # noqa: C901 """Evaluate a URL as a possible download, and maybe retrieve it""" if url in self.scanned_urls and not retrieve: return self.scanned_urls[url] = True if not URL_SCHEME(url): self.process_filename(url) return else: dists = list(distros_for_url(url)) if dists: if not self.url_ok(url): return self.debug("Found link: %s", url) if dists or not retrieve or url in self.fetched_urls: list(map(self.add, dists)) return # don't need the actual page if not self.url_ok(url): self.fetched_urls[url] = True return self.info("Reading %s", url) self.fetched_urls[url] = True # prevent multiple fetch attempts tmpl = "Download error on %s: %%s -- Some packages may not be found!" f = self.open_url(url, tmpl % url) if f is None: return if isinstance(f, urllib.error.HTTPError) and f.code == 401: self.info("Authentication error: %s" % f.msg) self.fetched_urls[f.url] = True if 'html' not in f.headers.get('content-type', '').lower(): f.close() # not html, we can't process it return base = f.url # handle redirects page = f.read() if not isinstance(page, str): # In Python 3 and got bytes but want str. if isinstance(f, urllib.error.HTTPError): # Errors have no charset, assume latin1: charset = 'latin-1' else: charset = f.headers.get_param('charset') or 'latin-1' page = page.decode(charset, "ignore") f.close() for match in HREF.finditer(page): link = urllib.parse.urljoin(base, htmldecode(match.group(1))) self.process_url(link) if url.startswith(self.index_url) and getattr(f, 'code', None) != 404: page = self.process_index(url, page) def process_filename(self, fn, nested=False): # process filenames or directories if not os.path.exists(fn): self.warn("Not found: %s", fn) return if os.path.isdir(fn) and not nested: path = os.path.realpath(fn) for item in os.listdir(path): self.process_filename(os.path.join(path, item), True) dists = distros_for_filename(fn) if dists: self.debug("Found: %s", fn) list(map(self.add, dists)) def url_ok(self, url, fatal=False): s = URL_SCHEME(url) is_file = s and s.group(1).lower() == 'file' if is_file or self.allows(urllib.parse.urlparse(url)[1]): return True msg = ( "\nNote: Bypassing %s (disallowed host; see " "http://bit.ly/2hrImnY for details).\n") if fatal: raise DistutilsError(msg % url) else: self.warn(msg, url) def scan_egg_links(self, search_path): dirs = filter(os.path.isdir, search_path) egg_links = ( (path, entry) for path in dirs for entry in os.listdir(path) if entry.endswith('.egg-link') ) list(itertools.starmap(self.scan_egg_link, egg_links)) def scan_egg_link(self, path, entry): with open(os.path.join(path, entry)) as raw_lines: # filter non-empty lines lines = list(filter(None, map(str.strip, raw_lines))) if len(lines) != 2: # format is not recognized; punt return egg_path, setup_path = lines for dist in find_distributions(os.path.join(path, egg_path)): dist.location = os.path.join(path, lines) dist.precedence = SOURCE_DIST self.add(dist) def _scan(self, link): # Process a URL to see if it's for a package page NO_MATCH_SENTINEL = None, None if not link.startswith(self.index_url): return NO_MATCH_SENTINEL parts = list(map( urllib.parse.unquote, link[len(self.index_url):].split('/') )) if len(parts) != 2 or '#' in parts[1]: return NO_MATCH_SENTINEL # it's a package page, sanitize and index it pkg = safe_name(parts[0]) ver = safe_version(parts[1]) self.package_pages.setdefault(pkg.lower(), {})[link] = True return to_filename(pkg), to_filename(ver) def process_index(self, url, page): """Process the contents of a PyPI page""" # process an index page into the package-page index for match in HREF.finditer(page): try: self._scan(urllib.parse.urljoin(url, htmldecode(match.group(1)))) except ValueError: pass pkg, ver = self._scan(url) # ensure this page is in the page index if not pkg: return "" # no sense double-scanning non-package pages # process individual package page for new_url in find_external_links(url, page): # Process the found URL base, frag = egg_info_for_url(new_url) if base.endswith('.py') and not frag: if ver: new_url += '#egg=%s-%s' % (pkg, ver) else: self.need_version_info(url) self.scan_url(new_url) return PYPI_MD5.sub( lambda m: '<a href="%s#md5=%s">%s</a>' % m.group(1, 3, 2), page ) def need_version_info(self, url): self.scan_all( "Page at %s links to .py file(s) without version info; an index " "scan is required.", url ) def scan_all(self, msg=None, args): if self.index_url not in self.fetched_urls: if msg: self.warn(msg, args) self.info( "Scanning index of all packages (this may take a while)" ) self.scan_url(self.index_url) def find_packages(self, requirement): self.scan_url(self.index_url + requirement.unsafe_name + '/') if not self.package_pages.get(requirement.key): # Fall back to safe version of the name self.scan_url(self.index_url + requirement.project_name + '/') if not self.package_pages.get(requirement.key): # We couldn't find the target package, so search the index page too self.not_found_in_index(requirement) for url in list(self.package_pages.get(requirement.key, ())): # scan each page that might be related to the desired package self.scan_url(url) def obtain(self, requirement, installer=None): self.prescan() self.find_packages(requirement) for dist in self[requirement.key]: if dist in requirement: return dist self.debug("%s does not match %s", requirement, dist) return super(PackageIndex, self).obtain(requirement, installer) def check_hash(self, checker, filename, tfp): """ checker is a ContentChecker """ checker.report( self.debug, "Validating %%s checksum for %s" % filename) if not checker.is_valid(): tfp.close() os.unlink(filename) raise DistutilsError( "%s validation failed for %s; " "possible download problem?" % (checker.hash.name, os.path.basename(filename)) ) def add_find_links(self, urls): """Add `urls` to the list that will be prescanned for searches""" for url in urls: if ( self.to_scan is None # if we have already "gone online" or not URL_SCHEME(url) # or it's a local file/directory or url.startswith('file:') or list(distros_for_url(url)) # or a direct package link ): # then go ahead and process it now self.scan_url(url) else: # otherwise, defer retrieval till later self.to_scan.append(url) def prescan(self): """Scan urls scheduled for prescanning (e.g. --find-links)""" if self.to_scan: list(map(self.scan_url, self.to_scan)) self.to_scan = None # from now on, go ahead and process immediately def not_found_in_index(self, requirement): if self[requirement.key]: # we've seen at least one distro meth, msg = self.info, "Couldn't retrieve index page for %r" else: # no distros seen for this name, might be misspelled meth, msg = ( self.warn, "Couldn't find index page for %r (maybe misspelled?)") meth(msg, requirement.unsafe_name) self.scan_all() def download(self, spec, tmpdir): """Locate and/or download `spec` to `tmpdir`, returning a local path `spec` may be a ``Requirement`` object, or a string containing a URL, an existing local filename, or a project/version requirement spec (i.e. the string form of a ``Requirement`` object). If it is the URL of a .py file with an unambiguous ``#egg=name-version`` tag (i.e., one that escapes ``-`` as ``_`` throughout), a trivial ``setup.py`` is automatically created alongside the downloaded file. If `spec` is a ``Requirement`` object or a string containing a project/version requirement spec, this method returns the location of a matching distribution (possibly after downloading it to `tmpdir`). If `spec` is a locally existing file or directory name, it is simply returned unchanged. If `spec` is a URL, it is downloaded to a subpath of `tmpdir`, and the local filename is returned. Various errors may be raised if a problem occurs during downloading. """ if not isinstance(spec, Requirement): scheme = URL_SCHEME(spec) if scheme: # It's a url, download it to tmpdir found = self._download_url(scheme.group(1), spec, tmpdir) base, fragment = egg_info_for_url(spec) if base.endswith('.py'): found = self.gen_setup(found, fragment, tmpdir) return found elif os.path.exists(spec): # Existing file or directory, just return it return spec else: spec = parse_requirement_arg(spec) return getattr(self.fetch_distribution(spec, tmpdir), 'location', None) def fetch_distribution( # noqa: C901 # is too complex (14) # FIXME self, requirement, tmpdir, force_scan=False, source=False, develop_ok=False, local_index=None): """Obtain a distribution suitable for fulfilling `requirement` `requirement` must be a ``pkg_resources.Requirement`` instance. If necessary, or if the `force_scan` flag is set, the requirement is searched for in the (online) package index as well as the locally installed packages. If a distribution matching `requirement` is found, the returned distribution's ``location`` is the value you would have gotten from calling the ``download()`` method with the matching distribution's URL or filename. If no matching distribution is found, ``None`` is returned. If the `source` flag is set, only source distributions and source checkout links will be considered. Unless the `develop_ok` flag is set, development and system eggs (i.e., those using the ``.egg-info`` format) will be ignored. """ # process a Requirement self.info("Searching for %s", requirement) skipped = {} dist = None def find(req, env=None): if env is None: env = self # Find a matching distribution; may be called more than once for dist in env[req.key]: if dist.precedence == DEVELOP_DIST and not develop_ok: if dist not in skipped: self.warn( "Skipping development or system egg: %s", dist, ) skipped[dist] = 1 continue test = ( dist in req and (dist.precedence <= SOURCE_DIST or not source) ) if test: loc = self.download(dist.location, tmpdir) dist.download_location = loc if os.path.exists(dist.download_location): return dist if force_scan: self.prescan() self.find_packages(requirement) dist = find(requirement) if not dist and local_index is not None: dist = find(requirement, local_index) if dist is None: if self.to_scan is not None: self.prescan() dist = find(requirement) if dist is None and not force_scan: self.find_packages(requirement) dist = find(requirement) if dist is None: self.warn( "No local packages or working download links found for %s%s", (source and "a source distribution of " or ""), requirement, ) else: self.info("Best match: %s", dist) return dist.clone(location=dist.download_location) def fetch(self, requirement, tmpdir, force_scan=False, source=False): """Obtain a file suitable for fulfilling `requirement` DEPRECATED; use the ``fetch_distribution()`` method now instead. For backward compatibility, this routine is identical but returns the ``location`` of the downloaded distribution instead of a distribution object. """ dist = self.fetch_distribution(requirement, tmpdir, force_scan, source) if dist is not None: return dist.location return None def gen_setup(self, filename, fragment, tmpdir): match = EGG_FRAGMENT.match(fragment) dists = match and [ d for d in interpret_distro_name(filename, match.group(1), None) if d.version ] or [] if len(dists) == 1: # unambiguous ``#egg`` fragment basename = os.path.basename(filename) # Make sure the file has been downloaded to the temp dir. if os.path.dirname(filename) != tmpdir: dst = os.path.join(tmpdir, basename) from setuptools.command.easy_install import samefile if not samefile(filename, dst): shutil.copy2(filename, dst) filename = dst with open(os.path.join(tmpdir, 'setup.py'), 'w') as file: file.write( "from setuptools import setup\n" "setup(name=%r, version=%r, py_modules=[%r])\n" % ( dists[0].project_name, dists[0].version, os.path.splitext(basename)[0] ) ) return filename elif match: raise DistutilsError( "Can't unambiguously interpret project/version identifier %r; " "any dashes in the name or version should be escaped using " "underscores. %r" % (fragment, dists) ) else: raise DistutilsError( "Can't process plain .py files without an '#egg=name-version'" " suffix to enable automatic setup script generation." ) dl_blocksize = 8192 def _download_to(self, url, filename): self.info("Downloading %s", url) # Download the file fp = None try: checker = HashChecker.from_url(url) fp = self.open_url(url) if isinstance(fp, urllib.error.HTTPError): raise DistutilsError( "Can't download %s: %s %s" % (url, fp.code, fp.msg) ) headers = fp.info() blocknum = 0 bs = self.dl_blocksize size = -1 if "content-length" in headers: # Some servers return multiple Content-Length headers :( sizes = headers.get_all('Content-Length') size = max(map(int, sizes)) self.reporthook(url, filename, blocknum, bs, size) with open(filename, 'wb') as tfp: while True: block = fp.read(bs) if block: checker.feed(block) tfp.write(block) blocknum += 1 self.reporthook(url, filename, blocknum, bs, size) else: break self.check_hash(checker, filename, tfp) return headers finally: if fp: fp.close() def reporthook(self, url, filename, blocknum, blksize, size): pass # no-op # FIXME: def open_url(self, url, warning=None): # noqa: C901 # is too complex (12) if url.startswith('file:'): return local_open(url) try: return open_with_auth(url, self.opener) except (ValueError, http.client.InvalidURL) as v: msg = ' '.join([str(arg) for arg in v.args]) if warning: self.warn(warning, msg) else: raise DistutilsError('%s %s' % (url, msg)) from v except urllib.error.HTTPError as v: return v except urllib.error.URLError as v: if warning: self.warn(warning, v.reason) else: raise DistutilsError("Download error for %s: %s" % (url, v.reason)) from v except http.client.BadStatusLine as v: if warning: self.warn(warning, v.line) else: raise DistutilsError( '%s returned a bad status line. The server might be ' 'down, %s' % (url, v.line) ) from v except (http.client.HTTPException, socket.error) as v: if warning: self.warn(warning, v) else: raise DistutilsError("Download error for %s: %s" % (url, v)) from v def _download_url(self, scheme, url, tmpdir): # Determine download filename # name, fragment = egg_info_for_url(url) if name: while '..' in name: name = name.replace('..', '.').replace('\\', '_') else: name = "__downloaded__" # default if URL has no path contents if name.endswith('.egg.zip'): name = name[:-4] # strip the extra .zip before download filename = os.path.join(tmpdir, name) # ensure path resolves within the tmpdir if not filename.startswith(str(tmpdir)): raise ValueError(f"Invalid filename {filename}") # Download the file # if scheme == 'svn' or scheme.startswith('svn+'): return self._download_svn(url, filename) elif scheme == 'git' or scheme.startswith('git+'): return self._download_git(url, filename) elif scheme.startswith('hg+'): return self._download_hg(url, filename) elif scheme == 'file': return urllib.request.url2pathname(urllib.parse.urlparse(url)[2]) else: self.url_ok(url, True) # raises error if not allowed return self._attempt_download(url, filename) def scan_url(self, url): self.process_url(url, True) def _attempt_download(self, url, filename): headers = self._download_to(url, filename) if 'html' in headers.get('content-type', '').lower(): return self._download_html(url, headers, filename) else: return filename def _download_html(self, url, headers, filename): file = open(filename) for line in file: if line.strip(): # Check for a subversion index page if re.search(r'<title>([^- ]+ - )?Revision \d+:', line): # it's a subversion index page: file.close() os.unlink(filename) return self._download_svn(url, filename) break # not an index page file.close() os.unlink(filename) raise DistutilsError("Unexpected HTML page found at " + url) def _download_svn(self, url, filename): warnings.warn("SVN download support is deprecated", UserWarning) url = url.split('#', 1)[0] # remove any fragment for svn's sake creds = [] if url.lower().startswith('svn:') and '@' in url: scheme, netloc, path, p, q, f = urllib.parse.urlparse(url) if not netloc and path.startswith('//') and '/' in path[2:]: netloc, path = path[2:].split('/', 1) auth, host = _splituser(netloc) if auth: if ':' in auth: user, pw = auth.split(':', 1) creds = [f"--username={user}", f"--password={pw}"] else: creds = [f"--username={auth}"] netloc = host parts = scheme, netloc, url, p, q, f url = urllib.parse.urlunparse(parts) self.info("Doing subversion checkout from %s to %s", url, filename) subprocess.check_call(["svn", "checkout"] + creds + ["-q", url, filename]) return filename @staticmethod def _vcs_split_rev_from_url(url, pop_prefix=False): scheme, netloc, path, query, frag = urllib.parse.urlsplit(url) scheme = scheme.split('+', 1)[-1] # Some fragment identification fails path = path.split('#', 1)[0] rev = None if '@' in path: path, rev = path.rsplit('@', 1) # Also, discard fragment url = urllib.parse.urlunsplit((scheme, netloc, path, query, '')) return url, rev def _download_git(self, url, filename): filename = filename.split('#', 1)[0] url, rev = self._vcs_split_rev_from_url(url, pop_prefix=True) self.info("Doing git clone from %s to %s", url, filename) subprocess.check_call(["git", "clone", "--quiet", url, filename]) if rev is not None: self.info("Checking out %s", rev) subprocess.check_call(["git", "-C", filename, "checkout", "--quiet", rev]) return filename def _download_hg(self, url, filename): filename = filename.split('#', 1)[0] url, rev = self._vcs_split_rev_from_url(url, pop_prefix=True) self.info("Doing hg clone from %s to %s", url, filename) subprocess.check_call(["hg", "clone", "--quiet", url, filename]) if rev is not None: self.info("Updating to %s", rev) subprocess.check_call(["hg", "--cwd", filename, "up", "-C", "-r", rev, "-q"]) return filename def debug(self, msg, args): log.debug(msg, args) def info(self, msg, args): log.info(msg, args) def warn(self, msg, args): log.warn(msg, args) # This pattern matches a character entity reference (a decimal numeric # references, a hexadecimal numeric reference, or a named reference). entity_sub = re.compile(r'&(#(\d+\|x[\da-fA-F]+)\|[\w.:-]+);?').sub def decode_entity(match): what = match.group(0) return html.unescape(what) def htmldecode(text): """ Decode HTML entities in the given text. >>> htmldecode( ... 'https://../package_name-0.1.2.tar.gz' ... '?tokena=A&tokenb=B">package_name-0.1.2.tar.gz') 'https://../package_name-0.1.2.tar.gz?tokena=A&tokenb=B">package_name-0.1.2.tar.gz' """ return entity_sub(decode_entity, text) def socket_timeout(timeout=15): def _socket_timeout(func): def _socket_timeout(args, kwargs): old_timeout = socket.getdefaulttimeout() socket.setdefaulttimeout(timeout) try: return func(args, *kwargs) finally: socket.setdefaulttimeout(old_timeout) return _socket_timeout return _socket_timeout def _encode_auth(auth): """ Encode auth from a URL suitable for an HTTP header. >>> str(_encode_auth('username%3Apassword')) 'dXNlcm5hbWU6cGFzc3dvcmQ=' Long auth strings should not cause a newline to be inserted. >>> long_auth = 'username:' + 'password'10 >>> chr(10) in str(_encode_auth(long_auth)) False """ auth_s = urllib.parse.unquote(auth) # convert to bytes auth_bytes = auth_s.encode() encoded_bytes = base64.b64encode(auth_bytes) # convert back to a string encoded = encoded_bytes.decode() # strip the trailing carriage return return encoded.replace('\n', '') class Credential: """ A username/password pair. Use like a namedtuple. """ def __init__(self, username, password): self.username = username self.password = password def __iter__(self): yield self.username yield self.password def __str__(self): return '%(username)s:%(password)s' % vars(self) class PyPIConfig(configparser.RawConfigParser): def __init__(self): """ Load from ~/.pypirc """ defaults = dict.fromkeys(['username', 'password', 'repository'], '') configparser.RawConfigParser.__init__(self, defaults) rc = os.path.join(os.path.expanduser('~'), '.pypirc') if os.path.exists(rc): self.read(rc) @property def creds_by_repository(self): sections_with_repositories = [ section for section in self.sections() if self.get(section, 'repository').strip() ] return dict(map(self._get_repo_cred, sections_with_repositories)) def _get_repo_cred(self, section): repo = self.get(section, 'repository').strip() return repo, Credential( self.get(section, 'username').strip(), self.get(section, 'password').strip(), ) def find_credential(self, url): """ If the URL indicated appears to be a repository defined in this config, return the credential for that repository. """ for repository, cred in self.creds_by_repository.items(): if url.startswith(repository): return cred def open_with_auth(url, opener=urllib.request.urlopen): """Open a urllib2 request, handling HTTP authentication""" parsed = urllib.parse.urlparse(url) scheme, netloc, path, params, query, frag = parsed # Double scheme does not raise on macOS as revealed by a # failing test. We would expect "nonnumeric port". Refs #20. if netloc.endswith(':'): raise http.client.InvalidURL("nonnumeric port: ''") if scheme in ('http', 'https'): auth, address = _splituser(netloc) else: auth = None if not auth: cred = PyPIConfig().find_credential(url) if cred: auth = str(cred) info = cred.username, url log.info('Authenticating as %s for %s (from .pypirc)', *info) if auth: auth = "Basic " + _encode_auth(auth) parts = scheme, address, path, params, query, frag new_url = urllib.parse.urlunparse(parts) request = urllib.request.Request(new_url) request.add_header("Authorization", auth) else: request = urllib.request.Request(url) request.add_header('User-Agent', user_agent) fp = opener(request) if auth: # Put authentication info back into request URL if same host, # so that links found on the page will work s2, h2, path2, param2, query2, frag2 = urllib.parse.urlparse(fp.url) if s2 == scheme and h2 == address: parts = s2, netloc, path2, param2, query2, frag2 fp.url = urllib.parse.urlunparse(parts) return fp # copy of urllib.parse._splituser from Python 3.8 def _splituser(host): """splituser('user[:passwd]@host[:port]') --> 'user[:passwd]', 'host[:port]'.""" user, delim, host = host.rpartition('@') return (user if delim else None), host # adding a timeout to avoid freezing package_index open_with_auth = socket_timeout(_SOCKET_TIMEOUT)(open_with_auth) def fix_sf_url(url): return url # backward compatibility def local_open(url): """Read a local path, with special support for directories""" scheme, server, path, param, query, frag = urllib.parse.urlparse(url) filename = urllib.request.url2pathname(path) if os.path.isfile(filename): return urllib.request.urlopen(url) elif path.endswith('/') and os.path.isdir(filename): files = [] for f in os.listdir(filename): filepath = os.path.join(filename, f) if f == 'index.html': with open(filepath, 'r') as fp: body = fp.read() break elif os.path.isdir(filepath): f += '/' files.append('<a href="{name}">{name}</a>'.format(name=f)) else: tmpl = ( "<html><head><title>{url}</title>" "</head><body>{files}</body></html>") body = tmpl.format(url=url, files='\n'.join(files)) status, message = 200, "OK" else: status, message, body = 404, "Path not found", "Not found" headers = {'content-type': 'text/html'} body_stream = io.StringIO(body) return urllib.error.HTTPError(url, status, message, headers, body_stream)	39.92 KB	2025-09-21 11:46
📄 py34compat.py import importlib try: import importlib.util except ImportError: pass try: module_from_spec = importlib.util.module_from_spec except AttributeError: def module_from_spec(spec): return spec.loader.load_module(spec.name)	245 B	2025-09-21 11:46
📄 sandbox.py import os import sys import tempfile import operator import functools import itertools import re import contextlib import pickle import textwrap import builtins import pkg_resources from distutils.errors import DistutilsError from pkg_resources import working_set if sys.platform.startswith('java'): import org.python.modules.posix.PosixModule as _os else: _os = sys.modules[os.name] try: _file = file except NameError: _file = None _open = open __all__ = [ "AbstractSandbox", "DirectorySandbox", "SandboxViolation", "run_setup", ] def _execfile(filename, globals, locals=None): """ Python 3 implementation of execfile. """ mode = 'rb' with open(filename, mode) as stream: script = stream.read() if locals is None: locals = globals code = compile(script, filename, 'exec') exec(code, globals, locals) @contextlib.contextmanager def save_argv(repl=None): saved = sys.argv[:] if repl is not None: sys.argv[:] = repl try: yield saved finally: sys.argv[:] = saved @contextlib.contextmanager def save_path(): saved = sys.path[:] try: yield saved finally: sys.path[:] = saved @contextlib.contextmanager def override_temp(replacement): """ Monkey-patch tempfile.tempdir with replacement, ensuring it exists """ os.makedirs(replacement, exist_ok=True) saved = tempfile.tempdir tempfile.tempdir = replacement try: yield finally: tempfile.tempdir = saved @contextlib.contextmanager def pushd(target): saved = os.getcwd() os.chdir(target) try: yield saved finally: os.chdir(saved) class UnpickleableException(Exception): """ An exception representing another Exception that could not be pickled. """ @staticmethod def dump(type, exc): """ Always return a dumped (pickled) type and exc. If exc can't be pickled, wrap it in UnpickleableException first. """ try: return pickle.dumps(type), pickle.dumps(exc) except Exception: # get UnpickleableException inside the sandbox from setuptools.sandbox import UnpickleableException as cls return cls.dump(cls, cls(repr(exc))) class ExceptionSaver: """ A Context Manager that will save an exception, serialized, and restore it later. """ def __enter__(self): return self def __exit__(self, type, exc, tb): if not exc: return # dump the exception self._saved = UnpickleableException.dump(type, exc) self._tb = tb # suppress the exception return True def resume(self): "restore and re-raise any exception" if '_saved' not in vars(self): return type, exc = map(pickle.loads, self._saved) raise exc.with_traceback(self._tb) @contextlib.contextmanager def save_modules(): """ Context in which imported modules are saved. Translates exceptions internal to the context into the equivalent exception outside the context. """ saved = sys.modules.copy() with ExceptionSaver() as saved_exc: yield saved sys.modules.update(saved) # remove any modules imported since del_modules = ( mod_name for mod_name in sys.modules if mod_name not in saved # exclude any encodings modules. See #285 and not mod_name.startswith('encodings.') ) _clear_modules(del_modules) saved_exc.resume() def _clear_modules(module_names): for mod_name in list(module_names): del sys.modules[mod_name] @contextlib.contextmanager def save_pkg_resources_state(): saved = pkg_resources.__getstate__() try: yield saved finally: pkg_resources.__setstate__(saved) @contextlib.contextmanager def setup_context(setup_dir): temp_dir = os.path.join(setup_dir, 'temp') with save_pkg_resources_state(): with save_modules(): with save_path(): hide_setuptools() with save_argv(): with override_temp(temp_dir): with pushd(setup_dir): # ensure setuptools commands are available __import__('setuptools') yield _MODULES_TO_HIDE = { 'setuptools', 'distutils', 'pkg_resources', 'Cython', '_distutils_hack', } def _needs_hiding(mod_name): """ >>> _needs_hiding('setuptools') True >>> _needs_hiding('pkg_resources') True >>> _needs_hiding('setuptools_plugin') False >>> _needs_hiding('setuptools.__init__') True >>> _needs_hiding('distutils') True >>> _needs_hiding('os') False >>> _needs_hiding('Cython') True """ base_module = mod_name.split('.', 1)[0] return base_module in _MODULES_TO_HIDE def hide_setuptools(): """ Remove references to setuptools' modules from sys.modules to allow the invocation to import the most appropriate setuptools. This technique is necessary to avoid issues such as #315 where setuptools upgrading itself would fail to find a function declared in the metadata. """ _distutils_hack = sys.modules.get('_distutils_hack', None) if _distutils_hack is not None: _distutils_hack.remove_shim() modules = filter(_needs_hiding, sys.modules) _clear_modules(modules) def run_setup(setup_script, args): """Run a distutils setup script, sandboxed in its directory""" setup_dir = os.path.abspath(os.path.dirname(setup_script)) with setup_context(setup_dir): try: sys.argv[:] = [setup_script] + list(args) sys.path.insert(0, setup_dir) # reset to include setup dir, w/clean callback list working_set.__init__() working_set.callbacks.append(lambda dist: dist.activate()) with DirectorySandbox(setup_dir): ns = dict(__file__=setup_script, __name__='__main__') _execfile(setup_script, ns) except SystemExit as v: if v.args and v.args[0]: raise # Normal exit, just return class AbstractSandbox: """Wrap 'os' module and 'open()' builtin for virtualizing setup scripts""" _active = False def __init__(self): self._attrs = [ name for name in dir(_os) if not name.startswith('_') and hasattr(self, name) ] def _copy(self, source): for name in self._attrs: setattr(os, name, getattr(source, name)) def __enter__(self): self._copy(self) if _file: builtins.file = self._file builtins.open = self._open self._active = True def __exit__(self, exc_type, exc_value, traceback): self._active = False if _file: builtins.file = _file builtins.open = _open self._copy(_os) def run(self, func): """Run 'func' under os sandboxing""" with self: return func() def _mk_dual_path_wrapper(name): original = getattr(_os, name) def wrap(self, src, dst, args, kw): if self._active: src, dst = self._remap_pair(name, src, dst, args, *kw) return original(src, dst, args, *kw) return wrap for name in ["rename", "link", "symlink"]: if hasattr(_os, name): locals()[name] = _mk_dual_path_wrapper(name) def _mk_single_path_wrapper(name, original=None): original = original or getattr(_os, name) def wrap(self, path, args, *kw): if self._active: path = self._remap_input(name, path, args, *kw) return original(path, args, *kw) return wrap if _file: _file = _mk_single_path_wrapper('file', _file) _open = _mk_single_path_wrapper('open', _open) for name in [ "stat", "listdir", "chdir", "open", "chmod", "chown", "mkdir", "remove", "unlink", "rmdir", "utime", "lchown", "chroot", "lstat", "startfile", "mkfifo", "mknod", "pathconf", "access" ]: if hasattr(_os, name): locals()[name] = _mk_single_path_wrapper(name) def _mk_single_with_return(name): original = getattr(_os, name) def wrap(self, path, args, *kw): if self._active: path = self._remap_input(name, path, args, *kw) return self._remap_output(name, original(path, args, *kw)) return original(path, args, *kw) return wrap for name in ['readlink', 'tempnam']: if hasattr(_os, name): locals()[name] = _mk_single_with_return(name) def _mk_query(name): original = getattr(_os, name) def wrap(self, args, *kw): retval = original(args, *kw) if self._active: return self._remap_output(name, retval) return retval return wrap for name in ['getcwd', 'tmpnam']: if hasattr(_os, name): locals()[name] = _mk_query(name) def _validate_path(self, path): """Called to remap or validate any path, whether input or output""" return path def _remap_input(self, operation, path, args, *kw): """Called for path inputs""" return self._validate_path(path) def _remap_output(self, operation, path): """Called for path outputs""" return self._validate_path(path) def _remap_pair(self, operation, src, dst, args, *kw): """Called for path pairs like rename, link, and symlink operations""" return ( self._remap_input(operation + '-from', src, args, *kw), self._remap_input(operation + '-to', dst, args, *kw) ) if hasattr(os, 'devnull'): _EXCEPTIONS = [os.devnull] else: _EXCEPTIONS = [] class DirectorySandbox(AbstractSandbox): """Restrict operations to a single subdirectory - pseudo-chroot""" write_ops = dict.fromkeys([ "open", "chmod", "chown", "mkdir", "remove", "unlink", "rmdir", "utime", "lchown", "chroot", "mkfifo", "mknod", "tempnam", ]) _exception_patterns = [ # Allow lib2to3 to attempt to save a pickled grammar object (#121) r'.lib2to3.\.pickle$', ] "exempt writing to paths that match the pattern" def __init__(self, sandbox, exceptions=_EXCEPTIONS): self._sandbox = os.path.normcase(os.path.realpath(sandbox)) self._prefix = os.path.join(self._sandbox, '') self._exceptions = [ os.path.normcase(os.path.realpath(path)) for path in exceptions ] AbstractSandbox.__init__(self) def _violation(self, operation, args, *kw): from setuptools.sandbox import SandboxViolation raise SandboxViolation(operation, args, kw) if _file: def _file(self, path, mode='r', args, *kw): if mode not in ('r', 'rt', 'rb', 'rU', 'U') and not self._ok(path): self._violation("file", path, mode, args, *kw) return _file(path, mode, args, *kw) def _open(self, path, mode='r', args, *kw): if mode not in ('r', 'rt', 'rb', 'rU', 'U') and not self._ok(path): self._violation("open", path, mode, args, *kw) return _open(path, mode, args, *kw) def tmpnam(self): self._violation("tmpnam") def _ok(self, path): active = self._active try: self._active = False realpath = os.path.normcase(os.path.realpath(path)) return ( self._exempted(realpath) or realpath == self._sandbox or realpath.startswith(self._prefix) ) finally: self._active = active def _exempted(self, filepath): start_matches = ( filepath.startswith(exception) for exception in self._exceptions ) pattern_matches = ( re.match(pattern, filepath) for pattern in self._exception_patterns ) candidates = itertools.chain(start_matches, pattern_matches) return any(candidates) def _remap_input(self, operation, path, args, *kw): """Called for path inputs""" if operation in self.write_ops and not self._ok(path): self._violation(operation, os.path.realpath(path), args, *kw) return path def _remap_pair(self, operation, src, dst, args, *kw): """Called for path pairs like rename, link, and symlink operations""" if not self._ok(src) or not self._ok(dst): self._violation(operation, src, dst, args, *kw) return (src, dst) def open(self, file, flags, mode=0o777, args, *kw): """Called for low-level os.open()""" if flags & WRITE_FLAGS and not self._ok(file): self._violation("os.open", file, flags, mode, args, *kw) return _os.open(file, flags, mode, args, kw) WRITE_FLAGS = functools.reduce( operator.or_, [ getattr(_os, a, 0) for a in "O_WRONLY O_RDWR O_APPEND O_CREAT O_TRUNC O_TEMPORARY".split()] ) class SandboxViolation(DistutilsError): """A setup script attempted to modify the filesystem outside the sandbox""" tmpl = textwrap.dedent(""" SandboxViolation: {cmd}{args!r} {kwargs} The package setup script has attempted to modify files on your system that are not within the EasyInstall build area, and has been aborted. This package cannot be safely installed by EasyInstall, and may not support alternate installation locations even if you run its setup script by hand. Please inform the package's author and the EasyInstall maintainers to find out if a fix or workaround is available. """).lstrip() def __str__(self): cmd, args, kwargs = self.args return self.tmpl.format(locals())	13.82 KB	2025-09-21 11:46
📄 script (dev).tmpl # EASY-INSTALL-DEV-SCRIPT: %(spec)r,%(script_name)r __requires__ = %(spec)r __import__('pkg_resources').require(%(spec)r) __file__ = %(dev_path)r with open(__file__) as f: exec(compile(f.read(), __file__, 'exec'))	218 B	2025-09-21 11:46
📄 script.tmpl # EASY-INSTALL-SCRIPT: %(spec)r,%(script_name)r __requires__ = %(spec)r __import__('pkg_resources').run_script(%(spec)r, %(script_name)r)	138 B	2025-09-21 11:46
📄 ssl_support.py import os import socket import atexit import re import functools import urllib.request import http.client from pkg_resources import ResolutionError, ExtractionError try: import ssl except ImportError: ssl = None __all__ = [ 'VerifyingHTTPSHandler', 'find_ca_bundle', 'is_available', 'cert_paths', 'opener_for' ] cert_paths = """ /etc/pki/tls/certs/ca-bundle.crt /etc/ssl/certs/ca-certificates.crt /usr/share/ssl/certs/ca-bundle.crt /usr/local/share/certs/ca-root.crt /etc/ssl/cert.pem /System/Library/OpenSSL/certs/cert.pem /usr/local/share/certs/ca-root-nss.crt /etc/ssl/ca-bundle.pem """.strip().split() try: HTTPSHandler = urllib.request.HTTPSHandler HTTPSConnection = http.client.HTTPSConnection except AttributeError: HTTPSHandler = HTTPSConnection = object is_available = ssl is not None and object not in ( HTTPSHandler, HTTPSConnection) try: from ssl import CertificateError, match_hostname except ImportError: try: from backports.ssl_match_hostname import CertificateError from backports.ssl_match_hostname import match_hostname except ImportError: CertificateError = None match_hostname = None if not CertificateError: class CertificateError(ValueError): pass if not match_hostname: # noqa: C901 # 'If 59' is too complex (21) # FIXME def _dnsname_match(dn, hostname, max_wildcards=1): """Matching according to RFC 6125, section 6.4.3 https://tools.ietf.org/html/rfc6125#section-6.4.3 """ pats = [] if not dn: return False # Ported from python3-syntax: # leftmost, remainder = dn.split(r'.') parts = dn.split(r'.') leftmost = parts[0] remainder = parts[1:] wildcards = leftmost.count('') if wildcards > max_wildcards: # Issue #17980: avoid denials of service by refusing more # than one wildcard per fragment. A survey of established # policy among SSL implementations showed it to be a # reasonable choice. raise CertificateError( "too many wildcards in certificate DNS name: " + repr(dn)) # speed up common case w/o wildcards if not wildcards: return dn.lower() == hostname.lower() # RFC 6125, section 6.4.3, subitem 1. # The client SHOULD NOT attempt to match a # presented identifier in which the wildcard # character comprises a label other than the # left-most label. if leftmost == '': # When '' is a fragment by itself, it matches a non-empty dotless # fragment. pats.append('[^.]+') elif leftmost.startswith('xn--') or hostname.startswith('xn--'): # RFC 6125, section 6.4.3, subitem 3. # The client SHOULD NOT attempt to match a presented identifier # where the wildcard character is embedded within an A-label or # U-label of an internationalized domain name. pats.append(re.escape(leftmost)) else: # Otherwise, '' matches any dotless string, e.g. www pats.append(re.escape(leftmost).replace(r'\', '[^.]')) # add the remaining fragments, ignore any wildcards for frag in remainder: pats.append(re.escape(frag)) pat = re.compile(r'\A' + r'\.'.join(pats) + r'\Z', re.IGNORECASE) return pat.match(hostname) def match_hostname(cert, hostname): """Verify that cert (in decoded format as returned by SSLSocket.getpeercert()) matches the hostname. RFC 2818 and RFC 6125 rules are followed, but IP addresses are not accepted for hostname. CertificateError is raised on failure. On success, the function returns nothing. """ if not cert: raise ValueError("empty or no certificate") dnsnames = [] san = cert.get('subjectAltName', ()) for key, value in san: if key == 'DNS': if _dnsname_match(value, hostname): return dnsnames.append(value) if not dnsnames: # The subject is only checked when there is no dNSName entry # in subjectAltName for sub in cert.get('subject', ()): for key, value in sub: # XXX according to RFC 2818, the most specific Common Name # must be used. if key == 'commonName': if _dnsname_match(value, hostname): return dnsnames.append(value) if len(dnsnames) > 1: raise CertificateError( "hostname %r doesn't match either of %s" % (hostname, ', '.join(map(repr, dnsnames)))) elif len(dnsnames) == 1: raise CertificateError( "hostname %r doesn't match %r" % (hostname, dnsnames[0])) else: raise CertificateError( "no appropriate commonName or " "subjectAltName fields were found") class VerifyingHTTPSHandler(HTTPSHandler): """Simple verifying handler: no auth, subclasses, timeouts, etc.""" def __init__(self, ca_bundle): self.ca_bundle = ca_bundle HTTPSHandler.__init__(self) def https_open(self, req): return self.do_open( lambda host, kw: VerifyingHTTPSConn(host, self.ca_bundle, kw), req ) class VerifyingHTTPSConn(HTTPSConnection): """Simple verifying connection: no auth, subclasses, timeouts, etc.""" def __init__(self, host, ca_bundle, kw): HTTPSConnection.__init__(self, host, kw) self.ca_bundle = ca_bundle def connect(self): sock = socket.create_connection( (self.host, self.port), getattr(self, 'source_address', None) ) # Handle the socket if a (proxy) tunnel is present if hasattr(self, '_tunnel') and getattr(self, '_tunnel_host', None): self.sock = sock self._tunnel() # http://bugs.python.org/issue7776: Python>=3.4.1 and >=2.7.7 # change self.host to mean the proxy server host when tunneling is # being used. Adapt, since we are interested in the destination # host for the match_hostname() comparison. actual_host = self._tunnel_host else: actual_host = self.host if hasattr(ssl, 'create_default_context'): ctx = ssl.create_default_context(cafile=self.ca_bundle) self.sock = ctx.wrap_socket(sock, server_hostname=actual_host) else: # This is for python < 2.7.9 and < 3.4? self.sock = ssl.wrap_socket( sock, cert_reqs=ssl.CERT_REQUIRED, ca_certs=self.ca_bundle ) try: match_hostname(self.sock.getpeercert(), actual_host) except CertificateError: self.sock.shutdown(socket.SHUT_RDWR) self.sock.close() raise def opener_for(ca_bundle=None): """Get a urlopen() replacement that uses ca_bundle for verification""" return urllib.request.build_opener( VerifyingHTTPSHandler(ca_bundle or find_ca_bundle()) ).open # from jaraco.functools def once(func): @functools.wraps(func) def wrapper(args, kwargs): if not hasattr(func, 'always_returns'): func.always_returns = func(args, **kwargs) return func.always_returns return wrapper @once def get_win_certfile(): try: import wincertstore except ImportError: return None class CertFile(wincertstore.CertFile): def __init__(self): super(CertFile, self).__init__() atexit.register(self.close) def close(self): try: super(CertFile, self).close() except OSError: pass _wincerts = CertFile() _wincerts.addstore('CA') _wincerts.addstore('ROOT') return _wincerts.name def find_ca_bundle(): """Return an existing CA bundle path, or None""" extant_cert_paths = filter(os.path.isfile, cert_paths) return ( get_win_certfile() or next(extant_cert_paths, None) or _certifi_where() ) def _certifi_where(): try: return __import__('certifi').where() except (ImportError, ResolutionError, ExtractionError): pass	8.36 KB	2025-09-21 11:46
📄 unicode_utils.py import unicodedata import sys # HFS Plus uses decomposed UTF-8 def decompose(path): if isinstance(path, str): return unicodedata.normalize('NFD', path) try: path = path.decode('utf-8') path = unicodedata.normalize('NFD', path) path = path.encode('utf-8') except UnicodeError: pass # Not UTF-8 return path def filesys_decode(path): """ Ensure that the given path is decoded, NONE when no expected encoding works """ if isinstance(path, str): return path fs_enc = sys.getfilesystemencoding() or 'utf-8' candidates = fs_enc, 'utf-8' for enc in candidates: try: return path.decode(enc) except UnicodeDecodeError: continue def try_encode(string, enc): "turn unicode encoding into a functional routine" try: return string.encode(enc) except UnicodeEncodeError: return None	941 B	2025-09-21 11:46
📄 version.py import pkg_resources try: __version__ = pkg_resources.get_distribution('setuptools').version except Exception: __version__ = 'unknown'	144 B	2025-09-21 11:46
📄 wheel.py """Wheels support.""" from distutils.util import get_platform from distutils import log import email import itertools import os import posixpath import re import zipfile import pkg_resources import setuptools from pkg_resources import parse_version from setuptools.extern.packaging.tags import sys_tags from setuptools.extern.packaging.utils import canonicalize_name from setuptools.command.egg_info import write_requirements WHEEL_NAME = re.compile( r"""^(?P<project_name>.+?)-(?P<version>\d.?) ((-(?P<build>\d.?))?-(?P<py_version>.+?)-(?P<abi>.+?)-(?P<platform>.+?) )\.whl$""", re.VERBOSE).match NAMESPACE_PACKAGE_INIT = \ "__import__('pkg_resources').declare_namespace(__name__)\n" def unpack(src_dir, dst_dir): '''Move everything under `src_dir` to `dst_dir`, and delete the former.''' for dirpath, dirnames, filenames in os.walk(src_dir): subdir = os.path.relpath(dirpath, src_dir) for f in filenames: src = os.path.join(dirpath, f) dst = os.path.join(dst_dir, subdir, f) os.renames(src, dst) for n, d in reversed(list(enumerate(dirnames))): src = os.path.join(dirpath, d) dst = os.path.join(dst_dir, subdir, d) if not os.path.exists(dst): # Directory does not exist in destination, # rename it and prune it from os.walk list. os.renames(src, dst) del dirnames[n] # Cleanup. for dirpath, dirnames, filenames in os.walk(src_dir, topdown=True): assert not filenames os.rmdir(dirpath) class Wheel: def __init__(self, filename): match = WHEEL_NAME(os.path.basename(filename)) if match is None: raise ValueError('invalid wheel name: %r' % filename) self.filename = filename for k, v in match.groupdict().items(): setattr(self, k, v) def tags(self): '''List tags (py_version, abi, platform) supported by this wheel.''' return itertools.product( self.py_version.split('.'), self.abi.split('.'), self.platform.split('.'), ) def is_compatible(self): '''Is the wheel is compatible with the current platform?''' supported_tags = set( (t.interpreter, t.abi, t.platform) for t in sys_tags()) return next((True for t in self.tags() if t in supported_tags), False) def egg_name(self): return pkg_resources.Distribution( project_name=self.project_name, version=self.version, platform=(None if self.platform == 'any' else get_platform()), ).egg_name() + '.egg' def get_dist_info(self, zf): # find the correct name of the .dist-info dir in the wheel file for member in zf.namelist(): dirname = posixpath.dirname(member) if (dirname.endswith('.dist-info') and canonicalize_name(dirname).startswith( canonicalize_name(self.project_name))): return dirname raise ValueError("unsupported wheel format. .dist-info not found") def install_as_egg(self, destination_eggdir): '''Install wheel as an egg directory.''' with zipfile.ZipFile(self.filename) as zf: self._install_as_egg(destination_eggdir, zf) def _install_as_egg(self, destination_eggdir, zf): dist_basename = '%s-%s' % (self.project_name, self.version) dist_info = self.get_dist_info(zf) dist_data = '%s.data' % dist_basename egg_info = os.path.join(destination_eggdir, 'EGG-INFO') self._convert_metadata(zf, destination_eggdir, dist_info, egg_info) self._move_data_entries(destination_eggdir, dist_data) self._fix_namespace_packages(egg_info, destination_eggdir) @staticmethod def _convert_metadata(zf, destination_eggdir, dist_info, egg_info): def get_metadata(name): with zf.open(posixpath.join(dist_info, name)) as fp: value = fp.read().decode('utf-8') return email.parser.Parser().parsestr(value) wheel_metadata = get_metadata('WHEEL') # Check wheel format version is supported. wheel_version = parse_version(wheel_metadata.get('Wheel-Version')) wheel_v1 = ( parse_version('1.0') <= wheel_version < parse_version('2.0dev0') ) if not wheel_v1: raise ValueError( 'unsupported wheel format version: %s' % wheel_version) # Extract to target directory. os.mkdir(destination_eggdir) zf.extractall(destination_eggdir) # Convert metadata. dist_info = os.path.join(destination_eggdir, dist_info) dist = pkg_resources.Distribution.from_location( destination_eggdir, dist_info, metadata=pkg_resources.PathMetadata(destination_eggdir, dist_info), ) # Note: Evaluate and strip markers now, # as it's difficult to convert back from the syntax: # foobar; "linux" in sys_platform and extra == 'test' def raw_req(req): req.marker = None return str(req) install_requires = list(sorted(map(raw_req, dist.requires()))) extras_require = { extra: sorted( req for req in map(raw_req, dist.requires((extra,))) if req not in install_requires ) for extra in dist.extras } os.rename(dist_info, egg_info) os.rename( os.path.join(egg_info, 'METADATA'), os.path.join(egg_info, 'PKG-INFO'), ) setup_dist = setuptools.Distribution( attrs=dict( install_requires=install_requires, extras_require=extras_require, ), ) # Temporarily disable info traces. log_threshold = log._global_log.threshold log.set_threshold(log.WARN) try: write_requirements( setup_dist.get_command_obj('egg_info'), None, os.path.join(egg_info, 'requires.txt'), ) finally: log.set_threshold(log_threshold) @staticmethod def _move_data_entries(destination_eggdir, dist_data): """Move data entries to their correct location.""" dist_data = os.path.join(destination_eggdir, dist_data) dist_data_scripts = os.path.join(dist_data, 'scripts') if os.path.exists(dist_data_scripts): egg_info_scripts = os.path.join( destination_eggdir, 'EGG-INFO', 'scripts') os.mkdir(egg_info_scripts) for entry in os.listdir(dist_data_scripts): # Remove bytecode, as it's not properly handled # during easy_install scripts install phase. if entry.endswith('.pyc'): os.unlink(os.path.join(dist_data_scripts, entry)) else: os.rename( os.path.join(dist_data_scripts, entry), os.path.join(egg_info_scripts, entry), ) os.rmdir(dist_data_scripts) for subdir in filter(os.path.exists, ( os.path.join(dist_data, d) for d in ('data', 'headers', 'purelib', 'platlib') )): unpack(subdir, destination_eggdir) if os.path.exists(dist_data): os.rmdir(dist_data) @staticmethod def _fix_namespace_packages(egg_info, destination_eggdir): namespace_packages = os.path.join( egg_info, 'namespace_packages.txt') if os.path.exists(namespace_packages): with open(namespace_packages) as fp: namespace_packages = fp.read().split() for mod in namespace_packages: mod_dir = os.path.join(destination_eggdir, *mod.split('.')) mod_init = os.path.join(mod_dir, '__init__.py') if not os.path.exists(mod_dir): os.mkdir(mod_dir) if not os.path.exists(mod_init): with open(mod_init, 'w') as fp: fp.write(NAMESPACE_PACKAGE_INIT)	8.09 KB	2025-09-21 11:46
📄 windows_support.py import platform import ctypes def windows_only(func): if platform.system() != 'Windows': return lambda args, *kwargs: None return func @windows_only def hide_file(path): """ Set the hidden attribute on a file or directory. From http://stackoverflow.com/questions/19622133/ `path` must be text. """ __import__('ctypes.wintypes') SetFileAttributes = ctypes.windll.kernel32.SetFileAttributesW SetFileAttributes.argtypes = ctypes.wintypes.LPWSTR, ctypes.wintypes.DWORD SetFileAttributes.restype = ctypes.wintypes.BOOL FILE_ATTRIBUTE_HIDDEN = 0x02 ret = SetFileAttributes(path, FILE_ATTRIBUTE_HIDDEN) if not ret: raise ctypes.WinError()	714 B	2025-09-21 11:46

Edit File:

Terminal