Как вывести в текстовый фрейм tkinter логи через loggerhandler

У меня есть 2 модуля (UI.py, LoggerHandler.py). При запуске любой кнопки из UI.py получаю событие записанное в файл logs_camera.txt:

ERROR | Не удалось записать сообщение в один из Frame'ов: 'LoggingHandler' object has no attribute 'first_log_text'`

Код модуля LoggerHandler.py

import logging
from datetime import datetime
import tkinter as tk

FORMAT = '%(asctime)s | %(levelname)s | %(message)s'
logging.basicConfig(format=FORMAT, filename="logs_camera.txt", filemode="w")
logger = logging.getLogger(__name__)

class LoggingHandler(logging.Handler):
    def __init__(self, message, frame_number):
        super().__init__()
        self.message = message
        self.frame_number = frame_number

    def set_first_log_text(self, first_log_text):
        self.first_log_text = first_log_text

    def set_secondary_log_text(self, secondary_log_text):
        self.secondary_log_text = secondary_log_text

    def emit(self, record):
        try:
            if self.frame_number == 1:
                text_field = self.first_log_text
            elif self.frame_number == 2:
                text_field = self.secondary_log_text
            
            text_field.configure(state=tk.NORMAL)
            text_field.insert(tk.END, f"{datetime.now()} | {record.getMessage()}\n")
            text_field.configure(state=tk.DISABLED)
        except Exception as e:
            logger.error(f"Не удалось записать сообщение в один из Frame'ов: {e}")

    def add_info_handler(message, frame_number):
        handler = LoggingHandler(message, frame_number)
        handler.setLevel(logging.INFO)
        handler.setFormatter(logging.Formatter('%(asctime)s | %(levelname)s | %(message)s'))
        record = logging.LogRecord(None, logging.INFO, None, 0, message, None, None)
        handler.emit(record)

    def add_warning_handler(message, frame_number):
        handler = LoggingHandler(message, frame_number)
        handler.setLevel(logging.WARNING)
        handler.setFormatter(logging.Formatter('%(asctime)s | %(levelname)s | %(message)s'))
        record = logging.LogRecord(None, logging.WARNING, None, 0, message, None, None)
        handler.emit(record)

    def add_error_handler(message, frame_number):
        handler = LoggingHandler(message, frame_number)
        handler.setLevel(logging.ERROR)
        handler.setFormatter(logging.Formatter('%(asctime)s | %(levelname)s | %(message)s'))
        record = logging.LogRecord(None, logging.ERROR, None, 0, message, None, None)
        handler.emit(record)

Код модуля UI.py

from tkinter import *
from tkinter import ttk
import socket
import threading
import tkinter as tk
from datetime import datetime
from LoggerHandler import LoggingHandler as logger

