scripts/obs/update_unstable.py

#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2021 Luca Beltrame <lbeltrame@kde.org>
# SPDX-License-Identifier: BSD-3-clause

import argparse
from datetime import date
from collections import defaultdict
import logging
import json
from pathlib import Path
from typing import Union, Dict, Any
from urllib.parse import quote

import git
import requests
import sarge


API_URL = "https://invent.kde.org/api/v4/projects/"
OBS_URL = "https://api.opensuse.org/trigger/runservice"
MATRIX_COMMANDER = "/home/mocker/local-venv/bin/matrix-commander.py"

MESSAGE_TEMPLATE = f"""
### OBS package update complete

Stats for {date.today().strftime('%Y-%m-%d')}:

"""


class GitHashCache:

    def __init__(self, cache_file: str) -> None:
        self.cache = cache_file
        self._data: Dict[str, Dict[str, str]] = dict()

    def __getitem__(self, key: str) -> Dict[str, str]:

        if key not in self._data:
            raise KeyError

        return self._data[key]

    def __setitem__(self, key: str, value: Dict[str, str]) -> None:
        self._data[key] = value

    def get(self, key: str,
            *args: Any, **kwargs: Any) -> Union[None, str, Dict[str, str]]:
        return self._data.get(key, *args, **kwargs)

    def save(self) -> None:
        logging.debug("Saving pickled data")
        with open(self.cache, "w") as handle:
            json.dump(self._data, handle, indent=4)

    def load(self) -> None:

        if not Path(self.cache).exists():
            logging.debug("File cache not found, not loading")
            return

        with open(self.cache) as handle:
            self._data = json.load(handle)


def project_exists(project: str) -> bool:
    # We want / to get quoted, so put safe to ""
    project_name = quote(project, safe="")
    request = requests.get(API_URL + project_name)

    if request:
        return True

    return False


def lsremote(url: str) -> Dict[str, str]:

    remote_refs = {}
    gitcmd = git.cmd.Git()
    for ref in gitcmd.ls_remote(url).split('\n'):
        hash_ref_list = ref.split('\t')
        remote_refs[hash_ref_list[1]] = hash_ref_list[0]

    return remote_refs


def get_remote_hash(url: str, branch: str = "master") -> str:
    refs = "refs/heads/{}".format(branch)
    return lsremote(url)[refs]


def trigger_update(repository: str, package_name: str,
                   token: str) -> Union[requests.Response, bool]:

    header = {"Authorization": f"Token {token}"}
    parameters = {"project": repository, "package": package_name}

    logging.info("Updating package %s", package_name)
    result = requests.post(OBS_URL, params=parameters, headers=header)

    if not result:
        logging.error(
            "Error during service run, package %s, error code %s, url %s",
            package_name, result.status_code, result.url)
        return False

    logging.debug("Package %s complete", package_name)
    return result


def update_package(hash_data: GitHashCache, package_name: str,
                   remote_name: str, obs_repository: str,
                   branch: str,
                   token: str,
                   stats: Dict[str, int]) -> None:

    repo_name = "https://invent.kde.org/{}".format(remote_name)

    if not project_exists(remote_name):
        logging.warning("Repository %s not found, skipping", remote_name)
        return

    remote_hash = get_remote_hash(repo_name, branch)

    if hash_data.get(obs_repository) is None:
        logging.debug("No prior data - initializing empty")
        hash_data[obs_repository] = dict()

    current_hash = hash_data[obs_repository].get(remote_name, "")

    logging.debug("Package %s, theirs %s, ours %s",
                  remote_name, remote_hash, current_hash)

    if remote_hash != current_hash:
        logging.debug("Hash doesn't match, updating")
        if trigger_update(obs_repository, package_name, token):
            hash_data[obs_repository][remote_name] = remote_hash
            stats["updated"] += 1
            hash_data.save()
        else:
            stats["errors"] += 1


def update_packages(cache_file: str,
                    repo_mapping_file: str, token: str) -> None:

    hash_data = GitHashCache(cache_file)
    hash_data.load()

    stats = dict()

    with open(repo_mapping_file, "r") as mapping:

        repo_data = json.load(mapping)

        for obs_repository, branch_data in repo_data.items():
            repo_stats: Dict[str, int] = defaultdict(int)

            logging.info("Updating packages for %s", obs_repository)

            for package in branch_data:
                kde_name = package["kde"]
                obs_name = package["obs"]
                branch = package["branch"]
                package_name = Path(kde_name).name
                logging.debug("Updating package %s (%s)",
                              package_name, obs_name)
                logging.debug("Using branch %s", branch)
                update_package(hash_data, obs_name, kde_name, obs_repository,
                               branch, token, repo_stats)
            stats[obs_repository] = repo_stats

    logging.debug("Saving data")
    hash_data.save()
    notify_matrix(stats)


def notify_matrix(stats: Dict[str, Dict[str, int]]) -> None:

    structure = [MESSAGE_TEMPLATE]
    for key, value in stats.items():
        row = (f"* {key}: {value['updated']} updated packages,"
               f" {value['errors']} errors")
        structure.append(row)

    message = "\n".join(structure)

    cmd = [MATRIX_COMMANDER, "--markdown", "-m", message]
    logging.debug("Sending Matrix notification")
    sarge.run(cmd)


def commit_changes(cache_file: str, repo_home: str) -> None:

    repo = git.Repo(repo_home)
    repo.index.add([cache_file])
    repo.index.commit("[OBS unstable update bot] Update caches")
    origin = repo.remotes["origin"]
    origin.pull(rebase=True)
    origin.push()


def main() -> None:

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--cache-file", help="Location of the cache file",
        default=Path.home() / ".local/share/obs_repo_cache.json")
    parser.add_argument(
        "--repo-root", help="Location of the git repository root")
    parser.add_argument("mapping_file", help="KDE:OBS repository mapping file")
    parser.add_argument("token", help="Authorization token file")
    parser.add_argument("--debug", help="Show debugging output",
                        action="store_true")
    options = parser.parse_args()

    level = logging.INFO if not options.debug else logging.DEBUG

    logging.basicConfig(format='%(levelname)s - %(message)s',
                        level=level)

    cache_file = options.cache_file

    with open(options.token) as handle:
        token = handle.read().strip()

    update_packages(cache_file, options.mapping_file, token)

    if options.repo_root is not None:
        logging.info("Committing changes")
        commit_changes(cache_file, options.repo_root)

    logging.info("Complete")


if __name__ == "__main__":
    main()