mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 10:05:28 +03:00
some fix
This commit is contained in:
@@ -1,3 +1,3 @@
|
|||||||
__all__ = ["App"]
|
__all__ = ["App"]
|
||||||
|
|
||||||
from .entity import App
|
from .models import App
|
||||||
|
|||||||
@@ -23,8 +23,3 @@ class NoRegisteredHandlersException(Exception):
|
|||||||
self.router_name = router_name
|
self.router_name = router_name
|
||||||
def __str__(self):
|
def __str__(self):
|
||||||
return f"No Registered Handlers Found For '{self.router_name}'"
|
return f"No Registered Handlers Found For '{self.router_name}'"
|
||||||
|
|
||||||
|
|
||||||
class IncorrectNumberOfHandlerArgsException(Exception):
|
|
||||||
def __str__(self):
|
|
||||||
return "Incorrect Input Flags Handler has incorrect number of arguments"
|
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
from typing import Callable
|
from typing import Callable
|
||||||
from inspect import getfullargspec
|
|
||||||
from rich.console import Console
|
from rich.console import Console
|
||||||
from art import text2art
|
from art import text2art
|
||||||
import re
|
import re
|
||||||
@@ -12,27 +11,23 @@ from argenta.command.exceptions import (UnprocessedInputFlagException,
|
|||||||
EmptyInputCommandException,
|
EmptyInputCommandException,
|
||||||
BaseInputCommandException)
|
BaseInputCommandException)
|
||||||
from argenta.app.exceptions import (InvalidRouterInstanceException,
|
from argenta.app.exceptions import (InvalidRouterInstanceException,
|
||||||
InvalidDescriptionMessagePatternException,
|
InvalidDescriptionMessagePatternException,
|
||||||
NoRegisteredRoutersException,
|
NoRegisteredRoutersException,
|
||||||
NoRegisteredHandlersException,
|
NoRegisteredHandlersException)
|
||||||
IncorrectNumberOfHandlerArgsException)
|
|
||||||
from argenta.app.registered_routers.entity import RegisteredRouters
|
from argenta.app.registered_routers.entity import RegisteredRouters
|
||||||
|
|
||||||
|
|
||||||
class App:
|
|
||||||
|
class BaseApp:
|
||||||
def __init__(self,
|
def __init__(self,
|
||||||
prompt: str = '[italic dim bold]What do you want to do?\n',
|
prompt: str = '[italic dim bold]What do you want to do?\n',
|
||||||
initial_message: str = f'\n[bold red]{text2art('Argenta', font='tarty1')}\n\n',
|
initial_message: str = 'Argenta',
|
||||||
farewell_message: str = f'[bold red]\n{text2art('\nSee you\n', font='chanky')}[/bold red]\n'
|
farewell_message: str = 'See you',
|
||||||
f'[red i]github.com/koloideal/Argenta[/red i] | '
|
|
||||||
f'[red bold i]made by kolo[/red bold i]\n',
|
|
||||||
exit_command: str = 'Q',
|
exit_command: str = 'Q',
|
||||||
exit_command_description: str = 'Exit command',
|
exit_command_description: str = 'Exit command',
|
||||||
system_points_title: str = 'System points:',
|
system_points_title: str = 'System points:',
|
||||||
ignore_exit_command_register: bool = True,
|
|
||||||
ignore_command_register: bool = True,
|
ignore_command_register: bool = True,
|
||||||
line_separate: str = f'\n[dim]{"--"*25}\n',
|
line_separate: str = '-----',
|
||||||
command_group_description_separate: str = '',
|
|
||||||
repeat_command_groups: bool = True,
|
repeat_command_groups: bool = True,
|
||||||
print_func: Callable[[str], None] = Console().print) -> None:
|
print_func: Callable[[str], None] = Console().print) -> None:
|
||||||
self._prompt = prompt
|
self._prompt = prompt
|
||||||
@@ -40,30 +35,103 @@ class App:
|
|||||||
self._exit_command = exit_command
|
self._exit_command = exit_command
|
||||||
self._exit_command_description = exit_command_description
|
self._exit_command_description = exit_command_description
|
||||||
self._system_points_title = system_points_title
|
self._system_points_title = system_points_title
|
||||||
self._ignore_exit_command_register = ignore_exit_command_register
|
|
||||||
self._farewell_message = farewell_message
|
|
||||||
self._initial_message = initial_message
|
|
||||||
self._line_separate = line_separate
|
self._line_separate = line_separate
|
||||||
self._command_group_description_separate = command_group_description_separate
|
|
||||||
self._ignore_command_register = ignore_command_register
|
self._ignore_command_register = ignore_command_register
|
||||||
self._repeat_command_groups_description = repeat_command_groups
|
self._repeat_command_groups_description = repeat_command_groups
|
||||||
|
|
||||||
|
self.farewell_message = farewell_message
|
||||||
|
self.initial_message = initial_message
|
||||||
|
|
||||||
self._description_message_pattern: str = '[bold red][{command}][/bold red] [blue dim]*=*=*[/blue dim] [bold yellow italic]{description}'
|
self._description_message_pattern: str = '[bold red][{command}][/bold red] [blue dim]*=*=*[/blue dim] [bold yellow italic]{description}'
|
||||||
self._registered_routers: RegisteredRouters = RegisteredRouters()
|
self._registered_routers: RegisteredRouters = RegisteredRouters()
|
||||||
self._invalid_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'[red bold]Incorrect flag syntax: {raw_command}')
|
|
||||||
self._repeated_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'[red bold]Repeated input flags: {raw_command}')
|
|
||||||
self._empty_input_command_handler: Callable[[], None] = lambda: print_func('[red bold]Empty input command')
|
|
||||||
self._unknown_command_handler: Callable[[InputCommand], None] = lambda command: print_func(f"[red bold]Unknown command: {command.get_trigger()}")
|
|
||||||
self._exit_command_handler: Callable[[], None] = lambda: print_func(self._farewell_message)
|
|
||||||
self._messages_on_startup = []
|
self._messages_on_startup = []
|
||||||
|
|
||||||
|
self.invalid_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'[red bold]Incorrect flag syntax: {raw_command}')
|
||||||
|
self.repeated_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'[red bold]Repeated input flags: {raw_command}')
|
||||||
|
self.empty_input_command_handler: Callable[[], None] = lambda: print_func('[red bold]Empty input command')
|
||||||
|
self.unknown_command_handler: Callable[[InputCommand], None] = lambda command: print_func(f"[red bold]Unknown command: {command.get_trigger()}")
|
||||||
|
self.exit_command_handler: Callable[[], None] = lambda: print_func(self.farewell_message)
|
||||||
|
|
||||||
|
self._setup_default_view(is_initial_message_default=initial_message == 'Argenta',
|
||||||
|
is_farewell_message_default=farewell_message == 'See you',
|
||||||
|
is_line_separate_default=line_separate == '-----')
|
||||||
|
|
||||||
|
|
||||||
|
def _setup_default_view(self, is_initial_message_default: bool,
|
||||||
|
is_farewell_message_default: bool,
|
||||||
|
is_line_separate_default: bool):
|
||||||
|
if is_initial_message_default:
|
||||||
|
self.initial_message = f'\n[bold red]{text2art('Argenta', font='tarty1')}\n\n'
|
||||||
|
if is_farewell_message_default:
|
||||||
|
self.farewell_message = (f'[bold red]\n{text2art('\nSee you\n', font='chanky')}[/bold red]\n'
|
||||||
|
f'[red i]github.com/koloideal/Argenta[/red i] | '
|
||||||
|
f'[red bold i]made by kolo[/red bold i]\n')
|
||||||
|
if is_line_separate_default:
|
||||||
|
self._line_separate = f'\n[dim]{"-" * 50}\n'
|
||||||
|
|
||||||
|
|
||||||
|
def _validate_number_of_routers(self) -> None:
|
||||||
|
if not self._registered_routers:
|
||||||
|
raise NoRegisteredRoutersException()
|
||||||
|
|
||||||
|
|
||||||
|
def _validate_included_routers(self) -> None:
|
||||||
|
for router in self._registered_routers:
|
||||||
|
if not router.get_command_handlers():
|
||||||
|
raise NoRegisteredHandlersException(router.get_name())
|
||||||
|
|
||||||
|
|
||||||
|
def _is_exit_command(self, command: InputCommand):
|
||||||
|
if command.get_trigger().lower() == self._exit_command.lower():
|
||||||
|
if self._ignore_command_register:
|
||||||
|
system_router.input_command_handler(command)
|
||||||
|
return True
|
||||||
|
elif command.get_trigger() == self._exit_command:
|
||||||
|
system_router.input_command_handler(command)
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def _is_unknown_command(self, command: InputCommand):
|
||||||
|
for router_entity in self._registered_routers:
|
||||||
|
for command_handler in router_entity.get_command_handlers():
|
||||||
|
handled_command_trigger = command_handler.get_handled_command().get_trigger()
|
||||||
|
if handled_command_trigger.lower() == command.get_trigger().lower() and self._ignore_command_register:
|
||||||
|
return False
|
||||||
|
elif handled_command_trigger == command.get_trigger():
|
||||||
|
return False
|
||||||
|
self.unknown_command_handler(command)
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def _print_command_group_description(self):
|
||||||
|
for registered_router in self._registered_routers:
|
||||||
|
self._print_func(registered_router.get_title())
|
||||||
|
for command_handler in registered_router.get_command_handlers():
|
||||||
|
self._print_func(self._description_message_pattern.format(
|
||||||
|
command=command_handler.get_handled_command().get_trigger(),
|
||||||
|
description=command_handler.get_handled_command().get_description()))
|
||||||
|
self._print_func('')
|
||||||
|
|
||||||
|
|
||||||
|
def _error_handler(self, error: BaseInputCommandException, raw_command: str) -> None:
|
||||||
|
match error:
|
||||||
|
case UnprocessedInputFlagException():
|
||||||
|
self.invalid_input_flags_handler(raw_command)
|
||||||
|
case RepeatedInputFlagsException():
|
||||||
|
self.repeated_input_flags_handler(raw_command)
|
||||||
|
case EmptyInputCommandException():
|
||||||
|
self.empty_input_command_handler()
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
class App(BaseApp):
|
||||||
def start_polling(self) -> None:
|
def start_polling(self) -> None:
|
||||||
self._setup_system_router()
|
self._setup_system_router()
|
||||||
self._validate_number_of_routers()
|
self._validate_number_of_routers()
|
||||||
self._validate_included_routers()
|
self._validate_included_routers()
|
||||||
|
|
||||||
self._print_func(self._initial_message)
|
self._print_func(self.initial_message)
|
||||||
|
|
||||||
for message in self._messages_on_startup:
|
for message in self._messages_on_startup:
|
||||||
self._print_func(message)
|
self._print_func(message)
|
||||||
@@ -107,68 +175,6 @@ class App:
|
|||||||
self._print_func(self._prompt)
|
self._print_func(self._prompt)
|
||||||
|
|
||||||
|
|
||||||
def set_initial_message(self, message: str) -> None:
|
|
||||||
self._initial_message: str = message
|
|
||||||
|
|
||||||
|
|
||||||
def set_farewell_message(self, message: str) -> None:
|
|
||||||
self._farewell_message: str = message
|
|
||||||
|
|
||||||
|
|
||||||
def set_description_message_pattern(self, pattern: str) -> None:
|
|
||||||
first_check = re.match(r'.*{command}.*', pattern)
|
|
||||||
second_check = re.match(r'.*{description}.*', pattern)
|
|
||||||
|
|
||||||
if bool(first_check) and bool(second_check):
|
|
||||||
self._description_message_pattern: str = pattern
|
|
||||||
else:
|
|
||||||
raise InvalidDescriptionMessagePatternException(pattern)
|
|
||||||
|
|
||||||
|
|
||||||
def set_invalid_input_flags_handler(self, handler: Callable[[str], None]) -> None:
|
|
||||||
args = getfullargspec(handler).args
|
|
||||||
if len(args) != 1:
|
|
||||||
raise IncorrectNumberOfHandlerArgsException()
|
|
||||||
else:
|
|
||||||
self._invalid_input_flags_handler = handler
|
|
||||||
|
|
||||||
|
|
||||||
def set_repeated_input_flags_handler(self, handler: Callable[[str], None]) -> None:
|
|
||||||
args = getfullargspec(handler).args
|
|
||||||
if len(args) != 1:
|
|
||||||
raise IncorrectNumberOfHandlerArgsException()
|
|
||||||
else:
|
|
||||||
self._repeated_input_flags_handler = handler
|
|
||||||
|
|
||||||
|
|
||||||
def set_unknown_command_handler(self, handler: Callable[[str], None]) -> None:
|
|
||||||
args = getfullargspec(handler).args
|
|
||||||
if len(args) != 1:
|
|
||||||
raise IncorrectNumberOfHandlerArgsException()
|
|
||||||
else:
|
|
||||||
self._unknown_command_handler = handler
|
|
||||||
|
|
||||||
|
|
||||||
def set_empty_command_handler(self, handler: Callable[[], None]) -> None:
|
|
||||||
args = getfullargspec(handler).args
|
|
||||||
if len(args) != 0:
|
|
||||||
raise IncorrectNumberOfHandlerArgsException()
|
|
||||||
else:
|
|
||||||
self._empty_input_command_handler = handler
|
|
||||||
|
|
||||||
|
|
||||||
def set_exit_command_handler(self, handler: Callable[[], None]) -> None:
|
|
||||||
args = getfullargspec(handler).args
|
|
||||||
if len(args) != 0:
|
|
||||||
raise IncorrectNumberOfHandlerArgsException()
|
|
||||||
else:
|
|
||||||
self._exit_command_handler = handler
|
|
||||||
|
|
||||||
|
|
||||||
def add_message_on_startup(self, message: str) -> None:
|
|
||||||
self._messages_on_startup.append(message)
|
|
||||||
|
|
||||||
|
|
||||||
def include_router(self, router: Router) -> None:
|
def include_router(self, router: Router) -> None:
|
||||||
if not isinstance(router, Router):
|
if not isinstance(router, Router):
|
||||||
raise InvalidRouterInstanceException()
|
raise InvalidRouterInstanceException()
|
||||||
@@ -182,69 +188,27 @@ class App:
|
|||||||
self.include_router(router)
|
self.include_router(router)
|
||||||
|
|
||||||
|
|
||||||
def _validate_number_of_routers(self) -> None:
|
def add_message_on_startup(self, message: str) -> None:
|
||||||
if not self._registered_routers:
|
self._messages_on_startup.append(message)
|
||||||
raise NoRegisteredRoutersException()
|
|
||||||
|
|
||||||
|
|
||||||
def _validate_included_routers(self) -> None:
|
def set_description_message_pattern(self, pattern: str) -> None:
|
||||||
for router in self._registered_routers:
|
first_check = re.match(r'.*{command}.*', pattern)
|
||||||
if not router.get_command_handlers():
|
second_check = re.match(r'.*{description}.*', pattern)
|
||||||
raise NoRegisteredHandlersException(router.get_name())
|
|
||||||
|
if bool(first_check) and bool(second_check):
|
||||||
|
self._description_message_pattern: str = pattern
|
||||||
|
else:
|
||||||
|
raise InvalidDescriptionMessagePatternException(pattern)
|
||||||
|
|
||||||
|
|
||||||
def _setup_system_router(self):
|
def _setup_system_router(self):
|
||||||
system_router.set_title(self._system_points_title)
|
system_router.set_title(self._system_points_title)
|
||||||
@system_router.command(Command(trigger=self._exit_command,
|
@system_router.command(Command(self._exit_command, self._exit_command_description))
|
||||||
description=self._exit_command_description))
|
|
||||||
def exit_command():
|
def exit_command():
|
||||||
self._exit_command_handler()
|
self.exit_command_handler()
|
||||||
|
|
||||||
if system_router not in self._registered_routers.get_registered_routers():
|
if system_router not in self._registered_routers.get_registered_routers():
|
||||||
self.include_router(system_router)
|
self.include_router(system_router)
|
||||||
|
|
||||||
|
|
||||||
def _is_exit_command(self, command: InputCommand):
|
|
||||||
if command.get_trigger().lower() == self._exit_command.lower():
|
|
||||||
if self._ignore_exit_command_register:
|
|
||||||
system_router.input_command_handler(command)
|
|
||||||
return True
|
|
||||||
elif command.get_trigger() == self._exit_command:
|
|
||||||
system_router.input_command_handler(command)
|
|
||||||
return True
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
def _is_unknown_command(self, command: InputCommand):
|
|
||||||
for router_entity in self._registered_routers:
|
|
||||||
for command_handler in router_entity.get_command_handlers():
|
|
||||||
handled_command_trigger = command_handler.get_handled_command().get_trigger()
|
|
||||||
if handled_command_trigger.lower() == command.get_trigger().lower():
|
|
||||||
if self._ignore_command_register:
|
|
||||||
return False
|
|
||||||
elif handled_command_trigger == command.get_trigger():
|
|
||||||
return False
|
|
||||||
self._unknown_command_handler(command)
|
|
||||||
return True
|
|
||||||
|
|
||||||
|
|
||||||
def _print_command_group_description(self):
|
|
||||||
for registered_router in self._registered_routers:
|
|
||||||
self._print_func(registered_router.get_title())
|
|
||||||
for command_handler in registered_router.get_command_handlers():
|
|
||||||
self._print_func(self._description_message_pattern.format(
|
|
||||||
command=command_handler.get_handled_command().get_trigger(),
|
|
||||||
description=command_handler.get_handled_command().get_description()))
|
|
||||||
self._print_func(self._command_group_description_separate)
|
|
||||||
|
|
||||||
|
|
||||||
def _error_handler(self, error: BaseInputCommandException, raw_command: str) -> None:
|
|
||||||
match error:
|
|
||||||
case UnprocessedInputFlagException():
|
|
||||||
self._invalid_input_flags_handler(raw_command)
|
|
||||||
case RepeatedInputFlagsException():
|
|
||||||
self._repeated_input_flags_handler(raw_command)
|
|
||||||
case EmptyInputCommandException():
|
|
||||||
self._empty_input_command_handler()
|
|
||||||
|
|
||||||
|
|
||||||
@@ -1,5 +1,4 @@
|
|||||||
from mock.mock_app.handlers.routers import work_router, settings_router
|
from mock.mock_app.handlers.routers import work_router, settings_router
|
||||||
from art import text2art
|
|
||||||
|
|
||||||
from argenta.app import App
|
from argenta.app import App
|
||||||
from argenta.app.defaults import PredeterminedMessages
|
from argenta.app.defaults import PredeterminedMessages
|
||||||
|
|||||||
Reference in New Issue
Block a user