yt-dlp/devscripts/update_requirements.py

#!/usr/bin/env python3
from __future__ import annotations

# Allow direct execution
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import collections.abc
import contextlib
import dataclasses
import hashlib
import io
import itertools
import json
import pathlib
import re
import typing
import urllib.parse
import zipfile

from devscripts.tomlparse import parse_toml
from devscripts.utils import (
    call_github_api,
    request,
    run_process,
    zipf_files_and_folders,
)


BASE_PATH = pathlib.Path(__file__).parent.parent
PYPROJECT_PATH = BASE_PATH / 'pyproject.toml'
MAKEFILE_PATH = BASE_PATH / 'Makefile'
LOCKFILE_PATH = BASE_PATH / 'uv.lock'
REQUIREMENTS_PATH = BASE_PATH / 'bundle/requirements'
REQS_OUTPUT_TMPL = 'requirements-{}.txt'
CUSTOM_COMPILE_COMMAND = 'python -m devscripts.update_requirements'

EXTRAS_TABLE = 'project.optional-dependencies'
GROUPS_TABLE = 'dependency-groups'

PINNED_EXTRAS = {
    'pin': 'default',
    'pin-curl-cffi': 'curl-cffi',
    'pin-secretstorage': 'secretstorage',
    'pin-deno': 'deno',
}

WELLKNOWN_PACKAGES = {
    'deno': {'owner': 'denoland', 'repo': 'deno'},
    'protobug': {'owner': 'yt-dlp', 'repo': 'protobug'},
    'yt-dlp-ejs': {'owner': 'yt-dlp', 'repo': 'ejs'},
}

EJS_ASSETS = {
    'yt.solver.lib.js': False,
    'yt.solver.lib.min.js': False,
    'yt.solver.deno.lib.js': True,
    'yt.solver.bun.lib.js': True,
    'yt.solver.core.min.js': False,
    'yt.solver.core.js': True,
}

EJS_TEMPLATE = '''\
# This file is generated by devscripts/update_requirements.py. DO NOT MODIFY!

VERSION = {version!r}
HASHES = {{
{hash_mapping}
}}
'''


@dataclasses.dataclass
class Target:
    extras: list[str] = dataclasses.field(default_factory=list)
    groups: list[str] = dataclasses.field(default_factory=list)
    prune_packages: list[str] = dataclasses.field(default_factory=list)
    omit_packages: list[str] = dataclasses.field(default_factory=list)


LINUX_TARGET = Target(
    extras=['default', 'curl-cffi', 'secretstorage'],
    groups=['pyinstaller'],
)
WIN64_TARGET = Target(
    extras=['default', 'curl-cffi'],
)

BUNDLE_TARGETS = {
    'linux-x86_64': LINUX_TARGET,
    'linux-aarch64': LINUX_TARGET,
    'linux-armv7l': LINUX_TARGET,
    'musllinux-x86_64': LINUX_TARGET,
    'musllinux-aarch64': LINUX_TARGET,
    'win-x64': WIN64_TARGET,
    'win-arm64': WIN64_TARGET,
    'win-x86': Target(extras=['default']),
    'macos': Target(
        extras=['default', 'curl-cffi'],
        # NB: Resolve delocate and PyInstaller together since they share dependencies
        groups=['delocate', 'pyinstaller'],
        # curl-cffi and cffi don't provide universal2 wheels, so only directly install their deps
        omit_packages=['curl-cffi', 'cffi'],
    ),
    # We fuse our own universal2 wheels for curl-cffi+cffi, so we need a separate requirements file
    'macos-curl_cffi': Target(
        extras=['curl-cffi'],
        # Only need curl-cffi+cffi in this requirements file; their deps are installed directly
        # XXX: Try to keep these in sync with curl-cffi's and cffi's transitive dependencies
        prune_packages=['rich'],
        omit_packages=['certifi', 'pycparser'],
    ),
}

PYINSTALLER_BUILDS_TARGETS = {
    'win-x64-pyinstaller': 'win_amd64',
    'win-x86-pyinstaller': 'win32',
    'win-arm64-pyinstaller': 'win_arm64',
}

PYINSTALLER_BUILDS_TMPL = '''\
{}pyinstaller @ {} \\
    --hash={}
'''


