Skip to content
agetpkg 14.6 KiB
Newer Older
Seblu's avatar
Seblu committed
#!/usr/bin/python3
# coding: utf-8

# agetpkg - Archive Get Package
Seblu's avatar
Seblu committed
# Copyright © 2017 Sébastien Luttringer
Seblu's avatar
Seblu committed
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.

'''Arch Linux Archive Get Package'''
Seblu's avatar
Seblu committed

from argparse import ArgumentParser
from collections import OrderedDict
from email.utils import parsedate
Seblu's avatar
Seblu committed
from logging import getLogger, error, warn, debug, DEBUG
Seblu's avatar
Seblu committed
from lzma import open as xzopen
Seblu's avatar
Seblu committed
from os import stat, uname, getcwd, chdir, geteuid, environ
Seblu's avatar
Seblu committed
from os.path import basename, exists, join
from pprint import pprint
from re import match, compile as recompile
from shutil import copyfileobj
Seblu's avatar
Seblu committed
from subprocess import call
from tempfile import TemporaryDirectory
Seblu's avatar
Seblu committed
from time import mktime, time
from urllib.request import urlopen, Request
from xdg.BaseDirectory import save_cache_path

# magics
NAME = "agetpkg"
Seblu's avatar
Seblu committed
VERSION = "3"
Seblu's avatar
Seblu committed
ARCHIVE_URL = "https://archive.archlinux.org/packages/.all/"
Seblu's avatar
Seblu committed
REPORT_URL = "https://github.com/seblu/agetpkg/issues"
Seblu's avatar
Seblu committed
INDEX_FILENAME = "index.0.xz"
PKG_EXT = ".pkg.tar.xz"
SIG_EXT = ".sig"

class Error(BaseException):
    """Error"""
    ERR_USAGE = 1
    ERR_FATAL = 2
    ERR_ABORT = 3
    ERR_UNKNOWN = 4

class Url(object):
    """Remote Ressource"""

    HTTP_HEADERS = {
Seblu's avatar
Seblu committed
        "User-Agent": f"{NAME} v{VERSION}",
Seblu's avatar
Seblu committed
    }

    def __init__(self, url, timeout):
        self.url = url
        self.timeout = timeout

    def __str__(self):
        return self.url

    @property
    def exists(self):
        try:
            self.headers
            return True
        except Exception:
            return False

    @property
    def size(self):
        """Return the remote file size"""
        try:
            return int(self.headers["Content-Length"])
        except Exception as exp:
Seblu's avatar
Seblu committed
            raise Error(f"Failed to get size of {self.url}: {exp}")
Seblu's avatar
Seblu committed

    @property
    def lastmod(self):
        try:
            return int(mktime(parsedate(self.headers["Last-Modified"])))
        except Exception as exp:
Seblu's avatar
Seblu committed
            raise Error(f"Failed to get last modification date of {self.url}: {exp}")
Seblu's avatar
Seblu committed

    @property
    def headers(self):
        """Return a dict with url headers"""
        if not hasattr(self, "_headers"):
            try:
Seblu's avatar
Seblu committed
                debug(f"Request headers on URL: {self.url}")
Seblu's avatar
Seblu committed
                url_req = Request(self.url, method="HEAD", headers=self.HTTP_HEADERS)
                remote_fd = urlopen(url_req, timeout=self.timeout)
                self._headers = dict(remote_fd.getheaders())
            except Exception as exp:
Seblu's avatar
Seblu committed
                raise Error(f"Failed to get headers at {self}: {exp}")
Seblu's avatar
Seblu committed
        return getattr(self, "_headers")

    def download(self, destination):
        """Download URL to destination"""
Seblu's avatar
Seblu committed
        debug(f"Downloading from : {self.url}")
        debug(f"            to   : {destination}")
Seblu's avatar
Seblu committed
        try:
            url_req = Request(self.url, headers=self.HTTP_HEADERS)
            remote_fd = urlopen(url_req, timeout=self.timeout)
            local_fd = open(destination, "wb")
            copyfileobj(remote_fd, local_fd)
        except Exception as exp:
Seblu's avatar
Seblu committed
            raise Error(f"Failed to download {self}: {exp}")
Seblu's avatar
Seblu committed

