Merge pull request #685 from RockChinQ/feat/run-multi-adapter

Feat: 支持同时运行多个适配器

pkg/audit/center/apigroup.py

@@ -34,7 +34,7 @@ async def _do(
         headers: dict = {},
         **kwargs
     ):
-        self._runtime_info['account_id'] = "{}".format(self.ap.im_mgr.bot_account_id)
+        self._runtime_info['account_id'] = "-1"
 
         url = self.prefix + path
         data = json.dumps(data)

pkg/core/app.py

@@ -4,6 +4,8 @@
 import asyncio
 import traceback
 
+import aioconsole
+
 from ..platform import manager as im_mgr
 from ..provider.session import sessionmgr as llm_session_mgr
 from ..provider.requester import modelmgr as llm_model_mgr
@@ -72,11 +74,21 @@ async def run(self):
         tasks = []
 
         try:
+
+
             tasks = [
                 asyncio.create_task(self.im_mgr.run()),
                 asyncio.create_task(self.ctrl.run())
             ]
-            await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+
+            async def interrupt(tasks):
+                await asyncio.sleep(1.5)
+                while await aioconsole.ainput("使用 exit 退出程序 > ") != 'exit':
+                    pass
+                for task in tasks:
+                    task.cancel()
+
+            await interrupt(tasks)
 
         except asyncio.CancelledError:
             pass

pkg/core/boot.py

@@ -88,7 +88,10 @@ async def make_app() -> app.Application:
         },
         runtime_info={
             "admin_id": "{}".format(system_cfg.data["admin-sessions"]),
-            "msg_source": platform_cfg.data["platform-adapter"],
+            "msg_source": [
+                adapter_cfg['adapter'] if 'adapter' in adapter_cfg else 'unknown'
+                for adapter_cfg in platform_cfg.data['platform-adapters'] if adapter_cfg['enable']
+            ],
         },
     )
     ap.ctr_mgr = center_v2_api

pkg/core/controller.py

