feat✨: add coser

zhenxun/builtin_plugins/hooks/withdraw_hook.py

@@ -1,46 +1,30 @@
 import asyncio
-from typing import Optional
 
 from nonebot.adapters import Bot
-from nonebot.adapters.discord import Bot as DiscordBot
-from nonebot.adapters.dodo import Bot as DodoBot
-from nonebot.adapters.kaiheila import Bot as KaiheilaBot
-from nonebot.adapters.onebot.v11 import Bot as v11Bot
-from nonebot.adapters.onebot.v12 import Bot as v12Bot
 from nonebot.matcher import Matcher
 from nonebot.message import run_postprocessor
 
-from zhenxun.services.log import logger
-from zhenxun.utils.utils import WithdrawManager
-
-# TODO: 其他平台撤回消息
+from zhenxun.utils.withdraw_manage import WithdrawManager
 
 
-# 消息撤回
 @run_postprocessor
 async def _(
     matcher: Matcher,
-    exception: Optional[Exception],
+    exception: Exception | None,
     bot: Bot,
 ):
     tasks = []
-    for message_id in WithdrawManager._data:
-        second = WithdrawManager._data[message_id]
-        tasks.append(asyncio.ensure_future(_withdraw_message(bot, message_id, second)))
-        WithdrawManager.remove(message_id)
+    index_list = list(WithdrawManager._data.keys())
+    for index in index_list:
+        (
+            bot,
+            message_id,
+            time,
+        ) = WithdrawManager._data[index]
+        tasks.append(
+            asyncio.ensure_future(
+                WithdrawManager.withdraw_message(bot, message_id, time)
+            )
+        )
+        WithdrawManager.remove(index)
     await asyncio.gather(*tasks)
-
-
-async def _withdraw_message(bot: Bot, message_id: str, time: int):
-    await asyncio.sleep(time)
-    logger.debug(f"撤回消息ID: {message_id}", "HOOK")
-    if isinstance(bot, v11Bot):
-        await bot.delete_msg(message_id=int(message_id))
-    elif isinstance(bot, v12Bot):
-        await bot.delete_message(message_id=message_id)
-    elif isinstance(bot, DodoBot):
-        pass
-    elif isinstance(bot, KaiheilaBot):
-        pass
-    elif isinstance(bot, DiscordBot):
-        pass

zhenxun/plugins/coser.py

@@ -0,0 +1,91 @@
+import time
+from typing import Tuple
+
+from nonebot.adapters import Bot
+from nonebot.params import RegexGroup
+from nonebot.plugin import PluginMetadata
+from nonebot_plugin_alconna import Alconna, Args, Arparma, Match, on_alconna
+from nonebot_plugin_saa import Image, Text
+from nonebot_plugin_session import EventSession
+
+from zhenxun.configs.config import Config
+from zhenxun.configs.path_config import TEMP_PATH
+from zhenxun.configs.utils import PluginExtraData, RegisterConfig
+from zhenxun.services.log import logger
+from zhenxun.utils.http_utils import AsyncHttpx
+from zhenxun.utils.withdraw_manage import WithdrawManager
+
+__plugin_meta__ = PluginMetadata(
+    name="coser",
+    description="三次元也不戳，嘿嘿嘿",
+    usage="""
+    ?N连cos/coser
+    示例: cos
+    示例: 5连cos （单次请求张数小于9）
+    """.strip(),
+    extra=PluginExtraData(
+        author="HibiKier",
+        version="0.1",
+        configs=[
+            RegisterConfig(
+                key="WITHDRAW_COS_MESSAGE",
+                value=(0, 1),
+                help="自动撤回，参1：延迟撤回色图时间(秒)，0 为关闭 | 参2：监控聊天类型，0(私聊) 1(群聊) 2(群聊+私聊)",
+                default_value=(0, 1),
+                type=Tuple[int, int],
+            ),
+        ],
+    ).dict(),
+)
+
+_matcher = on_alconna(Alconna("get-cos", Args["num", int, 1]), priority=5, block=True)
+
+_matcher.shortcut(
+    r"cos",
+    command="get-cos",
+    arguments=["1"],
+    prefix=True,
+)
+
+_matcher.shortcut(
+    r"(?P<num>\d)(张|个|条|连)cos",
+    command="get-cos",
+    arguments=["{num}"],
+    prefix=True,
+)
+
+
+# 纯cos，较慢:https://picture.yinux.workers.dev
+# 比较杂，有福利姬，较快:https://api.jrsgslb.cn/cos/url.php?return=img
+url = "https://picture.yinux.workers.dev"
+
+
+@_matcher.handle()
+async def _(
+    bot: Bot,
+    session: EventSession,
+    arparma: Arparma,
+    num: int,
+):
+    withdraw_time = Config.get_config("coser", "WITHDRAW_COS_MESSAGE")
+    for _ in range(num):
+        path = TEMP_PATH / f"cos_cc{int(time.time())}.jpeg"
+        try:
+            await AsyncHttpx.download_file(url, path)
+            receipt = await Image(path).send()
+            message_id = receipt.extract_message_id().dict().get("message_id")
+            if message_id and WithdrawManager.check(session, withdraw_time):
+                WithdrawManager.append(
+                    bot,
+                    message_id,
+                    withdraw_time[0],
+                )
+            logger.info(f"发送cos", arparma.header_result, session=session)
+        except Exception as e:
+            await Text("你cos给我看！").send()
+            logger.error(
+                f"cos错误",
+                arparma.header_result,
+                session=session,
+                e=e,
+            )

