Python Scripts

pypi-info.py

#!/usr/bin/env -S uv run --script

# /// script
# requires-python = ">=3.14"
# dependencies = [
#   "packaging",
# ]
# ///
"""Query and download packages from a PEP 503/691 simple index.

By default the client speaks the PEP 691 JSON simple API (sending the
``application/vnd.pypi.simple.v1+json`` Accept header), which works against both
pypi.org and libraries.cgr.dev. The legacy ``/<pkg>/json`` path API is no longer
used (libraries.cgr.dev does not serve it). Pass --simple to use the HTML simple
API instead.
"""

from __future__ import annotations

import argparse
import base64
import html.parser
import json
import netrc as netrc_module
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime
from pathlib import Path
from typing import Any, Optional

from packaging.utils import InvalidSdistFilename, InvalidWheelFilename, parse_sdist_filename, parse_wheel_filename
from packaging.version import InvalidVersion, Version

DEFAULT_PYPI_URL = os.environ.get("PYPI_URL", "https://pypi.org/simple")
ECO_URL = "https://libraries.cgr.dev/python/simple"

ACCEPT_JSON = "application/vnd.pypi.simple.v1+json"
ACCEPT_HTML = "application/vnd.pypi.simple.v1+html"


def split_url_auth(url: str) -> tuple[str, Optional[tuple[str, str]]]:
    """Split any 'user:password@' credentials out of a URL.

    urllib neither uses userinfo in a URL for authentication nor strips it before
    connecting, so a 'user:token@host' URL crashes http.client's host parsing.
    Pulling the credentials out here lets us support the convenient inline form
    while keeping the secret out of the URL we connect to or log.

    Returns (clean_url, (username, password)) if userinfo was present, else
    (url, None). Username and password are percent-decoded.
    """
    parsed = urllib.parse.urlsplit(url)
    if parsed.username is None:
        return url, None

    host = parsed.hostname or ""
    if parsed.port is not None:
        host = f"{host}:{parsed.port}"
    clean_url = parsed._replace(netloc=host).geturl()

    username = urllib.parse.unquote(parsed.username)
    password = urllib.parse.unquote(parsed.password) if parsed.password else ""
    return clean_url, (username, password)


def get_auth(
    netrc_file: Optional[str], url: str, inline_auth: Optional[tuple[str, str]] = None
) -> Optional[tuple[str, str]]:
    """Return (username, password) for the given URL, or None.

    Priority:
      1. Credentials embedded in the URL (user:password@host)
      2. --netrc FILE (entry for the URL's host)
      3. AUTH_TOKEN env var ('user:pass' or a bare token with user 'token')
      4. ~/.netrc (entry for the URL's host)
    """
    if inline_auth:
        return inline_auth

    host = urllib.parse.urlparse(url).hostname or ""

    if netrc_file:
        nrc = netrc_module.netrc(netrc_file)
        auth = nrc.authenticators(host)
        if auth:
            login, _, password = auth
            return (login or "token", password or "")

    token = os.environ.get("AUTH_TOKEN")
    if token:
        if ":" in token:
            username, _, password = token.partition(":")
            return (username, password)
        return ("token", token)

    default_netrc = os.path.expanduser("~/.netrc")
    if os.path.exists(default_netrc):
        nrc = netrc_module.netrc(default_netrc)
        auth = nrc.authenticators(host)
        if auth:
            login, _, password = auth
            return (login or "token", password or "")

    return None


def _build_request(url: str, auth: Optional[tuple[str, str]], accept: Optional[str] = None) -> urllib.request.Request:
    """Build a urllib Request with optional HTTP Basic auth and Accept header."""
    req = urllib.request.Request(url)
    if auth:
        username, password = auth
        credentials = base64.b64encode(f"{username}:{password}".encode()).decode()
        req.add_header("Authorization", f"Basic {credentials}")
    if accept:
        req.add_header("Accept", accept)
    return req


