add: логи внутри TUI

This commit is contained in:
Sergei Erjemin 2025-06-08 22:28:00 +03:00
parent e6dd940d8a
commit 22bd7ee303
3 changed files with 194 additions and 64 deletions

View File

@ -4,8 +4,10 @@ import logging
import yaml import yaml
from typing import Optional, Dict # Для Python < 3.9 используйте Dict, для Python 3.9+ можно просто dict from typing import Optional, Dict # Для Python < 3.9 используйте Dict, для Python 3.9+ можно просто dict
# --- Настройки и инициирование логирования модуля ---
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class ConfigError(Exception): class ConfigError(Exception):
"""Базовый класс для ошибок конфигурации.""" """Базовый класс для ошибок конфигурации."""
pass pass
@ -38,7 +40,7 @@ def load_config(path: str) -> Optional[Dict]:
logger.error(msg) logger.error(msg)
raise InvalidConfigFormatError(msg) raise InvalidConfigFormatError(msg)
logger.info(msg=f"Конфигурация успешно загружена из `{path}`") logger.info(msg=f"Конфигурация успешно загружена из: `{path}`")
return config_data return config_data
except FileNotFoundError: except FileNotFoundError:

View File

@ -3,69 +3,105 @@
import argparse import argparse
import logging import logging
import sys import sys
from config import load_config, ConfigError, ConfigNotFoundError, InvalidConfigFormatError from logging.handlers import MemoryHandler
from tui import PGanecApp from pganec.config import load_config, ConfigError, ConfigNotFoundError, InvalidConfigFormatError
from pganec.tui import PGanecApp
# --- Настройки и инициирование логирования --- # --- Настройки и инициирование логирования ---
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def print_hi(name):
# Use a breakpoint in the code line below to debug your script.
print(f'Hi, {name}') # Press ⌘F8 to toggle the breakpoint.
# --- Главная функция приложения ---
# Press the green button in the gutter to run the script.
if __name__ == '__main__': if __name__ == '__main__':
parser = argparse.ArgumentParser(description="PGanec — TUI резервного копирования и восстановления баз PostgreSQL") parser = argparse.ArgumentParser(description="PGanec — TUI резервного копирования и восстановления баз PostgreSQL")
parser.add_argument("-c", "--config", required=True, parser.add_argument("-c", "--config", required=True,
help="Путь к файлу конфигурации (YAML)") help="Путь к файлу конфигурации (YAML)")
parser.add_argument("-d", "--debug-level", default="QUIET", parser.add_argument("-d", "--debug-level", default="INFO",
help="Уровень отладки (QUIET, NOTSET, DEBUG, INFO, WARNING, ERROR, CRITICAL)") help="Уровень отладки (QUIET, NOTSET, DEBUG, INFO, WARNING, ERROR, CRITICAL)")
parser.add_argument("-l", "--loging", default="", parser.add_argument("-l", "--loging", default="",
help="Путь к файлу логирования (если не указан, логирование не будет вестись)") help="Путь к файлу логирования (если не указан, логирование не будет вестись)")
args = parser.parse_args() args = parser.parse_args()
# Настройка отладочных логов # Настройка отладочных логов
# --- Определение числового уровня логирования для передачи в TUI ---
level_input_str = args.debug_level.upper() level_input_str = args.debug_level.upper()
chosen_log_level = None log_level_to_pass = logging.INFO # Значение по умолчанию, если что-то пойдет не так
if level_input_str == "QUIET": # Опция "QUIET"
chosen_log_level = logging.CRITICAL + 1
else:
chosen_log_level = getattr(logging, level_input_str, None)
if chosen_log_level is None:
logger.warning(f"Неизвестный уровень отладки '{args.debug_level}'. Используется DEBUG.")
chosen_log_level = logging.DEBUG
logging.basicConfig( if level_input_str == "QUIET":
level=chosen_log_level, log_level_to_pass = logging.CRITICAL + 1
format="%(asctime)s - %(name)s - %(levelname)s - %(module)s.%(funcName)s:%(lineno)d - %(message)s" else:
) # Пытаемся получить числовое значение для стандартных уровней
logger.debug(f"Параметрами: {args}") parsed_level = getattr(logging, level_input_str, None)
logger.debug(f"Уровень: {chosen_log_level}") if parsed_level is not None and isinstance(parsed_level, int):
log_level_to_pass = parsed_level
else:
# Если уровень не распознан, выводим предупреждение в stderr,
# так как logger.warning может быть еще не настроен для вывода.
print(f"Предупреждение: Неизвестный уровень логирования '{args.debug_level}'. "
f"Используется INFO по умолчанию.", file=sys.stderr)
log_level_to_pass = logging.INFO # Возвращаемся к INFO по умолчанию
# --- Настройка раннего логирования с MemoryHandler ---
# 1. Устанавливаем уровень корневого логгера, чтобы он пропускал нужные сообщения
root_logger = logging.getLogger()
root_logger.setLevel(log_level_to_pass) # Важно для MemoryHandler
# 2. Создаем MemoryHandler
# Он будет копить логи до момента, когда TUI сможет их отобразить.
# `capacity` - сколько записей хранить.
# `flushLevel` - уровень, при котором MemoryHandler попытается сбросить логи в target.
# Ставим очень высокий, чтобы он не сбрасывал сам по себе, т.к. target еще нет.
# `target` - пока None, установим его в TUI.
memory_handler = MemoryHandler(capacity=200, flushLevel=logging.CRITICAL + 10, target=None)
memory_handler.setLevel(logging.NOTSET) # MemoryHandler должен принимать все, что пропускает root_logger
# 3. Добавляем MemoryHandler к корневому логгеру
root_logger.addHandler(memory_handler)
logger.debug(
f"main.py: MemoryHandler добавлен. Log Level: '{logging.getLevelName(root_logger.getEffectiveLevel())}'")
logging.getLogger().setLevel(log_level_to_pass)
logger.debug(f"Запущено с параметрами: {args}")
logger.info(f"Задан уровень логирования: '{logging.getLevelName(log_level_to_pass)}'"
f" ({log_level_to_pass})")
# logging.basicConfig(
# level=chosen_log_level_int,
# format="%(asctime)s - %(name)s - %(levelname)s - %(module)s.%(funcName)s:%(lineno)d - %(message)s"
# )
# Загрузка конфигурации # Загрузка конфигурации
config = None # Инициализируем config
try: try:
config = load_config(args.config) config = load_config(args.config)
logger.info(f"Конфигурация успешно загружена: {config}") if config: # Добавим проверку, что config не None, если load_config может вернуть None при успехе (хотя по типу не должен)
logger.debug(
f"Конфигурация успешно загружена: {list(config.keys())}") # Логируем ключи, а не всю конфигурацию
else:
# Эта ситуация не должна возникать, если load_config выбрасывает исключения при ошибках
# или возвращает dict. Но для полноты.
print(f"Ошибка: Конфигурация не была загружена из '{args.config}', но исключение не было выброшено.",
file=sys.stderr)
sys.exit(1)
except ConfigNotFoundError: except ConfigNotFoundError:
print(f"Ошибка: Файл конфигурации '{args.config}' не найден.", file=sys.stderr) print(f"Ошибка: Файл конфигурации '{args.config}' не найден.", file=sys.stderr)
# Возможно сюда стоит добавить TUI интерфейс для ввода (выбора) файла конфигурации...
# Но так как ошибка может быть связана с правами доступа (например, TUI запущен от неправильного
# пользователя), оставим идею "на-подумать", а пока просто завершим выполнение программы.
sys.exit(1) sys.exit(1)
except InvalidConfigFormatError as e: except InvalidConfigFormatError as e:
print(f"Ошибка: Файл конфигурации '{args.config}' имеет неверный формат. {e}", file=sys.stderr) print(f"Ошибка: Файл конфигурации '{args.config}' имеет неверный формат. {e}", file=sys.stderr)
sys.exit(1) sys.exit(1)
except ConfigError as e: # Общая ошибка конфигурации (должна быть последней в цепочке) except ConfigError as e:
print(f"Ошибка при загрузке конфигурации: {e}", file=sys.stderr) print(f"Ошибка при загрузке конфигурации: {e}", file=sys.stderr)
sys.exit(1) sys.exit(1)
except Exception as e: # Для совсем непредвиденных, не ConfigError except Exception as e:
logger.critical(f"Необработанная ошибка во время инициализации: {e}", exc_info=True) logger.critical(f"Необработанная ошибка во время инициализации: {e}", exc_info=True)
print(f"Произошла критическая и непредвиденная ошибка: {e}", file=sys.stderr) print(f"Произошла критическая и непредвиденная ошибка: {e}", file=sys.stderr)
sys.exit(2) sys.exit(2)
# Запуск главного меню TUI # Запуск главного меню TUI с передачей числового уровня логирования
PGanecApp().run() # app = PGanecApp(log_level_int=log_level_to_pass) # Передаем и конфигурацию
app = PGanecApp(log_level_int=log_level_to_pass, app_config=config, early_log_handler=memory_handler)
app.run()
# PGanecApp().run()

