from __future__ import annotations import typing import asyncio import traceback import datetime from pkg.platform.adapter import MessagePlatformAdapter from pkg.platform.types import events as platform_events, message as platform_message from pkg.core import app from .. import adapter from ...core import app from ..types import message as platform_message from ..types import events as platform_events from ..types import entities as platform_entities from ...command.errors import ParamNotEnoughError from libs.qq_official_api.api import QQOfficialClient from libs.qq_official_api.qqofficialevent import QQOfficialEvent from ...utils import image class QQOfficialMessageConverter(adapter.MessageConverter): @staticmethod async def yiri2target(message_chain: platform_message.MessageChain): content_list = [] #只实现了发文字 for msg in message_chain: if type(msg) is platform_message.Plain: content_list.append({ "type":"text", "content":msg.text, }) return content_list @staticmethod async def target2yiri(message:str,message_id:str,pic_url:str,content_type): yiri_msg_list = [] yiri_msg_list.append( platform_message.Source(id=message_id,time=datetime.datetime.now()) ) if pic_url is not None: base64_url = await image.get_qq_official_image_base64(pic_url=pic_url,content_type=content_type) yiri_msg_list.append( platform_message.Image(base64=base64_url) ) yiri_msg_list.append(platform_message.Plain(text=message)) chain = platform_message.MessageChain(yiri_msg_list) return chain class QQOfficialEventConverter(adapter.EventConverter): @staticmethod async def yiri2target(event:platform_events.MessageEvent) -> QQOfficialEvent: return event.source_platform_object @staticmethod async def target2yiri(event:QQOfficialEvent): """ QQ官方消息转换为LB对象 """ yiri_chain = await QQOfficialMessageConverter.target2yiri( message=event.content,message_id=event.d_id,pic_url=event.attachments,content_type=event.content_type ) if event.t == 'C2C_MESSAGE_CREATE': friend = platform_entities.Friend( id = event.user_openid, nickname = event.t, remark = "", ) return platform_events.FriendMessage( sender = friend,message_chain = yiri_chain,time = int( datetime.datetime.strptime( event.timestamp, "%Y-%m-%dT%H:%M:%S%z" ).timestamp() ), source_platform_object=event ) if event.t == 'DIRECT_MESSAGE_CREATE': friend = platform_entities.Friend( id = event.guild_id, nickname = event.t, remark = "", ) return platform_events.FriendMessage( sender = friend,message_chain = yiri_chain, source_platform_object=event ) if event.t == 'GROUP_AT_MESSAGE_CREATE': yiri_chain.insert(0, platform_message.At(target="justbot")) sender = platform_entities.GroupMember( id = event.group_openid, member_name= event.t, permission= 'MEMBER', group = platform_entities.Group( id = event.group_openid, name = 'MEMBER', permission= platform_entities.Permission.Member ), special_title='', join_timestamp=0, last_speak_timestamp=0, mute_time_remaining=0 ) time = int( datetime.datetime.strptime( event.timestamp, "%Y-%m-%dT%H:%M:%S%z" ).timestamp() ) return platform_events.GroupMessage( sender = sender, message_chain=yiri_chain, time = time, source_platform_object=event ) if event.t =='AT_MESSAGE_CREATE': yiri_chain.insert(0, platform_message.At(target="justbot")) sender = platform_entities.GroupMember( id = event.channel_id, member_name=event.t, permission= 'MEMBER', group = platform_entities.Group( id = event.channel_id, name = 'MEMBER', permission=platform_entities.Permission.Member ), special_title='', join_timestamp=0, last_speak_timestamp=0, mute_time_remaining=0 ) time = int( datetime.datetime.strptime( event.timestamp, "%Y-%m-%dT%H:%M:%S%z" ).timestamp() ) return platform_events.GroupMessage( sender =sender, message_chain = yiri_chain, time = time, source_platform_object=event ) class QQOfficialAdapter(adapter.MessagePlatformAdapter): bot:QQOfficialClient ap:app.Application config:dict bot_account_id:str message_converter: QQOfficialMessageConverter = QQOfficialMessageConverter() event_converter: QQOfficialEventConverter = QQOfficialEventConverter() def __init__(self, config:dict, ap:app.Application): self.config = config self.ap = ap required_keys = [ "appid", "secret", ] missing_keys = [key for key in required_keys if key not in config] if missing_keys: raise ParamNotEnoughError("QQ官方机器人缺少相关配置项，请查看文档或联系管理员") self.bot = QQOfficialClient( app_id=config["appid"], secret=config["secret"], token=config["token"], ) async def reply_message( self, message_source: platform_events.MessageEvent, message: platform_message.MessageChain, quote_origin: bool = False, ): qq_official_event = await QQOfficialEventConverter.yiri2target( message_source, ) content_list = await QQOfficialMessageConverter.yiri2target(message) #私聊消息 if qq_official_event.t == 'C2C_MESSAGE_CREATE': for content in content_list: if content["type"] == 'text': await self.bot.send_private_text_msg(qq_official_event.user_openid,content['content'],qq_official_event.d_id) #群聊消息 if qq_official_event.t == 'GROUP_AT_MESSAGE_CREATE': for content in content_list: if content["type"] == 'text': await self.bot.send_group_text_msg(qq_official_event.group_openid,content['content'],qq_official_event.d_id) #频道群聊 if qq_official_event.t == 'AT_MESSAGE_CREATE': for content in content_list: if content["type"] == 'text': await self.bot.send_channle_group_text_msg(qq_official_event.channel_id,content['content'],qq_official_event.d_id) #频道私聊 if qq_official_event.t == 'DIRECT_MESSAGE_CREATE': for content in content_list: if content["type"] == 'text': await self.bot.send_channle_private_text_msg(qq_official_event.guild_id,content['content'],qq_official_event.d_id) async def send_message( self, target_type: str, target_id: str, message: platform_message.MessageChain ): pass def register_listener( self, event_type: typing.Type[platform_events.Event], callback: typing.Callable[ [platform_events.Event, adapter.MessagePlatformAdapter], None ], ): async def on_message(event:QQOfficialEvent): self.bot_account_id = "justbot" try: return await callback( await self.event_converter.target2yiri(event),self ) except: traceback.print_exc() if event_type == platform_events.FriendMessage: self.bot.on_message("DIRECT_MESSAGE_CREATE")(on_message) self.bot.on_message("C2C_MESSAGE_CREATE")(on_message) elif event_type == platform_events.GroupMessage: self.bot.on_message("GROUP_AT_MESSAGE_CREATE")(on_message) self.bot.on_message("AT_MESSAGE_CREATE")(on_message) async def run_async(self): async def shutdown_trigger_placeholder(): while True: await asyncio.sleep(1) await self.bot.run_task( host='0.0.0.0', port=self.config["port"], shutdown_trigger=shutdown_trigger_placeholder, ) async def kill(self) -> bool: return False def unregister_listener( self, event_type: type, callback: typing.Callable[[platform_events.Event, MessagePlatformAdapter], None], ): return super().unregister_listener(event_type, callback)