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
 7if TYPE_CHECKING:
 8    from .ayaka import AyakaApp
 9    from .group import 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
20app_list: List["AyakaApp"] = []
21group_list: List["AyakaGroup"] = []
22bot_list: List[Bot] = []
23
24# 监听私聊
25private_listener_dict: Dict[int, List[int]] = defaultdict(list)
26
27
28def get_bot(bot_id: int):
29    '''获取已连接的bot'''
30    bot_id = str(bot_id)
31    for bot in bot_list:
32        if bot.self_id == bot_id:
33            return bot
34
35
36def get_app(app_name: str):
37    for app in app_list:
38        if app.name == app_name:
39            return app
40
41
42first_bot_connect = True
43driver = get_driver()
44
45
46@driver.on_startup
47async def startup():
48    # 可有可无
49    app_list.sort(key=lambda x: x.name)
50
51
52@driver.on_bot_connect
53async def bot_connect(bot: Bot):
54    bot_list.append(bot)
55
56    # 在第一个bot就绪后，开启插件中的定时模块
57    global first_bot_connect
58    if first_bot_connect:
59        first_bot_connect = False
60        for app in app_list:
61            for t in app.timers:
62                t.start()
63
64
65@driver.on_bot_disconnect
66async def bot_disconnect(bot: Bot):
67    bot_list.remove(bot)
def get_bot(bot_id: int): 
29def get_bot(bot_id: int):
30    '''获取已连接的bot'''
31    bot_id = str(bot_id)
32    for bot in bot_list:
33        if bot.self_id == bot_id:
34            return bot

获取已连接的bot

def get_app(app_name: str): 
37def get_app(app_name: str):
38    for app in app_list:
39        if app.name == app_name:
40            return app
@driver.on_startup
async def startup(): 
47@driver.on_startup
48async def startup():
49    # 可有可无
50    app_list.sort(key=lambda x: x.name)
@driver.on_bot_connect
async def bot_connect(bot: nonebot.adapters.onebot.v11.bot.Bot): 
53@driver.on_bot_connect
54async def bot_connect(bot: Bot):
55    bot_list.append(bot)
56
57    # 在第一个bot就绪后，开启插件中的定时模块
58    global first_bot_connect
59    if first_bot_connect:
60        first_bot_connect = False
61        for app in app_list:
62            for t in app.timers:
63                t.start()
@driver.on_bot_disconnect
async def bot_disconnect(bot: nonebot.adapters.onebot.v11.bot.Bot): 
66@driver.on_bot_disconnect
67async def bot_disconnect(bot: Bot):
68    bot_list.remove(bot)