release_post.py

#!/usr/bin/env python
"""Generate the package update section of the XCP-ng release post.

Read the json report generated by pkg_in_pipe.py and, for each package of a
given koji tag, print the "Explain the change to users" sections of the pull
requests related to the package builds.
"""

from __future__ import annotations

import argparse
import os
import re
import signal
import sys
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from string import Template
from textwrap import dedent
from typing import Literal, cast

import diskcache  # type: ignore[import-untyped]
import requests
from pydantic import BaseModel
from tqdm import tqdm


class Warnings(BaseModel):
    plane: bool
    github: bool


class Issue(BaseModel):
    sequence_id: int
    url: str
    milestones: list[str]


class PullRequest(BaseModel):
    number: int
    title: str
    url: str
    linked: bool


class Build(BaseModel):
    nvr: str
    previous_nvr: str | None = None
    package: str
    url: str
    built_by: str
    maintained_by: str | None
    issues: list[Issue]
    pull_requests: list[PullRequest]
    build_linked: bool


class TagReport(BaseModel):
    tag: str
    builds: list[Build]


class Report(BaseModel):
    generated_at: datetime
    generated_info: str | None
    warnings: Warnings
    error: Literal['koji', 'unknown'] | None
    tags: list[TagReport]


PR_URL_RE = re.compile(r'^https://github\.com/([^/]+)/([^/]+)/pull/(\d+)$')

RETENTION_TIME = 24 * 60 * 60  # 24 hours


def find_tag_report(report: Report, tag: str) -> TagReport:
    for tag_report in report.tags:
        if tag_report.tag == tag:
            return tag_report
    raise SystemExit(f'error: the tag {tag} is not present in the report')


def prs_by_package(tag_report: TagReport) -> dict[str, list[PullRequest]]:
    prs_by_package: dict[str, list[PullRequest]] = defaultdict(list)
    for build in tag_report.builds:
        for pr in build.pull_requests:
            if all(existing.url != pr.url for existing in prs_by_package[build.package]):
                prs_by_package[build.package].append(pr)
    return dict(prs_by_package)


def fetch_pr_description(
    url: str, token: str | None, re_cache: bool, cache: diskcache.Cache
) -> str | None:
    cache_key = f'pr-body-1-{url}'
    if not re_cache and cache_key in cache:
        return cast(str | None, cache[cache_key])
    match = PR_URL_RE.fullmatch(url)
    if match is None:
        raise RuntimeError(f'not a github pull request url: {url}')
    owner, repo, number = cast(tuple[str, str, str], match.groups())
    headers = {'Accept': 'application/vnd.github+json'}
    if token:
        headers['Authorization'] = f'Bearer {token}'
    response = requests.get(f'https://api.github.com/repos/{owner}/{repo}/pulls/{number}', headers=headers, timeout=30)
    if response.status_code != 200:
        raise RuntimeError(f'got a {response.status_code} response')
    body = cast(str | None, response.json().get('body'))
    cache.set(cache_key, body, expire=RETENTION_TIME)
    return body


def fetch_pr_descriptions(
    prs: list[PullRequest], token: str | None, re_cache: bool, cache: diskcache.Cache
) -> list[tuple[PullRequest, str]]:
    descriptions: list[tuple[PullRequest, str]] = []
    for pr in prs:
        try:
            description = fetch_pr_description(pr.url, token, re_cache, cache)
        except (RuntimeError, requests.RequestException) as e:
            print(f'warning: could not fetch the description of {pr.url}: {e}', file=sys.stderr)
            continue
        if description is None:
            print(f'warning: the pull request {pr.url} has no description', file=sys.stderr)
            continue
        descriptions.append((pr, description))
    return descriptions


USER_SECTION_RE = re.compile(r'(?i)^#{1,6}\s*Explain the change to users\s*$')
HEADING_RE = re.compile(r'^#{1,6}(\s|$)')
CODE_FENCE_RE = re.compile(r'^\s*(```+|~~~+)')


def extract_user_section(description: str) -> list[str] | None:
    """Return the lines of the "Explain the change to users" section of a pull request description.

    The section is returned verbatim, except for the HTML comments (template boilerplate)
    and the blank lines around them, which are removed.
    """
    in_code = False
    section: list[str] | None = None
    for line in description.splitlines():
        fence = CODE_FENCE_RE.match(line)
        if fence is not None:
            in_code = not in_code
        if section is not None:
            if not in_code and HEADING_RE.match(line):
                break
            section.append(line)
        elif not in_code and USER_SECTION_RE.match(line):
            section = []
    if section is None:
        return None
    section = strip_html_comments(section)
    while section and not section[0].strip():
        section.pop(0)
    while section and not section[-1].strip():
        section.pop()
    return section or None


