ubuntu-dev-tools/seeded-in-ubuntu

#!/usr/bin/python3
#
# Copyright (C) 2011, Stefano Rivera <stefanor@ubuntu.com>
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

# pylint: disable=invalid-name
# pylint: enable=invalid-name

import argparse
import collections
import gzip
import json
import os
import time
import urllib.request

from ubuntutools import getLogger
from ubuntutools.lp.lpapicache import Distribution, Launchpad, PackageNotFoundException

Logger = getLogger()

DATA_URL = "http://qa.ubuntuwire.org/ubuntu-seeded-packages/seeded.json.gz"


def load_index(url):
    """Download a new copy of the image contents index, if necessary,
    and read it.
    """
    cachedir = os.path.expanduser("~/.cache/ubuntu-dev-tools")
    seeded = os.path.join(cachedir, "seeded.json.gz")

    if not os.path.isfile(seeded) or time.time() - os.path.getmtime(seeded) > 60 * 60 * 2:
        if not os.path.isdir(cachedir):
            os.makedirs(cachedir)
        urllib.request.urlretrieve(url, seeded)

    try:
        with gzip.open(seeded, "r") as f:
            return json.load(f)
    except Exception as e:  # pylint: disable=broad-except
        Logger.error(
            "Unable to parse seed data: %s. Deleting cached data, please try again.", str(e)
        )
        os.unlink(seeded)
    return None


def resolve_binaries(sources):
    """Return a dict of source:binaries for all binary packages built by
    sources
    """
    archive = Distribution("ubuntu").getArchive()
    binaries = {}
    for source in sources:
        try:
            spph = archive.getSourcePackage(source)
        except PackageNotFoundException as e:
            Logger.error(str(e))
            continue
        binaries[source] = sorted(set(bpph.getPackageName() for bpph in spph.getBinaries()))

    return binaries


def present_on(appearences):
    """Format a list of (flavor, type) tuples into a human-readable string"""
    present = collections.defaultdict(set)
    for flavor, type_ in appearences:
        present[flavor].add(type_)
    for flavor, types in present.items():
        if len(types) > 1:
            types.discard("supported")
    output = [f"  {flavor}: {', '.join(sorted(types))}" for flavor, types in present.items()]
    output.sort()
    return "\n".join(output)


def output_binaries(index, binaries):
    """Print binaries found in index"""
    for binary in binaries:
        if binary in index:
            Logger.info("%s is seeded in:", binary)
            Logger.info(present_on(index[binary]))
        else:
            Logger.info("%s is not seeded (and may not exist).", binary)


def output_by_source(index, by_source):
    """Logger.Info(binaries found in index. Grouped by source"""
    for source, binaries in by_source.items():
        seen = False
        if not binaries:
            Logger.info(
                "Status unknown: No binary packages built by the latest "
                "%s.\nTry again using -b and the expected binary packages.",
                source,
            )
            continue
        for binary in binaries:
            if binary in index:
                seen = True
                Logger.info("%s (from %s) is seeded in:", binary, source)
                Logger.info(present_on(index[binary]))
        if not seen:
            Logger.info("%s's binaries are not seeded.", source)


def main():
    """Query which images the specified packages are on"""
    parser = argparse.ArgumentParser(usage="%(prog)s [options] package...")
    parser.add_argument(
        "-b",
        "--binary",
        default=False,
        action="store_true",
        help="Binary packages are being specified, not source packages (fast)",
    )
    parser.add_argument(
        "-u",
        "--data-url",
        metavar="URL",
        default=DATA_URL,
        help="URL for the seeded packages index. Default: UbuntuWire",
    )
    parser.add_argument("packages", metavar="package", nargs="+", help=argparse.SUPPRESS)
    args = parser.parse_args()

    # Login anonymously to LP
    Launchpad.login_anonymously()

    index = load_index(args.data_url)
    if args.binary:
        output_binaries(index, args.packages)
    else:
        binaries = resolve_binaries(args.packages)
        output_by_source(index, binaries)


if __name__ == "__main__":
    main()