mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 18:15:28 +03:00
Compare commits
44 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 8fbf651223 | |||
| 459c16ec87 | |||
| d9c74310c3 | |||
| 5e6cdc342e | |||
| a378163431 | |||
| fd4f2e1570 | |||
| 3f7c577c29 | |||
| b72fcc6a11 | |||
| 158e5eb75a | |||
| 46cb2f70c5 | |||
| ef242a5732 | |||
| f87f102ced | |||
| ddf2a2fb10 | |||
| eac5358ead | |||
| 76c18ddbff | |||
| 4eeb4eb182 | |||
| 905698384a | |||
| 79ccfbb3b1 | |||
| a63c46a78b | |||
| a3a7cbf2e6 | |||
| a9e545f3d8 | |||
| 37b62fd69b | |||
| 0ae86d0b2b | |||
| ebfd5a80b3 | |||
| 250704fc88 | |||
| 87239f1501 | |||
| 48c117dd72 | |||
| 1d782f6213 | |||
| ddf396a644 | |||
| c10af04280 | |||
| 890d863391 | |||
| 29b184b2ed | |||
| 6d331a57c1 | |||
| b0eb1e3e6c | |||
| 2d088caaaf | |||
| b4b7d5442c | |||
| 684760121c | |||
| dfc3c472ce | |||
| 8a3f742636 | |||
| 213525915d | |||
| 6174de3bc0 | |||
| c78609caca | |||
| 3f88c5278e | |||
| 93438f8e45 |
+4
-4
@@ -1,6 +1,6 @@
|
||||
.venv
|
||||
.idea
|
||||
argenta/router/__pycache__/
|
||||
argenta/app/__pycache__/
|
||||
argenta/__pycache__/
|
||||
dist
|
||||
dist
|
||||
poetry.lock
|
||||
*__pycache__/
|
||||
|
||||
|
||||
@@ -1,2 +1,273 @@
|
||||
# Argenta
|
||||
Python library for creating cli apps
|
||||
|
||||
---
|
||||
|
||||
## Описание
|
||||
**Argenta** — это библиотека для создания CLI-приложений на Python. Она предоставляет удобные инструменты для маршрутизации команд и обработки пользовательского ввода.
|
||||
|
||||
---
|
||||
|
||||
# Установка
|
||||
```bash
|
||||
pip install argenta
|
||||
```
|
||||
or
|
||||
```bash
|
||||
poetry add argenta
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
# Быстрый старт
|
||||
Пример базового CLI-приложения с Argenta:
|
||||
```python
|
||||
#routers.py
|
||||
from argenta.router import Router
|
||||
|
||||
router = Router()
|
||||
|
||||
@router.command("hello")
|
||||
def hello():
|
||||
print("Hello, world!")
|
||||
|
||||
@router.unknown_command
|
||||
def unlnown_command(command):
|
||||
print(f'Command "{command}" undefined')
|
||||
```
|
||||
```python
|
||||
#main.py
|
||||
from argenta.app import App
|
||||
from routers import router
|
||||
|
||||
app: App = App()
|
||||
|
||||
def main() -> None:
|
||||
app.include_router(router)
|
||||
app.start_polling()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
# Техническая документация
|
||||
|
||||
---
|
||||
|
||||
## declared *classes* :
|
||||
|
||||
---
|
||||
|
||||
### *class* :: `App`
|
||||
Класс, определяющий поведение и состояние приложения
|
||||
|
||||
#### Конструктор
|
||||
```python
|
||||
App(prompt: str = 'Enter a command',
|
||||
initial_greeting: str = '\nHello, I am Argenta\n',
|
||||
farewell_message: str = '\nGoodBye\n',
|
||||
exit_command: str = 'Q',
|
||||
exit_command_description: str = 'Exit command',
|
||||
exit_command_title: str = 'System points:',
|
||||
ignore_exit_command_register: bool = True,
|
||||
ignore_command_register: bool = False,
|
||||
line_separate: str = '',
|
||||
command_group_description_separate: str = '',
|
||||
repeat_command_groups: bool = True,
|
||||
print_func: Callable[[str], None] = print)
|
||||
```
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `prompt` (`str`): Сообщение перед вводом команды.
|
||||
- `initial_greeting` (`str`): Приветственное сообщение при запуске.
|
||||
- `farewell_message` (`str`): Сообщение при выходе.
|
||||
- `exit_command` (`str`): Команда выхода (по умолчанию `'Q'`).
|
||||
- `exit_command_description` (`str`): Описание команды выхода.
|
||||
- `exit_command_title` (`str`): Заголовок перед списком команд выхода.
|
||||
- `ignore_exit_command_register` (`bool`): Игнорировать регистр команды выхода.
|
||||
- `ignore_command_register` (`bool`): Игнорировать регистр всех команд.
|
||||
- `line_separate` (`str`): Разделительная строка между командами.
|
||||
- `command_group_description_separate` (`str`): Разделитель между группами команд.
|
||||
- `repeat_command_groups` (`bool`): Повторять описание команд перед вводом.
|
||||
- `print_func` (`Callable[[str], None]`): Функция вывода текста в терминал (по умолчанию `print`).
|
||||
|
||||
---
|
||||
|
||||
#### **declared *methods***
|
||||
|
||||
---
|
||||
|
||||
**App().**`start_polling() -> None`
|
||||
|
||||
*method mean* **::** запускает жизненный цикл приложения
|
||||
|
||||
---
|
||||
|
||||
**App().**`include_router(router: Router, is_main: bool = False) -> None`
|
||||
|
||||
*param* `router: Router` **::** регистрируемый роутер
|
||||
|
||||
*param* `is_main: bool` **::** будет ли являться регистрируемый роутер главным
|
||||
*example* **::** `True` или `False`
|
||||
|
||||
*method mean* **::** регистрирует роутер в приложении
|
||||
|
||||
---
|
||||
|
||||
**App().**`set_initial_message(message: str) -> None`
|
||||
|
||||
*param* `message: str` **::** устанавливаемое приветственное сообщение
|
||||
*example* **::** `"Hello, I'm a cli example app"`
|
||||
|
||||
*method mean* **::** устанавливает сообщение, которое будет отображено при запуске программы
|
||||
|
||||
---
|
||||
|
||||
**App().**`set_farewell_message(message: str) -> None`
|
||||
|
||||
*param* `message: str` **::** устанавливаемое сообщение при выходе
|
||||
*example* **::** `"GoodBye !"`
|
||||
|
||||
*method mean* **::** устанавливает сообщение, которое будет отображено при выходе
|
||||
|
||||
---
|
||||
|
||||
**App().**`set_description_message_pattern(pattern: str) -> None`
|
||||
|
||||
*param* `pattern: str` **::** паттерн описания команды при её выводе в консоль
|
||||
*example* **::** `"[{command}] *=*=* {description}"`
|
||||
|
||||
*method mean* **::** устанавливает приветственное сообщение
|
||||
|
||||
---
|
||||
|
||||
**App().**`get_main_router() -> Router`
|
||||
|
||||
*method mean* **::** возвращает `Router()`, который является главным в приложении
|
||||
|
||||
---
|
||||
|
||||
**App().**`get_all_app_commands() -> list[str]`
|
||||
|
||||
*method mean* **::** возвращает список команд всех зарегистрированных роутеров, сохраняя их регистр
|
||||
|
||||
---
|
||||
|
||||
#### Примечания
|
||||
|
||||
- Среди зарегистрированных в приложении роутеров должен быть один главный, является ли роутер главным
|
||||
определяется значением аргумента `is_main` равным `True`, в методе `App().include_router()`, который по умолчанию равен
|
||||
`False`, если в приложении зарегистрирован лишь один роутер, то он неявно устанавливается главным, если
|
||||
зарегистрировано больше одного роутера, то требуется явное указание главного. При регистрации более одного
|
||||
главного роутера вызывается исключение `OnlyOneMainRouterIsAllowedException`. При регистрации более одного
|
||||
роутера и отсутствии указания главного вызывается исключение `MissingMainRouterException`
|
||||
|
||||
- В устанавливаемом паттерне сообщения описания команды необходимы быть два ключевых слова:
|
||||
`command` и `description`, каждое из которых должно быть заключено в фигурные скобки, после обработки
|
||||
паттерна на места этих ключевых слов будут подставлены соответствующие значения команды, при отсутствии
|
||||
этих двух ключевых слов будет вызвано исключение `InvalidDescriptionMessagePatternException`
|
||||
|
||||
- Команды приложения не должны повторяться, при значении атрибута `ignore_command_register` равным `True`
|
||||
допускается создание обработчиков для разных регистров одинаковых символов в команде, для примера `u` и `U`,
|
||||
при значении атрибута `ignore_command_register` класса `App` равным `False` тот же пример вызывает исключение
|
||||
`RepeatedCommandInDifferentRoutersException`. Исключение вызывается только при наличии пересекающихся команд
|
||||
у __<u>разных</u>__ роутеров
|
||||
|
||||
- У главного обработчика должен быть зарегистрирован обработчик неизвестных команд:
|
||||
```python
|
||||
router = Router()
|
||||
|
||||
@router.unknown_command
|
||||
def unknown_command(command):
|
||||
print(f'Command "{command}" undefined')
|
||||
```
|
||||
При отсутствии обработчика неизвестных команд у главного роутера будет вызвано исключение
|
||||
`MissingHandlerForUnknownCommandsException`. При регистрации обработчика неизвестных команд у
|
||||
__<u>не</u>__ главного роутера будет вызвано исключение `HandlerForUnknownCommandsOnNonMainRouterException`
|
||||
|
||||
|
||||
|
||||
|
||||
#### Исключения
|
||||
|
||||
- `InvalidRouterInstanceException` — Переданный объект в метод `App().include_router()` не является экземпляром класса `Router`.
|
||||
- `InvalidDescriptionMessagePatternException` — Неправильный формат паттерна описания команд.
|
||||
- `OnlyOneMainRouterIsAllowedException` — Регистрация более одного главного роутера.
|
||||
- `MissingMainRouterException` — Отсутствует главный роутер.
|
||||
- `MissingHandlerForUnknownCommandsException` — В основном роутере отсутствует обработчик неизвестных команд.
|
||||
- `HandlerForUnknownCommandsOnNonMainRouterException` — Обработчик неизвестных команд определён не у основного роутера.
|
||||
- `NoRegisteredRoutersException` — Отсутствуют зарегистрированные роутеры.
|
||||
- `NoRegisteredHandlersException` — У роутера нет ни одного обработчика команд.
|
||||
- `RepeatedCommandInDifferentRoutersException` — Одна и та же команда зарегистрирована в разных роутерах.
|
||||
|
||||
---
|
||||
|
||||
### *class* :: `Router`
|
||||
Класс, который определяет и конфигурирует обработчики команд
|
||||
|
||||
#### Конструктор
|
||||
```python
|
||||
Router(title: str = 'Commands group title:',
|
||||
name: str = 'subordinate')
|
||||
```
|
||||
|
||||
**Аргументы:**
|
||||
- **name : mean**
|
||||
- `title` (`str`): Заголовок группы команд.
|
||||
- `name` (`str`): Персональное название роутера
|
||||
|
||||
|
||||
|
||||
#### **declared *methods***
|
||||
|
||||
---
|
||||
|
||||
**`@`Router().**`command(command: str, description: str = None)`
|
||||
|
||||
*param* `command: str` **::** строковый триггер, который будет выполнять указанные действия
|
||||
*example* **::** `U` / `update` / `ExaMPLE`
|
||||
|
||||
*param* `description: str` **::** описание команды, которое будет выведено в консоль
|
||||
*example* **::** `description for update command` или `example description`
|
||||
|
||||
*method mean* **::** декоратор регистрирует функцию как обработчик команды
|
||||
|
||||
---
|
||||
|
||||
**`@`Router().**`unknown_command`
|
||||
|
||||
*method mean* **::** декоратор регистрирует функцию как обработчик неизвестных команд
|
||||
|
||||
---
|
||||
|
||||
**Router().**`get_name() -> str`
|
||||
|
||||
*method mean* **::** возвращает установленное название роутера
|
||||
|
||||
---
|
||||
|
||||
**Router().**`get_title() -> str`
|
||||
|
||||
*method mean* **::** возвращает установленный заголовок группы команд данного роутера
|
||||
|
||||
---
|
||||
|
||||
**Router().**`get_router_info() -> dict`
|
||||
|
||||
*method mean* **::** возвращает информацию о роутере
|
||||
|
||||
---
|
||||
|
||||
**Router().**`get_all_commands() -> list[str]`
|
||||
|
||||
*method mean* **::** возвращает все зарегистрированные команды для данного роутера
|
||||
|
||||
---
|
||||
|
||||
#### Исключения
|
||||
- `InvalidCommandInstanceException` - Переданный объект для регистрации команды не является строкой
|
||||
- `InvalidDescriptionInstanceException` - Переданный объект для регистрации описания команды не является строкой
|
||||
- `UnknownCommandHandlerHasAlreadyBeenCreatedException` - Обработчик неизвестных команд уже создан
|
||||
- `RepeatedCommandException` - Одна и та же команда зарегистрирована в одном роутере
|
||||
|
||||
@@ -1,2 +0,0 @@
|
||||
from .router import *
|
||||
from .app import *
|
||||
@@ -0,0 +1,3 @@
|
||||
from .entity import App
|
||||
from .exceptions import (InvalidDescriptionMessagePatternException,
|
||||
InvalidRouterInstanceException)
|
||||
|
||||
+187
-82
@@ -1,145 +1,250 @@
|
||||
from typing import Callable
|
||||
from inspect import getfullargspec
|
||||
import re
|
||||
|
||||
from ..command.entity import Command
|
||||
from ..router.entity import Router
|
||||
from ..command.exceptions import (UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException)
|
||||
from .exceptions import (InvalidRouterInstanceException,
|
||||
InvalidDescriptionMessagePatternException,
|
||||
OnlyOneMainRouterIsAllowedException,
|
||||
MissingMainRouterException,
|
||||
MissingHandlersForUnknownCommandsOnMainRouterException,
|
||||
HandlerForUnknownCommandsCanOnlyBeDeclaredForMainRouterException)
|
||||
NoRegisteredRoutersException,
|
||||
NoRegisteredHandlersException,
|
||||
RepeatedCommandInDifferentRoutersException,
|
||||
IncorrectNumberOfHandlerArgsException)
|
||||
|
||||
|
||||
class App:
|
||||
def __init__(self,
|
||||
prompt: str = 'Enter a command',
|
||||
exit_command: str = 'q',
|
||||
initial_message: str = '\nHello, I am Argenta\n',
|
||||
farewell_message: str = '\nGoodBye\n',
|
||||
invalid_input_flags_message: str = 'Invalid input flags',
|
||||
exit_command: str = 'Q',
|
||||
exit_command_description: str = 'Exit command',
|
||||
exit_command_title: str = 'System points:',
|
||||
ignore_exit_command_register: bool = True,
|
||||
initial_greeting: str = 'Hello',
|
||||
goodbye_message: str = 'GoodBye',
|
||||
line_separate: str = '\n',
|
||||
command_group_description_separate: str = '\n',
|
||||
ignore_command_register: bool = False,
|
||||
line_separate: str = '',
|
||||
command_group_description_separate: str = '',
|
||||
repeat_command_groups: bool = True,
|
||||
print_func: Callable[[str], None] = print) -> None:
|
||||
self.prompt = prompt
|
||||
self.print_func = print_func
|
||||
self.exit_command = exit_command
|
||||
self.exit_command_description = exit_command_description
|
||||
self.exit_command_title = exit_command_title
|
||||
self.ignore_exit_command_register = ignore_exit_command_register
|
||||
self.goodbye_message = goodbye_message
|
||||
self.initial_greeting = initial_greeting
|
||||
self.farewell_message = farewell_message
|
||||
self.initial_message = initial_message
|
||||
self.invalid_input_flags_message = invalid_input_flags_message
|
||||
self.line_separate = line_separate
|
||||
self.command_group_description_separate = command_group_description_separate
|
||||
self.ignore_command_register = ignore_command_register
|
||||
self.repeat_command_groups = repeat_command_groups
|
||||
|
||||
self.routers: list[Router] = []
|
||||
self.registered_commands: list[dict[str, str | list[dict[str, Callable[[], None] | str]] | Router]] = []
|
||||
self.main_app_router: Router | None = None
|
||||
self._description_message_pattern = '[{command}] *=*=* {description}'
|
||||
self._routers: list[Router] = []
|
||||
self._description_message_pattern: str = '[{command}] *=*=* {description}'
|
||||
self._registered_router_entities: list[dict[str, str | list[dict[str, Callable[[], None] | Command]] | Router]] = []
|
||||
self._invalid_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'Incorrect flag syntax: "{raw_command}"')
|
||||
self._repeated_input_flags_handler: Callable[[str], None] = lambda raw_command: print_func(f'Repeated input flags: "{raw_command}"')
|
||||
self._empty_input_command_handler: Callable[[], None] = lambda: print_func(f'Empty input command')
|
||||
self._unknown_command_handler: Callable[[Command], None] = lambda command: print_func(f"Unknown command: {command.get_string_entity()}")
|
||||
|
||||
|
||||
def start_polling(self) -> None:
|
||||
self.print_func(self.initial_greeting)
|
||||
self.validate_main_router()
|
||||
self._validate_number_of_routers()
|
||||
self._validate_included_routers()
|
||||
self._validate_all_router_commands()
|
||||
|
||||
while True:
|
||||
self.print_command_group_description()
|
||||
self.print_func(self.initial_message)
|
||||
|
||||
if not self.repeat_command_groups:
|
||||
self._print_command_group_description()
|
||||
self.print_func(self.prompt)
|
||||
|
||||
command: str = input()
|
||||
while True:
|
||||
if self.repeat_command_groups:
|
||||
self._print_command_group_description()
|
||||
self.print_func(self.prompt)
|
||||
|
||||
self.checking_command_for_exit_command(command)
|
||||
self.print_func(self.line_separate)
|
||||
raw_command: str = input()
|
||||
|
||||
is_unknown_command: bool = self.check_is_command_unknown(command)
|
||||
try:
|
||||
input_command: Command = Command.parse_input_command(raw_command=raw_command)
|
||||
except UnprocessedInputFlagException:
|
||||
self.print_func(self.line_separate)
|
||||
self._invalid_input_flags_handler(raw_command)
|
||||
self.print_func(self.line_separate)
|
||||
|
||||
if is_unknown_command:
|
||||
if not self.repeat_command_groups:
|
||||
self.print_func(self.prompt)
|
||||
continue
|
||||
|
||||
for router in self.routers:
|
||||
router.input_command_handler(command)
|
||||
except RepeatedInputFlagsException:
|
||||
self.print_func(self.line_separate)
|
||||
self._repeated_input_flags_handler(raw_command)
|
||||
self.print_func(self.line_separate)
|
||||
|
||||
if not self.repeat_command_groups:
|
||||
self.print_func(self.prompt)
|
||||
continue
|
||||
|
||||
except EmptyInputCommandException:
|
||||
self.print_func(self.line_separate)
|
||||
self._empty_input_command_handler()
|
||||
self.print_func(self.line_separate)
|
||||
|
||||
if not self.repeat_command_groups:
|
||||
self.print_func(self.prompt)
|
||||
continue
|
||||
|
||||
self._check_command_for_exit_command(input_command.get_string_entity())
|
||||
|
||||
self.print_func(self.line_separate)
|
||||
is_unknown_command: bool = self._check_is_command_unknown(input_command)
|
||||
|
||||
if is_unknown_command:
|
||||
self.print_func(self.line_separate)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
if not self.repeat_command_groups:
|
||||
self.print_func(self.prompt)
|
||||
continue
|
||||
|
||||
for router in self._routers:
|
||||
router.input_command_handler(input_command)
|
||||
|
||||
self.print_func(self.line_separate)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
if not self.repeat_command_groups:
|
||||
self.print_func(self.prompt)
|
||||
|
||||
|
||||
def set_initial_greeting(self, greeting: str) -> None:
|
||||
self.initial_greeting = greeting
|
||||
def set_initial_message(self, message: str) -> None:
|
||||
self.initial_message: str = message
|
||||
|
||||
|
||||
def set_goodbye_message(self, message: str) -> None:
|
||||
self.goodbye_message = message
|
||||
def set_farewell_message(self, message: str) -> None:
|
||||
self.farewell_message: str = message
|
||||
|
||||
|
||||
def set_description_message_pattern(self, pattern: str) -> None:
|
||||
try:
|
||||
pattern.format(command='command',
|
||||
description='description')
|
||||
except KeyError:
|
||||
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)
|
||||
self._description_message_pattern = pattern
|
||||
|
||||
|
||||
def validate_main_router(self):
|
||||
if not self.main_app_router:
|
||||
raise MissingMainRouterException()
|
||||
|
||||
if not self.main_app_router.unknown_command_func:
|
||||
raise MissingHandlersForUnknownCommandsOnMainRouterException()
|
||||
|
||||
for router in self.routers:
|
||||
if router.unknown_command_func and self.main_app_router is not router:
|
||||
raise HandlerForUnknownCommandsCanOnlyBeDeclaredForMainRouterException()
|
||||
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 checking_command_for_exit_command(self, command: str):
|
||||
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 include_router(self, router: Router) -> None:
|
||||
if not isinstance(router, Router):
|
||||
raise InvalidRouterInstanceException()
|
||||
|
||||
router.set_ignore_command_register(self.ignore_command_register)
|
||||
self._routers.append(router)
|
||||
|
||||
command_entities: list[dict[str, Callable[[], None] | Command]] = router.get_command_entities()
|
||||
self._registered_router_entities.append({'name': router.get_name(),
|
||||
'title': router.get_title(),
|
||||
'entity': router,
|
||||
'commands': command_entities})
|
||||
|
||||
|
||||
def _validate_number_of_routers(self) -> None:
|
||||
if not self._routers:
|
||||
raise NoRegisteredRoutersException()
|
||||
|
||||
|
||||
def _validate_included_routers(self) -> None:
|
||||
for router in self._routers:
|
||||
if not router.get_command_entities():
|
||||
raise NoRegisteredHandlersException(router.get_name())
|
||||
|
||||
|
||||
def _validate_all_router_commands(self) -> None:
|
||||
for idx in range(len(self._registered_router_entities)):
|
||||
current_router: Router = self._registered_router_entities[idx]['entity']
|
||||
routers_without_current_router = self._registered_router_entities.copy()
|
||||
routers_without_current_router.pop(idx)
|
||||
|
||||
current_router_all_commands: list[str] = current_router.get_all_commands()
|
||||
|
||||
for router_entity in routers_without_current_router:
|
||||
if len(set(current_router_all_commands).intersection(set(router_entity['entity'].get_all_commands()))) > 0:
|
||||
raise RepeatedCommandInDifferentRoutersException()
|
||||
if self.ignore_command_register:
|
||||
if len(set([x.lower() for x in current_router_all_commands]).intersection(set([x.lower() for x in router_entity['entity'].get_all_commands()]))) > 0:
|
||||
raise RepeatedCommandInDifferentRoutersException()
|
||||
|
||||
|
||||
def _check_command_for_exit_command(self, command: str):
|
||||
if command.lower() == self.exit_command.lower():
|
||||
if self.ignore_exit_command_register:
|
||||
self.print_func(self.goodbye_message)
|
||||
self.print_func(self.farewell_message)
|
||||
exit(0)
|
||||
else:
|
||||
if command == self.exit_command:
|
||||
self.print_func(self.goodbye_message)
|
||||
self.print_func(self.farewell_message)
|
||||
exit(0)
|
||||
|
||||
|
||||
def check_is_command_unknown(self, command: str):
|
||||
registered_commands = self.registered_commands
|
||||
for router in registered_commands:
|
||||
for command_entity in router['commands']:
|
||||
if command_entity['command'].lower() == command.lower():
|
||||
if router['router'].ignore_command_register:
|
||||
def _check_is_command_unknown(self, command: Command):
|
||||
registered_router_entities: list[dict[str, str | list[dict[str, Callable[[], None] | Command]] | Router]] = self._registered_router_entities
|
||||
for router_entity in registered_router_entities:
|
||||
for command_entity in router_entity['commands']:
|
||||
if command_entity['command'].get_string_entity().lower() == command.get_string_entity().lower():
|
||||
if self.ignore_command_register:
|
||||
return False
|
||||
else:
|
||||
if command_entity['command'] == command:
|
||||
if command_entity['command'].get_string_entity() == command.get_string_entity():
|
||||
return False
|
||||
self.main_app_router.unknown_command_handler(command)
|
||||
self.print_func(self.line_separate)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
self._unknown_command_handler(command)
|
||||
return True
|
||||
|
||||
|
||||
def print_command_group_description(self):
|
||||
for router in self.registered_commands:
|
||||
self.print_func(router['name'])
|
||||
for command_entity in router['commands']:
|
||||
def _print_command_group_description(self):
|
||||
for router_entity in self._registered_router_entities:
|
||||
self.print_func(router_entity['title'])
|
||||
for command_entity in router_entity['commands']:
|
||||
self.print_func(self._description_message_pattern.format(
|
||||
command=command_entity['command'],
|
||||
description=command_entity['description']
|
||||
command=command_entity['command'].get_string_entity(),
|
||||
description=command_entity['command'].get_description()
|
||||
)
|
||||
)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
|
||||
|
||||
def include_router(self, router: Router, is_main: bool = False) -> None:
|
||||
if not isinstance(router, Router):
|
||||
raise InvalidRouterInstanceException()
|
||||
|
||||
if is_main:
|
||||
if not self.main_app_router:
|
||||
self.main_app_router = router
|
||||
router.set_router_as_main()
|
||||
else:
|
||||
raise OnlyOneMainRouterIsAllowedException(router)
|
||||
|
||||
self.routers.append(router)
|
||||
|
||||
registered_commands: list[dict[str, Callable[[], None] | str]] = router.get_registered_commands()
|
||||
self.registered_commands.append({'name': router.get_name(),
|
||||
'router': router,
|
||||
'commands': registered_commands})
|
||||
self.print_func(self.exit_command_title)
|
||||
self.print_func(self._description_message_pattern.format(
|
||||
command=self.exit_command,
|
||||
description=self.exit_command_description
|
||||
)
|
||||
)
|
||||
self.print_func(self.command_group_description_separate)
|
||||
|
||||
|
||||
+10
-14
@@ -13,27 +13,23 @@ class InvalidDescriptionMessagePatternException(Exception):
|
||||
f"Your pattern: {self.pattern}")
|
||||
|
||||
|
||||
class OnlyOneMainRouterIsAllowedException(Exception):
|
||||
def __init__(self, existing_main_router):
|
||||
self.existing_main_router = existing_main_router
|
||||
|
||||
class NoRegisteredRoutersException(Exception):
|
||||
def __str__(self):
|
||||
return ("Only One Main Router Allowed\n"
|
||||
f"Existing main router is: {self.existing_main_router}")
|
||||
return "No Registered Router Found"
|
||||
|
||||
|
||||
class MissingMainRouterException(Exception):
|
||||
class NoRegisteredHandlersException(Exception):
|
||||
def __init__(self, router_name):
|
||||
self.router_name = router_name
|
||||
def __str__(self):
|
||||
return ("Missing Main Router\n"
|
||||
"One of the registered routers must be the main one")
|
||||
return f"No Registered Handlers Found For '{self.router_name}'"
|
||||
|
||||
|
||||
class MissingHandlersForUnknownCommandsOnMainRouterException(Exception):
|
||||
class RepeatedCommandInDifferentRoutersException(Exception):
|
||||
def __str__(self):
|
||||
return ("Missing Handlers For Unknown Commands On The Main Router\n"
|
||||
"The main router must have a declared handler for unknown commands")
|
||||
return "Commands in different handlers cannot be repeated"
|
||||
|
||||
|
||||
class HandlerForUnknownCommandsCanOnlyBeDeclaredForMainRouterException(Exception):
|
||||
class IncorrectNumberOfHandlerArgsException(Exception):
|
||||
def __str__(self):
|
||||
return '\nThe handler for unknown commands can only be declared for the main router'
|
||||
return "Incorrect Input Flags Handler has incorrect number of arguments"
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
from .entity import Command
|
||||
@@ -0,0 +1,121 @@
|
||||
from .params.flag.entity import Flag
|
||||
from .params.flag.flags_group.entity import FlagsGroup
|
||||
from .exceptions import (InvalidCommandInstanceException,
|
||||
InvalidDescriptionInstanceException,
|
||||
InvalidFlagsInstanceException,
|
||||
UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException)
|
||||
|
||||
from typing import Generic, TypeVar
|
||||
|
||||
|
||||
T = TypeVar('T')
|
||||
|
||||
|
||||
class Command(Generic[T]):
|
||||
def __init__(self, command: str,
|
||||
description: str = None,
|
||||
flags: Flag | FlagsGroup | None = None):
|
||||
self._command = command
|
||||
self._description = f'description for "{self._command}" command' if not description else description
|
||||
self._registered_flags: FlagsGroup | None = flags if isinstance(flags, FlagsGroup) else FlagsGroup([flags]) if isinstance(flags, Flag) else flags
|
||||
|
||||
self._input_flags: FlagsGroup | None = None
|
||||
|
||||
|
||||
def get_string_entity(self):
|
||||
return self._command
|
||||
|
||||
|
||||
def get_description(self):
|
||||
return self._description
|
||||
|
||||
|
||||
def get_registered_flags(self):
|
||||
return self._registered_flags
|
||||
|
||||
|
||||
def validate_commands_params(self):
|
||||
if not isinstance(self._command, str):
|
||||
raise InvalidCommandInstanceException(self._command)
|
||||
if not isinstance(self._description, str):
|
||||
raise InvalidDescriptionInstanceException()
|
||||
if not any([(isinstance(self._registered_flags, FlagsGroup)), not self._registered_flags]):
|
||||
raise InvalidFlagsInstanceException
|
||||
|
||||
|
||||
def validate_input_flag(self, flag: Flag):
|
||||
registered_flags: FlagsGroup | None = self.get_registered_flags()
|
||||
if registered_flags:
|
||||
if isinstance(registered_flags, Flag):
|
||||
if registered_flags.get_string_entity() == flag.get_string_entity():
|
||||
is_valid = registered_flags.validate_input_flag_value(flag.get_value())
|
||||
if is_valid:
|
||||
return True
|
||||
else:
|
||||
for registered_flag in registered_flags:
|
||||
if registered_flag.get_string_entity() == flag.get_string_entity():
|
||||
is_valid = registered_flag.validate_input_flag_value(flag.get_value())
|
||||
if is_valid:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def set_input_flags(self, input_flags: FlagsGroup):
|
||||
self._input_flags = input_flags
|
||||
|
||||
def get_input_flags(self) -> FlagsGroup:
|
||||
return self._input_flags
|
||||
|
||||
@staticmethod
|
||||
def parse_input_command(raw_command: str) -> 'Command[T]':
|
||||
if not raw_command:
|
||||
raise EmptyInputCommandException()
|
||||
list_of_tokens = raw_command.split()
|
||||
command = list_of_tokens[0]
|
||||
list_of_tokens.pop(0)
|
||||
|
||||
flags: FlagsGroup = FlagsGroup()
|
||||
current_flag_name = None
|
||||
current_flag_value = None
|
||||
for _ in list_of_tokens:
|
||||
if _.startswith('-'):
|
||||
flag_prefix_last_symbol_index = _.rfind('-')
|
||||
if current_flag_name or len(_) < 2 or len(_[:flag_prefix_last_symbol_index]) > 3:
|
||||
raise UnprocessedInputFlagException()
|
||||
else:
|
||||
current_flag_name = _
|
||||
else:
|
||||
if not current_flag_name:
|
||||
raise UnprocessedInputFlagException()
|
||||
else:
|
||||
current_flag_value = _
|
||||
if current_flag_name and current_flag_value:
|
||||
flag_prefix_last_symbol_index = current_flag_name.rfind('-')
|
||||
flag_prefix = current_flag_name[:flag_prefix_last_symbol_index]
|
||||
flag_name = current_flag_name[flag_prefix_last_symbol_index:]
|
||||
|
||||
input_flag = Flag(flag_name=flag_name,
|
||||
flag_prefix=flag_prefix)
|
||||
input_flag.set_value(current_flag_value)
|
||||
|
||||
all_flags = [x.get_string_entity() for x in flags.get_flags()]
|
||||
if input_flag.get_string_entity() not in all_flags:
|
||||
flags.add_flag(input_flag)
|
||||
else:
|
||||
raise RepeatedInputFlagsException(input_flag)
|
||||
|
||||
current_flag_name = None
|
||||
current_flag_value = None
|
||||
if any([current_flag_name, current_flag_value]):
|
||||
raise UnprocessedInputFlagException()
|
||||
if len(flags.get_flags()) == 0:
|
||||
return Command(command=command)
|
||||
else:
|
||||
input_command = Command(command=command)
|
||||
input_command.set_input_flags(flags)
|
||||
return input_command
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
from .params.flag.entity import Flag
|
||||
|
||||
|
||||
class InvalidCommandInstanceException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Command Instance"
|
||||
|
||||
|
||||
class InvalidDescriptionInstanceException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Description Instance"
|
||||
|
||||
|
||||
class InvalidFlagsInstanceException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Flags Instance"
|
||||
|
||||
|
||||
class UnprocessedInputFlagException(Exception):
|
||||
def __str__(self):
|
||||
return "Unprocessed Input Flags"
|
||||
|
||||
|
||||
class RepeatedInputFlagsException(Exception):
|
||||
def __init__(self, flag: Flag):
|
||||
self.flag = flag
|
||||
def __str__(self):
|
||||
return ("Repeated Input Flags\n"
|
||||
f"Duplicate flag was detected in the input: '{self.flag.get_string_entity()}'")
|
||||
|
||||
|
||||
class EmptyInputCommandException(Exception):
|
||||
def __str__(self):
|
||||
return "Input Command is empty"
|
||||
@@ -0,0 +1,2 @@
|
||||
from .entity import Flag
|
||||
from .flags_group.entity import FlagsGroup
|
||||
@@ -0,0 +1,51 @@
|
||||
from typing import Literal, Pattern
|
||||
|
||||
|
||||
class Flag:
|
||||
def __init__(self, flag_name: str,
|
||||
flag_prefix: Literal['-', '--', '---'] = '-',
|
||||
ignore_flag_value_register: bool = False,
|
||||
possible_flag_values: list[str] | Pattern[str] = False):
|
||||
self._flag_name = flag_name
|
||||
self._flag_prefix = flag_prefix
|
||||
self.possible_flag_values = possible_flag_values
|
||||
self.ignore_flag_value_register = ignore_flag_value_register
|
||||
|
||||
self._value = None
|
||||
|
||||
def get_string_entity(self):
|
||||
string_entity: str = self._flag_prefix + self._flag_name
|
||||
return string_entity
|
||||
|
||||
def get_flag_name(self):
|
||||
return self._flag_name
|
||||
|
||||
def get_flag_prefix(self):
|
||||
return self._flag_prefix
|
||||
|
||||
def get_value(self):
|
||||
return self._value
|
||||
|
||||
def set_value(self, value):
|
||||
self._value = value
|
||||
|
||||
def validate_input_flag_value(self, input_flag_value: str):
|
||||
if isinstance(self.possible_flag_values, Pattern):
|
||||
is_valid = bool(self.possible_flag_values.match(input_flag_value))
|
||||
if bool(is_valid):
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
if isinstance(self.possible_flag_values, list):
|
||||
if self.ignore_flag_value_register:
|
||||
if input_flag_value.lower() in [x.lower() for x in self.possible_flag_values]:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
else:
|
||||
if input_flag_value in self.possible_flag_values:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
return True
|
||||
@@ -0,0 +1,24 @@
|
||||
from argenta.command.params.flag.entity import Flag
|
||||
|
||||
|
||||
class FlagsGroup:
|
||||
def __init__(self, flags: list[Flag] = None):
|
||||
self._flags: list[Flag] = [] if not flags else flags
|
||||
|
||||
def get_flags(self):
|
||||
return self._flags
|
||||
|
||||
def add_flag(self, flag: Flag):
|
||||
self._flags.append(flag)
|
||||
|
||||
def add_flags(self, flags: list[Flag]):
|
||||
self._flags.extend(flags)
|
||||
|
||||
def __iter__(self):
|
||||
return iter(self._flags)
|
||||
|
||||
def __next__(self):
|
||||
return next(iter(self))
|
||||
|
||||
def __getitem__(self, item):
|
||||
return self._flags[item]
|
||||
@@ -0,0 +1,2 @@
|
||||
from .entity import Router
|
||||
from .exceptions import InvalidDescriptionInstanceException
|
||||
+100
-49
@@ -1,70 +1,121 @@
|
||||
from typing import Callable, Any
|
||||
from ..router.exceptions import (InvalidCommandInstanceException,
|
||||
UnknownCommandHandlerHasAlreadyBeenCreatedException,
|
||||
InvalidDescriptionInstanceException)
|
||||
from inspect import getfullargspec
|
||||
|
||||
from ..command.entity import Command
|
||||
from ..command.params.flag.entity import Flag
|
||||
from ..command.params.flag.flags_group.entity import FlagsGroup
|
||||
from ..router.exceptions import (RepeatedCommandException, RepeatedFlagNameException,
|
||||
TooManyTransferredArgsException,
|
||||
RequiredArgumentNotPassedException,
|
||||
IncorrectNumberOfHandlerArgsException)
|
||||
|
||||
|
||||
class Router:
|
||||
def __init__(self,
|
||||
name: str,
|
||||
ignore_command_register: bool = False):
|
||||
title: str = 'Commands group title:',
|
||||
name: str = 'subordinate'):
|
||||
|
||||
self.ignore_command_register: bool = ignore_command_register
|
||||
self._name = name
|
||||
self.title = title
|
||||
self.name = name
|
||||
|
||||
self.processed_commands: list[dict[str, Callable[[], None] | str]] = []
|
||||
self.unknown_command_func: Callable[[str], None] | None = None
|
||||
self._is_main_router: bool = False
|
||||
self._command_entities: list[dict[str, Callable[[], None] | Command]] = []
|
||||
self._ignore_command_register: bool = False
|
||||
|
||||
self._not_valid_flag_handler: Callable[[Flag], None] = lambda flag: print(f"Undefined or incorrect input flag: '{flag.get_string_entity()} {flag.get_value()}'")
|
||||
|
||||
|
||||
def command(self, command: str, description: str) -> Callable[[Any], Any]:
|
||||
if not isinstance(command, str):
|
||||
raise InvalidCommandInstanceException()
|
||||
if not isinstance(description, str):
|
||||
raise InvalidDescriptionInstanceException()
|
||||
def command(self, command: Command) -> Callable[[Any], Any]:
|
||||
command.validate_commands_params()
|
||||
self._validate_command(command)
|
||||
|
||||
def command_decorator(func):
|
||||
Router._validate_func_args(command, func)
|
||||
self._command_entities.append({'handler_func': func,
|
||||
'command': command})
|
||||
def wrapper(*args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
return wrapper
|
||||
|
||||
return command_decorator
|
||||
|
||||
def set_invalid_input_flag_handler(self, func):
|
||||
processed_args = getfullargspec(func).args
|
||||
if len(processed_args) != 1:
|
||||
raise IncorrectNumberOfHandlerArgsException()
|
||||
else:
|
||||
def command_decorator(func):
|
||||
self.processed_commands.append({'func': func,
|
||||
'command': command,
|
||||
'description': description})
|
||||
def wrapper(*args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
return wrapper
|
||||
return command_decorator
|
||||
self._not_valid_flag_handler = func
|
||||
|
||||
|
||||
def unknown_command(self, func):
|
||||
if self.unknown_command_func is not None:
|
||||
raise UnknownCommandHandlerHasAlreadyBeenCreatedException()
|
||||
|
||||
self.unknown_command_func = func
|
||||
|
||||
def wrapper(*args, **kwargs):
|
||||
return func(*args, **kwargs)
|
||||
return wrapper
|
||||
|
||||
|
||||
def input_command_handler(self, input_command):
|
||||
for command_entity in self.processed_commands:
|
||||
if input_command.lower() == command_entity['command'].lower():
|
||||
if self.ignore_command_register:
|
||||
return command_entity['func']()
|
||||
def input_command_handler(self, input_command: Command):
|
||||
input_command_name: str = input_command.get_string_entity()
|
||||
input_command_flags: FlagsGroup = input_command.get_input_flags()
|
||||
for command_entity in self._command_entities:
|
||||
if input_command_name.lower() == command_entity['command'].get_string_entity().lower():
|
||||
if command_entity['command'].get_registered_flags():
|
||||
if input_command_flags:
|
||||
for flag in input_command_flags:
|
||||
is_valid = command_entity['command'].validate_input_flag(flag)
|
||||
if not is_valid:
|
||||
self._not_valid_flag_handler(flag)
|
||||
return
|
||||
return command_entity['handler_func'](input_command_flags)
|
||||
else:
|
||||
return command_entity['handler_func'](FlagsGroup(None))
|
||||
else:
|
||||
if input_command == command_entity['command']:
|
||||
return command_entity['func']()
|
||||
|
||||
def unknown_command_handler(self, unknown_command):
|
||||
self.unknown_command_func(unknown_command)
|
||||
if input_command_flags:
|
||||
self._not_valid_flag_handler(input_command_flags[0])
|
||||
return
|
||||
else:
|
||||
return command_entity['handler_func']()
|
||||
|
||||
|
||||
def set_router_as_main(self):
|
||||
self._is_main_router = True
|
||||
def _validate_command(self, command: Command):
|
||||
command_name: str = command.get_string_entity()
|
||||
if command_name in self.get_all_commands():
|
||||
raise RepeatedCommandException()
|
||||
if self._ignore_command_register:
|
||||
if command_name.lower() in [x.lower() for x in self.get_all_commands()]:
|
||||
raise RepeatedCommandException()
|
||||
|
||||
flags: FlagsGroup = command.get_registered_flags()
|
||||
if flags:
|
||||
flags_name: list = [x.get_string_entity().lower() for x in flags]
|
||||
if len(set(flags_name)) < len(flags_name):
|
||||
raise RepeatedFlagNameException()
|
||||
|
||||
|
||||
def get_registered_commands(self) -> list[dict[str, Callable[[], None] | str]]:
|
||||
return self.processed_commands
|
||||
@staticmethod
|
||||
def _validate_func_args(command: Command, func: Callable):
|
||||
registered_args = command.get_registered_flags()
|
||||
transferred_args = getfullargspec(func).args
|
||||
if registered_args and transferred_args:
|
||||
if len(transferred_args) != 1:
|
||||
raise TooManyTransferredArgsException()
|
||||
elif registered_args and not transferred_args:
|
||||
raise RequiredArgumentNotPassedException()
|
||||
elif not registered_args and transferred_args:
|
||||
raise TooManyTransferredArgsException()
|
||||
|
||||
|
||||
def set_ignore_command_register(self, ignore_command_register: bool):
|
||||
self._ignore_command_register = ignore_command_register
|
||||
|
||||
|
||||
def get_command_entities(self) -> list[dict[str, Callable[[], None] | Command]]:
|
||||
return self._command_entities
|
||||
|
||||
|
||||
def get_name(self) -> str:
|
||||
return self._name
|
||||
return self.name
|
||||
|
||||
|
||||
def get_title(self) -> str:
|
||||
return self.title
|
||||
|
||||
|
||||
def get_all_commands(self) -> list[str]:
|
||||
all_commands: list[str] = []
|
||||
for command_entity in self._command_entities:
|
||||
all_commands.append(command_entity['command'].get_string_entity())
|
||||
|
||||
return all_commands
|
||||
|
||||
@@ -1,13 +1,33 @@
|
||||
class InvalidCommandInstanceException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Command Instance"
|
||||
|
||||
|
||||
class InvalidDescriptionInstanceException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Description Instance"
|
||||
|
||||
|
||||
class UnknownCommandHandlerHasAlreadyBeenCreatedException(Exception):
|
||||
class RepeatedCommandException(Exception):
|
||||
def __str__(self):
|
||||
return "Only one unknown command handler can be declared"
|
||||
return "Commands in handler cannot be repeated"
|
||||
|
||||
|
||||
class RepeatedFlagNameException(Exception):
|
||||
def __str__(self):
|
||||
return "Repeated flag name in register command"
|
||||
|
||||
|
||||
class TooManyTransferredArgsException(Exception):
|
||||
def __str__(self):
|
||||
return "Too many transferred arguments"
|
||||
|
||||
|
||||
class RequiredArgumentNotPassedException(Exception):
|
||||
def __str__(self):
|
||||
return "Required argument not passed"
|
||||
|
||||
|
||||
class NotValidInputFlagHandlerHasBeenAlreadyCreatedException(Exception):
|
||||
def __str__(self):
|
||||
return "Invalid Input Flag Handler has already been created"
|
||||
|
||||
|
||||
class IncorrectNumberOfHandlerArgsException(Exception):
|
||||
def __str__(self):
|
||||
return "Incorrect Input Flags Handler has incorrect number of arguments"
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
import re
|
||||
|
||||
|
||||
def set_description_message_pattern(pattern: str) -> None:
|
||||
first_check = re.match(r'.*command.*', pattern)
|
||||
second_check = re.match(r'.*{description}.*', pattern)
|
||||
|
||||
|
||||
set_description_message_pattern('Invalid des{ommand}cription pattern')
|
||||
@@ -0,0 +1,75 @@
|
||||
import os
|
||||
import shutil
|
||||
import time
|
||||
from zipfile import ZipFile
|
||||
import requests
|
||||
from ..local_data_func.get_script_release_tag import get_script_tag
|
||||
from tqdm import tqdm
|
||||
|
||||
|
||||
class UpdateScript:
|
||||
|
||||
GITHUB_REPO = "https://api.github.com/repos/koloideal/WordMath"
|
||||
|
||||
@staticmethod
|
||||
def get_latest_release() -> dict:
|
||||
response = requests.get(f"{UpdateScript.GITHUB_REPO}/releases/latest")
|
||||
data = response.json()
|
||||
return {'tag': data['tag_name'],
|
||||
'url': data['zipball_url']}
|
||||
|
||||
|
||||
@staticmethod
|
||||
def download_new_release(zip_url):
|
||||
response = requests.get(zip_url, stream=True)
|
||||
response.raise_for_status()
|
||||
|
||||
total_size = int(response.headers.get("content-length", 0))
|
||||
block_size = 1024
|
||||
|
||||
with tqdm(total=total_size, unit="B", unit_scale=True) as progress_bar:
|
||||
with open('new_release.zip', "wb") as file:
|
||||
for data in response.iter_content(block_size):
|
||||
progress_bar.update(len(data))
|
||||
file.write(data)
|
||||
time.sleep(0.3)
|
||||
|
||||
with ZipFile('new_release.zip') as zip_file:
|
||||
zip_file.extractall("new_release")
|
||||
|
||||
|
||||
@staticmethod
|
||||
def upgrade_script():
|
||||
excluded_files = ['venv', '.venv', '.git', 'new_release']
|
||||
for obj in os.listdir():
|
||||
if obj not in excluded_files:
|
||||
if os.path.isfile(obj):
|
||||
os.remove(obj)
|
||||
elif os.path.isdir(obj):
|
||||
shutil.rmtree(obj)
|
||||
|
||||
new_release = os.listdir('new_release')
|
||||
new_release_name = new_release[0] if new_release[0].startswith('koloideal-WordMath') else None
|
||||
path = f'new_release/{new_release_name}'
|
||||
all_files = os.listdir(path)
|
||||
for file in all_files:
|
||||
shutil.move(path + '/' + file, './' + file)
|
||||
|
||||
shutil.rmtree('new_release')
|
||||
|
||||
|
||||
@staticmethod
|
||||
def start_update() -> bool:
|
||||
existing_release_tag: str = get_script_tag()
|
||||
latest_release: dict = UpdateScript.get_latest_release()
|
||||
latest_release_tag = latest_release['tag']
|
||||
latest_release_url = latest_release['url']
|
||||
|
||||
if latest_release_tag != existing_release_tag:
|
||||
#UpdateScript.download_new_release(latest_release_url)
|
||||
#UpdateScript.upgrade_script()
|
||||
return latest_release_tag
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
from numpy import ndarray
|
||||
from word2number import w2n
|
||||
from ..local_data_func.get_operator_synonyms import get_operator_synonyms
|
||||
import numexpr
|
||||
|
||||
|
||||
def word2num_math(string: str) -> ndarray | ZeroDivisionError | OverflowError:
|
||||
operator_synonyms: dict[str, list[str]] = get_operator_synonyms()
|
||||
|
||||
def variables_to_operator(synonym):
|
||||
for key in operator_synonyms.keys():
|
||||
if synonym in operator_synonyms[key]:
|
||||
return key
|
||||
|
||||
action = {
|
||||
"plus": '+',
|
||||
"minus": '-',
|
||||
"divide": '/',
|
||||
"multiply": '*',
|
||||
"degree": '**',
|
||||
}
|
||||
|
||||
result_string: str = ''
|
||||
all_variables_of_operators: list[str] = [x for l in operator_synonyms.values() for x in l]
|
||||
operators = {}
|
||||
|
||||
while True:
|
||||
is_clear = True
|
||||
number_of_operator = 1
|
||||
for word in string.split():
|
||||
try:
|
||||
if word in all_variables_of_operators:
|
||||
ope_index = string.index(word)
|
||||
if ope_index in operators.keys():
|
||||
is_clear = True
|
||||
break
|
||||
else:
|
||||
raise ValueError
|
||||
except ValueError:
|
||||
continue
|
||||
else:
|
||||
is_clear = False
|
||||
operators[number_of_operator] = variables_to_operator(word)
|
||||
number_of_operator += 1
|
||||
string = string.replace(word, "&&", 1)
|
||||
if is_clear:
|
||||
break
|
||||
|
||||
num_of_ope = 1
|
||||
for number in string.split("&&"):
|
||||
try:
|
||||
int_num = w2n.word_to_num(number.strip())
|
||||
except ValueError:
|
||||
return ValueError("Invalid input expression")
|
||||
if num_of_ope in operators.keys():
|
||||
result_string += f'{int_num} {action[operators[num_of_ope]]} '
|
||||
num_of_ope += 1
|
||||
else:
|
||||
result_string += f'{int_num}'
|
||||
|
||||
try:
|
||||
result = numexpr.evaluate(result_string)
|
||||
except ZeroDivisionError:
|
||||
return ZeroDivisionError('Except divide by zero')
|
||||
except OverflowError:
|
||||
return OverflowError('Too big result')
|
||||
else:
|
||||
return result
|
||||
@@ -0,0 +1,10 @@
|
||||
from rich.console import Console
|
||||
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
def help_command():
|
||||
console.print("[italic bold]The main functionality of the script is to convert an expression from a string "
|
||||
"to a mathematical one and then calculate this expression. "
|
||||
"Project GitHub: https://github.com/koloideal/WordMath[/italic bold]")
|
||||
@@ -0,0 +1,20 @@
|
||||
from rich.console import Console
|
||||
from ...business_logic.word2num_math import word2num_math
|
||||
|
||||
|
||||
console = Console()
|
||||
print_line_separator = lambda: console.print('\n[bold blue]--------------------------------------[/bold blue]\n')
|
||||
|
||||
|
||||
def start_solving_command():
|
||||
while True:
|
||||
console.print(
|
||||
"\n[italic]Enter a string expression or [bold italic green] Q [/bold italic green] for exit:[/italic]")
|
||||
string_expression = input()
|
||||
if string_expression.lower() == 'q':
|
||||
break
|
||||
else:
|
||||
print_line_separator()
|
||||
console.print(
|
||||
f'[bold green]Answer:[/bold green] [bold blue]{word2num_math(string_expression)}[/bold blue]')
|
||||
print_line_separator()
|
||||
@@ -0,0 +1,26 @@
|
||||
import requests
|
||||
from rich.console import Console
|
||||
from ...business_logic.script_updater import UpdateScript
|
||||
|
||||
|
||||
console = Console()
|
||||
print_line_separator = lambda: console.print('\n[bold green]--------------------------------------[/bold green]\n')
|
||||
|
||||
|
||||
def upgrade_command():
|
||||
try:
|
||||
requests.get('https://ya.ru')
|
||||
except requests.exceptions.ConnectionError:
|
||||
console.print('[bold red]No internet connection[/bold red]')
|
||||
else:
|
||||
latest_tag = UpdateScript.start_update()
|
||||
if latest_tag:
|
||||
print_line_separator()
|
||||
console.print(f"[bold yellow]The newest version ({latest_tag}) of the script has been successfully installed![/bold yellow]")
|
||||
print_line_separator()
|
||||
console.print("[bold yellow]Rerun the script for the changes to take effect[/bold yellow]")
|
||||
print_line_separator()
|
||||
exit(0)
|
||||
else:
|
||||
console.print('[bold red]You have the latest version installed[/bold red]')
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
import re
|
||||
|
||||
from rich.console import Console
|
||||
|
||||
from argenta.command.entity import Command
|
||||
from argenta.command.params.flag.entity import Flag
|
||||
from argenta.command.params.flag.flags_group.entity import FlagsGroup
|
||||
from argenta.router import Router
|
||||
|
||||
|
||||
work_router: Router = Router(title='Work nts:')
|
||||
settings_router: Router = Router(title='Settings points:')
|
||||
|
||||
console = Console()
|
||||
|
||||
flagi = FlagsGroup(flags=[
|
||||
Flag(flag_name='host',
|
||||
flag_prefix='--',
|
||||
possible_flag_values=re.compile(r'^192.168.\d{1,3}.\d{1,3}$')),
|
||||
Flag(flag_name='port',
|
||||
flag_prefix='--', )
|
||||
])
|
||||
|
||||
|
||||
@work_router.command(Command(command='0', description='Get Help'))
|
||||
def command_help():
|
||||
print('Help command')
|
||||
'''flags = args.get_flags()
|
||||
for flag in flags:
|
||||
print(f'name: "{flag.get_string_entity()}", value: "{flag.get_value()}"')'''
|
||||
#help_command()
|
||||
|
||||
|
||||
@work_router.command(Command(command='P', description='Start Solving', flags=flagi))
|
||||
def command_start_solving(argrrtrts: FlagsGroup | None):
|
||||
print('Solving...')
|
||||
flags = argrrtrts.get_flags()
|
||||
for flag in flags:
|
||||
print(f'name: "{flag.get_string_entity()}", value: "{flag.get_value()}"')
|
||||
#start_solving_command()
|
||||
|
||||
|
||||
@settings_router.command(Command(command='G', description='Update WordMath'))
|
||||
def command_update():
|
||||
print('uefi')
|
||||
# upgrade_command()
|
||||
|
||||
|
||||
def invalid_input_flag(flag: Flag):
|
||||
print(f'Invalid inpuuuuuuuuuuuuuuuuuuuuuuuut flag: "{flag.get_string_entity()} {flag.get_value()}"')
|
||||
|
||||
|
||||
work_router.set_invalid_input_flag_handler(invalid_input_flag)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
{
|
||||
"minus": [
|
||||
"without",
|
||||
"lacking",
|
||||
"deprived_of",
|
||||
"bereft_of",
|
||||
"destitute_of",
|
||||
"minus"
|
||||
],
|
||||
"plus": [
|
||||
"added_to",
|
||||
"add",
|
||||
"coupled_with",
|
||||
"with_the_addition_of",
|
||||
"plus"
|
||||
],
|
||||
"divide": [
|
||||
"separate",
|
||||
"part",
|
||||
"split",
|
||||
"divide",
|
||||
"divide_by"
|
||||
],
|
||||
"multiply": [
|
||||
"extend",
|
||||
"expand",
|
||||
"spread",
|
||||
"build_up",
|
||||
"accumulate",
|
||||
"augment",
|
||||
"multiply",
|
||||
"times"
|
||||
],
|
||||
"degree": [
|
||||
"stage",
|
||||
"extent",
|
||||
"grade",
|
||||
"proportion",
|
||||
"gradation",
|
||||
"to_the_power_of",
|
||||
"degree",
|
||||
"to_the"
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"tag": "v4.1.1"
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
import json
|
||||
|
||||
|
||||
def get_operator_synonyms() -> dict[str, list[str]]:
|
||||
with open("tests/mock_app/local_data/operator_synonyms.json", "r", encoding="utf-8") as file:
|
||||
operator_synonyms: dict = json.load(file)
|
||||
|
||||
return operator_synonyms
|
||||
@@ -0,0 +1,8 @@
|
||||
import json
|
||||
|
||||
|
||||
def get_script_tag() -> str:
|
||||
with open("tests/mock_app/local_data/script_release_tag.json", "r", encoding="utf-8") as file:
|
||||
script_release_tag: str = json.load(file)['tag']
|
||||
|
||||
return script_release_tag
|
||||
@@ -0,0 +1,40 @@
|
||||
from tests.mock_app.handlers.routers import work_router, settings_router
|
||||
from art import text2art
|
||||
from rich.console import Console
|
||||
|
||||
from argenta.app import App
|
||||
from argenta.router import Router
|
||||
from argenta.command import Command
|
||||
from argenta.command.params.flag import Flag, FlagsGroup
|
||||
|
||||
|
||||
app: App = App(prompt='[italic white bold]What do you want to do(enter number of action)?',
|
||||
line_separate=f'\n{"[bold green]-[/bold green][bold red]-[/bold red]"*25}\n',
|
||||
print_func=Console().print,
|
||||
command_group_description_separate='',
|
||||
repeat_command_groups=True)
|
||||
|
||||
|
||||
def main():
|
||||
ascii_name: str = text2art('WordMath', font='nancyj')
|
||||
initial_greeting: str = f'[bold red]\n\n{ascii_name}'
|
||||
|
||||
ascii_goodbye_message: str = text2art('GoodBye', font='small')
|
||||
goodbye_message: str = f'[bold red]\n{ascii_goodbye_message}{' '*12}made by kolo\n'
|
||||
|
||||
app.include_router(work_router)
|
||||
app.include_router(settings_router)
|
||||
|
||||
app.set_initial_message(initial_greeting)
|
||||
app.set_farewell_message(goodbye_message)
|
||||
|
||||
app.set_invalid_input_flags_handler(lambda raw_command: print(f"Invalid input flags: {raw_command}"))
|
||||
app.set_unknown_command_handler(lambda command: print(f"Unknown command: {command.get_string_entity()}"))
|
||||
app.set_repeated_input_flags_handler(lambda raw_command: print(f"Repeated input flags: {raw_command}"))
|
||||
|
||||
app.set_description_message_pattern('[bold red][{command}][/bold red] [blue]*=*=*[/blue] [bold yellow italic]{description}')
|
||||
|
||||
app.start_polling()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Generated
-7
@@ -1,7 +0,0 @@
|
||||
# This file is automatically @generated by Poetry 2.0.1 and should not be changed by hand.
|
||||
package = []
|
||||
|
||||
[metadata]
|
||||
lock-version = "2.1"
|
||||
python-versions = ">=3.11"
|
||||
content-hash = "f5666f5625d676c506924a57dc0520a1f3ed2b2c774baed3dc85353594f8473d"
|
||||
+24
-5
@@ -1,17 +1,36 @@
|
||||
[project]
|
||||
name = "argenta"
|
||||
version = "0.1.1"
|
||||
description = "python library for creating cli apps"
|
||||
version = "0.3.5"
|
||||
description = "python library for creating custom shells"
|
||||
authors = [
|
||||
{name = "kolo",email = "kolo.is.main@gmail.com"}
|
||||
{name = "kolo", email = "kolo.is.main@gmail.com"}
|
||||
]
|
||||
license = {text = "MIT"}
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.11"
|
||||
dependencies = [
|
||||
]
|
||||
dependencies = [] # no dependencies
|
||||
|
||||
|
||||
[tool.ruff]
|
||||
exclude = [
|
||||
".idea",
|
||||
"venv",
|
||||
".git",
|
||||
"poetry.lock",
|
||||
".__pycache__",
|
||||
"tests"
|
||||
]
|
||||
|
||||
|
||||
[build-system]
|
||||
requires = ["poetry-core>=2.0.0,<3.0.0"]
|
||||
build-backend = "poetry.core.masonry.api"
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
art = "^6.4"
|
||||
rich = "^13.9.4"
|
||||
numpy = "^2.2.2"
|
||||
word2number = "^1.1"
|
||||
numexpr = "^2.10.2"
|
||||
requests = "^2.32.3"
|
||||
|
||||
|
||||
@@ -1,23 +0,0 @@
|
||||
from setuptools import setup, find_packages
|
||||
|
||||
with open("README.md", "r", encoding="utf-8") as fh:
|
||||
long_description = fh.read()
|
||||
|
||||
setup(
|
||||
name="argenta",
|
||||
version="0.1.1",
|
||||
author="kolo",
|
||||
author_email="kolo.is.main@gmail.com",
|
||||
description="Python library for creating CLI apps",
|
||||
long_description=long_description,
|
||||
long_description_content_type="text/markdown",
|
||||
packages=find_packages(),
|
||||
install_requires=[
|
||||
"requests",
|
||||
],
|
||||
classifiers=[
|
||||
"Programming Language :: Python :: 3",
|
||||
"Operating System :: OS Independent",
|
||||
],
|
||||
python_requires='>=3.11',
|
||||
)
|
||||
@@ -1,5 +0,0 @@
|
||||
from argenta.app.entity import *
|
||||
|
||||
|
||||
def test():
|
||||
assert App().exit_command == 'q'
|
||||
@@ -0,0 +1,20 @@
|
||||
from argenta.app import App
|
||||
from argenta.app.exceptions import (InvalidDescriptionMessagePatternException,
|
||||
NoRegisteredRoutersException)
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestApp(unittest.TestCase):
|
||||
def test_set_invalid_description_message_pattern(self):
|
||||
with self.assertRaises(InvalidDescriptionMessagePatternException):
|
||||
App().set_description_message_pattern('Invalid description pattern')
|
||||
|
||||
def test_set_invalid_description_message_pattern2(self):
|
||||
with self.assertRaises(InvalidDescriptionMessagePatternException):
|
||||
App().set_description_message_pattern('Invalid {desription} description {comand} pattern')
|
||||
|
||||
def test_no_registered_router(self):
|
||||
with self.assertRaises(NoRegisteredRoutersException):
|
||||
App()._validate_number_of_routers()
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
from argenta.command import Command
|
||||
from argenta.command.params.flag import Flag, FlagsGroup
|
||||
from argenta.command.exceptions import (UnprocessedInputFlagException,
|
||||
RepeatedInputFlagsException,
|
||||
EmptyInputCommandException)
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestCommand(unittest.TestCase):
|
||||
def test_parse_correct_raw_command(self):
|
||||
self.assertEqual(Command.parse_input_command('ssh --host 192.168.0.3').get_string_entity(), 'ssh')
|
||||
|
||||
def test_parse_raw_command_with_flag_name_without_value(self):
|
||||
with self.assertRaises(UnprocessedInputFlagException):
|
||||
Command.parse_input_command('ssh --host')
|
||||
|
||||
def test_parse_raw_command_without_flag_name_with_value(self):
|
||||
with self.assertRaises(UnprocessedInputFlagException):
|
||||
Command.parse_input_command('ssh 192.168.0.3')
|
||||
|
||||
def test_parse_raw_command_with_repeated_flag_name(self):
|
||||
with self.assertRaises(RepeatedInputFlagsException):
|
||||
Command.parse_input_command('ssh --host 192.168.0.3 --host 172.198.0.43')
|
||||
|
||||
def test_parse_empty_raw_command(self):
|
||||
with self.assertRaises(EmptyInputCommandException):
|
||||
Command.parse_input_command('')
|
||||
|
||||
def test_get_command_description(self):
|
||||
self.assertEqual(Command(command='test', description='test description').get_description(), 'test description')
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
from argenta.command.params.flag import Flag
|
||||
|
||||
import unittest
|
||||
import re
|
||||
|
||||
|
||||
class TestFlag(unittest.TestCase):
|
||||
def test_get_string_entity(self):
|
||||
self.assertEqual(Flag(flag_name='test').get_string_entity(),
|
||||
'-test')
|
||||
|
||||
def test_get_string_entity2(self):
|
||||
self.assertEqual(Flag(flag_name='test',
|
||||
flag_prefix='---').get_string_entity(),
|
||||
'---test')
|
||||
|
||||
def test_get_flag_name(self):
|
||||
self.assertEqual(Flag(flag_name='test').get_flag_name(),
|
||||
'test')
|
||||
|
||||
def test_get_flag_prefix(self):
|
||||
self.assertEqual(Flag(flag_name='test').get_flag_prefix(),
|
||||
'-')
|
||||
|
||||
def test_get_flag_prefix2(self):
|
||||
self.assertEqual(Flag(flag_name='test',
|
||||
flag_prefix='--').get_flag_prefix(),
|
||||
'--')
|
||||
|
||||
def test_get_flag_value_without_set(self):
|
||||
self.assertEqual(Flag(flag_name='test').get_value(),
|
||||
None)
|
||||
|
||||
def test_get_flag_value_with_set(self):
|
||||
flag = Flag(flag_name='test')
|
||||
flag.set_value('example')
|
||||
self.assertEqual(flag.get_value(), 'example')
|
||||
|
||||
def test_validate_incorrect_flag_value_with_list_of_possible_flag_values(self):
|
||||
flag = Flag(flag_name='test', possible_flag_values=['1', '2', '3'])
|
||||
self.assertEqual(flag.validate_input_flag_value('bad value'), False)
|
||||
|
||||
def test_validate_correct_flag_value_with_list_of_possible_flag_values(self):
|
||||
flag = Flag(flag_name='test', possible_flag_values=['1', '2', '3'])
|
||||
self.assertEqual(flag.validate_input_flag_value('1'), True)
|
||||
|
||||
def test_validate_incorrect_flag_value_with_pattern_of_possible_flag_values(self):
|
||||
flag = Flag(flag_name='test', possible_flag_values=re.compile(r'192.168.\d+.\d+'))
|
||||
self.assertEqual(flag.validate_input_flag_value('152.123.9.8'), False)
|
||||
|
||||
def test_validate_correct_flag_value_with_pattern_of_possible_flag_values(self):
|
||||
flag = Flag(flag_name='test', possible_flag_values=re.compile(r'192.168.\d+.\d+'))
|
||||
self.assertEqual(flag.validate_input_flag_value('192.168.9.8'), True)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
from argenta.command.params.flag import Flag, FlagsGroup
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestFlagsGroup(unittest.TestCase):
|
||||
def test_get_flags(self):
|
||||
flags = FlagsGroup()
|
||||
list_of_flags = [
|
||||
Flag('test1'),
|
||||
Flag('test2'),
|
||||
Flag('test3'),
|
||||
]
|
||||
flags.add_flags(list_of_flags)
|
||||
self.assertEqual(flags.get_flags(),
|
||||
list_of_flags)
|
||||
|
||||
def test_add_flag(self):
|
||||
flags = FlagsGroup()
|
||||
flags.add_flag(Flag('test'))
|
||||
self.assertEqual(len(flags.get_flags()), 1)
|
||||
|
||||
def test_add_flags(self):
|
||||
flags = FlagsGroup()
|
||||
flags.add_flags([Flag('test'), Flag('test2')])
|
||||
self.assertEqual(len(flags.get_flags()), 2)
|
||||
@@ -0,0 +1,34 @@
|
||||
from argenta.command.params.flag import FlagsGroup, Flag
|
||||
from argenta.router import Router
|
||||
from argenta.command import Command
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class TestRouter(unittest.TestCase):
|
||||
def test_get_router_name(self):
|
||||
self.assertEqual(Router(name='test name').get_name(), 'test name')
|
||||
|
||||
def test_get_router_title(self):
|
||||
self.assertEqual(Router(title='test title').get_title(), 'test title')
|
||||
|
||||
def test_input_correct_command(self):
|
||||
router = Router()
|
||||
@router.command(Command(command='test'))
|
||||
def test():
|
||||
return 'correct result'
|
||||
|
||||
self.assertEqual(router.input_command_handler(Command(command='test')), 'correct result')
|
||||
|
||||
def test_input_command_with_invalid_flag(self):
|
||||
router = Router()
|
||||
router.set_invalid_input_flag_handler(lambda x: x)
|
||||
|
||||
@router.command(Command(command='test'))
|
||||
def test():
|
||||
return 'correct result'
|
||||
|
||||
input_command = Command(command='test')
|
||||
input_command.set_input_flags(FlagsGroup([Flag('host')]))
|
||||
|
||||
self.assertEqual(router.input_command_handler(input_command), None)
|
||||
Reference in New Issue
Block a user