mirror of
https://git.launchpad.net/ubuntu-dev-tools
synced 2025-03-14 00:21:08 +00:00
pull-pkg: update to use previous SourcePackage improvements
New pull-pkg allows pulling source, debs, ddebs, or udebs, or just listing all package files. Also, package lookup by binary name is done automatically.
This commit is contained in:
Dan Streetman
Dan Streetman
parent
ec72cf1538
commit
d7bcb012f6
281
pull-pkg
281
pull-pkg
@ -1,10 +1,11 @@
|
||||
#!/usr/bin/python3
|
||||
#
|
||||
# pull-lp-source -- pull a source package from Launchpad
|
||||
# Basic usage: pull-lp-source <source package> [<release>]
|
||||
# pull-pkg -- pull package files for debian/ubuntu/uca
|
||||
# Basic usage: pull-pkg -D distro -p type <package name> [version] [release]
|
||||
#
|
||||
# Copyright (C) 2008, Iain Lane <iain@orangesquash.org.uk>,
|
||||
# 2010-2011, Stefano Rivera <stefanor@ubuntu.com>
|
||||
# 2017, Dan Streetman <dan.streetman@canonical.com>
|
||||
#
|
||||
# ##################################################################
|
||||
#
|
||||
@ -23,124 +24,252 @@
|
||||
# ##################################################################
|
||||
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from optparse import OptionParser
|
||||
|
||||
from distro_info import UbuntuDistroInfo, DistroDataOutdated
|
||||
from distro_info import DebianDistroInfo
|
||||
|
||||
from ubuntutools.archive import UbuntuSourcePackage, DownloadError
|
||||
from ubuntutools.archive import (UbuntuSourcePackage, DebianSourcePackage,
|
||||
UbuntuCloudArchiveSourcePackage,
|
||||
DownloadError)
|
||||
from ubuntutools.config import UDTConfig
|
||||
from ubuntutools.lp.lpapicache import Distribution, Launchpad
|
||||
from ubuntutools.lp.lpapicache import (Distribution, Launchpad)
|
||||
from ubuntutools.lp.udtexceptions import (SeriesNotFoundException,
|
||||
PackageNotFoundException,
|
||||
PocketDoesNotExistError)
|
||||
from ubuntutools.logger import Logger
|
||||
from ubuntutools.misc import split_release_pocket
|
||||
from ubuntutools.misc import (split_release_pocket, host_architecture)
|
||||
|
||||
PULL_SOURCE = 'source'
|
||||
PULL_DEBS = 'debs'
|
||||
PULL_DDEBS = 'ddebs'
|
||||
PULL_UDEBS = 'udebs'
|
||||
PULL_LIST = 'list'
|
||||
|
||||
DEFAULT_PULL = PULL_SOURCE
|
||||
VALID_PULLS = [PULL_SOURCE, PULL_DEBS, PULL_DDEBS, PULL_UDEBS, PULL_LIST]
|
||||
|
||||
|
||||
def source_package_for(binary, release):
|
||||
"""Query DDE to find the source package for a particular binary
|
||||
Should really do this with LP, but it's not possible LP: #597041
|
||||
"""
|
||||
url = ('http://dde.debian.net/dde/q/udd/dist/d:ubuntu/r:%s/p:%s/?t=json'
|
||||
% (release, binary))
|
||||
data = None
|
||||
DISTRO_DEBIAN = 'debian'
|
||||
DISTRO_UBUNTU = 'ubuntu'
|
||||
DISTRO_UCA = 'uca'
|
||||
|
||||
DEFAULT_DISTRO = DISTRO_UBUNTU
|
||||
DISTRO_PKG_CLASS = {
|
||||
DISTRO_DEBIAN: DebianSourcePackage,
|
||||
DISTRO_UBUNTU: UbuntuSourcePackage,
|
||||
DISTRO_UCA: UbuntuCloudArchiveSourcePackage,
|
||||
}
|
||||
VALID_DISTROS = DISTRO_PKG_CLASS.keys()
|
||||
|
||||
|
||||
def parse_pull(pull):
|
||||
if not pull:
|
||||
pull = DEFAULT_PULL
|
||||
Logger.normal("Defaulting to pull %s", pull)
|
||||
|
||||
# allow 'dbgsym' as alias for 'ddebs'
|
||||
if pull == 'dbgsym':
|
||||
Logger.debug("Pulling '%s' for '%s'", PULL_DDEBS, pull)
|
||||
pull = PULL_DDEBS
|
||||
# assume anything starting with 'bin' means 'debs'
|
||||
if str(pull).startswith('bin'):
|
||||
Logger.debug("Pulling '%s' for '%s'", PULL_DEBS, pull)
|
||||
pull = PULL_DEBS
|
||||
# verify pull action is valid
|
||||
if pull not in VALID_PULLS:
|
||||
Logger.error("Invalid pull action '%s'", pull)
|
||||
sys.exit(1)
|
||||
|
||||
return pull
|
||||
|
||||
|
||||
def parse_distro(distro):
|
||||
if not distro:
|
||||
distro = DEFAULT_DISTRO
|
||||
Logger.normal("Defaulting to distro %s", distro)
|
||||
|
||||
distro = distro.lower()
|
||||
|
||||
# allow 'lp' for 'ubuntu'
|
||||
if distro == 'lp':
|
||||
Logger.debug("Using distro '%s' for '%s'", DISTRO_UBUNTU, distro)
|
||||
distro = DISTRO_UBUNTU
|
||||
# assume anything with 'cloud' is UCA
|
||||
if re.match(r'.*cloud.*', distro):
|
||||
Logger.debug("Using distro '%s' for '%s'", DISTRO_UCA, distro)
|
||||
distro = DISTRO_UCA
|
||||
# verify distro is valid
|
||||
if distro not in VALID_DISTROS:
|
||||
Logger.error("Invalid distro '%s'", distro)
|
||||
sys.exit(1)
|
||||
|
||||
return distro
|
||||
|
||||
|
||||
def parse_release(release, distro):
|
||||
if distro == DISTRO_UCA:
|
||||
# UCA is special; it is specified UBUNTURELEASE-UCARELEASE or just
|
||||
# UCARELEASE. The user could also specify UCARELEASE-POCKET. But UCA
|
||||
# archives always correspond to only one UBUNTURELEASE, and UCA archives
|
||||
# have only the Release pocket, so only UCARELEASE matters to us.
|
||||
for r in release.split('-'):
|
||||
if r in UbuntuCloudArchiveSourcePackage.getReleaseNames():
|
||||
Logger.debug("Using UCA release '%s'", r)
|
||||
return (r, None)
|
||||
raise SeriesNotFoundException('UCA release {} not found.'.format(release))
|
||||
|
||||
# Check if release[-pocket] is specified
|
||||
(release, pocket) = split_release_pocket(release, default=None)
|
||||
Logger.debug("Parsed release '%s' pocket '%s'", release, pocket)
|
||||
|
||||
if distro == DISTRO_DEBIAN:
|
||||
# This converts from the aliases like 'unstable'
|
||||
debian_info = DebianDistroInfo()
|
||||
codename = debian_info.codename(release)
|
||||
if codename:
|
||||
Logger.normal("Using release '%s' for '%s'", codename, release)
|
||||
release = codename
|
||||
|
||||
try:
|
||||
data = json.load(urllib.request.urlopen(url))['r']
|
||||
except urllib.error.URLError as e:
|
||||
Logger.error('Unable to retrieve package information from DDE: '
|
||||
'%s (%s)', url, str(e))
|
||||
except ValueError as e:
|
||||
Logger.error('Unable to parse JSON response from DDE: '
|
||||
'%s (%s)', url, str(e))
|
||||
if not data:
|
||||
return None
|
||||
return data[0]['source']
|
||||
d = Distribution(distro)
|
||||
Logger.debug("Distro '%s' is valid", distro)
|
||||
except:
|
||||
Logger.debug("Distro '%s' not valid", distro)
|
||||
raise SeriesNotFoundException("Distro {} not found".format(distro))
|
||||
|
||||
# let SeriesNotFoundException flow up
|
||||
d.getSeries(release)
|
||||
|
||||
Logger.debug("Using distro '%s' release '%s' pocket '%s'",
|
||||
distro, release, pocket)
|
||||
return (release, pocket)
|
||||
|
||||
|
||||
def main():
|
||||
usage = "Usage: %prog <package> [release|version]"
|
||||
usage = "Usage: %prog <package> [release[-pocket]|version]"
|
||||
opt_parser = OptionParser(usage)
|
||||
opt_parser.add_option('-v', '--verbose',
|
||||
dest='verbose', default=False,
|
||||
action='store_true',
|
||||
help="Print verbose/debug messages")
|
||||
opt_parser.add_option('-d', '--download-only',
|
||||
dest='download_only', default=False,
|
||||
action='store_true',
|
||||
help="Do not extract the source package")
|
||||
opt_parser.add_option('-m', '--mirror', metavar='UBUNTU_MIRROR',
|
||||
dest='ubuntu_mirror',
|
||||
help='Preferred Ubuntu mirror (default: Launchpad)')
|
||||
opt_parser.add_option('-m', '--mirror', dest='mirror',
|
||||
help='Preferred mirror')
|
||||
opt_parser.add_option('--no-conf',
|
||||
dest='no_conf', default=False, action='store_true',
|
||||
help="Don't read config files or environment "
|
||||
"variables")
|
||||
opt_parser.add_option('-a', '--arch',
|
||||
dest='arch', default=None,
|
||||
help="Get binary packages for specified architecture "
|
||||
"(default: {})".format(host_architecture()))
|
||||
opt_parser.add_option('-p', '--pull',
|
||||
dest='pull', default=None,
|
||||
help="What to pull: {} (default: {})"
|
||||
.format(", ".join(VALID_PULLS), DEFAULT_PULL))
|
||||
opt_parser.add_option('-D', '--distro',
|
||||
dest='distro', default=None,
|
||||
help="Pull from: {} (default: {})"
|
||||
.format(", ".join(VALID_DISTROS), DEFAULT_DISTRO))
|
||||
(options, args) = opt_parser.parse_args()
|
||||
if not args:
|
||||
opt_parser.error("Must specify package name")
|
||||
|
||||
distro = parse_distro(options.distro)
|
||||
mirrors = []
|
||||
|
||||
config = UDTConfig(options.no_conf)
|
||||
if options.ubuntu_mirror is None:
|
||||
options.ubuntu_mirror = config.get_value('UBUNTU_MIRROR')
|
||||
if options.mirror is None:
|
||||
options.mirror = config.get_value(distro.upper() + '_MIRROR')
|
||||
if options.mirror:
|
||||
mirrors.append(options.mirror)
|
||||
|
||||
pull = parse_pull(options.pull)
|
||||
if pull == PULL_DDEBS:
|
||||
ddebs_mirror = config.get_value(distro.upper() + '_DDEBS_MIRROR')
|
||||
if ddebs_mirror:
|
||||
mirrors.append(ddebs_mirror)
|
||||
|
||||
# Login anonymously to LP
|
||||
Launchpad.login_anonymously()
|
||||
|
||||
Logger.set_verbosity(options.verbose)
|
||||
|
||||
package = str(args[0]).lower()
|
||||
|
||||
ubuntu_info = UbuntuDistroInfo()
|
||||
if len(args) > 1: # Custom distribution specified.
|
||||
version = str(args[1])
|
||||
else:
|
||||
try:
|
||||
version = os.getenv('DIST') or ubuntu_info.devel()
|
||||
except DistroDataOutdated as e:
|
||||
Logger.warn("%s\nOr specify a distribution.", e)
|
||||
sys.exit(1)
|
||||
component = None
|
||||
|
||||
# Release, not package version number:
|
||||
version = None
|
||||
release = None
|
||||
pocket = None
|
||||
try:
|
||||
(release, pocket) = split_release_pocket(version, default=None)
|
||||
except PocketDoesNotExistError:
|
||||
pass
|
||||
if release in ubuntu_info.all:
|
||||
archive = Distribution('ubuntu').getArchive()
|
||||
|
||||
if len(args) > 1:
|
||||
try:
|
||||
spph = archive.getSourcePackage(package, release, pocket)
|
||||
except SeriesNotFoundException as e:
|
||||
Logger.error(str(e))
|
||||
sys.exit(1)
|
||||
except PackageNotFoundException as e:
|
||||
source_package = source_package_for(package, release)
|
||||
if source_package is not None and source_package != package:
|
||||
(release, pocket) = parse_release(args[1], distro)
|
||||
if len(args) > 2:
|
||||
version = args[2]
|
||||
except (SeriesNotFoundException, PocketDoesNotExistError):
|
||||
version = args[1]
|
||||
Logger.debug("Param '%s' not valid series, must be version", version)
|
||||
if len(args) > 2:
|
||||
try:
|
||||
spph = archive.getSourcePackage(source_package, release,
|
||||
pocket)
|
||||
package = source_package
|
||||
except PackageNotFoundException:
|
||||
Logger.error(str(e))
|
||||
(release, pocket) = parse_release(args[2], distro)
|
||||
except (SeriesNotFoundException, PocketDoesNotExistError):
|
||||
Logger.error("Can't find series for '%s' or '%s'",
|
||||
args[1], args[2])
|
||||
sys.exit(1)
|
||||
else:
|
||||
Logger.error(str(e))
|
||||
sys.exit(1)
|
||||
|
||||
version = spph.getVersion()
|
||||
component = spph.getComponent()
|
||||
|
||||
Logger.normal('Downloading %s version %s', package, version)
|
||||
srcpkg = UbuntuSourcePackage(package, version, component=component,
|
||||
mirrors=[options.ubuntu_mirror])
|
||||
try:
|
||||
srcpkg.pull()
|
||||
pkgcls = DISTRO_PKG_CLASS[distro]
|
||||
srcpkg = pkgcls(package=package, version=version,
|
||||
series=release, pocket=pocket,
|
||||
mirrors=mirrors)
|
||||
spph = srcpkg.lp_spph
|
||||
except PackageNotFoundException as e:
|
||||
Logger.error(str(e))
|
||||
sys.exit(1)
|
||||
|
||||
Logger.normal('Found %s', spph.display_name)
|
||||
|
||||
if pull == PULL_LIST:
|
||||
Logger.normal("Source files:")
|
||||
for f in srcpkg.dsc['Files']:
|
||||
Logger.normal(" %s", f['name'])
|
||||
Logger.normal("Binary files:")
|
||||
for f in spph.getBinaries(options.arch):
|
||||
Logger.normal(" %s", f.getFileName())
|
||||
sys.exit(0)
|
||||
|
||||
try:
|
||||
if pull == PULL_SOURCE:
|
||||
srcpkg.pull()
|
||||
if not options.download_only:
|
||||
srcpkg.unpack()
|
||||
else:
|
||||
name = '.*'
|
||||
if package != spph.getPackageName():
|
||||
Logger.normal("Pulling binary package '%s'", package)
|
||||
Logger.normal("Use package name '%s' to pull all binary packages",
|
||||
spph.getPackageName())
|
||||
name = package
|
||||
if pull == PULL_DEBS:
|
||||
name = r'{}(?<!-di)(?<!-dbgsym)$'.format(name)
|
||||
elif pull == PULL_DDEBS:
|
||||
name += '-dbgsym$'
|
||||
elif pull == PULL_UDEBS:
|
||||
name += '-di$'
|
||||
else:
|
||||
Logger.error("Unknown action '%s'", pull)
|
||||
sys.exit(1)
|
||||
total = srcpkg.pull_binaries(name=name, arch=options.arch)
|
||||
if total < 1:
|
||||
Logger.error("No %s found for %s", pull, spph.display_name)
|
||||
sys.exit(1)
|
||||
except DownloadError as e:
|
||||
Logger.error('Failed to download: %s', str(e))
|
||||
sys.exit(1)
|
||||
if not options.download_only:
|
||||
srcpkg.unpack()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user