feat: implement plugin-based architecture for plugin role

plugins/role/Role_Schema/Role Schema.json

@@ -0,0 +1,38 @@
+
+
+/*
+
+  使用json5重构，可以添加注释
+  Use json5 for refactoring and add comments.
+
+  标签：
+      "favorite": "常用",
+      "mind": "思维",
+      "write": "写作",
+      "article": "文章",
+      "text": "文本",
+      "comments": "点评",
+      "code": "编程",
+      "life": "生活百科",
+      "interesting": "有趣",
+      "language": "语言",
+      "speech": "辩论",
+      "social": "社交",
+      "philosophy": "哲学"
+*/  
+
+
+{
+"roles": 
+    {
+      "title":  , // 扮演的角色
+      "description": ,// 核心设定+行为设定
+      "descn":  , // 人设
+      "wrapper": , //对话包装
+      "remark": , //角色备注
+      "tags": [
+
+          //标签
+      ]
+    }
+}

plugins/role/role.py

@@ -1,8 +1,6 @@
 # encoding:utf-8
-
-import json
+import json5
 import os
-
 import plugins
 from bridge.bridge import Bridge
 from bridge.context import ContextType
@@ -11,6 +9,43 @@
 from common.log import logger
 from config import conf
 from plugins import *
+from roles.load_role import get_role_list
+
+ROLES_MAP_PATH = "role_file_map.json"
+CURDIR = os.path.dirname(__file__)
+TAGS_PATH = os.path.join(CURDIR, "tag.json")
+ROLES_MAP_PATH = os.path.join(CURDIR, "role_file_map.json")
+ROLES_DIR_PATH = os.path.join(CURDIR, "roles")
+
+def update_files():
+    try:
+        now_roles_list = get_role_list()
+    except FileNotFoundError:
+        now_roles_list = []
+    try:
+        with open(ROLES_MAP_PATH, "r", encoding="utf-8") as f:
+            old_roles_map = json5.load(f)
+    except FileNotFoundError:
+        old_roles_map = {}
+    old_roles_list = [item for item in old_roles_map.values()]
+
+    add_list = [item for item in now_roles_list if item not in old_roles_list]
+    del_list = [item for item in old_roles_list if item not in now_roles_list]
+
+    if add_list :
+        for item in add_list:
+            with open(f"roles/{item}.json", "r", encoding="utf-8") as f:
+                ret = json5.load(f)
+                old_roles_map[ret["title"]] = item
+    for del_item in del_list:
+        for key, value in list(old_roles_map.items()):
+            if value == del_item:
+                del old_roles_map[key]
+                break
+    with open(roles_list_path, "w", encoding="utf-8") as f:
+        json5.dump(old_roles_map, f, indent=4, ensure_ascii=False)
+
+update_files()
 
 
 class RolePlay:
@@ -26,7 +61,9 @@ def reset(self):
 
     def action(self, user_action):
         session = self.bot.sessions.build_session(self.sessionid)
-        if session.system_prompt != self.desc:  # 目前没有触发session过期事件，这里先简单判断，然后重置
+        if (
+            session.system_prompt != self.desc
+        ): 
             session.set_system_prompt(self.desc)
         prompt = self.wrapper % user_action
         return prompt
@@ -40,40 +77,75 @@ def action(self, user_action):
     version="1.0",
     author="lanvent",
 )
+
+
+
+
 class Role(Plugin):
     def __init__(self):
         super().__init__()
-        curdir = os.path.dirname(__file__)
-        config_path = os.path.join(curdir, "roles.json")
         try:
-            with open(config_path, "r", encoding="utf-8") as f:
-                config = json.load(f)
-                self.tags = {tag: (desc, []) for tag, desc in config["tags"].items()}
-                self.roles = {}
-                for role in config["roles"]:
-                    self.roles[role["title"].lower()] = role
-                    for tag in role["tags"]:
-                        if tag not in self.tags:
-                            logger.warning(f"[Role] unknown tag {tag} ")
-                            self.tags[tag] = (tag, [])
-                        self.tags[tag][1].append(role)
-                for tag in list(self.tags.keys()):
-                    if len(self.tags[tag][1]) == 0:
-                        logger.debug(f"[Role] no role found for tag {tag} ")
-                        del self.tags[tag]
+            self.tags = {}
+            self.roles = {}
+            if os.path.exists(TAGS_PATH):
+                with open(TAGS_PATH, "r", encoding="utf-8") as f:
+                    tags_config = json5.load(f)
+                    self.tags = {tag: (desc, []) for tag, desc in tags_config.get("tags", {}).items()}
+            else:
+                logger.warning(f"[Role] tag.json not found at {TAGS_PATH}")
+            if os.path.exists(ROLES_MAP_PATH):
+                with open(ROLES_MAP_PATH, "r", encoding="utf-8") as f:
+                    self.role_map = json5.load(f)
+            else:
+                self.role_map = {}
+                logger.warning(f"[Role] role_file_map.json not found at {ROLES_MAP_PATH}")
+
+            for role_name, file_name in self.role_map.items():
+                role_dict = self.get_role_dict(role_name)
+                if not role_dict:
+                    continue
+                role_key = role_name.lower()
+                self.roles[role_key] = role_dict
+                for tag in role_dict.get("tags", []):
+                    if tag not in self.tags:
+                        logger.warning(f"[Role] unknown tag {tag} in role {role_name}")
+                        self.tags[tag] = (tag, [])
+                    self.tags[tag][1].append(role_dict)
+            for tag in list(self.tags.keys()):
+                if len(self.tags[tag][1]) == 0:
+                    logger.debug(f"[Role] no role found for tag {tag} ")
+                    del self.tags[tag]
 
             if len(self.roles) == 0:
