scout/plugins/event.py

Rama main · réplica de lectura del repositorio original.

"""事件系统 - 在插件间通信"""

from typing import List, Callable, Awaitable, Dict, Any
from collections import defaultdict
import asyncio
import logging

logger = logging.getLogger(__name__)


class Event:
    """事件对象"""
    def __init__(self, name: str, data: Dict[str, Any] = None):
        self.name = name
        self.data = data or {}
        self._stop = False
        self.result: Any = None
    
    def stop_propagation(self):
        """停止事件传播"""
        self._stop = True
    
    @property
    def stopped(self) -> bool:
        return self._stop


class EventBus:
    """事件总线 - 管理事件的发布和订阅"""
    
    def __init__(self):
        self._handlers: Dict[str, List[Callable[[Event], Awaitable[None]]]] = defaultdict(list)
        self._history: List[Event] = []
        self._max_history = 100
    
    def on(self, event_name: str):
        """装饰器：注册事件处理器"""
        def decorator(func: Callable[[Event], Awaitable[None]]):
            self._handlers[event_name].append(func)
            return func
        return decorator
    
    def subscribe(self, event_name: str, handler: Callable[[Event], Awaitable[None]]):
        """订阅事件"""
        self._handlers[event_name].append(handler)
    
    def unsubscribe(self, event_name: str, handler: Callable[[Event], Awaitable[None]]):
        """取消订阅"""
        if event_name in self._handlers and handler in self._handlers[event_name]:
            self._handlers[event_name].remove(handler)
    
    async def emit(self, event: Event) -> Event:
        """发布事件"""
        logger.debug(f"发出事件: {event.name}, 数据: {event.data}")
        
        # 记录历史
        self._history.append(event)
        if len(self._history) > self._max_history:
            self._history = self._history[-self._max_history:]
        
        # 调用所有处理器
        handlers = self._handlers.get(event.name, [])
        for handler in handlers:
            if event.stopped:
                break
            
            try:
                await handler(event)
            except Exception as e:
                logger.error(f"事件处理器异常: {handler.__name__} - {e}", exc_info=True)
        
        return event
    
    def emit_sync(self, event: Event) -> Event:
        """同步发布事件（在非异步上下文中使用）"""

        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None

        if loop and loop.is_running():
            # 在异步上下文中，创建任务
            # ★ 2026-09-01：事件循环对任务仅持弱引用，未保存引用的任务可能
            # 在执行中途被 GC 静默丢弃 —— 存入后台任务集，完成后自动移除。
            if not hasattr(self, "_bg_tasks"):
                self._bg_tasks: set = set()
            _t = asyncio.create_task(self.emit(event))
            self._bg_tasks.add(_t)
            _t.add_done_callback(self._bg_tasks.discard)
            return event
        else:
            # 在同步上下文中，运行事件循环
            return asyncio.run(self.emit(event))
    
    async def emit_with_result(self, event: Event) -> Any:
        """发布事件并返回结果"""
        await self.emit(event)
        return event.result
    
    def clear(self, event_name: str = None):
        """清除处理器"""
        if event_name:
            self._handlers[event_name].clear()
        else:
            self._handlers.clear()
    
    def get_history(self, limit: int = 10) -> List[Event]:
        """获取事件历史"""
        return self._history[-limit:] if self._history else []
    
    @property
    def handlers_count(self) -> Dict[str, int]:
        """获取各事件的处理器数量"""
        return {name: len(handlers) for name, handlers in self._handlers.items()}


# 全局事件总线实例
event_bus = EventBus()