hiso-qq-bot/bot.py

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 dotenv import load_dotenv
import os 
import types
from bobby import bobbySession
from mcwatcher import watcher  
from threading import Thread
import asyncio
#环境配置  

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://starptr.top')

# 触发概率不大所以可以直接提取关键字
    async def parseCommand(self,msg :Message):
        commands:dict[str,types.MethodType|types.FunctionType] = {
            r'/邀请码' : bobby.getInviteCode,
            r'吃什么' : lambda :"是啊吃什么"
        }
        required = msg.content
        reply = ""
        for cmd,func in commands.items():
            if required.find(cmd) != -1:
                try:
                    reply = func()
                except:
                    reply = "服务异常"
                break
        if reply.__len__() == 0:
            reply = "你在发什么东西我怎么看不懂?"
        await msg.reply(content=reply)

# 触发概率大，以后要做好解析防止刷屏
    async def parseGlobalMessage(self,payload):
        try :
            msg = payload['d']
            gid = msg['group_id']
            content = msg['content']
        except:
            return
        # 加入群聊
        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)
        # 解析指令
        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))