Исходный код EJIO.ctk_extender.notification.desktop_controller

## /ctk_extender/notification/desktop_controller.py

import customtkinter as ctk
from threading import Lock
from loguru import logger
from typing import Any

from EJIO.utils import StaticClass, Queue, Disposable
from .base import BaseNotification
from .flags import StyleFlags
from .position import get_position_strategy

__all__ = [
    'DesktopNotificationController',
]




class DesktopNotificationController(StaticClass, Disposable):
    """
    Статический контроллер внешних ООП-уведомлений рабочего стола.
    Использует потокобезопасную Queue буферизации для FIFO-ротации.
    """
    __slots__ = ()

    _LOCK: Lock = Lock()
    _max_visible: int = 3
    _active_nodes: list[tuple[ctk.CTkToplevel, BaseNotification]] = []
    _queue: Queue[tuple[type[BaseNotification], StyleFlags, str, float, dict[str, Any]]] = Queue()



    @classmethod
    def push(
            cls,
            notification_cls: type[BaseNotification],
            flags: StyleFlags = StyleFlags.DEFAULT,
            position: str = "top_right",
            duration: float = 4.0,
            **kwargs: Any
    ) -> None:
        """
        Потокобезопасно пушит ООП-карточку на рабочий стол ОС или в Queue буфер.

        :param notification_cls: Класс уведомления.
        :param flags: Флаги стиля уведомления.
        :param position: Позиция уведомления.
        :param duration: Длительность показа уведомления в с.
        :param kwargs: Дополнительные параметры инициализации класса уведомления.
        """
        with cls._LOCK:
            if len(cls._active_nodes) < cls._max_visible:
                cls._display(notification_cls, flags, position, duration, kwargs)
            else:
                cls._queue.push((notification_cls, flags, position, duration, kwargs))


    @classmethod
    def _display(
            cls,
            cls_type: type[BaseNotification],
            flags: StyleFlags,
            pos_name: str,
            dur: float,
            kwargs: dict[str, Any]
    ) -> None:
        """
        Отображает уведомление на экране.

        :param cls_type: Класс уведомления.
        :param flags: Флаги стиля уведомления.
        :param pos_name: Позиция уведомления.
        :param dur: Длительность показа уведомления в с.
        :param kwargs: Дополнительные параметры инициализации класса уведомления.
        """
        # 1. Создаем внешнюю Си-раму Toplevel без системных границ
        win = ctk.CTkToplevel()
        win.overrideredirect(True)
        win.attributes("-topmost", True)
        win.withdraw()  # Скрываем на время конструирования внутренних стилей

        # 2. Инстанцируем переданный ООП-класс уведомления, отдавая win как parent
        widget_instance = cls_type(win, flags=flags, **kwargs)

        # Растягиваем внутренний фрейм виджета на весь размер Toplevel окна
        widget_instance.ctk_instance.pack(fill="both", expand=True)

        cls._active_nodes.append((win, widget_instance))

        # Вычисляем абсолютные экранные координаты геометрии
        pos_strategy = get_position_strategy(pos_name)
        cls._reposition(pos_strategy)

        win.deiconify()
        logger.debug(f"Showing desktop notification {cls_type.__name__}")
        # Планируем каскадное закрытие по таймеру
        win.after(int(dur * 1000), lambda: cls._expire(win, widget_instance, pos_strategy))

    @classmethod
    def _reposition(cls, pos_strategy: Any) -> None:
        """Пересчитывает абсолютные экранные координаты стопки окон."""
        gap = 10
        for idx, (win, widget) in enumerate(reversed(cls._active_nodes)):
            # Вычисляем сдвиг с учетом реальной кастомной высоты каждого переданного виджета
            offset_y = 40 + idx * (widget.height + gap)
            pos_strategy.apply_desktop(win, offset_y, widget.width, widget.height)

    @classmethod
    def _expire(cls, win: ctk.CTkToplevel, widget: BaseNotification, pos_strategy: Any) -> None:
        """Уничтожает отгоревшее окно и вытягивает следующее уведомление из очереди."""
        with cls._LOCK:
            for active_win, active_widget in list(cls._active_nodes):
                if active_win == win:
                    cls._active_nodes.remove((active_win, active_widget))
                    try:
                        active_widget.destroy()
                        active_win.destroy()
                    except Exception:
                        pass

                    # Пересчитываем сдвиг для оставшихся окон на экране
                    cls._reposition(pos_strategy)

                    # Извлекаем отложенное уведомление из Queue буфера
                    if not cls._queue.empty:
                        next_notify = cls._queue.pop()
                        cls_type, flags, pos_name, dur, kwargs = next_notify
                        cls._display(cls_type, flags, pos_name, dur, kwargs)
                    logger.debug(f"Closing desktop notification {widget.__class__.__name__}")
                    break



    @classmethod
    def clear(cls) -> None:
        """Полная выгрузка всех внешних окон при экстренном закрытии приложения."""
        with cls._LOCK:
            for win, widget in list(cls._active_nodes):
                try:
                    widget.destroy()
                    win.destroy()
                except Exception:
                    pass
            cls._active_nodes.clear()
            while not cls._queue.empty:
                cls._queue.pop()