jp (shell): shell command (REPL mode), first draft:
This command launch jp in REPL mode, allowing do normal jp commands with some facilities. Command can be selected with "cmd" (e.g. "cmd blog"). An argument can be fixed with "use" (e.g. "use output fancy"). Command is launched with "do", or directly with its name if it doesn't conflict with a shell command. Arguments completion is still TODO (only shell commands are completed so far).
This commit is contained in:
Goffi
parent
73ed5094b9
commit
6672bfc57b
3 changed files with 269 additions and 3 deletions
|
|
@ -165,6 +165,9 @@ class Jp(object):
|
|||
|
||||
self.bridge = bridge_module.Bridge()
|
||||
self.bridge.bridgeConnect(callback=self._bridgeCb, errback=self._bridgeEb)
|
||||
# _no_exit is used in shell mode because SystemExit can't
|
||||
# catched when loop is in use
|
||||
self._no_exit = False
|
||||
|
||||
def _bridgeCb(self):
|
||||
self.parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
|
|
@ -402,8 +405,8 @@ class Jp(object):
|
|||
cls = getattr(module, class_name)
|
||||
cls(self)
|
||||
|
||||
def run(self, args=None):
|
||||
self.args = self.parser.parse_args(args)
|
||||
def run(self, args=None, namespace=None):
|
||||
self.args = self.parser.parse_args(args, namespace=None)
|
||||
try:
|
||||
self.args.func()
|
||||
if self._need_loop or self._auto_loop:
|
||||
|
|
@ -442,7 +445,8 @@ class Jp(object):
|
|||
callback(*args, **kwargs)
|
||||
|
||||
self.stop_loop()
|
||||
sys.exit(errcode)
|
||||
if not self._no_exit:
|
||||
sys.exit(errcode)
|
||||
|
||||
def check_jids(self, jids):
|
||||
"""Check jids validity, transform roster name to corresponding jids
|
||||
|
|
|
|||
259
frontends/src/jp/cmd_shell.py
Normal file
259
frontends/src/jp/cmd_shell.py
Normal file
|
|
@ -0,0 +1,259 @@
|
|||
#!/usr/bin/env python2
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# jp: a SàT command line tool
|
||||
# Copyright (C) 2009-2016 Jérôme Poisson (goffi@goffi.org)
|
||||
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as published by
|
||||
# the Free Software Foundation, either version 3 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 Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
|
||||
import base
|
||||
import cmd
|
||||
from sat.core.i18n import _
|
||||
from sat.core import exceptions
|
||||
from sat_frontends.jp.constants import Const as C
|
||||
from sat.tools.common.ansi import ANSI as A
|
||||
import shlex
|
||||
|
||||
__commands__ = ["Shell"]
|
||||
INTRO = _(u"""Welcome to {app_name} shell, the Salut à Toi shell !
|
||||
|
||||
This enrironment helps you using several {app_name} commands with similar parameters.
|
||||
|
||||
To quit, just enter "quit" or press C-c.
|
||||
Enter "help" or "?" to know what to do
|
||||
""").format(app_name = C.APP_NAME)
|
||||
|
||||
|
||||
class Shell(base.CommandBase, cmd.Cmd):
|
||||
|
||||
def __init__(self, host):
|
||||
base.CommandBase.__init__(self, host, 'shell', help=_(u'launch jp in shell (REPL) mode'))
|
||||
cmd.Cmd.__init__(self)
|
||||
|
||||
def parse_args(self, args):
|
||||
"""parse line arguments"""
|
||||
return shlex.split(args, posix=True)
|
||||
|
||||
def get_cmd_choices(self, cmd=None, parser=None):
|
||||
if parser is None:
|
||||
parser = self._cur_parser
|
||||
try:
|
||||
choices = parser._subparsers._group_actions[0].choices
|
||||
return choices[cmd] if cmd is not None else choices
|
||||
except (KeyError, AttributeError):
|
||||
raise exceptions.NotFound
|
||||
|
||||
def update_path(self):
|
||||
self._cur_parser = self.host.parser
|
||||
self.help = u''
|
||||
for idx, path_elt in enumerate(self.path):
|
||||
try:
|
||||
self._cur_parser = self.get_cmd_choices(path_elt)
|
||||
except exceptions.NotFound:
|
||||
self.disp(_(u'bad command path'), error=True)
|
||||
self.path=self.path[:idx]
|
||||
break
|
||||
else:
|
||||
self.help = self._cur_parser
|
||||
|
||||
self.prompt = A.color(C.A_PROMPT_PATH, u'/'.join(self.path)) + A.color(C.A_PROMPT_SUF, u'> ')
|
||||
try:
|
||||
self.actions = self.get_cmd_choices().keys()
|
||||
except exceptions.NotFound:
|
||||
self.actions = []
|
||||
|
||||
def add_parser_options(self):
|
||||
pass
|
||||
|
||||
def escape_arg(self, arg):
|
||||
"""format arg with quotes"""
|
||||
return u'"' + arg.replace(u'"',u'\\"') + u'"'
|
||||
|
||||
def format_args(self, args):
|
||||
"""format argument to be printed with quotes if needed"""
|
||||
for arg in args:
|
||||
if " " in arg:
|
||||
yield self.escape_arg(arg)
|
||||
else:
|
||||
yield arg
|
||||
|
||||
def get_use_args(self, args):
|
||||
"""format args for current parser according to self.use"""
|
||||
parser = self._cur_parser
|
||||
|
||||
# we check not optional args to see if there
|
||||
# is a corresonding parser
|
||||
# else USE args would not work correctly (only for current parser)
|
||||
cmd_args = []
|
||||
for arg in args:
|
||||
if arg.startswith(u'-'):
|
||||
break
|
||||
try:
|
||||
parser = self.get_cmd_choices(arg, parser)
|
||||
except exceptions.NotFound:
|
||||
break
|
||||
cmd_args.append(arg)
|
||||
|
||||
# we remove command args
|
||||
# they'll be in returned list (use_args)
|
||||
del args[:len(cmd_args)]
|
||||
|
||||
opt_args = []
|
||||
pos_args = []
|
||||
actions = {a.dest: a for a in parser._actions}
|
||||
for arg, value in self.use.iteritems():
|
||||
try:
|
||||
action = actions[arg]
|
||||
except KeyError:
|
||||
if self.verbose:
|
||||
self.disp(_(u'ignoring {name}={value}, not corresponding to any argument (in USE)').format(
|
||||
name=arg,
|
||||
value=self.escape_arg(value)))
|
||||
else:
|
||||
if self.verbose:
|
||||
self.disp(_(u'arg {name}={value} set (in USE)').format(name=arg, value=self.escape_arg(value)))
|
||||
if not action.option_strings:
|
||||
pos_args.append(value)
|
||||
else:
|
||||
opt_args.append(action.option_strings[0])
|
||||
opt_args.append(value)
|
||||
return cmd_args + opt_args + pos_args
|
||||
|
||||
def default(self, args):
|
||||
"""called when no shell command is recognized
|
||||
|
||||
will launch the command with args on the line
|
||||
(i.e. will launch do [args])
|
||||
"""
|
||||
self.do_do(args)
|
||||
|
||||
def do_help(self, args):
|
||||
"""show help message"""
|
||||
self.disp(A.color(C.A_HEADER, _(u'Shell commands:')), no_lf=True)
|
||||
super(Shell, self).do_help(args)
|
||||
if not args:
|
||||
self.disp(A.color(C.A_HEADER, _(u'Action commands:')))
|
||||
help_list = self._cur_parser.format_help().split('\n\n')
|
||||
print('\n\n'.join(help_list[1 if self.path else 2:]))
|
||||
|
||||
def do_debug(self, args):
|
||||
"""launch internal debugger"""
|
||||
try:
|
||||
import ipdb as pdb
|
||||
except ImportError:
|
||||
import pdb
|
||||
pdb.set_trace()
|
||||
|
||||
def do_verbose(self, args):
|
||||
"""show verbose mode, or (de)activate it"""
|
||||
args = self.parse_args(args)
|
||||
if args:
|
||||
self.verbose = C.bool(args[0])
|
||||
self.disp(_(u'verbose mode is {status}').format(
|
||||
status = _(u'ENABLED') if self.verbose else _(u'DISABLED')))
|
||||
|
||||
def do_cmd(self, args):
|
||||
"""change command path"""
|
||||
if args == '..':
|
||||
self.path = self.path[:-1]
|
||||
else:
|
||||
if not args or args[0] == '/':
|
||||
self.path = []
|
||||
args = '/'.join(args.split())
|
||||
for path_elt in args.split('/'):
|
||||
path_elt = path_elt.strip()
|
||||
if not path_elt:
|
||||
continue
|
||||
self.path.append(path_elt)
|
||||
self.update_path()
|
||||
|
||||
def do_version(self, args):
|
||||
"""show current SàT/jp version"""
|
||||
try:
|
||||
self.host.run(['--version'])
|
||||
except SystemExit:
|
||||
pass
|
||||
|
||||
def do_do(self, args):
|
||||
"""lauch a command"""
|
||||
# we don't want host to really exit
|
||||
# we want to stay in the loop
|
||||
self.host._no_exit = True
|
||||
args = self.parse_args(args)
|
||||
# args may be modified by use_args
|
||||
# to remove subparsers from it
|
||||
use_args = self.get_use_args(args)
|
||||
cmd_args = self.path + use_args + args
|
||||
if self.verbose:
|
||||
self.disp(u"COMMAND => {args}".format(args=u' '.join(self.format_args(cmd_args))))
|
||||
try:
|
||||
self.host.run(cmd_args)
|
||||
except SystemExit as e:
|
||||
if e.code != 0:
|
||||
self.disp(A.color(C.A_FAILURE, u'command failed with an error code of {err_no}'.format(err_no=e.code)), error=True)
|
||||
except Exception as e:
|
||||
self.disp(A.color(C.A_FAILURE, u'command failed with an exception: {msg}'.format(msg=e)), error=True)
|
||||
finally:
|
||||
self.host._no_exit = False
|
||||
|
||||
def do_use(self, args):
|
||||
"""fix an argument"""
|
||||
args = self.parse_args(args)
|
||||
if not args:
|
||||
if not self.use:
|
||||
self.disp(_(u'no argument in USE'))
|
||||
else:
|
||||
self.disp(_(u'arguments in USE:'))
|
||||
for arg, value in self.use.iteritems():
|
||||
self.disp(_(A.color(C.A_SUBHEADER, arg, A.RESET, u' = ', self.escape_arg(value))))
|
||||
elif len(args) != 2:
|
||||
self.disp(u'bad syntax, please use:\nuse [arg] [value]', error=True)
|
||||
else:
|
||||
self.use[args[0]] = u' '.join(args[1:])
|
||||
if self.verbose:
|
||||
self.disp('set {name} = {value}'.format(
|
||||
name = args[0], value=self.escape_arg(args[1])))
|
||||
|
||||
def do_use_clear(self, args):
|
||||
"""unset one or many argument(s) in USE, or all of them if no arg is specified"""
|
||||
args = self.parse_args(args)
|
||||
if not args:
|
||||
self.use.clear()
|
||||
else:
|
||||
for arg in args:
|
||||
try:
|
||||
del self.use[arg]
|
||||
except KeyError:
|
||||
self.disp(A.color(C.A_FAILURE, _(u'argument {name} not found').format(name=arg)), error=True)
|
||||
else:
|
||||
if self.verbose:
|
||||
self.disp(_(u'argument {name} removed').format(name=arg))
|
||||
|
||||
def do_quit(self, args):
|
||||
u"""quit the shell"""
|
||||
self.disp(_(u'good bye!'))
|
||||
self.host.quit()
|
||||
|
||||
def do_exit(self, args):
|
||||
u"""alias for quit"""
|
||||
self.do_quit(args)
|
||||
|
||||
def start(self):
|
||||
self.path = []
|
||||
self._cur_parser = self.host.parser
|
||||
self.use = {}
|
||||
self.verbose = False
|
||||
self.update_path()
|
||||
self.cmdloop(INTRO.encode('utf-8'))
|
||||
|
|
@ -41,6 +41,9 @@ class Const(constants.Const):
|
|||
A_SUBHEADER = A.BOLD + A.FG_RED
|
||||
A_SUCCESS = A.BOLD + A.FG_GREEN
|
||||
A_FAILURE = A.BOLD + A.FG_RED
|
||||
# A_PROMPT_* is for shell
|
||||
A_PROMPT_PATH = A_HEADER
|
||||
A_PROMPT_SUF = A.BOLD
|
||||
|
||||
# exit codes
|
||||
EXIT_OK = 0
|
||||
|
|
|
|||
Loading…
Reference in a new issue