bzrtools 2.6.0-1 / usr / lib / python2.7 / dist-packages / bzrlib / plugins / bzrtools / shell.py

# Copyright (C) 2004, 2005 Aaron Bentley
# <aaron@aaronbentley.com>
#
#    This program is free software; you can redistribute it and/or modify
#    it under the terms of the GNU General Public License as published by
#    the Free Software Foundation; either version 2 of the License, or
#    (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU General Public License for more details.
#
#    You should have received a copy of the GNU General Public License
#    along with this program; if not, write to the Free Software
#    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

import cmd
from itertools import chain
import os
try:
    import readline
except ImportError:
    _has_readline = False
else:
    _has_readline = True
import shlex
import stat
import string
import sys

from bzrlib import osutils, trace
from bzrlib.branch import Branch
from bzrlib.config import config_dir, ensure_config_dir_exists
from bzrlib.commands import get_cmd_object, all_command_names, get_alias
from bzrlib.errors import BzrError
from bzrlib.workingtree import WorkingTree

import terminal


SHELL_BLACKLIST = set(['rm', 'ls'])
COMPLETION_BLACKLIST = set(['shell'])


class BlackListedCommand(BzrError):
    def __init__(self, command):
        BzrError.__init__(self, "The command %s is blacklisted for shell use" %
                          command)


class CompletionContext(object):
    def __init__(self, text, command=None, prev_opt=None, arg_pos=None):
        self.text = text
        self.command = command
        self.prev_opt = prev_opt
        self.arg_pos = None

    def get_completions(self):
        try:
            return self.get_completions_or_raise()
        except Exception, e:
            print e, type(e)
            return []

    def get_option_completions(self):
        try:
            command_obj = get_cmd_object(self.command)
        except BzrError:
            return []
        opts = [o+" " for o in iter_opt_completions(command_obj)]
        return list(filter_completions(opts, self.text))

    def get_completions_or_raise(self):
        if self.command is None:
            if '/' in self.text:
                iter = iter_executables(self.text)
            else:
                iter = (c+" " for c in iter_command_names() if
                        c not in COMPLETION_BLACKLIST)
            return list(filter_completions(iter, self.text))
        if self.prev_opt is None:
            completions = self.get_option_completions()
            if self.command == "cd":
                iter = iter_dir_completions(self.text)
                completions.extend(list(filter_completions(iter, self.text)))
            else:
                iter = iter_file_completions(self.text)
                completions.extend(filter_completions(iter, self.text))
            return completions


class PromptCmd(cmd.Cmd):

    def __init__(self):
        cmd.Cmd.__init__(self)
        self.prompt = "bzr> "
        try:
            self.tree = WorkingTree.open_containing('.')[0]
        except:
            self.tree = None
        self.set_title()
        self.set_prompt()
        self.identchars += '-'
        ensure_config_dir_exists()
        self.history_file = osutils.pathjoin(config_dir(), 'shell-history')
        whitespace = ''.join(c for c in string.whitespace if c < chr(127))
        if _has_readline:
            readline.set_completer_delims(whitespace)
            if os.access(self.history_file, os.R_OK) and \
                os.path.isfile(self.history_file):
                readline.read_history_file(self.history_file)
        self.cwd = os.getcwd()

    def write_history(self):
        if _has_readline:
            readline.write_history_file(self.history_file)

    def do_quit(self, args):
        self.write_history()
        raise StopIteration

    def do_exit(self, args):
        self.do_quit(args)

    def do_EOF(self, args):
        print
        self.do_quit(args)

    def postcmd(self, line, bar):
        self.set_title()
        self.set_prompt()

    def set_prompt(self):
        if self.tree is not None:
            try:
                prompt_data = (self.tree.branch.nick, self.tree.branch.revno(),
                               self.tree.relpath('.'))
                prompt = " %s:%d/%s" % prompt_data
            except:
                prompt = ""
        else:
            prompt = ""
        self.prompt = "bzr%s> " % prompt

    def set_title(self, command=None):
        try:
            b = Branch.open_containing('.')[0]
            version = "%s:%d" % (b.nick, b.revno())
        except:
            version = "[no version]"
        if command is None:
            command = ""
        sys.stdout.write(terminal.term_title("bzr %s %s" % (command, version)))

    def do_cd(self, line):
        if line == "":
            line = "~"
        line = os.path.expanduser(line)
        if os.path.isabs(line):
            newcwd = line
        else:
            newcwd = self.cwd+'/'+line
        newcwd = os.path.normpath(newcwd)
        try:
            os.chdir(newcwd)
            self.cwd = newcwd
        except Exception, e:
            print e
        try:
            self.tree = WorkingTree.open_containing(".")[0]
        except:
            self.tree = None

    def do_help(self, line):
        self.default("help "+line)

    def default(self, line):
        try:
            args = shlex.split(line)
        except ValueError, e:
            print 'Parse error:', e
            return

        alias_args = get_alias(args[0])
        if alias_args is not None:
            args[0] = alias_args.pop(0)

        commandname = args.pop(0)
        for char in ('|', '<', '>'):
            commandname = commandname.split(char)[0]
        if commandname[-1] in ('|', '<', '>'):
            commandname = commandname[:-1]
        try:
            if commandname in SHELL_BLACKLIST:
                raise BlackListedCommand(commandname)
            cmd_obj = get_cmd_object(commandname)
        except (BlackListedCommand, BzrError):
            return os.system(line)

        try:
            is_qbzr = cmd_obj.__module__.startswith('bzrlib.plugins.qbzr.')
            if too_complicated(line) or is_qbzr:
                return os.system("bzr "+line)
            else:
                return (cmd_obj.run_argv_aliases(args, alias_args) or 0)
        except BzrError, e:
            trace.log_exception_quietly()
            print e
        except KeyboardInterrupt, e:
            print "Interrupted"
        except Exception, e:
            trace.log_exception_quietly()
            print "Unhandled error:\n%s" % (e)


    def completenames(self, text, line, begidx, endidx):
        return CompletionContext(text).get_completions()

    def completedefault(self, text, line, begidx, endidx):
        """Perform completion for native commands.

        :param text: The text to complete
        :type text: str
        :param line: The entire line to complete
        :type line: str
        :param begidx: The start of the text in the line
        :type begidx: int
        :param endidx: The end of the text in the line
        :type endidx: int
        """
        (cmd, args, foo) = self.parseline(line)
        if cmd == "bzr":
            cmd = None
        return CompletionContext(text, command=cmd).get_completions()