def call_pypi_api(project: str) -> dict[str, dict[str, typing.Any]]:
    print(f'Fetching package info from PyPI API: {project}', file=sys.stderr)
    headers = {
        'Accept': 'application/json',
        'User-Agent': 'yt-dlp',
    }
    with request(f'https://pypi.org/pypi/{project}/json', headers=headers) as resp:
        return json.load(resp)


def fetch_latest_github_release(owner: str, repo: str) -> dict[str, dict[str, typing.Any]]:
    print(f'Fetching latest release from Github API: {owner}/{repo}', file=sys.stderr)
    return call_github_api(f'/repos/{owner}/{repo}/releases/latest')


def fetch_github_tags(
    owner: str,
    repo: str,
    *,
    fetch_tags: list[str] | None = None,
) -> dict[str, dict[str, typing.Any]]:

    needed_tags = set(fetch_tags or [])
    results = {}

    for page in itertools.count(1):
        print(f'Fetching tags list page {page} from Github API: {owner}/{repo}', file=sys.stderr)
        tags = call_github_api(
            f'/repos/{owner}/{repo}/tags',
            query={'per_page': '100', 'page': page})

        if not tags:
            break

        if not fetch_tags:
            # Fetch all tags
            results.update({tag['name']: tag for tag in tags})
            continue

        for tag in tags:
            clean_tag = tag['name'].removeprefix('v')
            possible_matches = {tag['name'], clean_tag}
            # Normalize calver tags like 2024.01.01 to 2024.1.1
            with contextlib.suppress(ValueError):
                possible_matches.add('.'.join(map(str, map(int, clean_tag.split('.')))))

            for name in possible_matches:
                if name in needed_tags:
                    needed_tags.remove(name)
                    results[name] = tag
                    break  # from inner loop

            if not needed_tags:
                break

        if not needed_tags:
            break

    return results


def generate_table_lines(
    table_name: str,
    table: dict[str, str | bool | int | float | list[str | dict[str, str]]],
) -> collections.abc.Iterator[str]:
    SUPPORTED_TYPES = (str, bool, int, float, list)

    yield f'[{table_name}]\n'
    for name, value in table.items():
        if not isinstance(value, SUPPORTED_TYPES):
            raise TypeError(
                f'expected {"/".join(t.__name__ for t in SUPPORTED_TYPES)} value, '
                f'got {type(value).__name__}')

        if not isinstance(value, list):
            yield f'{name} = {json.dumps(value)}\n'
            continue

        yield f'{name} = ['
        if value:
            yield '\n'
        for element in value:
            yield '    '
            if isinstance(element, dict):
                yield '{ ' + ', '.join(f'{k} = {json.dumps(v)}' for k, v in element.items()) + ' }'
            else:
                yield f'"{element}"'
            yield ',\n'
        yield ']\n'
    yield '\n'


def replace_table_in_pyproject(
    pyproject_text: str,
    table_name: str,
    table: dict[str, str | bool | int | float | list[str | dict[str, str]]],
) -> collections.abc.Iterator[str]:
    INSIDE = 1
    BEYOND = 2

    state = 0
    for line in pyproject_text.splitlines(True):
        if state == INSIDE:
            if line == '\n':
                state = BEYOND
            continue
        if line != f'[{table_name}]\n' or state == BEYOND:
            yield line
            continue
        yield from generate_table_lines(table_name, table)
        state = INSIDE


def modify_and_write_pyproject(
    pyproject_text: str,
    table_name: str,
    table: dict[str, str | bool | int | float | list[str | dict[str, str]]],
) -> None:
    with PYPROJECT_PATH.open(mode='w') as f:
        f.writelines(replace_table_in_pyproject(pyproject_text, table_name, table))


