tools/examples/svnshell.py

#!/usr/bin/env python
#
# svnshell.py : a Python-based shell interface for cruising 'round in
#               the filesystem.
#
######################################################################
#    Licensed to the Apache Software Foundation (ASF) under one
#    or more contributor license agreements.  See the NOTICE file
#    distributed with this work for additional information
#    regarding copyright ownership.  The ASF licenses this file
#    to you under the Apache License, Version 2.0 (the
#    "License"); you may not use this file except in compliance
#    with the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing,
#    software distributed under the License is distributed on an
#    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#    KIND, either express or implied.  See the License for the
#    specific language governing permissions and limitations
#    under the License.
######################################################################
#

import sys
import time
import re
from cmd import Cmd
from random import randint
from svn import fs, core, repos


class SVNShell(Cmd):
  def __init__(self, path):
    """initialize an SVNShell object"""
    Cmd.__init__(self)
    path = core.svn_path_canonicalize(path)
    self.fs_ptr = repos.fs(repos.open(path))
    self.is_rev = 1
    self.rev = fs.youngest_rev(self.fs_ptr)
    self.txn = None
    self.root = fs.revision_root(self.fs_ptr, self.rev)
    self.path = "/"
    self._setup_prompt()
    self.cmdloop()

  def precmd(self, line):
    if line == "EOF":
      # Ctrl-D is a command without a newline.  Print a newline, so the next
      # shell prompt is not on the same line as the last svnshell prompt.
      print("")
      return "exit"
    return line

  def postcmd(self, stop, line):
    self._setup_prompt()

  _errors = ["Huh?",
             "Whatchoo talkin' 'bout, Willis?",
             "Say what?",
             "Nope.  Not gonna do it.",
             "Ehh...I don't think so, chief."]

  def default(self, line):
    print(self._errors[randint(0, len(self._errors) - 1)])

  def do_cat(self, arg):
    """dump the contents of a file"""
    if not len(arg):
      print("You must supply a file path.")
      return
    catpath = self._parse_path(arg)
    kind = fs.check_path(self.root, catpath)
    if kind == core.svn_node_none:
      print("Path '%s' does not exist." % catpath)
      return
    if kind == core.svn_node_dir:
      print("Path '%s' is not a file." % catpath)
      return
    ### be nice to get some paging in here.
    stream = fs.file_contents(self.root, catpath)
    while True:
      data = core.svn_stream_read(stream, core.SVN_STREAM_CHUNK_SIZE)
      sys.stdout.write(data)
      if len(data) < core.SVN_STREAM_CHUNK_SIZE:
        break

  def do_cd(self, arg):
    """change directory"""
    newpath = self._parse_path(arg)

    # make sure that path actually exists in the filesystem as a directory
    kind = fs.check_path(self.root, newpath)
    if kind != core.svn_node_dir:
      print("Path '%s' is not a valid filesystem directory." % newpath)
      return
    self.path = newpath

  def do_ls(self, arg):
    """list the contents of the current directory or provided path"""
    parent = self.path
    if not len(arg):
      # no arg -- show a listing for the current directory.
      entries = fs.dir_entries(self.root, self.path)
    else:
      # arg?  show a listing of that path.
      newpath = self._parse_path(arg)
      kind = fs.check_path(self.root, newpath)
      if kind == core.svn_node_dir:
        parent = newpath
        entries = fs.dir_entries(self.root, parent)
      elif kind == core.svn_node_file:
        parts = self._path_to_parts(newpath)
        name = parts.pop(-1)
        parent = self._parts_to_path(parts)
        print(parent + ':' + name)
        tmpentries = fs.dir_entries(self.root, parent)
        if not tmpentries.get(name, None):
          return
        entries = {}
        entries[name] = tmpentries[name]
      else:
        print("Path '%s' not found." % newpath)
        return

    keys = sorted(entries.keys())

    print("   REV   AUTHOR  NODE-REV-ID     SIZE         DATE NAME")
    print("----------------------------------------------------------------------------")

    for entry in keys:
      fullpath = parent + '/' + entry
      size = ''
      is_dir = fs.is_dir(self.root, fullpath)
      if is_dir:
        name = entry + '/'
      else:
        size = str(fs.file_length(self.root, fullpath))
        name = entry
      node_id = fs.unparse_id(entries[entry].id)
      created_rev = fs.node_created_rev(self.root, fullpath)
      author = fs.revision_prop(self.fs_ptr, created_rev,
                                core.SVN_PROP_REVISION_AUTHOR)
      if not author:
        author = ""
      date = fs.revision_prop(self.fs_ptr, created_rev,
                              core.SVN_PROP_REVISION_DATE)
      if not date:
        date = ""
      else:
        date = self._format_date(date)

      print("%6s %8s %12s %8s %12s %s" % (created_rev, author[:8],
                                          node_id, size, date, name))

  def do_lstxns(self, arg):
    """list the transactions available for browsing"""
    txns = sorted(fs.list_transactions(self.fs_ptr))
    counter = 0
    for txn in txns:
      counter = counter + 1
      sys.stdout.write("%8s   " % txn)
      if counter == 6:
        print("")
        counter = 0
    print("")

  def do_pcat(self, arg):
    """list the properties of a path"""
    catpath = self.path
    if len(arg):
      catpath = self._parse_path(arg)
    kind = fs.check_path(self.root, catpath)
    if kind == core.svn_node_none:
      print("Path '%s' does not exist." % catpath)
      return
    plist = fs.node_proplist(self.root, catpath)
    if not plist:
      return
    for pkey, pval in plist.items():
      print('K ' + str(len(pkey)))
      print(pkey)
      print('P ' + str(len(pval)))
      print(pval)
    print('PROPS-END')

  def do_setrev(self, arg):
    """set the current revision to view"""
    try:
      if arg.lower() == 'head':
        rev = fs.youngest_rev(self.fs_ptr)
      else:
        rev = int(arg)
      newroot = fs.revision_root(self.fs_ptr, rev)
    except:
      print("Error setting the revision to '" + arg + "'.")
      return
    fs.close_root(self.root)
    self.root = newroot
    self.rev = rev
    self.is_rev = 1
    self._do_path_landing()

  def do_settxn(self, arg):
    """set the current transaction to view"""
    try:
      txnobj = fs.open_txn(self.fs_ptr, arg)
      newroot = fs.txn_root(txnobj)
    except:
      print("Error setting the transaction to '" + arg + "'.")
      return
    fs.close_root(self.root)
    self.root = newroot
    self.txn = arg
    self.is_rev = 0
    self._do_path_landing()

  def do_youngest(self, arg):
    """list the youngest revision available for browsing"""
    rev = fs.youngest_rev(self.fs_ptr)
    print(rev)

  def do_exit(self, arg):
    sys.exit(0)

  def _path_to_parts(self, path):
    return [_f for _f in path.split('/') if _f]

  def _parts_to_path(self, parts):
    return '/' + '/'.join(parts)

  def _parse_path(self, path):
    # cleanup leading, trailing, and duplicate '/' characters
    newpath = self._parts_to_path(self._path_to_parts(path))

    # if PATH is absolute, use it, else append it to the existing path.
    if path.startswith('/') or self.path == '/':
      newpath = '/' + newpath
    else:
      newpath = self.path + '/' + newpath

    # cleanup '.' and '..'
    parts = self._path_to_parts(newpath)
    finalparts = []
    for part in parts:
      if part == '.':
        pass
      elif part == '..':
        if len(finalparts) != 0:
          finalparts.pop(-1)
      else:
        finalparts.append(part)

    # finally, return the calculated path
    return self._parts_to_path(finalparts)

  def _format_date(self, date):
    date = core.svn_time_from_cstring(date)
    date = time.asctime(time.localtime(date / 1000000))
    return date[4:-8]

  def _do_path_landing(self):
    """try to land on self.path as a directory in root, failing up to '/'"""
    not_found = 1
    newpath = self.path
    while not_found:
      kind = fs.check_path(self.root, newpath)
      if kind == core.svn_node_dir:
        not_found = 0
      else:
        parts = self._path_to_parts(newpath)
        parts.pop(-1)
        newpath = self._parts_to_path(parts)
    self.path = newpath

  def _setup_prompt(self):
    """present the prompt and handle the user's input"""
    if self.is_rev:
      self.prompt = "<rev: " + str(self.rev)
    else:
      self.prompt = "<txn: " + self.txn
    self.prompt += " " + self.path + ">$ "

  def _complete(self, text, line, begidx, endidx, limit_node_kind=None):
    """Generic tab completer.  Takes the 4 standard parameters passed to a
    cmd.Cmd completer function, plus LIMIT_NODE_KIND, which should be a
    svn.core.svn_node_foo constant to restrict the returned completions to, or
    None for no limit.  Catches and displays exceptions, because otherwise
    they are silently ignored - which is quite frustrating when debugging!"""
    try:
      args = line.split()
      if len(args) > 1:
        arg = args[1]
      else:
        arg = ""
      dirs = arg.split('/')
      user_elem = dirs[-1]
      user_dir = "/".join(dirs[:-1] + [''])

      canon_dir = self._parse_path(user_dir)

      entries = fs.dir_entries(self.root, canon_dir)
      acceptable_completions = []
      for name, dirent_t in entries.items():
        if not name.startswith(user_elem):
          continue
        if limit_node_kind and dirent_t.kind != limit_node_kind:
          continue
        if dirent_t.kind == core.svn_node_dir:
          name += '/'
        acceptable_completions.append(name)
      if limit_node_kind == core.svn_node_dir or not limit_node_kind:
        if user_elem in ('.', '..'):
          for extraname in ('.', '..'):
            if extraname.startswith(user_elem):
              acceptable_completions.append(extraname + '/')
      return acceptable_completions
    except:
      ei = sys.exc_info()
      sys.stderr.write("EXCEPTION WHILST COMPLETING\n")
      import traceback
      traceback.print_tb(ei[2])
      sys.stderr.write("%s: %s\n" % (ei[0], ei[1]))
      raise

  def complete_cd(self, text, line, begidx, endidx):
    return self._complete(text, line, begidx, endidx, core.svn_node_dir)

  def complete_cat(self, text, line, begidx, endidx):
    return self._complete(text, line, begidx, endidx, core.svn_node_file)

  def complete_ls(self, text, line, begidx, endidx):
    return self._complete(text, line, begidx, endidx)

  def complete_pcat(self, text, line, begidx, endidx):
    return self._complete(text, line, begidx, endidx)


def _basename(path):
  "Return the basename for a '/'-separated path."
  idx = path.rfind('/')
  if idx == -1:
    return path
  return path[idx+1:]


def usage(exit):
  if exit:
    output = sys.stderr
  else:
    output = sys.stdout
  output.write(
    "usage: %s REPOS_PATH\n"
    "\n"
    "Once the program has started, type 'help' at the prompt for hints on\n"
    "using the shell.\n" % sys.argv[0])
  sys.exit(exit)

def main():
  if len(sys.argv) != 2:
    usage(1)

  SVNShell(sys.argv[1])

if __name__ == '__main__':
  main()