mirror-update

#! /usr/bin/env python3
# -*- coding: utf-8; py-indent-offset: 4 -*-
#
# Author:  Linuxfabrik GmbH, Zurich, Switzerland
# Contact: info (at) linuxfabrik (dot) ch
#          https://www.linuxfabrik.ch/
# License: The Unlicense, see LICENSE file.

import argparse
import logging
import os
import sys

SYSTEMD = bool(os.getppid() == 1)
if SYSTEMD:
    import systemd.journal
import yaml.scanner

import lib.base3
import lib.shell3
import lib.url3

__author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
__version__ = '2025012701'

DESCRIPTION = 'A script to create and update mirrors of RPM repositories.'
DEFAULT_CONFIG = '/etc/mirror.yml'
DEFAULT_RPM_REGEX = r'.*{latest_version}.*\.rpm'

createrepo_command = "createrepo '{TARGET_PATH}' --update"


def parse_args():
    """Parse command line arguments using argparse.
    """
    parser = argparse.ArgumentParser(description=DESCRIPTION)

    parser.add_argument(
        '-V', '--version',
        action='version',
        version='%(prog)s: {__version__} by {__author__}'
    )

    parser.add_argument(
        '--config',
        help='Path to the the config file. Default: %(default)s',
        dest='CONFIG',
        type=str,
        default=DEFAULT_CONFIG,
    )

    parser.add_argument(
        '--check',
        help='Enable check-only mode. The config file will be checked for syntax errors. If the check completes successfully, mirror-update will exit with a value of 0. If an error is encountered, mirror-update will exit with a value of 1.',
        dest='CHECK_MODE',
        action='store_true',
        default=False,
    )

    return parser.parse_args()


def init_logging():
    logger = logging.getLogger()
    logging.basicConfig(
        format='%(levelname)s: %(message)s'
    )
    logger.setLevel(logging.INFO)

    if SYSTEMD:
        logger.addHandler(systemd.journal.JournalHandler())

    return logger




class MirrorUpdate:
    def __init__(self, config, logger):
        self.config = config
        self.logger = logger


    def run_cmd(self, cmd):
        success, result = lib.shell3.shell_exec(cmd)
        self.logger.debug(f'Running command "{cmd}".')
        if not success:
            self.logger.error(f'Failed to run "{cmd}": {result}')
            return False

        _, stderr, retc = result
        if retc != 0:
            self.logger.error(f'"{cmd}" failed with: {stderr}')
            return False

        if stderr:
            self.logger.warning(f'"{cmd}" had errors: {stderr}')
            return False

        return True


    def mkdir(self, path):
        try:
            os.makedirs(path)
        except OSError:
            if not os.path.isdir(path):
                self.logger.exception(f'failed to create {path}')
                return False
        return True


    def validate_config(self):
        valid = True
        base_path = self.config.get('base_path')
        if not base_path:
            self.logger.error('The config is missing the "base_path" key, or it is empty.')
            valid = False

        if not os.path.isdir(base_path):
            self.logger.error(f'The "base_path" "{base_path}" is not a directory.')
            valid = False

        reposync_repos = self.config.get('reposync_repos', [])
        reposync_repo_keys = [
            'repoid',
            'relative_target_path',
        ]
        reposync_repos_repoids = set()
        for repo in reposync_repos:
            repoid = repo.get('repoid')
            if not repoid:
                self.logger.error('There is a reposync repo without a "repoid" in the config. Skipping validation for this repo.')
                valid = False
                continue

            if repoid in reposync_repos_repoids:
                self.logger.warning(f'There are multiple reposync repos with the repoid "{repoid}".')
                valid = False
            reposync_repos_repoids.add(repoid)

            for key in reposync_repo_keys:
                try:
                    repo[key]
                except KeyError:
                    self.logger.error(f'The reposync repo "{repoid}" is missing the "{key}" key.')
                    valid = False

        return valid


    def run(self):
        self.clean_repo_data()
        self.update_reposync_repos()


    def clean_repo_data(self):
        self.run_cmd('sudo dnf clean expire-cache')
        self.run_cmd('sudo dnf repolist')


    def update_reposync_repos(self):
        self.logger.info('--- Start of reposync repos ---')
        for repo in self.config.get('reposync_repos', []):
            self.logger.info(f'{repo["repoid"]} start')

            target_path = os.path.join(self.config["base_path"], repo["relative_target_path"])
            if not self.mkdir(target_path):
                continue

            cmd = "reposync --assumeyes --delete --download-metadata --download-path='{TARGET_PATH}' --downloadcomps --newest-only --norepopath --repoid='{REPOID}'".format(
                REPOID=repo["repoid"],
                TARGET_PATH=target_path,
            )
            if not self.run_cmd(cmd):
                continue

            if repo.get('createrepo', False) is True:
                cmd = createrepo_command.format(
                    REPOID=repo["repoid"],
                    TARGET_PATH=target_path,
                )
                if not self.run_cmd(cmd):
                    continue

            self.logger.info(f'{repo["repoid"]} end')
        self.logger.info('--- End of reposync repos ---')


def main():
    """The main function. Hier spielt die Musik.
    """
    args = parse_args()
    logger = init_logging()

    logger.info(f'Using config at {args.CONFIG}.')
    with open(args.CONFIG, 'rb') as file:
        try:
            config = yaml.safe_load(file)
        except yaml.scanner.ScannerError:
            logger.exception('Could not parse config file. Aborting...')
            sys.exit(1)

    mirror_update = MirrorUpdate(config, logger)

    if not mirror_update.validate_config():
        logger.critical('Config is invalid. Aborting.')
        sys.exit(1)

    if args.CHECK_MODE:
        logger.info('Config is valid.')
        sys.exit()

    mirror_update.run()


if __name__ == '__main__':
    main()