feat: 添加了网页监控面板

requirements.txt

@@ -1,4 +1,5 @@
-arclet-alconna==2.0.0a1
+arclet-alconna==1.8.5
+arclet-alconna-tools==0.7.0
 dash==2.16.1
 nonebot2[fastapi]==2.2.1
 nonebot-adapter-onebot==2.4.3

src/plugins/liteyuki_markdowntest.py

@@ -27,4 +27,4 @@ async def _(bot: T_Bot, event: T_MessageEvent, arg: v11.Message = CommandArg()):
         bot,
         message_type=event.message_type,
         session_id=event.user_id if event.message_type == "private" else event.group_id
-    )
+)

src/plugins/liteyuki_plugin_eventpush.py

@@ -0,0 +1,99 @@
+from typing import Optional
+
+import nonebot
+from nonebot import on_message
+from arclet.alconna import Arparma, Alconna, Args, Option, Subcommand, Arg
+from nonebot_plugin_alconna import on_alconna
+from src.utils.data import LiteModel
+from src.utils.message import send_markdown
+from src.utils.typing import T_Bot, T_MessageEvent
+
+pushes = []
+
+
+class Node(LiteModel):
+    bot_id: str
+    session_type: str
+    session_id: str
+
+    def __str__(self):
+        return f"{self.bot_id}.{self.session_type}.{self.session_id}"
+
+
+class Push(LiteModel):
+    source: Node
+    target: Node
+
+
+alc = Alconna(
+    "lep",
+    Subcommand(
+        "add",
+        Args["source", str],
+        Args["target", str],
+        Option("-b", Args["bidirectional", bool])
+    ),
+    Subcommand(
+        "rm",
+        Args["index", int],
+
+    ),
+    Subcommand(
+        "list",
+    )
+)
+
+add_push = on_alconna(alc)
+
+
+@add_push.handle()
+async def _(result: Arparma):
+    """bot_id.session_type.session_id"""
+    if result.subcommands.get("add"):
+        source = result.subcommands["add"].args.get("source")
+        target = result.subcommands["add"].args.get("target")
+        if source and target:
+            source = source.split(".")
+            target = target.split(".")
+            pushes.append(Push(
+                source=Node(bot_id=source[0], session_type=source[1], session_id=source[2]),
+                target=Node(bot_id=target[0], session_type=target[1], session_id=target[2])
+            ))
+
+            if result.subcommands["add"].args.get("-b"):
+                pushes.append(Push(
+                    source=Node(bot_id=target[0], session_type=target[1], session_id=target[2]),
+                    target=Node(bot_id=source[0], session_type=source[1], session_id=source[2])
+                ))
+
+            await add_push.finish("添加成功")
+        else:
+            await add_push.finish("参数缺失")
+    elif result.subcommands.get("rm"):
+        index = result.subcommands["rm"].args.get("index")
+        if index is not None:
+            try:
+                pushes.pop(index)
+                await add_push.finish("删除成功")
+            except IndexError:
+                await add_push.finish("索引错误")
+        else:
+            await add_push.finish("参数缺失")
+    elif result.subcommands.get("list"):
+        await add_push.finish("\n".join([f"{i} {push.source.bot_id}.{push.source.session_type}.{push.source.session_id} -> "
+                                       f"{push.target.bot_id}.{push.target.session_type}.{push.target.session_id}" for i, push in enumerate(pushes)]))
+    else:
+        await add_push.finish("参数错误")
+
+
+@on_message(block=False).handle()
+async def _(event: T_MessageEvent, bot: T_Bot):
+    for push in pushes:
+        if str(push.source) == f"{bot.self_id}.{event.message_type}.{event.user_id if event.message_type == 'private' else event.group_id}":
+            bot2 = nonebot.get_bot(push.target.bot_id)
+            msg_formatted = ""
+            for l in str(event.message).split("\n"):
+                msg_formatted += f"**{l.strip()}**\n"
+            push_message = f"{msg_formatted}\n\n> From {event.sender.nickname}@{push.source.session_type}.{push.source.session_id}\n> Bot {bot.self_id}"
+            await send_markdown(push_message, bot2, push.target.session_type, push.target.session_id)
+    return

src/utils/message.py

@@ -6,8 +6,9 @@ from .tools import de_escape
 from .typing import T_Bot
 
 
-async def send_markdown(markdown: str, bot: T_Bot, message_type: str, session_id: str) -> dict[str, Any]:
+async def send_markdown(markdown: str, bot: T_Bot, message_type: str, session_id: str | int) -> dict[str, Any]:
     formatted_md = de_escape(markdown).replace("\n", r"\n").replace("\"", r'\\\"')
+    try:
         forward_data = await bot.call_api(
             api="send_private_forward_msg",
             user_id=bot.self_id,
@@ -15,8 +16,8 @@ async def send_markdown(markdown: str, bot: T_Bot, message_type: str, session_id
                     v11.MessageSegment(
                         type="node",
                         data={
-                    "name": "Liteyuki.OneBot",
+                                "name"   : "Liteyuki.OneBot",
-                    "uin": bot.self_id,
+                                "uin"    : bot.self_id,
                                 "content": [
                                         {
                                                 "type": "markdown",
@@ -27,11 +28,11 @@ async def send_markdown(markdown: str, bot: T_Bot, message_type: str, session_id
                                 ]
                         },
                     )
-
             ]
         )
-    try:
         data = await bot.send_msg(
+            user_id=session_id,
+            group_id=session_id,
             message_type=message_type,
             message=[
                     v11.MessageSegment(
@@ -42,16 +43,28 @@ async def send_markdown(markdown: str, bot: T_Bot, message_type: str, session_id
                     ),
             ],
 
-
         )
     except Exception as e:
-        nonebot.logger.warning("send_markdown error, send as plane text: %s", e)
+        nonebot.logger.warning("send_markdown error, send as plane text: %s" % e)
+        if isinstance(bot, v11.Bot):
             data = await bot.send_msg(
                 message_type=message_type,
                 message=markdown,
-            user_id=session_id if message_type == "private" else None,
+                user_id=int(session_id),
-            group_id=session_id if message_type == "group" else None
+                group_id=int(session_id)
-
-
             )
+        elif isinstance(bot, v12.Bot):
+            data = await bot.send_message(
+                detail_type=message_type,
+                message=v12.Message(
+                    v12.MessageSegment.text(
+                        text=markdown
+                    )
+                ),
+                user_id=str(session_id),
+                group_id=str(session_id)
+            )
+        else:
+            nonebot.logger.error("send_markdown: bot type not supported")
+            data = {}
     return data

src/utils/typing.py

@@ -1,6 +1,5 @@
 from nonebot.adapters.onebot import v11, v12
 
-
 T_Bot = v11.Bot | v12.Bot
 T_GroupMessageEvent = v11.GroupMessageEvent | v12.GroupMessageEvent
 T_PrivateMessageEvent = v11.PrivateMessageEvent | v12.PrivateMessageEvent