class App(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title("Makro test camera")
        icon = PhotoImage(file="makrologo.png")
        self.iconphoto(TRUE, icon)  
        self.geometry("1000x500+400+100")
        self.resizable(False, False)

       # Создаем фрейм для логов
        self.log_frame = tk.Frame(self)
        self.log_frame.pack(pady=10)

        # Создаем фрейм для первого лога
        self.first_log_frame = ttk.Frame(self.log_frame)
        self.first_log_frame.pack(side=tk.LEFT, padx=15)
        # Создаем фрейм для второго лога
        self.secondary_log_frame = ttk.Frame(self.log_frame)
        self.secondary_log_frame.pack(side=tk.RIGHT, padx=15)
        
        # Описание блока вывода логов первого фрейма
        self.label_first_log_frame = ttk.Label(self.first_log_frame, text="Cостояние подключений", compound="bottom", font=("Arial", 12))
        self.label_first_log_frame.pack()
        # Описание блока вывода логов первого фрейма
        self.label_secondary_log_frame = ttk.Label(self.secondary_log_frame, text="Трафик моста", compound="bottom", font=("Arial", 12))
        self.label_secondary_log_frame.pack()


        # Текстовое поле для логов первого фрейма
        self.first_log_text = tk.Text(self.first_log_frame, wrap=tk.WORD, height=20, width=60, state=tk.NORMAL)
        self.first_log_text.pack(fill=tk.BOTH, expand=True)
        # Текстовое поле для логов второго фрейма
        self.secondary_log_text = tk.Text(self.secondary_log_frame, wrap=tk.WORD, height=20, width=60, state=tk.NORMAL)
        self.secondary_log_text.pack(fill=tk.BOTH, expand=True)


        # Кнопки запуска сервера и клиента
        self.button_frame = tk.Frame(self)
        self.button_frame.pack(side=tk.BOTTOM, fill=tk.X)

        # Верхняя группа иконок
        self.top_buttons_frame = tk.Frame(self.button_frame)
        self.top_buttons_frame.pack(side=tk.TOP, anchor=tk.W, pady=(5, 10), fill=tk.X)

        self.start_all_service_button = ttk.Button(self.top_buttons_frame, text="Запустить всё", command=self.start_all_services)
        self.start_all_service_button.pack(side=tk.BOTTOM, padx=5, expand=True, fill=tk.X, ipady=10)
        
        # Нижняя группа иконок
        self.bottom_buttons_frame = tk.Frame(self.button_frame)
        self.bottom_buttons_frame.pack(side=tk.BOTTOM, anchor=tk.S, pady=(0, 10), fill=tk.X)

        self.start_server_button = ttk.Button(self.bottom_buttons_frame, text="Запустить Сервер", command=self.start_server)
        self.start_server_button.pack(side=tk.LEFT, padx=5, expand=True, fill=tk.X, ipady=5)

        self.start_client_button = ttk.Button(self.bottom_buttons_frame, text="Запустить Клиент", command=self.start_client)
        self.start_client_button.pack(side=tk.RIGHT, padx=5, expand=True, fill=tk.X, ipady=5)

        # Переменные для хранения потоков
        self.server_thread = None
        self.client_thread = None

    def start_server(self):
        if self.server_thread is None:
            self.server_thread = threading.Thread(target=self.run_server)
            self.server_thread.start()
            self.start_server_button.config(state=tk.DISABLED)  # Деактивируем кнопку
            logger.add_info_handler("Сервер запущен start_server()", 1)
        else:
            logger.add_warning_handler("Сервер уже запущен start_server()", 1)

    def start_client(self):
        if self.client_thread is None:
            self.client_thread = threading.Thread(target=self.run_client)
            self.client_thread.start()
            self.start_client_button.config(state=tk.DISABLED)  # Деактивируем кнопку
            logger.add_info_handler("Клиент запущен start_client()", 1)
        else:
            logger.add_error_handler("Клиент уже запущен start_client()", 1)

    def start_all_services(self):
        if self.client_thread or self.server_thread is None:
        # Запуск оба сервиса
            self.start_client()
            self.start_server()
            self.start_all_service_button.config(state=tk.DISABLED) # Деактивируем кнопку 
            logger.add_info_handler("Все сервисы запущены + ", 1)
        else:
            self.start_all_service_button.config(state=tk.DISABLED) # Деактивируем кнопку 
            logger.add_warning_handler("Один из сервисов уже запущен +", 1)

    def run_server(self):
        server = Server('172.16.10.33', 50500)  # Замените на свои IP и порт
        server.run()
        logger.add_info_handler("Сервер запущен run_server()", 1)

    def run_client(self):
        client = Client('172.16.10.220', 50000)  # Замените на свои IP и порт
        client.run() 
        logger.add_info_handler("Клиент запущен run_client()", 1)

class Server:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.clients = []  # Список для хранения подключенных клиентов

    def run(self):
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.bind((self.host, self.port))
                s.listen()
                logger.add_info_handler(f"Сервер запущен на {self.host}:{self.port}", 1)
                while True:
                    conn, addr = s.accept()
                    self.clients.append(conn)  # Добавляем нового клиента в список
                    logger.add_info_handler(f"Подключено: {addr}", 1)
                    thread = threading.Thread(target=self.handle_connection, args=(conn, addr))
                    thread.start()
        except Exception as e:
            logger.add_error_handler("Ошибка сервера: {}".format(e), 1)

    def send_data(self, data):
        # Отправляем данные всем подключенным клиентам
        for client in self.clients:
            try:
                client.sendall(data)
                logger.add_info_handler(f"Отправлено в ПО: {data}", 2)
            except Exception as e:
                logger.add_error_handler(f"Ошибка отправки данных клиенту: {e}", 1)

    def handle_connection(self, conn, addr):
        while True:
            try:
                data = conn.recv(1024)
                if not data:
                    break
                timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                logger.add_info_handler(f"Получено в обработчик: {timestamp} {data.decode('utf-8')}", 2)
            except Exception as e:
                logger.add_error_handler(f"Ошибка обработки данных в handle_connection: {e} ", 1)
                break

        conn.close()
        self.clients.remove(conn)  # Удаляем клиента из списка
        logger.add_info_handler(f"Соединение с {addr} закрыто", 1)
class Client:
    def __init__(self, camera_ip, camera_port):
        self.camera_ip = camera_ip
        self.camera_port = camera_port

    def run(self):
        global global_data
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.connect((self.camera_ip, self.camera_port))
                while True:
                    data = s.recv(1024)
                    if not data:
                        break
                    global_data = data  # Используем ключевое слово global
                    # Отправляем данные на сервер
                    Server.send_data(global_data)  # Используем метод send_data сервера
                    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                    logger.add_info_handler(f"Получено от TCP сервера камеры: {timestamp} {data.decode('utf-8')}", 2, self)
        except Exception as e:
            logger.add_error_handler("Ошибка подключения клиента к TCP серверу камеры:".format(e), 1)

        finally:
            s.close()
            logger.add_info_handler(f"Соединение клиента с сервером {self.camera_ip, self.camera_port} закрыто", 1)


if __name__ == "__main__":
    app = App()
    app.mainloop()

Ответы (1 шт):

Автор решения: Amgarak

Каждый вызов метода [add_info_handler, add_warning_handler, add_error_handler] создает новый обработчик для записи каждого сообщения. В этом совершенно нет необходимости, вам будет достаточно всего одного обработчика.

За основу взял свой старый пример и немного адаптировал его для вашего случая:

import logging
import tkinter as tk
from tkinter.scrolledtext import ScrolledText
from typing import List

class TextHandler(logging.Handler):
    """Переопределяем вывод оброботчика"""
    def __init__(self, text_widget):
        super().__init__()
        self.text_widget = text_widget
        self.text_widget.config(state=tk.NORMAL)

    def emit(self, record):
        msg = self.format(record)
        self.text_widget.insert(tk.END, msg + '\n')
        self.text_widget.see(tk.END)

class MultiLevelFilter(logging.Filter):
    """Фильтр для разрешения нескольких уровней логирования."""
    def __init__(self, levels: List[int]):
        super().__init__()
        self.levels = levels

    def filter(self, record: logging.LogRecord) -> bool:
        return record.levelno in self.levels

class LoggerManager:
    LEVEL_MAPPING = {
        'DEBUG': logging.DEBUG,
        'INFO': logging.INFO,
        'WARNING': logging.WARNING,
        'ERROR': logging.ERROR,
        'CRITICAL': logging.CRITICAL
    }

    def __init__(self, text_widget, name="TkinterLogger", level=logging.DEBUG):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(level)

        # Настройка обработчика для текстового виджета
        self.text_handler = TextHandler(text_widget)
        self.text_handler.setLevel(level)
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        self.text_handler.setFormatter(formatter)
        self.logger.addHandler(self.text_handler)

        self.log_filter = None  # Для хранения активного фильтра

    def set_filter(self, levels: List[str]):
        """Установка фильтра для логирования по уровням."""
        valid_levels = [self.LEVEL_MAPPING[level] for level in levels if level in self.LEVEL_MAPPING]
        if valid_levels:
            self.clear_filter()
            self.log_filter = MultiLevelFilter(valid_levels)
            self.logger.addFilter(self.log_filter)
            self.logger.info(f"Фильтр установлен для уровней: {', '.join(levels)}")
        else:
            self.logger.error(f"Некорректные уровни логирования: {levels}")

    def clear_filter(self):
        """Очистка фильтра."""
        if self.log_filter:
            self.logger.removeFilter(self.log_filter)
            self.log_filter = None
            self.logger.info("Фильтр логирования очищен")


if __name__ == "__main__":
    
    root = tk.Tk()
    root.title("Логирование в окно")

    log_text = ScrolledText(root, width=80, height=20)
    log_text.pack()

    logger_manager = LoggerManager(log_text)
    logger = logger_manager.logger

    def apply_filters():
        selected_levels = []
        if debug_var.get(): selected_levels.append("DEBUG")
        if info_var.get(): selected_levels.append("INFO")
        if warning_var.get(): selected_levels.append("WARNING")
        if error_var.get(): selected_levels.append("ERROR")
        if critical_var.get(): selected_levels.append("CRITICAL")
        logger_manager.set_filter(selected_levels)
        
    def print_logger():
        logger.debug("Пример сообщения DEBUG")
        logger.info("Пример сообщения INFO")
        logger.warning("Пример сообщения WARNING")
        logger.error("Пример сообщения ERROR")
        logger.critical("Пример сообщения CRITICAL")
        
    def clear_text():
        log_text.delete(1.0, 'end')
        
    debug_var = tk.IntVar(value=1)
    info_var = tk.IntVar(value=1)
    warning_var = tk.IntVar(value=1)
    error_var = tk.IntVar(value=1)
    critical_var = tk.IntVar(value=1)

    tk.Checkbutton(root, text="DEBUG", variable=debug_var).pack(anchor=tk.W)
    tk.Checkbutton(root, text="INFO", variable=info_var).pack(anchor=tk.W)
    tk.Checkbutton(root, text="WARNING", variable=warning_var).pack(anchor=tk.W)
    tk.Checkbutton(root, text="ERROR", variable=error_var).pack(anchor=tk.W)
    tk.Checkbutton(root, text="CRITICAL", variable=critical_var).pack(anchor=tk.W)
    
    button_frame = tk.Frame(root)
    button_frame.pack()

    apply_button = tk.Button(button_frame, text="Применить фильтры", command=apply_filters)
    apply_button.pack(side=tk.LEFT, padx=10)

    logger_button = tk.Button(button_frame, text="Проверить логгер", command=print_logger)
    logger_button.pack(side=tk.LEFT, padx=10)

    clear_button = tk.Button(button_frame, text="Очистить текст", command=clear_text)
    clear_button.pack(side=tk.LEFT, padx=10)
    
    root.mainloop()

введите сюда описание изображения

→ Ссылка