risotto/ansible/sbin/make_changelog

#!/usr/bin/env python3


import logging
from dnf.conf import Conf
from dnf.cli.cli import BaseCli, Cli
from dnf.cli.output import Output
from dnf.cli.option_parser import OptionParser
from dnf.i18n import _, ucd
from datetime import datetime, timezone
from sys import argv
from os import getcwd, unlink
from os.path import isfile, join
from glob import glob
from subprocess import run


# List new or removed file
def read_dnf_pkg_file(os_name, filename1, filename2):
    if os_name == 'debian':
        idx_pkg = 0, 1
        idx_version = 1, 2
        header_idx = 0, 0
    else:
        idx_pkg = 0, 0
        idx_version = 2, 2
        header_idx = 2, 2
        pass
    pkgs = {}
    for fidx, filename in enumerate((filename1, filename2)):
        if not isfile(filename):
            continue
        with open(filename, 'r') as pkgs_fh:
            for idx, pkg_line in enumerate(pkgs_fh.readlines()):
                if idx < header_idx[fidx]:
                    # header
                    continue
                sp_line = pkg_line.strip().split()
                if len(sp_line) < idx_version[fidx] + 1:
                    continue
                pkg = sp_line[idx_pkg[fidx]]
                version = sp_line[idx_version[fidx]]
                #if pkg in pkgs:
                #    raise Exception(f'package already set {pkg}?')
                if os_name == 'debian' and version.startswith('('):
                    version = version[1:]
                pkgs[pkg] = version
    return pkgs


def list_packages(title, packages, packages_info):
    print(f'# {title}\n')
    if not packages:
        print('*Aucun*')
    packages = list(packages)
    packages = sorted(packages)
    for idx, pkg in enumerate(packages):
        print(f' - {pkg} ({packages_info[pkg]})')
    print()


# List updated packages
class CustomOutput(Output):
    def listPkgs(self, *args, **kwargs):
        # do not display list
        pass


def format_changelog_markdown(changelog):
    """Return changelog formatted as in spec file"""
    text = '\n'.join([f'    {line}' for line in changelog['text'].split('\n')])
    chlog_str = '  - %s %s\n\n%s\n' % (
        changelog['timestamp'].strftime("%a %b %d %X %Y"),
        ucd(changelog['author']),
        ucd(text))
    return chlog_str


def print_changelogs_markdown(packages):
    # group packages by src.rpm to avoid showing duplicate changelogs
    self = BASE
    bysrpm = dict()
    for p in packages:
        # there are packages without source_name, use name then.
        bysrpm.setdefault(p.source_name or p.name, []).append(p)
    for source_name in sorted(bysrpm.keys()):
        bin_packages = bysrpm[source_name]
        print('- ' + _("Changelogs for {}").format(', '.join([str(pkg) for pkg in bin_packages])))
        print()
        for chl in self.latest_changelogs(bin_packages[0]):
            print(format_changelog_markdown(chl))


def dnf_update(image_name, releasever):
    conf = Conf()
    # obsoletes are already listed
    conf.obsoletes = False
    with BaseCli(conf) as base:
        global BASE
        BASE = base
        base.print_changelogs = print_changelogs_markdown
        custom_output = CustomOutput(base.output.base, base.output.conf)
        base.output = custom_output
        cli = Cli(base)
        image_dir = join(getcwd(), image_name)
        cli.configure(['--setopt=install_weak_deps=False', '--nodocs', '--noplugins', '--installroot=' + image_dir, '--releasever', releasever, 'check-update', '--changelog'], OptionParser())
        logger = logging.getLogger("dnf")
        for h in logger.handlers:
            logger.removeHandler(h)
        logger.addHandler(logging.NullHandler())
        cli.run()


def main(os_name, image_name, old_version, releasever):
    date = datetime.now(timezone.utc).isoformat()
    if old_version == 0:
        title = f"Création de l'image {image_name}"
        subtitle = f"Les paquets de la première image {image_name} sur base Fedora {releasever}"
    else:
        title = f"Nouvelle version de l'image {image_name}"
        subtitle = f"Différence des paquets de l'image {image_name} sur base Fedora {releasever} entre la version {old_version} et {old_version + 1}"
    print(f"""+++
title = "{title}"
description = "{subtitle}"
date = {date}
updated = {date}
draft = false
template = "blog/page.html"

[taxonomies]
authors = ["Automate"]

[extra]
lead = "{subtitle}."
type = "installe"
+++
""")
    new_dict = read_dnf_pkg_file(os_name, f'/var/lib/risotto/images/image_bases-{os_name}-{releasever}.pkgs', f'/var/lib/risotto/images/{image_name}.pkgs.new')
    new_pkg = new_dict.keys()
    old_file = f'/var/lib/risotto/images/{image_name}.pkgs'
    if not old_version or not isfile(old_file):
        list_packages('Liste des paquets', new_pkg, new_dict)
    else:
        ori_dict = read_dnf_pkg_file(os_name, f'/var/lib/risotto/images/{image_name}.base.pkgs', old_file)
        ori_pkg = ori_dict.keys()
        list_packages('Les paquets supprimés', ori_pkg - new_pkg, ori_dict)
        list_packages('Les paquets ajoutés', new_pkg - ori_pkg, new_dict)
        print('# Les paquets mises à jour\n')
        if os_name == 'fedora':
            dnf_update(image_name, releasever)
        else:
            for filename in glob('*.deb'):
                unlink(filename)
            for package in ori_pkg & new_dict:
                if ori_dict[package] == new_dict[package]:
                    continue
                info = run(['apt', 'download', package], capture_output=True)
                if info.returncode:
                    raise Exception(f'cannot download {package}: {info}')
            packages = list(glob('*.deb'))
            packages.sort()
            for package in packages:
                info = run(['chroot', '.', 'apt-listchanges', '--which', 'both', '-f', 'text', package], capture_output=True)
                if info.returncode:
                    raise Exception(f'cannot list changes for {package}: {info}')
                header = True
                for line in info.stdout.decode().split('\n'):
                    if not header:
                        print(line)
                    if line.startswith('-----------------------'):
                        header = False
                print()
                unlink(package)



if __name__ == "__main__":
    image_name = argv[1]
    old_version = int(argv[2])
    os_name = argv[3]
    releasever = argv[4]
    main(os_name, image_name, old_version, releasever)