Lib.core.EventManager

  1"""
  2事件管理器，用于管理事件与事件监听器
  3"""
  4import traceback
  5from typing import Any, TypeVar
  6
  7from collections.abc import Callable
  8from dataclasses import dataclass, field
  9
 10from Lib.core.ThreadPool import async_task
 11from Lib.core import ConfigManager
 12from Lib.utils import Logger
 13import inspect
 14
 15logger = Logger.get_logger()
 16
 17
 18class _Event:
 19    """
 20    请勿使用此事件类，使用Event继承以创建自定义事件
 21    """
 22    pass
 23
 24
 25class Hook(_Event):
 26    """
 27    钩子事件，用于在事件处理过程中跳过某些监听器
 28    """
 29    def __init__(self, event, listener):
 30        self.event = event
 31        self.listener = listener
 32
 33    def call(self):
 34        """
 35        按优先级顺序同步触发所有监听器
 36        """
 37        if self.__class__ in event_listeners:
 38            for listener in sorted(event_listeners[self.__class__], key=lambda i: i.priority, reverse=True):
 39                if not ConfigManager.GlobalConfig().debug.enable:
 40                    try:
 41                        res = listener.func(self, **listener.kwargs)
 42                    except Exception as e:
 43                        logger.error(f"Error occurred in listener: {repr(e)}\n{traceback.format_exc()}")
 44                        continue
 45                else:
 46                    res = listener.func(self, **listener.kwargs)
 47                if res is True:
 48                    return True
 49            return False
 50
 51
 52T = TypeVar('T', bound='_Event')
 53
 54
 55# 定义事件监听器的数据类
 56@dataclass(order=True)
 57class EventListener:
 58    """
 59    事件监听器数据类
 60    """
 61    priority: int  # 优先级，默认为排序的依据
 62    func: Callable[[T, ...], Any]  # 监听器函数
 63    kwargs: dict[str, Any] = field(default_factory=dict)  # 附加参数
 64
 65    def __post_init__(self):
 66        # 确保监听器函数至少有一个参数
 67        assert len(inspect.signature(self.func).parameters) >= 1, "The listener takes at least 1 parameter"
 68
 69
 70# 定义监听器的类型和存储
 71event_listeners: dict[type[T], list[EventListener]] = {}
 72
 73
 74# 装饰器，用于注册监听器
 75def event_listener(event_class: type[T], priority: int = 0, **kwargs):
 76    """
 77    用于注册监听器
 78
 79    Args:
 80        event_class: 事件类型
 81        priority: 优先级，默认为0
 82        **kwargs: 附加参数
 83    """
 84    assert issubclass(event_class, _Event), "Event class must be a subclass of Event"
 85
 86    def wrapper(func: Callable[[T, ...], Any]):
 87        # 注册事件监听器
 88        listener = EventListener(priority=priority, func=func, kwargs=kwargs)
 89        event_listeners.setdefault(event_class, []).append(listener)
 90        return func
 91
 92    return wrapper
 93
 94
 95class Event(_Event):
 96    """
 97    基事件类，所有自定义事件均继承自此类，继承自此类以创建自定义事件
 98    """
 99
100    def _call_hook(self, listener):
101        return Hook(self, listener).call()
102
103    def call(self):
104        """
105        按优先级顺序同步触发所有监听器
106        """
107        if self.__class__ in event_listeners:
108            res_list = []
109            for listener in sorted(event_listeners[self.__class__], key=lambda i: i.priority, reverse=True):
110                if self._call_hook(listener):
111                    logger.debug(f"Skipped listener: {listener.func.__name__}")
112                    continue
113                if not ConfigManager.GlobalConfig().debug.enable:
114                    try:
115                        res = listener.func(self, **listener.kwargs)
116                    except Exception as e:
117                        logger.error(f"Error occurred in listener: {repr(e)}\n{traceback.format_exc()}")
118                        continue
119                else:
120                    res = listener.func(self, **listener.kwargs)
121                res_list.append(res)
122
123    @async_task
124    def call_async(self):
125        """
126        无需等待的异步按优先级顺序触发所有监听器
127        """
128        self.call()
129
130
131if __name__ == "__main__":
132    # 示例：自定义事件
133    """
134    class MyEvent(Event):
135        def __init__(self, message):
136            self.message = message
137
138
139    # 监听器函数
140    @event_listener(MyEvent, priority=10, other_message="priority is 10")
141    @event_listener(MyEvent, priority=100, other_message="priority is 100")
142    @event_listener(MyEvent, other_message="I'm going to be skipped")
143    def on_my_event(event, other_message=""):
144        print(f"Received event: {event.message}!", other_message)
145
146
147    @event_listener(Hook)
148    def on_hook(event):
149        if event.event.__class__ == MyEvent and event.listener.kwargs["other_message"] == "I'm going to be skipped":
150            return True
151
152
153    # 触发事件
154    event = MyEvent("Hello, World")
155    event.call()
156"""