python-cdd 0.0.11 / usr / share / pyshared / cdd / PackageList.py

##########################################################################
#
# PackageList.py: manage the Package object list information.
#
#  Python-CDD is a library to make easier to build applications to
#  Custom Debian Distributions.
#  See http://projetos.ossystems.com.br/python-cdd for more information.
#
# ====================================================================
# Copyright (c) 2002-2005 O.S. Systems.	 All rights reserved.
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution.
#
#########################################################################
# Authors: Otavio Salvador <otavio@ossystems.com.br>
#		   Nat Budin <natb@brandeis.edu>
#		   Free Ekanayaka <free@agnula.org>
#		   Everton J. Carpes <everton@ossystems.com.br>

"""
PackageList classes and functions

This file implements a class to represent lists of packages and operations
related to those lists, like dependencies resolution and package filtering
"""
import os
import re
import sys
import logging
log = logging.getLogger("cdd.PackageList")

import apt_pkg
apt_pkg.init_system()


# Constants to resolve_depends
DEPENDS	   = 2 # Use Depends field to resolve
SUGGESTS   = 4 # Use Suggests field to resolve
RECOMMENDS = 8 # Use Recommends field to resolve


class PackageAlreadyExists(Exception):
	"""
	Exception called when someone tries to add a package but it was already
	included.

	Parameters:

	- *package*: The name of package.
	"""
	def __init__(self, package):
		Exception.__init__(self)
		self.package = package


class PackageDoesNotExist(Exception):
	"""
	Exception called when someone tries to remove a package that was not
	included.

	Parameters:

	- *package*: The name of package.
	"""
	def __init__(self, package):
		Exception.__init__(self)
		self.package = package


class BrokenDependencies(Exception):
	"""
	Exception called when, after trying to solve dependencies, we are left with
	broken ones.

	Parameters:

	- *deps*: List of unresolved dependencies.
	"""
	def __init__(self, deps, broken):
		Exception.__init__(self)
		self.deps = deps
		self.broken = broken


class InvalidFilter(Exception):
	"""
	Exception called when someone try to use a invalid filter option.

	Parameters:

	- *option*: The filter option.
	"""
	def __init__(self, option):
		Exception.__init__(self)
		self.option = option