class _AuthStrippingRedirectHandler(urllib.request.HTTPRedirectHandler):
    """Drop the Authorization header when a redirect crosses to another host.

    A download URL on the index host (e.g. libraries.cgr.dev) commonly 302s to a
    pre-signed object-store URL; forwarding our index credentials there both
    leaks the secret and is rejected (HTTP 400) by the signed-URL host.
    """

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        new = super().redirect_request(req, fp, code, msg, headers, newurl)
        if new is not None:
            old_host = urllib.parse.urlparse(req.full_url).hostname
            new_host = urllib.parse.urlparse(newurl).hostname
            if old_host != new_host:
                for key in list(new.headers):
                    if key.lower() == "authorization":
                        del new.headers[key]
        return new


_OPENER = urllib.request.build_opener(_AuthStrippingRedirectHandler())


def _urlopen(req: urllib.request.Request):
    return _OPENER.open(req)


def fetch_json(url: str, auth: Optional[tuple[str, str]] = None) -> dict[str, Any]:
    """Fetch a PEP 691 JSON document from a simple-index URL."""
    req = _build_request(url, auth, ACCEPT_JSON)
    try:
        with _urlopen(req) as response:
            return json.loads(response.read().decode())
    except urllib.error.HTTPError as e:
        print(f"Error fetching {url}: {e.code}", file=sys.stderr)
        sys.exit(1)
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)


class _LinkParser(html.parser.HTMLParser):
    """Parse anchor links from PEP 503 simple index HTML."""

    def __init__(self) -> None:
        super().__init__()
        self.links: list[tuple[str, str]] = []  # (href, text)
        self._current_href: Optional[str] = None

    def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None:
        if tag == "a":
            for name, value in attrs:
                if name == "href" and value:
                    self._current_href = value
                    break

    def handle_data(self, data: str) -> None:
        if self._current_href is not None:
            self.links.append((self._current_href, data.strip()))

    def handle_endtag(self, tag: str) -> None:
        if tag == "a":
            self._current_href = None


def fetch_html_files(url: str, auth: Optional[tuple[str, str]] = None) -> list[dict[str, Any]]:
    """Fetch an HTML simple-index page and return file dicts (filename, url)."""
    req = _build_request(url, auth, ACCEPT_HTML)
    try:
        with _urlopen(req) as response:
            content = response.read().decode("utf-8", errors="replace")
    except urllib.error.HTTPError as e:
        print(f"Error fetching {url}: {e.code}", file=sys.stderr)
        sys.exit(1)
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

    parser = _LinkParser()
    parser.feed(content)
    files = []
    for href, text in parser.links:
        if not text:
            continue
        absolute_url = urllib.parse.urljoin(url, href.split("#")[0])
        files.append({"filename": text, "url": absolute_url})
    return files


# Fallback for artifacts packaging can't parse (eggs, exes), e.g.
#   Pillow-6.1.0-py2.7-win32.egg, package-1.0.exe
_artifact_regex = re.compile(r"(?P<pkg>.*)-(?P<ver>[0-9][0-9.]*)(?:[-.][-\w.]*)?\.(?P<ext>[a-zA-Z]+)")


def parse_version(filename: str) -> Optional[Version]:
    """Parse a Version out of an artifact filename, or None if unrecognized."""
    try:
        _, ver, _, _ = parse_wheel_filename(filename)
        return ver
    except InvalidWheelFilename:
        pass
    except InvalidVersion:
        return None
    try:
        _, ver = parse_sdist_filename(filename)
        return ver
    except (InvalidSdistFilename, InvalidVersion):
        pass
    m = _artifact_regex.search(filename)
    if m:
        try:
            return Version(m.group("ver"))
        except InvalidVersion:
            return None
    return None


def parse_filename(filename: str) -> tuple[str, Version]:
    try:
        pkg, ver, _, _ = parse_wheel_filename(filename)
        return pkg, ver
    except InvalidWheelFilename:
        return parse_sdist_filename(filename)


