tools/examples/svnshell.py

   1 #!/usr/bin/env python
   2 #
   3 # svnshell.py : a Python-based shell interface for cruising 'round in
   4 #               the filesystem.
   5 #
   6 ######################################################################
   7 #
   8 # Copyright (c) 2000-2004 CollabNet.  All rights reserved.
   9 #
  10 # This software is licensed as described in the file COPYING, which
  11 # you should have received as part of this distribution.  The terms
  12 # are also available at http://subversion.tigris.org/license-1.html.
  13 # If newer versions of this license are posted there, you may use a
  14 # newer version instead, at your option.
  15 #
  16 ######################################################################
  17 #
  18
  19 import sys
  20 import string
  21 import time
  22 import re
  23 from cmd import Cmd
  24 from random import randint
  25 from svn import fs, core, repos
  26
  27
  28 class SVNShell(Cmd):
  29   def __init__(self, path):
  30     """initialize an SVNShell object"""
  31     Cmd.__init__(self)
  32     path = core.svn_path_canonicalize(path)
  33     self.fs_ptr = repos.fs(repos.open(path))
  34     self.is_rev = 1
  35     self.rev = fs.youngest_rev(self.fs_ptr)
  36     self.txn = None
  37     self.root = fs.revision_root(self.fs_ptr, self.rev)
  38     self.path = "/"
  39     self._setup_prompt()
  40     self.cmdloop()
  41
  42   def precmd(self, line):
  43     if line == "EOF":
  44       # Ctrl-D is a command without a newline.  Print a newline, so the next
  45       # shell prompt is not on the same line as the last svnshell prompt.
  46       print
  47       return "exit"
  48     return line
  49
  50   def postcmd(self, stop, line):
  51     self._setup_prompt()
  52
  53   _errors = ["Huh?",
  54              "Whatchoo talkin' 'bout, Willis?",
  55              "Say what?",
  56              "Nope.  Not gonna do it.",
  57              "Ehh...I don't think so, chief."]
  58
  59   def default(self, line):
  60     print self._errors[randint(0, len(self._errors) - 1)]
  61
  62   def do_cat(self, arg):
  63     """dump the contents of a file"""
  64     if not len(arg):
  65       print "You must supply a file path."
  66       return
  67     catpath = self._parse_path(arg)
  68     kind = fs.check_path(self.root, catpath)
  69     if kind == core.svn_node_none:
  70       print "Path '%s' does not exist." % catpath
  71       return
  72     if kind == core.svn_node_dir:
  73       print "Path '%s' is not a file." % catpath
  74       return
  75     ### be nice to get some paging in here.
  76     stream = fs.file_contents(self.root, catpath)
  77     while 1:
  78       data = core.svn_stream_read(stream, core.SVN_STREAM_CHUNK_SIZE)
  79       sys.stdout.write(data)
  80       if len(data) < core.SVN_STREAM_CHUNK_SIZE:
  81         break
  82
  83   def do_cd(self, arg):
  84     """change directory"""
  85     newpath = self._parse_path(arg)
  86
  87     # make sure that path actually exists in the filesystem as a directory
  88     kind = fs.check_path(self.root, newpath)
  89     if kind != core.svn_node_dir:
  90       print "Path '%s' is not a valid filesystem directory." % newpath
  91       return
  92     self.path = newpath
  93
  94   def do_ls(self, arg):
  95     """list the contents of the current directory or provided path"""
  96     parent = self.path
  97     if not len(arg):
  98       # no arg -- show a listing for the current directory.
  99       entries = fs.dir_entries(self.root, self.path)
 100     else:
 101       # arg?  show a listing of that path.
 102       newpath = self._parse_path(arg)
 103       kind = fs.check_path(self.root, newpath)
 104       if kind == core.svn_node_dir:
 105         parent = newpath
 106         entries = fs.dir_entries(self.root, parent)
 107       elif kind == core.svn_node_file:
 108         parts = self._path_to_parts(newpath)
 109         name = parts.pop(-1)
 110         parent = self._parts_to_path(parts)
 111         print parent + ':' + name
 112         tmpentries = fs.dir_entries(self.root, parent)
 113         if not tmpentries.get(name, None):
 114           return
 115         entries = {}
 116         entries[name] = tmpentries[name]
 117       else:
 118         print "Path '%s' not found." % newpath
 119         return
 120
 121     keys = entries.keys()
 122     keys.sort()
 123
 124     print "   REV   AUTHOR  NODE-REV-ID     SIZE         DATE NAME"
 125     print "----------------------------------------------------------------------------"
 126
 127     for entry in keys:
 128       fullpath = parent + '/' + entry
 129       size = ''
 130       is_dir = fs.is_dir(self.root, fullpath)
 131       if is_dir:
 132         name = entry + '/'
 133       else:
 134         size = str(fs.file_length(self.root, fullpath))
 135         name = entry
 136       node_id = fs.unparse_id(entries[entry].id)
 137       created_rev = fs.node_created_rev(self.root, fullpath)
 138       author = fs.revision_prop(self.fs_ptr, created_rev,
 139                                 core.SVN_PROP_REVISION_AUTHOR)
 140       if not author:
 141         author = ""
 142       date = fs.revision_prop(self.fs_ptr, created_rev,
 143                               core.SVN_PROP_REVISION_DATE)
 144       if not date:
 145         date = ""
 146       else:
 147         date = self._format_date(date)
 148
 149       print "%6s %8s %12s %8s %12s %s" % (created_rev, author[:8],
 150                                           node_id, size, date, name)
 151
 152   def do_lstxns(self, arg):
 153     """list the transactions available for browsing"""
 154     txns = fs.list_transactions(self.fs_ptr)
 155     txns.sort()
 156     counter = 0
 157     for txn in txns:
 158       counter = counter + 1
 159       print "%8s  " % txn,
 160       if counter == 6:
 161         print ""
 162         counter = 0
 163     print ""
 164
 165   def do_pcat(self, arg):
 166     """list the properties of a path"""
 167     catpath = self.path
 168     if len(arg):
 169       catpath = self._parse_path(arg)
 170     kind = fs.check_path(self.root, catpath)
 171     if kind == core.svn_node_none:
 172       print "Path '%s' does not exist." % catpath
 173       return
 174     plist = fs.node_proplist(self.root, catpath)
 175     if not plist:
 176       return
 177     for pkey, pval in plist.items():
 178       print 'K ' + str(len(pkey))
 179       print pkey
 180       print 'P ' + str(len(pval))
 181       print pval
 182     print 'PROPS-END'
 183
 184   def do_setrev(self, arg):
 185     """set the current revision to view"""
 186     try:
 187       if arg.lower() == 'head':
 188         rev = fs.youngest_rev(self.fs_ptr)
 189       else:
 190         rev = int(arg)
 191       newroot = fs.revision_root(self.fs_ptr, rev)
 192     except:
 193       print "Error setting the revision to '" + arg + "'."
 194       return
 195     fs.close_root(self.root)
 196     self.root = newroot
 197     self.rev = rev
 198     self.is_rev = 1
 199     self._do_path_landing()
 200
 201   def do_settxn(self, arg):
 202     """set the current transaction to view"""
 203     try:
 204       txnobj = fs.open_txn(self.fs_ptr, arg)
 205       newroot = fs.txn_root(txnobj)
 206     except:
 207       print "Error setting the transaction to '" + arg + "'."
 208       return
 209     fs.close_root(self.root)
 210     self.root = newroot
 211     self.txn = arg
 212     self.is_rev = 0
 213     self._do_path_landing()
 214
 215   def do_youngest(self, arg):
 216     """list the youngest revision available for browsing"""
 217     rev = fs.youngest_rev(self.fs_ptr)
 218     print rev
 219
 220   def do_exit(self, arg):
 221     sys.exit(0)
 222
 223   def _path_to_parts(self, path):
 224     return filter(None, string.split(path, '/'))
 225
 226   def _parts_to_path(self, parts):
 227     return '/' + string.join(parts, '/')
 228
 229   def _parse_path(self, path):
 230     # cleanup leading, trailing, and duplicate '/' characters
 231     newpath = self._parts_to_path(self._path_to_parts(path))
 232
 233     # if PATH is absolute, use it, else append it to the existing path.
 234     if path.startswith('/') or self.path == '/':
 235       newpath = '/' + newpath
 236     else:
 237       newpath = self.path + '/' + newpath
 238
 239     # cleanup '.' and '..'
 240     parts = self._path_to_parts(newpath)
 241     finalparts = []
 242     for part in parts:
 243       if part == '.':
 244         pass
 245       elif part == '..':
 246         if len(finalparts) != 0:
 247           finalparts.pop(-1)
 248       else:
 249         finalparts.append(part)
 250
 251     # finally, return the calculated path
 252     return self._parts_to_path(finalparts)
 253
 254   def _format_date(self, date):
 255     date = core.svn_time_from_cstring(date)
 256     date = time.asctime(time.localtime(date / 1000000))
 257     return date[4:-8]
 258
 259   def _do_path_landing(self):
 260     """try to land on self.path as a directory in root, failing up to '/'"""
 261     not_found = 1
 262     newpath = self.path
 263     while not_found:
 264       kind = fs.check_path(self.root, newpath)
 265       if kind == core.svn_node_dir:
 266         not_found = 0
 267       else:
 268         parts = self._path_to_parts(newpath)
 269         parts.pop(-1)
 270         newpath = self._parts_to_path(parts)
 271     self.path = newpath
 272
 273   def _setup_prompt(self):
 274     """present the prompt and handle the user's input"""
 275     if self.is_rev:
 276       self.prompt = "<rev: " + str(self.rev)
 277     else:
 278       self.prompt = "<txn: " + self.txn
 279     self.prompt += " " + self.path + ">$ "
 280
 281   def _complete(self, text, line, begidx, endidx, limit_node_kind=None):
 282     """Generic tab completer.  Takes the 4 standard parameters passed to a
 283     cmd.Cmd completer function, plus LIMIT_NODE_KIND, which should be a
 284     svn.core.svn_node_foo constant to restrict the returned completions to, or
 285     None for no limit.  Catches and displays exceptions, because otherwise
 286     they are silently ignored - which is quite frustrating when debugging!"""
 287     try:
 288       args = line.split()
 289       if len(args) > 1:
 290         arg = args[1]
 291       else:
 292         arg = ""
 293       dirs = arg.split('/')
 294       user_elem = dirs[-1]
 295       user_dir = "/".join(dirs[:-1] + [''])
 296
 297       canon_dir = self._parse_path(user_dir)
 298
 299       entries = fs.dir_entries(self.root, canon_dir)
 300       acceptable_completions = []
 301       for name, dirent_t in entries.items():
 302         if not name.startswith(user_elem):
 303           continue
 304         if limit_node_kind and dirent_t.kind != limit_node_kind:
 305           continue
 306         if dirent_t.kind == core.svn_node_dir:
 307           name += '/'
 308         acceptable_completions.append(name)
 309       if limit_node_kind == core.svn_node_dir or not limit_node_kind:
 310         if user_elem in ('.', '..'):
 311           for extraname in ('.', '..'):
 312             if extraname.startswith(user_elem):
 313               acceptable_completions.append(extraname + '/')
 314       return acceptable_completions
 315     except:
 316       ei = sys.exc_info()
 317       sys.stderr.write("EXCEPTION WHILST COMPLETING\n")
 318       import traceback
 319       traceback.print_tb(ei[2])
 320       sys.stderr.write("%s: %s\n" % (ei[0], ei[1]))
 321       raise
 322
 323   def complete_cd(self, text, line, begidx, endidx):
 324     return self._complete(text, line, begidx, endidx, core.svn_node_dir)
 325
 326   def complete_cat(self, text, line, begidx, endidx):
 327     return self._complete(text, line, begidx, endidx, core.svn_node_file)
 328
 329   def complete_ls(self, text, line, begidx, endidx):
 330     return self._complete(text, line, begidx, endidx)
 331
 332   def complete_pcat(self, text, line, begidx, endidx):
 333     return self._complete(text, line, begidx, endidx)
 334
 335
 336 def _basename(path):
 337   "Return the basename for a '/'-separated path."
 338   idx = string.rfind(path, '/')
 339   if idx == -1:
 340     return path
 341   return path[idx+1:]
 342
 343
 344 def usage(exit):
 345   if exit:
 346     output = sys.stderr
 347   else:
 348     output = sys.stdout
 349   output.write(
 350     "usage: %s REPOS_PATH\n"
 351     "\n"
 352     "Once the program has started, type 'help' at the prompt for hints on\n"
 353     "using the shell.\n" % sys.argv[0])
 354   sys.exit(exit)
 355
 356 def main():
 357   if len(sys.argv) != 2:
 358     usage(1)
 359
 360   SVNShell(sys.argv[1])
 361
 362 if __name__ == '__main__':
 363   main()