929 lines
38 KiB
Python
929 lines
38 KiB
Python
# -*- 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"]) |