Wala shell

Advanced PHP File Manager

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

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py from .acquire import get_wheel, pip_wheel_env_run from .util import Version, Wheel __all__ = [ "get_wheel", "pip_wheel_env_run", "Version", "Wheel", ]	168 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 acquire.py """Bootstrap""" import logging import sys from operator import eq, lt from pathlib import Path from subprocess import PIPE, CalledProcessError, Popen from .bundle import from_bundle from .periodic_update import add_wheel_to_update_log from .util import Version, Wheel, discover_wheels def get_wheel(distribution, version, for_py_version, search_dirs, download, app_data, do_periodic_update, env): """ Get a wheel with the given distribution-version-for_py_version trio, by using the extra search dir + download """ # not all wheels are compatible with all python versions, so we need to py version qualify it wheel = None if not download or version != Version.bundle: # 1. acquire from bundle wheel = from_bundle(distribution, version, for_py_version, search_dirs, app_data, do_periodic_update, env) if download and wheel is None and version != Version.embed: # 2. download from the internet wheel = download_wheel( distribution=distribution, version_spec=Version.as_version_spec(version), for_py_version=for_py_version, search_dirs=search_dirs, app_data=app_data, to_folder=app_data.house, env=env, ) if wheel is not None and app_data.can_update: add_wheel_to_update_log(wheel, for_py_version, app_data) return wheel def download_wheel(distribution, version_spec, for_py_version, search_dirs, app_data, to_folder, env): to_download = f"{distribution}{version_spec or ''}" logging.debug("download wheel %s %s to %s", to_download, for_py_version, to_folder) cmd = [ sys.executable, "-m", "pip", "download", "--progress-bar", "off", "--disable-pip-version-check", "--only-binary=:all:", "--no-deps", "--python-version", for_py_version, "-d", str(to_folder), to_download, ] # pip has no interface in python - must be a new sub-process env = pip_wheel_env_run(search_dirs, app_data, env) process = Popen(cmd, env=env, stdout=PIPE, stderr=PIPE, universal_newlines=True, encoding="utf-8") out, err = process.communicate() if process.returncode != 0: kwargs = {"output": out, "stderr": err} raise CalledProcessError(process.returncode, cmd, **kwargs) result = _find_downloaded_wheel(distribution, version_spec, for_py_version, to_folder, out) logging.debug("downloaded wheel %s", result.name) return result def _find_downloaded_wheel(distribution, version_spec, for_py_version, to_folder, out): for line in out.splitlines(): line = line.lstrip() for marker in ("Saved ", "File was already downloaded "): if line.startswith(marker): return Wheel(Path(line[len(marker) :]).absolute()) # if for some reason the output does not match fallback to the latest version with that spec return find_compatible_in_house(distribution, version_spec, for_py_version, to_folder) def find_compatible_in_house(distribution, version_spec, for_py_version, in_folder): wheels = discover_wheels(in_folder, distribution, None, for_py_version) start, end = 0, len(wheels) if version_spec is not None and version_spec != "": if version_spec.startswith("<"): from_pos, op = 1, lt elif version_spec.startswith("=="): from_pos, op = 2, eq else: raise ValueError(version_spec) version = Wheel.as_version_tuple(version_spec[from_pos:]) start = next((at for at, w in enumerate(wheels) if op(w.version_tuple, version)), len(wheels)) return None if start == end else wheels[start] def pip_wheel_env_run(search_dirs, app_data, env): env = env.copy() env.update({"PIP_USE_WHEEL": "1", "PIP_USER": "0", "PIP_NO_INPUT": "1"}) wheel = get_wheel( distribution="pip", version=None, for_py_version=f"{sys.version_info.major}.{sys.version_info.minor}", search_dirs=search_dirs, download=False, app_data=app_data, do_periodic_update=False, env=env, ) if wheel is None: raise RuntimeError("could not find the embedded pip") env["PYTHONPATH"] = str(wheel.path) return env __all__ = [ "get_wheel", "download_wheel", "pip_wheel_env_run", ]	4.31 KB	2026-01-20 12:58
📄 bundle.py from ..wheels.embed import get_embed_wheel from .periodic_update import periodic_update from .util import Version, Wheel, discover_wheels def from_bundle(distribution, version, for_py_version, search_dirs, app_data, do_periodic_update, env): """ Load the bundled wheel to a cache directory. """ of_version = Version.of_version(version) wheel = load_embed_wheel(app_data, distribution, for_py_version, of_version) if version != Version.embed: # 2. check if we have upgraded embed if app_data.can_update: per = do_periodic_update wheel = periodic_update(distribution, of_version, for_py_version, wheel, search_dirs, app_data, per, env) # 3. acquire from extra search dir found_wheel = from_dir(distribution, of_version, for_py_version, search_dirs) if found_wheel is not None: if wheel is None: wheel = found_wheel elif found_wheel.version_tuple > wheel.version_tuple: wheel = found_wheel return wheel def load_embed_wheel(app_data, distribution, for_py_version, version): wheel = get_embed_wheel(distribution, for_py_version) if wheel is not None: version_match = version == wheel.version if version is None or version_match: with app_data.ensure_extracted(wheel.path, lambda: app_data.house) as wheel_path: wheel = Wheel(wheel_path) else: # if version does not match ignore wheel = None return wheel def from_dir(distribution, version, for_py_version, directories): """ Load a compatible wheel from a given folder. """ for folder in directories: for wheel in discover_wheels(folder, distribution, version, for_py_version): return wheel return None __all__ = [ "load_embed_wheel", "from_bundle", ]	1.84 KB	2026-01-20 12:58
📁 embed	-	2026-03-21 15:45
📄 periodic_update.py """ Periodically update bundled versions. """ import json import logging import os import ssl import sys from datetime import datetime, timedelta from itertools import groupby from pathlib import Path from shutil import copy2 from subprocess import PIPE, Popen from textwrap import dedent from threading import Thread from urllib.error import URLError from urllib.request import urlopen from virtualenv.app_data import AppDataDiskFolder from virtualenv.util.subprocess import CREATE_NO_WINDOW from ..wheels.embed import BUNDLE_SUPPORT from ..wheels.util import Wheel GRACE_PERIOD_CI = timedelta(hours=1) # prevent version switch in the middle of a CI run GRACE_PERIOD_MINOR = timedelta(days=28) UPDATE_PERIOD = timedelta(days=14) UPDATE_ABORTED_DELAY = timedelta(hours=1) def periodic_update(distribution, of_version, for_py_version, wheel, search_dirs, app_data, do_periodic_update, env): if do_periodic_update: handle_auto_update(distribution, for_py_version, wheel, search_dirs, app_data, env) now = datetime.now() def _update_wheel(ver): updated_wheel = Wheel(app_data.house / ver.filename) logging.debug("using %supdated wheel %s", "periodically " if updated_wheel else "", updated_wheel) return updated_wheel u_log = UpdateLog.from_app_data(app_data, distribution, for_py_version) if of_version is None: for _, group in groupby(u_log.versions, key=lambda v: v.wheel.version_tuple[0:2]): # use only latest patch version per minor, earlier assumed to be buggy all_patches = list(group) ignore_grace_period_minor = any(version for version in all_patches if version.use(now)) for version in all_patches: if wheel is not None and Path(version.filename).name == wheel.name: return wheel if version.use(now, ignore_grace_period_minor): return _update_wheel(version) else: for version in u_log.versions: if version.wheel.version == of_version: return _update_wheel(version) return wheel def handle_auto_update(distribution, for_py_version, wheel, search_dirs, app_data, env): embed_update_log = app_data.embed_update_log(distribution, for_py_version) u_log = UpdateLog.from_dict(embed_update_log.read()) if u_log.needs_update: u_log.periodic = True u_log.started = datetime.now() embed_update_log.write(u_log.to_dict()) trigger_update(distribution, for_py_version, wheel, search_dirs, app_data, periodic=True, env=env) def add_wheel_to_update_log(wheel, for_py_version, app_data): embed_update_log = app_data.embed_update_log(wheel.distribution, for_py_version) logging.debug("adding %s information to %s", wheel.name, embed_update_log.file) u_log = UpdateLog.from_dict(embed_update_log.read()) if any(version.filename == wheel.name for version in u_log.versions): logging.warning("%s already present in %s", wheel.name, embed_update_log.file) return # we don't need a release date for sources other than "periodic" version = NewVersion(wheel.name, datetime.now(), None, "download") u_log.versions.append(version) # always write at the end for proper updates embed_update_log.write(u_log.to_dict()) DATETIME_FMT = "%Y-%m-%dT%H:%M:%S.%fZ" def dump_datetime(value): return None if value is None else value.strftime(DATETIME_FMT) def load_datetime(value): return None if value is None else datetime.strptime(value, DATETIME_FMT) class NewVersion: def __init__(self, filename, found_date, release_date, source): self.filename = filename self.found_date = found_date self.release_date = release_date self.source = source @classmethod def from_dict(cls, dictionary): return cls( filename=dictionary["filename"], found_date=load_datetime(dictionary["found_date"]), release_date=load_datetime(dictionary["release_date"]), source=dictionary["source"], ) def to_dict(self): return { "filename": self.filename, "release_date": dump_datetime(self.release_date), "found_date": dump_datetime(self.found_date), "source": self.source, } def use(self, now, ignore_grace_period_minor=False, ignore_grace_period_ci=False): if self.source == "manual": return True elif self.source == "periodic": if self.found_date < now - GRACE_PERIOD_CI or ignore_grace_period_ci: if not ignore_grace_period_minor: compare_from = self.release_date or self.found_date return now - compare_from >= GRACE_PERIOD_MINOR return True return False def __repr__(self): return ( f"{self.__class__.__name__}(filename={self.filename}), found_date={self.found_date}, " f"release_date={self.release_date}, source={self.source})" ) def __eq__(self, other): return type(self) == type(other) and all( getattr(self, k) == getattr(other, k) for k in ["filename", "release_date", "found_date", "source"] ) def __ne__(self, other): return not (self == other) @property def wheel(self): return Wheel(Path(self.filename)) class UpdateLog: def __init__(self, started, completed, versions, periodic): self.started = started self.completed = completed self.versions = versions self.periodic = periodic @classmethod def from_dict(cls, dictionary): if dictionary is None: dictionary = {} return cls( load_datetime(dictionary.get("started")), load_datetime(dictionary.get("completed")), [NewVersion.from_dict(v) for v in dictionary.get("versions", [])], dictionary.get("periodic"), ) @classmethod def from_app_data(cls, app_data, distribution, for_py_version): raw_json = app_data.embed_update_log(distribution, for_py_version).read() return cls.from_dict(raw_json) def to_dict(self): return { "started": dump_datetime(self.started), "completed": dump_datetime(self.completed), "periodic": self.periodic, "versions": [r.to_dict() for r in self.versions], } @property def needs_update(self): now = datetime.now() if self.completed is None: # never completed return self._check_start(now) else: if now - self.completed <= UPDATE_PERIOD: return False return self._check_start(now) def _check_start(self, now): return self.started is None or now - self.started > UPDATE_ABORTED_DELAY def trigger_update(distribution, for_py_version, wheel, search_dirs, app_data, env, periodic): wheel_path = None if wheel is None else str(wheel.path) cmd = [ sys.executable, "-c", dedent( """ from virtualenv.report import setup_report, MAX_LEVEL from virtualenv.seed.wheels.periodic_update import do_update setup_report(MAX_LEVEL, show_pid=True) do_update({!r}, {!r}, {!r}, {!r}, {!r}, {!r}) """, ) .strip() .format(distribution, for_py_version, wheel_path, str(app_data), [str(p) for p in search_dirs], periodic), ] debug = env.get("_VIRTUALENV_PERIODIC_UPDATE_INLINE") == "1" pipe = None if debug else PIPE kwargs = {"stdout": pipe, "stderr": pipe} if not debug and sys.platform == "win32": kwargs["creationflags"] = CREATE_NO_WINDOW process = Popen(cmd, *kwargs) logging.info( "triggered periodic upgrade of %s%s (for python %s) via background process having PID %d", distribution, "" if wheel is None else f"=={wheel.version}", for_py_version, process.pid, ) if debug: process.communicate() # on purpose not called to make it a background process def do_update(distribution, for_py_version, embed_filename, app_data, search_dirs, periodic): versions = None try: versions = _run_do_update(app_data, distribution, embed_filename, for_py_version, periodic, search_dirs) finally: logging.debug("done %s %s with %s", distribution, for_py_version, versions) return versions def _run_do_update(app_data, distribution, embed_filename, for_py_version, periodic, search_dirs): from virtualenv.seed.wheels import acquire wheel_filename = None if embed_filename is None else Path(embed_filename) embed_version = None if wheel_filename is None else Wheel(wheel_filename).version_tuple app_data = AppDataDiskFolder(app_data) if isinstance(app_data, str) else app_data search_dirs = [Path(p) if isinstance(p, str) else p for p in search_dirs] wheelhouse = app_data.house embed_update_log = app_data.embed_update_log(distribution, for_py_version) u_log = UpdateLog.from_dict(embed_update_log.read()) now = datetime.now() update_versions, other_versions = [], [] for version in u_log.versions: if version.source in {"periodic", "manual"}: update_versions.append(version) else: other_versions.append(version) if periodic: source = "periodic" else: source = "manual" # mark the most recent one as source "manual" if update_versions: update_versions[0].source = source if wheel_filename is not None: dest = wheelhouse / wheel_filename.name if not dest.exists(): copy2(str(wheel_filename), str(wheelhouse)) last, last_version, versions, filenames = None, None, [], set() while last is None or not last.use(now, ignore_grace_period_ci=True): download_time = datetime.now() dest = acquire.download_wheel( distribution=distribution, version_spec=None if last_version is None else f"<{last_version}", for_py_version=for_py_version, search_dirs=search_dirs, app_data=app_data, to_folder=wheelhouse, env=os.environ, ) if dest is None or (update_versions and update_versions[0].filename == dest.name): break release_date = release_date_for_wheel_path(dest.path) last = NewVersion(filename=dest.path.name, release_date=release_date, found_date=download_time, source=source) logging.info("detected %s in %s", last, datetime.now() - download_time) versions.append(last) filenames.add(last.filename) last_wheel = last.wheel last_version = last_wheel.version if embed_version is not None: if embed_version >= last_wheel.version_tuple: # stop download if we reach the embed version break u_log.periodic = periodic if not u_log.periodic: u_log.started = now # update other_versions by removing version we just found other_versions = [version for version in other_versions if version.filename not in filenames] u_log.versions = versions + update_versions + other_versions u_log.completed = datetime.now() embed_update_log.write(u_log.to_dict()) return versions def release_date_for_wheel_path(dest): wheel = Wheel(dest) # the most accurate is to ask PyPi - e.g. https://pypi.org/pypi/pip/json, # see https://warehouse.pypa.io/api-reference/json/ for more details content = _pypi_get_distribution_info_cached(wheel.distribution) if content is not None: try: upload_time = content["releases"][wheel.version][0]["upload_time"] return datetime.strptime(upload_time, "%Y-%m-%dT%H:%M:%S") except Exception as exception: logging.error("could not load release date %s because %r", content, exception) return None def _request_context(): yield None # fallback to non verified HTTPS (the information we request is not sensitive, so fallback) yield ssl._create_unverified_context() _PYPI_CACHE = {} def _pypi_get_distribution_info_cached(distribution): if distribution not in _PYPI_CACHE: _PYPI_CACHE[distribution] = _pypi_get_distribution_info(distribution) return _PYPI_CACHE[distribution] def _pypi_get_distribution_info(distribution): content, url = None, f"https://pypi.org/pypi/{distribution}/json" try: for context in _request_context(): try: with urlopen(url, context=context) as file_handler: content = json.load(file_handler) break except URLError as exception: logging.error("failed to access %s because %r", url, exception) except Exception as exception: logging.error("failed to access %s because %r", url, exception) return content def manual_upgrade(app_data, env): threads = [] for for_py_version, distribution_to_package in BUNDLE_SUPPORT.items(): # load extra search dir for the given for_py for distribution in distribution_to_package.keys(): thread = Thread(target=_run_manual_upgrade, args=(app_data, distribution, for_py_version, env)) thread.start() threads.append(thread) for thread in threads: thread.join() def _run_manual_upgrade(app_data, distribution, for_py_version, env): start = datetime.now() from .bundle import from_bundle current = from_bundle( distribution=distribution, version=None, for_py_version=for_py_version, search_dirs=[], app_data=app_data, do_periodic_update=False, env=env, ) logging.warning( "upgrade %s for python %s with current %s", distribution, for_py_version, "" if current is None else current.name, ) versions = do_update( distribution=distribution, for_py_version=for_py_version, embed_filename=current.path, app_data=app_data, search_dirs=[], periodic=False, ) args = [ distribution, for_py_version, datetime.now() - start, ] if versions: args.append("\n".join(f"\t{v}" for v in versions)) ver_update = "new entries found:\n%s" if versions else "no new versions found" logging.warning(f"upgraded %s for python %s in %s {ver_update}", args) __all__ = [ "add_wheel_to_update_log", "periodic_update", "do_update", "manual_upgrade", "NewVersion", "UpdateLog", "load_datetime", "dump_datetime", "trigger_update", "release_date_for_wheel_path", ]	14.47 KB	2026-01-20 12:58
📄 util.py from operator import attrgetter from zipfile import ZipFile class Wheel: def __init__(self, path): # https://www.python.org/dev/peps/pep-0427/#file-name-convention # The wheel filename is {distribution}-{version}(-{build tag})?-{python tag}-{abi tag}-{platform tag}.whl self.path = path self._parts = path.stem.split("-") @classmethod def from_path(cls, path): if path is not None and path.suffix == ".whl" and len(path.stem.split("-")) >= 5: return cls(path) return None @property def distribution(self): return self._parts[0] @property def version(self): return self._parts[1] @property def version_tuple(self): return self.as_version_tuple(self.version) @staticmethod def as_version_tuple(version): result = [] for part in version.split(".")[0:3]: try: result.append(int(part)) except ValueError: break if not result: raise ValueError(version) return tuple(result) @property def name(self): return self.path.name def support_py(self, py_version): name = f"{'-'.join(self.path.stem.split('-')[0:2])}.dist-info/METADATA" with ZipFile(str(self.path), "r") as zip_file: metadata = zip_file.read(name).decode("utf-8") marker = "Requires-Python:" requires = next((i[len(marker) :] for i in metadata.splitlines() if i.startswith(marker)), None) if requires is None: # if it does not specify a python requires the assumption is compatible return True py_version_int = tuple(int(i) for i in py_version.split(".")) for require in (i.strip() for i in requires.split(",")): # https://www.python.org/dev/peps/pep-0345/#version-specifiers for operator, check in [ ("!=", lambda v: py_version_int != v), ("==", lambda v: py_version_int == v), ("<=", lambda v: py_version_int <= v), (">=", lambda v: py_version_int >= v), ("<", lambda v: py_version_int < v), (">", lambda v: py_version_int > v), ]: if require.startswith(operator): ver_str = require[len(operator) :].strip() version = tuple((int(i) if i != "*" else None) for i in ver_str.split("."))[0:2] if not check(version): return False break return True def __repr__(self): return f"{self.__class__.__name__}({self.path})" def __str__(self): return str(self.path) def discover_wheels(from_folder, distribution, version, for_py_version): wheels = [] for filename in from_folder.iterdir(): wheel = Wheel.from_path(filename) if wheel and wheel.distribution == distribution: if version is None or wheel.version == version: if wheel.support_py(for_py_version): wheels.append(wheel) return sorted(wheels, key=attrgetter("version_tuple", "distribution"), reverse=True) class Version: #: the version bundled with virtualenv bundle = "bundle" embed = "embed" #: custom version handlers non_version = (bundle, embed) @staticmethod def of_version(value): return None if value in Version.non_version else value @staticmethod def as_pip_req(distribution, version): return f"{distribution}{Version.as_version_spec(version)}" @staticmethod def as_version_spec(version): of_version = Version.of_version(version) return "" if of_version is None else f"=={of_version}" __all__ = [ "discover_wheels", "Version", "Wheel", ]	3.75 KB	2026-01-20 12:58

Terminal