diff options
| author | Michael Vogt <michael.vogt@ubuntu.com> | 2009-01-13 17:22:27 +0100 |
|---|---|---|
| committer | Michael Vogt <michael.vogt@ubuntu.com> | 2009-01-13 17:22:27 +0100 |
| commit | 38d602dc83006c51dfe4ed594d691ea9b0679498 (patch) | |
| tree | b7aedfba82c44cad6c3012f879b5d6d7e8ad1425 /aptsources | |
| parent | 12cf58d12b969010f3d98b2974d72bbb950b775f (diff) | |
| parent | 614897f798d9f16591fbd29ebe2a6c5674102d2d (diff) | |
| download | python-apt-38d602dc83006c51dfe4ed594d691ea9b0679498.tar.gz | |
* apt/*.py:
- Almost complete cleanup of the code
- Remove inconsistent use of tabs and spaces (Closes: #505443)
- Improved documentation
* apt/debfile.py:
- Drop get*() methods, as they are deprecated and were
never in a stable release
- Make DscSrcPackage working
* apt/gtk/widgets.py:
- Fix the code and document the signals
* Introduce new documentation build with Sphinx
- Contains style Guide (Closes: #481562)
- debian/rules: Build the documentation here
- setup.py: Remove pydoc building and add new docs.
- debian/examples: Include examples from documentation
- debian/python-apt.docs:
+ Change html/ to build/doc/html.
+ Add build/doc/text for the text-only documentation
* setup.py:
- Only create build/data when building, not all the time
- Remove build/mo and build/data on clean -a
* debian/control:
- Remove the Conflicts on python2.3-apt, python2.4-apt, as
they are only needed for oldstable (sarge)
- Build-Depend on python-sphinx (>= 0.5)
* aptsources/distinfo.py:
- Allow @ in mirror urls (Closes: #478171) (LP: #223097)
* Merge Ben Finney's whitespace changes (Closes: #481563)
* Merge Ben Finney's do not use has_key() (Closes: #481878)
* Do not use deprecated form of raise statement (Closes: #494259)
* Add support for PkgRecords.SHA256Hash (Closes: #456113)
Diffstat (limited to 'aptsources')
| -rw-r--r-- | aptsources/__init__.py | 3 | ||||
| -rw-r--r-- | aptsources/distinfo.py | 94 | ||||
| -rw-r--r-- | aptsources/distro.py | 777 | ||||
| -rw-r--r-- | aptsources/sourceslist.py | 803 |
4 files changed, 863 insertions, 814 deletions
diff --git a/aptsources/__init__.py b/aptsources/__init__.py index d6b3605c..c02341c2 100644 --- a/aptsources/__init__.py +++ b/aptsources/__init__.py @@ -1,5 +1,4 @@ - import apt_pkg - + # init the package system apt_pkg.init() diff --git a/aptsources/distinfo.py b/aptsources/distinfo.py index a6c0faf8..59fa7e02 100644 --- a/aptsources/distinfo.py +++ b/aptsources/distinfo.py @@ -1,5 +1,5 @@ #!/usr/bin/env python -# +# # distinfo.py - provide meta information for distro repositories # # Copyright (c) 2005 Gustavo Noronha Silva @@ -7,17 +7,17 @@ # # Author: Gustavo Noronha Silva <kov@debian.org> # Sebastian Heinlein <glatzor@ubuntu.com> -# -# This program is free software; you can redistribute it and/or -# modify it under the terms of the GNU General Public License as +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License as # published by the Free Software Foundation; either version 2 of the # License, or (at your option) any later version. -# +# # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. -# +# # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 @@ -27,16 +27,17 @@ import os import gettext from os import getenv import ConfigParser -import string +import re + import apt_pkg -#from gettext import gettext as _ -import gettext -def _(s): return gettext.dgettext("python-apt", s) -import re +def _(s): + return gettext.dgettext("python-apt", s) + class Template: + def __init__(self): self.name = None self.child = False @@ -55,45 +56,56 @@ class Template: def has_component(self, comp): ''' Check if the distribution provides the given component ''' return comp in map(lambda c: c.name, self.components) - + def is_mirror(self, url): ''' Check if a given url of a repository is a valid mirror ''' proto, hostname, dir = split_url(url) - if self.mirror_set.has_key(hostname): + if hostname in self.mirror_set: return self.mirror_set[hostname].has_repository(proto, dir) else: return False + class Component: + def __init__(self, name, desc=None, long_desc=None): self.name = name self.description = desc self.description_long = long_desc + def get_description(self): - if self.description_long != None: + if self.description_long is not None: return self.description_long - elif self.description != None: + elif self.description is not None: return self.description else: return None + def set_description(self, desc): self.description = desc + def set_description_long(self, desc): self.description_long = desc + def get_description_long(self): return self.description_long + class Mirror: ''' Storage for mirror related information ''' + def __init__(self, proto, hostname, dir, location=None): self.hostname = hostname self.repositories = [] self.add_repository(proto, dir) self.location = location + def add_repository(self, proto, dir): self.repositories.append(Repository(proto, dir)) + def get_repositories_for_proto(self, proto): return filter(lambda r: r.proto == proto, self.repositories) + def has_repository(self, proto, dir): if dir is None: return False @@ -101,28 +113,38 @@ class Mirror: if r.proto == proto and dir in r.dir: return True return False + def get_repo_urls(self): return map(lambda r: r.get_url(self.hostname), self.repositories) + def get_location(self): return self.location + def set_location(self, location): self.location = location + class Repository: + def __init__(self, proto, dir): self.proto = proto self.dir = dir + def get_info(self): return self.proto, self.dir + def get_url(self, hostname): return "%s://%s/%s" % (self.proto, hostname, self.dir) + def split_url(url): ''' split a given URL into the protocoll, the hostname and the dir part ''' - return map(lambda a,b: a, re.split(":*\/+", url, maxsplit=2), - [None, None, None]) + return map(lambda a, b: a, re.split(":*\/+", url, maxsplit=2), + [None, None, None]) + class DistInfo: + def __init__(self, dist = None, base_dir = "/usr/share/python-apt/templates"): @@ -132,7 +154,9 @@ class DistInfo: location = None match_loc = re.compile(r"^#LOC:(.+)$") - match_mirror_line = re.compile(r"^(#LOC:.+)|(((http)|(ftp)|(rsync)|(file)|(https))://[A-Za-z/\.:\-_]+)$") + match_mirror_line = re.compile( + r"^(#LOC:.+)|(((http)|(ftp)|(rsync)|(file)|(https))://" + r"[A-Za-z0-9/\.:\-_@]+)$") #match_mirror_line = re.compile(r".+") if not dist: @@ -143,21 +167,20 @@ class DistInfo: self.dist = dist - map_mirror_sets = {} dist_fname = "%s/%s.info" % (base_dir, dist) - dist_file = open (dist_fname) + dist_file = open(dist_fname) if not dist_file: return template = None component = None for line in dist_file: - tokens = line.split (':', 1) - if len (tokens) < 2: + tokens = line.split(':', 1) + if len(tokens) < 2: continue - field = tokens[0].strip () - value = tokens[1].strip () + field = tokens[0].strip() + value = tokens[1].strip() if field == 'ChangelogURI': self.changelogs_uri = _(value) elif field == 'MetaReleaseURI': @@ -190,14 +213,14 @@ class DistInfo: template.match_uri = value elif field == 'MatchURI-%s' % self.arch: template.match_uri = value - elif (field == 'MirrorsFile' or + elif (field == 'MirrorsFile' or field == 'MirrorsFile-%s' % self.arch): - if not map_mirror_sets.has_key(value): + if value not in map_mirror_sets: mirror_set = {} try: mirror_data = filter(match_mirror_line.match, - map(string.strip, open(value))) - except: + [x.strip() for x in open(value)]) + except Exception: print "WARNING: Failed to read mirror file" mirror_data = [] for line in mirror_data: @@ -205,10 +228,11 @@ class DistInfo: location = match_loc.sub(r"\1", line) continue (proto, hostname, dir) = split_url(line) - if mirror_set.has_key(hostname): + if hostname in mirror_set: mirror_set[hostname].add_repository(proto, dir) else: - mirror_set[hostname] = Mirror(proto, hostname, dir, location) + mirror_set[hostname] = Mirror( + proto, hostname, dir, location) map_mirror_sets[value] = mirror_set template.mirror_set = map_mirror_sets[value] elif field == 'Description': @@ -230,7 +254,7 @@ class DistInfo: if not template: return # reuse some properties of the parent template - if template.match_uri == None and template.child: + if template.match_uri is None and template.child: for t in template.parents: if t.match_uri: template.match_uri = t.match_uri @@ -243,11 +267,11 @@ class DistInfo: if component and not template.has_component(component.name): template.components.append(component) component = None - self.templates.append(template) + self.templates.append(template) if __name__ == "__main__": - d = DistInfo ("Ubuntu", "/usr/share/python-apt/templates") + d = DistInfo("Ubuntu", "/usr/share/python-apt/templates") print d.changelogs_uri for template in d.templates: print "\nSuite: %s" % template.name @@ -257,8 +281,8 @@ if __name__ == "__main__": if template.mirror_set != {}: print "Mirrors: %s" % template.mirror_set.keys() for comp in template.components: - print " %s -%s -%s" % (comp.name, - comp.description, + print " %s -%s -%s" % (comp.name, + comp.description, comp.description_long) for child in template.children: print " %s" % child.description diff --git a/aptsources/distro.py b/aptsources/distro.py index f271bbc4..d8f191d6 100644 --- a/aptsources/distro.py +++ b/aptsources/distro.py @@ -2,439 +2,454 @@ # # Copyright (c) 2004-2007 Canonical Ltd. # 2006-2007 Sebastian Heinlein -# +# # Authors: Sebastian Heinlein <glatzor@ubuntu.com> # Michael Vogt <mvo@debian.org> -# -# This program is free software; you can redistribute it and/or -# modify it under the terms of the GNU General Public License as +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License as # published by the Free Software Foundation; either version 2 of the # License, or (at your option) any later version. -# +# # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. -# +# # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 # USA -import string import gettext import re import os import sys import gettext -def _(s): return gettext.dgettext("python-apt", s) + + +def _(s): + return gettext.dgettext("python-apt", s) class NoDistroTemplateException(Exception): - pass + pass + class Distribution: - def __init__(self, id, codename, description, release): - """ Container for distribution specific informations """ - # LSB information - self.id = id - self.codename = codename - self.description = description - self.release = release - - self.binary_type = "deb" - self.source_type = "deb-src" - - def get_sources(self, sourceslist): - """ - Find the corresponding template, main and child sources - for the distribution - """ - self.sourceslist = sourceslist - # corresponding sources - self.source_template = None - self.child_sources = [] - self.main_sources = [] - self.disabled_sources = [] - self.cdrom_sources = [] - self.download_comps = [] - self.enabled_comps = [] - self.cdrom_comps = [] - self.used_media = [] - self.get_source_code = False - self.source_code_sources = [] - - # location of the sources - self.default_server = "" - self.main_server = "" - self.nearest_server = "" - self.used_servers = [] - - # find the distro template - for template in self.sourceslist.matcher.templates: - if self.is_codename(template.name) and\ - template.distribution == self.id: - #print "yeah! found a template for %s" % self.description - #print template.description, template.base_uri, template.components - self.source_template = template - break - if self.source_template == None: - raise (NoDistroTemplateException, - "Error: could not find a distribution template") - - # find main and child sources - media = [] - comps = [] - cdrom_comps = [] - enabled_comps = [] - source_code = [] - for source in self.sourceslist.list: - if source.invalid == False and\ - self.is_codename(source.dist) and\ - source.template and\ - self.is_codename(source.template.name): - #print "yeah! found a distro repo: %s" % source.line - # cdroms need do be handled differently - if source.uri.startswith("cdrom:") and \ - source.disabled == False: - self.cdrom_sources.append(source) - cdrom_comps.extend(source.comps) - elif source.uri.startswith("cdrom:") and \ - source.disabled == True: - self.cdrom_sources.append(source) - elif source.type == self.binary_type and \ - source.disabled == False: - self.main_sources.append(source) - comps.extend(source.comps) - media.append(source.uri) - elif source.type == self.binary_type and \ - source.disabled == True: - self.disabled_sources.append(source) - elif source.type == self.source_type and source.disabled == False: - self.source_code_sources.append(source) - elif source.type == self.source_type and source.disabled == True: - self.disabled_sources.append(source) - if source.invalid == False and\ - source.template in self.source_template.children: - if source.disabled == False and source.type == self.binary_type: - self.child_sources.append(source) - elif source.disabled == False and source.type == self.source_type: - self.source_code_sources.append(source) - else: - self.disabled_sources.append(source) - self.download_comps = set(comps) - self.cdrom_comps = set(cdrom_comps) - enabled_comps.extend(comps) - enabled_comps.extend(cdrom_comps) - self.enabled_comps = set(enabled_comps) - self.used_media = set(media) - - self.get_mirrors() - - def get_mirrors(self, mirror_template=None): - """ - Provide a set of mirrors where you can get the distribution from - """ - # the main server is stored in the template - self.main_server = self.source_template.base_uri + def __init__(self, id, codename, description, release): + """ Container for distribution specific informations """ + # LSB information + self.id = id + self.codename = codename + self.description = description + self.release = release - # other used servers - for medium in self.used_media: - if not medium.startswith("cdrom:"): - # seems to be a network source - self.used_servers.append(medium) + self.binary_type = "deb" + self.source_type = "deb-src" - if len(self.main_sources) == 0: - self.default_server = self.main_server - else: - self.default_server = self.main_sources[0].uri - - # get a list of country codes and real names - self.countries = {} - try: - f = open("/usr/share/iso-codes/iso_3166.tab", "r") - lines = f.readlines() - for line in lines: - parts = line.split("\t") - self.countries[parts[0].lower()] = parts[1].strip() - except: - print "could not open file '%s'" % file - else: - f.close() - - # try to guess the nearest mirror from the locale - self.country = None - self.country_code = None - locale = os.getenv("LANG", default="en_UK") - a = locale.find("_") - z = locale.find(".") - if z == -1: - z = len(locale) - country_code = locale[a+1:z].lower() - - if mirror_template: - self.nearest_server = mirror_template % country_code - - if self.countries.has_key(country_code): - self.country = self.countries[country_code] - self.country_code = country_code - - def _get_mirror_name(self, server): - ''' Try to get a human readable name for the main mirror of a country - Customize for different distributions ''' - country = None - i = server.find("://") - l = server.find(".archive.ubuntu.com") - if i != -1 and l != -1: - country = server[i+len("://"):l] - if self.countries.has_key(country): - # TRANSLATORS: %s is a country - return _("Server for %s") % \ - gettext.dgettext("iso_3166", - self.countries[country].rstrip()).rstrip() - else: - return("%s" % server.rstrip("/ ")) - - def get_server_list(self): - ''' Return a list of used and suggested servers ''' - def compare_mirrors(mir1, mir2): - '''Helper function that handles comaprision of mirror urls - that could contain trailing slashes''' - return re.match(mir1.strip("/ "), mir2.rstrip("/ ")) - - # Store all available servers: - # Name, URI, active - mirrors = [] - if len(self.used_servers) < 1 or \ - (len(self.used_servers) == 1 and \ - compare_mirrors(self.used_servers[0], self.main_server)): - mirrors.append([_("Main server"), self.main_server, True]) - mirrors.append([self._get_mirror_name(self.nearest_server), - self.nearest_server, False]) - elif len(self.used_servers) == 1 and not \ - compare_mirrors(self.used_servers[0], self.main_server): - mirrors.append([_("Main server"), self.main_server, False]) - # Only one server is used - server = self.used_servers[0] - - # Append the nearest server if it's not already used - if not compare_mirrors(server, self.nearest_server): - mirrors.append([self._get_mirror_name(self.nearest_server), - self.nearest_server, False]) - mirrors.append([self._get_mirror_name(server), server, True]) - - elif len(self.used_servers) > 1: - # More than one server is used. Since we don't handle this case - # in the user interface we set "custom servers" to true and - # append a list of all used servers - mirrors.append([_("Main server"), self.main_server, False]) - mirrors.append([self._get_mirror_name(self.nearest_server), - self.nearest_server, False]) - mirrors.append([_("Custom servers"), None, True]) - for server in self.used_servers: - if compare_mirrors(server, self.nearest_server) or\ - compare_mirrors(server, self.main_server): - continue - elif not [self._get_mirror_name(server), server, False] in mirrors: - mirrors.append([self._get_mirror_name(server), server, False]) - - return mirrors - - def add_source(self, type=None, - uri=None, dist=None, comps=None, comment=""): - """ - Add distribution specific sources - """ - if uri == None: - # FIXME: Add support for the server selector - uri = self.default_server - if dist == None: - dist = self.codename - if comps == None: - comps = list(self.enabled_comps) - if type == None: - type = self.binary_type - new_source = self.sourceslist.add(type, uri, dist, comps, comment) - # if source code is enabled add a deb-src line after the new - # source - if self.get_source_code == True and type == self.binary_type: - self.sourceslist.add(self.source_type, uri, dist, comps, comment, - file=new_source.file, - pos=self.sourceslist.list.index(new_source)+1) - - def enable_component(self, comp): - """ - Enable a component in all main, child and source code sources - (excluding cdrom based sources) + def get_sources(self, sourceslist): + """ + Find the corresponding template, main and child sources + for the distribution + """ - comp: the component that should be enabled - """ - def add_component_only_once(source, comps_per_dist): + self.sourceslist = sourceslist + # corresponding sources + self.source_template = None + self.child_sources = [] + self.main_sources = [] + self.disabled_sources = [] + self.cdrom_sources = [] + self.download_comps = [] + self.enabled_comps = [] + self.cdrom_comps = [] + self.used_media = [] + self.get_source_code = False + self.source_code_sources = [] + + # location of the sources + self.default_server = "" + self.main_server = "" + self.nearest_server = "" + self.used_servers = [] + + # find the distro template + for template in self.sourceslist.matcher.templates: + if self.is_codename(template.name) and\ + template.distribution == self.id: + #print "yeah! found a template for %s" % self.description + #print template.description, template.base_uri, \ + # template.components + self.source_template = template + break + if self.source_template is None: + raise NoDistroTemplateException("Error: could not find a " + "distribution template") + + # find main and child sources + media = [] + comps = [] + cdrom_comps = [] + enabled_comps = [] + source_code = [] + for source in self.sourceslist.list: + if source.invalid == False and\ + self.is_codename(source.dist) and\ + source.template and\ + self.is_codename(source.template.name): + #print "yeah! found a distro repo: %s" % source.line + # cdroms need do be handled differently + if source.uri.startswith("cdrom:") and \ + source.disabled == False: + self.cdrom_sources.append(source) + cdrom_comps.extend(source.comps) + elif source.uri.startswith("cdrom:") and \ + source.disabled == True: + self.cdrom_sources.append(source) + elif source.type == self.binary_type and \ + source.disabled == False: + self.main_sources.append(source) + comps.extend(source.comps) + media.append(source.uri) + elif source.type == self.binary_type and \ + source.disabled == True: + self.disabled_sources.append(source) + elif source.type == self.source_type \ + and source.disabled == False: + self.source_code_sources.append(source) + elif source.type == self.source_type \ + and source.disabled == True: + self.disabled_sources.append(source) + if source.invalid == False and\ + source.template in self.source_template.children: + if source.disabled == False \ + and source.type == self.binary_type: + self.child_sources.append(source) + elif source.disabled == False \ + and source.type == self.source_type: + self.source_code_sources.append(source) + else: + self.disabled_sources.append(source) + self.download_comps = set(comps) + self.cdrom_comps = set(cdrom_comps) + enabled_comps.extend(comps) + enabled_comps.extend(cdrom_comps) + self.enabled_comps = set(enabled_comps) + self.used_media = set(media) + + self.get_mirrors() + + def get_mirrors(self, mirror_template=None): """ - Check if we already added the component to the repository, since - a repository could be splitted into different apt lines. If not - add the component + Provide a set of mirrors where you can get the distribution from """ - # if we don't that distro, just reutnr (can happen for e.g. - # dapper-update only in deb-src - if not comps_per_dist.has_key(source.dist): - return - # if we have seen this component already for this distro, - # return (nothing to do - if comp in comps_per_dist[source.dist]: - return - # add it - source.comps.append(comp) - comps_per_dist[source.dist].add(comp) - - sources = [] - sources.extend(self.main_sources) - sources.extend(self.child_sources) - # store what comps are enabled already per distro (where distro is - # e.g. "dapper", "dapper-updates") - comps_per_dist = {} - comps_per_sdist = {} - for s in sources: - if s.type == self.binary_type: - if not comps_per_dist.has_key(s.dist): - comps_per_dist[s.dist] = set() - map(comps_per_dist[s.dist].add, s.comps) - for s in self.source_code_sources: - if s.type == self.source_type: - if not comps_per_sdist.has_key(s.dist): - comps_per_sdist[s.dist] = set() - map(comps_per_sdist[s.dist].add, s.comps) - - # check if there is a main source at all - if len(self.main_sources) < 1: - # create a new main source - self.add_source(comps=["%s"%comp]) - else: - # add the comp to all main, child and source code sources - for source in sources: - add_component_only_once(source, comps_per_dist) + # the main server is stored in the template + self.main_server = self.source_template.base_uri - # check if there is a main source code source at all - if self.get_source_code == True: - if len(self.source_code_sources) < 1: - # create a new main source - self.add_source(type=self.source_type, comps=["%s"%comp]) + # other used servers + for medium in self.used_media: + if not medium.startswith("cdrom:"): + # seems to be a network source + self.used_servers.append(medium) + + if len(self.main_sources) == 0: + self.default_server = self.main_server + else: + self.default_server = self.main_sources[0].uri + + # get a list of country codes and real names + self.countries = {} + try: + f = open("/usr/share/iso-codes/iso_3166.tab", "r") + lines = f.readlines() + for line in lines: + parts = line.split("\t") + self.countries[parts[0].lower()] = parts[1].strip() + except: + print "could not open file '%s'" % file + else: + f.close() + + # try to guess the nearest mirror from the locale + self.country = None + self.country_code = None + locale = os.getenv("LANG", default="en_UK") + a = locale.find("_") + z = locale.find(".") + if z == -1: + z = len(locale) + country_code = locale[a+1:z].lower() + + if mirror_template: + self.nearest_server = mirror_template % country_code + + if country_code in self.countries: + self.country = self.countries[country_code] + self.country_code = country_code + + def _get_mirror_name(self, server): + ''' Try to get a human readable name for the main mirror of a country + Customize for different distributions ''' + country = None + i = server.find("://") + l = server.find(".archive.ubuntu.com") + if i != -1 and l != -1: + country = server[i+len("://"):l] + if country in self.countries: + # TRANSLATORS: %s is a country + return _("Server for %s") % \ + gettext.dgettext("iso_3166", + self.countries[country].rstrip()).rstrip() else: - # add the comp to all main, child and source code sources - for source in self.source_code_sources: - add_component_only_once(source, comps_per_sdist) + return("%s" % server.rstrip("/ ")) + + def get_server_list(self): + ''' Return a list of used and suggested servers ''' + + def compare_mirrors(mir1, mir2): + ''' Helper function that handles comaprision of mirror urls + that could contain trailing slashes''' + return re.match(mir1.strip("/ "), mir2.rstrip("/ ")) + + # Store all available servers: + # Name, URI, active + mirrors = [] + if len(self.used_servers) < 1 or \ + (len(self.used_servers) == 1 and \ + compare_mirrors(self.used_servers[0], self.main_server)): + mirrors.append([_("Main server"), self.main_server, True]) + mirrors.append([self._get_mirror_name(self.nearest_server), + self.nearest_server, False]) + elif len(self.used_servers) == 1 and not \ + compare_mirrors(self.used_servers[0], self.main_server): + mirrors.append([_("Main server"), self.main_server, False]) + # Only one server is used + server = self.used_servers[0] + + # Append the nearest server if it's not already used + if not compare_mirrors(server, self.nearest_server): + mirrors.append([self._get_mirror_name(self.nearest_server), + self.nearest_server, False]) + mirrors.append([self._get_mirror_name(server), server, True]) + + elif len(self.used_servers) > 1: + # More than one server is used. Since we don't handle this case + # in the user interface we set "custom servers" to true and + # append a list of all used servers + mirrors.append([_("Main server"), self.main_server, False]) + mirrors.append([self._get_mirror_name(self.nearest_server), + self.nearest_server, False]) + mirrors.append([_("Custom servers"), None, True]) + for server in self.used_servers: + mirror_entry = [self._get_mirror_name(server), server, False] + if compare_mirrors(server, self.nearest_server) or\ + compare_mirrors(server, self.main_server): + continue + elif not mirror_entry in mirrors: + mirrors.append(mirror_entry) + + return mirrors + + def add_source(self, type=None, + uri=None, dist=None, comps=None, comment=""): + """ + Add distribution specific sources + """ + if uri is None: + # FIXME: Add support for the server selector + uri = self.default_server + if dist is None: + dist = self.codename + if comps is None: + comps = list(self.enabled_comps) + if type is None: + type = self.binary_type + new_source = self.sourceslist.add(type, uri, dist, comps, comment) + # if source code is enabled add a deb-src line after the new + # source + if self.get_source_code == True and type == self.binary_type: + self.sourceslist.add(self.source_type, uri, dist, comps, comment, + file=new_source.file, + pos=self.sourceslist.list.index(new_source)+1) + + def enable_component(self, comp): + """ + Enable a component in all main, child and source code sources + (excluding cdrom based sources) - def disable_component(self, comp): - """ - Disable a component in all main, child and source code sources - (excluding cdrom based sources) - """ - sources = [] - sources.extend(self.main_sources) - sources.extend(self.child_sources) - sources.extend(self.source_code_sources) - if comp in self.cdrom_comps: + comp: the component that should be enabled + """ + + def add_component_only_once(source, comps_per_dist): + """ + Check if we already added the component to the repository, since + a repository could be splitted into different apt lines. If not + add the component + """ + # if we don't that distro, just reutnr (can happen for e.g. + # dapper-update only in deb-src + if source.dist not in comps_per_dist: + return + # if we have seen this component already for this distro, + # return (nothing to do + if comp in comps_per_dist[source.dist]: + return + # add it + source.comps.append(comp) + comps_per_dist[source.dist].add(comp) + + sources = [] + sources.extend(self.main_sources) + sources.extend(self.child_sources) + # store what comps are enabled already per distro (where distro is + # e.g. "dapper", "dapper-updates") + comps_per_dist = {} + comps_per_sdist = {} + for s in sources: + if s.type == self.binary_type: + if s.dist not in comps_per_dist: + comps_per_dist[s.dist] = set() + map(comps_per_dist[s.dist].add, s.comps) + for s in self.source_code_sources: + if s.type == self.source_type: + if s.dist not in comps_per_sdist: + comps_per_sdist[s.dist] = set() + map(comps_per_sdist[s.dist].add, s.comps) + + # check if there is a main source at all + if len(self.main_sources) < 1: + # create a new main source + self.add_source(comps=["%s"%comp]) + else: + # add the comp to all main, child and source code sources + for source in sources: + add_component_only_once(source, comps_per_dist) + + # check if there is a main source code source at all + if self.get_source_code == True: + if len(self.source_code_sources) < 1: + # create a new main source + self.add_source(type=self.source_type, comps=["%s"%comp]) + else: + # add the comp to all main, child and source code sources + for source in self.source_code_sources: + add_component_only_once(source, comps_per_sdist) + + def disable_component(self, comp): + """ + Disable a component in all main, child and source code sources + (excluding cdrom based sources) + """ sources = [] sources.extend(self.main_sources) - for source in sources: - if comp in source.comps: - source.comps.remove(comp) - if len(source.comps) < 1: - self.sourceslist.remove(source) - - def change_server(self, uri): - ''' Change the server of all distro specific sources to - a given host ''' - def change_server_of_source(source, uri, seen): - # Avoid creating duplicate entries - source.uri = uri - for comp in source.comps: - if [source.uri, source.dist, comp] in seen: + sources.extend(self.child_sources) + sources.extend(self.source_code_sources) + if comp in self.cdrom_comps: + sources = [] + sources.extend(self.main_sources) + for source in sources: + if comp in source.comps: source.comps.remove(comp) - else: - seen.append([source.uri, source.dist, comp]) - if len(source.comps) < 1: - self.sourceslist.remove(source) - seen_binary = [] - seen_source = [] - self.default_server = uri - for source in self.main_sources: - change_server_of_source(source, uri, seen_binary) - for source in self.child_sources: - # Do not change the forces server of a child source - if source.template.base_uri == None or \ - source.template.base_uri != source.uri: + if len(source.comps) < 1: + self.sourceslist.remove(source) + + def change_server(self, uri): + ''' Change the server of all distro specific sources to + a given host ''' + + def change_server_of_source(source, uri, seen): + # Avoid creating duplicate entries + source.uri = uri + for comp in source.comps: + if [source.uri, source.dist, comp] in seen: + source.comps.remove(comp) + else: + seen.append([source.uri, source.dist, comp]) + if len(source.comps) < 1: + self.sourceslist.remove(source) + + seen_binary = [] + seen_source = [] + self.default_server = uri + for source in self.main_sources: change_server_of_source(source, uri, seen_binary) - for source in self.source_code_sources: - change_server_of_source(source, uri, seen_source) + for source in self.child_sources: + # Do not change the forces server of a child source + if source.template.base_uri is None or \ + source.template.base_uri != source.uri: + change_server_of_source(source, uri, seen_binary) + for source in self.source_code_sources: + change_server_of_source(source, uri, seen_source) + + def is_codename(self, name): + ''' Compare a given name with the release codename. ''' + if name == self.codename: + return True + else: + return False - def is_codename(self, name): - ''' Compare a given name with the release codename. ''' - if name == self.codename: - return True - else: - return False class DebianDistribution(Distribution): - ''' Class to support specific Debian features ''' + ''' Class to support specific Debian features ''' + + def is_codename(self, name): + ''' Compare a given name with the release codename and check if + if it can be used as a synonym for a development releases ''' + if name == self.codename or self.release in ("testing", "unstable"): + return True + else: + return False + + def _get_mirror_name(self, server): + ''' Try to get a human readable name for the main mirror of a country + Debian specific ''' + country = None + i = server.find("://ftp.") + l = server.find(".debian.org") + if i != -1 and l != -1: + country = server[i+len("://ftp."):l] + if country in self.countries: + # TRANSLATORS: %s is a country + return _("Server for %s") % gettext.dgettext( + "iso_3166", self.countries[country].rstrip()).rstrip() + else: + return("%s" % server.rstrip("/ ")) + + def get_mirrors(self): + Distribution.get_mirrors( + self, mirror_template="http://ftp.%s.debian.org/debian/") - def is_codename(self, name): - ''' Compare a given name with the release codename and check if - if it can be used as a synonym for a development releases ''' - if name == self.codename or self.release in ("testing", "unstable"): - return True - else: - return False - - def _get_mirror_name(self, server): - ''' Try to get a human readable name for the main mirror of a country - Debian specific ''' - country = None - i = server.find("://ftp.") - l = server.find(".debian.org") - if i != -1 and l != -1: - country = server[i+len("://ftp."):l] - if self.countries.has_key(country): - # TRANSLATORS: %s is a country - return _("Server for %s") % \ - gettext.dgettext("iso_3166", - self.countries[country].rstrip()).rstrip() - else: - return("%s" % server.rstrip("/ ")) - - def get_mirrors(self): - Distribution.get_mirrors(self, - mirror_template="http://ftp.%s.debian.org/debian/") class UbuntuDistribution(Distribution): - ''' Class to support specific Ubuntu features ''' - def get_mirrors(self): - Distribution.get_mirrors(self, - mirror_template="http://%s.archive.ubuntu.com/ubuntu/") + ''' Class to support specific Ubuntu features ''' + + def get_mirrors(self): + Distribution.get_mirrors( + self, mirror_template="http://%s.archive.ubuntu.com/ubuntu/") + -def get_distro(id=None,codename=None,description=None,release=None): - """ +def get_distro(id=None, codename=None, description=None, release=None): + """ Check the currently used distribution and return the corresponding - distriubtion class that supports distro specific features. - + distriubtion class that supports distro specific features. + If no paramter are given the distro will be auto detected via a call to lsb-release """ # make testing easier if not (id and codename and description and release): - lsb_info = [] - for lsb_option in ["-i", "-c", "-d", "-r"]: - pipe = os.popen("lsb_release %s -s" % lsb_option) - lsb_info.append(pipe.read().strip()) - del pipe - (id, codename, description, release) = lsb_info + lsb_info = [] + for lsb_option in ["-i", "-c", "-d", "-r"]: + pipe = os.popen("lsb_release %s -s" % lsb_option) + lsb_info.append(pipe.read().strip()) + del pipe + (id, codename, description, release) = lsb_info if id == "Ubuntu": - return UbuntuDistribution(id, codename, description, - release) + return UbuntuDistribution(id, codename, description, release) elif id == "Debian": return DebianDistribution(id, codename, description, release) else: return Distribution(id, codename, description, release) - diff --git a/aptsources/sourceslist.py b/aptsources/sourceslist.py index e0f32f8d..dc2a5d45 100644 --- a/aptsources/sourceslist.py +++ b/aptsources/sourceslist.py @@ -1,438 +1,449 @@ # aptsource.py - Provide an abstraction of the sources.list -# +# # Copyright (c) 2004-2007 Canonical Ltd. # 2004 Michiel Sikkes # 2006-2007 Sebastian Heinlein -# +# # Author: Michiel Sikkes <michiel@eyesopened.nl> # Michael Vogt <mvo@debian.org> # Sebastian Heinlein <glatzor@ubuntu.com> -# -# This program is free software; you can redistribute it and/or -# modify it under the terms of the GNU General Public License as +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License as # published by the Free Software Foundation; either version 2 of the # License, or (at your option) any later version. -# +# # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. -# +# # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 # USA - -import string + import gettext -import re -import apt_pkg import glob -import shutil -import time import os.path +import re +import shutil import sys +import time + +import apt_pkg +from aptsources.distinfo import DistInfo -#from UpdateManager.Common.DistInfo import DistInfo -from distinfo import DistInfo # some global helpers + + def is_mirror(master_uri, compare_uri): - """check if the given add_url is idential or a mirror of orig_uri - e.g. master_uri = archive.ubuntu.com - compare_uri = de.archive.ubuntu.com - -> True - """ - # remove traling spaces and "/" - compare_uri = compare_uri.rstrip("/ ") - master_uri = master_uri.rstrip("/ ") - # uri is identical - if compare_uri == master_uri: - #print "Identical" - return True - # add uri is a master site and orig_uri has the from "XX.mastersite" - # (e.g. de.archive.ubuntu.com) - try: - compare_srv = compare_uri.split("//")[1] - master_srv = master_uri.split("//")[1] - #print "%s == %s " % (add_srv, orig_srv) - except IndexError: # ok, somethings wrong here - #print "IndexError" + """ check if the given add_url is idential or a mirror of orig_uri + e.g. master_uri = archive.ubuntu.com + compare_uri = de.archive.ubuntu.com + -> True + """ + # remove traling spaces and "/" + compare_uri = compare_uri.rstrip("/ ") + master_uri = master_uri.rstrip("/ ") + # uri is identical + if compare_uri == master_uri: + #print "Identical" + return True + # add uri is a master site and orig_uri has the from "XX.mastersite" + # (e.g. de.archive.ubuntu.com) + try: + compare_srv = compare_uri.split("//")[1] + master_srv = master_uri.split("//")[1] + #print "%s == %s " % (add_srv, orig_srv) + except IndexError: # ok, somethings wrong here + #print "IndexError" + return False + # remove the leading "<country>." (if any) and see if that helps + if "." in compare_srv and \ + compare_srv[compare_srv.index(".")+1:] == master_srv: + #print "Mirror" + return True return False - # remove the leading "<country>." (if any) and see if that helps - if "." in compare_srv and \ - compare_srv[compare_srv.index(".")+1:] == master_srv: - #print "Mirror" - return True - return False + def uniq(s): - """ simple and efficient way to return uniq list """ - return list(set(s)) + """ simple and efficient way to return uniq list """ + return list(set(s)) + class SourceEntry: - """ single sources.list entry """ - def __init__(self, line,file=None): - self.invalid = False # is the source entry valid - self.disabled = False # is it disabled ('#' in front) - self.type = "" # what type (deb, deb-src) - self.uri = "" # base-uri - self.dist = "" # distribution (dapper, edgy, etc) - self.comps = [] # list of available componetns (may empty) - self.comment = "" # (optional) comment - self.line = line # the original sources.list line - if file == None: - file = apt_pkg.Config.FindDir("Dir::Etc")+apt_pkg.Config.Find("Dir::Etc::sourcelist") - self.file = file # the file that the entry is located in - self.parse(line) - self.template = None # type DistInfo.Suite - self.children = [] - - def __eq__(self, other): - """ equal operator for two sources.list entries """ - return (self.disabled == other.disabled and - self.type == other.type and - self.uri == other.uri and - self.dist == other.dist and - self.comps == other.comps) - - - def mysplit(self, line): - """ a split() implementation that understands the sources.list - format better and takes [] into account (for e.g. cdroms) """ - line = string.strip(line) - pieces = [] - tmp = "" - # we are inside a [..] block - p_found = False - space_found = False - for i in range(len(line)): - if line[i] == "[": - p_found=True - tmp += line[i] - elif line[i] == "]": - p_found=False - tmp += line[i] - elif space_found and not line[i].isspace(): # we skip one or more space + """ single sources.list entry """ + + def __init__(self, line, file=None): + self.invalid = False # is the source entry valid + self.disabled = False # is it disabled ('#' in front) + self.type = "" # what type (deb, deb-src) + self.uri = "" # base-uri + self.dist = "" # distribution (dapper, edgy, etc) + self.comps = [] # list of available componetns + # (may empty) + self.comment = "" # (optional) comment + self.line = line # the original sources.list line + if file is None: + file = apt_pkg.Config.FindDir( + "Dir::Etc")+apt_pkg.Config.Find("Dir::Etc::sourcelist") + self.file = file # the file that the entry is located in + self.parse(line) + self.template = None # type DistInfo.Suite + self.children = [] + + def __eq__(self, other): + """ equal operator for two sources.list entries """ + return (self.disabled == other.disabled and + self.type == other.type and + self.uri == other.uri and + self.dist == other.dist and + self.comps == other.comps) + + def mysplit(self, line): + """ a split() implementation that understands the sources.list + format better and takes [] into account (for e.g. cdroms) """ + line = line.strip() + pieces = [] + tmp = "" + # we are inside a [..] block + p_found = False space_found = False - pieces.append(tmp) - tmp = line[i] - elif line[i].isspace() and not p_found: # found a whitespace - space_found = True - else: - tmp += line[i] - # append last piece - if len(tmp) > 0: - pieces.append(tmp) - return pieces - - def parse(self,line): - """ parse a given sources.list (textual) line and break it up - into the field we have """ - line = string.strip(self.line) - #print line - # check if the source is enabled/disabled - if line == "" or line == "#": # empty line - self.invalid = True - return - if line[0] == "#": - self.disabled = True - pieces = string.split(line[1:]) - # if it looks not like a disabled deb line return - if not pieces[0] in ("rpm", "rpm-src", "deb", "deb-src"): - self.invalid = True - return - else: - line = line[1:] - # check for another "#" in the line (this is treated as a comment) - i = line.find("#") - if i > 0: - self.comment = line[i+1:] - line = line[:i] - # source is ok, split it and see what we have - pieces = self.mysplit(line) - # Sanity check - if len(pieces) < 3: - self.invalid = True - return - # Type, deb or deb-src - self.type = string.strip(pieces[0]) - # Sanity check - if self.type not in ("deb", "deb-src", "rpm", "rpm-src"): - self.invalid = True - return - # URI - self.uri = string.strip(pieces[1]) - if len(self.uri) < 1: - self.invalid = True - # distro and components (optional) - # Directory or distro - self.dist = string.strip(pieces[2]) - if len(pieces) > 3: - # List of components - self.comps = pieces[3:] - else: - self.comps = [] - - def set_enabled(self, new_value): - """ set a line to enabled or disabled """ - self.disabled = not new_value - # enable, remove all "#" from the start of the line - if new_value == True: - i=0 - self.line = string.lstrip(self.line) - while self.line[i] == "#": - i += 1 - self.line = self.line[i:] - else: - # disabled, add a "#" - if string.strip(self.line)[0] != "#": - self.line = "#" + self.line - - def __str__(self): - """ debug helper """ - return self.str().strip() - - def str(self): - """ return the current line as string """ - if self.invalid: - return self.line - line = "" - if self.disabled: - line = "# " - line += "%s %s %s" % (self.type, self.uri, self.dist) - if len(self.comps) > 0: - line += " " + " ".join(self.comps) - if self.comment != "": - line += " #"+self.comment - line += "\n" - return line - + for i in range(len(line)): + if line[i] == "[": + p_found=True + tmp += line[i] + elif line[i] == "]": + p_found=False + tmp += line[i] + elif space_found and not line[i].isspace(): + # we skip one or more space + space_found = False + pieces.append(tmp) + tmp = line[i] + elif line[i].isspace() and not p_found: + # found a whitespace + space_found = True + else: + tmp += line[i] + # append last piece + if len(tmp) > 0: + pieces.append(tmp) + return pieces + + def parse(self, line): + """ parse a given sources.list (textual) line and break it up + into the field we have """ + line = self.line.strip() + #print line + # check if the source is enabled/disabled + if line == "" or line == "#": # empty line + self.invalid = True + return + if line[0] == "#": + self.disabled = True + pieces = line[1:].strip() + # if it looks not like a disabled deb line return + if not pieces[0] in ("rpm", "rpm-src", "deb", "deb-src"): + self.invalid = True + return + else: + line = line[1:] + # check for another "#" in the line (this is treated as a comment) + i = line.find("#") + if i > 0: + self.comment = line[i+1:] + line = line[:i] + # source is ok, split it and see what we have + pieces = self.mysplit(line) + # Sanity check + if len(pieces) < 3: + self.invalid = True + return + # Type, deb or deb-src + self.type = pieces[0].strip() + # Sanity check + if self.type not in ("deb", "deb-src", "rpm", "rpm-src"): + self.invalid = True + return + # URI + self.uri = pieces[1].strip() + if len(self.uri) < 1: + self.invalid = True + # distro and components (optional) + # Directory or distro + self.dist = pieces[2].strip() + if len(pieces) > 3: + # List of components + self.comps = pieces[3:] + else: + self.comps = [] + + def set_enabled(self, new_value): + """ set a line to enabled or disabled """ + self.disabled = not new_value + # enable, remove all "#" from the start of the line + if new_value: + self.line = self.line.lstrip().lstrip('#') + else: + # disabled, add a "#" + if self.line.strip()[0] != "#": + self.line = "#" + self.line + + def __str__(self): + """ debug helper """ + return self.str().strip() + + def str(self): + """ return the current line as string """ + if self.invalid: + return self.line + line = "" + if self.disabled: + line = "# " + line += "%s %s %s" % (self.type, self.uri, self.dist) + if len(self.comps) > 0: + line += " " + " ".join(self.comps) + if self.comment != "": + line += " #"+self.comment + line += "\n" + return line + + class NullMatcher(object): - """ a Matcher that does nothing """ - def match(self, s): - return True + """ a Matcher that does nothing """ + + def match(self, s): + return True + class SourcesList: - """ represents the full sources.list + sources.list.d file """ - def __init__(self, - withMatcher=True, - matcherPath="/usr/share/python-apt/templates/"): - self.list = [] # the actual SourceEntries Type - if withMatcher: - self.matcher = SourceEntryMatcher(matcherPath) - else: - self.matcher = NullMatcher() - self.refresh() - - def refresh(self): - """ update the list of known entries """ - self.list = [] - # read sources.list - dir = apt_pkg.Config.FindDir("Dir::Etc") - file = apt_pkg.Config.Find("Dir::Etc::sourcelist") - self.load(dir+file) - # read sources.list.d - partsdir = apt_pkg.Config.FindDir("Dir::Etc::sourceparts") - for file in glob.glob("%s/*.list" % partsdir): - self.load(file) - # check if the source item fits a predefined template - for source in self.list: - if source.invalid == False: - self.matcher.match(source) - - def __iter__(self): - """ simple iterator to go over self.list, returns SourceEntry - types """ - for entry in self.list: - yield entry - raise StopIteration - - def add(self, type, uri, dist, orig_comps, comment="", pos=-1, file=None): - """ - Add a new source to the sources.list. - The method will search for existing matching repos and will try to - reuse them as far as possible - """ - # create a working copy of the component list so that - # we can modify it later - comps = orig_comps[:] - # check if we have this source already in the sources.list - for source in self.list: - if source.disabled == False and source.invalid == False and \ - source.type == type and uri == source.uri and \ - source.dist == dist: - for new_comp in comps: - if new_comp in source.comps: - # we have this component already, delete it from the new_comps - # list - del comps[comps.index(new_comp)] - if len(comps) == 0: - return source - for source in self.list: - # if there is a repo with the same (type, uri, dist) just add the - # components - if source.disabled == False and source.invalid == False and \ - source.type == type and uri == source.uri and \ - source.dist == dist: - comps = uniq(source.comps + comps) - source.comps = comps - return source - # if there is a corresponding repo which is disabled, enable it - elif source.disabled == True and source.invalid == False and \ - source.type == type and uri == source.uri and \ - source.dist == dist and \ - len(set(source.comps) & set(comps)) == len(comps): - source.disabled = False - return source - # there isn't any matching source, so create a new line and parse it - line = "%s %s %s" % (type,uri,dist) - for c in comps: - line = line + " " + c; - if comment != "": - line = "%s #%s\n" %(line,comment) - line = line + "\n" - new_entry = SourceEntry(line) - if file != None: - new_entry.file = file - self.matcher.match(new_entry) - self.list.insert(pos, new_entry) - return new_entry - - def remove(self, source_entry): - """ remove the specified entry from the sources.list """ - self.list.remove(source_entry) - - def restoreBackup(self, backup_ext): - " restore sources.list files based on the backup extension " - dir = apt_pkg.Config.FindDir("Dir::Etc") - file = apt_pkg.Config.Find("Dir::Etc::sourcelist") - if os.path.exists(dir+file+backup_ext) and \ - os.path.exists(dir+file): - shutil.copy(dir+file+backup_ext,dir+file) - # now sources.list.d - partsdir = apt_pkg.Config.FindDir("Dir::Etc::sourceparts") - for file in glob.glob("%s/*.list" % partsdir): - if os.path.exists(file+backup_ext): - shutil.copy(file+backup_ext,file) - - def backup(self, backup_ext=None): - """ make a backup of the current source files, if no backup extension - is given, the current date/time is used (and returned) """ - already_backuped = set() - if backup_ext == None: - backup_ext = time.strftime("%y%m%d.%H%M") - for source in self.list: - if not source.file in already_backuped and os.path.exists(source.file): - shutil.copy(source.file,"%s%s" % (source.file,backup_ext)) - return backup_ext - - def load(self,file): - """ (re)load the current sources """ - try: - f = open(file, "r") - lines = f.readlines() - for line in lines: - source = SourceEntry(line,file) - self.list.append(source) - except: - print "could not open file '%s'" % file - else: - f.close() - - def save(self): - """ save the current sources """ - files = {} - # write an empty default config file if there aren't any sources - if len(self.list) == 0: - path = "%s%s" % (apt_pkg.Config.FindDir("Dir::Etc"), - apt_pkg.Config.Find("Dir::Etc::sourcelist")) - header = ("## See sources.list(5) for more information, especialy\n" + """ represents the full sources.list + sources.list.d file """ + + def __init__(self, + withMatcher=True, + matcherPath="/usr/share/python-apt/templates/"): + self.list = [] # the actual SourceEntries Type + if withMatcher: + self.matcher = SourceEntryMatcher(matcherPath) + else: + self.matcher = NullMatcher() + self.refresh() + + def refresh(self): + """ update the list of known entries """ + self.list = [] + # read sources.list + dir = apt_pkg.Config.FindDir("Dir::Etc") + file = apt_pkg.Config.Find("Dir::Etc::sourcelist") + self.load(dir+file) + # read sources.list.d + partsdir = apt_pkg.Config.FindDir("Dir::Etc::sourceparts") + for file in glob.glob("%s/*.list" % partsdir): + self.load(file) + # check if the source item fits a predefined template + for source in self.list: + if not source.invalid: + self.matcher.match(source) + + def __iter__(self): + """ simple iterator to go over self.list, returns SourceEntry + types """ + for entry in self.list: + yield entry + raise StopIteration + + def add(self, type, uri, dist, orig_comps, comment="", pos=-1, file=None): + """ + Add a new source to the sources.list. + The method will search for existing matching repos and will try to + reuse them as far as possible + """ + # create a working copy of the component list so that + # we can modify it later + comps = orig_comps[:] + # check if we have this source already in the sources.list + for source in self.list: + if not source.disabled and not source.invalid and \ + source.type == type and uri == source.uri and \ + source.dist == dist: + for new_comp in comps: + if new_comp in source.comps: + # we have this component already, delete it + # from the new_comps list + del comps[comps.index(new_comp)] + if len(comps) == 0: + return source + for source in self.list: + # if there is a repo with the same (type, uri, dist) just add the + # components + if not source.disabled and not source.invalid and \ + source.type == type and uri == source.uri and \ + source.dist == dist: + comps = uniq(source.comps + comps) + source.comps = comps + return source + # if there is a corresponding repo which is disabled, enable it + elif source.disabled and not source.invalid and \ + source.type == type and uri == source.uri and \ + source.dist == dist and \ + len(set(source.comps) & set(comps)) == len(comps): + source.disabled = False + return source + # there isn't any matching source, so create a new line and parse it + line = "%s %s %s" % (type, uri, dist) + for c in comps: + line = line + " " + c + if comment != "": + line = "%s #%s\n" % (line, comment) + line = line + "\n" + new_entry = SourceEntry(line) + if file is not None: + new_entry.file = file + self.matcher.match(new_entry) + self.list.insert(pos, new_entry) + return new_entry + + def remove(self, source_entry): + """ remove the specified entry from the sources.list """ + self.list.remove(source_entry) + + def restoreBackup(self, backup_ext): + " restore sources.list files based on the backup extension " + dir = apt_pkg.Config.FindDir("Dir::Etc") + file = apt_pkg.Config.Find("Dir::Etc::sourcelist") + if os.path.exists(dir+file+backup_ext) and \ + os.path.exists(dir+file): + shutil.copy(dir+file+backup_ext, dir+file) + # now sources.list.d + partsdir = apt_pkg.Config.FindDir("Dir::Etc::sourceparts") + for file in glob.glob("%s/*.list" % partsdir): + if os.path.exists(file+backup_ext): + shutil.copy(file+backup_ext, file) + + def backup(self, backup_ext=None): + """ make a backup of the current source files, if no backup extension + is given, the current date/time is used (and returned) """ + already_backuped = set() + if backup_ext is None: + backup_ext = time.strftime("%y%m%d.%H%M") + for source in self.list: + if not source.file in already_backuped \ + and os.path.exists(source.file): + shutil.copy(source.file, "%s%s" % (source.file, backup_ext)) + return backup_ext + + def load(self, file): + """ (re)load the current sources """ + try: + f = open(file, "r") + lines = f.readlines() + for line in lines: + source = SourceEntry(line, file) + self.list.append(source) + except: + print "could not open file '%s'" % file + else: + f.close() + + def save(self): + """ save the current sources """ + files = {} + # write an empty default config file if there aren't any sources + if len(self.list) == 0: + path = "%s%s" % (apt_pkg.Config.FindDir("Dir::Etc"), + apt_pkg.Config.Find("Dir::Etc::sourcelist")) + header = ( + "## See sources.list(5) for more information, especialy\n" "# Remember that you can only use http, ftp or file URIs\n" "# CDROMs are managed through the apt-cdrom tool.\n") - open(path,"w").write(header) - return - for source in self.list: - if not files.has_key(source.file): - files[source.file]=open(source.file,"w") - files[source.file].write(source.str()) - for f in files: - files[f].close() - - def check_for_relations(self, sources_list): - """get all parent and child channels in the sources list""" - parents = [] - used_child_templates = {} - for source in sources_list: - # try to avoid checking uninterressting sources - if source.template == None: - continue - # set up a dict with all used child templates and corresponding - # source entries - if source.template.child == True: - key = source.template - if not used_child_templates.has_key(key): - used_child_templates[key] = [] - temp = used_child_templates[key] - temp.append(source) - else: - # store each source with children aka. a parent :) - if len(source.template.children) > 0: - parents.append(source) - #print self.used_child_templates - #print self.parents - return (parents, used_child_templates) - -# matcher class to make a source entry look nice -# lots of predefined matchers to make it i18n/gettext friendly + open(path, "w").write(header) + return + for source in self.list: + if source.file not in files: + files[source.file] = open(source.file, "w") + files[source.file].write(source.str()) + for f in files: + files[f].close() + + def check_for_relations(self, sources_list): + """get all parent and child channels in the sources list""" + parents = [] + used_child_templates = {} + for source in sources_list: + # try to avoid checking uninterressting sources + if source.template is None: + continue + # set up a dict with all used child templates and corresponding + # source entries + if source.template.child: + key = source.template + if key not in used_child_templates: + used_child_templates[key] = [] + temp = used_child_templates[key] + temp.append(source) + else: + # store each source with children aka. a parent :) + if len(source.template.children) > 0: + parents.append(source) + #print self.used_child_templates + #print self.parents + return (parents, used_child_templates) + + class SourceEntryMatcher: - def __init__(self, matcherPath): - self.templates = [] - # Get the human readable channel and comp names from the channel .infos - spec_files = glob.glob("%s/*.info" % matcherPath) - for f in spec_files: - f = os.path.basename(f) - i = f.find(".info") - f = f[0:i] - dist = DistInfo(f,base_dir=matcherPath) - for template in dist.templates: - if template.match_uri != None: - self.templates.append(template) - return - - def match(self, source): - """Add a matching template to the source""" - _ = gettext.gettext - found = False - for template in self.templates: - if (re.search(template.match_uri, source.uri) and - re.match(template.match_name, source.dist)): - found = True - source.template = template - break - elif (template.is_mirror(source.uri) and - re.match(template.match_name, source.dist)): - found = True - source.template = template - break - return found + """ matcher class to make a source entry look nice + lots of predefined matchers to make it i18n/gettext friendly + """ + + def __init__(self, matcherPath): + self.templates = [] + # Get the human readable channel and comp names from the channel .infos + spec_files = glob.glob("%s/*.info" % matcherPath) + for f in spec_files: + f = os.path.basename(f) + i = f.find(".info") + f = f[0:i] + dist = DistInfo(f, base_dir=matcherPath) + for template in dist.templates: + if template.match_uri is not None: + self.templates.append(template) + return + + def match(self, source): + """Add a matching template to the source""" + _ = gettext.gettext + found = False + for template in self.templates: + if (re.search(template.match_uri, source.uri) and + re.match(template.match_name, source.dist)): + found = True + source.template = template + break + elif (template.is_mirror(source.uri) and + re.match(template.match_name, source.dist)): + found = True + source.template = template + break + return found # some simple tests if __name__ == "__main__": - apt_pkg.InitConfig() - sources = SourcesList() - - for entry in sources: - print entry.str() - #print entry.uri - - mirror = is_mirror("http://archive.ubuntu.com/ubuntu/", - "http://de.archive.ubuntu.com/ubuntu/") - print "is_mirror(): %s" % mirror - - print is_mirror("http://archive.ubuntu.com/ubuntu", - "http://de.archive.ubuntu.com/ubuntu/") - print is_mirror("http://archive.ubuntu.com/ubuntu/", - "http://de.archive.ubuntu.com/ubuntu") + apt_pkg.InitConfig() + sources = SourcesList() + + for entry in sources: + print entry.str() + #print entry.uri + + mirror = is_mirror("http://archive.ubuntu.com/ubuntu/", + "http://de.archive.ubuntu.com/ubuntu/") + print "is_mirror(): %s" % mirror + print is_mirror("http://archive.ubuntu.com/ubuntu", + "http://de.archive.ubuntu.com/ubuntu/") + print is_mirror("http://archive.ubuntu.com/ubuntu/", + "http://de.archive.ubuntu.com/ubuntu") |