def run_uv_export(
    *,
    extras: list[str] | None = None,
    groups: list[str] | None = None,
    prune_packages: list[str] | None = None,
    omit_packages: list[str] | None = None,
    bare: bool = False,
    output_file: pathlib.Path | None = None,
) -> str:
    return run_process(
        'uv', 'export',
        '--no-python-downloads',
        '--quiet',
        '--no-progress',
        '--color=never',
        '--format=requirements.txt',
        '--frozen',
        '--refresh',
        '--no-emit-project',
        '--no-default-groups',
        *(f'--extra={extra}' for extra in (extras or [])),
        *(f'--group={group}' for group in (groups or [])),
        *(f'--prune={package}' for package in (prune_packages or [])),
        *(f'--no-emit-package={package}' for package in (omit_packages or [])),
        *(['--no-annotate', '--no-hashes', '--no-header'] if bare else []),
        *([f'--output-file={output_file.relative_to(BASE_PATH)}'] if output_file else []),
    ).stdout


def run_pip_compile(
    *args: str,
    input_line: str,
    output_file: pathlib.Path | None = None,
    env: dict[str, str] | None = None,
) -> str:
    return run_process(
        'uv', 'pip', 'compile',
        '--no-python-downloads',
        '--quiet',
        '--no-progress',
        '--color=never',
        '--format=requirements.txt',
        '--refresh',
        '--generate-hashes',
        '--no-strip-markers',
        f'--custom-compile-command={CUSTOM_COMPILE_COMMAND}',
        '--universal',
        *args,
        *([f'--output-file={output_file.relative_to(BASE_PATH)}'] if output_file else []),
        '-',  # Read from stdin
        input=f'{input_line}\n',
        env=env,
    ).stdout


def makefile_variables(
    prefix: str,
    filetypes: list[str] | None = None,
    *,
    version: str | None = None,
    name: str | None = None,
    digest: str | None = None,
    data: bytes | None = None,
    keys_only: bool = False,
) -> dict[str, str | None]:

    variables = {
        f'{prefix}_VERSION': version,
        f'{prefix}_WHEEL_NAME': name,
        f'{prefix}_WHEEL_HASH': digest,
    }
    for ft in filetypes or []:
        variables.update({
            f'{prefix}_{ft.upper()}_FOLDERS': None,
            f'{prefix}_{ft.upper()}_FILES': None,
        })

    if keys_only:
        return variables

    if not all(arg is not None for arg in (version, name, digest, not filetypes or data)):
        raise ValueError(
            'makefile_variables requires version, name, digest, '
            f'{"and data, " if filetypes else ""}OR keys_only=True')

    if filetypes:
        with io.BytesIO(data) as buf, zipfile.ZipFile(buf) as zipf:
            for ft in filetypes:
                files, folders = zipf_files_and_folders(zipf, f'*.{ft.lower()}')
                variables[f'{prefix}_{ft.upper()}_FOLDERS'] = ' '.join(folders)
                variables[f'{prefix}_{ft.upper()}_FILES'] = ' '.join(files)

    return variables


def ejs_makefile_variables(**kwargs) -> dict[str, str | None]:
    return makefile_variables('EJS', filetypes=['PY', 'JS'], **kwargs)


