mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 18:15:28 +03:00
Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| cbf7d3c578 | |||
| ea2d068022 | |||
| 5991851207 | |||
| f628c3b5b5 |
@@ -52,6 +52,14 @@ def main() -> None:
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
# Фичи в разработке
|
||||
|
||||
- Полноценная поддержка автокомплитера на Linux
|
||||
- Возможность настройки захвата stdout при обработке хэндлером ввода
|
||||
|
||||
## Полная [документация](https://argenta-docs.vercel.app) | MIT 2025 kolo | made by [kolo](https://t.me/kolo_id)
|
||||
|
||||
|
||||
|
||||
+78
-78
@@ -1,89 +1,89 @@
|
||||
from argenta.app import App
|
||||
from argenta.app.autocompleter import AutoCompleter
|
||||
from argenta.router import Router
|
||||
from argenta.command import Command
|
||||
from argenta.orchestrator import Orchestrator
|
||||
from argenta.app.dividing_line import DynamicDividingLine
|
||||
from argenta.response import Response
|
||||
import platform
|
||||
import psutil
|
||||
import os
|
||||
import subprocess
|
||||
import socket
|
||||
from argenta.response.status import Status
|
||||
from argenta.command.flag import Flag
|
||||
from argenta.command.flags import Flags
|
||||
from argenta.app import App
|
||||
from argenta.orchestrator import Orchestrator
|
||||
|
||||
# Маршрутизатор для работы с файлами
|
||||
file_router = Router("Файловые операции")
|
||||
# Создание маршрутизатора
|
||||
file_router = Router("Операции с файлами")
|
||||
|
||||
|
||||
@file_router.command(Command("list", "Список файлов"))
|
||||
def list_files(response: Response):
|
||||
files = os.listdir()
|
||||
for file in files:
|
||||
print(file)
|
||||
|
||||
|
||||
@file_router.command(Command("size", "Размер файла"))
|
||||
def file_size(response: Response):
|
||||
file_name = input("Введите имя файла: ")
|
||||
if os.path.exists(file_name):
|
||||
size = os.path.getsize(file_name)
|
||||
print(f"Размер файла {file_name}: {size} байт")
|
||||
else:
|
||||
print(f"Файл {file_name} не найден")
|
||||
|
||||
|
||||
# Маршрутизатор для системных операций
|
||||
system_router = Router("Системные операции")
|
||||
|
||||
|
||||
@system_router.command(Command("info", "Информация о системе"))
|
||||
def system_info(response: Response):
|
||||
print(f"Система: {platform.system()}")
|
||||
print(f"Версия: {platform.version()}")
|
||||
print(f"Архитектура: {platform.architecture()}")
|
||||
print(f"Процессор: {platform.processor()}")
|
||||
|
||||
|
||||
@system_router.command(Command("memory", "Информация о памяти"))
|
||||
def memory_info(response: Response):
|
||||
memory = psutil.virtual_memory()
|
||||
print(f"Всего памяти: {memory.total / (1024**3):.2f} ГБ")
|
||||
print(f"Доступно: {memory.available / (1024**3):.2f} ГБ")
|
||||
print(f"Использовано: {memory.used / (1024**3):.2f} ГБ ({memory.percent}%)")
|
||||
|
||||
|
||||
# Маршрутизатор для сетевых операций
|
||||
network_router = Router("Сетевые операции")
|
||||
|
||||
|
||||
@network_router.command(Command("ping", "Проверка доступности хоста"))
|
||||
def ping_host(response: Response):
|
||||
host = input("Введите имя хоста: ")
|
||||
print(f"Пингую {host}...")
|
||||
subprocess.run(["ping", "-c", "4", host])
|
||||
|
||||
|
||||
@network_router.command(Command("ip", "Показать IP-адреса"))
|
||||
def show_ip(response: Response):
|
||||
hostname = socket.gethostname()
|
||||
print(f"Имя хоста: {hostname}")
|
||||
print(f"IP-адрес: {socket.gethostbyname(hostname)}")
|
||||
|
||||
|
||||
# Создание приложения и регистрация маршрутизаторов
|
||||
app = App(
|
||||
prompt="System> ",
|
||||
initial_message="Pingator",
|
||||
dividing_line=DynamicDividingLine("*"),
|
||||
autocompleter=AutoCompleter(".hist", "e"),
|
||||
# Определение флагов для команды копирования
|
||||
copy_flags = Flags(
|
||||
Flag('source', '--'),
|
||||
Flag('destination', '--'),
|
||||
Flag('recursive', '--', False), # Булевый флаг без значения
|
||||
Flag('force', '-', False) # Короткий булевый флаг
|
||||
)
|
||||
@file_router.command(Command('case', aliases=['cp', 'ch']))
|
||||
def handler(response: Response):
|
||||
print('test')
|
||||
|
||||
# Добавляем все маршрутизаторы
|
||||
app.include_routers(file_router, system_router, network_router)
|
||||
# Регистрация команды копирования
|
||||
@file_router.command(Command(
|
||||
trigger="ch",
|
||||
description="Копирование файлов",
|
||||
flags=copy_flags,
|
||||
aliases=["cp"]
|
||||
))
|
||||
def copy_files(response: Response):
|
||||
# Получаем значения корректных флагов
|
||||
source = None
|
||||
destination = None
|
||||
recursive = False
|
||||
force = False
|
||||
|
||||
# Добавляем сообщение при запуске
|
||||
app.add_message_on_startup("Для просмотра доступных команд нажмите Enter")
|
||||
for flag in response.valid_flags:
|
||||
if flag.get_name() == "source":
|
||||
source = flag.get_value()
|
||||
elif flag.get_name() == "destination":
|
||||
destination = flag.get_value()
|
||||
elif flag.get_name() == "recursive":
|
||||
recursive = True
|
||||
elif flag.get_name() == "force":
|
||||
force = True
|
||||
|
||||
# Запускаем приложение
|
||||
# Проверка обязательных параметров
|
||||
if not source or not destination:
|
||||
print("Ошибка: необходимо указать источник и назначение")
|
||||
return
|
||||
|
||||
print(f"Копирование из {source} в {destination}")
|
||||
if recursive:
|
||||
print("Рекурсивное копирование включено")
|
||||
if force:
|
||||
print("Принудительное копирование включено")
|
||||
|
||||
# Обработка неопределенных флагов
|
||||
if response.undefined_flags:
|
||||
print("\nПредупреждение: обнаружены незарегистрированные флаги:")
|
||||
for flag in response.undefined_flags:
|
||||
print(f" - {flag.get_name()}" +
|
||||
(f" = {flag.get_value()}" if flag.get_value() else ""))
|
||||
|
||||
# Обработка флагов с некорректными значениями
|
||||
if response.invalid_value_flags:
|
||||
print("\nПредупреждение: обнаружены флаги с некорректными значениями:")
|
||||
for flag in response.invalid_value_flags:
|
||||
print(f" - {flag.get_name()} = {flag.get_value()}")
|
||||
|
||||
# Принятие решения на основе статуса
|
||||
if response.status != Status.ALL_FLAGS_VALID:
|
||||
print("\nВыполнение с предупреждениями из-за проблем с флагами.")
|
||||
|
||||
|
||||
|
||||
app = App()
|
||||
app.include_router(file_router)
|
||||
orchestrator = Orchestrator()
|
||||
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
+2
-2
@@ -1,9 +1,9 @@
|
||||
[project]
|
||||
name = "argenta"
|
||||
version = "1.0.1"
|
||||
version = "1.0.3"
|
||||
description = "Python library for building modular CLI applications"
|
||||
authors = [{ name = "kolo", email = "kolo.is.main@gmail.com" }]
|
||||
requires-python = ">=3.11, <4.0"
|
||||
requires-python = ">=3.8"
|
||||
readme = "README.md"
|
||||
license = { text = "MIT" }
|
||||
dependencies = [
|
||||
|
||||
@@ -1,12 +1,10 @@
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
|
||||
|
||||
@dataclass
|
||||
class PredefinedMessages:
|
||||
class PredefinedMessages(Enum):
|
||||
"""
|
||||
Public. A dataclass with predetermined messages for quick use
|
||||
"""
|
||||
|
||||
USAGE = "[b dim]Usage[/b dim]: [i]<command> <[green]flags[/green]>[/i]"
|
||||
HELP = "[b dim]Help[/b dim]: [i]<command>[/i] [b red]--help[/b red]"
|
||||
AUTOCOMPLETE = "[b dim]Autocomplete[/b dim]: [i]<part>[/i] [bold]<tab>"
|
||||
|
||||
+59
-76
@@ -22,20 +22,18 @@ from argenta.response import Response
|
||||
|
||||
|
||||
class BaseApp:
|
||||
def __init__(
|
||||
self,
|
||||
prompt: str,
|
||||
initial_message: str,
|
||||
farewell_message: str,
|
||||
exit_command: Command,
|
||||
system_router_title: str | None,
|
||||
ignore_command_register: bool,
|
||||
dividing_line: StaticDividingLine | DynamicDividingLine,
|
||||
repeat_command_groups: bool,
|
||||
override_system_messages: bool,
|
||||
autocompleter: AutoCompleter,
|
||||
print_func: Callable[[str], None],
|
||||
) -> None:
|
||||
def __init__(self,
|
||||
prompt: str,
|
||||
initial_message: str,
|
||||
farewell_message: str,
|
||||
exit_command: Command,
|
||||
system_router_title: str | None,
|
||||
ignore_command_register: bool,
|
||||
dividing_line: StaticDividingLine | DynamicDividingLine,
|
||||
repeat_command_groups: bool,
|
||||
override_system_messages: bool,
|
||||
autocompleter: AutoCompleter,
|
||||
print_func: Callable[[str], None]) -> None:
|
||||
self._prompt = prompt
|
||||
self._print_func = print_func
|
||||
self._exit_command = exit_command
|
||||
@@ -49,30 +47,18 @@ class BaseApp:
|
||||
self._farewell_message = farewell_message
|
||||
self._initial_message = initial_message
|
||||
|
||||
self._description_message_gen: Callable[[str, str], str] = (
|
||||
lambda command, description: f"[{command}] *=*=* {description}"
|
||||
)
|
||||
self._description_message_gen: Callable[[str, str], str] = (lambda command, description: f"[{command}] *=*=* {description}")
|
||||
self._registered_routers: RegisteredRouters = RegisteredRouters()
|
||||
self._messages_on_startup: list[str] = []
|
||||
|
||||
self._all_registered_triggers_in_lower: list[str] = []
|
||||
self._all_registered_triggers_in_lower_case: list[str] = []
|
||||
self._all_registered_triggers_in_default_case: list[str] = []
|
||||
|
||||
self._incorrect_input_syntax_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(
|
||||
"Empty input command"
|
||||
)
|
||||
self._unknown_command_handler: Callable[[InputCommand], None] = (
|
||||
lambda command: print_func(f"Unknown command: {command.get_trigger()}")
|
||||
)
|
||||
self._exit_command_handler: Callable[[Response], None] = (
|
||||
lambda response: print_func(self._farewell_message)
|
||||
)
|
||||
self._incorrect_input_syntax_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("Empty input command")
|
||||
self._unknown_command_handler: Callable[[InputCommand], None] = (lambda command: print_func(f"Unknown command: {command.get_trigger()}"))
|
||||
self._exit_command_handler: Callable[[Response], None] = (lambda response: print_func(self._farewell_message))
|
||||
|
||||
def set_description_message_pattern(self, _: Callable[[str, str], str]) -> None:
|
||||
"""
|
||||
@@ -208,7 +194,7 @@ class BaseApp:
|
||||
"""
|
||||
input_command_trigger = command.get_trigger()
|
||||
if self._ignore_command_register:
|
||||
if input_command_trigger.lower() in self._all_registered_triggers_in_lower:
|
||||
if input_command_trigger.lower() in self._all_registered_triggers_in_lower_case:
|
||||
return False
|
||||
else:
|
||||
if input_command_trigger in self._all_registered_triggers_in_default_case:
|
||||
@@ -249,7 +235,7 @@ class BaseApp:
|
||||
|
||||
def _most_similar_command(self, unknown_command: str) -> str | None:
|
||||
all_commands = (
|
||||
self._all_registered_triggers_in_lower
|
||||
self._all_registered_triggers_in_lower_case
|
||||
if self._ignore_command_register
|
||||
else self._all_registered_triggers_in_default_case
|
||||
)
|
||||
@@ -318,21 +304,22 @@ class BaseApp:
|
||||
self._setup_system_router()
|
||||
|
||||
for router_entity in self._registered_routers:
|
||||
self._all_registered_triggers_in_default_case.extend(
|
||||
router_entity.get_triggers()
|
||||
)
|
||||
self._all_registered_triggers_in_default_case.extend(
|
||||
router_entity.get_aliases()
|
||||
)
|
||||
self._all_registered_triggers_in_default_case.extend(router_entity.get_triggers())
|
||||
self._all_registered_triggers_in_default_case.extend(router_entity.get_aliases())
|
||||
|
||||
self._all_registered_triggers_in_lower.extend(
|
||||
[x.lower() for x in router_entity.get_triggers()]
|
||||
)
|
||||
self._all_registered_triggers_in_lower.extend(
|
||||
[x.lower() for x in router_entity.get_aliases()]
|
||||
)
|
||||
self._all_registered_triggers_in_lower_case.extend([x.lower() for x in router_entity.get_triggers()])
|
||||
self._all_registered_triggers_in_lower_case.extend([x.lower() for x in router_entity.get_aliases()])
|
||||
|
||||
self._autocompleter.initial_setup(self._all_registered_triggers_in_lower)
|
||||
self._autocompleter.initial_setup(self._all_registered_triggers_in_lower_case)
|
||||
|
||||
if self._ignore_command_register:
|
||||
for cmd in set(self._all_registered_triggers_in_lower_case):
|
||||
if self._all_registered_triggers_in_lower_case.count(cmd) != 1:
|
||||
Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{cmd}[/b blue]")
|
||||
else:
|
||||
for cmd in set(self._all_registered_triggers_in_default_case):
|
||||
if self._all_registered_triggers_in_default_case.count(cmd) != 1:
|
||||
Console().print(f"\n[b red]WARNING:[/b red] Overlapping trigger or alias: [b blue]{cmd}[/b blue]")
|
||||
|
||||
if not self._override_system_messages:
|
||||
self._setup_default_view()
|
||||
@@ -349,20 +336,18 @@ class BaseApp:
|
||||
|
||||
|
||||
class App(BaseApp):
|
||||
def __init__(
|
||||
self,
|
||||
prompt: str = "What do you want to do?\n",
|
||||
initial_message: str = "\nArgenta\n",
|
||||
farewell_message: str = "\nSee you\n",
|
||||
exit_command: Command = Command("Q", "Exit command"),
|
||||
system_router_title: str | None = "System points:",
|
||||
ignore_command_register: bool = True,
|
||||
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
|
||||
repeat_command_groups: bool = True,
|
||||
override_system_messages: bool = False,
|
||||
autocompleter: AutoCompleter = AutoCompleter(),
|
||||
print_func: Callable[[str], None] = Console().print,
|
||||
) -> None:
|
||||
def __init__(self,
|
||||
prompt: str = "What do you want to do?\n",
|
||||
initial_message: str = "Argenta\n",
|
||||
farewell_message: str = "\nSee you\n",
|
||||
exit_command: Command = Command("Q", "Exit command"),
|
||||
system_router_title: str | None = "System points:",
|
||||
ignore_command_register: bool = True,
|
||||
dividing_line: StaticDividingLine | DynamicDividingLine = StaticDividingLine(),
|
||||
repeat_command_groups: bool = True,
|
||||
override_system_messages: bool = False,
|
||||
autocompleter: AutoCompleter = AutoCompleter(),
|
||||
print_func: Callable[[str], None] = Console().print) -> None:
|
||||
"""
|
||||
Public. The essence of the application itself.
|
||||
Configures and manages all aspects of the behavior and presentation of the user interacting with the user
|
||||
@@ -379,19 +364,17 @@ class App(BaseApp):
|
||||
:param print_func: system messages text output function
|
||||
:return: None
|
||||
"""
|
||||
super().__init__(
|
||||
prompt=prompt,
|
||||
initial_message=initial_message,
|
||||
farewell_message=farewell_message,
|
||||
exit_command=exit_command,
|
||||
system_router_title=system_router_title,
|
||||
ignore_command_register=ignore_command_register,
|
||||
dividing_line=dividing_line,
|
||||
repeat_command_groups=repeat_command_groups,
|
||||
override_system_messages=override_system_messages,
|
||||
autocompleter=autocompleter,
|
||||
print_func=print_func,
|
||||
)
|
||||
super().__init__(prompt=prompt,
|
||||
initial_message=initial_message,
|
||||
farewell_message=farewell_message,
|
||||
exit_command=exit_command,
|
||||
system_router_title=system_router_title,
|
||||
ignore_command_register=ignore_command_register,
|
||||
dividing_line=dividing_line,
|
||||
repeat_command_groups=repeat_command_groups,
|
||||
override_system_messages=override_system_messages,
|
||||
autocompleter=autocompleter,
|
||||
print_func=print_func)
|
||||
|
||||
def run_polling(self) -> None:
|
||||
"""
|
||||
@@ -420,7 +403,7 @@ class App(BaseApp):
|
||||
system_router.finds_appropriate_handler(input_command)
|
||||
if self._ignore_command_register:
|
||||
self._autocompleter.exit_setup(
|
||||
self._all_registered_triggers_in_lower
|
||||
self._all_registered_triggers_in_lower_case
|
||||
)
|
||||
else:
|
||||
self._autocompleter.exit_setup(
|
||||
|
||||
@@ -192,14 +192,12 @@ class Router:
|
||||
pass
|
||||
else:
|
||||
file_path: str = getsourcefile(func)
|
||||
source_line: int = getsourcelines(func)[1] + 1
|
||||
source_line: int = getsourcelines(func)[1]
|
||||
fprint = Console().print
|
||||
fprint(
|
||||
f'\nFile "{file_path}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
|
||||
fprint(f'\nFile "{file_path}", line {source_line}\n[b red]WARNING:[/b red] [i]The typehint '
|
||||
f"of argument([green]{transferred_arg}[/green]) passed to the handler is [/i][bold blue]{Response}[/bold blue],"
|
||||
f" [i]but[/i] [bold blue]{arg_annotation}[/bold blue] [i]is specified[/i]\n",
|
||||
highlight=False,
|
||||
)
|
||||
f" [i]but[/i] [bold blue]{arg_annotation}[/bold blue] [i]is specified[/i]",
|
||||
highlight=False)
|
||||
|
||||
def set_command_register_ignore(self, _: bool) -> None:
|
||||
"""
|
||||
|
||||
@@ -33,13 +33,13 @@ class MyTestCase(unittest.TestCase):
|
||||
def test_is_unknown_command1(self):
|
||||
app = App()
|
||||
app.set_unknown_command_handler(lambda command: None)
|
||||
app._all_registered_triggers_in_lower = ['fr', 'tr', 'de']
|
||||
app._all_registered_triggers_in_lower_case = ['fr', 'tr', 'de']
|
||||
self.assertEqual(app._is_unknown_command(InputCommand('fr')), False)
|
||||
|
||||
def test_is_unknown_command2(self):
|
||||
app = App()
|
||||
app.set_unknown_command_handler(lambda command: None)
|
||||
app._all_registered_triggers_in_lower = ['fr', 'tr', 'de']
|
||||
app._all_registered_triggers_in_lower_case = ['fr', 'tr', 'de']
|
||||
self.assertEqual(app._is_unknown_command(InputCommand('cr')), True)
|
||||
|
||||
def test_is_unknown_command3(self):
|
||||
|
||||
Reference in New Issue
Block a user