#!/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"]