View File

@ -2,73 +2,123 @@
# src/pganec/tui.py # src/pganec/tui.py
import logging import logging
from logging.handlers import MemoryHandler
from typing import Dict, Any, Optional
from textual.app import App, ComposeResult from textual.app import App, ComposeResult
from textual.containers import Vertical, Horizontal from textual.containers import Vertical, Horizontal
from textual.widgets import Button, Header, Footer, Static from textual.widgets import Button, Header, Footer, Static, Log
# --- Настройки и инициирование логирования модуля ---
logger = logging.getLogger(__name__) 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): class MainMenu(Static):
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
yield Horizontal( yield Horizontal(
Button(label="Backup (DB -> Dump)", id="backup", variant="primary"), Button(label="Backup (DB -> Dump)", id="backup"),
Button(label="Restore (Dump -> DB)", id="restore", variant="primary"), Button(label="Restore (Dump -> DB)", id="restore"),
Button(label="Copy (DB -> DB)", id="copy", variant="error"), Button(label="Copy (DB -> DB)", id="copy", variant="error"),
Button(label="Quit", id="quit", variant="error"), Button(label="Quit", id="quit", variant="error"),
id="menu", id="menu",
) )
yield Footer()
class PGanecApp(App): class PGanecApp(App):
CSS = """ CSS = """
#menu { Screen {
width: 100%; layout: vertical; /* Используем вертикальный layout для всего экрана */
align: left top;
padding: 0;
} }
#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 { Button {
align: left middle; align: center middle;
text-align: left;
width: 25%; width: 25%;
color: orange; color: orange;
background: transparent; background: transparent;
margin: 0; margin: 0; padding: 0;
padding: 0;
} }
Button.-primary { Button#backup, Button#restore { color: orange; border: round orange; }
color: orange;
border: round orange;
text-align: left;
}
Button.-error { Button#copy, Button#quit { color: grey; border: round orange; }
color: grey;
border: round orange;
text-align: left;
}
Button:focus { Button:focus { border: heavy green; }
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%; }
Footer { dock: bottom; height: 1; }
Header { dock: top; height: 1; }
""" """
BINDINGS = [ BINDINGS = [
("b", "backup", "Backup"), ("b", "backup", "Backup"),
("r", "restore", "Restore"), ("r", "restore", "Restore"),
("с", "copy", "Copy"), ("c", "copy", "Copy"),
("q", "quit", "Quit"), ("q", "quit", "Quit"),
("right", "focus_next"), ("down", "focus_next"), ("right", "focus_next"), ("down", "focus_next"), # Стрелочки для навигации (вниз/вправо -- вперед)
("left", "focus_previous"), ("up", "focus_previous"), ("left", "focus_previous"), ("up", "focus_previous"), # Стрелочки для навигации (вверх/влево -- назад)
("enter", "activate"), ("enter", "activate", "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: def compose(self) -> ComposeResult:
yield Header() yield Header()
yield Static("PGanec", id="title") with Vertical(id="main_content_area"):
yield Static("PGanec TUI", id="title")
yield MainMenu() yield MainMenu()
# Другие элементы основного интерфейса могут быть здесь
yield Log(id="app_log_viewer", highlight=True) # Виджет лога
yield Footer() yield Footer()
async def on_button_pressed(self, event: Button.Pressed) -> None: async def on_button_pressed(self, event: Button.Pressed) -> None:
@ -82,6 +132,48 @@ class PGanecApp(App):
elif button_id == "quit": elif button_id == "quit":
await self.action_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.debug("TUI: DEBUG сообщение для проверки")
logger.info("TUI-логгер инициализирован")
logger.warning("TUI: WARNING для проверки")
logger.error("TUI: ERROR тоже для проверки")
async def action_activate(self) -> None: async def action_activate(self) -> None:
focused = self.focused focused = self.focused
if isinstance(focused, Button): if isinstance(focused, Button):