########################################################################## # # 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.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)