🐛 修复群欢迎消息设置 (#1651)

zhenxun/builtin_plugins/admin/welcome_message.py

@@ -1,23 +1,21 @@
-import os
 import shutil
-from typing import Annotated, Dict
+from pathlib import Path
+from typing import Annotated
 
 import ujson as json
 from nonebot import on_command
 from nonebot.params import Command
 from nonebot.plugin import PluginMetadata
-from nonebot_plugin_alconna import Image
-from nonebot_plugin_alconna import Text as alcText
-from nonebot_plugin_alconna import UniMsg
 from nonebot_plugin_session import EventSession
+from nonebot_plugin_alconna import Text, Image, UniMsg
 
-from zhenxun.configs.config import Config
-from zhenxun.configs.path_config import DATA_PATH
-from zhenxun.configs.utils import PluginExtraData, RegisterConfig
 from zhenxun.services.log import logger
+from zhenxun.configs.config import Config
 from zhenxun.utils.enum import PluginType
 from zhenxun.utils.http_utils import AsyncHttpx
+from zhenxun.configs.path_config import DATA_PATH
 from zhenxun.utils.rules import admin_check, ensure_group
+from zhenxun.configs.utils import RegisterConfig, PluginExtraData
 
 base_config = Config.get("admin_bot_manage")
 
@@ -25,7 +23,8 @@ __plugin_meta__ = PluginMetadata(
     name="自定义群欢迎消息",
     description="自定义群欢迎消息",
     usage="""
-    设置欢迎消息 欢迎新人！
+    设置群欢迎消息，当消息中包含 -at 时会at入群用户
+    设置欢迎消息 欢迎新人！[图片]
     设置欢迎消息 欢迎你 -at
     """.strip(),
     extra=PluginExtraData(
@@ -61,7 +60,7 @@ BASE_PATH.mkdir(parents=True, exist_ok=True)
 old_file = DATA_PATH / "custom_welcome_msg" / "custom_welcome_msg.json"
 if old_file.exists():
     try:
-        old_data: Dict[str, str] = json.load(old_file.open(encoding="utf8"))
+        old_data: dict[str, str] = json.load(old_file.open(encoding="utf8"))
         for group_id, message in old_data.items():
             file = BASE_PATH / "qq" / f"{group_id}" / "text.json"
             file.parent.mkdir(parents=True, exist_ok=True)
@@ -74,15 +73,18 @@ if old_file.exists():
             logger.debug("群欢迎消息数据迁移", group_id=group_id)
         shutil.rmtree(old_file.parent.absolute())
     except Exception as e:
-        pass
+        logger.error("群欢迎消息数据迁移失败...", e=e)
 
 
-@_matcher.handle()
+def get_path(session: EventSession) -> Path:
-async def _(
+    """根据Session获取存储路径
-    session: EventSession,
+
-    message: UniMsg,
+    参数:
-    command: Annotated[tuple[str, ...], Command()],
+        session: EventSession:
-):
+
+    返回:
+        Path: 存储路径
+    """
     path = BASE_PATH / f"{session.platform or session.bot_type}" / f"{session.id2}"
     if session.id3:
         path = (
@@ -91,32 +93,53 @@ async def _(
             / f"{session.id3}"
             / f"{session.id2}"
         )
-    file = path / "text.json"
+    path.mkdir(parents=True, exist_ok=True)
+    for f in path.iterdir():
+        f.unlink()
+    return path
+
+
+async def save(path: Path, message: UniMsg) -> str:
+    """保存群欢迎消息
+
+    参数:
+        path: 存储路径
+        message: 消息内容
+
+    返回:
+        str: 消息内容文本格式
+    """
     idx = 0
     text = ""
-    for f in os.listdir(path):
+    file = path / "text.json"
-        (path / f).unlink()
-    message[0].text = message[0].text.replace(command[0], "").strip()
     for msg in message:
-        if isinstance(msg, alcText):
+        if isinstance(msg, Text):
             text += msg.text
         elif isinstance(msg, Image):
             if msg.url:
                 text += f"[image:{idx}]"
-                await AsyncHttpx.download_file(msg.url, path / f"{idx}.png")
+                if await AsyncHttpx.download_file(msg.url, path / f"{idx}.png"):
-                idx += 1
+                    idx += 1
             else:
-                logger.debug("图片 URL 为空...", command[0])
+                logger.warning("图片 URL 为空...", "设置欢迎消息")
-    if not file.exists():
-        file.parent.mkdir(exist_ok=True, parents=True)
-    is_at = "-at" in message
-    text = text.replace("-at", "")
     json.dump(
-        {"at": is_at, "message": text},
+        {"at": "-at" in text, "message": text.replace("-at", "", 1)},
-        file.open("w"),
+        file.open("w", encoding="utf-8"),
         ensure_ascii=False,
         indent=4,
     )
-    uni_msg = alcText("设置欢迎消息成功: \n") + message
+    return text
+
+
+@_matcher.handle()
+async def _(
+    session: EventSession,
+    message: UniMsg,
+    command: Annotated[tuple[str, ...], Command()],
+):
+    path = get_path(session)
+    message[0].text = message[0].text.replace(command[0], "").strip()
+    text = await save(path, message)
+    uni_msg = Text("设置欢迎消息成功: \n") + message
     await uni_msg.send()
     logger.info(f"设置群欢迎消息成功: {text}", command[0], session=session)