def update_ejs(verify: bool = False) -> dict[str, tuple[str | None, str | None]] | None:
    PACKAGE_NAME = 'yt-dlp-ejs'
    PREFIX = f'    "{PACKAGE_NAME}=='
    LIBRARY_NAME = PACKAGE_NAME.replace('-', '_')
    PACKAGE_PATH = BASE_PATH / 'yt_dlp/extractor/youtube/jsc/_builtin/vendor'

    current_version = None
    with PYPROJECT_PATH.open() as file:
        for line in file:
            if not line.startswith(PREFIX):
                continue
            current_version, _, _ = line.removeprefix(PREFIX).partition('"')

    if not current_version:
        raise ValueError(f'{PACKAGE_NAME} dependency line could not be found')

    makefile_info = ejs_makefile_variables(keys_only=True)
    prefixes = tuple(f'{key} = ' for key in makefile_info)
    with MAKEFILE_PATH.open() as file:
        for line in file:
            if not line.startswith(prefixes):
                continue
            key, _, val = line.partition(' = ')
            makefile_info[key] = val.rstrip()

    info = fetch_latest_github_release('yt-dlp', 'ejs')
    version = info['tag_name']
    if version == current_version:
        print(f'{PACKAGE_NAME} is up to date! ({version})', file=sys.stderr)
        return

    print(f'Updating {PACKAGE_NAME} from {current_version} to {version}', file=sys.stderr)
    hashes = []
    wheel_info = {}
    for asset in info['assets']:
        name = asset['name']
        digest = asset['digest']

        is_wheel = name.startswith(f'{LIBRARY_NAME}-') and name.endswith('.whl')
        if not is_wheel and name not in EJS_ASSETS:
            continue

        with request(asset['browser_download_url']) as resp:
            data = resp.read()

        # verify digest from github
        algo, _, expected = digest.partition(':')
        hexdigest = hashlib.new(algo, data).hexdigest()
        if hexdigest != expected:
            raise ValueError(f'downloaded attest mismatch ({hexdigest!r} != {expected!r})')

        if is_wheel:
            wheel_info = ejs_makefile_variables(version=version, name=name, digest=digest, data=data)
            continue

        # calculate sha3-512 digest
        asset_hash = hashlib.sha3_512(data).hexdigest()
        hashes.append(f'    {name!r}: {asset_hash!r},')

        if EJS_ASSETS[name]:
            (PACKAGE_PATH / name).write_bytes(data)

    hash_mapping = '\n'.join(hashes)
    if missing_assets := [asset_name for asset_name in EJS_ASSETS if asset_name not in hash_mapping]:
        raise ValueError(f'asset(s) not found in release: {", ".join(missing_assets)}')

    if missing_fields := [key for key in makefile_info if not wheel_info.get(key)]:
        raise ValueError(f'wheel info not found in release: {", ".join(missing_fields)}')

    (PACKAGE_PATH / '_info.py').write_text(EJS_TEMPLATE.format(
        version=version,
        hash_mapping=hash_mapping,
    ))

    content = PYPROJECT_PATH.read_text()
    updated = content.replace(PREFIX + current_version, PREFIX + version)
    PYPROJECT_PATH.write_text(updated)

    makefile = MAKEFILE_PATH.read_text()
    for key in wheel_info:
        makefile = makefile.replace(f'{key} = {makefile_info[key]}', f'{key} = {wheel_info[key]}')
    MAKEFILE_PATH.write_text(makefile)

    return update_requirements(upgrade_only=PACKAGE_NAME, verify=verify)


@dataclasses.dataclass
class Dependency:
    name: str
    exact_version: str | None
    direct_reference: str | None
    specifier: str | None
    markers: str | None


def parse_version_from_dist(filename: str, name: str, *, require: bool = False) -> str | None:
    # Ref: https://packaging.python.org/en/latest/specifications/binary-distribution-format/#escaping-and-unicode
    normalized_name = re.sub(r'[-_.]+', '-', name).lower().replace('-', '_')

    # Ref: https://packaging.python.org/en/latest/specifications/version-specifiers/#version-specifiers
    if mobj := re.match(rf'{normalized_name}-(?P<version>[^-]+)-', filename):
        return mobj.group('version')

    if require:
        raise ValueError(f'unable to parse version from distribution filename: {filename}')

    return None


def parse_dependency(line: str, *, require_version: bool = False) -> Dependency:
    # Ref: https://packaging.python.org/en/latest/specifications/name-normalization/
    NAME_RE = re.compile(r'^(?P<name>[A-Z0-9](?:[A-Z0-9._-]*[A-Z0-9])?)', re.IGNORECASE)

    line = line.rstrip().removesuffix('\\')
    mobj = NAME_RE.match(line)
    if not mobj:
        raise ValueError(f'unable to parse Dependency.name from line:\n    {line}')

    name = mobj.group('name')
    rest = line[len(name):].lstrip()
    specifier_or_direct_reference, _, markers = map(str.strip, rest.partition(';'))
    specifier, _, direct_reference = map(str.strip, specifier_or_direct_reference.partition('@'))

    exact_version = None
    if ',' not in specifier and specifier.startswith('=='):
        exact_version = specifier[2:]

    # Ref: https://packaging.python.org/en/latest/specifications/binary-distribution-format/
    if direct_reference and not exact_version:
        filename = urllib.parse.urlparse(direct_reference).path.rpartition('/')[2]
        if filename.endswith(('.tar.gz', '.whl')):
            exact_version = parse_version_from_dist(filename, name)

    if require_version and not exact_version:
        raise ValueError(f'unable to parse Dependency.exact_version from line:\n    {line}')

    return Dependency(
        name=name,
        exact_version=exact_version,
        direct_reference=direct_reference or None,
        specifier=specifier or None,
        markers=markers or None)


