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

## /ctk_extender/notification/in_app_controller.py

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__ = ['InAppNotificationController']




class InAppNotificationController(StaticClass, Disposable):
    """
    Статический контроллер внешних ООП-уведомлений главного окна.
    Использует потокобезопасную Queue буферизации для FIFO-ротации.
    """
    _LOCK: Lock = Lock()
    _max_visible: int = 3
    _active_widgets: list[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 = 3.5,
            **kwargs: Any
    ) -> None:
        """
        Потокобезопасно пушит ООП-карточку на главное окно или в Queue буфер.

        :param notification_cls: Класс уведомления.
        :param flags: Флаги стиля уведомления.
        :param position: Позиция уведомления.
        :param duration: Длительность показа уведомления в с.
        :param kwargs: Дополнительные параметры инициализации класса уведомления.
        """
        with cls._LOCK:
            if len(cls._active_widgets) < 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: Дополнительные параметры инициализации класса уведомления.
        """
        from ..app.base import Application
        from ..window import WindowController

        app = Application()
        root = app.context_root
        if not root:
            logger.warning(f"No root window found for notification {cls_type.__name__}.")
            return

        try:
            active_win = WindowController.get_active(app._main_window_name)
            notification_parent = active_win.active_layout_canvas
        except Exception:
            notification_parent = root

        if "width" not in kwargs:
            kwargs["width"] = 240
        if "height" not in kwargs:
            kwargs["height"] = 46

        # Создаем пустой фон на базе флагов
        node = cls_type(notification_parent, flags=flags, **kwargs)
        cls._active_widgets.append(node)

        try:
            node.ctk_instance.lift()
            notification_parent.update_idletasks()
        except Exception:
            logger.error(f"Lifting in-app notification {cls_type.__name__} failed.")

        # Обновляем геометрию
        pos_strategy = get_position_strategy(pos_name)
        cls._reposition(pos_strategy)

        logger.debug(f"Showing in-app notification {cls_type.__name__}.")

        notification_parent.after(int(dur * 1000), lambda: cls._expire(node, pos_strategy))

    @classmethod
    def _reposition(cls, pos_strategy: Any) -> None:
        """Пересчитывает абсолютные экранные координаты стопки уведомлений."""
        gap = 8
        for idx, widget in enumerate(reversed(cls._active_widgets)):
            try:
                offset_y = 15 + idx * (widget.height + gap)
                pos_strategy.apply_in_app(widget.ctk_instance, offset_y)
            except Exception:
                continue

    @classmethod
    def _expire(cls, widget: BaseNotification, pos_strategy: Any) -> None:
        """Уничтожает отгоревшее уведомление и вытягивает следующее из очереди."""
        with cls._LOCK:
            if widget in cls._active_widgets:
                cls._active_widgets.remove(widget)
                cls._reposition(pos_strategy)

                try:
                    widget.destroy()
                except Exception:
                    pass

                if not cls._queue.empty:
                    next_node = cls._queue.pop()
                    cls_type, flags, pos_name, dur, kwargs = next_node
                    cls._display(cls_type, flags, pos_name, dur, kwargs)
                logger.debug(f"Closing in-app notification {widget.__class__.__name__}")



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