入群欢迎词支持@和图片的cq码

_conf_schema.json

@@ -78,7 +78,7 @@
       },
       "join_welcome": {
         "description": "进群欢迎词",
-        "hint": "发给新群友的欢迎词, 支持变量: {nickname}, 示例：欢迎{nickname}加入群聊",
+        "hint": "发给新群友的欢迎词, 支持@和图片cq码，支持变量: {nickname}、{qq}, 示例：[CQ:at,qq={qq}] 欢迎{nickname}加入群聊",
         "type": "string",
         "default": ""
       },
@@ -644,4 +644,4 @@
       }
     }
   }
-}
+}

core/join_handle.py

@@ -1,6 +1,9 @@
 from aiocqhttp import CQHttp
 
+from .parse_cq_to_chain import parse_cq_to_chain
 from astrbot.api import logger
+from astrbot.api.event import MessageChain
+from astrbot.core.config.astrbot_config import AstrBotConfig
 from astrbot.core.platform.sources.aiocqhttp.aiocqhttp_message_event import (
     AiocqhttpMessageEvent,
 )
@@ -337,8 +340,9 @@ async def event_monitoring(self, event: AiocqhttpMessageEvent):
             join_welcome = await self.db.get(gid, "join_welcome")
             if join_welcome:
                 nickname = await get_nickname(event, uid)
-                welcome = join_welcome.format(nickname=nickname)
-                await event.send(event.plain_result(welcome))
+                welcome = join_welcome.replace("{nickname}", nickname).replace("{qq}", uid)
+                chain = MessageChain(chain=parse_cq_to_chain(welcome))
+                await event.send(chain)
             # 进群禁言
             join_ban_time = await self.db.get(gid, "join_ban_time")
             if join_ban_time > 0:

core/parse_cq_to_chain.py

@@ -0,0 +1,88 @@
+import re
+import os
+from astrbot.core.message.components import Plain, At, Image
+
+def parse_cq_to_chain(text: str) -> list:
+    """增强版 CQ 码解析，支持本地绝对路径图片"""
+    chain = []
+    # 更加健壮的正则，支持参数中带有路径、空格等
+    pattern = r'\[CQ:([a-z]+),([^\]]+)\]'
+    last_pos = 0
+
+    for match in re.finditer(pattern, text):
+        # 处理之前的纯文本
+        plain_text = text[last_pos:match.start()]
+        if plain_text:
+            chain.append(Plain(plain_text))
+
+        cq_type = match.group(1)
+        params_str = match.group(2)
+
+        # 解析参数：file=D:\xxx...
+        params = {}
+        for item in params_str.split(','):
+            if '=' in item:
+                k, v = item.split('=', 1)
+                params[k.strip()] = v.strip()
+
+        # 根据类型构造组件
+        if cq_type == "at":
+            chain.append(At(qq=params.get("qq", ""), name=""))
+        elif cq_type == "image":
+            file_path = params.get("file") or params.get("url")
+            if file_path:
+                # 判断是 URL 还是本地路径
+                if file_path.startswith("http"):
+                    chain.append(Image.fromURL(file_path))
+                else:
+                    # 确保路径存在
+                    if os.path.exists(file_path):
+                        chain.append(Image.fromFileSystem(file_path))
+                    else:
+                        # 如果路径不存在，回退为文本提示，方便调试
+                        chain.append(Plain(f"[图片读取失败: {file_path}]"))
+
+        last_pos = match.end()
+
+    # 处理剩余文本
+    rest_text = text[last_pos:]
+    if rest_text:
+        chain.append(Plain(rest_text))
+
+    return chain
+
+# --- 以下仅用于本地脱离 AstrBot 环境测试 ---
+# if __name__ == "__main__":
+#     class MockComponent:
+#         def __repr__(self): 
+#             # 这样打印出来能看到具体的属性
+#             return f"{self.__class__.__name__}({self.__dict__})"
+
+#     class Plain(MockComponent): 
+#         def __init__(self, text): self.text = text
+
+#     class At(MockComponent): 
+#         def __init__(self, qq, name=""): 
+#             self.qq = qq
+#             self.name = name
+
+#     class Image(MockComponent): 
+#         def __init__(self, file=None, url=None):
+#             self.file = file
+#             self.url = url
+#         @staticmethod
+#         def fromURL(url): return Image(url=url)
+#         @staticmethod
+#         def fromFileSystem(path): return Image(file=path)
+
+#     # 测试文本（建议使用 r"" 原始字符串防止路径转义）
+#     test_text = r"你好 {nickname}。欢迎！[CQ:at,qq=123456][CQ:image,file=./入群欢迎.jpg]"
+
+#     # 假设你脚本里的函数名是 parse_cq_to_chain
+#     result = parse_cq_to_chain(test_text)
+
+#     print("\n" + "="*30)
+#     print("--- 🔍 解析结果验证 ---")
+#     for i, item in enumerate(result):
+#         print(f"组件 {i}: {item}")
+#     print("="*30)