def strip_html_comments(lines: list[str]) -> list[str]:
    """Remove the HTML comments (template boilerplate) and the blank lines around them."""
    res: list[str] = []
    in_comment = False
    skip_blanks = False
    for line in lines:
        if in_comment:
            if '-->' in line:
                in_comment = False
                skip_blanks = True
            continue
        if '<!--' in line:
            in_comment = '-->' not in line
            skip_blanks = True
            while res and not res[-1].strip():
                res.pop()
            continue
        if skip_blanks and not line.strip():
            continue
        skip_blanks = False
        res.append(line)
    while res and not res[-1].strip():
        res.pop()
    return res


def warn(pbar: tqdm, message: str) -> None:
    pbar.clear()
    print(f'warning: {message}', file=sys.stderr)


def collect_sections(descriptions: list[tuple[PullRequest, str]], pbar: tqdm) -> list[str]:
    lines: list[str] = []
    for pr, description in descriptions:
        section = extract_user_section(description)
        if section is None:
            warn(pbar, f'no "Explain the change to users" section in {pr.url}')
            continue
        lines.extend(section)
    return lines


def verbatim_lines(package: str, lines: list[str]) -> list[str]:
    if len(lines) > 1:
        return [f'- `{package}`:'] + [f'\t{line}' if line else '' for line in lines]
    if lines:
        return [f'- `{package}`: {lines[0]}']
    return [f'- `{package}`:']


def versions_by_package(tag_report: TagReport) -> list[tuple[str, str | None, str]]:
    """Return the latest build of each package, as (package, previous_nvr, nvr), sorted by package name.

    The report lists the builds of a tag newest first, so the first occurrence of a package is its latest build.
    """
    latest: dict[str, tuple[str | None, str]] = {}
    for build in tag_report.builds:
        latest.setdefault(build.package, (build.previous_nvr, build.nvr))
    return [(package, previous_nvr, nvr) for package, (previous_nvr, nvr) in sorted(latest.items())]


def version_release(package: str, nvr: str) -> str:
    """Return the version-release part of a package nvr."""
    return nvr.removeprefix(package + '-')


def format_versions(versions: list[tuple[str, str | None, str]]) -> str:
    lines = []
    for package, previous_nvr, nvr in versions:
        new_version = version_release(package, nvr)
        item = f'{version_release(package, previous_nvr)} -> {new_version}' if previous_nvr is not None else new_version
        lines.append(f'* `{package}`: {item}')
    return '\n'.join(lines)


POST_TEMPLATE = dedent('''\
    # New maintenance update candidates for XCP-ng $version LTS

    This batch of updates contains mostly fixes, tools version update, a some improvements.

    ## What changed

    $what_changed

    ## Versions

    $versions

    ## Test on XCP-ng $version

    ```bash
    yum clean metadata --enablerepo=xcp-ng-testing,xcp-ng-candidates
    yum update --enablerepo=xcp-ng-testing,xcp-ng-candidates
    reboot
    ```

    The usual update rules apply: pool coordinator first, etc.

    ## What to test

    As usual, normal use and anything else you want to test.

    ## Test window before official release of the updates

    **X days**

    We would like to thank users who shared feedback since our last call for testing:
    ''')


def format_post(what_changed: str, versions: str, version: str) -> str:
    return Template(POST_TEMPLATE).substitute(what_changed=what_changed, versions=versions, version=version)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description='Generate the package update section of the XCP-ng release post')
    parser.add_argument('--report', help='The json report generated by pkg_in_pipe.py', default='report.json')
    parser.add_argument('--tag', help='The koji tag to include in the release post', default='v8.3-ci')
    parser.add_argument('--version', help='The XCP-ng version of the release post, e.g. 8.3', default=None)
    parser.add_argument('--cache', help='The cache path', default='/tmp/pkg_in_pipe.cache')
    parser.add_argument('--re-cache', help='Refresh the cache', action='store_true')
    parser.add_argument(
        '--github-token', help='The token used to access the Github api', default=os.environ.get('GITHUB_TOKEN')
    )
    return parser.parse_args()


def main() -> None:
    signal.signal(signal.SIGPIPE, signal.SIG_DFL)
    args = parse_args()
    cache = diskcache.Cache(args.cache)
    report = Report.model_validate_json(Path(args.report).read_text())
    tag_report = find_tag_report(report, args.tag)
    what_changed: list[str] = []
    with tqdm(
        sorted(prs_by_package(tag_report).items()),
        desc='packages',
        unit='package',
        file=sys.stderr,
        leave=False,
    ) as pbar:
        for package, prs in pbar:
            descriptions = fetch_pr_descriptions(prs, args.github_token, args.re_cache, cache)
            if not descriptions:
                warn(pbar, f'no description available for the pull requests of {package}, skipping it')
                continue
            lines = collect_sections(descriptions, pbar)
            if not lines:
                warn(pbar, f'no "Explain the change to users" section available for the pull requests of {package}')
            what_changed.extend(verbatim_lines(package, lines))
            what_changed.append('')
    version = args.version or args.tag[1:].split('-')[0]
    print(format_post('\n'.join(what_changed).rstrip(), format_versions(versions_by_package(tag_report)), version))


if __name__ == '__main__':
    main()