class Package(Url):
    """Abstract a multi versionned package"""

    def __init__(self, url, timeout):
        self.url = Url(url, timeout)
        self.sigurl = Url(url + SIG_EXT, timeout)
        self.timeout = timeout
        self.filename = basename(url)
        self.sigfilename = self.filename + SIG_EXT
        # regex is not strict, but we are not validating something here
Seblu's avatar
Seblu committed
        m = match(r"^([\w@._+-]+)-((?:(\d+):)?([^-]+)-([^-]+))-(\w+)", self.filename)
Seblu's avatar
Seblu committed
        if m is None:
Seblu's avatar
Seblu committed
            raise Error(f"Unable to parse package info from filename {self.filename}")
Seblu's avatar
Seblu committed
        (self.name, self.full_version, self.epoch, self.version, self.release,
         self.arch) = m.groups()
        # no epoch means 0 (man PKGBUILD)
        if self.epoch is None:
            self.epoch = 0

    def __str__(self):
        return self.filename

    def __getitem__(self, key):
        try:
            return getattr(self, key)
        except AttributeError:
            raise KeyError()

    @property
    def size(self):
        """Return package Content-Length (size in bytes)"""
        return self.url.size

    @property
    def lastmod(self):
        """Return package Last-Modified date (in seconds since epoch)"""
        return self.url.lastmod

    def get(self, sign=True, force=False):
        """Download the package locally"""
        if not force:
            if exists(self.filename):
Seblu's avatar
Seblu committed
                raise Error(f"Local file {self.filename} already exists")
Seblu's avatar
Seblu committed
            if sign and exists(self.sigfilename):
Seblu's avatar
Seblu committed
                raise Error(f"Local file {self.sigfilename} already exists")
Seblu's avatar
Seblu committed
        self.url.download(self.filename)
        if sign and self.sigurl.exists:
            self.sigurl.download(self.sigfilename)

class Archive(object):
    """Abstract access to the package Archive"""

    def __init__(self, url, timeout, update=1):
        """Init the Archive interface
        url of the archive (flat style)
        update = update the local index cache (0: never, 1: when needed, 2: always)
        timeout = the socket timeout for network requests
        """
        if url[-1] != "/":
            raise Error("Archive URL must end with a /")
        self.url = url
        self.remote_index = Url(self.url + INDEX_FILENAME, timeout)
        self.local_index = join(save_cache_path(NAME), INDEX_FILENAME)
        self.timeout = timeout
        if update > 0:
            self.update_index(update == 2)
        self._load_index()

    def _load_index(self):
Seblu's avatar
Seblu committed
        debug(f"Loading index from {self.local_index}")
Seblu's avatar
Seblu committed
        fd = xzopen(self.local_index, "rb")
        self._index = OrderedDict()
        for line in fd.readlines():
            key = line.decode().rstrip()
Seblu's avatar
Seblu committed
            self._index[key] = Package(f"{self.url}{key}{PKG_EXT}", self.timeout)
        debug(f"Index loaded: {len(self._index)} packages")
Seblu's avatar
Seblu committed

    def update_index(self, force=False):
        """Update index remotely when needed"""
        debug("Check remote index for upgrade")
        if force:
            return self.remote_index.download(self.local_index)
        # get remote info
        try:
            remote_size = self.remote_index.size
            remote_lastmod = self.remote_index.lastmod
        except Exception as exp:
Seblu's avatar
Seblu committed
            debug(f"Failed to get remote index size/lastmod: {exp}")
Seblu's avatar
Seblu committed
            return self.remote_index.download(self.local_index)
        # get local info
        try:
            local_st = stat(self.local_index)
        except Exception as exp:
Seblu's avatar
Seblu committed
            debug(f"Failed to get local stat: {exp}")
Seblu's avatar
Seblu committed
            return self.remote_index.download(self.local_index)
        # compare size
        if remote_size != local_st.st_size:
Seblu's avatar
Seblu committed
            debug(f"Size differ between remote and local index ({remote_size} vs {local_st.st_size})")
Seblu's avatar
Seblu committed
            return self.remote_index.download(self.local_index)
        # compare date
        elif remote_lastmod > local_st.st_mtime:
Seblu's avatar
Seblu committed
            debug(f"Remote index is newer than local, updating it ({remote_lastmod} vs {local_st.st_mtime})")
