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