222 lines
10 KiB
Python
222 lines
10 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
|
||
|
||
# --- Настройки и инициирование логирования модуля ---
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Кастомный обработчик логирования для 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)
|
||
|
||
|
||
class MainMenu(Static):
|
||
def compose(self) -> ComposeResult:
|
||
yield Horizontal(
|
||
Button(label="Backup (DB -> Dump)", id="backup"),
|
||
Button(label="Restore (Dump -> DB)", id="restore"),
|
||
Button(label="Copy (DB -> DB)", id="copy", variant="error"),
|
||
Button(label="Quit", id="quit", variant="error"),
|
||
id="menu",
|
||
)
|
||
|
||
# Новый простой экран для Backup
|
||
class SimpleBackupScreen(Screen):
|
||
"""
|
||
Простой экран для демонстрации перехода при выборе Backup.
|
||
"""
|
||
|
||
BINDINGS = [
|
||
("escape", "pop_screen", "Назад"), # Позволяет вернуться на главный экран по Escape
|
||
]
|
||
|
||
def compose(self) -> ComposeResult:
|
||
yield Header(name="Экран Бэкапа") # Заголовок для нового экрана
|
||
yield Label("Привет, мир! Это экран для бэкапа.", classes="greeting-label")
|
||
yield Footer()
|
||
|
||
def on_mount(self) -> None:
|
||
logger.info(f"Экран '{self.id}' (SimpleBackupScreen) смонтирован.")
|
||
|
||
class PGanecApp(App):
|
||
CSS = """
|
||
Screen {
|
||
layout: vertical; /* Используем вертикальный layout для всего экрана */
|
||
}
|
||
#main_content_area { /* Новый контейнер для всего, что НЕ прибито к краям */
|
||
/* Этот контейнер будет занимать все доступное пространство
|
||
между Header и Log/Footer */
|
||
height: 1fr; /* 1fr означает "одна доля доступного пространства" */
|
||
overflow-y: auto; /* Если контент не влезет, появится прокрутка */
|
||
}
|
||
#title { width: 100%; text-align: left; background: purple; padding: 0 1; }
|
||
#menu { width: 100%; height: 3; align: left top; padding: 0; }
|
||
Button {
|
||
align: center middle;
|
||
width: 25%;
|
||
color: orange;
|
||
background: transparent;
|
||
margin: 0; padding: 0;
|
||
}
|
||
|
||
Button#backup, Button#restore, 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;
|
||
margin-top: 1;
|
||
border-top: double grey;
|
||
background: 15%;
|
||
}
|
||
Log#app_log_viewer:focus { height: 50%; border-top: solid green; background: 45%; }
|
||
|
||
/* Стили для нового экрана */
|
||
SimpleBackupScreen .greeting-label {
|
||
width: 100%;
|
||
text-align: center;
|
||
padding: 2 0; /* Немного отступов сверху и снизу */
|
||
text-style: bold;
|
||
}
|
||
|
||
Footer { dock: bottom; height: 1; }
|
||
Header { dock: top; height: 1; }
|
||
|
||
"""
|
||
|
||
BINDINGS = [
|
||
("b", "backup", "Backup"),
|
||
("r", "restore", "Restore"),
|
||
("c", "copy", "Copy"),
|
||
("q", "quit", "Quit"),
|
||
("right", "focus_next"), ("down", "focus_next"), # Стрелочки для навигации (вниз/вправо -- вперед)
|
||
("left", "focus_previous"), ("up", "focus_previous"), # Стрелочки для навигации (вверх/влево -- назад)
|
||
("enter", "activate"),
|
||
]
|
||
|
||
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 # <--- Сохраняем
|
||
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 Log(id="app_log_viewer", highlight=True) # Виджет лога
|
||
yield Footer()
|
||
|
||
async def on_button_pressed(self, event: Button.Pressed) -> None:
|
||
button_id = event.button.id
|
||
if button_id == "backup":
|
||
await self.push_screen(SimpleBackupScreen())
|
||
elif button_id == "restore":
|
||
logger.info("Действие 'Restore' пока не реализовано с новым экраном.")
|
||
self.app.bell() # Сигнал, что действие не выполнено
|
||
elif button_id == "copy":
|
||
logger.warning("Copy action is currently a placeholder for quit.")
|
||
await self.action_quit()
|
||
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("TUI: DEBUG сообщение для проверки")
|
||
logger.warning("TUI: WARNING для проверки")
|
||
logger.error("TUI: ERROR тоже для проверки")
|
||
|
||
|
||
async def action_activate(self) -> None:
|
||
focused = self.focused
|
||
if isinstance(focused, Button):
|
||
await self.on_button_pressed(Button.Pressed(focused))
|
||
|
||
# Заглушки для других экранов:
|
||
class BackupScreen(Static):
|
||
def on_mount(self) -> None:
|
||
self.update("🛠 Здесь будет экран выбора сервера для бэкапа.")
|
||
|
||
class RestoreScreen(Static):
|
||
def on_mount(self) -> None:
|
||
self.update("♻️ Здесь будет экран выбора сервера и бэкапа для восстановления.")
|