playwrite/gui/utils/state_manager.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
State Manager - 状态管理器

实现简单的观察者模式，用于在组件间共享状态
"""
from typing import Any, Callable, Dict, List, Set
from threading import Lock
import logging

logger = logging.getLogger(__name__)


class StateManager:
    """
    简单的状态管理器

    使用观察者模式，允许组件订阅状态变更通知

    Usage:
        # 创建全局实例
        state = StateManager()

        # 设置状态
        state.set("current_user", "admin")

        # 获取状态
        user = state.get("current_user")

        # 订阅状态变更
        def on_user_change(new_user):
            print(f"User changed to: {new_user}")

        state.subscribe("current_user", on_user_change)

        # 取消订阅
        state.unsubscribe("current_user", on_user_change)
    """

    def __init__(self):
        """初始化状态管理器"""
        self._state: Dict[str, Any] = {}
        self._listeners: Dict[str, List[Callable]] = {}
        self._lock = Lock()

    def set(self, key: str, value: Any, notify: bool = True):
        """
        设置状态并通知监听器

        Args:
            key: 状态键
            value: 状态值
            notify: 是否通知监听器（默认 True）
        """
        with self._lock:
            # 检查值是否实际变更
            if key in self._state and self._state[key] == value:
                return

            # 更新状态
            old_value = self._state.get(key)
            self._state[key] = value

            logger.debug(f"State changed: {key} = {value} (was: {old_value})")

        # 通知监听器（在锁外部执行，避免死锁）
        if notify:
            self._notify(key, value, old_value)

    def get(self, key: str, default: Any = None) -> Any:
        """
        获取状态

        Args:
            key: 状态键
            default: 默认值（如果键不存在）

        Returns:
            状态值或默认值
        """
        with self._lock:
            return self._state.get(key, default)

    def delete(self, key: str):
        """
        删除状态

        Args:
            key: 状态键
        """
        with self._lock:
            if key in self._state:
                del self._state[key]
                logger.debug(f"State deleted: {key}")

    def subscribe(self, key: str, callback: Callable[[Any], None]):
        """
        订阅状态变更

        Args:
            key: 状态键
            callback: 回调函数，接收新值作为参数
        """
        with self._lock:
            if key not in self._listeners:
                self._listeners[key] = []
            self._listeners[key].append(callback)
            logger.debug(f"New subscriber for {key}: {callback.__name__}")

    def unsubscribe(self, key: str, callback: Callable[[Any], None]):
        """
        取消订阅

        Args:
            key: 状态键
            callback: 要移除的回调函数
        """
        with self._lock:
            if key in self._listeners:
                try:
                    self._listeners[key].remove(callback)
                    logger.debug(f"Unsubscribed from {key}: {callback.__name__}")

                    # 如果没有监听器了，删除键
                    if not self._listeners[key]:
                        del self._listeners[key]
                except ValueError:
                    logger.warning(f"Callback not found in subscribers for {key}")

    def subscribe_all(self, callback: Callable[[str, Any, Any], None]):
        """
        订阅所有状态变更

        回调函数签名：callback(key, new_value, old_value)

        Args:
            callback: 回调函数
        """
        # 使用特殊的键来存储"全部"监听器
        with self._lock:
            special_key = "__all__"
            if special_key not in self._listeners:
                self._listeners[special_key] = []
            self._listeners[special_key].append(callback)
            logger.debug(f"New subscriber for all changes: {callback.__name__}")

    def unsubscribe_all(self, callback: Callable[[str, Any, Any], None]):
        """
        取消订阅所有状态变更

        Args:
            callback: 要移除的回调函数
        """
        special_key = "__all__"
        with self._lock:
            if special_key in self._listeners:
                try:
                    self._listeners[special_key].remove(callback)
                    logger.debug(f"Unsubscribed from all changes: {callback.__name__}")

                    if not self._listeners[special_key]:
                        del self._listeners[special_key]
                except ValueError:
                    logger.warning(f"Callback not found in all subscribers")

    def _notify(self, key: str, new_value: Any, old_value: Any):
        """
        通知所有订阅者

        Args:
            key: 状态键
            new_value: 新值
            old_value: 旧值
        """
        with self._lock:
            # 获取该键的监听器
            listeners = self._listeners.get(key, []).copy()

            # 获取"全部"监听器
            all_listeners = self._listeners.get("__all__", []).copy()

        # 在锁外部调用回调，避免死锁
        for callback in listeners:
            try:
                callback(new_value)
            except Exception as e:
                logger.error(f"Error in state change listener for {key}: {e}", exc_info=True)

        for callback in all_listeners:
            try:
                callback(key, new_value, old_value)
            except Exception as e:
                logger.error(f"Error in all-state listener for {key}: {e}", exc_info=True)

    def get_all(self) -> Dict[str, Any]:
        """
        获取所有状态的副本

        Returns:
            包含所有状态的字典
        """
        with self._lock:
            return self._state.copy()

    def clear(self):
        """清空所有状态"""
        with self._lock:
            self._state.clear()
            self._listeners.clear()
            logger.debug("All state cleared")

    def has_key(self, key: str) -> bool:
        """
        检查是否存在指定键

        Args:
            key: 状态键

        Returns:
            True 如果键存在，False 否则
        """
        with self._lock:
            return key in self._state


# 全局状态管理器实例
_global_state_manager: StateManager = None


def get_global_state_manager() -> StateManager:
    """
    获取全局状态管理器实例（单例模式）

    Returns:
        全局 StateManager 实例
    """
    global _global_state_manager
    if _global_state_manager is None:
        _global_state_manager = StateManager()
    return _global_state_manager