# -*- coding: utf-8 -*- import asyncio import os from jieba import posseg import re import json 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 from datetime import datetime, timedelta import requests from Player_Query import Character_Query import csv from Vechile import Vechile_Character_Query from Specific_vehicle import Specific_query,sorting from pathlib import Path 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: url = message.attachments[0].url response = requests.get(url) image_dir = os.path.join(os.path.dirname(__file__)) if response.status_code == 200: # 将图片数据写入本地文件 file_base = message.content.replace("投稿", "").strip() if file_base != '': image_name = file_base + '.jpg' image_path = os.path.join(os.path.dirname(__file__),image_name) else: image_name = message.id[:-8] + '.jpg' image_path = os.path.join(os.path.dirname(__file__),image_name) with open(image_path, 'wb') as file: file.write(response.content) messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=f"图片下载成功！", ) _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) _log.info(messageResult) elif '角色查询' in text: stripped_msg = text.strip() stripped_msg = stripped_msg.replace("，", ",") cleaned_text = re.sub(r'[,\s]+', ' ', stripped_msg) user_text = cleaned_text.split() _log.info(user_text) words = posseg.lcut(stripped_msg) msg = { 'ID': '', 'Num':20, 'Mode':'Normal', "Status":'', "TimeLine":'', } msg['ID'] = user_text[1] for word in words: # 每个元素是一个 pair 对象，包含 word 和 flag 两个属性，分别表示词和词性 if word.flag == "que": msg['Mode'] = (word.word).lower() break elif (word.word) == '死亡': msg['Mode'] = "death" elif (word.word) == '评分': msg['Mode'] = "ivi" elif (word.word) == '时间': msg['Mode'] = "time" elif (word.word) == '击杀': msg['Mode'] = "kill" #查询数量记录 try: msg['num'] = user_text[3] except: msg['num'] = 20 if msg['ID'] != '' : #start = await Character_Query().test_re(msg) messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 1, content=str('数据检索中...'), ) _log.info(messageResult) try: date = await Character_Query().Character_date(msg['ID']) except: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 2, content=str('API沟通失败，请重试...'), ) _log.info(messageResult) try: code = date['code'] if code == 10000: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 2, content=str('未找到该玩家数据，请检查ID是否正确'), ) _log.info(messageResult) except: Player_BasicDate = date['basicInfo'] Player_LatestSessionInfo = date['latestSessionInfo'] Character_Id = Player_BasicDate['characterFullName'] if Player_BasicDate['outfitAlias'] == None: Character_outfitAlias = '' else: Character_outfitAlias = '[' + Player_BasicDate['outfitAlias'] + ']' Online_flag = Player_BasicDate['online'] try: if Online_flag == True : logintimeline = int(Player_LatestSessionInfo['latestLoginDate']) logintime = datetime.fromtimestamp(logintimeline) formatted_time = logintime.strftime("%Y-%m-%d %H:%M") elif Online_flag == False : Logouttimeline = int(Player_LatestSessionInfo['latestLogoutDate'])/1000 Logouttime = datetime.fromtimestamp(int(Logouttimeline)) formatted_time = Logouttime.strftime("%Y-%m-%d %H:%M") except: formatted_time = '数据库无记录' msg['Character_ID'] = Character_outfitAlias + ' ' + Character_Id msg['ID'] = Character_Id msg['Status'] = Online_flag msg['TimeLine'] = formatted_time web_url = 'https://bot.file.sunderer.games' if msg['Mode'] == 'Normal': print(msg['Mode']) overname = await Character_Query().Date_OverView(msg,date) url = web_url + '/' + overname try: uploadMedia = await message._api.post_group_file( group_openid=message.group_openid, file_type=1, # 文件类型要对应上，具体支持的类型见方法说明 url=url # 文件Url ) messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=7, msg_id=message.id, msg_seq = 4, #content=f"收到了消息：{message.content}", media=uploadMedia, ) _log.info(messageResult) except: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 3909, content=str('QQbot接口请求错误，请重试'), ) _log.info(messageResult) else: print(msg['Mode']) cycle = await Character_Query().Date_list(msg,date) #web_url = 'http://xm1.dc.sunderer.games:3100/' for i in range(0,cycle): img_id = str(Character_Id) + "({}).jpg".format(i) Character_url = web_url + '/' + img_id try: uploadMedia = await message._api.post_group_file( group_openid=message.group_openid, file_type=1, # 文件类型要对应上，具体支持的类型见方法说明 url=Character_url # 文件Url ) messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=7, msg_id=message.id, msg_seq = 3 + (i-1), #content=f"收到了消息：{message.content}", media=uploadMedia, ) _log.info(messageResult) except: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 3 + (i-1), content=str('QQbot接口请求错误，请重试'), ) _log.info(messageResult) for j in range(0,cycle): img_id = str(Character_Id) + "({}).jpg".format(j) file_path = os.path.join('/home/bot/QQbot/Charter_Sundy_Query/Output',img_id) os.remove(file_path) #User_Name = 'MMMMMMMMMM' else: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 2, content=str('请检查是否输入ID'), ) _log.info(messageResult) elif '指令帮助' in text: textt = "查询指令格式为：【角色ID】，【筛选模式】，【数量】\r查询指令通过逗号（，）连接\r例如：MicroYX，击杀，20\r Aokiwasuke，kdr，40\r筛选模式为：\r击杀，死亡，\r 倒地，评分，\r 时间，KDR，\r KPM，HSR\r ACC， IVI\r数量可以任意填写，但数量多会导致生成时间过慢。\r如有不理解或者数据错误，请@群管理。" messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=str(textt), ) elif '在线验证' in text: stripped_msg = text.strip() words = posseg.lcut(stripped_msg) msg = { 'ID': '', 'Num':20, 'Mode':'Normal', "Status":'', "TimeLine":'', } for word in words: if word.flag == 'eng': msg['ID'] = word.word break flag = await Fuck_tx().Online_Query(msg['ID']) print(flag) messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, content=str(flag), ) _log.info(messageResult) elif '查询人口' in text: file_url = 'https://bot.file.sunderer.games/error.png'#'test.png error.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: stripped_msg = text.strip() stripped_msg = stripped_msg.replace("，", ",") cleaned_text = re.sub(r'[,\s]+', ' ', stripped_msg) user_text = cleaned_text.split() _log.info(user_text) words = posseg.lcut(stripped_msg) msg = { 'ID': '', 'Num':20, 'Mode':'Normal', "Status":'', "TimeLine":'', } msg['ID'] = user_text[1] for word in words: # 每个元素是一个 pair 对象，包含 word 和 flag 两个属性，分别表示词和词性 if word.flag == "que": msg['Mode'] = (word.word).lower() break elif (word.word) == '死亡': msg['Mode'] = "death" elif (word.word) == '评分': msg['Mode'] = "ivi" elif (word.word) == '时间': msg['Mode'] = "time" elif (word.word) == '击杀': msg['Mode'] = "kill" if msg['ID'] != '' : #start = await Character_Query().test_re(msg) messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 1, content=str('数据检索中...'), ) _log.info(messageResult) try: date = await Character_Query().Character_date(msg['ID']) except: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 2, content=str('API沟通失败，请重试...'), ) _log.info(messageResult) try: code = date['code'] if code == 10000: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 2, content=str('未找到该玩家数据，请检查ID是否正确'), ) _log.info(messageResult) except: web_body = await Vechile_Character_Query().Date_Rinse_common(date) web_url = 'https://bot.file.sunderer.games/' file_url = web_url + web_body 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_seq= 3, msg_id=message.id, #content=f"收到了消息：{message.content}", media=uploadMedia, ) _log.info(messageResult) file_path = os.path.join('/home/bot/QQbot/Charter_Sundy_Query/Output',web_body) os.remove(file_path) elif '载具查询' in text: stripped_msg = text.strip() stripped_msg = stripped_msg.replace("，", ",") cleaned_text = re.sub(r'[,\s]+', ' ', stripped_msg) user_text = cleaned_text.split() _log.info(user_text) words = posseg.lcut(stripped_msg) msg = { 'ID': '', 'Num':20, 'Mode':'kill', "Status":'', "TimeLine":'', 'Select':'', 'user_give_faction':'' } msg['ID'] = user_text[1] for word in words: if word.flag == "vea": msg['Select'] = word.word if word.word == '蚊子': msg["Select"] = 'Mosquito' elif word.word == '镰刀': msg["Select"] = 'Scythe' elif word.word == '海盗': msg["Select"] = 'Reaver' elif word.word == '电磁': msg["Select"] = 'Magrider' elif word.word == '徘徊': msg["Select"] = 'Prowler' elif word.word == '决裂': msg["Select"] = 'Sunderer' elif word.word == '先驱': msg["Select"] = 'Vanguard' elif word.word == '奇美拉': msg["Select"] = 'Chimera' elif word.word == '突袭': msg["Select"] = 'Harasser' elif word.word == '银河': msg["Select"] = 'Galaxy' elif word.word == '女武神': msg["Select"] = 'Valkyrie' elif word.word == '解放': msg["Select"] = 'Liberator' elif word.word == '北极星': msg["Select"] = 'Lodestar Prototype' elif word.word == '火花': msg["Select"] = 'Flash' elif word.word == '闪电': msg["Select"] = 'Lightning' elif word.word == '巨像': msg["Select"] = 'Colossus' elif word.word == '私掠者': msg["Select"] = 'Corsair' elif word.word == '标枪': msg["Select"] = 'Javelin' elif word.word == '苦行僧': msg["Select"] = 'Dervish' if msg['Select'] == 'Mosquito' or msg['Select'] =='Prowler' : msg['user_give_faction'] = "3" elif msg['Select'] == 'Scythe' or msg['Select'] == 'Magrider' : msg['user_give_faction'] = "1" elif msg['Select'] == 'Reaver' or msg['Select'] == 'Vanguard' : msg['user_give_faction'] = "2" elif msg['Select'] == 'Dervish' or msg['Select'] == 'Chimera' : msg['user_give_faction'] = "4" _log.info(msg) for word in words: # 每个元素是一个 pair 对象，包含 word 和 flag 两个属性，分别表示词和词性 if word.flag == "que": msg['Mode'] = (word.word).lower() break elif word.flag == 'qua': msg['Mode'] = (word.word) if msg['Mode']== '击杀':# msg['Mode'] = "kill" elif msg['Mode']== '开火次数':# msg['Mode'] = "fire" elif msg['Mode']== '命中次数':# msg['Mode'] = "hit" elif msg['Mode']== '命中率':# msg['Mode'] = "acc" elif msg['Mode']== '击杀平均开火数':# msg['Mode'] = "spk" elif msg['Mode']== '击杀平均命中数':# msg['Mode'] = "hpk" elif msg['Mode']== '伤害量':# msg['Mode'] = "dmg" elif msg['Mode']== '分钟平均得分数':# msg['Mode'] = "spm" elif msg['Mode']== '累计使用时间':# msg['Mode'] = "time" if msg['ID'] != '' : #start = await Character_Query().test_re(msg) messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 1, content=str('数据检索中...'), ) _log.info(messageResult) try: date = await Character_Query().Character_date(msg['ID']) except: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 2, content=str('API沟通失败，请重试...'), ) _log.info(messageResult) try: code = date['code'] if code == 10000: messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 2, content=str('未找到该玩家数据，请检查ID是否正确'), ) _log.info(messageResult) except: Vechile_BasicDate_ori = date["battleData"]['vehicleData'] airvichile = ['Mosquito','Scythe','Reaver','Dervish'] landvichile = ['Chimera','Vanguard','Magrider','Prowler'] if msg['user_give_faction'] != None: Chart_faction = str(date["basicInfo"]["factionId"]) _log.info(Chart_faction) if Chart_faction == msg['user_give_faction']: _log.info('阵营与查询目标吻合') else: _log.info('阵营与查询目标不吻合') if Chart_faction == '1': if msg['Select'] in airvichile: msg['Select'] = 'Scythe' elif msg['Select'] in landvichile: msg['Select'] = 'Magrider' elif Chart_faction == '2': if msg['Select'] in airvichile: msg['Select'] = 'Reaver' elif msg['Select'] in landvichile: msg['Select'] = 'Vanguard' elif Chart_faction == '3': if msg['Select'] in airvichile: msg['Select'] = 'Mosquito' elif msg['Select'] in landvichile: msg['Select'] = 'Prowler' elif Chart_faction == '4': if msg['Select'] in airvichile: msg['Select'] = 'Dervish' elif msg['Select'] in landvichile: msg['Select'] = 'Chimera' else: _log.info('line 693 error') messageResult = await message._api.post_group_message( group_openid=message.group_openid, msg_type=0, msg_id=message.id, msg_seq= 2, content=str('看起来这个角色并没有敌方阵营的载具，让我帮你换一下。'), ) _log.info(messageResult) _log.info(msg) Vechile_BasicDate = await Specific_query().select_vehicle(Vechile_BasicDate_ori, Select=msg["Select"]) #_log.info(Vechile_BasicDate) Vechile_weapon_data = Vechile_BasicDate[0]['weaponStat'] sorting_data = await sorting().Data_sorting(Vechile_weapon_data, msg['Mode']) list_num = len(Vechile_weapon_data)+1 web_body = await Specific_query().vechile_img(date, Vechile_BasicDate, sorting_data, list_num, msg['Mode']) web_url = 'https://bot.file.sunderer.games/' file_url = web_url + web_body 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_seq= 3, msg_id=message.id, #content=f"收到了消息：{message.content}", media=uploadMedia, ) _log.info(messageResult) file_path = os.path.join('/home/bot/QQbot/Charter_Sundy_Query/Output',web_body) os.remove(file_path) 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 = 'https://bot.file.sunderer.games/' 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 = 'error.png'#'/Soltech_Sundy.jpg'#'error.png'# elif Server == "温怀特": web_body = 'error.png'#'/Wainwright_Sundy.jpg'#'error.png'# elif Server == "奥斯普雷": web_body = 'error.png'#'/Osprey_Sundy.jpg'#'error.png'# 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 async def Online_Query(self,id): url = "https://api.sunderer.games/census/character/search" querystring = {"name": ""} querystring["name"] = id response = requests.request("GET", url, params=querystring) response = response.json() online_flag = response['basicInfo']['online'] return online_flag 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"])