def package_diff_dict(
    old_dict: dict[str, str],
    new_dict: dict[str, str],
) -> dict[str, tuple[str | None, str | None]]:
    """
    @param old_dict: Dictionary w/ package names as keys and old package versions as values
    @param new_dict: Dictionary w/ package names as keys and new package versions as values
    @returns         Dictionary w/ package names as keys and tuples of (old_ver, new_ver) as values
    """
    ret_dict = {}

    for name, new_version in new_dict.items():
        if name not in old_dict:
            ret_dict[name] = (None, new_version)
            continue

        old_version = old_dict[name]
        if new_version != old_version:
            ret_dict[name] = (old_version, new_version)

    for name, old_version in old_dict.items():
        if name not in new_dict:
            ret_dict[name] = (old_version, None)

    return ret_dict


def get_lock_packages(lock: dict[str, typing.Any]) -> dict[str, str]:
    return {
        package['name']: package['version']
        for package in lock['package']
        if package.get('version')
    }


def evaluate_requirements_txt(old_txt: str, new_txt: str) -> dict[str, tuple[str | None, str | None]]:
    old_dict = {}
    new_dict = {}

    for txt, dct in [(old_txt, old_dict), (new_txt, new_dict)]:
        for line in txt.splitlines():
            if not line.strip() or line.startswith(('#', ' ')):
                continue
            dep = parse_dependency(line, require_version=True)
            dct.update({dep.name: dep.exact_version})

    return package_diff_dict(old_dict, new_dict)