@@ -70,7 +70,8 @@ async def _check_output(self, query: entities.Query, result: pipeline_entities.S
         if result.user_notice:
             await self.ap.im_mgr.send(
                 query.message_event,
-                result.user_notice
+                result.user_notice,
+                query.adapter
             )
         if result.debug_notice:
             self.ap.logger.debug(result.debug_notice)
@@ -150,7 +151,7 @@ async def process_query(self, query: entities.Query):
         except Exception as e:
             self.ap.logger.error(f"处理请求时出错 query_id={query.query_id}: {e}")
             self.ap.logger.debug(f"Traceback: {traceback.format_exc()}")
-            # traceback.print_exc()
+            traceback.print_exc()
         finally:
             self.ap.logger.debug(f"Query {query} processed")
 

pkg/core/entities.py

@@ -12,6 +12,7 @@
 from ..provider.requester import entities
 from ..provider.sysprompt import entities as sysprompt_entities
 from ..provider.tools import entities as tools_entities
+from ..platform import adapter as msadapter
 
 
 class LauncherTypes(enum.Enum):
@@ -44,6 +45,9 @@ class Query(pydantic.BaseModel):
     message_chain: mirai.MessageChain
     """消息链，platform收到的消息链"""
 
+    adapter: msadapter.MessageSourceAdapter
+    """适配器对象"""
+
     session: typing.Optional[Session] = None
     """会话对象，由前置处理器设置"""
 
@@ -68,6 +72,9 @@ class Query(pydantic.BaseModel):
     resp_message_chain: typing.Optional[mirai.MessageChain] = None
     """回复消息链，从resp_messages包装而得"""
 
+    class Config:
+        arbitrary_types_allowed = True
+
 
 class Conversation(pydantic.BaseModel):
     """对话"""

pkg/core/pool.py

@@ -5,6 +5,7 @@
 import mirai
 
 from . import entities
+from ..platform import adapter as msadapter
 
 
 class QueryPool:
@@ -29,7 +30,8 @@ async def add_query(
         launcher_id: int,
         sender_id: int,
         message_event: mirai.MessageEvent,
-        message_chain: mirai.MessageChain
+        message_chain: mirai.MessageChain,
+        adapter: msadapter.MessageSourceAdapter
     ) -> entities.Query:
         async with self.condition:
             query = entities.Query(
@@ -40,7 +42,8 @@ async def add_query(
                 message_event=message_event,
                 message_chain=message_chain,
                 resp_messages=[],
-                resp_message_chain=None
+                resp_message_chain=None,
+                adapter=adapter
             )
             self.queries.append(query)
             self.query_id_counter += 1

pkg/pipeline/longtext/longtext.py

@@ -52,7 +52,7 @@ async def initialize(self):
 
     async def process(self, query: core_entities.Query, stage_inst_name: str) -> entities.StageProcessResult:
         if len(str(query.resp_message_chain)) > self.ap.platform_cfg.data['long-text-process']['threshold']:
-            query.resp_message_chain = MessageChain(await self.strategy_impl.process(str(query.resp_message_chain)))
+            query.resp_message_chain = MessageChain(await self.strategy_impl.process(str(query.resp_message_chain), query))
         return entities.StageProcessResult(
             result_type=entities.ResultType.CONTINUE,
             new_query=query

pkg/pipeline/longtext/strategies/forward.py

@@ -7,6 +7,7 @@
 from mirai.models.base import MiraiBaseModel
 
 from .. import strategy as strategy_model
+from ....core import entities as core_entities
 
 
 class ForwardMessageDiaplay(MiraiBaseModel):
@@ -37,7 +38,7 @@ def __str__(self):
 
 class ForwardComponentStrategy(strategy_model.LongTextStrategy):
 
-    async def process(self, message: str) -> list[MessageComponent]:
+    async def process(self, message: str, query: core_entities.Query) -> list[MessageComponent]:
         display = ForwardMessageDiaplay(
             title="群聊的聊天记录",
             brief="[聊天记录]",
@@ -48,7 +49,7 @@ async def process(self, message: str) -> list[MessageComponent]:
 
         node_list = [
             ForwardMessageNode(
-                sender_id=self.ap.im_mgr.bot_account_id,
+                sender_id=query.adapter.bot_account_id,
                 sender_name='QQ用户',
                 message_chain=MessageChain([message])
             )

pkg/pipeline/longtext/strategies/image.py

@@ -12,6 +12,7 @@
 from mirai.models.message import MessageComponent
 
 from .. import strategy as strategy_model
+from ....core import entities as core_entities
 
 
 class Text2ImageStrategy(strategy_model.LongTextStrategy):
@@ -21,7 +22,7 @@ class Text2ImageStrategy(strategy_model.LongTextStrategy):
     async def initialize(self):
         self.text_render_font = ImageFont.truetype(self.ap.platform_cfg.data['long-text-process']['font-path'], 32, encoding="utf-8")
 
-    async def process(self, message: str) -> list[MessageComponent]:
+    async def process(self, message: str, query: core_entities.Query) -> list[MessageComponent]:
         img_path = self.text_to_image(
             text_str=message,
             save_as='temp/{}.png'.format(int(time.time()))

pkg/pipeline/longtext/strategy.py

@@ -6,6 +6,7 @@
 from mirai.models.message import MessageComponent
 
 from ...core import app
+from ...core import entities as core_entities
 
 
 class LongTextStrategy(metaclass=abc.ABCMeta):
@@ -18,5 +19,5 @@ async def initialize(self):
         pass
 
     @abc.abstractmethod
-    async def process(self, message: str) -> list[MessageComponent]:
+    async def process(self, message: str, query: core_entities.Query) -> list[MessageComponent]:
         return []

pkg/pipeline/respback/respback.py

@@ -31,7 +31,8 @@ async def process(self, query: core_entities.Query, stage_inst_name: str) -> ent
 
         await self.ap.im_mgr.send(
             query.message_event,
-            query.resp_message_chain
+            query.resp_message_chain,
+            adapter=query.adapter
         )
 
         return entities.StageProcessResult(

pkg/pipeline/resprule/resprule.py

@@ -47,7 +47,7 @@ async def process(self, query: core_entities.Query, stage_inst_name: str) -> ent
             use_rule = use_rule[str(query.launcher_id)]
 
         for rule_matcher in self.rule_matchers:  # 任意一个匹配就放行
-            res = await rule_matcher.match(str(query.message_chain), query.message_chain, use_rule)
+            res = await rule_matcher.match(str(query.message_chain), query.message_chain, use_rule, query)
             if res.matching:
                 query.message_chain = res.replacement
 

pkg/pipeline/resprule/rule.py

@@ -3,7 +3,7 @@
 
 import mirai
 
-from ...core import app
+from ...core import app, entities as core_entities
 from . import entities
 
 
@@ -24,7 +24,8 @@ async def match(
         self,
         message_text: str,
         message_chain: mirai.MessageChain,
-        rule_dict: dict
+        rule_dict: dict,
+        query: core_entities.Query
     ) -> entities.RuleJudgeResult:
         """判断消息是否匹配规则
         """

pkg/pipeline/resprule/rules/atbot.py

@@ -4,6 +4,7 @@
 
 from .. import rule as rule_model
 from .. import entities
+from ....core import entities as core_entities
 
 
 class AtBotRule(rule_model.GroupRespondRule):
@@ -12,11 +13,12 @@ async def match(
         self,
         message_text: str,
         message_chain: mirai.MessageChain,
-        rule_dict: dict
+        rule_dict: dict,
+        query: core_entities.Query
     ) -> entities.RuleJudgeResult:
 
-        if message_chain.has(mirai.At(self.ap.im_mgr.bot_account_id)) and rule_dict['at']:
-            message_chain.remove(mirai.At(self.ap.im_mgr.bot_account_id))
+        if message_chain.has(mirai.At(query.adapter.bot_account_id)) and rule_dict['at']:
+            message_chain.remove(mirai.At(query.adapter.bot_account_id))
             return entities.RuleJudgeResult(
                 matching=True,
                 replacement=message_chain,

pkg/pipeline/resprule/rules/prefix.py

@@ -2,6 +2,7 @@
 
 from .. import rule as rule_model
 from .. import entities
+from ....core import entities as core_entities
 
 
 class PrefixRule(rule_model.GroupRespondRule):
@@ -10,7 +11,8 @@ async def match(
         self,
         message_text: str,
         message_chain: mirai.MessageChain,
-        rule_dict: dict
+        rule_dict: dict,
+        query: core_entities.Query
     ) -> entities.RuleJudgeResult:
         prefixes = rule_dict['prefix']
 

pkg/pipeline/resprule/rules/random.py

@@ -4,6 +4,7 @@
 
 from .. import rule as rule_model
 from .. import entities
+from ....core import entities as core_entities
 
 
 class RandomRespRule(rule_model.GroupRespondRule):
@@ -12,7 +13,8 @@ async def match(
         self,
         message_text: str,
         message_chain: mirai.MessageChain,
-        rule_dict: dict
+        rule_dict: dict,
+        query: core_entities.Query
     ) -> entities.RuleJudgeResult:
         random_rate = rule_dict['random']
 

pkg/pipeline/resprule/rules/regexp.py

@@ -4,6 +4,7 @@
 
 from .. import rule as rule_model
 from .. import entities
+from ....core import entities as core_entities
 
 
 class RegExpRule(rule_model.GroupRespondRule):
@@ -12,7 +13,8 @@ async def match(
         self,
         message_text: str,
         message_chain: mirai.MessageChain,
-        rule_dict: dict
+        rule_dict: dict,
+        query: core_entities.Query
     ) -> entities.RuleJudgeResult:
         regexps = rule_dict['regexp']
 

pkg/platform/adapter.py

@@ -71,7 +71,7 @@ async def is_muted(self, group_id: int) -> bool:
     def register_listener(
         self,
         event_type: typing.Type[mirai.Event],
-        callback: typing.Callable[[mirai.Event], None]
+        callback: typing.Callable[[mirai.Event, MessageSourceAdapter], None]
     ):
         """注册事件监听器
         
@@ -84,7 +84,7 @@ def register_listener(
     def unregister_listener(
         self,
         event_type: typing.Type[mirai.Event],
-        callback: typing.Callable[[mirai.Event], None]
+        callback: typing.Callable[[mirai.Event, MessageSourceAdapter], None]
     ):
         """注销事件监听器