def fetch_files(args: argparse.Namespace, pkg: str, auth: Optional[tuple[str, str]]) -> list[dict[str, Any]]:
    """Return the list of file dicts for a package in the configured mode.

    JSON mode dicts carry filename/url/size/upload-time; HTML mode only
    filename/url.
    """
    pkg_url = f"{args.url.rstrip('/')}/{pkg}/"
    if args.simple:
        return fetch_html_files(pkg_url, auth)
    data = fetch_json(pkg_url, auth)
    return data.get("files", [])


def files_to_rows(files: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Normalize PEP 691 file dicts into display rows, parsing version from name."""
    rows = []
    for f in files:
        filename = f["filename"]
        ver = parse_version(filename)
        if ver is None:
            print(f"Skipping unrecognized artifact {filename}", file=sys.stderr)
            continue
        rows.append(
            {
                "version": str(ver),
                "_version": ver,
                "upload_time": f.get("upload-time"),
                "filename": filename,
                "size": f.get("size"),
                "url": f.get("url", ""),
            }
        )
    return rows


def cmd_ls(args: argparse.Namespace) -> None:
    """List releases and files for a package (optionally filtered by version)."""
    auth = get_auth(args.netrc, args.url, args.inline_auth)
    rows = files_to_rows(fetch_files(args, args.package, auth))

    if args.version:
        try:
            want = Version(args.version)
        except InvalidVersion:
            print(f"Error: invalid version {args.version!r}", file=sys.stderr)
            sys.exit(1)
        rows = [r for r in rows if r["_version"] == want]

    if args.long:
        compute_deltas(rows)
        rows.sort(key=lambda r: (r["_version"], r["delta"], r["filename"]))
        print_rows(rows)
    else:
        rows.sort(key=lambda r: (r["_version"], r["filename"]))
        for row in rows:
            print(row["filename"])


def compute_deltas(rows: list[dict[str, Any]]) -> None:
    """Add a delta field (seconds from the oldest upload within each version)."""
    if not rows:
        return

    versions: dict[str, list[dict[str, Any]]] = {}
    for row in rows:
        versions.setdefault(row["version"], []).append(row)

    for version_rows in versions.values():
        timestamps = []
        for row in version_rows:
            iso = row["upload_time"]
            if iso:
                dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
                timestamps.append(int(dt.timestamp()))
            else:
                timestamps.append(None)

        known = [t for t in timestamps if t is not None]
        oldest = min(known) if known else 0

        for row, ts in zip(version_rows, timestamps):
            row["delta"] = (ts - oldest) if ts is not None else 0
            iso = row["upload_time"]
            row["upload_time"] = iso.split(".")[0] if iso else "-"


def print_rows(rows: list[dict[str, Any]]) -> None:
    """Print rows in the format: version upload_time delta filename size"""
    for row in rows:
        size = row["size"] if row["size"] is not None else "-"
        print(f"{row['version']} {row['upload_time']} {row['delta']} {row['filename']} {size}")


def cmd_json(args: argparse.Namespace) -> None:
    """Print the raw PEP 691 JSON for a package (optionally filtered by version)."""
    auth = get_auth(args.netrc, args.url, args.inline_auth)
    url = f"{args.url.rstrip('/')}/{args.package}/"
    data = fetch_json(url, auth)

    if args.version:
        try:
            want = Version(args.version)
        except InvalidVersion:
            print(f"Error: invalid version {args.version!r}", file=sys.stderr)
            sys.exit(1)
        data["files"] = [
            f for f in data.get("files", []) if parse_version(f["filename"]) == want
        ]

    json.dump(data, sys.stdout, indent=2)
    sys.stdout.write("\n")


def cmd_get(args: argparse.Namespace) -> None:
    """Download a file by name."""
    filename = args.filename
    output = args.output or filename

    try:
        pkg, _ = parse_filename(filename)
    except (InvalidWheelFilename, InvalidSdistFilename):
        print(f"Error: Could not parse package/version from {filename}", file=sys.stderr)
        sys.exit(1)

    auth = get_auth(args.netrc, args.url, args.inline_auth)
    files = fetch_files(args, pkg, auth)

    file_url = None
    for f in files:
        if f["filename"] == filename:
            file_url = f["url"]
            break

    if not file_url:
        print(f"Error: File {filename} not found in {pkg} index", file=sys.stderr)
        sys.exit(1)

    if Path(output).is_dir():
        output = Path(output) / filename

    # Only send the Basic auth header to the download URL if it shares the index
    # host; never leak index credentials to a third-party file host.
    file_host = urllib.parse.urlparse(file_url).hostname or ""
    index_host = urllib.parse.urlparse(args.url).hostname or ""
    file_auth = auth if file_host == index_host else None
    req = _build_request(file_url, file_auth)
    try:
        with _urlopen(req) as response, open(output, "wb") as fh:
            while True:
                chunk = response.read(65536)
                if not chunk:
                    break
                fh.write(chunk)
        print(f"wrote {output}", file=sys.stderr)
    except Exception as e:
        print(f"Error downloading {file_url} to {output}: {e}", file=sys.stderr)
        sys.exit(1)


def main() -> None:
    parser = argparse.ArgumentParser(
        prog="pypi-info",
        description="Query and download packages from a PEP 503/691 simple index",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=f"""
Default URL: {DEFAULT_PYPI_URL} (override with PYPI_URL env, --url, or --eco)

By default the PEP 691 JSON simple API is used (Accept:
{ACCEPT_JSON}). Pass --simple to use the HTML simple API instead.

Authentication (HTTP Basic; checked in order):
  1. Credentials embedded in the URL (https://user:password@host/...)
  2. Netrc file specified via --netrc
  3. AUTH_TOKEN environment variable ('user:pass' or bare token with user 'token')
  4. ~/.netrc (entry for the URL's host)
""",
    )
    url_group = parser.add_mutually_exclusive_group()
    url_group.add_argument(
        "-u", "--url",
        default=DEFAULT_PYPI_URL,
        metavar="URL",
        help=f"Base simple-index URL (default: {DEFAULT_PYPI_URL})",
    )
    url_group.add_argument(
        "-E", "--eco",
        dest="url",
        action="store_const",
        const=ECO_URL,
        help=f"shortcut for --url {ECO_URL}",
    )
    parser.add_argument(
        "-n", "--netrc",
        metavar="FILE",
        help="Netrc file for authentication (default: ~/.netrc)",
    )
    parser.add_argument(
        "--simple",
        action="store_true",
        help="Use the HTML simple API instead of PEP 691 JSON (incompatible with -l)",
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    # ls subcommand
    ls_parser = subparsers.add_parser("ls", help="List releases and files")
    ls_parser.add_argument("package", help="Package name")
    ls_parser.add_argument("version", nargs="?", help="Specific version (optional)")
    ls_parser.add_argument("-l", "--long", action="store_true", help="Long format output")

    # json subcommand
    json_parser = subparsers.add_parser("json", help="Print raw PEP 691 JSON")
    json_parser.add_argument("package", help="Package name")
    json_parser.add_argument("version", nargs="?", help="Specific version (optional)")

    # get subcommand
    get_parser = subparsers.add_parser("get", help="Download a file by name")
    get_parser.add_argument("filename", help="Filename to download")
    get_parser.add_argument("output", nargs="?", help="Output path (optional)")

    args = parser.parse_args()

    # Pull any 'user:password@' credentials out of the URL so the secret never ends
    # up in the URLs we connect to, join against, or log; route them to get_auth.
    args.url, args.inline_auth = split_url_auth(args.url)

    if args.simple and getattr(args, "long", False):
        parser.error("--long is incompatible with --simple (HTML lacks size/upload-time)")
    if args.simple and args.command == "json":
        parser.error("the 'json' command requires the JSON API; do not use --simple")

    if args.command == "ls":
        cmd_ls(args)
    elif args.command == "json":
        cmd_json(args)
    elif args.command == "get":
        cmd_get(args)


if __name__ == "__main__":
    main()