########################################################################## # # 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 # Nat Budin # Free Ekanayaka # Everton J. Carpes """ 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.InitSystem() # 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): log.debug("PackageList: constructor") 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): """ Resolve our dependencies adding needed packages to our list from the `pkglist`. """ try: resolve_depends(self, pkglist, field) except: raise 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. 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(data, regexp): regexp = re.compile(regexp) for k in data.keys(): if k and regexp.match(k): # Add all match packages if isinstance(data[k], list): for i in data[k]: try: packages.add(i) except PackageAlreadyExists: pass # We might repeat packages else: packages.add(data[k]) # 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 for (item, value) in condition.items(): if item == "include-from": condition_found = True process_task(value, item) continue elif item == "exclude-from": continue # we need to do it at end! elif item == "subsection": condition_found = True data = self._subsection process_regexp(data, value) continue elif item == "priority": condition_found = True data = self._priority process_regexp(data, value) continue elif item == "name": condition_found = True data = self._name process_regexp(data, value) continue else: raise InvalidFilter, item for (item, value) in condition.items(): if item == "exclude-from": # If we only passed 'exclude-from' condition, we use # our packages as initial state. if not condition_found: for package in self.values(): packages.add(package) process_task(value, item) continue 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: # First, if there is a non-vitual package, add it if pkgname in pkglist: new_dep.append((pkgname, pkgver, cond)) # And later 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): 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 scores.has_key(pkgname): for pkglist in with_lists: if pkgname in pkglist: found = True if pkgname not in backtrace and \ pkgname != package["Package"]: backtrace.append(pkgname) scores[pkgname] = calc_scores(pkglist[pkgname], backtrace) if pkglist._rdepends.has_key(pkgname): goodness[pkgname] = len(pkglist._rdepends[pkgname]) else: goodness[pkgname] = 0 break if not depends: # Packages without depends has score 1 (itself) return 1 else: # 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: if scores.has_key(pkgname): my_score = scores[pkgname] if my_score < depend_score or depend_score == 0: depend_score = my_score score += depend_score return score def select_field(package, backtrace = []): # Select packages to minimize the total score depends = _virtual_as_alternative(package[field], with_lists) for depend in depends: found = False depend_score = 0 goodness_score = 0 for pkgname, pkgver, cond in depend: # If any alternative is already selected, use it if pkgname in from_list: found = True selected = pkgname break # Choose the alternative with best goodness and smalest score if scores.has_key(pkgname): found = True my_score = scores[pkgname] my_goodness = goodness[pkgname] if depend_score == 0 or \ (my_score < depend_score or my_goodness > goodness_score): depend_score = my_score goodness_score = my_goodness selected = pkgname if not found: if depends: if package not in broken: broken.append(package) broken_depends[package["Package"]] = depends return False for pkglist in with_lists: if selected in pkglist: if selected not in from_list: from_list.add(pkglist[selected], pkglist) for selected in from_list.values(): if selected not in backtrace: backtrace.append(selected) select_field(selected, backtrace) # 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 # Number of depends for each package scores = {} # Goodness of each package goodness = {} # Broken packages broken = [] broken_depends = {} for package in from_list.values(): calc_scores(package) select_field(package) if broken: raise BrokenDependencies(broken, broken_depends)