|
| 1 | +from __future__ import annotations |
| 2 | + |
1 | 3 | import asyncio |
| 4 | +import io |
2 | 5 | import pathlib |
3 | 6 | import sys |
| 7 | +import tarfile |
| 8 | +from typing import Any |
4 | 9 |
|
5 | 10 | import httpx |
6 | 11 |
|
7 | | -DOWNLOAD_URL = "https://bitbucket.org/vinay.sajip/simple_launcher/downloads/{}" |
| 12 | +DISTLIB_URL = "https://pypi.org/simple/distlib" |
8 | 13 | VENDOR_DIR = ( |
9 | 14 | pathlib.Path(__file__) |
10 | 15 | .parent.parent.joinpath("src", "installer", "_scripts") |
|
23 | 28 | ] |
24 | 29 |
|
25 | 30 |
|
26 | | -async def _download(client: httpx.AsyncClient, name: str) -> None: |
27 | | - url = DOWNLOAD_URL.format(name) |
28 | | - print(f" Fetching {url}") |
29 | | - resp = await client.get(url) |
30 | | - data = await resp.aread() |
31 | | - VENDOR_DIR.joinpath(name).write_bytes(data) |
| 31 | +async def _get_distlib_page(client: httpx.AsyncClient) -> Any: |
| 32 | + resp = await client.get( |
| 33 | + DISTLIB_URL, |
| 34 | + headers={"ACCEPT": "application/vnd.pypi.simple.v1+json"}, |
| 35 | + follow_redirects=True, |
| 36 | + ) |
| 37 | + return resp.json() |
| 38 | + |
| 39 | + |
| 40 | +def _get_link_from_response(json_response: dict[str, Any]) -> tuple[str, str] | None: |
| 41 | + version = max(version_str.split(".") for version_str in json_response["versions"]) |
| 42 | + filename = f'distlib-{".".join(version)}.tar.gz' |
| 43 | + for file_info in json_response["files"]: |
| 44 | + if file_info["filename"] == filename: |
| 45 | + return file_info["url"], filename |
| 46 | + return None |
| 47 | + |
| 48 | + |
| 49 | +async def _download_distlib(client: httpx.AsyncClient) -> bytes | None: |
| 50 | + distlib_page = await _get_distlib_page(client) |
| 51 | + data = None |
| 52 | + if pair := _get_link_from_response(distlib_page): |
| 53 | + url, filename = pair |
| 54 | + print(f" Fetching {filename}") |
| 55 | + resp = await client.get(url) |
| 56 | + data = await resp.aread() |
| 57 | + return data |
| 58 | + |
| 59 | + |
| 60 | +def _get_launcher_path(names: list[str], launcher: str) -> str | None: |
| 61 | + if paths := [name for name in names if launcher in name]: |
| 62 | + return paths[0] |
| 63 | + return None |
| 64 | + |
| 65 | + |
| 66 | +def _unpack_launchers_to_dir(distlib_tar: bytes) -> None: |
| 67 | + print("Unpacking launchers") |
| 68 | + with tarfile.open(fileobj=io.BytesIO(distlib_tar)) as file: |
| 69 | + for launcher_name in LAUNCHERS: |
| 70 | + if (path := _get_launcher_path(file.getnames(), launcher_name)) and ( |
| 71 | + launcher := file.extractfile(path) |
| 72 | + ): |
| 73 | + print(f" Unpacking {launcher_name}") |
| 74 | + VENDOR_DIR.joinpath(launcher_name).write_bytes(launcher.read()) |
32 | 75 |
|
33 | 76 |
|
34 | 77 | async def main() -> None: |
35 | 78 | print(f"Downloading into {VENDOR_DIR} ...") |
36 | 79 | async with httpx.AsyncClient() as client: |
37 | | - await asyncio.gather(*(_download(client, name) for name in LAUNCHERS)) |
| 80 | + data = await _download_distlib(client) |
| 81 | + if data is not None: |
| 82 | + _unpack_launchers_to_dir(data) |
| 83 | + print("Scripts update failed!") |
38 | 84 |
|
39 | 85 |
|
40 | 86 | def _patch_windows() -> None: |
|
0 commit comments