454 lines
24 KiB
Python
454 lines
24 KiB
Python
# -*- coding: utf-8 -*-
|
||
# src/pganec/tui.py
|
||
|
||
import logging
|
||
from logging.handlers import MemoryHandler
|
||
from typing import Dict, Any, Optional, List, Tuple
|
||
from textual.app import App, ComposeResult
|
||
from textual.containers import Vertical, Horizontal, Container
|
||
from textual.css.query import DOMQuery
|
||
from textual.reactive import reactive
|
||
from textual.screen import Screen
|
||
from textual.widgets import Button, Header, Footer, Static, Log, Label, ListView, ListItem, OptionList
|
||
from textual.widget import Widget
|
||
from typing import TYPE_CHECKING
|
||
if TYPE_CHECKING:
|
||
from .tui import PGanecApp
|
||
|
||
# --- Настройки и инициирование логирования модуля ---
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# --- КОНФИГ_ПЕРЕМЕННЕЫ которые использоваться в приложении (возможно стоит перенести в другой файл) ---
|
||
DB_FOR_BACKUP = "db_4b"
|
||
|
||
# --- Кастомный обработчик логирования (для Textual) ---
|
||
class TextualLogHandler(logging.Handler):
|
||
"""
|
||
Обработчик логирования, который направляет записи в виджет Textual Log.
|
||
"""
|
||
def __init__(self, textual_log_widget: Log):
|
||
super().__init__()
|
||
self.textual_log_widget = textual_log_widget
|
||
|
||
def emit(self, record: logging.LogRecord):
|
||
"""
|
||
Отправляет отформатированную запись лога в виджет.
|
||
"""
|
||
try:
|
||
msg = self.format(record)
|
||
# Метод Log.write() является потоко-безопасным, и его можно вызывать напрямую.
|
||
self.textual_log_widget.write(msg)
|
||
except Exception:
|
||
# Стандартная обработка ошибок для Handler, если что-то пошло не так (например, виджет был удален
|
||
# или произошла ошибка форматирования)
|
||
self.handleError(record)
|
||
|
||
|
||
# --- Верхнее меню приложения (для Textual) ---
|
||
class MainMenu(Static):
|
||
def compose(self) -> ComposeResult:
|
||
yield Horizontal(
|
||
Button(label="Backup (DB\u2192Dump)", id="backup", classes="main-menu-button"),
|
||
Button(label="Restore (Dump\u2192DB)", id="restore"),
|
||
Button(label="Copy (DB\u2192DB)", id="copy", variant="error"),
|
||
Button(label="Service", id="service"),
|
||
Button(label="Quit", id="quit", variant="error"),
|
||
id="menu",
|
||
)
|
||
|
||
# --- Виджет для "встроенного" выбора ---
|
||
class SelectionWidget(Static): # Наследуем от Static
|
||
"""
|
||
Виджет для отображения опций выбора бекапа внутри основного экрана.
|
||
"""
|
||
@property
|
||
def app(self) -> "PGanecApp":
|
||
"""
|
||
Возвращает ссылку на приложение PGanecApp , к которому принадлежит этот виджет. Нужно для доступа к методам
|
||
и состоянию приложения (и статического анализатора, т.к. класс PGanecApp описан ниже и "отсюда не виден").
|
||
:return:
|
||
"""
|
||
return super().app # type: ignore
|
||
|
||
|
||
def __init__(self,
|
||
app_config: Optional[Dict[str, Any]] = None,
|
||
action_type=None,
|
||
**kwargs):
|
||
super().__init__(**kwargs)
|
||
self.app_config = app_config if app_config is not None else {}
|
||
self.action_type = action_type # Сохраняем тип действия
|
||
self._button_id_to_data_map: Dict[str, Any] = {} # Карта для хранения значений опций
|
||
# SelectionWidget(Static)
|
||
|
||
def compose(self) -> ComposeResult:
|
||
# options: List[Tuple[str, str]] = [] # Список опций для выбора
|
||
self._button_id_to_data_map.clear() # Очищаем карту значений
|
||
num_count = 1
|
||
if self.action_type == "backup":
|
||
yield Label("Выберите BD-сервер для бэкапа:")
|
||
# ... тут будет список серверов
|
||
if "servers" in self.app_config:
|
||
logger.info(f"TUI: Обнаружено {len(self.app_config['servers'])} серверов в конфигурации.")
|
||
for server_conf in self.app_config["servers"]:
|
||
server_name = server_conf.get("name")
|
||
button_id = server_conf.get("id")
|
||
if not button_id:
|
||
button_id = f"server_{num_count:02d}" # Генерируем временный id, если не указан
|
||
logger.warning(
|
||
f"Сервер '{server_name}' не имеет 'id' в yaml-конфигурации. Установлен id='{button_id}'.")
|
||
# continue # пропускаем?
|
||
if button_id in self._button_id_to_data_map:
|
||
logger.error(
|
||
f"TUI: Дублирующийся ID кнопки '{button_id}' для сервера '{server_name}' из yaml-конфигурации. ПРОРУСКАЕМ.")
|
||
continue
|
||
# Добавляем кнопку для каждого сервера
|
||
yield Button(label=f"{num_count:02d}: {server_name}", id=button_id, classes="list-button")
|
||
self._button_id_to_data_map[button_id] = {"type": DB_FOR_BACKUP, "config": server_conf}
|
||
num_count += 1
|
||
|
||
elif self.action_type == "restore":
|
||
yield Label("Выберите том для восстановления:")
|
||
if "targets" in self.app_config:
|
||
logger.info(f"TUI: Обнаружено {len(self.app_config['targets'])} томов в конфигурации.")
|
||
yield Static("Тут будет список...", classes="placeholder-text") # Временная заглушка
|
||
# ... тут будет список томов
|
||
elif self.action_type == "copy":
|
||
yield Label("Выберите BD-сервер (откуда копировать):")
|
||
# ... тут будет список серверов
|
||
elif self.action_type == "service":
|
||
yield Label("Служебные функции:")
|
||
# ... тут будут служебные функции
|
||
|
||
# Кнопка "Отмена" в конце любого списка
|
||
yield Button(label=f"{num_count:02d}: Отмена", id="cancel_widget", classes="list-button")
|
||
|
||
|
||
def on_mount(self) -> None:
|
||
"""
|
||
Вызывается после монтирования виджета в DOM.
|
||
:return:
|
||
"""
|
||
logger.info(f"Виджет '{self.id}' (SelectionWidget) смонтирован.")
|
||
try:
|
||
# Пытаемся установить фокус на первую кнопку-опцию
|
||
# Ищем первую кнопку с классом "list-button" внутри этого виджета
|
||
first_option_button = self.query_one("Button.list-button", Button)
|
||
first_option_button.focus()
|
||
logger.debug(f"TUI: SelectionWidget: Фокус на кнопке-опцию: `{first_option_button.id}`")
|
||
except Exception as e:
|
||
# except DOMQuery.DoesNotExist:
|
||
logger.error(f"SelectionWidget: Ошибка при установке фокуса в Виджете: {e}")
|
||
# logger.warning(f"SelectionWidget: OptionList не найден для установки фокуса (action_type: {self.action_type}).")
|
||
|
||
# Можно сразу установить фокус на первый элемент, если он есть
|
||
# self.query_one(OptionList).focus() или self.query_one(Button).focus()
|
||
|
||
|
||
async def on_button_pressed(self, event: Button.Pressed) -> None:
|
||
button_id = event.button.id
|
||
button_data = self._button_id_to_data_map.get(button_id) # <--- Извлекаем данные по ID кнопки
|
||
|
||
logger.info(f"SelectionWidget ({self.action_type}): нажата кнопка-опция id='{button_id}', data={button_data}")
|
||
|
||
if button_data:
|
||
action_type_from_data = button_data.get("type")
|
||
if action_type_from_data == DB_FOR_BACKUP:
|
||
server_config = button_data.get("config") # <--- Получаем полную конфигурацию сервера
|
||
logger.info(
|
||
f"SelectionWidget ({self.action_type}): выбран сервер {server_config.get('name') if isinstance(server_config, dict) else server_config}. Далее - выбор БД/дампа.")
|
||
# Тут будет логика для перехода к следующему шагу, используя server_config
|
||
self.app.bell()
|
||
# ... другие обработчики ...
|
||
|
||
if event.button.id == "cancel_widget":
|
||
logger.debug("Нажата 'Отмена' в SelectionWidget.")
|
||
await self.remove() # Удаляем сам виджет
|
||
self.app.selection_widget_active = False # <-- Сброс флага что виджет активен
|
||
# Можно также послать сообщение приложению, чтобы оно обновило состояние, если нужно
|
||
# self.app.post_message(BackupSelectionCancelled())
|
||
self.app.set_main_menu_enabled(True) # Включаем верхнее меню обратно
|
||
|
||
|
||
|
||
class PGanecApp(App):
|
||
CSS = """
|
||
Screen { layout: vertical; } /* Вертикальная компоновка для всего приложения */
|
||
#main_content_area { /* Новый контейнер для всего, что НЕ прибито к краям */
|
||
/* Этот контейнер будет занимать все доступное пространство между Header и Log/Footer */
|
||
height: 1fr; /* 1fr означает "одна доля доступного пространства" */
|
||
overflow-y: auto; /* Если контент не влезет, появится прокрутка */
|
||
}
|
||
#title { width: 100%; text-align: left; background: purple; padding: 0; }
|
||
#menu { width: 100%; height: 3; align: left top; padding: 0; }
|
||
#menu Button {
|
||
align: center middle;
|
||
width: 20%;
|
||
color: orange;
|
||
background: transparent;
|
||
margin: 0; padding: 0;
|
||
}
|
||
|
||
Button#backup, Button#restore, Button#service, Button#quit { color: orange; border: round orange; }
|
||
|
||
Button#copy { color: grey; border: round orange; }
|
||
|
||
Button:focus { border: heavy green; }
|
||
|
||
Log#app_log_viewer {
|
||
dock: bottom;
|
||
height: 10;
|
||
border-top: double grey;
|
||
background: 15%;
|
||
}
|
||
Log#app_log_viewer:focus { height: 50%; border-top: solid green; background: 45%; }
|
||
|
||
/* Стили для встроенного виджета выбора */
|
||
SelectionWidget {
|
||
padding: 1;
|
||
margin-top: 0;
|
||
background: $panel; /* Используем системный цвет панели */
|
||
/* height: auto; /* Высота будет по содержимому */
|
||
}
|
||
|
||
SelectionWidget .placeholder-text {
|
||
padding: 0;
|
||
color: $text-muted;
|
||
}
|
||
Button.list-button { /* Стили для кнопок внутри виджета */
|
||
width: auto;
|
||
height: 1;
|
||
align: left middle;
|
||
border: hidden;
|
||
background: transparent;
|
||
color: $text;
|
||
margin-left: 0; padding-left: 1;
|
||
}
|
||
Button#cancel_widget {margin-left: -1; color: red; } /* Отдельный стиль для кнопки "Отмена" */
|
||
|
||
Footer { dock: bottom; height: 1; }
|
||
Header { dock: top; height: 1; }
|
||
|
||
"""
|
||
|
||
BINDINGS = [
|
||
("b", "backup", "Backup"),
|
||
("r", "restore", "Restore"),
|
||
("c", "copy", "Copy"),
|
||
("s", "service", "Service"),
|
||
("q", "quit", "Quit"),
|
||
("right", "focus_next"), ("down", "focus_next"), # Стрелочки для навигации (вниз/вправо -- вперед)
|
||
("left", "focus_previous"), ("up", "focus_previous"), # Стрелочки для навигации (вверх/влево -- назад)
|
||
("enter", "activate"),
|
||
]
|
||
|
||
# Атрибут для хранения ссылки на виджет выбора, если он отображен
|
||
selection_widget_instance: Optional[SelectionWidget] = None
|
||
|
||
def __init__(self,
|
||
log_level_int: int = logging.INFO,
|
||
app_config: Dict[str, Any] = None,
|
||
early_log_handler: Optional[MemoryHandler] = None,
|
||
**kwargs):
|
||
"""
|
||
Конструктор приложения.
|
||
:param cli_log_level_str: Уровень логирования из CLI (задаётся через параметры при запуске приложения).
|
||
"""
|
||
super().__init__(**kwargs)
|
||
self.log_level_int = log_level_int
|
||
self.app_config = app_config if app_config is not None else {} # Сохраняем конфигурацию
|
||
self.early_log_handler = early_log_handler # <--- Сохраняем
|
||
self.selection_widget_instance = None # Инициализируем атрибут для виджета выбора
|
||
logger.debug(
|
||
f"TUI инициализирован. Log Level: `{logging.getLevelName(self.log_level_int)}`; Config Keys: {list(self.app_config.keys())}"
|
||
)
|
||
|
||
|
||
def compose(self) -> ComposeResult:
|
||
yield Header()
|
||
with Vertical(id="main_content_area"):
|
||
yield Static("PGanec TUI", id="title")
|
||
yield MainMenu()
|
||
# Другие элементы основного интерфейса могут быть здесь
|
||
# Место для нашего виджета выбора будет здесь.
|
||
# Мы не добавляем его сразу, а будем делать это динамически.
|
||
# Можно добавить "якорь" - пустой контейнер, если нужно точное позиционирование
|
||
yield Container(id="selection_widget_placeholder")
|
||
yield Log(id="app_log_viewer", highlight=True) # Виджет лога
|
||
yield Footer()
|
||
|
||
|
||
def set_main_menu_enabled(self, enabled: bool):
|
||
"""
|
||
Включает или отключает кнопки верхнего меню
|
||
:param enabled:
|
||
:return:
|
||
"""
|
||
menu = self.query_one(MainMenu)
|
||
for btn in menu.query(Button):
|
||
btn.disabled = not enabled
|
||
|
||
|
||
async def show_selection_widget(self, action_type: str) -> None:
|
||
"""Показывает виджет выбора для ."""
|
||
# Сначала удаляем старый виджет, если он есть, чтобы не было дублей
|
||
if self.selection_widget_instance:
|
||
try:
|
||
await self.selection_widget_instance.remove()
|
||
except Exception as e: # DOMObjectMissingError может возникнуть, если уже удален
|
||
logger.debug(f"Ошибка при удалении старого selection_widget_instance (возможно, уже удален): {e}")
|
||
self.selection_widget_instance = None
|
||
|
||
# Создаем новый экземпляр виджета
|
||
self.selection_widget_instance = SelectionWidget(
|
||
app_config=self.app_config,
|
||
action_type=action_type,
|
||
id="select_any" # Даем ему id для возможного поиска
|
||
)
|
||
|
||
self.selection_widget_active = True # Выставляем флаг, что виджет активен
|
||
self.set_main_menu_enabled(False) # Отключаем верхнее меню
|
||
# Находим контейнер, куда его добавить.
|
||
# Будем добавлять после MainMenu внутри #main_content_area
|
||
main_content_area = self.query_one("#main_content_area", Vertical)
|
||
main_menu = self.query_one(MainMenu)
|
||
|
||
# Добавляем виджет в DOM после MainMenu
|
||
# `mount` добавляет виджет и вызывает его on_mount
|
||
await main_content_area.mount(self.selection_widget_instance, after=main_menu)
|
||
self.selection_widget_instance.focus() # Пытаемся установить фокус
|
||
logger.info("SelectionWidget добавлен в DOM.")
|
||
|
||
|
||
async def on_button_pressed(self, event: Button.Pressed) -> None:
|
||
"""
|
||
Обработчик нажатия (Enter или Click) кнопки в верхнем меню.
|
||
:param event: Полученное событие нажатия кнопки на верхнем меню.
|
||
:return:
|
||
"""
|
||
button_id = event.button.id
|
||
if button_id == "backup":
|
||
logger.debug("TUI: Enter или Click на кнопку 'Backup' -- Инициализируем виджет выбора.")
|
||
await self.show_selection_widget("backup")
|
||
elif button_id == "restore":
|
||
logger.debug("TUI: Enter или Click на кнопку 'Restore' -- Инициализируем виджет выбора.")
|
||
await self.show_selection_widget("restore")
|
||
elif button_id == "copy":
|
||
logger.debug("TUI: Enter или Click на кнопку 'Copy' -- Инициализируем виджет выбора.")
|
||
await self.show_selection_widget("copy")
|
||
elif button_id == "service":
|
||
logger.debug("TUI: Enter или Click на кнопку 'Service' -- Инициализируем виджет выбора.")
|
||
await self.show_selection_widget("service")
|
||
elif button_id == "quit":
|
||
await self.action_quit()
|
||
|
||
def on_mount(self) -> None:
|
||
"""Вызывается после монтирования DOM приложения."""
|
||
# --- Обжужукиваем логирование в #app_log_viewer ---
|
||
log_viewer_widget = self.query_one("#app_log_viewer", Log)
|
||
textual_handler = TextualLogHandler(log_viewer_widget)
|
||
if self.log_level_int >= logging.INFO:
|
||
formatter = logging.Formatter(
|
||
fmt="%(asctime)s [%(levelname)-8s]\t %(message)s\n",
|
||
datefmt="%H:%M:%S" # Формат времени: часы:минуты:секунды
|
||
)
|
||
else:
|
||
formatter = logging.Formatter(
|
||
fmt="%(asctime)s [%(levelname)-8s] %(module)s.%(funcName)s:%(lineno)04d\t- %(message)s\n",
|
||
datefmt="%H:%M:%S.%s" # Формат времени: часы:минуты:секунды.милисекунды
|
||
)
|
||
textual_handler.setFormatter(formatter)
|
||
textual_handler.setLevel(self.log_level_int)
|
||
|
||
# Получаем корневой логгер
|
||
target_logger = logging.getLogger()
|
||
target_logger.addHandler(textual_handler) # Добавляем наш обработчик
|
||
|
||
# --- -- Перенос и сброс ранних логов из MemoryHandler -- ---
|
||
if self.early_log_handler:
|
||
# Устанавливаем наш textual_handler как цель для MemoryHandler
|
||
self.early_log_handler.setTarget(textual_handler)
|
||
# Сбрасываем все накопленные записи
|
||
self.early_log_handler.flush()
|
||
# Закрываем и удаляем MemoryHandler из корневого логера, так как он больше не нужен
|
||
self.early_log_handler.close()
|
||
logger.debug("TUI: Перенесли и закрыли ранние логи…")
|
||
target_logger.removeHandler(self.early_log_handler)
|
||
logger.debug("TUI: Удалили ранний MemoryHandler из корневого логгера…")
|
||
self.early_log_handler = None # Очищаем ссылку
|
||
# --- -- Конец переноса и сброса -- ---
|
||
# --- Конец обжужукивания логирования ---
|
||
|
||
logger.info("TUI-логгер инициализирован")
|
||
logger.debug("test TUI-logger: DEBUG")
|
||
logger.warning("test TUI-logger: WARNING")
|
||
logger.error("test TUI-logger: ERROR")
|
||
|
||
|
||
async def action_activate(self) -> None:
|
||
"""
|
||
Обработчик нажатия клавиши Enter.
|
||
:return:
|
||
"""
|
||
focused = self.focused
|
||
if not focused:
|
||
logger.debug("TUI: action_activate вызван, но нет сфокусированного элемента.")
|
||
return
|
||
else:
|
||
if isinstance(focused, Button):
|
||
# Проверяем, не является ли кнопка частью нашего виджета выбора
|
||
if not isinstance(focused, Button): # Если не кнопка - ничего не делаем
|
||
logger.debug("TUI: Что-то нажато (Enter), но это не кнопка.")
|
||
return
|
||
|
||
# if focused.parent and isinstance(focused.parent, SelectionWidget):
|
||
# await focused.parent.on_button_pressed(Button.Pressed(focused))
|
||
# else:
|
||
# await self.on_button_pressed(Button.Pressed(focused))
|
||
|
||
# --- Обработчики быстрых клавиш ---
|
||
# Для 'q' (Quit) уже есть в BINDINGS срабатывающий метод action_quit, он сработает автоматически (не ясно почему).
|
||
async def action_backup(self):
|
||
"""
|
||
Обработчик нажатия быстрой клавиши 'b' для инициализации виджета создания бэкапа.
|
||
:return:
|
||
"""
|
||
if getattr(self, "selection_widget_active", False): # Проверяем, активен ли виджет выбора
|
||
return # Если активен, ничего не делаем (отменяем нажатие клавиши)
|
||
self.query_one("#backup", Button).focus() # подсветим кнопку
|
||
logger.debug("TUI: Быстрая клавиша 'b' -- Инициализируем виджет.")
|
||
await self.show_selection_widget("backup")
|
||
|
||
async def action_restore(self):
|
||
"""
|
||
Обработчик нажатия быстрой клавиши 'r' для восстановления.
|
||
:return:
|
||
"""
|
||
if getattr(self, "selection_widget_active", False): # Проверяем, активен ли виджет выбора
|
||
return
|
||
self.query_one("#restore", Button).focus() # подсветим кнопку
|
||
logger.debug("TUI: Быстрая клавиша 'r' -- Инициализируем виджет.")
|
||
await self.show_selection_widget("restore")
|
||
|
||
async def action_copy(self):
|
||
"""
|
||
Обработчик нажатия быстрой клавиши 'c' для копирования.
|
||
:return:
|
||
"""
|
||
if getattr(self, "selection_widget_active", False): # Проверяем, активен ли виджет выбора
|
||
return
|
||
self.query_one("#copy", Button).focus()
|
||
logger.debug("TUI: Быстрая клавиша 'c' -- Инициализируем виджет.")
|
||
await self.show_selection_widget("copy")
|
||
|
||
async def action_service(self):
|
||
"""
|
||
Обработчик нажатия быстрой клавиши 's' для сервисных действий.
|
||
:return:
|
||
"""
|
||
if getattr(self, "selection_widget_active", False): # Проверяем, активен ли виджет выбора
|
||
return
|
||
self.query_one("#service", Button).focus()
|
||
logger.debug("TUI: Быстрая клавиша 's' -- Инициализируем виджет.")
|
||
await self.show_selection_widget("service")
|