import warnings warnings.filterwarnings("ignore", "apt API not stable yet", FutureWarning) import apt import apt_pkg import gettext import sys, os, shutil import copy import time from apt.progress import InstallProgress from gettext import gettext as _ SYNAPTIC_PINFILE = "/var/lib/synaptic/preferences" # from update manager class MyCache(apt.Cache): def __init__(self, progress, rootdir=None): apt.Cache.__init__(self, progress, rootdir) self._initDepCache() assert self._depcache.BrokenCount == 0 and self._depcache.DelCount == 0 self.all_changes = {} def commit(self, fetchProgress=None, installProgress=None): if fetchProgress == None: fetchProgress = apt.progress.FetchProgress() if installProgress == None: installProgress = apt.progress.InstallProgress() pm = apt_pkg.GetPackageManager(self._depcache) fetcher = apt_pkg.GetAcquire(fetchProgress) while True: # fetch archives first res = self._fetchArchives(fetcher, pm) if (res == pm.ResultCompleted): break fetcher.Shutdown() return (res == pm.ResultCompleted) def _initDepCache(self): #apt_pkg.Config.Set("Debug::pkgPolicy","1") #self.depcache = apt_pkg.GetDepCache(self.cache) #self._depcache = apt_pkg.GetDepCache(self._cache) self._depcache.ReadPinFile() if os.path.exists(SYNAPTIC_PINFILE): self._depcache.ReadPinFile(SYNAPTIC_PINFILE) self._depcache.Init() def clear(self): self._initDepCache() @property def requiredDownload(self): """ get the size of the packages that are required to download """ pm = apt_pkg.GetPackageManager(self._depcache) fetcher = apt_pkg.GetAcquire() pm.GetArchives(fetcher, self._list, self._records) return fetcher.FetchNeeded @property def installCount(self): return self._depcache.InstCount def saveDistUpgrade(self): """ this functions mimics a upgrade but will never remove anything """ self._depcache.Upgrade(True) wouldDelete = self._depcache.DelCount if self._depcache.DelCount > 0: self.clear() assert self._depcache.BrokenCount == 0 and self._depcache.DelCount == 0 self._depcache.Upgrade() return wouldDelete def matchPackageOrigin(self, pkg, matcher): """ match 'pkg' origin against 'matcher', take versions between installedVersion and candidateVersion into account too Useful if installed pkg A v1.0 is available in both -updates (as v1.2) and -security (v1.1). we want to display it as a security update then """ inst_ver = pkg._pkg.CurrentVer cand_ver = self._depcache.GetCandidateVer(pkg._pkg) # init with empty match update_origin = matcher[(None,None)] for ver in pkg._pkg.VersionList: # discard is < than installed ver if (inst_ver and apt_pkg.VersionCompare(ver.VerStr, inst_ver.VerStr) <= 0): #print "skipping '%s' " % ver.VerStr continue # check if we have a match for(verFileIter,index) in ver.FileList: if matcher.has_key((verFileIter.Archive, verFileIter.Origin)): indexfile = pkg._list.FindIndex(verFileIter) if indexfile: # and indexfile.IsTrusted: match = matcher[verFileIter.Archive, verFileIter.Origin] if match.importance > update_origin.importance: update_origin = match return update_origin def get_changelog(self, name, lock): # don't touch the gui in this function, it needs to be thread-safe pkg = self[name] # get the src package name srcpkg = pkg.sourcePackageName # assume "main" section src_section = "main" # use the section of the candidate as a starting point section = pkg._depcache.GetCandidateVer(pkg._pkg).Section # get the source version, start with the binaries version binver = pkg.candidateVersion srcver = pkg.candidateVersion #print "bin: %s" % binver try: # try to get the source version of the pkg, this differs # for some (e.g. libnspr4 on ubuntu) # this feature only works if the correct deb-src are in the # sources.list # otherwise we fall back to the binary version number srcrecords = apt_pkg.GetPkgSrcRecords() srcrec = srcrecords.Lookup(srcpkg) if srcrec: srcver = srcrecords.Version if apt_pkg.VersionCompare(binver, srcver) > 0: srcver = binver #print "srcver: %s" % srcver section = srcrecords.Section #print "srcsect: %s" % section else: # fail into the error handler raise SystemError except SystemError, e: srcver = binver l = section.split("/") if len(l) > 1: src_section = l[0] # lib is handled special prefix = srcpkg[0] if srcpkg.startswith("lib"): prefix = "lib" + srcpkg[3] # stip epoch l = string.split(srcver,":") if len(l) > 1: srcver = "".join(l[1:]) try: uri = CHANGELOGS_URI % (src_section,prefix,srcpkg,srcpkg, srcver) # print "Trying: %s " % uri changelog = urllib2.urlopen(uri) #print changelog.read() # do only get the lines that are new alllines = "" regexp = "^%s \((.*)\)(.*)$" % (re.escape(srcpkg)) i=0 while True: line = changelog.readline() if line == "": break match = re.match(regexp,line) if match: # strip epoch from installed version # and from changelog too installed = pkg.installedVersion if installed and ":" in installed: installed = installed.split(":",1)[1] changelogver = match.group(1) if changelogver and ":" in changelogver: changelogver = changelogver.split(":",1)[1] if installed and \ apt_pkg.VersionCompare(changelogver,installed)<=0: break # EOF (shouldn't really happen) alllines = alllines + line # Print an error if we failed to extract a changelog if len(alllines) == 0: alllines = _("The list of changes is not available") # only write if we where not canceld if lock.locked(): self.all_changes[name] = [alllines, srcpkg] except urllib2.HTTPError: if lock.locked(): self.all_changes[name] = [_("The list of changes is not " "available yet.\nPlease try again " "later."), srcpkg] except IOError: if lock.locked(): self.all_changes[name] = [_("Failed to download the list " "of changes. \nPlease " "check your Internet " "connection."), srcpkg] if lock.locked(): lock.release() class UpdateList: class UpdateOrigin: def __init__(self, desc, importance): self.packages = [] self.importance = importance self.description = desc def __init__(self): # a map of packages under their origin pipe = os.popen("lsb_release -c -s") dist = pipe.read().strip() del pipe self.distUpgradeWouldDelete = 0 self.pkgs = {} self.num_updates = 0 self.matcher = self.initMatcher(dist) def initMatcher(self, dist): # (origin, archive, description, importance) matcher_templates = [ ("%s-security" % dist, "Ubuntu", _("Important security updates"),10), ("%s-updates" % dist, "Ubuntu", _("Recommended updates"), 9), ("%s-proposed" % dist, "Ubuntu", _("Proposed updates"), 8), ("%s-backports" % dist, "Ubuntu", _("Backports"), 7), (dist, "Ubuntu", _("Distribution updates"), 6) ] matcher = {} for (origin, archive, desc, importance) in matcher_templates: matcher[(origin, archive)] = self.UpdateOrigin(desc, importance) matcher[(None,None)] = self.UpdateOrigin(_("Other updates"), -1) return matcher def update(self, cache): self.held_back = [] # do the upgrade self.distUpgradeWouldDelete = cache.saveDistUpgrade() # sort by origin for pkg in cache: if pkg.isUpgradable or pkg.markedInstall: if pkg.candidateOrigin == None: # can happen for e.g. loged packages # FIXME: do something more sensible here (but what?) print "WARNING: upgradable but no canidateOrigin?!?: ", pkg.name continue # check where the package belongs origin_node = cache.matchPackageOrigin(pkg, self.matcher) if not self.pkgs.has_key(origin_node): self.pkgs[origin_node] = [] self.pkgs[origin_node].append(pkg) self.num_updates = self.num_updates + 1 if pkg.isUpgradable and not (pkg.markedUpgrade or pkg.markedInstall): self.held_back.append(pkg.name) for l in self.pkgs.keys(): self.pkgs[l].sort(lambda x,y: cmp(x.name,y.name)) self.keepcount = cache._depcache.KeepCount # my code class my_TextFetchProgress(apt.progress.TextFetchProgress): def __init__(self): apt.progress.TextFetchProgress.__init__(self) def done(self): self def update(self, percent): self def error(self, errorstr): print "got dpkg error: '%s'" % errorstr def list_update(): c = apt.Cache() res = c.update(apt.progress.TextFetchProgress()) def select_upgrade(): progress = my_TextFetchProgress() cache = MyCache(progress) list = UpdateList() list.update(cache) cache.commit() def dl_package(package_name): progress = my_TextFetchProgress() cache = MyCache(progress) cache[package_name].markInstall() cache.commit() def create_aptbackup(target_dir): if not os.path.exists(target_dir): os.mkdir(target_dir) os.mkdir(target_dir + "/download") os.mkdir(target_dir + "/download/partial") # /etc/apt/ os.mkdir(target_dir + "/etc") shutil.copytree("/etc/apt/sources.list.d", target_dir + "/etc/sources.list.d") shutil.copy("/etc/apt/sources.list", target_dir + "/etc/") # /var/lib/dpkg/status os.mkdir(target_dir + "/dpkg") shutil.copy("/var/lib/dpkg/status", target_dir + "/dpkg/status") # /var/lib/apt try: shutil.copytree("/var/lib/apt", target_dir + "/libapt") except shutil.Error: shutil.Error def set_aptbackup(target_dir): apt_pkg.Config.Set("Dir::Etc", target_dir + "/etc/") apt_pkg.Config.Set("Dir::State::status", target_dir + "/dpkg/status") # target apt_pkg.Config.Set("Dir::State", target_dir + "/libapt") # target or host? target i think apt_pkg.Config.Set("Dir::Cache::archives", target_dir + "/download") # packages dl-ed to here set_aptbackup("/media/stuffs/dev/python/apt_backup") dl_package("abiword")