zhenxun/utils/utils.py

@@ -12,34 +12,6 @@ import pytz
 from zhenxun.services.log import logger
 
 
-class WithdrawManager:
-    """
-    消息撤回
-    """
-
-    _data = {}
-
-    @classmethod
-    def append(cls, message_id: str, second: int):
-        """添加一个撤回消息id和时间
-
-        参数:
-            message_id: 撤回消息id
-            time: 延迟时间
-        """
-        cls._data[message_id] = second
-
-    @classmethod
-    def remove(cls, message_id: str):
-        """删除一个数据
-
-        参数:
-            message_id: 撤回消息id
-        """
-        if message_id in cls._data:
-            del cls._data[message_id]
-
-
 class ResourceDirManager:
     """
     临时文件管理器

zhenxun/utils/withdraw_manage.py

@@ -0,0 +1,90 @@
+import asyncio
+
+from nonebot.adapters import Bot
+from nonebot.adapters.discord import Bot as DiscordBot
+from nonebot.adapters.dodo import Bot as DodoBot
+from nonebot.adapters.kaiheila import Bot as KaiheilaBot
+from nonebot.adapters.onebot.v11 import Bot as v11Bot
+from nonebot.adapters.onebot.v12 import Bot as v12Bot
+from nonebot_plugin_session import EventSession
+
+from zhenxun.services.log import logger
+
+
+class WithdrawManager:
+
+    _data = {}
+    _index = 0
+
+    @classmethod
+    def check(cls, session: EventSession, withdraw_time: tuple[int, int]) -> bool:
+        """配置项检查
+
+        参数:
+            session: Session
+            withdraw_time: 配置项数据, (0, 1)
+
+        返回:
+            bool: 是否允许撤回
+        """
+        if withdraw_time[0] and withdraw_time[0] > 0:
+            if withdraw_time[1] == 2:
+                return True
+            if withdraw_time[1] == 1 and (session.id2 or session.id3):
+                return True
+            if withdraw_time[1] == 0 and not (session.id2 or session.id3):
+                return True
+        return False
+
+    @classmethod
+    def append(cls, bot: Bot, message_id: str | int, time: int):
+        """添加消息撤回
+
+        参数:
+            bot: Bot
+            message_id: 消息Id
+            time: 延迟时间
+        """
+        cls._data[cls._index] = (
+            bot,
+            message_id,
+            time,
+        )
+        cls._index += 1
+
+    @classmethod
+    def remove(cls, index: int):
+        """移除
+
+        参数:
+            index: index
+        """
+        if index in cls._data:
+            del cls._data[index]
+
+    @classmethod
+    async def withdraw_message(
+        cls, bot: Bot, message_id: str | int, time: int | None = None
+    ):
+        """消息撤回
+
+        参数:
+            bot: Bot
+            message_id: 消息Id
+            time: 延迟时间
+        """
+        if time:
+            logger.debug(f"将在 {time}秒 内撤回消息ID: {message_id}", "WithdrawManager")
+            await asyncio.sleep(time)
+        if isinstance(bot, v11Bot):
+            logger.debug(f"v11Bot 撤回消息ID: {message_id}", "WithdrawManager")
+            await bot.delete_msg(message_id=int(message_id))
+        elif isinstance(bot, v12Bot):
+            logger.debug(f"v12Bot 撤回消息ID: {message_id}", "WithdrawManager")
+            await bot.delete_message(message_id=str(message_id))
+        elif isinstance(bot, KaiheilaBot):
+            pass
+        elif isinstance(bot, DodoBot):
+            pass
+        elif isinstance(bot, DiscordBot):
+            pass