pip-from.py

import httpx

import argparse
from http import HTTPStatus
from datetime import datetime
from contextlib import suppress
from collections import namedtuple

from typing import Dict, List, Optional, Tuple, Union

PackageVersion = namedtuple("PackageVersion", ["package_name", "release_name"])

DUMMY_DATE = datetime.fromtimestamp(99999999999)

PYPI_API_ENDPOINT = "https://pypi.org/pypi/{package_name}/json"
DOI_ENDPOINT = "https://doi.org/api/handles/{doi}"


def get_package_releases(
    package_name: str,
) -> Optional[Dict[str, List[Dict[str, Union[str, Dict[str, str]]]]]]:
    api_response = httpx.get(PYPI_API_ENDPOINT.format(package_name=package_name))

    if api_response.status_code != HTTPStatus.OK:
        return None

    if releases := api_response.json().get("releases"):
        return releases


def find_package_release(package_name: str, pivot_date: datetime) -> Optional[PackageVersion]:
    def release_entry_to_date(
        entry: List[Dict[str, Union[str, Dict[str, str]]]]
    ) -> datetime:
        if entry and (upload_time := entry[0].get("upload_time")):
            return datetime.fromisoformat(upload_time)
        return DUMMY_DATE

    if not (releases := get_package_releases(package_name)):
        return None

    release_name, _release_information = min(
        releases.items(),
        key=lambda key_value: abs(release_entry_to_date(key_value[1]) - pivot_date),
    )

    return PackageVersion(package_name, release_name)


def build_command(packages: List[PackageVersion]) -> str:
    packages = " ".join(
        f"{package_name}=={package_version}"
        for package_name, package_version in packages
    )
    return f"pip install {packages}"


def try_parse_date(input_date: str) -> Optional[datetime]:
    with suppress(ValueError):
        return datetime.strptime(input_date, "%Y/%m/%d")  # YYYY/MM/DD


def try_doi2date(input_doi: str) -> Optional[datetime]:
    response = httpx.get(DOI_ENDPOINT.format(doi=input_doi))

    if response.status_code != HTTPStatus.OK:
        return

    if not (response_json := response.json()) or response_json.get("responseCode") != 1:
        return

    for value in response_json["values"]:
        if timestamp := value.get("timestamp"):
            return datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%SZ")


def parse_arguments() -> Optional[Tuple[datetime, List[str]]]:
    parser = argparse.ArgumentParser()
    parser.add_argument("package_names", metavar="package_names", nargs="+")
    date_group = parser.add_mutually_exclusive_group(required=True)
    date_group.add_argument("--date", type=str)
    date_group.add_argument("--doi", type=str)

    args = parser.parse_args()

    if args.date:
        if not (date := try_parse_date(args.date)):
            return

    elif args.doi:
        if not (date := try_doi2date(args.doi)):
            return

    return date, args.package_names


if __name__ == "__main__":
    if not (parsed_arguments := parse_arguments()):
        print(parsed_arguments)
        exit()

    date, package_names = parsed_arguments
    package_versions = []

    for package_name in package_names:
        if not (package_version := find_package_release(package_name, date)):
            print(
                f"There is no release before {date.strftime('%Y/%m/%d')} for package {package_name}."
            )
            break
        package_versions.append(package_version)
    else:
        print(build_command(package_versions))