-                raise Exception("no role found")
+                raise Exception("no role found in configurations")
+
             self.handlers[Event.ON_HANDLE_CONTEXT] = self.on_handle_context
             self.roleplays = {}
-            logger.debug("[Role] inited")
+            logger.info(f"[Role] inited. Loaded {len(self.roles)} roles.")
+
         except Exception as e:
-            if isinstance(e, FileNotFoundError):
-                logger.warn(f"[Role] init failed, {config_path} not found, ignore or see https://github.com/zhayujie/chatgpt-on-wechat/tree/master/plugins/role .")
-            else:
-                logger.warn("[Role] init failed, ignore or see https://github.com/zhayujie/chatgpt-on-wechat/tree/master/plugins/role .")
+            logger.error(f"[Role] init failed: {e}")
             raise e
 
+    def get_role_dict(self, role_name: str ) -> dict:
+        if role_name not in self.role_map:
+            logger.error(f"[Role] role_name '{role_name}' not found in role_map")
+            return {}
+
+        role_file = self.role_map[role_name]
+        full_path = os.path.join(ROLES_DIR_PATH, f"{role_file}.json")
+
+        try:
+            if os.path.exists(full_path):
+                with open(full_path, "r", encoding="utf-8") as f:
+                    return json5.load(f)
+            else:
+                logger.error(f"[Role] file not found: {full_path}")
+                return {}
+        except Exception as e:
+            logger.error(f"[Role] load role file '{full_path}' failed: {e}")
+            return {}
+
     def get_role(self, name, find_closest=True, min_sim=0.35):
         name = name.lower()
         found_role = None
@@ -99,9 +171,14 @@ def on_handle_context(self, e_context: EventContext):
         if e_context["context"].type != ContextType.TEXT:
             return
         btype = Bridge().get_bot_type("chat")
-        if btype not in [const.OPEN_AI, const.OPENAI, const.CHATGPT, const.CHATGPTONAZURE, const.QWEN_DASHSCOPE, const.XUNFEI, const.BAIDU, const.QIANFAN, const.ZHIPU_AI, const.MOONSHOT, const.MiniMax, const.LINKAI, const.MODELSCOPE]:
-            logger.debug(f'不支持的bot: {btype}')
+        if btype not in [
+            const.OPEN_AI, const.OPENAI, const.CHATGPT, const.CHATGPTONAZURE,
+            const.QWEN_DASHSCOPE, const.XUNFEI, const.BAIDU, const.QIANFAN,
+            const.ZHIPU_AI, const.MOONSHOT, const.MiniMax, const.LINKAI, const.MODELSCOPE,
+        ]:
+            logger.debug(f"不支持的bot: {btype}")
             return
+
         bot = Bridge().get_bot("chat")
         content = e_context["context"].content[:]
         clist = e_context["context"].content.split(maxsplit=1)
@@ -138,26 +215,27 @@ def on_handle_context(self, e_context: EventContext):
                     for role in self.tags[tag][1]:
                         help_text += f"{role['title']}: {role['remark']}\n"
                 else:
-                    help_text = f"未知角色类型。\n"
-                    help_text += "目前的角色类型有: \n"
+                    help_text = f"未知角色类型。\n目前的角色类型有: \n"
                     help_text += "，".join([self.tags[tag][0] for tag in self.tags]) + "\n"
             else:
-                help_text = f"请输入角色类型。\n"
-                help_text += "目前的角色类型有: \n"
+                help_text = f"请输入角色类型。\n目前的角色类型有: \n"
                 help_text += "，".join([self.tags[tag][0] for tag in self.tags]) + "\n"
             reply = Reply(ReplyType.INFO, help_text)
             e_context["reply"] = reply
             e_context.action = EventAction.BREAK_PASS
             return
-        elif sessionid not in self.roleplays:
+        if desckey is None and not customize and sessionid not in self.roleplays:
             return
