#!/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()