555 lines
24 KiB
Python
555 lines
24 KiB
Python
# -*- coding: utf-8 -*-
|
||
import asyncio
|
||
import os
|
||
from jieba import posseg
|
||
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
|
||
|
||
|
||
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()
|
||
words = posseg.lcut(stripped_msg)
|
||
msg = { 'ID': '',
|
||
'Num':20,
|
||
'Mode':'Normal',
|
||
"Status":'',
|
||
"TimeLine":'',
|
||
}
|
||
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"
|
||
|
||
for word in words:
|
||
if word.flag == 'm':
|
||
msg['Num'] = int(word.word)
|
||
break
|
||
|
||
for word in words:
|
||
if word.flag == 'eng':
|
||
msg['ID'] = word.word
|
||
break
|
||
|
||
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:
|
||
|
||
file_url = 'https://bot.file.sunderer.games/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 = '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 = '/Soltech_Sundy.jpg'
|
||
elif Server == "米勒":
|
||
web_body = '/沉痛悼念米勒.png'
|
||
elif Server == "艾莫若德":
|
||
web_body = '/沉痛悼念艾莫若德.png'
|
||
elif Server == "康纳利":
|
||
web_body = '/沉痛悼念康纳利.png'
|
||
elif Server == "柯伯特":
|
||
web_body = '/沉痛悼念柯伯特.png'
|
||
elif Server == "奥斯普雷":
|
||
web_body = '/Connery.jpg'
|
||
elif Server == "温怀特":
|
||
web_body = '/Miller.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"]) |