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