# -*- coding: utf-8 -*- import asyncio import os from jieba import posseg import yaml import jieba import botpy from botpy import logging, BotAPI from botpy.ext.cog_yaml import read from botpy.message import GroupMessage, Message,C2CMessage from botpy.manage import C2CManageEvent from botpy.ext.command_util import Commands import time import requests test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml")) jieba.load_userdict(os.path.join(os.path.dirname(__file__), "dict.txt")) _log = logging.get_logger() class MyClient(botpy.Client): async def on_ready(self): _log.info(f"robot 「{self.robot.name}」 on_ready!") async def on_group_at_message_create(self, message: GroupMessage): text = (message.content).strip() if 'Error' in text or '正常' in text or '维护模式' in text or '当前模式' in text or '测试' in text: stripped_msg = text.strip() words = posseg.lcut(stripped_msg) for word in words: # 每个元素是一个 pair 对象，包含 word 和 flag 两个属性，分别表示词和词性 if word.flag == "m": # ns 词性表示地名 Verification = str(int(word.word)) Verification = Verification.zfill(6) break Ver = await Fuck_tx().Code_Verifica(Verification) if Ver == True: if 'Error' in text: flag = await Fuck_tx().Error_Mode() if flag == True: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=f"Reply_Mode修改为1", ) _log.info(messageResult) else: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=f"Reply_Mode修改失败", ) _log.info(messageResult) elif '正常' in text: flag = await Fuck_tx().Common_Mode() if flag == True: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=f"Reply_Mode修改为0", ) _log.info(messageResult) else: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=f"Reply_Mode修改失败", ) _log.info(messageResult) elif '当前模式' in text: text = await Fuck_tx().Now_Cfg() if text['Reply_Mode'] == '0': messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=f"当前为普通模式，Reply_Mode为:{text['Reply_Mode']}", ) _log.info(messageResult) elif text['Reply_Mode'] == '1': messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=f"当前为Error模式，Reply_Mode为:{text['Reply_Mode']}", ) _log.info(messageResult) elif '测试' in text: text = await Fuck_tx().Now_Cfg() if text['Reply_Mode'] == '0': messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, #不包含message_id为主动发文 content=f"当前为普通模式，Reply_Mode为:{text['Reply_Mode']}", ) _log.info(messageResult) elif text['Reply_Mode'] == '1': messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=f"当前为Error模式，Reply_Mode为:{text['Reply_Mode']}", ) _log.info(messageResult) elif '维护模式' in text: flag = await Fuck_tx().Server_Mode() if flag == True: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=f"Reply_Mode修改为2", ) _log.info(messageResult) else: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=f"Reply_Mode修改失败", ) _log.info(messageResult) else: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=f"你谁啊你", ) _log.info(messageResult) elif '你瞅啥' in text: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id,#'ROBOT1.0_iqHc6IePSUh91RXfdrmBvNQE6XX6zpF9LAGEqhTRX5JcmtPXIy7o3e6hAHskAxaMNjMB4nSmAPGOUqoyZisnpw!!', content=f"瞅你咋地", ) _log.info(messageResult) elif '人口查询' in text: file_url = 'http://ly1.dc.sunderer.games:3000/test.png' uploadMedia = await message._api.post_group_file( group_openid=message.group_openid, file_type=1, # 文件类型要对应上，具体支持的类型见方法说明 url=file_url # 文件Url ) messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=7, msg_id=message.id, #content=f"收到了消息：{message.content}", media=uploadMedia, ) _log.info(messageResult) elif '查询' in text: file_url = await Fuck_tx().Server_query(message) uploadMedia = await message._api.post_group_file( group_openid=message.group_openid, file_type=1, # 文件类型要对应上，具体支持的类型见方法说明 url=file_url # 文件Url ) messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=7, msg_id=message.id, #content=f"收到了消息：{message.content}", media=uploadMedia, ) _log.info(messageResult) class Fuck_tx(): async def Server_query(self,message): text = message.content web_url = 'http://ly1.dc.sunderer.games:3000/' stripped_msg = text.strip() words = posseg.lcut(stripped_msg) Server = None for word in words: # 每个元素是一个 pair 对象，包含 word 和 flag 两个属性，分别表示词和词性 if word.flag == "ser": # ns 词性表示地名 Server = word.word break if Server != None: test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml")) if test_config['Reply_Mode'] == '0': if Server == "索泰克": web_body = '/Soltech.jpg' elif Server == "米勒": web_body = '/Miller.jpg' elif Server == "艾莫若德": web_body = '/Emerald.jpg' elif Server == "康纳利": web_body = '/Connery.jpg' elif Server == "柯伯特": web_body = '/Cobalt.jpg' file_url = web_url + web_body return file_url elif test_config['Reply_Mode'] == '1': web_body = 'error.png' file_url = web_url + web_body return file_url elif test_config['Reply_Mode'] == '2': web_body = 'Server.jpg' file_url = web_url + web_body return file_url async def Error_Mode(self): try: test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml")) test_config['Reply_Mode'] = '1' with open((os.path.join(os.path.dirname(__file__), "config.yaml")), "w") as file: yaml.dump(test_config, file) return True except: return False async def Server_Mode(self): try: test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml")) test_config['Reply_Mode'] = '2' with open((os.path.join(os.path.dirname(__file__), "config.yaml")), "w") as file: yaml.dump(test_config, file) return True except: return False async def Common_Mode(self): try: test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml")) test_config['Reply_Mode'] = '0' with open((os.path.join(os.path.dirname(__file__), "config.yaml")), "w") as file: yaml.dump(test_config, file) return True except: return False async def Now_Cfg(self): try: test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml")) return test_config except: return '出错了' async def Code_Verifica(self,code): time_now = str(int(time.time())) time00 = time_now[0:4] time1 = str(code)[::-1] time_ver = time00+time1 if int(time_now) < int(time_ver) and (int(time_ver)-300) < int(time_now): return True else: return False async def Code_Gen(self): time1 = str(int(time.time())+300) time3 = time1[::-1][:-4] return time3 async def Clean_All_Server(self): url = "https://api.sunderer.games/census/restart" payload = "-----011000010111000001101001\r\nContent-Disposition: form-data; name=\"page_no\"\r\n\r\n1\r\n-----011000010111000001101001\r\nContent-Disposition: form-data; name=\"page_size\"\r\n\r\n15\r\n-----011000010111000001101001--\r\n\r\n" headers = {"content-type": "multipart/form-data; boundary=---011000010111000001101001"} response = requests.request("POST", url, data=payload, headers=headers) return response.text async def Clean_Specific_Server(self,server): url = "https://api.sunderer.games/census/restart/server" payload = {"ServerName": ""} payload['ServerName'] = str(server) headers = {"content-type": "application/json"} response = requests.request("POST", url, json=payload, headers=headers) return response.text if __name__ == "__main__": # 通过预设置的类型，设置需要监听的事件通道 # intents = botpy.Intents.none() # intents.public_messages=True # 通过kwargs，设置需要监听的事件通道 intents = botpy.Intents(public_messages=True) client = MyClient(intents=intents) client.run(appid=test_config["appid"], secret=test_config["secret"])