1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86
| import botpy from botpy import logging, Intents from botpy.message import GroupMessage
class Bot(botpy.Client): img_gen_running = False app_id: int = 0 app_secret: str = ''
def __init__(self, intents: Intents): super().__init__(intents) self.app_id, self.app_secret = load_qqbot_cfg() tc_app_id, tc_app_secret = load_tecentcloud_cfg() tran = lang_translate.LanguageTranslater(tc_app_id, tc_app_secret) if tc_app_id != '' else None self.qq_msg_parser = qq_msg_parser.QQMsgParser(tran) self.ai_chat = ai_chat.AiChat(tc_app_id, tc_app_secret)
async def on_group_at_message_create(self, message: GroupMessage): print(message.content) prompt_type, prompt = self.qq_msg_parser.msg_to_prompt(message.content) msg_req = 0 if prompt_type == qq_msg_parser.PROMPT_TYPE_IMG: await self.resolve_img_gen(message, msg_req, prompt) else: await self.resolve_chat(message, msg_req, prompt)
async def resolve_img_gen(self, message: GroupMessage, msg_req: int, img_prompt: str): if self.img_gen_running: return self.group_reply_txt(message, msg_req, "aris手里的画没画完呐~等画完再发一遍给aris喵~") msg_req += 1 await self.group_reply_txt(message, msg_req, "aris画得比较慢,要等半分钟喵~") msg_req += 1 self.img_gen_running = True gen_code, file_or_msg = await img_gen.generate_img(img_prompt) self.img_gen_running = False _log.info(f'generate_img("{img_prompt}"): {gen_code} {file_or_msg}') if gen_code == 200: jpg_data = await img_gen.get_jpg_from_png(file_or_msg) jpg_data_b64 = base64.standard_b64encode(jpg_data).decode() await self.group_reply_image_data(message, msg_req, jpg_data_b64) else: await self.group_reply_txt(message, msg_req, "aris这张不想画了,换别的吧 T_T")
async def resolve_chat(self, message: GroupMessage, msg_req: int, prompt: str): reply_txt = self.ai_chat.chat(prompt) await self.group_reply_txt(message, msg_req, reply_txt)
async def group_reply_txt(self, message: GroupMessage, msg_req: int, txt: str): await self.api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq=msg_req, content=txt)
async def group_reply_image_data(self, message: GroupMessage, msg_req: int, img_b64: str): upload_media = await self.api.post_group_file( group_openid=message.group_openid, file_type=1, url='', file_data=img_b64, ) await self.api.post_group_message( group_openid=message.group_openid, msg_type=7, msg_id=message.id, msg_seq=msg_req, media=upload_media, )
async def group_reply_image_url(self, message: GroupMessage, msg_req: int, file_url: str): upload_media = await self.api.post_group_file( group_openid=message.group_openid, file_type=1, url=file_url, file_data='', ) await self.api.post_group_message( group_openid=message.group_openid, msg_type=7, msg_id=message.id, msg_seq=msg_req, media=upload_media, )
|