class PackageList:
	"""
	This class is used to store a list of packages and provide ways to select
	subsets of those packages applying filters to the list.

	It also provides methods to resolve depencencies (Depends, Recommends and
	Suggests) against a full set of packages and include the needed packages
	on the list.

	To make filtering faster the class includes caches by name, subsection and
	priority.
	"""

	def __init__(self, package_list = None):
		self._name = {}
		self._subsection = {}
		self._priority = {}
		self._provides = {}
		self._rdepends = {}
		self._whereis = {}

		if package_list:
			for package in package_list.values():
				package_from = package_list.whereis(package['Package'])
				if package_from == self:
					self.add(package)
				else:
					self.add(package, package_from)

	def __len__(self):
		return len(self._name)

	def __getitem__(self, key):
		return self.get(key)

	# allow uses like ... if x in y:
	def __contains__(self, item):
		return self._name.has_key(item)

	def has_key(self, key):
		return self._name.has_key(key)

	def keys(self):
		"""Returns a copy of the list of package names"""
		return self._name.keys()

	def values(self):
		"""Returns a copy of the list of packages"""
		return self._name.values()

	def items(self):
		"""Returns a copy of the list of (name, package) pairs"""
		return self._name.items()

	def get(self, key):
		"""Returns a package given it's name"""
		if not self._name.has_key(key):
			raise PackageDoesNotExist, key
		return self._name[key]

	def add(self, package, package_list = None):
		"""
		Method to add a package to the list (updating caches).

		Note that the package is only added if it's name is not already on the
		list of packages, or else a PackageAlreadyExists exception is raised.
		"""
		if self._name.has_key(package['Package']):
			raise PackageAlreadyExists, package['Package']

		# Initialize the need lists
		if not self._subsection.has_key(package['Section']):
			self._subsection[package['Section']] = []
		if not self._priority.has_key(package['Priority']):
			self._priority[package['Priority']] = []

		# Update all basic cache information
		self._name[package['Package']] = package
		self._subsection[package['Section']].append(package)
		self._priority[package['Priority']].append(package)

		# Update rdepends cache information
		if package.has_key('Depends'):
			for depend in package['Depends']:
				for pkgname, version, cond in depend:
					if not self._rdepends.has_key(pkgname):
						self._rdepends[pkgname] = []
					self._rdepends[pkgname].append(package['Package'])

		# Add our foreign package reference on self._whereis
		if package_list:
			self._whereis[package['Package']] = package_list

		# Update provides information
		if package.has_key('Provides'):
			for prov in package['Provides']:
				if not self._provides.has_key(prov):
					self._provides[prov] = []
				self._provides[prov].append(package)

	def whereis(self, package):
		"""
		Method to get where a package come from.

		If the package was not on the list a PackageDoesNotExist exception is
		raised.

		Parameters:

		- *package*: package name to be localized
		"""
		if not isinstance(package, str):
			package = self._name[package]['Package']

		if not self._name.has_key(package):
			raise PackageDoesNotExist, package

		if not self._whereis.has_key(package):
			return self # package is native

		return self._whereis[package] # package is foreign

	def remove(self, package):
		"""
		Method to remove a package from the list (updating caches).

		If the package was not on the list a PackageDoesNotExist exception is
		raised.

		Parameters:

		- *package*: package to be removed
		"""
		if isinstance(package, str):
			if not self._name.has_key(package):
				raise PackageDoesNotExist, package
			package = self._name[package]
		else:
			if not self._name.has_key(package['Package']):
				raise PackageDoesNotExist, package['Package']

		del self._name[package['Package']]

		self._subsection[package['Section']].remove(package)
		self._priority[package['Priority']].remove(package)

		if package.has_key('Provides'):
			for prov in package['Provides']:
				self._provides[prov].remove(package)

	def extend(self, pkglist):
		"""
		Merge packages from pkglist with our list
		"""
		for pkg in pkglist.values():
			if	pkg['Package'] not in self.keys():
				self.add(pkg)

	def orphan_packages(self):
		"""
		"""
		dependencies = {}
		for package in self.values():
			if package.has_key('Depends'):
				for depend in _virtual_as_alternative(package['Depends'],
													  [self]):
					for pkgname, pkgver, cond in depend:
						if not dependencies.has_key(pkgname):
							dependencies[pkgname] = []
						if package['Package'] not in dependencies[pkgname]:
							dependencies[pkgname].append(package['Package'])

		orphan_list = self.keys()
		map(lambda x: orphan_list.remove(x), dependencies.keys())
		return orphan_list

	def resolve_depends(self, pkglist, field=DEPENDS, fastFail=True):
		"""
		Resolve our dependencies adding needed packages to our list from the
		`pkglist`.
		"""
		resolve_depends(self, pkglist, field, fastFail)

	def filter(self, condition, *architectures):
		"""
		Return a PackageList that contains only the packages that verify a set
		of conditions.

		The condition argument is a dictionary that can have the following keys:

		- *include-from*: includes packages from the given debian-cd task

		- *exclude-from*: excludes packages from the given debian-cd task

		- *subsection*: includes packages that belong to a section that match
		  the given regular expression

		- *priority*: includes packages that have a priority that match the
		  given regular expression

		- *name*: includes packages that have a name that match the given
		  regular expression.

		- *architectures*: architectures to use while processing the
		   include-from and exclude-from conditions. If you left this
		   blank, i386 will be used as default.

		- *field-XXX*: incldues package that have a control field XXX (eg Tag)
		   that match the givven regular expression

		- *exclude-subsection* *exclude-priority* *exclude-name* *exclude-field-XXX
			as above but exclude matching packages

		If the condition was not on the above list a InvalidFilter exception is
		raised.
		"""
		# Process the given regexp for givin data field.
		def process_regexp(regexp, cache, field, exclude=False):
			def addAction(toAdd):
				try:
					packages.add(toAdd)
				except PackageAlreadyExists:
					pass # We might repeat packages

			def removeAction(toRemove):
				if toRemove['Package'] in packages:
					packages.remove(toRemove['Package'])

			regexp = re.compile(regexp)
			if exclude:
				updateAction = removeAction
				candidatePackages = packages.values()
			else:
				updateAction = addAction
				candidatePackages = self.values()

			if cache is not None:
				for k, v in cache.iteritems():
					if k and regexp.match(k):
						# Add all match packages
						if isinstance(v, list):
							for i in v:
								updateAction(i)
						else:
							updateAction(v)

			if field is not None:
				for package in candidatePackages:
					fieldValue = package[field]
					if fieldValue is not None and regexp.match(fieldValue):
						updateAction(package)


		# Process a taskfile for the givin operation.
		def process_task(filename, operation):
			for arch in architectures:
				if arch not in valid_architectures:
					raise ValueError, "Invalid architecture name."

				raw = os.popen("cpp -undef -I %s \
									-I /usr/share/debian-cd/tasks \
									-DARCH_%s %s | \
								grep -v \'^$\|^#.*$\' | tr \'\n\' \' \'"
							   % (os.path.dirname(filename), arch, filename),
							   'r' ).readlines()[0].split()
				for praw in raw:
					pkg = self._name.get(praw, None)
					if operation == 'include-from':
						if pkg and pkg['Package'] not in packages:
							packages.add(pkg)
					else:
						if pkg and pkg['Package'] in packages:
							packages.remove(pkg)

		valid_architectures = ('alpha', 'amd64', 'arm', 'hppa', 'hurd-i386',
							   'i386', 'ia64', 'm68k', 'mips', 'mipsel',
							   'powerpc', 's390', 'sh', 'sparc')

		# Set i386 as default architecture
		if not len(architectures):
			architectures = ['i386']

		packages = PackageList()
		condition_found = False
		cacheForCondition = {
			"subsection" : self._subsection,
			"priority" : self._priority,
			"name" : self._name,
		}

		fieldRe = re.compile("field-(.*)")

		def invokeRegexpFilter(item, value, **kw):
			fieldMatch = fieldRe.match(item)
			field = None
			if fieldMatch is not None:
				field = fieldMatch.group(1)

			cache = cacheForCondition.get(item)
			if cache is None and field is None:
				raise InvalidFilter, item
			process_regexp(value, cache, field, **kw)


		for (item, value) in condition.items():
			if item == "include-from":
				condition_found = True
				process_task(value, item)
				continue
			if item.startswith("exclude-"):
				continue # we need to do it at end!

			invokeRegexpFilter(item, value)
			condition_found = True

		excludeRe = re.compile("exclude-(.*)")
		for (item, value) in condition.items():
			match = excludeRe.match(item)
			if match:
				# If we passed 'exclude-' conditions, we use
				# our packages as initial state.
				if not condition_found:
					for package in self.values():
						packages.add(package)
					condition_found = True

				if item == "exclude-from":
					process_task(value, item)
				else:
					invokeRegexpFilter(match.group(1), value, exclude=True)
		return packages



