import pysvn.ra import posixpath from UserDict import DictMixin class Repository(object): def __init__(self, url, verbose=False): self.url = url self.ra = pysvn.ra.connect(url) self.headrev = self.ra.get_latest_rev() self.verbose = verbose self.revisions = [] prev = None for rev in range(self.headrev+1): dirnode = Dir(self, '', rev) dirnode.prev = prev self.revisions.append(dirnode) prev = dirnode def getpath(self, path, rev): result = self.revisions[rev] assert result.rev == rev return result.getpath(path) def getrev(self): return self.headrev class Node(object): def __init__(self, repo, path, rev): self.repo = repo self.path = path self.rev = rev def __getattr__(self, name): if name == 'prev': self._force_prev() return self.prev raise AttributeError(name) def _force_prev(self): if self.repo.verbose: print '[prev /%s@%d]' % (self.path, self.rev) ra = self.repo.ra locations = ra.get_locations(self.path, self.rev, [self.rev-1]) if locations: path = ra.relpath(ra.join(locations[self.rev-1])) self.prev = self.repo.getpath(path, self.rev-1) else: self.prev = None class Dir(Node): def __init__(self, *args, **kwds): Node.__init__(self, *args, **kwds) self.entries = DirEntries(self) def getpath(self, path): result = self for p in path.split('/'): if p: result = result.entries[p] return result def getentries(self): result = self.entries.items() result.sort() return result class File(Node): pass # ____________________________________________________________ class DirEntries(DictMixin): _ra_entries = None def __init__(self, dir): self.dir = dir self.contents = {} def get_ra_entries(self): if self._ra_entries is None: ra = self.dir.repo.ra if self.dir.repo.verbose: print '[list /%s@%d]' % (self.dir.path, self.dir.rev) _, props, entries = ra.get_dir(self.dir.path, self.dir.rev) assert self.dir.rev == int(props['svn:entry:committed-rev']) self._ra_entries = entries return self._ra_entries ## def force(self): ## if not self._complete: ## for name in self.get_ra_entries(): ## self.force_entry(name) ## self._complete = True def force_entry(self, name): if name not in self.contents: stats = self.get_ra_entries()[name] rev = int(stats['svn:entry:committed-rev']) path = posixpath.join(self.dir.path, name) ra = self.dir.repo.ra if rev < self.dir.rev: if self.dir.repo.verbose: print '[path /%s@%d -> %d]' % (path, self.dir.rev, rev) locations = ra.get_locations(path, self.dir.rev, [rev]) path = ra.relpath(ra.join(locations[rev])) path = posixpath.dirname(path) dir_with_this_change = self.dir.repo.getpath(path, rev) else: dir_with_this_change = self.dir assert rev == dir_with_this_change.rev node = dir_with_this_change.entries.ensure_node(name, stats) self.contents[name] = node def ensure_node(self, name, stats): try: node = self.contents[name] except KeyError: dir = self.dir node = newnode(dir.repo, posixpath.join(dir.path, name), dir.rev, stats) self.contents[name] = node return node def copy(self): return dict(self.items()) def __getitem__(self, key): self.force_entry(key) return self.contents[key] def keys(self): return self.get_ra_entries().keys() def newnode(repo, path, rev, stats): kind = stats['svn:entry:kind'] return KIND2CLS[kind](repo, path, rev) KIND2CLS = {'file': File, 'dir': Dir}