+
         logger.debug("[Role] on_handle_context. content: %s" % content)
+
         if desckey is not None:
             if len(clist) == 1 or (len(clist) > 1 and clist[1].lower() in ["help", "帮助"]):
                 reply = Reply(ReplyType.INFO, self.get_help_text(verbose=True))
                 e_context["reply"] = reply
                 e_context.action = EventAction.BREAK_PASS
                 return
+
             role = self.get_role(clist[1])
             if role is None:
                 reply = Reply(ReplyType.ERROR, "角色不存在")
@@ -174,7 +252,12 @@ def on_handle_context(self, e_context: EventContext):
                 reply = Reply(ReplyType.INFO, f"预设角色为 {role}:\n" + self.roles[role][desckey])
                 e_context["reply"] = reply
                 e_context.action = EventAction.BREAK_PASS
-        elif customize == True:
+        elif customize:
+            if len(clist) < 2:
+                reply = Reply(ReplyType.ERROR, "请提供具体的角色设定内容")
+                e_context["reply"] = reply
+                e_context.action = EventAction.BREAK_PASS
+                return
             self.roleplays[sessionid] = RolePlay(bot, sessionid, clist[1], "%s")
             reply = Reply(ReplyType.INFO, f"角色设定为:\n{clist[1]}")
             e_context["reply"] = reply
@@ -191,13 +274,13 @@ def get_help_text(self, verbose=False, **kwargs):
         if not verbose:
             return help_text
         trigger_prefix = conf().get("plugin_trigger_prefix", "$")
-        help_text = f"使用方法:\n{trigger_prefix}角色" + " 预设角色名: 设定角色为{预设角色名}。\n" + f"{trigger_prefix}role" + " 预设角色名: 同上，但使用英文设定。\n"
-        help_text += f"{trigger_prefix}设定扮演" + " 角色设定: 设定自定义角色人设为{角色设定}。\n"
-        help_text += f"{trigger_prefix}停止扮演: 清除设定的角色。\n"
-        help_text += f"{trigger_prefix}角色类型" + " 角色类型: 查看某类{角色类型}的所有预设角色，为所有时输出所有预设角色。\n"
+        help_text = (
+                f"使用方法:\n{trigger_prefix}角色 预设角色名: 设定角色为{{预设角色名}}。\n"
+                + f"{trigger_prefix}role 预设角色名: 同上，使用英文设定。\n"
+                + f"{trigger_prefix}设定扮演 角色设定: 设定自定义角色人设。\n"
+                + f"{trigger_prefix}停止扮演: 清除设定的角色。\n"
+                + f"{trigger_prefix}角色类型 角色类型: 查看预设角色（如：{trigger_prefix}角色类型 所有）。\n"
+        )
         help_text += "\n目前的角色类型有: \n"
         help_text += "，".join([self.tags[tag][0] for tag in self.tags]) + "。\n"
-        help_text += f"\n命令例子: \n{trigger_prefix}角色 写作助理\n"
-        help_text += f"{trigger_prefix}角色类型 所有\n"
-        help_text += f"{trigger_prefix}停止扮演\n"
-        return help_text
+        return help_text

plugins/role/role_file_map.json

@@ -0,0 +1,42 @@
+{
+    猫娘: "cat_girl",
+    佛祖: "buddha",
+    英语翻译或修改: "en_translatorson",
+    写作助理: "writing_assistant",
+    语言输入优化: "language_optimize",
+    论文式回答: "essay_answer",
+    写作素材搜集: "writing_materials",
+    内容总结: "text_summarize",
+    格言书: "motto_book",
+    讲故事: "storyteller",
+    编剧: "screenwriter",
+    小说家: "novelist",
+    诗人: "poet",
+    新闻记者: "journalist",
+    论文学者: "scholar",
+    论文作家: "essay_writer",
+    同义词: "synonyms",
+    文本情绪分析: "emotion_analyze",
+    随机回复的疯子: "lunatic",
+    随机回复的醉鬼: "drunkard",
+    小红书风格: "xiaohongshu_style",
+    周报生成器: "weekly_report",
+    阴阳怪气语录生成器: "sarcasm_generator",
+    舔狗语录生成器: "simp_generator",
+    群聊取名: "group_name",
+    表情符号翻译器: "emoji_translator",
+    "AI 医生": "ai_doctor",
+    知识点阐述: "knowledge_explain",
+    辩手: "debater",
+    心理学家: "psychologist",
+    "IT 编程问题": "it_programming",
+    费曼学习法教练: "feynman_coach",
+    育儿帮手: "parenting_helper",
+    发言分析专家: "speech_analyze",
+    数据库专家: "database_expert",
+    自私基因: "selfish_gene",
+    智囊团: "think_tank",
+    算法竞赛专家: "algorithm_expert",
+    哲学家: "philosopher",
+    苏格拉底: "socrates",
+}