#!/usr/bin/env -S uv run --script # /// script # requires-python = "==3.13" # dependencies = [ # "pydantic==2.11.7", # "requests==2.32.5", # "docker==7.1.0", # "urllib3==2.5.0", # "siphash24==1.8", # ] # /// from __future__ import annotations import concurrent.futures import json import logging import os.path import py_compile import random import shutil import struct import tarfile import time import zipfile from datetime import datetime from pathlib import Path from threading import Lock from zoneinfo import ZoneInfo import docker import requests from docker.errors import APIError, NotFound from siphash24 import siphash24 from models import ( CacheFile, Distribution, DistributionKind, InvalidationMode, MajorMinorVersion, Package, ParseResponse, PythonVersion, ReproductionResult, DistributionResult, FileReproductionResult, DistributionStatus, FileStatus, PythonVersionChoice, ) INPUT_PATH = Path(__file__).parent.joinpath("results/2025-09-09-compact.json") BERLIN_TZ = ZoneInfo("Europe/Berlin") UNPACKED_PACKAGES_DIR = str(Path(__file__).parent.joinpath("Temp").absolute()) # NOTE: Maps magic bytecode numbers to a sorted list of releases MAGIC_NUMBER_VERSIONS = { 3280: ["3.4.0a3", "3.4.0a2", "3.4.0a1"], 3290: ["3.4.0a4"], 3300: ["3.4.0rc1", "3.4.0b3", "3.4.0b2", "3.4.0b1"], 3310: [ "3.4.9rc1", "3.4.9", "3.4.8rc1", "3.4.8", "3.4.7rc1", "3.4.7", "3.4.6rc1", "3.4.6", "3.4.5rc1", "3.4.5", "3.4.4rc1", "3.4.4", "3.4.3rc1", "3.4.3", "3.4.2rc1", "3.4.2", "3.4.1rc1", "3.4.10rc1", "3.4.10", "3.4.1", "3.4.0rc3", "3.4.0rc2", "3.4.0", ], 3320: ["3.5.0a4", "3.5.0a3", "3.5.0a2", "3.5.0a1"], 3330: ["3.5.0b1"], 3340: ["3.5.0b2"], 3350: [ "3.5.2rc1", "3.5.2", "3.5.1rc1", "3.5.1", "3.5.0rc4", "3.5.0rc3", "3.5.0rc2", "3.5.0rc1", "3.5.0b4", "3.5.0b3", "3.5.0", ], 3351: [ "3.5.9", "3.5.8rc2", "3.5.8rc1", "3.5.8", "3.5.7rc1", "3.5.7", "3.5.6rc1", "3.5.6", "3.5.5rc1", "3.5.5", "3.5.4rc1", "3.5.4", "3.5.3rc1", "3.5.3", "3.5.10rc1", "3.5.10", ], 3361: ["3.6.0a1"], 3372: ["3.6.0a4", "3.6.0a3", "3.6.0a2"], 3377: ["3.6.0b1"], 3378: ["3.6.0b4", "3.6.0b3", "3.6.0b2"], 3379: [ "3.6.9rc1", "3.6.9", "3.6.8rc1", "3.6.8", "3.6.7rc2", "3.6.7rc1", "3.6.7", "3.6.6rc1", "3.6.6", "3.6.5rc1", "3.6.5", "3.6.4rc1", "3.6.4", "3.6.3rc1", "3.6.3", "3.6.2rc2", "3.6.2rc1", "3.6.2", "3.6.1rc1", "3.6.15", "3.6.14", "3.6.13", "3.6.12", "3.6.11rc1", "3.6.11", "3.6.10rc1", "3.6.10", "3.6.1", "3.6.0rc2", "3.6.0rc1", "3.6.0", ], 3390: ["3.7.0a1"], 3391: ["3.7.0a3", "3.7.0a2"], 3392: ["3.7.0a4"], 3393: ["3.7.0b4", "3.7.0b3", "3.7.0b2", "3.7.0b1"], 3394: [ "3.7.9", "3.7.8rc1", "3.7.8", "3.7.7rc1", "3.7.7", "3.7.6rc1", "3.7.6", "3.7.5rc1", "3.7.5", "3.7.4rc2", "3.7.4rc1", "3.7.4", "3.7.3rc1", "3.7.3", "3.7.2rc1", "3.7.2", "3.7.1rc2", "3.7.1rc1", "3.7.17", "3.7.16", "3.7.15", "3.7.14", "3.7.13", "3.7.12", "3.7.11", "3.7.10", "3.7.1", "3.7.0rc1", "3.7.0b5", "3.7.0", ], 3401: ["3.8.0a3", "3.8.0a2", "3.8.0a1"], 3410: ["3.8.0b1", "3.8.0a4"], 3411: ["3.8.0b2"], 3412: ["3.8.0b3"], 3413: [ "3.8.9", "3.8.8rc1", "3.8.8", "3.8.7rc1", "3.8.7", "3.8.6rc1", "3.8.6", "3.8.5", "3.8.4rc1", "3.8.4", "3.8.3rc1", "3.8.3", "3.8.2rc2", "3.8.2rc1", "3.8.20", "3.8.2", "3.8.1rc1", "3.8.19", "3.8.18", "3.8.17", "3.8.16", "3.8.15", "3.8.14", "3.8.13", "3.8.12", "3.8.11", "3.8.10", "3.8.1", "3.8.0rc1", "3.8.0b4", "3.8.0", ], 3420: ["3.9.0a1"], 3422: ["3.9.0a2"], 3424: ["3.9.0a3"], 3425: [ "3.9.9", "3.9.8", "3.9.7", "3.9.6", "3.9.5", "3.9.4", "3.9.3", "3.9.2rc1", "3.9.23", "3.9.22", "3.9.21", "3.9.20", "3.9.2", "3.9.1rc1", "3.9.19", "3.9.18", "3.9.17", "3.9.16", "3.9.15", "3.9.14", "3.9.13", "3.9.12", "3.9.11", "3.9.10", "3.9.1", "3.9.0rc2", "3.9.0rc1", "3.9.0b5", "3.9.0b4", "3.9.0b3", "3.9.0b2", "3.9.0b1", "3.9.0a6", "3.9.0a5", "3.9.0a4", "3.9.0", "3.10.0a1", ], 3430: ["3.10.0a2"], 3432: ["3.10.0a3"], 3433: ["3.10.0a5", "3.10.0a4"], 3434: ["3.10.0a6"], 3435: ["3.10.0a7"], 3439: [ "3.10.9", "3.10.8", "3.10.7", "3.10.6", "3.10.5", "3.10.4", "3.10.3", "3.10.2", "3.10.18", "3.10.17", "3.10.16", "3.10.15", "3.10.14", "3.10.13", "3.10.12", "3.10.11", "3.10.10", "3.10.1", "3.10.0rc2", "3.10.0rc1", "3.10.0b4", "3.10.0b3", "3.10.0b2", "3.10.0b1", "3.10.0", ], 3460: ["3.11.0a1"], 3462: ["3.11.0a2"], 3465: ["3.11.0a3"], 3474: ["3.11.0a4"], 3478: ["3.11.0a5"], 3485: ["3.11.0a6"], 3491: ["3.11.0a7"], 3494: ["3.11.0b3", "3.11.0b2", "3.11.0b1"], 3495: [ "3.11.9", "3.11.8", "3.11.7", "3.11.6", "3.11.5", "3.11.4", "3.11.3", "3.11.2", "3.11.13", "3.11.12", "3.11.11", "3.11.10", "3.11.1", "3.11.0rc2", "3.11.0rc1", "3.11.0b5", "3.11.0b4", "3.11.0", ], 3509: ["3.12.0a1"], 3512: ["3.12.0a3", "3.12.0a2"], 3514: ["3.12.0a4"], 3517: ["3.12.0a5"], 3520: ["3.12.0a6"], 3524: ["3.12.0a7"], 3531: [ "3.12.9", "3.12.8", "3.12.7", "3.12.6", "3.12.5", "3.12.4", "3.12.3", "3.12.2", "3.12.11", "3.12.10", "3.12.1", "3.12.0rc3", "3.12.0rc2", "3.12.0rc1", "3.12.0b4", "3.12.0b3", "3.12.0b2", "3.12.0b1", "3.12.0", ], 3563: ["3.13.0a1"], 3565: ["3.13.0a2"], 3566: ["3.13.0a3"], 3568: ["3.13.0a4"], 3569: ["3.13.0a6", "3.13.0a5"], 3570: ["3.13.0b1"], 3571: [ "3.13.7", "3.13.6", "3.13.5", "3.13.4", "3.13.3", "3.13.2", "3.13.1", "3.13.0rc3", "3.13.0rc2", "3.13.0rc1", "3.13.0b4", "3.13.0b3", "3.13.0b2", "3.13.0", ], } PYTHON_RELEASE_TABLE = { "2.0": ["2.0.1"], "2.1": ["2.1.3"], "2.2": ["2.2.3", "2.2.2", "2.2.1", "2.2.0"], "2.3": ["2.3.7", "2.3.6", "2.3.5", "2.3.4", "2.3.3", "2.3.2", "2.3.1", "2.3.0"], "2.4": ["2.4.6", "2.4.5", "2.4.4", "2.4.3", "2.4.2", "2.4.1", "2.4.0"], "2.5": ["2.5.6", "2.5.5", "2.5.4", "2.5.3", "2.5.2", "2.5.1", "2.5.0"], "2.6": [ "2.6.9", "2.6.8", "2.6.7", "2.6.6", "2.6.5", "2.6.4", "2.6.3", "2.6.2", "2.6.1", "2.6.0", ], "2.7": [ "2.7.18", "2.7.17", "2.7.16", "2.7.15", "2.7.14", "2.7.13", "2.7.12", "2.7.11", "2.7.10", "2.7.9", "2.7.8", "2.7.7", "2.7.6", "2.7.5", "2.7.4", "2.7.3", "2.7.2", "2.7.1", "2.7.0", ], "3.0": ["3.0.1", "3.0.0"], "3.1": ["3.1.5", "3.1.4", "3.1.3", "3.1.2", "3.1.1", "3.1.0"], "3.10": [ "3.10.18", "3.10.17", "3.10.16", "3.10.15", "3.10.14", "3.10.13", "3.10.12", "3.10.11", "3.10.10", "3.10.9", "3.10.8", "3.10.7", "3.10.6", "3.10.5", "3.10.4", "3.10.3", "3.10.2", "3.10.1", "3.10.0", ], "3.11": [ "3.11.13", "3.11.12", "3.11.11", "3.11.10", "3.11.9", "3.11.8", "3.11.7", "3.11.6", "3.11.5", "3.11.4", "3.11.3", "3.11.2", "3.11.1", "3.11.0", ], "3.12": [ "3.12.11", "3.12.10", "3.12.9", "3.12.8", "3.12.7", "3.12.6", "3.12.5", "3.12.4", "3.12.3", "3.12.2", "3.12.1", "3.12.0", ], "3.13": [ "3.13.7", "3.13.6", "3.13.5", "3.13.4", "3.13.3", "3.13.2", "3.13.1", "3.13.0", ], "3.2": ["3.2.6", "3.2.5", "3.2.4", "3.2.3", "3.2.2", "3.2.1", "3.2.0"], "3.3": ["3.3.7", "3.3.6", "3.3.5", "3.3.4", "3.3.3", "3.3.2", "3.3.1", "3.3.0"], "3.4": [ "3.4.10", "3.4.9", "3.4.8", "3.4.7", "3.4.6", "3.4.5", "3.4.4", "3.4.3", "3.4.2", "3.4.1", "3.4.0", ], "3.5": [ "3.5.10", "3.5.9", "3.5.8", "3.5.7", "3.5.6", "3.5.5", "3.5.4", "3.5.3", "3.5.2", "3.5.1", "3.5.0", ], "3.6": [ "3.6.15", "3.6.14", "3.6.13", "3.6.12", "3.6.11", "3.6.10", "3.6.9", "3.6.8", "3.6.7", "3.6.6", "3.6.5", "3.6.4", "3.6.3", "3.6.2", "3.6.1", "3.6.0", ], "3.7": [ "3.7.17", "3.7.16", "3.7.15", "3.7.14", "3.7.13", "3.7.12", "3.7.11", "3.7.10", "3.7.9", "3.7.8", "3.7.7", "3.7.6", "3.7.5", "3.7.4", "3.7.3", "3.7.2", "3.7.1", "3.7.0", ], "3.8": [ "3.8.20", "3.8.19", "3.8.18", "3.8.17", "3.8.16", "3.8.15", "3.8.14", "3.8.13", "3.8.12", "3.8.11", "3.8.10", "3.8.9", "3.8.8", "3.8.7", "3.8.6", "3.8.5", "3.8.4", "3.8.3", "3.8.2", "3.8.1", "3.8.0", ], "3.9": [ "3.9.23", "3.9.22", "3.9.21", "3.9.20", "3.9.19", "3.9.18", "3.9.17", "3.9.16", "3.9.15", "3.9.14", "3.9.13", "3.9.12", "3.9.11", "3.9.10", "3.9.9", "3.9.8", "3.9.7", "3.9.6", "3.9.5", "3.9.4", "3.9.2", "3.9.1", "3.9.0", ], "3.14": ["3.14.0rc2"], } class ParsingError(BaseException): pass class CompilationError(BaseException): pass class PythonExecutor: def __init__(self): self.version_port_map = {} self.client = docker.from_env() self.available_ports = list(range(38_500, 40_000)) self.containers = [] self.lock = Lock() self.unavailable_versions = {} def _ensure_running(self, version: str) -> int: with self.lock: if version in self.unavailable_versions: raise self.unavailable_versions[version] if version not in self.version_port_map: self._start_container(version) return self.version_port_map[version] def _take_port(self, version: str) -> int: # HACK: this is not very resillient. # It takes a port *before* the container starts, and if it fails to start for some reason, # the port stays in the version map. # Right now i manually remove the port for every exception that occurs when trying to start # a container, but surely there is a more elegant solution. port = self.available_ports.pop( random.randint(0, len(self.available_ports) - 1) ) self.version_port_map[version] = port return port def _start_container(self, version: str): logging.info("Starting docker container for version %s.", version) image_name = f"python:{version}-slim" port = self._take_port(version) file_path = Path(__file__).parent while True: try: container = self.client.containers.run( image_name, "/app/rust-server/target/x86_64-unknown-linux-gnu/release/server", ports={"8080": ("127.0.0.1", port)}, volumes={ str(file_path.joinpath("docker-files").absolute()): { "bind": "/app", "mode": "ro", }, str(file_path.joinpath("pyc_parsing").absolute()): { "bind": "/pyc_parsing", "mode": "ro", }, str(file_path.joinpath("pyc_compiling").absolute()): { "bind": "/pyc_compiling", "mode": "ro", }, UNPACKED_PACKAGES_DIR: { "bind": UNPACKED_PACKAGES_DIR, "mode": "ro", }, }, detach=True, remove=True, stop_signal="SIGKILL", ) break except NotFound as e: self.unavailable_versions[version] = e del self.version_port_map[version] logging.warning( "Tried to start unavailable container for python version %s.", version, ) raise except APIError as e: if "failed to bind host port" in str(e): logging.error("Failed to bind container to new port -- reassigning") port = self._take_port(version) else: del self.version_port_map[version] raise while container.status != "created": logging.debug( "Waiting for container to start. Status: %s", container.status ) time.sleep(1) self.containers.append(container) time.sleep(2) # wait for http server to get ready def parse(self, pyc_file_path: Path, python_version: str) -> ParseResponse: """ Returns the (raw) magic bytes, as well as a mapping from the "path" of code objects in the module to their bytecode. """ script = match_script(Path("./pyc_parsing"), python_version) port = self._ensure_running(python_version) logging.debug( "Sending parse request to port %d for source %s with script %s pyton-version %s", port, pyc_file_path, script, python_version, ) resp = requests.post( f"http://127.0.0.1:{port}/parse", params={ "pyc_file_path": str(pyc_file_path.absolute()), "script_name": script, }, timeout=30, ) logging.debug("Got parsing response: %s", resp.text) if not resp.ok: logging.error( "Failed to parse %s with %s via port %s: %s", pyc_file_path, python_version, port, resp.content, ) raise ParsingError return ParseResponse(**resp.json()) def compile( self, source_path: Path, out_path: Path, optimization_level: int, invalidation_mode: py_compile.PycInvalidationMode | None, python_version: str, ): """ Compiles the given file with the given python versions and the parameters, and writes it to out_path. """ script_name = match_script(Path("./pyc_compiling"), python_version) port = self._ensure_running(python_version) params = { "source_path": str(source_path.absolute()), "optimization_level": str(optimization_level), "invalidation_mode": str(invalidation_mode.value) if invalidation_mode is not None else str(py_compile.PycInvalidationMode.TIMESTAMP.value), "script_name": script_name, } logging.debug("Sending compile request to port %d with params %s", port, params) resp = requests.post( f"http://127.0.0.1:{port}/compile", params=params, timeout=30, ) logging.debug("Got compile response: %s", resp.content) if not resp.ok: logging.error( "Failed to compile %s with %s via port %s: %s", source_path, python_version, port, resp.content, ) raise CompilationError with out_path.open("wb") as fp: fp.write(resp.content) def stop_all(self): for container in self.containers: container.stop() def main(): logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-8.8s [%(name)-10.10s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logging.info("Ensuring python versions are installed...") python = PythonExecutor() logging.info("Loading data...") with INPUT_PATH.open() as fp: data = json.load(fp) logging.info("Deserializing packages...") packages = [Package(**package) for package in data] try: results = work_threaded(packages, python) # results = work_serial(packages, python) finally: logging.info("Stopping python containers...") python.stop_all() logging.info("Serializing results...") with open("./results/reproducibility-results.json", "w") as fp: json.dump([result.model_dump(mode="json") for result in results], fp, indent=2) def work_threaded( packages: list[Package], python: PythonExecutor ) -> list[ReproductionResult]: results = [] no_packages = len(packages) with concurrent.futures.ThreadPoolExecutor(max_workers=200) as executor: futures = [ executor.submit(reproduce_package_cache, i, no_packages, package, python) for i, package in enumerate(packages) ] for future in concurrent.futures.as_completed(futures): result: ReproductionResult = future.result() results.append(result) return results def work_serial( packages: list[Package], python: PythonExecutor ) -> list[ReproductionResult]: """ Mainly intended for debugging purposes. """ results = [] no_packages = len(packages) for i, package in enumerate(packages): result = reproduce_package_cache(i, no_packages, package, python) results.append(result) return results def reproduce_package_cache( index: int, no_packages: int, package: Package, python: PythonExecutor ) -> ReproductionResult: logging.info( "[%d/%d] Working on package %s@%s.", index, no_packages, package.name, package.version, ) distribution_results = [] for distribution in package.distributions: if distribution.kind not in ( DistributionKind.SourceDistribution, DistributionKind.Wheel, DistributionKind.Egg ): distribution_results.append( DistributionResult.unsupported(distribution.kind) ) continue matches = reproduce_distribution_cache(package, distribution, python) distribution_results.append( DistributionResult( distribution_kind=distribution.kind, status=DistributionStatus.Success, files=matches, ) ) return ReproductionResult( package_name=package.name, package_version=package.version, distributions=distribution_results, ) def reproduce_distribution_cache( package: Package, distribution: Distribution, python: PythonExecutor ) -> list[FileReproductionResult]: unpacked_path = unpack_distribution(package, distribution) file_results = [] no_files = len(distribution.cache_files) for i, file in enumerate(distribution.cache_files): logging.debug("[%d/%d] Working on %s", i, no_files, file) result = reproduce_file(unpacked_path, file, python) file_results.append(result) shutil.rmtree(unpacked_path) return file_results def reproduce_file( unpacked_path: Path, file: CacheFile, python: PythonExecutor ) -> FileReproductionResult: # NOTE: ignoring python3.4 and older due to unavailable docker images (v1 no longer # compatible) if file.python_version <= PythonVersion.CPython34: return FileReproductionResult.unsupported_python(file) if file.source_path == Path("."): return FileReproductionResult.source_unavailable(file) source_file_path = unpacked_path.joinpath(file.source_path) cache_file_path = unpacked_path.joinpath(file.cache_path) source_mtime = datetime.fromtimestamp( os.path.getmtime(source_file_path), tz=BERLIN_TZ ) cache_mtime = datetime.fromtimestamp( os.path.getmtime(cache_file_path), tz=BERLIN_TZ ) if not source_file_path.exists(): return FileReproductionResult.source_unavailable(file) # NOTE: If for a single python version we get an actual parsing or compilation error, the file # is marked as unparsable / uncompilable. It is unclear whether trying multiple python versions # could still help here, and if so, how we would want to deal with errors. One idea could be to # have a flag whether parsing / compiling has worked at least once, and if so, we mark the file # as irreproducible. Otherwise, we return the respective parsing error? try: original_parsed_file = python.parse( cache_file_path, # latest release for this bytecode version PYTHON_RELEASE_TABLE[file.python_version.major_minor_string][0], ) except ParsingError: return FileReproductionResult.original_cache_unparsable(file) except NotFound: return FileReproductionResult.unsupported_python(file) except Exception: logging.exception( "Failed to parse %s for some reason for python version %s", cache_file_path, PYTHON_RELEASE_TABLE[file.python_version.major_minor_string][0], ) raise cache_valid = determine_cache_validity(source_file_path, original_parsed_file) cache_file_path = source_file_path.parent.joinpath(f"{source_file_path.stem}.pyc") for python_version_choice, exact_python_version in determine_python_versions( original_parsed_file.header.magic, file.python_version ): logging.debug("Using python version %s for compilation", exact_python_version) try: python.compile( source_file_path, cache_file_path, file.optimization_level, None, exact_python_version, ) except CompilationError: return FileReproductionResult.cache_uncompilable( file, original_parsed_file.header.invalidation_mode, cache_valid, source_mtime, cache_mtime, ) except NotFound: continue except Exception: logging.exception( "Failed to compile %s to %s for some reason for python version %s", source_file_path, cache_file_path, exact_python_version, ) raise try: rebuilt_parsed_file = python.parse( cache_file_path, # latest release for this bytecode version PYTHON_RELEASE_TABLE[file.python_version.major_minor_string][0], ) except ParsingError: return FileReproductionResult.rebuilt_cache_unparsable( file, rebuilt_parsed_file.header.invalidation_mode, cache_valid, ) except NotFound: continue except Exception: logging.exception( "Failed to parse %s for python version %s for some reason", cache_file_path, PYTHON_RELEASE_TABLE[file.python_version.major_minor_string][0], ) raise # NOTE: It is arguable whether this is a fitting equivalence test. if original_parsed_file.bytecode_table == rebuilt_parsed_file.bytecode_table: return FileReproductionResult( file_name=str(file.cache_path), status=FileStatus.Reproducible, python_version=exact_python_version, python_version_choice=python_version_choice, invalidation_mode=InvalidationMode.from_pyc_invalidation_mode( original_parsed_file.header.invalidation_mode ), cache_valid=cache_valid, source_modification_time=source_mtime, cache_modification_time=cache_mtime, optimization_level=file.optimization_level, additional_tag=file.tag, ) return FileReproductionResult( file_name=str(file.cache_path), status=FileStatus.Irreproducible, python_version="", python_version_choice=PythonVersionChoice.Unknown, invalidation_mode=InvalidationMode.from_pyc_invalidation_mode( original_parsed_file.header.invalidation_mode ), cache_valid=cache_valid, source_modification_time=source_mtime, cache_modification_time=cache_mtime, optimization_level=file.optimization_level, additional_tag=file.tag, ) def determine_cache_validity(source_file_path: Path, cache_file: ParseResponse) -> bool: # Following https://docs.python.org/3/reference/import.html#pyc-invalidation and PEP 552 match cache_file.header.invalidation_mode: case py_compile.PycInvalidationMode.UNCHECKED_HASH: return True case py_compile.PycInvalidationMode.CHECKED_HASH: hash = siphash24() with source_file_path.open("rb") as fp: hash.update(fp.read()) return hash.digest == cache_file.header.hash case py_compile.PycInvalidationMode.TIMESTAMP: header_mtime = cache_file.header.source_modification_time return header_mtime is not None and int( os.path.getmtime(source_file_path) ) == int(header_mtime.timestamp()) case _: return False def unpack_distribution(package: Package, distribution: Distribution) -> Path: match distribution.kind: case DistributionKind.SourceDistribution: return unpack_tarfile(distribution.dist_path, package.name, package.version) case DistributionKind.Wheel | DistributionKind.Egg: return unpack_zipfile(distribution.dist_path, package.name, package.version) case _: raise NotImplementedError def unpack_tarfile(tarfile_path: Path, package_name: str, package_version: str) -> Path: unpack_dir = Path(UNPACKED_PACKAGES_DIR).joinpath( f"{package_name}@{package_version}" ) unpack_dir.mkdir(parents=True, exist_ok=True) with tarfile.open(tarfile_path) as tf: tf.extractall(unpack_dir, filter="data") return unpack_dir def unpack_zipfile(zipfile_path: Path, package_name: str, package_version: str) -> Path: unpack_dir = Path(UNPACKED_PACKAGES_DIR).joinpath( f"{package_name}@{package_version}" ) unpack_dir.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(zipfile_path, "r") as zip_ref: zip_ref.extractall(unpack_dir) # Set modified and access times manually from the zipinfo, since python's ZipFile.extractall # rewrites (or perhaps rather, does not rewrite?) the timestamps when writing to disk to the # current time. for member in zip_ref.infolist(): if not member.is_dir(): file_path = unpack_dir.joinpath(member.filename) modtime = time.mktime(member.date_time + (0, 0, -1)) os.utime(file_path, (0, modtime)) return unpack_dir def match_script(path: Path, python_str_version: str) -> str: """ Returns the name of the `python_MIN_MAX.py` script that matches the given python version. """ major, minor, _ = python_str_version.split(".") python_version = MajorMinorVersion(int(major), int(minor)) for file in path.iterdir(): if file.name.startswith("python"): _, file_min, file_max = file.stem.split("_") file_min_version = MajorMinorVersion(int(file_min[0]), int(file_min[1:])) file_max_version = MajorMinorVersion(int(file_max[0]), int(file_max[1:])) if ( file_min_version <= python_version and file_max_version >= python_version ): return file.name raise RuntimeError("Matching python file should exist.") def determine_python_versions( magic: bytes, file_version: PythonVersion ) -> list[tuple[PythonVersionChoice, str]]: magic_number = struct.unpack(" 0: magic_number_tuples[0] = ( PythonVersionChoice.LatestMagic, magic_number_tuples[0][1], ) if len(minor_tuples) > 0: minor_tuples[0] = (PythonVersionChoice.LatestFileName, minor_tuples[0][1]) return magic_number_tuples + minor_tuples if __name__ == "__main__": main()