Python3 RPM changelog helper PoC

changes.py

#!/usr/bin/python3
"""Manipulate the changelog entries for all code-streams.

Different code streams that use the same version of the source code (same git
commit) can have different RPM changelogs. They differ in structure due to
different release times: entries that are grouped in one code stream might be
split into multiple entries in another.

This tools helps with adding a new entry to all changelogs and with editing the
latest changelog entry for all code streams at once.
"""
import argparse
import io
import os
import re
import subprocess
import sys
import tempfile

if sys.version_info < (3, 11):
    raise RuntimeError("Python 3.11 or greater required.")

SEPARATOR = "-" * 67 + "\n"


def partition_list_by(lst, fun):
    """Applies fun to each value in lst, splitting lst when f returns True."""
    ret = []
    cur = []
    for i in lst:
        cur.append(i)
        if fun(i):
            ret.append(cur)
            cur = []

    return ret


def list_changelogs():
    """List all .changes files for (open)SUSE packages in the repository."""
    cp = subprocess.run(
        ["git", "ls-files", "*.changes"], capture_output=True, text=True
    )
    return cp.stdout.splitlines()


def first_changelog_entry(filename: str) -> str | None:
    """Return first changelog entry in .changes file.

    Args:
      filename: File name of .changes file.

    Returns:
      First changelog entry or None of no changelog entries were found.
    """

    def _match_separator(line: str):
        return line == SEPARATOR

    with open(filename, "r", encoding="utf-8") as file:
        # split at separator of next entry
        entries = partition_list_by(file, _match_separator)

    if len(entries) > 1:
        entry = [SEPARATOR] + entries[1]
        # return entry without separator of next entry
        return "".join(entry[:-1])

    return None


def update_changelog_from_file(changelog_file: str, entry_file: str, edit=False) -> int:
    """Update a .changes files with an entry from a separate file.

    The file containing the new changelog is prepended to the .changes file.

    Internally, both the new entry and the .changes files are read into memory
    and then written to disk, overriding the old .changes file.

    Args:
      changelog_file: Name of .changes file.
      entry_file: Name of file containing the new entry.
      edit: If True, replace the first entry. If False (default), add the new entry to
        the top of the changelog.

    Returns:
      Number of bytes written back to changelog_file.
    """
    with open(entry_file, "r", encoding="utf-8") as new_entry:
        new_changelog = io.StringIO()
        new_changelog.write(new_entry.read())

    with open(changelog_file, "r", encoding="utf-8") as old_changelog:
        if edit:
            # skip first separator
            old_changelog.readline()
            # skip until the second separator was read
            while line := old_changelog.readline() != SEPARATOR:
                continue

            # add back the separator
            new_changelog.write(SEPARATOR)
        else:
            # new changelog entry, needs an empty line for correct spacing
            new_changelog.write("\n")

        new_changelog.write(old_changelog.read())

    with open(changelog_file, "w", encoding="utf-8") as f:
        return f.write(new_changelog.getvalue())


def new_entry():
    """Use `osc vc` to add a new entry.

    The new entry will be added to all tracked .changes files.
    """
    changelogs = list_changelogs()

    with tempfile.TemporaryDirectory() as temp:
        temp_changes = os.path.join(temp, "temp.changes")
        subprocess.run(["osc", "vc", temp_changes], check=True)

        for changelog in changelogs:
            update_changelog_from_file(changelog, temp_changes)


def edit_latest_entry():
    """Use `osc vc` to edit the latest entry.

    The changed entry will be written to all tracked .changes files.

    Raises:
      RuntimeError when no .changes files are found, the latest entries
      are empty or differ.

    """
    changelogs = list_changelogs()
    if not changelogs:
        raise RuntimeError("No changelogs found!")

    latest_entries = [first_changelog_entry(chlog) for chlog in changelogs]
    if not latest_entries:
        raise RuntimeError("No changlog entries found!")
    elif len(set(latest_entries)) != 1:
        raise RuntimeError("Latest changelog entries differ!")
    elif latest_entries[0] is None:
        raise RuntimeError("No changlog entries found!")

    entry = latest_entries[0]
    editor = os.getenv("EDITOR", os.getenv("VISUAL", "emacs"))
    edit_cmd = editor.split()

    with tempfile.TemporaryDirectory() as temp:
        # Write entry to a temporary file, then edit it with $EDITOR, $VISUAL, or `emacs`.
        temp_changes = os.path.join(temp, "temp.changes")
        with open(temp_changes, "w", encoding="utf-8") as f:
            f.write(entry)

        # len(entry) can differ from number of written bytes, utf-8 takes up to
        # 4 bytes per code point
        initial_size = os.path.getsize(temp_changes)

        edit_cmd.append(temp_changes)
        subprocess.run(edit_cmd, check=True)

        if os.path.getsize(temp_changes) == initial_size:
            # no changes
            return None

        for changelog in changelogs:
            update_changelog_from_file(changelog, temp_changes, edit=True)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    # TODO: add "remove" to drop the latest changelog
    parser.add_argument("action", choices=["add", "edit", "list"], default="add")
    args = parser.parse_args()

    if args.action == "list":
        print(" ".join(list_changelogs()))
    elif args.action == "add":
        new_entry()
    elif args.action == "edit":
        edit_latest_entry()