Seblu's avatar
Seblu committed
            return self.remote_index.download(self.local_index)
        debug("Remote and local indexes seems equal. No update.")

    def search(self, name_pattern, version_pattern, release_pattern, arch_list=None):
Seblu's avatar
Seblu committed
        """Search for a package """
        name_regex = recompile(name_pattern)
        version_regex = recompile(version_pattern) if version_pattern is not None else None
        release_regex = recompile(release_pattern) if release_pattern is not None else None
Seblu's avatar
Seblu committed
        res = list()
        for pkg in self._index.values():
            if name_regex.search(pkg.name):
                # check against arch
                if arch_list is not None and len(arch_list) > 0:
                    if pkg.arch not in arch_list:
                        continue
                # check against version
                if version_regex is not None:
                    if not version_regex.search(pkg.version):
                        continue
                # check against release
                if release_regex is not None:
                    if not release_regex.search(pkg.release):
Seblu's avatar
Seblu committed
                        continue
                res += [pkg]
        return res

Seblu's avatar
Seblu committed
def which(binary):
    """lookup if bin exists into PATH"""
    dirs = environ.get("PATH", "").split(":")
    for d in dirs:
        if exists(join(d, binary)):
            return True
    return False

def pacman(args, asroot=True):
    """execute pacman (optionally as root)"""
    cmd = ["pacman" ] + args
    # add sudo or su if not root and
    if asroot and geteuid() != 0:
        if which("sudo"):
            cmd = ["sudo"] + cmd
        elif which("su"):
Seblu's avatar
Seblu committed
            cmd = ["su", "root", f"-c={' '.join(cmd)}" ]
Seblu's avatar
Seblu committed
        else:
Seblu's avatar
Seblu committed
            error(f"Unable to execute as root: {' '.join(cmd)}")
    debug(f"calling: {cmd}")
Seblu's avatar
Seblu committed
    call(cmd, close_fds=True)

Seblu's avatar
Seblu committed
def list_packages(packages, long=False):
    """display a list of packages on stdout"""
    if long:
        pattern = "%(name)s %(full_version)s %(arch)s %(size)s %(lastmod)s %(url)s"
    else:
        pattern = "%(name)s %(full_version)s %(arch)s"
    for package in packages:
        print(pattern % package)

def select_packages(packages, select_all=False):
Seblu's avatar
Seblu committed
    """select a package in a list"""
    # shortcut to one package
Seblu's avatar
Seblu committed
    if len(packages) == 1:
Seblu's avatar
Seblu committed
        yield packages[0]
    elif select_all:
        for pkg in packages:
            yield pkg
Seblu's avatar
Seblu committed
    else:
Seblu's avatar
Seblu committed
        # display a list of packages to select
Seblu's avatar
Seblu committed
        index = dict(enumerate(packages))
Seblu's avatar
Seblu committed
        pad = len(f"{max(index.keys()):d}")
Seblu's avatar
Seblu committed
        for i, pkg in index.items():
Seblu's avatar
Seblu committed
            print(f"{i:{pad}} {pkg}")
Seblu's avatar
Seblu committed
        selection = ""
Seblu's avatar
Seblu committed
        while not match(r"^(\d+ ){0,}\d+$", selection):
            try:
                selection = input("Select packages (* for all): ").strip()
            except EOFError:
                return
Seblu's avatar
Seblu committed
            if selection == "":
                return
            # shortcut to select all packages
            if selection == "*":
                for pkg in packages:
                    yield pkg
                return
        # parse selection
Seblu's avatar
Seblu committed
        numbers = [ int(x) for x in selection.split(" ") ]
        for num in numbers:
            if num in index.keys():
                yield index[num]
            else:
Seblu's avatar
Seblu committed
                warn(f"No package n°{num}")
Seblu's avatar
Seblu committed

def get_packages(packages, select_all=False):
Seblu's avatar
Seblu committed
    """download packages"""
    for pkg in select_packages(packages, select_all):
Seblu's avatar
Seblu committed
        pkg.get()
Seblu's avatar
Seblu committed

def install_packages(packages, select_all=False):
Seblu's avatar
Seblu committed
    """install packages in one shot to allow deps to work"""
    packages = list(select_packages(packages, select_all))