def _virtual_as_alternative(original, full_lists):
	# Replace all dependency of a virtual package with a or'ed list of
	# real packages
	new = []
	if original:
		# Package has depends
		for dep in original:
			new_dep	 = []
			for (pkgname, pkgver, cond) in dep:
				found = False
				for pkglist in full_lists:
					if pkgname in pkglist._provides:
						# if there is a non-vitual package, add it
						if pkgname in pkglist:
							new_dep.append((pkgname, pkgver, cond))
						# add the packages that provide the package
						for alternative in pkglist._provides[pkgname]:
							new_dep.append((alternative['Package'], '', ''))
						found = True

				# It wasn't a virtual package
				if not found:
					new_dep.append((pkgname, pkgver, cond))
			new.append(new_dep)

	return new


def resolve_depends(from_list, with_lists, field=DEPENDS, fastFail=True):
	def calc_scores(package, backtrace=[]):
		# Calc packages scores depending of its depends number
		depends = _virtual_as_alternative(package[field], with_lists)
		for depend in depends:
			for pkgname, pkgver, cond in depend:
				if not solutions.has_key(pkgname):
					for pkglist in with_lists:
						if pkgname in pkglist:
							if cond and not allow_version(pkglist[pkgname]["Version"], pkgver, cond):
								continue
							if pkgname not in backtrace and pkgname != package["Package"]:
								backtrace.append(pkgname)
								solution = Solution(pkgname)
								solution.pkglist = pkglist
								solution.score = calc_scores(pkglist[pkgname],backtrace)
								if pkglist._rdepends.has_key(pkgname):
									solution.goodness = len(pkglist._rdepends[pkgname])
								else:
									solution.goodness = 0
								solutions[pkgname] = solution
							break

		# Packages with depends has score 1 plus the minimum score
		# of all alternatives for its depends
		score = 1
		for depend in depends:
			depend_score = 0
			for pkgname, pkgver, cond in depend:
				solution = solutions.get(pkgname)
				if solution is not None:
					my_score = solution.score
					if my_score < depend_score or depend_score == 0:
						depend_score = my_score
			score += depend_score

		return score

	def select_field(package):
		# Select packages to minimize the total score
		depends = _virtual_as_alternative(package[field], with_lists)
		selected = {}
		for depend in depends:
			best_solution = Solution()
			for pkgname, pkgver, cond in depend:
				# If any alternative is already selected, use it
				if pkgname in from_list:
					best_solution.pkgname = pkgname
					best_solution.pkglist = from_list
					break

				if pkgname in selected:
					best_solution.pkgname = pkgname
					best_solution.pkglist = None
					break

				# Choose the alternative with best goodness and smalest score
				solution = solutions.get(pkgname)
				if solution is not None:
					best_solution.update(solution)

			if best_solution.pkgname and best_solution.pkgname not in selected:
				if best_solution.pkgname not in from_list:
					best_solution.addToList(from_list)
					selected[best_solution.pkgname] = best_solution.getPackage()
			else:
				if depends:
					if package not in broken:
						broken.append(package)
					broken_depends.setdefault(package["Package"], []).append(depend)
					if fastFail:
						return False
					continue

		for package in selected.itervalues():
			select_field(package)

	# Convert with_list to a list
	if not isinstance(with_lists, list):
		with_lists = [with_lists]

	# Our from_list is also used as alternative
	with_lists = [from_list] + with_lists

	# Set field to be use
	if field == DEPENDS:
		field = "Depends"
	elif field == SUGGESTS:
		field = "Suggests"
	elif field == RECOMMENDS:
		field = "Recommends"
	else:
		raise SyntaxError, "%s is invalid for field." % field

	class Solution:
		def __init__(self, pkgname=None):
			# Number of depends for package
			self.pkgname = pkgname
			self.score = 0
			self.goodness = 0
			self.pkglist = None

		def update(self, solution):
			if self.score == 0 or \
					(solution.score < self.score or
					 solution.goodness > self.goodness):
				self.score = solution.score
				self.goodness = solution.goodness
				self.pkglist = solution.pkglist
				self.pkgname = solution.pkgname

		def getPackage(self):
			return self.pkglist[self.pkgname]

		def addToList(self, pkglist):
			pkglist.add(self.getPackage(), self.pkglist)

	solutions = {}

	# Broken packages
	broken = []
	broken_depends = {}
	for package in from_list.values():
		calc_scores(package)
		select_field(package)

	if broken:
		raise BrokenDependencies(broken, broken_depends)


def allow_version(versionA, versionB, operator):
	return {
		"<" : lambda a,b : apt_pkg.version_compare(a,b) < 0,
		"<=" : lambda a,b : apt_pkg.version_compare(a,b) <= 0,
		"=" : lambda a,b : apt_pkg.version_compare(a,b) == 0,
		">=" : lambda a,b : apt_pkg.version_compare(a,b) >= 0,
		">" : lambda a,b : apt_pkg.version_compare(a,b) > 0,
	}[operator](versionA, versionB)