--- /dev/null
+import atexit
+import json
+import subprocess
+import tarfile
+import tempfile
+import traceback
+import venv
+import zipfile
+from argparse import ArgumentParser, Namespace
+from concurrent.futures import ThreadPoolExecutor
+from functools import lru_cache, partial
+from pathlib import Path
+from typing import ( # type: ignore # typing can't see Literal
+ Generator,
+ List,
+ Literal,
+ NamedTuple,
+ Optional,
+ Tuple,
+ Union,
+ cast,
+)
+from urllib.request import urlopen, urlretrieve
+
+PYPI_INSTANCE = "https://pypi.org/pypi"
+PYPI_TOP_PACKAGES = (
+ "https://hugovk.github.io/top-pypi-packages/top-pypi-packages-{days}-days.json"
+)
+INTERNAL_BLACK_REPO = f"{tempfile.gettempdir()}/__black"
+
+ArchiveKind = Union[tarfile.TarFile, zipfile.ZipFile]
+Days = Union[Literal[30], Literal[365]]
+
+subprocess.run = partial(subprocess.run, check=True) # type: ignore
+# https://github.com/python/mypy/issues/1484
+
+
+class BlackVersion(NamedTuple):
+ version: str
+ config: Optional[str] = None
+
+
+def get_pypi_download_url(package: str, version: Optional[str]) -> str:
+ with urlopen(PYPI_INSTANCE + f"/{package}/json") as page:
+ metadata = json.load(page)
+
+ if version is None:
+ sources = metadata["urls"]
+ else:
+ if version in metadata["releases"]:
+ sources = metadata["releases"][version]
+ else:
+ raise ValueError(
+ f"No releases found with version ('{version}') tag. "
+ f"Found releases: {metadata['releases'].keys()}"
+ )
+
+ for source in sources:
+ if source["python_version"] == "source":
+ break
+ else:
+ raise ValueError(f"Couldn't find any sources for {package}")
+
+ return cast(str, source["url"])
+
+
+def get_top_packages(days: Days) -> List[str]:
+ with urlopen(PYPI_TOP_PACKAGES.format(days=days)) as page:
+ result = json.load(page)
+
+ return [package["project"] for package in result["rows"]]
+
+
+def get_package_source(package: str, version: Optional[str]) -> str:
+ if package == "cpython":
+ if version is None:
+ version = "main"
+ return f"https://github.com/python/cpython/archive/{version}.zip"
+ elif package == "pypy":
+ if version is None:
+ version = "branch/default"
+ return (
+ f"https://foss.heptapod.net/pypy/pypy/repository/{version}/archive.tar.bz2"
+ )
+ else:
+ return get_pypi_download_url(package, version)
+
+
+def get_archive_manager(local_file: str) -> ArchiveKind:
+ if tarfile.is_tarfile(local_file):
+ return tarfile.open(local_file)
+ elif zipfile.is_zipfile(local_file):
+ return zipfile.ZipFile(local_file)
+ else:
+ raise ValueError("Unknown archive kind.")
+
+
+def get_first_archive_member(archive: ArchiveKind) -> str:
+ if isinstance(archive, tarfile.TarFile):
+ return archive.getnames()[0]
+ elif isinstance(archive, zipfile.ZipFile):
+ return archive.namelist()[0]
+
+
+def download_and_extract(package: str, version: Optional[str], directory: Path) -> Path:
+ source = get_package_source(package, version)
+
+ local_file, _ = urlretrieve(source, directory / f"{package}-src")
+ with get_archive_manager(local_file) as archive:
+ archive.extractall(path=directory)
+ result_dir = get_first_archive_member(archive)
+ return directory / result_dir
+
+
+def get_package(
+ package: str, version: Optional[str], directory: Path
+) -> Optional[Path]:
+ try:
+ return download_and_extract(package, version, directory)
+ except Exception:
+ print(f"Caught an exception while downloading {package}.")
+ traceback.print_exc()
+ return None
+
+
+DEFAULT_SLICE = slice(None) # for flake8
+
+
+def download_and_extract_top_packages(
+ directory: Path,
+ days: Days = 365,
+ workers: int = 8,
+ limit: slice = DEFAULT_SLICE,
+) -> Generator[Path, None, None]:
+ with ThreadPoolExecutor(max_workers=workers) as executor:
+ bound_downloader = partial(get_package, version=None, directory=directory)
+ for package in executor.map(bound_downloader, get_top_packages(days)[limit]):
+ if package is not None:
+ yield package
+
+
+def git_create_repository(repo: Path) -> None:
+ subprocess.run(["git", "init"], cwd=repo)
+ git_add_and_commit(msg="Initial commit", repo=repo)
+
+
+def git_add_and_commit(msg: str, repo: Path) -> None:
+ subprocess.run(["git", "add", "."], cwd=repo)
+ subprocess.run(["git", "commit", "-m", msg, "--allow-empty"], cwd=repo)
+
+
+def git_switch_branch(
+ branch: str, repo: Path, new: bool = False, from_branch: Optional[str] = None
+) -> None:
+ args = ["git", "checkout"]
+ if new:
+ args.append("-b")
+ args.append(branch)
+ if from_branch:
+ args.append(from_branch)
+ subprocess.run(args, cwd=repo)
+
+
+def init_repos(options: Namespace) -> Tuple[Path, ...]:
+ options.output.mkdir(exist_ok=True)
+
+ if options.top_packages:
+ source_directories = tuple(
+ download_and_extract_top_packages(
+ directory=options.output,
+ workers=options.workers,
+ limit=slice(None, options.top_packages),
+ )
+ )
+ else:
+ source_directories = (
+ download_and_extract(
+ package=options.pypi_package,
+ version=options.version,
+ directory=options.output,
+ ),
+ )
+
+ for source_directory in source_directories:
+ git_create_repository(source_directory)
+
+ if options.black_repo is None:
+ subprocess.run(
+ ["git", "clone", "https://github.com/psf/black.git", INTERNAL_BLACK_REPO],
+ cwd=options.output,
+ )
+ options.black_repo = options.output / INTERNAL_BLACK_REPO
+
+ return source_directories
+
+
+@lru_cache(8)
+def black_runner(version: str, black_repo: Path) -> Path:
+ directory = tempfile.TemporaryDirectory()
+ venv.create(directory.name, with_pip=True)
+
+ python = Path(directory.name) / "bin" / "python"
+ subprocess.run([python, "-m", "pip", "install", "-e", black_repo])
+
+ atexit.register(directory.cleanup)
+ return python
+
+
+def format_repo_with_version(
+ repo: Path,
+ from_branch: Optional[str],
+ black_repo: Path,
+ black_version: BlackVersion,
+ input_directory: Path,
+) -> str:
+ current_branch = f"black-{black_version.version}"
+ git_switch_branch(black_version.version, repo=black_repo)
+ git_switch_branch(current_branch, repo=repo, new=True, from_branch=from_branch)
+
+ format_cmd: List[Union[Path, str]] = [
+ black_runner(black_version.version, black_repo),
+ (black_repo / "black.py").resolve(),
+ ".",
+ ]
+ if black_version.config:
+ format_cmd.extend(["--config", input_directory / black_version.config])
+
+ subprocess.run(format_cmd, cwd=repo, check=False) # ensure the process
+ # continuess to run even it can't format some files. Reporting those
+ # should be enough
+ git_add_and_commit(f"Format with black:{black_version.version}", repo=repo)
+
+ return current_branch
+
+
+def format_repos(repos: Tuple[Path, ...], options: Namespace) -> None:
+ black_versions = tuple(
+ BlackVersion(*version.split(":")) for version in options.versions
+ )
+
+ for repo in repos:
+ from_branch = None
+ for black_version in black_versions:
+ from_branch = format_repo_with_version(
+ repo=repo,
+ from_branch=from_branch,
+ black_repo=options.black_repo,
+ black_version=black_version,
+ input_directory=options.input,
+ )
+ git_switch_branch("main", repo=repo)
+
+ git_switch_branch("main", repo=options.black_repo)
+
+
+def main() -> None:
+ parser = ArgumentParser(
+ description="""Black Gallery is a script that
+ automates the process of applying different Black versions to a selected
+ PyPI package and seeing the results between versions."""
+ )
+
+ group = parser.add_mutually_exclusive_group(required=True)
+ group.add_argument("-p", "--pypi-package", help="PyPI package to download.")
+ group.add_argument(
+ "-t", "--top-packages", help="Top n PyPI packages to download.", type=int
+ )
+
+ parser.add_argument("-b", "--black-repo", help="Black's Git repository.", type=Path)
+ parser.add_argument(
+ "-v",
+ "--version",
+ help=(
+ "Version for given PyPI package. Will be discarded if used with -t option."
+ ),
+ )
+ parser.add_argument(
+ "-w",
+ "--workers",
+ help=(
+ "Maximum number of threads to download with at the same time. "
+ "Will be discarded if used with -p option."
+ ),
+ )
+ parser.add_argument(
+ "-i",
+ "--input",
+ default=Path("/input"),
+ type=Path,
+ help="Input directory to read configuration.",
+ )
+ parser.add_argument(
+ "-o",
+ "--output",
+ default=Path("/output"),
+ type=Path,
+ help="Output directory to download and put result artifacts.",
+ )
+ parser.add_argument("versions", nargs="*", default=("main",), help="")
+
+ options = parser.parse_args()
+ repos = init_repos(options)
+ format_repos(repos, options)
+
+
+if __name__ == "__main__":
+ main()
projects / etc / vim.git / blobdiff