def run_shell(directory=None):
    try:
        if not directory is None:
            os.chdir(directory)
        prompt = PromptCmd()
        while True:
            try:
                try:
                    prompt.cmdloop()
                except KeyboardInterrupt:
                    print
            finally:
                prompt.write_history()
    except StopIteration:
        pass


def iter_opt_completions(command_obj):
    for option_name, option in command_obj.options().items():
        yield "--" + option_name
        short_name = option.short_name()
        if short_name:
            yield "-" + short_name


def iter_file_completions(arg, only_dirs = False):
    """Generate an iterator that iterates through filename completions.

    :param arg: The filename fragment to match
    :type arg: str
    :param only_dirs: If true, match only directories
    :type only_dirs: bool
    """
    cwd = os.getcwd()
    if cwd != "/":
        extras = [".", ".."]
    else:
        extras = []
    (dir, file) = os.path.split(arg)
    if dir != "":
        listingdir = os.path.expanduser(dir)
    else:
        listingdir = cwd
    for file in chain(os.listdir(listingdir), extras):
        if dir != "":
            userfile = dir+'/'+file
        else:
            userfile = file
        if userfile.startswith(arg):
            if os.path.isdir(listingdir+'/'+file):
                userfile+='/'
                yield userfile
            elif not only_dirs:
                yield userfile + ' '


def iter_dir_completions(arg):
    """Generate an iterator that iterates through directory name completions.

    :param arg: The directory name fragment to match
    :type arg: str
    """
    return iter_file_completions(arg, True)


def iter_command_names(hidden=False):
    for real_cmd_name in all_command_names():
        cmd_obj = get_cmd_object(real_cmd_name)
        if not hidden and cmd_obj.hidden:
            continue
        for name in [real_cmd_name] + cmd_obj.aliases:
            # Don't complete on aliases that are prefixes of the canonical name
            if name == real_cmd_name or not real_cmd_name.startswith(name):
                yield name


def iter_executables(path):
    dirname, partial = os.path.split(path)
    for filename in os.listdir(dirname):
        if not filename.startswith(partial):
            continue
        fullpath = os.path.join(dirname, filename)
        mode=os.lstat(fullpath)[stat.ST_MODE]
        if stat.S_ISREG(mode) and 0111 & mode:
            yield fullpath + ' '


def filter_completions(iter, arg):
    return (c for c in iter if c.startswith(arg))


def iter_munged_completions(iter, arg, text):
    for completion in iter:
        completion = str(completion)
        if completion.startswith(arg):
            yield completion[len(arg)-len(text):]+" "


def too_complicated(line):
    for char in '|<>*?':
        if char in line:
            return True
    return False