支持更多格式优化

nicecoze.py

@@ -1,11 +1,8 @@
 # encoding:utf-8
 
-import json
+import re
-import os
-from urllib.parse import urlparse
 
 import plugins
-from bridge.context import ContextType
 from bridge.reply import Reply, ReplyType
 from common.log import logger
 from plugins import *
@@ -13,10 +10,10 @@ from plugins import *
 
 @plugins.register(
     name="Nicecoze",
-    desire_priority=100,
+    desire_priority=66,
-    hidden=True,
+    hidden=False,
     desc="一款优化coze-discord-proxy返回结果的插件。",
-    version="1.0",
+    version="1.1",
     author="空心菜",
 )
 class Nicecoze(Plugin):
@@ -32,37 +29,60 @@ class Nicecoze(Plugin):
     def on_decorate_reply(self, e_context: EventContext):
         if e_context["reply"].type != ReplyType.TEXT:
             return
-        reply = e_context["reply"]
         try:
-            content_list = reply.content.strip().split('\n')
+            channel = e_context["channel"]
+            context = e_context["context"]
+            content = e_context["reply"].content.strip()
+            # 避免图片无法下载时，重复调用插件导致没有响应的问题
+            if content.startswith("[NICECOZE提醒您图片下载失败了，点击网址将就看吧]"):
+                return
             # 提取CDP返回的Markdown图片链接中的网址，并修改ReplyType为IMAGE_URL，以便CoW自动下载Markdown链接中的图片
-            if len(content_list)==2 and self.is_url(content_list[0]):
+            if all(x in content for x in ['![', 'http']) and any(x in content for x in ['jpg', 'jpeg', 'png', 'gif', 'bmp', 'webp']):
-                if content_list[1] == f"![Image]({content_list[0]})":
+                logger.debug(f"[Nicecoze] starting decorate_markdown_image, content={content}")
-                    reply = Reply(ReplyType.IMAGE_URL, content_list[0])
+                replies = self.decorate_markdown_image(content)
-                    e_context["reply"] = reply
+                if replies:
-                    e_context.action = EventAction.CONTINUE
+                    logger.info(f"[Nicecoze] sending {len(replies)} images ...")
-                    logger.info(f"[Nicecoze] Change ReplyType from TEXT to IMAGE_URL: {content_list[0]}")
+                    e_context["reply"].content = "[NICECOZE提醒您图片下载失败了，点击网址将就看吧]\n" + e_context["reply"].content
-                else:
+                    for reply in replies:
-                    logger.debug(f"[Nicecoze] URL in content but not a markdown image url.")
+                        channel.send(reply, context)
+                    e_context["reply"] = Reply(ReplyType.TEXT, f"{len(replies)}张图片已发送，收到了吗？")
+                    e_context.action = EventAction.BREAK_PASS
+                    return
             # 去掉每行结尾的Markdown链接中网址部分的小括号，避免微信误以为“)”是网址的一部分导致微信中无法打开该页面
-            else:
+            content_list = content.split('\n')
-                new_content_list = [re.sub(r'\((https?://[^\s]+)\)$', r' \1', line) for line in content_list]
+            new_content_list = [re.sub(r'\((https?://[^\s]+)\)$', r' \1', line) for line in content_list]
-                reply = Reply(ReplyType.TEXT, '\n'.join(new_content_list))
+            if new_content_list != content_list:
+                logger.info(f"[Nicecoze] parenthesis in the url has been removed, content={content}")
+                reply = Reply(ReplyType.TEXT, '\n'.join(new_content_list).strip())
                 e_context["reply"] = reply
-                e_context.action = EventAction.CONTINUE
+        except Exception as e:
-        except Exception:
+            logger.warn(f"[Nicecoze] on_decorate_reply failed, content={content}, error={e}")
-            pass
+        finally:
+            e_context.action = EventAction.CONTINUE
 
-    def is_url(self, string):
+    def decorate_markdown_image(self, content):
-        """
+        # 完全匹配一张Markdown图片（格式：`![name](url)`）
-        判断字符串是否为URL的urllib库方法
+        markdown_image1 = r"\!\[(?P<image_name>.*)\]\((?P<image_url>https?\:\/\/[a-zA-Z0-9\-\.]+\.[a-zA-Z]{2,}(:[0-9]{1,5})?(\/[\S]*)\.(jpg|jpeg|png|gif|bmp|webp)(\?[\S]*)?)\)"
-        """
+        match_obj1 = re.fullmatch(markdown_image1, content, re.DOTALL)
-        try:
+        if match_obj1 and match_obj1.group('image_url'):
-            result = urlparse(string)
+            image_name, image_url = match_obj1.group('image_name'), match_obj1.group('image_url')
-            return all([result.scheme, result.netloc])
+            logger.info(f"[Nicecoze] markdown_image1 found, image_name={image_name}, image_url={image_url}")
-        except ValueError:
+            reply = Reply(ReplyType.IMAGE_URL, image_url)
-            return False
+            return [reply]
+        # 匹配多张Markdown图片(格式：`url\n![Image](url)`)
+        markdown_image2 = r"(?P<image_url>https?\:\/\/[a-zA-Z0-9\-\.]+\.[a-zA-Z]{2,}(:[0-9]{1,5})?(\/[\S]*)\.(jpg|jpeg|png|gif|bmp|webp)(\?[\S]*)?)\n*\!\[Image\]\((?P=image_url)\)"
+        match_iter2 = re.finditer(markdown_image2, content)
+        replies = []
+        for match in match_iter2:
+            image_url = match.group('image_url')
+            logger.info(f"[Nicecoze] markdown_image2 found, image_url={image_url}")
+            reply = Reply(ReplyType.IMAGE_URL, image_url)
+            replies.append(reply)
+        if replies:
+            return replies
+        if content.startswith('![') and 'http' in content and any(img in content for img in ['jpg', 'jpeg', 'png', 'gif', 'bmp', 'webp']):
+            logger.info(f"[Nicecoze] it seems markdown image in the content but not matched, content={content}.")
 
     def get_help_text(self, **kwargs):
-        return "一款优化coze-discord-proxy返回结果的插件。"
+        return "优化coze-discord-proxy返回结果的插件。"