Coverage for merco/scheduler/delivery.py: 0%

1"""任务投递""" 

2 

3from typing import Callable 

4 

5 

6class DeliveryManager: 

7 """任务结果投递管理""" 

8 

9 def __init__(self): 

10 self._channels: dict[str, Callable] = {} 

11 

12 def register_channel(self, name: str, handler: Callable): 

13 """注册投递渠道（如 telegram, discord, email 等）""" 

14 self._channels[name] = handler 

15 

16 async def deliver(self, channel: str, message: str, **kwargs): 

17 """投递消息到指定渠道""" 

18 handler = self._channels.get(channel) 

19 if handler is None: 

20 return {"error": f"Channel '{channel}' not found"} 

21 

22 try: 

23 await handler(message, **kwargs) 

24 return {"success": True} 

25 except Exception as e: 

26 return {"error": str(e)} 

27 

28 def list_channels(self) -> list[str]: 

29 """列出可用渠道""" 

30 return list(self._channels.keys())