Seblu's avatar
Seblu committed
    with TemporaryDirectory() as tmpdir:
        cwd = getcwd()
        chdir(tmpdir)
        for pkg in packages:
            pkg.get()
        pacman(["-U"] + [ pkg.filename for pkg in packages ])
        chdir(cwd)
Seblu's avatar
Seblu committed

Seblu's avatar
Seblu committed
def parse_argv():
    '''Parse command line arguments'''
    local_arch = uname().machine
Seblu's avatar
Seblu committed
    p_main = ArgumentParser(prog=NAME)
Seblu's avatar
Seblu committed
    # update index options
Seblu's avatar
Seblu committed
    g_update = p_main.add_mutually_exclusive_group()
    g_update.add_argument("-u", "--force-update",
        action="store_const", dest="update", const=2, default=1,
        help="force index update")
    g_update.add_argument("-U", "--no-update",
        action="store_const", dest="update", const=0,
        help="disable index update")
Seblu's avatar
Seblu committed
    # action mode options
    g_action = p_main.add_mutually_exclusive_group()
    g_action.add_argument("-g", "--get", action="store_const", dest="mode",
        const="get", help="get matching packages (default mode)")
    g_action.add_argument("-l", "--list", action="store_const", dest="mode",
        const="list", help="only list matching packages")
    g_action.add_argument("-i", "--install", action="store_const", dest="mode",
        const="install", help="install matching packages")
    # common options
    p_main.add_argument("-a", "--all", action="store_true",
        help="select all packages without prompting")
Seblu's avatar
Seblu committed
    p_main.add_argument("-A", "--arch", nargs="*", default=[local_arch, "any"],
Seblu's avatar
Seblu committed
        help=f"filter by architectures (default: {local_arch} and any. empty means all)")
Seblu's avatar
Seblu committed
    p_main.add_argument("-v", "--verbose", action="store_true",
        help="display more information")
Seblu's avatar
Seblu committed
    p_main.add_argument("--url", help=f"archive URL, default: {ARCHIVE_URL}",
Seblu's avatar
Seblu committed
        default=environ.get("ARCHIVE_URL", ARCHIVE_URL))
Seblu's avatar
Seblu committed
    p_main.add_argument("-t", "--timeout", default=10,
        help="connection timeout (default: 10s)")
    p_main.add_argument("--version", action="version",
Seblu's avatar
Seblu committed
                        version=f"{NAME} version {VERSION}")
Seblu's avatar
Seblu committed
    p_main.add_argument("--debug", action="store_true",
                        help="debug mode")
    # positional args
Seblu's avatar
Seblu committed
    p_main.add_argument("package",
        help="regex to match a package name")
    p_main.add_argument("version", nargs="?",
        help="regex to match a package version")
    p_main.add_argument("release", nargs="?",
        help="regex to match a package release")
Seblu's avatar
Seblu committed
    return p_main.parse_args()

def main():
    '''Program entry point'''
    try:
        # parse command line
        args = parse_argv()
        # set global debug mode
        if args.debug:
            getLogger().setLevel(DEBUG)
        # init archive interface
        archive = Archive(args.url, args.timeout, args.update)
        # select target pacakges
        packages = archive.search(args.package, args.version, args.release, args.arch)
Seblu's avatar
Seblu committed
        if len(packages) == 0:
            print("No match found.")
            exit(0)
        if args.mode == "list":
            list_packages(packages, long=args.verbose)
Seblu's avatar
Seblu committed
        elif args.mode == "install":
            install_packages(packages, args.all)
Seblu's avatar
Seblu committed
        else:
            get_packages(packages, args.all)
Seblu's avatar
Seblu committed
    except KeyboardInterrupt:
        exit(Error.ERR_ABORT)
    except Error as exp:
        error(exp)
        exit(Error.ERR_FATAL)
    except Exception as exp:
Seblu's avatar
Seblu committed
        error(f"Unknown error. Please report it with --debug at {REPORT_URL}.")
Seblu's avatar
Seblu committed
        error(exp)
        if getLogger().getEffectiveLevel() == DEBUG:
            raise
        exit(Error.ERR_UNKNOWN)

if __name__ == '__main__':
    main()

# vim:set ts=4 sw=4 et ai: