ok
This commit is contained in:
@@ -0,0 +1,448 @@
|
||||
# pyright: reportArgumentType=false
|
||||
# pyright: reportAttributeAccessIssue=false
|
||||
# pyright: reportIndexIssue=false
|
||||
# pyright: reportOptionalMemberAccess=false
|
||||
# pyright: reportCallIssue=false
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Dict, Any, Optional, List, Union
|
||||
import base64
|
||||
|
||||
from astrbot.api import logger
|
||||
from astrbot.api import AstrBotConfig
|
||||
|
||||
from .request import APIClient
|
||||
from .sqlite import AsyncSQLiteDB
|
||||
from .fun_basic import load_template,extract_fields
|
||||
|
||||
class GOKServer:
|
||||
def __init__(self, api_config, config:AstrBotConfig, sqlite:AsyncSQLiteDB ):
|
||||
self._api = APIClient()
|
||||
# 引用API配置文件
|
||||
self._api_config = api_config
|
||||
# 引用插件配置文件
|
||||
self._config = config
|
||||
# 引用数据库类
|
||||
self._sql_db = sqlite
|
||||
|
||||
# 获取配置中的 Token
|
||||
self.token = self._config.get("ytapi_token", "")
|
||||
if self.token == "":
|
||||
logger.info("获取配置token失败，请正确填写token,否则部分功能无法正常使用")
|
||||
else:
|
||||
logger.debug(f"获取配置token成功。{self.token}")
|
||||
|
||||
|
||||
async def close(self):
|
||||
"""释放底层 APIClient 资源"""
|
||||
if self._api:
|
||||
await self._api.close()
|
||||
self._api = None
|
||||
|
||||
|
||||
def _init_return_data(self) -> Dict[str, Any]:
|
||||
"""初始化标准的返回数据结构"""
|
||||
return {
|
||||
"code": 0,
|
||||
"msg": "功能函数未执行",
|
||||
"data": {}
|
||||
}
|
||||
|
||||
|
||||
async def _base_request(
|
||||
self,
|
||||
config_key: str,
|
||||
method: str,
|
||||
params: Optional[Dict[str, Any]] = None,
|
||||
out_key: Optional[str] = "data"
|
||||
) -> Optional[Any]:
|
||||
"""
|
||||
基础请求封装，处理配置获取和API调用。
|
||||
|
||||
:param config_key: 配置字典中对应 API 的键名。
|
||||
:param method: HTTP方法 ('GET' 或 'POST')。
|
||||
:param params: 请求参数或 Body 数据。
|
||||
:param out_key: 响应数据中需要提取的字段。
|
||||
:return: 成功时返回提取后的数据，失败时返回 None。
|
||||
"""
|
||||
try:
|
||||
api_config = self._api_config.get(config_key)
|
||||
if not api_config:
|
||||
logger.error(f"配置文件中未找到 key: {config_key}")
|
||||
return None
|
||||
|
||||
# 复制 params，避免修改原始配置模板
|
||||
request_params = api_config.get("params", {}).copy()
|
||||
if params:
|
||||
request_params.update(params)
|
||||
|
||||
url = api_config.get("url", "")
|
||||
if not url:
|
||||
logger.error(f"API配置缺少 URL: {config_key}")
|
||||
return None
|
||||
|
||||
if method.upper() == 'POST':
|
||||
data = await self._api.post(url, data=request_params, out_key=out_key)
|
||||
else: # 默认为 GET
|
||||
data = await self._api.get(url, params=request_params, out_key=out_key)
|
||||
|
||||
if not data:
|
||||
logger.warning(f"获取接口信息失败或返回空数据: {config_key}")
|
||||
|
||||
return data
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"基础请求调用出错 ({config_key}): {e}")
|
||||
return None
|
||||
|
||||
|
||||
# --- 业务功能函数 ---
|
||||
async def helps(self) -> Dict[str, Any]:
|
||||
"""功能"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 加载模板
|
||||
try:
|
||||
return_data["temp"] = await load_template("helps.html")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
|
||||
return_data["code"] = 200
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def add(self,gokid: int, name: str) -> Dict[str, Any]:
|
||||
"""角色添加 王者营地ID 角色"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 添加数据
|
||||
try:
|
||||
await self._sql_db.insert(
|
||||
"users",
|
||||
{
|
||||
"gokid": gokid,
|
||||
"name": name,
|
||||
}
|
||||
)
|
||||
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"添加角色失败: {e}")
|
||||
return_data["msg"] = "添加角色失败"
|
||||
return return_data
|
||||
|
||||
return_data["data"] = (
|
||||
"角色添加成功\n"
|
||||
f"王者营地ID：{gokid}\n"
|
||||
f"角色名称：{name}\n"
|
||||
)
|
||||
|
||||
return_data["code"] = 200
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def all(self) -> Dict[str, Any]:
|
||||
"""角色查看"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
|
||||
# 查询数据
|
||||
try:
|
||||
data = await self._sql_db.select_all("users")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"查看角色失败: {e}")
|
||||
return_data["msg"] = "查看角色失败"
|
||||
return return_data
|
||||
|
||||
if not data:
|
||||
return_data["msg"] = "未找到角色数据"
|
||||
return return_data
|
||||
|
||||
|
||||
# 加载模板
|
||||
try:
|
||||
return_data["temp"] = await load_template("jueshe.html")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
|
||||
# 数据处理
|
||||
return_data["data"]["lists"] = data
|
||||
|
||||
return_data["code"] = 200
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def select(self, name) -> Dict[str, Any]:
|
||||
"""角色查询 名称"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 判断输入是否为整数
|
||||
try:
|
||||
int(name)
|
||||
if int(name) >=100000000:
|
||||
like = "gokid LIKE ?"
|
||||
else:
|
||||
raise Exception("id数据异常")
|
||||
except (ValueError, TypeError, Exception):
|
||||
like = "name LIKE ?"
|
||||
|
||||
# 模糊拼接
|
||||
like_name = f"%{name}%"
|
||||
|
||||
# 查询数据
|
||||
try:
|
||||
data = await self._sql_db.select_all(
|
||||
"users",
|
||||
like,
|
||||
(like_name,)
|
||||
)
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"查询角色失败: {e}")
|
||||
return_data["msg"] = "查询角色失败"
|
||||
return return_data
|
||||
|
||||
if not data:
|
||||
return_data["msg"] = "未查询到角色数据"
|
||||
return return_data
|
||||
|
||||
|
||||
# 加载模板
|
||||
try:
|
||||
return_data["temp"] = await load_template("jueshe.html")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
|
||||
# 数据处理
|
||||
return_data["data"]["lists"] = data
|
||||
|
||||
return_data["code"] = 200
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def update(self, gokid:int, name: str) -> Dict[str, Any]:
|
||||
"""角色修改 王者营地ID 角色"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
data = await self._sql_db.select_one(
|
||||
"users",
|
||||
"gokid=?",
|
||||
(gokid,)
|
||||
)
|
||||
|
||||
if not data:
|
||||
return_data["msg"] = "没有当前ID"
|
||||
return return_data
|
||||
|
||||
|
||||
# 修改数据
|
||||
try:
|
||||
await self._sql_db.update(
|
||||
"users",
|
||||
{
|
||||
"name": name,
|
||||
},
|
||||
"gokid=?",
|
||||
(gokid,)
|
||||
)
|
||||
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"避雷修改失败: {e}")
|
||||
return_data["msg"] = "避雷修改失败"
|
||||
return return_data
|
||||
|
||||
return_data["data"] = (
|
||||
"角色修改成功\n"
|
||||
f"王者营地ID：{gokid}\n"
|
||||
f"角色名称：{name}\n"
|
||||
)
|
||||
|
||||
return_data["code"] = 200
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def delete(self, gokid:int) -> Dict[str, Any]:
|
||||
"""角色删除 王者营地ID"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
data = await self._sql_db.select_one(
|
||||
"users",
|
||||
"gokid=?",
|
||||
(gokid,)
|
||||
)
|
||||
|
||||
if not data:
|
||||
return_data["msg"] = "没有当前ID"
|
||||
return return_data
|
||||
|
||||
# 删除
|
||||
try:
|
||||
await self._sql_db.delete(
|
||||
"users",
|
||||
"gokid=?",
|
||||
(gokid,)
|
||||
)
|
||||
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"角色删除失败: {e}")
|
||||
return_data["msg"] = "角色删除失败"
|
||||
return return_data
|
||||
|
||||
return_data["data"] = f"角色删除成功。王者营地ID：{gokid}"
|
||||
|
||||
return_data["code"] = 200
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def get_gokid(self,name: str):
|
||||
# 判断输入是否为整数
|
||||
try:
|
||||
# 直接返回输入
|
||||
int(name)
|
||||
if int(name) >=100000000:
|
||||
gokid = name
|
||||
return gokid
|
||||
else:
|
||||
raise Exception("输入不是ID")
|
||||
except (ValueError, TypeError, Exception):
|
||||
# 查询数据
|
||||
try:
|
||||
data = await self._sql_db.select_one(
|
||||
"users",
|
||||
"name=?",
|
||||
(name,)
|
||||
)
|
||||
|
||||
if not data:
|
||||
gokid = None
|
||||
return gokid
|
||||
|
||||
gokid = data['gokid']
|
||||
return gokid
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"查询角色失败: {e}")
|
||||
gokid = None
|
||||
return gokid
|
||||
|
||||
|
||||
async def zhanji(self,name: str ,option: str):
|
||||
"""
|
||||
战绩查询
|
||||
"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 获取配置中的 Token
|
||||
token = self._config.get("ytapi_token", "")
|
||||
if token == "":
|
||||
return_data["msg"] = "系统未配置API访问Token"
|
||||
return return_data
|
||||
|
||||
# ID查询
|
||||
gokid = await self.get_gokid(name)
|
||||
|
||||
if not gokid :
|
||||
return_data["msg"] = "未查询到该用户，请确认输入正确的角色或营地ID"
|
||||
return return_data
|
||||
|
||||
#更新参数
|
||||
params = {"id": gokid, "option": option, "key": token}
|
||||
|
||||
# 需要提取的字段
|
||||
fields = ["gametime","killcnt","deadcnt","assistcnt","gameresult","mvpcnt","losemvp","mapName",
|
||||
"oldMasterMatchScore","newMasterMatchScore","usedTime","winNum","failNum","roleJobName","stars","desc",
|
||||
"gradeGame","heroIcon","godLikeCnt", "firstBlood","hero1TripleKillCnt","hero1UltraKillCnt","hero1RampageCnt","evaluateUrlV3","mvpUrlV3"]
|
||||
|
||||
# 获取数据
|
||||
data: Optional[List[Dict[str, Any]]] = await self._base_request("gok_zhanji", "GET", params=params)
|
||||
if not data:
|
||||
return_data["msg"] = "获取接口信息失败"
|
||||
return return_data
|
||||
|
||||
# 处理返回数据
|
||||
try:
|
||||
|
||||
# 提取字段
|
||||
result = extract_fields(data['list'], fields)
|
||||
result = result[:25]
|
||||
# 数据处理
|
||||
for m in result:
|
||||
minutes = m["usedTime"] // 60
|
||||
seconds = m["usedTime"] % 60
|
||||
m["time_str"] = f"{minutes}:{seconds:02d}"
|
||||
|
||||
return_data["data"]["data"] = result
|
||||
except Exception as e:
|
||||
logger.error(f"处理数据时出错: {e}")
|
||||
return_data["msg"] = "处理接口返回信息时出错"
|
||||
return return_data
|
||||
|
||||
# 加载模板
|
||||
try:
|
||||
return_data["temp"] = await load_template("wangzhezhanji.html")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
|
||||
return_data["code"] = 200
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def ziliao(self, name: str):
|
||||
return_data = self._init_return_data()
|
||||
# 获取配置中的 Token
|
||||
token = self._config.get("ytapi_token", "")
|
||||
if token == "":
|
||||
return_data["msg"] = "系统未配置API访问Token"
|
||||
return return_data
|
||||
|
||||
# ID查询
|
||||
gokid = await self.get_gokid(name)
|
||||
|
||||
if not gokid :
|
||||
return_data["msg"] = "未查询到该用户，请确认输入正确的角色或营地ID"
|
||||
return return_data
|
||||
|
||||
#更新参数
|
||||
params = {"id": gokid, "key": token}
|
||||
|
||||
# 获取数据
|
||||
data: Optional[List[Dict[str, Any]]] = await self._base_request("gok_ziliao", "GET", params=params, out_key="")
|
||||
logger.debug(data)
|
||||
|
||||
if not data:
|
||||
return_data["msg"] = "获取接口信息失败"
|
||||
return return_data
|
||||
|
||||
try:
|
||||
# 转 base64
|
||||
return_data["data"]["img_base64"] = base64.b64encode(data).decode("utf-8")
|
||||
except Exception as e:
|
||||
logger.error(f"处理数据时出错: {e}")
|
||||
return_data["msg"] = "处理接口返回信息时出错"
|
||||
return return_data
|
||||
|
||||
# 加载模板
|
||||
try:
|
||||
return_data["temp"] = await load_template("wzry_zl.html")
|
||||
except FileNotFoundError:
|
||||
logger.error(f"加载模板失败")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
|
||||
return_data["code"] = 200
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
Reference in New Issue
Block a user