🚧 sunset get_bot

This commit is contained in:
felinae98
2023-05-08 21:09:28 +08:00
parent 39c045c63f
commit 5010ca1ac5
7 changed files with 30 additions and 172 deletions
+2 -68
View File
@@ -1,18 +1,11 @@
""" 提供获取 Bot 的方法 """
import random
from collections import defaultdict
from typing import Any, Optional
from typing import Any
import nonebot
from nonebot import get_driver, on_notice
from nonebot.adapters import Bot
from nonebot.adapters.onebot.v11 import Bot as Ob11Bot
from nonebot.adapters.onebot.v11 import (
FriendAddNoticeEvent,
GroupDecreaseNoticeEvent,
GroupIncreaseNoticeEvent,
)
from nonebot_plugin_saa import PlatformTarget, TargetQQGroup, TargetQQPrivate
from nonebot_plugin_saa import PlatformTarget
GROUP: dict[int, list[Bot]] = {}
USER: dict[int, list[Bot]] = {}
@@ -29,65 +22,6 @@ def get_bots() -> list[Bot]:
return bots
async def _refresh_ob11(bot: Ob11Bot):
# 获取群列表
groups = await bot.get_group_list()
for group in groups:
group_id = group["group_id"]
target = TargetQQGroup(group_id=group_id)
BOT_CACHE[target].append(bot)
# 获取好友列表
users = await bot.get_friend_list()
for user in users:
user_id = user["user_id"]
target = TargetQQPrivate(user_id=user_id)
BOT_CACHE[target].append(bot)
async def refresh_bots():
"""刷新缓存的 Bot 数据"""
BOT_CACHE.clear()
for bot in get_bots():
match bot:
case Ob11Bot():
await _refresh_ob11(bot)
driver = get_driver()
@driver.on_bot_connect
@driver.on_bot_disconnect
async def _(bot: Bot):
await refresh_bots()
change_notice = on_notice(priority=1)
@change_notice.handle()
async def _(bot: Bot, event: FriendAddNoticeEvent):
await refresh_bots()
# 01-06 16:56:51 [SUCCESS] nonebot | OneBot V11 **** | [notice.group_increase.approve]: {'time': 1672995411, 'self_id': ****, 'post_type': 'notice', 'notice_type': 'group_increase', 'sub_type': 'approve', 'user_id': ****, 'group_id': ****, 'operator_id': 0}
# 01-06 16:58:09 [SUCCESS] nonebot | OneBot V11 **** | [notice.group_decrease.kick_me]: {'time': 1672995489, 'self_id': ****, 'post_type': 'notice', 'notice_type': 'group_decrease', 'sub_type': 'kick_me', 'user_id': ****, 'group_id': ****, 'operator_id': ****}
@change_notice.handle()
async def _(bot: Bot, event: GroupDecreaseNoticeEvent | GroupIncreaseNoticeEvent):
if bot.self_id == event.user_id:
await refresh_bots()
def get_bot(user: PlatformTarget) -> Optional[Bot]:
"""获取 Bot"""
bots = BOT_CACHE.get(user)
if not bots:
return
return random.choice(bots)
async def get_groups() -> list[dict[str, Any]]:
"""获取所有群号"""
# TODO