tools/update_repo.py

#! /usr/bin/env python3
import os
import subprocess


class Repo:
    _repo:str

    def __init__(self, repo:str):
        assert isinstance(repo, str)
        self._repo = repo
    
    @property
    def flatpaks(self):
        tmp = subprocess.run(["ostree", "refs", "--repo", self._repo], capture_output=True, check=True)
        return list(filter(lambda x: x.startswith(("runtime/", "app/")) or x == "empty", tmp.stdout.decode().splitlines()))

    def move_commit(self, repo, commit:str, no_fsync:bool=False, gpg_key:str=None, collection:str=None):
        assert isinstance(repo, Repo)
        assert isinstance(commit, str)
        assert isinstance(no_fsync, bool)
        assert gpg_key is None or isinstance(gpg_key, str)
        assert collection is None or isinstance(collection, str)

        run_args = []
        if no_fsync:
            run_args.append("--disable-fsync")
        if gpg_key:
            run_args.append("--gpg-sign=" + gpg_key)
        if collection:
            run_args.append("--extra-collection-id=" + collection)
        subprocess.run(["flatpak", "build-commit-from"] + run_args + ["--src-repo=" + repo._repo, "--src-ref=" + commit, self._repo, commit], check=True)
    
    def make_eol(self, commit:str, no_fsync:bool=False, gpg_key:str=None, collection:str=None):
        assert isinstance(commit, str)
        assert isinstance(no_fsync, bool)
        assert gpg_key is None or isinstance(gpg_key, str)
        assert collection is None or isinstance(collection, str)

        run_args = []
        if no_fsync:
            run_args.append("--disable-fsync")
        if gpg_key:
            run_args.append("--gpg-sign=" + gpg_key)
        if collection:
            run_args.append("--extra-collection-id=" + collection)
        subprocess.run(["flatpak", "build-commit-from", "--no-update-summary", "--no-summary-index"] + run_args + ["--end-of-life=Base commit doesn't exists any more.", self._repo, commit], check=True)
    
    def delete(self, ref:str):
        assert isinstance(ref, str)
        subprocess.run(["ostree", "refs", "--repo", self._repo, "--delete", ref], check=True)


if __name__ == "__main__":
    import argparse

    # Argument parser
    parser = argparse.ArgumentParser()
    parser.add_argument("repo", nargs=1, help="Repo to manage")
    parser.add_argument("sources", nargs="+", help="Path with repos")
    parser.add_argument("--collection", "-c", type=str, default=None, nargs="?", help="Set collection")
    parser.add_argument("--gpg", "-g", type=str, default=None, nargs="?", help="GPG-Key")
    parser.add_argument("--no-fsync", dest="no_fsync", action="store_const", const=True, default=False, help="Not fsync")
    parser.add_argument("--delete", action="store_const", const=True, default=False, help="Delete not existing branches")
    args = parser.parse_args()

    # Gen sources
    source_repos = []
    for i in args.sources:
        for j in map(lambda x: os.path.join(i, x), os.listdir(i)):
            source_repos.append(Repo(j))
    target_repo = Repo(args.repo[0])

    # Move commit
    commits_added = []
    for i in source_repos:
        tmp = i.flatpaks
        commits_added += tmp
        for j in tmp:
            target_repo.move_commit(i, j, no_fsync=args.no_fsync, gpg_key=args.gpg, collection=args.collection)

    # Remove old
    commits_added = set(commits_added)
    for i in filter(lambda x: x not in commits_added, target_repo.flatpaks):
        if args.delete:
            target_repo.delete(i)
        else:
            target_repo.make_eol(i, no_fsync=args.no_fsync, gpg_key=args.gpg, collection=args.collection)
    
    # Update summary and appstream
    subprocess.run(["ostree", "summary", "--repo", args.repo, "-u"] + (["--gpg-sign=" + args.gpg] if args.gpg else []), check=True)
    subprocess.run(["flatpak", "build-update-repo"] + (["--deploy-collection-id", "--collection-id", args.collection] if args.collection else []) + (["--gpg-sign", args.gpg] if args.gpg else []) + [args.repo], check=True)