ayaka.constant

 1'''所有上下文变量'''
 2from contextvars import ContextVar
 3from collections import defaultdict
 4from typing import List, TYPE_CHECKING, Dict
 5from .driver import Message, MessageSegment, Bot, MessageEvent, get_driver
 6
 7
 8if TYPE_CHECKING:
 9    from .ayaka import AyakaApp, AyakaGroup
10
11
12_bot: ContextVar[Bot] = ContextVar("_bot")
13_event: ContextVar[MessageEvent] = ContextVar("_event")
14_group: ContextVar["AyakaGroup"] = ContextVar("_group")
15_arg: ContextVar[Message] = ContextVar("_arg")
16_args: ContextVar[List[MessageSegment]] = ContextVar("_args")
17_message: ContextVar[Message] = ContextVar("_message")
18_cmd: ContextVar[str] = ContextVar("_cmd")
19_enter_exit_during: ContextVar[int] = ContextVar(
20    "_enter_exit_during", default=0)
21
22app_list: List["AyakaApp"] = []
23group_list: List["AyakaGroup"] = []
24bot_list: List[Bot] = []
25
26# 监听私聊
27private_listener_dict: Dict[int, List[int]] = defaultdict(list)
28
29
30first_bot_connect = True
31driver = get_driver()
32
33
34@driver.on_bot_connect
35async def bot_connect(bot: Bot):
36    bot_list.append(bot)
37
38    # 在第一个bot就绪后，开启插件中的定时模块
39    global first_bot_connect
40    if first_bot_connect:
41        first_bot_connect = False
42        for app in app_list:
43            for t in app.timers:
44                t.start()
45
46
47@driver.on_bot_disconnect
48async def bot_disconnect(bot: Bot):
49    bot_list.remove(bot)