def update_requirements(
    upgrade_only: str | None = None,
    verify: bool = False,
) -> dict[str, tuple[str | None, str | None]]:
    # Are we upgrading all packages or only one (e.g. 'yt-dlp-ejs' or 'protobug')?
    upgrade_arg = f'--upgrade-package={upgrade_only}' if upgrade_only else '--upgrade'

    pyproject_text = PYPROJECT_PATH.read_text()
    pyproject_toml = parse_toml(pyproject_text)
    extras = pyproject_toml['project']['optional-dependencies']

    # Remove pinned extras so they don't muck up the lockfile during generation/upgrade
    for pinned_extra_name in PINNED_EXTRAS:
        extras.pop(pinned_extra_name, None)

    # Write an intermediate pyproject.toml to use for generating lockfile and bundle requirements
    modify_and_write_pyproject(pyproject_text, table_name=EXTRAS_TABLE, table=extras)

    old_lock = None
    if LOCKFILE_PATH.is_file():
        old_lock = parse_toml(LOCKFILE_PATH.read_text())

    # If verifying, set UV_EXCLUDE_NEWER env var with the last timestamp recorded in uv.lock
    env = None
    if verify or upgrade_only in pyproject_toml['tool']['uv']['exclude-newer-package']:
        env = os.environ.copy()
        env['UV_EXCLUDE_NEWER'] = old_lock['options']['exclude-newer']
        print(f'Setting UV_EXCLUDE_NEWER={env["UV_EXCLUDE_NEWER"]}', file=sys.stderr)

    # Generate/upgrade lockfile
    print(f'Running: uv lock {upgrade_arg}', file=sys.stderr)
    run_process('uv', 'lock', upgrade_arg, env=env)

    # Record diff in uv.lock packages
    old_packages = get_lock_packages(old_lock) if old_lock else {}
    new_packages = get_lock_packages(parse_toml(LOCKFILE_PATH.read_text()))
    all_updates = package_diff_dict(old_packages, new_packages)

    # Update Windows PyInstaller requirements; need to compare before & after .txt's for reporting
    if not upgrade_only or upgrade_only.lower() == 'pyinstaller':
        info = fetch_latest_github_release('yt-dlp', 'Pyinstaller-Builds')
        for target_suffix, asset_tag in PYINSTALLER_BUILDS_TARGETS.items():
            asset_info = next(asset for asset in info['assets'] if asset_tag in asset['name'])
            pyinstaller_version = parse_version_from_dist(
                asset_info['name'], 'pyinstaller', require=True)
            pyinstaller_builds_deps = run_pip_compile(
                '--no-emit-package=pyinstaller',
                upgrade_arg,
                input_line=f'pyinstaller=={pyinstaller_version}',
                env=env)
            requirements_path = REQUIREMENTS_PATH / REQS_OUTPUT_TMPL.format(target_suffix)
            if requirements_path.is_file():
                old_requirements_txt = requirements_path.read_text()
            else:
                old_requirements_txt = ''

            new_requirements_txt = PYINSTALLER_BUILDS_TMPL.format(
                pyinstaller_builds_deps, asset_info['browser_download_url'], asset_info['digest'])
            requirements_path.write_text(new_requirements_txt)
            all_updates.update(evaluate_requirements_txt(old_requirements_txt, new_requirements_txt))

    # Export bundle requirements; any updates to these are already recorded w/ uv.lock package diff
    for target_suffix, target in BUNDLE_TARGETS.items():
        run_uv_export(
            extras=target.extras,
            groups=target.groups,
            prune_packages=target.prune_packages,
            omit_packages=target.omit_packages,
            output_file=REQUIREMENTS_PATH / REQS_OUTPUT_TMPL.format(target_suffix))

    # Export group requirements; any updates to these are already recorded w/ uv.lock package diff
    for group in ('build',):
        run_uv_export(
            groups=[group],
            output_file=REQUIREMENTS_PATH / REQS_OUTPUT_TMPL.format(group))

    # Compile requirements for single packages; need to compare before & after .txt's for reporting
    for package in ('pip',):
        requirements_path = REQUIREMENTS_PATH / REQS_OUTPUT_TMPL.format(package)
        if requirements_path.is_file():
            old_requirements_txt = requirements_path.read_text()
        else:
            old_requirements_txt = ''

        run_pip_compile(
            upgrade_arg,
            input_line=package,
            output_file=REQUIREMENTS_PATH / REQS_OUTPUT_TMPL.format(package),
            env=env)

        new_requirements_txt = requirements_path.read_text()
        all_updates.update(evaluate_requirements_txt(old_requirements_txt, new_requirements_txt))

    # Generate new pinned extras; any updates to these are already recorded w/ uv.lock package diff
    for pinned_name, extra_name in PINNED_EXTRAS.items():
        extras[pinned_name] = run_uv_export(extras=[extra_name], bare=True).splitlines()

    # Write the finalized pyproject.toml
    modify_and_write_pyproject(pyproject_text, table_name=EXTRAS_TABLE, table=extras)

    return all_updates


def generate_report(
    all_updates: dict[str, tuple[str | None, str | None]],
) -> collections.abc.Iterator[str]:
    GITHUB_RE = re.compile(r'https://github\.com/(?P<owner>[0-9a-zA-Z_-]+)/(?P<repo>[0-9a-zA-Z_-]+)')

    yield 'package | old | new | diff | changelog'
    yield '--------|-----|-----|------|----------'
    for package, (old, new) in sorted(all_updates.items()):
        if package in WELLKNOWN_PACKAGES:
            github_info = WELLKNOWN_PACKAGES[package]
            changelog = ''

        else:
            project_urls = call_pypi_api(package)['info']['project_urls']
            github_info = next((
                mobj.groupdict() for url in project_urls.values()
                if (mobj := GITHUB_RE.match(url))), None)
            changelog = next((
                url for key, url in project_urls.items()
                if key.lower().startswith(('change', 'history', 'release '))), '')
            if changelog:
                name = urllib.parse.urlparse(changelog).path.rstrip('/').rpartition('/')[2] or 'changelog'
                changelog = f'[{name}](<{changelog}>)'

        md_old = old = old or ''
        md_new = new = new or ''
        if old and new:
            # bolden and italicize the differing parts
            old_parts = old.split('.')
            new_parts = new.split('.')

            offset = None
            for index, (old_part, new_part) in enumerate(zip(old_parts, new_parts, strict=False)):
                if old_part != new_part:
                    offset = index
                    break

            if offset is not None:
                md_old = '.'.join(old_parts[:offset]) + '.***' + '.'.join(old_parts[offset:]) + '***'
                md_new = '.'.join(new_parts[:offset]) + '.***' + '.'.join(new_parts[offset:]) + '***'

        compare = ''
        if github_info:
            tags_info = fetch_github_tags(
                github_info['owner'], github_info['repo'], fetch_tags=[old, new])
            old_tag = tags_info.get(old) and tags_info[old]['name']
            new_tag = tags_info.get(new) and tags_info[new]['name']
            github_url = 'https://github.com/{owner}/{repo}'.format(**github_info)
            if new_tag:
                md_new = f'[{md_new}]({github_url}/releases/tag/{new_tag})'
            if old_tag:
                md_old = f'[{md_old}]({github_url}/releases/tag/{old_tag})'
            if new_tag and old_tag:
                compare = f'[`{old_tag}...{new_tag}`]({github_url}/compare/{old_tag}...{new_tag})'

        yield ' | '.join((
            f'[**`{package}`**](https://pypi.org/project/{package})',
            md_old,
            md_new,
            compare,
            changelog,
        ))


