233 lines
8.5 KiB
Python
233 lines
8.5 KiB
Python
|
#!/usr/bin/env python
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
|
||
|
|
from re import compile as re_compile
|
||
|
|
|
||
|
|
if "LibreTranslatePlus" not in globals():
|
||
|
|
class LibreTranslatePlus:pass
|
||
|
|
|
||
|
|
class Anonymous(LibreTranslatePlus.Abstracts.Base): # LibreTranslatePlus.Terminal
|
||
|
|
|
||
|
|
re_command = re_compile(r'^([^\s]+)(\s+(.*))?$')
|
||
|
|
re_command_attributes = re_compile(r'\s*("(([^\\\\"]+|\\\\.)*)"|\'(([^\\\\\']+|\\\\.)*)\'|([^\s\:\=]+))([\:\=]("(([^\\\\"]+|\\\\.)*)"|\'(([^\\\\\']+|\\\\.)*)\'|([^\s]+))?)?')
|
||
|
|
|
||
|
|
def __init__(self, ltp, _input = None):
|
||
|
|
super().__init__(ltp)
|
||
|
|
|
||
|
|
self._print("info", "ltp_terminal_building")
|
||
|
|
|
||
|
|
self.__started = False
|
||
|
|
self.__closed = False
|
||
|
|
self.__commands = {}
|
||
|
|
self.__thread = None
|
||
|
|
self.__set_common_variables()
|
||
|
|
|
||
|
|
self._print("ok", "ltp_terminal_built")
|
||
|
|
|
||
|
|
def start(self, callback = None):
|
||
|
|
|
||
|
|
self._print("info", "ltp_terminal_starting")
|
||
|
|
|
||
|
|
end = lambda status:callable(callback) and callback(status)
|
||
|
|
|
||
|
|
if self.__started:
|
||
|
|
self._print("warn", "ltp_terminal_already_started")
|
||
|
|
end(False)
|
||
|
|
return False
|
||
|
|
self.__started = True
|
||
|
|
|
||
|
|
self.__set_common_variables()
|
||
|
|
|
||
|
|
for key in ("default_commands", "commands"):
|
||
|
|
self.add_commands(self.settings(key), True)
|
||
|
|
|
||
|
|
self.__thread = self.ltp.threads.add(self.__listen, {
|
||
|
|
"bucle" : True
|
||
|
|
})
|
||
|
|
|
||
|
|
self._print("ok", "ltp_terminal_started")
|
||
|
|
|
||
|
|
end(True)
|
||
|
|
|
||
|
|
def __set_common_variables(self):
|
||
|
|
self.__show_exception = self.settings(("terminal_show_exception", "show_exception"))
|
||
|
|
self.__show_add_error = self.settings(("terminal_show_add_command_error_message", "show_error_message"))
|
||
|
|
self.__show_add_ok = self.settings(("terminal_show_add_command_ok_message", "show_ok_message"))
|
||
|
|
self.__show_add_exception = self.settings(("terminal_show_add_command_exception_message", "show_exception_message"))
|
||
|
|
|
||
|
|
def close(self, callback = None):
|
||
|
|
|
||
|
|
self._print("info", "ltp_terminal_closing")
|
||
|
|
|
||
|
|
end = lambda status:callable(callback) and callback(status)
|
||
|
|
|
||
|
|
if self.__closed:
|
||
|
|
end(False)
|
||
|
|
self._print("warn", "ltp_terminal_already_closed")
|
||
|
|
return False
|
||
|
|
self.__closed = True
|
||
|
|
|
||
|
|
self.ltp.threads.remove(self.__thread)
|
||
|
|
|
||
|
|
self._print("ok", "ltp_terminal_closed")
|
||
|
|
|
||
|
|
end(True)
|
||
|
|
|
||
|
|
return True
|
||
|
|
|
||
|
|
def __get_method(self, method):
|
||
|
|
|
||
|
|
level = self.ltp
|
||
|
|
|
||
|
|
for fragment in method.split("."):
|
||
|
|
if hasattr(level, fragment):
|
||
|
|
level = getattr(level, fragment)
|
||
|
|
else:
|
||
|
|
level = None
|
||
|
|
break
|
||
|
|
|
||
|
|
return level
|
||
|
|
|
||
|
|
def add_commands(self, inputs, overwrite = None):
|
||
|
|
if isinstance(inputs, (list, tuple)):
|
||
|
|
for subinputs in inputs:
|
||
|
|
self.add_commands(subinputs, overwrite)
|
||
|
|
elif isinstance(inputs, dict):
|
||
|
|
if not isinstance(overwrite, bool):
|
||
|
|
overwrite = self.settings(("terminal_overwrite", "overwrite"))
|
||
|
|
for key, command in inputs.items():
|
||
|
|
|
||
|
|
error = 1 << 1 if not overwrite and key in self.__commands else 0
|
||
|
|
|
||
|
|
if not error:
|
||
|
|
try:
|
||
|
|
|
||
|
|
if isinstance(command, str):
|
||
|
|
|
||
|
|
if command in self.__commands:
|
||
|
|
self.__commands[key] = self.__command[command]
|
||
|
|
else:
|
||
|
|
|
||
|
|
method = self.__get_method(command)
|
||
|
|
error |= (
|
||
|
|
1 << 0 if method == None else
|
||
|
|
1 << 1 if not callable(method) else
|
||
|
|
0) << 2
|
||
|
|
|
||
|
|
if not error:
|
||
|
|
self.__commands[key] = {"method" : method}
|
||
|
|
|
||
|
|
elif isinstance(command, dict):
|
||
|
|
|
||
|
|
error |= ((
|
||
|
|
1 << 0 if not len(list(command.keys())) else
|
||
|
|
1 << 1 if "method" not in command else
|
||
|
|
1 << 2 if command["method"] == None else
|
||
|
|
0) | (
|
||
|
|
(1 << 3 if not command["method"] else 0) if isinstance(command["method"], str) else
|
||
|
|
(1 << 4 if not callable(command["method"]) else 0)
|
||
|
|
)) << 4
|
||
|
|
|
||
|
|
if not error:
|
||
|
|
|
||
|
|
if isinstance(command["method"], str):
|
||
|
|
command["method"] = self.__get_method(command["method"])
|
||
|
|
error |= (
|
||
|
|
1 << 0 if command["method"] == None else
|
||
|
|
1 << 1 if not callable(command["method"]) else
|
||
|
|
0) << 9
|
||
|
|
if error:
|
||
|
|
break
|
||
|
|
|
||
|
|
if "aliases" in command:
|
||
|
|
aliases = command["aliases"]
|
||
|
|
del command["aliases"]
|
||
|
|
else:
|
||
|
|
aliases = None
|
||
|
|
|
||
|
|
self.__commands[key] = command
|
||
|
|
|
||
|
|
if aliases:
|
||
|
|
for alias in (
|
||
|
|
aliases if isinstance(aliases, (list, tuple)) else
|
||
|
|
(aliases,) if isinstance(aliases, str) and aliases else
|
||
|
|
tuple()):
|
||
|
|
if overwrite or alias not in self.__commands:
|
||
|
|
self.__commands[alias] = self.__commands[key]
|
||
|
|
|
||
|
|
except Exception as exception:
|
||
|
|
error |= 1 << 0
|
||
|
|
self.exception(exception, self.__show_add_command_exception_message and "ltp_terminal_add_command_exception", {"key" : key})
|
||
|
|
|
||
|
|
self.validate(
|
||
|
|
error,
|
||
|
|
(
|
||
|
|
"exception",
|
||
|
|
"command_key_exists",
|
||
|
|
"method_not_found",
|
||
|
|
"method_not_function",
|
||
|
|
"command_empty",
|
||
|
|
"command_not_method",
|
||
|
|
"method_null",
|
||
|
|
"method_empty",
|
||
|
|
"method_not_function",
|
||
|
|
"method_not_found",
|
||
|
|
"method_not_function"
|
||
|
|
),
|
||
|
|
{"key" : key},
|
||
|
|
self.__show_add_error and "ltp_terminal_add_command_error",
|
||
|
|
self.__show_add_ok and "ltp_terminal_add_command_ok"
|
||
|
|
)
|
||
|
|
|
||
|
|
elif isinstance(inputs, str):
|
||
|
|
self.add_commands(self.ltp.load_json(inputs, self.__show_exception)[0], overwrite)
|
||
|
|
|
||
|
|
def __listen(self, i):
|
||
|
|
|
||
|
|
command = input()
|
||
|
|
|
||
|
|
if not command:
|
||
|
|
return
|
||
|
|
|
||
|
|
matches = LibreTranslatePlus.Terminal.re_command.match(command)
|
||
|
|
command_name = matches.group(1)
|
||
|
|
command_attributes = matches.group(3)
|
||
|
|
attributes = {}
|
||
|
|
parameters = []
|
||
|
|
|
||
|
|
if command_attributes:
|
||
|
|
while True:
|
||
|
|
|
||
|
|
matches = LibreTranslatePlus.Terminal.re_command_attributes.match(command_attributes)
|
||
|
|
|
||
|
|
if not matches:
|
||
|
|
break
|
||
|
|
|
||
|
|
length = matches.start() + matches.end()
|
||
|
|
matches = matches.groups()
|
||
|
|
key = matches[1] or matches[3] or matches[5]
|
||
|
|
value = matches[
|
||
|
|
8 if matches[8] != None else
|
||
|
|
10 if matches[10] != None else
|
||
|
|
12]
|
||
|
|
|
||
|
|
if value != None:
|
||
|
|
attributes[key] = value
|
||
|
|
else:
|
||
|
|
parameters += [key]
|
||
|
|
|
||
|
|
command_attributes = command_attributes[length:]
|
||
|
|
|
||
|
|
if command_name not in self.__commands:
|
||
|
|
if command in ("bye", "close", "exit"):
|
||
|
|
self.anprm.close()
|
||
|
|
self._print("warn", "ltp_terminal_unknown_command", {"command" : command_name})
|
||
|
|
return
|
||
|
|
|
||
|
|
try:
|
||
|
|
self.__commands[command_name]["method"](attributes, parameters)
|
||
|
|
except Exception as exception:
|
||
|
|
self.exception(exception, "ltp_terminal_command_exception", {"command" : command_name})
|
||
|
|
|
||
|
|
LibreTranslatePlus.Terminal = Anonymous
|
||
|
|
del globals()["Anonymous"]
|