aurrepo.py

#!/usr/bin/env python
# coding: utf-8
"""Automate managing an Archlinux repository of AUR-Only packages

Automatically checks the AUR for version changes, rebuilds packages and adds
them to a proper Archlinux repository so you can have your computers use the
latest version without the hassle of manually checking for updates, rebuilding,
package distribution etc pp


The MIT License (MIT)

Copyright (c) 2016 Leonard Techel

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
"""
from urllib import request, parse
import json
import os
import shutil
import tarfile
import subprocess
import platform

def urlencode(obj):
    """Encodes an object as query string

    As difference to urllib.parse.urlencode, this function encodes lists
    Rails-style as key[]=listvalue&key[]=listvalue etc

    Args:
        obj (dict): Dict that should be converted to a query string
    """
    qargs = []

    for k, v in obj.items():
        if type(v) is list:
            for el in v:
                qargs.append(("{0}[]".format(k), el))
        else:
            qargs.append((k, v))

    return parse.urlencode(qargs)

def exec(cmd, *args, **kwargs):
    """Execs a command on the shell

    If the command fails, an exception is raised. You can specify all
    arguments ``subprocess.run`` would take.

    Args:
        cmd (str): Command to run
    """
    return subprocess.run(cmd, shell=True, check=True, *args, **kwargs)

class PacmanRepo:
    """Pacman repository management
    """
    def __init__(self, repo_path):
        self.repo_path = repo_path

    def addPackage(self, package_path):
        """Adds or updates a package

        As difference to a manual `repo-add` call, this function also copies
        the corresponding package file to the same directory the repo database
        is stored in.

        Args:
            package_path (str): Path to the package
        """
        shutil.copy(package_path, os.path.dirname(self.repo_path))

        sig = "{0}.sig".format(package_path)
        if os.path.isfile(sig):
            shutil.copy(sig, os.path.dirname(self.repo_path))

        exec("repo-add -v -s {0} {1}".format(self.repo_path, package_path))

    def removePackage(self, package_name):
        """Removes a package

        TODO remove the corresponding package files
        """
        exec("repo-remove -v -s {0} {1}".format(self.repo_path, package_name))

class AurRepo:
    """Keeps package version states and only rebuilds them if it changed
    """
    def __init__(self):
        # Path config. Should maybe be loaded from somewhere else.
        self.aur_url = "https://aur4.archlinux.org"
        self.tmp_path = "/tmp/aurrepo"
        self.state_path = "aurrepo.json"
        self.repo_path = "repo/repo.db.tar.gz"

        # Do not touch
        self.repo = PacmanRepo(self.repo_path)
        self.packages = {}

    def load(self):
        """Loads a saved packages state file

        The JSON representation of each AUR package that is managed can be
        saved in a JSON file. These definitions are used to determine if a
        package has a new version or not.
        """
        if not os.path.isfile(self.state_path):
            return

        with open(self.state_path, "r") as f:
            self.packages = json.load(f)

    def save(self):
        """Saves the current package states into a file

        This function should always be called after using ``self.check()` to
        save updated packages
        """
        with open(self.state_path, "w") as f:
            json.dump(self.packages, f)

    def addPackage(self, name):
        """Add a new package

        Automatically runs the first check so that the package gets build
        and added to the repository.

        Args:
            name (str): Exact package name from AUR
        """
        if name in self.packages:
            return

        self.packages[name] = {
            "Version": None,
            "URLPath": None
        }

        self.check()

    def updatePackage(self, name, new):
        """Download, build, repository update, state saving

        Downloads the AUR PKGBUILD archive, unpacks it, calls makepkg,
        updates the repository and in the end saves the new AurRepo state.

        Args:
            name (str): Exact package name from AUR
            new (dict): Dict representing the AUR API JSON package
        """
        if not os.path.exists(self.tmp_path):
            os.makedirs(self.tmp_path)

        # Download the tar file and unpack it
        url = "{0}{1}".format(self.aur_url, new["URLPath"])
        with request.urlopen(url) as req:
            tar = tarfile.open(fileobj=req, mode="r|gz")
            tar.extractall(path=self.tmp_path)

        build_path = "{0}/{1}".format(self.tmp_path, name)
        pkg_path = "{0}/{1}/{1}-{2}-{3}.pkg.tar.xz".format(self.tmp_path, name, new["Version"], platform.machine())

        # Call makepkg
        exec("makepkg -s -f --noconfirm --sign", cwd=build_path)

        # Update the repository
        self.repo.addPackage(pkg_path)

        # Remove the build folder
        shutil.rmtree(build_path)

        # Save the new configuration
        self.packages[name] = new
        self.save()

    def check(self):
        """Checks all known packages for updates

        Calls the AUR RPC batch API and then compares known versions with
        the ones from the API. If the version has changed, calls
        ``self.updatePackage()`.

        If some packages are no longer inside the AUR, there is an error message
        printed out.
        """
        q = {"type": "multiinfo", "arg": []}
        for name, obj in self.packages.items():
            q["arg"].append(name)

        url = "{0}/rpc.php?{1}".format(self.aur_url, urlencode(q))

        checked = []
        with request.urlopen(url) as req:
            pp = json.loads(req.read().decode())

            for package in pp["results"]:
                local = self.packages[package["Name"]]
                checked.append(package["Name"])

                # Update packages with changed version
                if package["Version"] != local["Version"]:
                    self.updatePackage(package["Name"], package)

        # Check if there are some packages lost in the AUR
        lost = list(set(self.packages.keys()) - set(checked))
        if lost:
            print("Packages lost from AUR: {0}".format(", ".join(lost)))

if __name__ == "__main__":
    repo = AurRepo()
    repo.load()

    repo.addPackage("knot")