自动同意小团体的好友请求

konabot/common/nb/wzq_conflict.py

@@ -0,0 +1,34 @@
+from typing import cast
+from nonebot import get_bot, get_plugin_config, logger
+from nonebot.adapters import Event as BaseEvent
+from nonebot.adapters.onebot.v11.event import GroupMessageEvent
+from nonebot.adapters.onebot.v11.bot import Bot as OnebotBot
+from nonebot.rule import Rule
+from pydantic import BaseModel
+
+
+class WZQConflictConfig(BaseModel):
+    wzq_bot_qq: int = 0
+
+config = get_plugin_config(WZQConflictConfig)
+
+
+async def no_wzqbot(evt: BaseEvent):
+    if config.wzq_bot_qq <= 0:
+        return True
+    if not isinstance(evt, GroupMessageEvent):
+        return True
+    gid = evt.group_id
+    sid = evt.self_id
+    bot = cast(OnebotBot, get_bot(str(sid)))
+
+    members = await bot.get_group_member_list(group_id=gid)
+
+    members = set((m.get("user_id", -1) for m in members))
+    if config.wzq_bot_qq in members:
+        return False
+    return True
+
+
+no_wzqbot_rule = Rule(no_wzqbot)
+

konabot/plugins/auto_accept.py

@@ -0,0 +1,25 @@
+import asyncio
+import random
+from typing import cast
+
+from loguru import logger
+from nonebot import get_bot, on_request
+from nonebot.adapters.onebot.v11.event import FriendRequestEvent
+from nonebot.adapters.onebot.v11.bot import Bot as OnebotBot
+
+from konabot.common.nb.is_admin import cfg as adminConfig
+
+add_request = on_request()
+
+@add_request.handle()
+async def _(req: FriendRequestEvent):
+    bot = cast(OnebotBot, get_bot(str(req.self_id)))
+    ok_member_ls: set[int] = set()
+    for group in adminConfig.admin_qq_group:
+        members = await bot.get_group_member_list(group_id=group)
+        ok_member_ls |= cast(set[int], set((m.get("user_id") for m in members)))
+    if req.user_id in ok_member_ls:
+        await asyncio.sleep(random.randint(5, 10))
+        await req.approve(bot)
+        logger.info(f"已经自动同意 {req.user_id} 的好友请求")
+

konabot/plugins/gen_qrcode/__init__.py

@@ -8,6 +8,8 @@ from nonebot_plugin_alconna import (Alconna, Args, Field, MultiVar, UniMessage,
                                     on_alconna)
 from nonebot_plugin_alconna.uniseg import UniMsg, At, Reply
 
+from konabot.common.nb.wzq_conflict import no_wzqbot_rule
+
 async def download_img(url):
     resp = requests.get(url.replace("https://multimedia.nt.qq","http://multimedia.nt.qq")) # bim获取QQ的图片时避免SSLv3报错
     img_bytes = BytesIO()
@@ -42,7 +44,7 @@ gqrc = on_alconna(Alconna(
         missing_tips=lambda: "请输入你要转换为二维码的文字！"
     )],
     # UniMessage[]
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"生成二维码","genqrcode"})
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"生成二维码","genqrcode"}, rule=no_wzqbot_rule)
 
 @gqrc.handle()
 async def _(saying: list):
@@ -66,4 +68,4 @@ async def _(saying: list):
     else:
     """
         # genqr("\n".join(saying))
-    await gqrc.send(await UniMessage().image(raw=genqr("\n".join(saying))).export())
+    await gqrc.send(await UniMessage().image(raw=genqr("\n".join(saying))).export())

konabot/plugins/poll/__init__.py

@@ -1,8 +1,10 @@
 import json, time
 
+from nonebot.rule import Rule
 from nonebot_plugin_alconna import Alconna, Args, Field, MultiVar, on_alconna
 from nonebot.adapters.onebot.v11 import Event
 
+from konabot.common.nb.wzq_conflict import no_wzqbot_rule
 from konabot.common.path import ASSETS_PATH, DATA_PATH
 
 
@@ -66,7 +68,7 @@ poll = on_alconna(Alconna(
     Args["saying", MultiVar(str, '+'), Field(
         missing_tips=lambda: "参数错误。用法：发起投票 <投票标题> <选项1> <选项2> ..."
     )],
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"发起投票","createpoll"})
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"发起投票","createpoll"}, rule=no_wzqbot_rule)
 @poll.handle()
 async def _(saying: list, event: Event):
     if (len(saying) < 3):
@@ -88,7 +90,7 @@ viewpoll = on_alconna(Alconna(
     Args["saying", MultiVar(str, '+'), Field(
         missing_tips=lambda: "请指定投票ID或标题！。用法：查看投票 <投票ID或标题>"
     )],
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"查看投票"})
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"查看投票"}, rule=no_wzqbot_rule)
 @viewpoll.handle()
 async def _(saying: list):
     # 参数，投票ID或者标题
@@ -130,7 +132,7 @@ vote = on_alconna(Alconna(
     Args["saying", MultiVar(str, '+'), Field(
         missing_tips=lambda: "参数错误。用法：投票 <投票ID/标题> <选项文本>"
     )],
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"投票","参与投票"})
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"投票","参与投票"}, rule=no_wzqbot_rule)
 @vote.handle()
 async def _(saying: list, event: Event):
     if (len(saying) < 2):