from typing import Any, List import group from botpy.connection import ConnectionSession from botpy.message import (Message,GroupMessage) from botpy.flags import (Flag,fill_with_flags) from botpy import( logger,Intents,Client) from botpy.gateway import (BotWebSocket) from botpy.types.gateway import (MessagePayload) import queue from botpy.types.session import Session from botpy.types.message import MarkdownPayload from dotenv import load_dotenv import os import types from bobby import bobbySession from mcwatcher import watcher from threading import Thread import asyncio from serverGET import ServerGet #环境配置 class Intents_ex(Intents): def __init__(self, **kwargs: bool) -> None: arg_tmp = dict[str,bool]() arg_ex = dict[str,bool]() for key, value in kwargs.items(): if key in ['member_event']: arg_ex[key] = value else: arg_tmp[key] = value super().__init__(**arg_tmp) for key , value in arg_ex.items() : setattr(self, key, value) @Flag def member_event(self): return 1 << 24 class BotWebsocket_ex(BotWebSocket): def __init__(self, session: Session, _connection: ConnectionSession): super().__init__(session, _connection) self._parser['group_member_add'] = lambda aaw:self._connection.dispatch("group_member_add",aaw) self._parser['group_message_create'] = lambda aaw:self._connection.dispatch("group_message_create",aaw) class djtuWebsocket (Client) : def __init__( self, intents: Intents, timeout: int = 5, is_sandbox=False, log_config: str | dict | None = None, log_format: str| None = None, log_level: int |None= None, bot_log: bool | None = True, ext_handlers: dict | List[dict] | bool = True): self.q = queue.Queue() super().__init__(intents, timeout, is_sandbox, log_config, log_format, log_level, bot_log, ext_handlers) async def bot_connect(self, session): """ newConnect 启动一个新的连接，如果连接在监听过程中报错了，或者被远端关闭了链接，需要识别关闭的原因，能否继续 resume 如果能够 resume，则往 sessionChan 中放入带有 sessionID 的 session 如果不能，则清理掉 sessionID，将 session 放入 sessionChan 中 session 的启动，交给 start 中的 for 循环执行，session 不自己递归进行重连，避免递归深度过深 param session: session对象 """ logger.info("[botpy] 会话启动中...") client = BotWebsocket_ex(session, self._connection) try: await client.ws_connect() except (Exception, KeyboardInterrupt, SystemExit) as e: await client.on_error(e) async def on_group_at_message_create(self, message: Message): await self.parseCommand(message) async def on_group_message_create(self,message): # 检验是否有自身@消息 try: msg = message['d'] mention = msg['mentions'] for m in mention : if m['is_you'] == True: return await self.parseCommand(GroupMessage(self.api,msg['id'],MessagePayload(message['d']))) except: '没有@消息欸' return await self.parseGlobalMessage(message) async def on_group_member_add(self,payload): # await message.reply(content="awdwa") openid = 'null' memberid = 'null' eid ='null' try : eid = payload['id'] data = payload['d'] openid = data['group_openid'] memberid = data['member_openid'] except: logger.warning("data解析失败") return logger.info("入群信息:"+' 群: '+openid+" openid: "+memberid) await self.api.post_group_message(openid,event_id=eid,content='欢迎加入大海啸！\n入服教程请看链接:\nhttps://hiso.wgfire.top/archives/ifnew') # 触发概率不大所以可以直接提取关键字 async def parseCommand(self,msg :GroupMessage): # 加入群聊 if msg.group_openid: self.group_add(msg.group_openid) commands:dict[str,types.MethodType|types.FunctionType] = { r'/邀请码' : bobby.getInviteCode, r'吃什么' : lambda ctx:"是啊吃什么" , r'/服务器' : ServerGet.GetNames, } required = msg.content reply = "" for cmd,func in commands.items(): if required.find(cmd) != -1: try: ctx = required.split(cmd)[1] reply = func(ctx) except: reply = "服务异常" break if reply.__len__() == 0: reply = "你在发什么东西我怎么看不懂?" md = MarkdownPayload(content=reply) await msg.reply(msg_type=2, markdown=md) def group_add(self,gid :str): # 加入群聊 new_group = True if not groups is None : for i in groups.groups : if i.find(gid) != -1 : new_group = False if new_group : groups.groups.append(gid) groups.save() # 触发概率大，以后要做好解析防止刷屏 async def parseGlobalMessage(self,payload): try : msg = payload['d'] gid = msg['group_id'] content = msg['content'] except: return # 群聊加入 self.group_add(gid) # 解析指令 commands:dict[str,types.MethodType|types.FunctionType] = { r'吃什么' : lambda :"是啊吃什么" } required = content reply = "" for cmd,func in commands.items(): if required.find(cmd) != -1: try: reply = func() except: reply = "服务异常" break if reply.__len__() == 0: return await self.api.post_group_message(gid,content=reply) # await msg.reply(content="awda") # 监听队列，接受其他线程的消息 async def mcloop(self): while True: if self.q.empty(): # time.sleep会阻塞整个线程我去 await asyncio.sleep(20) else: # 广播 try : msg = self.q.get() for i in groups.groups: asyncio.create_task(self.api.post_group_message(i,content=msg)) except Exception as e: logger.error("MCWatcher发送群聊任务失败:\n"+str(e)) # source load_dotenv(".env") bbyAccount = os.getenv("BOBBY_ACCOUNT") bbyPasswd = os.getenv("BOBBY_PASSWORD") appid = os.getenv("APP_ID") secret_bot = os.getenv("SECRET") groups = group.GroupCtl() if appid is None or secret_bot is None : raise SystemExit("未找到bot配置!") logger.info("登录账号:"+appid) # bobby Session if bbyAccount is None or bbyPasswd is None : raise SystemError("未找到账密配置") if appid is None or secret_bot is None : raise SystemError("未找到bot配置") bobby = bobbySession(bbyAccount,bbyPasswd) async def main(ws:djtuWebsocket, *args: Any, **kwargs: Any) -> None: async with ws as ws: logger.info("启动并发异步ing...") mcloop = ws.mcloop() # 你*的为什么retcoro true的话会报错啊 # 协程对象不会不是这个异步回话而是httpx吧？ bot = ws.start(*args, **kwargs,ret_coro=False) if bot is None: logger.error("bot不给协程啊") return try: await asyncio.gather(bot,mcloop) except Exception as e : logger.error(e) # bot初始化 ws = djtuWebsocket(Intents_ex(public_messages=True,member_event=True)) # 启动mcwatcher线程 def pushQuene(cont:str): ws.q.put(cont) Thread(target=watcher.MCWatcher,name="watcher",kwargs={ "hookfunction":pushQuene, "logging":logger }).start() # bot线程运行 asyncio.run(main(ws,appid=appid,secret=secret_bot))