def make_pull_request_description(all_updates: dict[str, tuple[str | None, str | None]]) -> str:
    return '\n'.join((
        '<!-- BEGIN update_requirements generated section -->\n',
        *generate_report(all_updates),
        '\n<!-- END update_requirements generated section -->\n\n',
    ))


def make_commit_message(all_updates: dict[str, tuple[str | None, str | None]]) -> str:
    return '\n\n'.join((
        make_commit_title(all_updates),
        make_commit_body(all_updates),
    ))


def make_commit_title(all_updates: dict[str, tuple[str | None, str | None]]) -> str:
    count = len(all_updates)
    return f'[build] Update {count} dependenc{"ies" if count > 1 else "y"}'


def make_commit_body(all_updates: dict[str, tuple[str | None, str | None]]) -> str:
    return '\n'.join(make_commit_line(package, old, new) for package, (old, new) in all_updates.items())


def make_commit_line(package: str, old: str | None, new: str | None) -> str:
    if old is None:
        return f'* Add {package} {new}'

    if new is None:
        return f'* Remove {package} {old}'

    return f'* Bump {package} {old} => {new}'


def table_a_raza(header: tuple[str, ...], rows: list[tuple[str, ...]]) -> collections.abc.Generator[str]:
    widths = [len(col) for col in header]

    for row in rows:
        for index, (width, col) in enumerate(zip(widths, row, strict=True)):
            if len(col) > width:
                widths[index] = len(col)

    yield ' | '.join(col.ljust(width) for width, col in zip(widths, header, strict=True))
    yield '-|-'.join(''.ljust(width, '-') for width in widths)
    for row in rows:
        yield ' | '.join(col.ljust(width) for width, col in zip(widths, row, strict=True))


def parse_args():
    import argparse
    parser = argparse.ArgumentParser(description='generate/update lockfile and requirements')
    parser.add_argument(
        'upgrade_only', nargs='?', metavar='PACKAGE',
        help='only upgrade this package. (by default, all packages will be upgraded)')
    parser.add_argument(
        '--verify', action='store_true',
        help='only verify the update(s) using the previously recorded cooldown timestamp')
    parser.add_argument(
        '--no-markdown-reports', action='store_true',
        help='do not generate markdown PR descriptions; avoids optional PyPI/GitHub API calls')
    return parser.parse_args()


def main():
    args = parse_args()

    if args.upgrade_only in ('ejs', 'yt-dlp-ejs'):
        all_updates = update_ejs(verify=args.verify)
    else:
        all_updates = update_requirements(upgrade_only=args.upgrade_only, verify=args.verify)

    if all_updates is None:
        return 1
    elif not all_updates:
        print('All requirements are up-to-date', file=sys.stderr)
        return 0

    if args.verify:
        print('Verification failed! Updates were made:', file=sys.stderr)
        for row in table_a_raza(('package', 'old', 'new'), [
            (package, old or '', new or '')
            for package, (old, new) in all_updates.items()
        ]):
            print(row)
        return 1

    else:
        if not args.no_markdown_reports:
            print(make_pull_request_description(all_updates))
        print(make_commit_message(all_updates))
        return 0


if __name__ == '__main__':
    sys.exit(main())