mirror of
https://github.com/koloideal/Argenta.git
synced 2026-06-10 18:15:28 +03:00
wotk
This commit is contained in:
+97
-8
@@ -1,13 +1,102 @@
|
||||
from enum import Enum
|
||||
from typing import Literal
|
||||
from rich.console import Console
|
||||
|
||||
from argenta.app import App
|
||||
from argenta.router import Router
|
||||
from argenta.command import Command
|
||||
from argenta.response import Response
|
||||
from argenta.orchestrator import Orchestrator
|
||||
from argenta.app.dividing_line import DynamicDividingLine, StaticDividingLine
|
||||
import platform
|
||||
import psutil
|
||||
import os
|
||||
import subprocess
|
||||
import socket
|
||||
|
||||
class PossibleValues(Enum):
|
||||
DISABLE: Literal[False] = False
|
||||
ALL: Literal[True] = True
|
||||
# Создаем маршрутизатор для работы с файлами
|
||||
file_router = Router("Файловые операции", disable_redirect_stdout=True)
|
||||
|
||||
def __eq__(self, other: bool) -> bool:
|
||||
return self.value == other
|
||||
@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} не найден")
|
||||
|
||||
print(PossibleValues.DISABLE == False)
|
||||
# Создаем маршрутизатор для системных операций
|
||||
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("Сетевые операции", disable_redirect_stdout=True)
|
||||
|
||||
@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)}")
|
||||
|
||||
# Создаем маршрутизатор для работы с пользователями
|
||||
user_router = Router("Операции с пользователями")
|
||||
|
||||
@user_router.command(Command("create", "Создать пользователя"))
|
||||
def create_user(response: Response):
|
||||
username = input("Введите имя пользователя: ")
|
||||
print(f"Пользователь {username} создан")
|
||||
|
||||
@user_router.command(Command("delete", "Удалить пользователя"))
|
||||
def delete_user(response: Response):
|
||||
username = input("Введите имя пользователя: ")
|
||||
print(f"Пользователь {username} удален")
|
||||
|
||||
# Создаем приложение и регистрируем маршрутизаторы
|
||||
app = App(
|
||||
prompt="MyApp> ",
|
||||
initial_message="Сложное тестовое приложение Argenta!",
|
||||
farewell_message="До свидания!",
|
||||
dividing_line=StaticDividingLine("*"),
|
||||
repeat_command_groups=False,
|
||||
ignore_command_register=True,
|
||||
override_system_messages=True,
|
||||
)
|
||||
|
||||
app.include_routers(file_router, system_router, network_router, user_router)
|
||||
|
||||
# Добавляем сообщение при запуске
|
||||
app.add_message_on_startup("\nДля просмотра доступных команд нажмите Enter")
|
||||
|
||||
# Пользовательский обработчик пустых команд
|
||||
def empty_command_handler():
|
||||
print("Для выхода введите Q")
|
||||
|
||||
app.set_empty_command_handler(empty_command_handler)
|
||||
|
||||
# Запускаем приложение
|
||||
orchestrator = Orchestrator()
|
||||
orchestrator.start_polling(app)
|
||||
|
||||
@@ -26,3 +26,8 @@ exclude = [
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
"psutil>=7.0.0",
|
||||
]
|
||||
|
||||
|
||||
@@ -50,7 +50,7 @@ class BaseApp:
|
||||
self._initial_message = initial_message
|
||||
|
||||
self._description_message_gen: Callable[[str, str], str] = (
|
||||
lambda command, description: f"[{command}] *=*=* {description}"
|
||||
lambda command, description: f"{command} *=*=* {description}"
|
||||
)
|
||||
self._registered_routers: RegisteredRouters = RegisteredRouters()
|
||||
self._messages_on_startup: list[str] = []
|
||||
@@ -274,7 +274,7 @@ class BaseApp:
|
||||
Private. Sets up default app view
|
||||
:return: None
|
||||
"""
|
||||
self._prompt = "[italic dim bold]What do you want to do?\n"
|
||||
self._prompt = f"[italic dim bold]{self._prompt}"
|
||||
self._initial_message = (
|
||||
"\n" + f"[bold red]{text2art(self._initial_message, font='tarty1')}" + "\n"
|
||||
)
|
||||
@@ -456,6 +456,7 @@ class App(BaseApp):
|
||||
for registered_router in self._registered_routers:
|
||||
if registered_router.disable_redirect_stdout:
|
||||
if isinstance(self._dividing_line, StaticDividingLine):
|
||||
print('k')
|
||||
self._print_func(
|
||||
self._dividing_line.get_full_static_line(
|
||||
self._override_system_messages
|
||||
@@ -468,6 +469,7 @@ class App(BaseApp):
|
||||
)
|
||||
)
|
||||
else:
|
||||
print('j')
|
||||
self._print_func(
|
||||
StaticDividingLine(
|
||||
self._dividing_line.get_unit_part()
|
||||
@@ -480,10 +482,12 @@ class App(BaseApp):
|
||||
).get_full_static_line(self._override_system_messages)
|
||||
)
|
||||
else:
|
||||
print('m')
|
||||
with redirect_stdout(io.StringIO()) as f:
|
||||
registered_router.finds_appropriate_handler(input_command)
|
||||
res: str = f.getvalue()
|
||||
if res:
|
||||
print('w')
|
||||
self._print_framed_text(res)
|
||||
|
||||
def include_router(self, router: Router) -> None:
|
||||
|
||||
Reference in New Issue
Block a user