dev
This commit is contained in:
@@ -2,10 +2,8 @@ import base64
|
||||
import hmac
|
||||
import json
|
||||
import time
|
||||
from typing import Optional
|
||||
from core.character.adventure import get_adv_info
|
||||
from core.basic.character import Character
|
||||
from fastapi import Header, Request
|
||||
from fastapi import Header
|
||||
from api.core.exception import ResponseException
|
||||
|
||||
|
||||
@@ -67,7 +65,7 @@ def deleteToken(token, redis, env):
|
||||
return token
|
||||
|
||||
|
||||
def uid(access_uid: Optional[str] = Header(None)):
|
||||
def uid(access_uid: str | None = Header(None)):
|
||||
try:
|
||||
return int(access_uid)
|
||||
except Exception:
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import gzip
|
||||
from fastapi import Request, APIRouter, FastAPI
|
||||
from fastapi.responses import Response
|
||||
from typing import Callable
|
||||
from collections.abc import Callable
|
||||
from fastapi.middleware.gzip import GZipMiddleware
|
||||
from fastapi.routing import APIRoute
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ def get_redis_info(redis, key, fun,without_gzip=False):
|
||||
info = json.loads(redis.get(key))
|
||||
else:
|
||||
info = json.loads(gzip_unzip(redis.get(key)))
|
||||
except Exception as ex:
|
||||
except Exception:
|
||||
info = None
|
||||
if info is None:
|
||||
info = fun()
|
||||
|
||||
@@ -1,20 +1,20 @@
|
||||
import base64
|
||||
import json
|
||||
from typing import Generic, TypeVar
|
||||
from pydantic import Field
|
||||
from pydantic.generics import GenericModel
|
||||
import gzip # 导入 gzip 模块
|
||||
from io import BytesIO # 从 io 模块导入 BytesIO 类
|
||||
|
||||
|
||||
T = TypeVar('T')
|
||||
|
||||
|
||||
class Return(GenericModel, Generic[T]):
|
||||
code: int = 200
|
||||
message: str = ''
|
||||
data: T | dict | str = None
|
||||
|
||||
|
||||
code: int = Field(..., description='响应状态码')
|
||||
"""响应状态码"""
|
||||
message: str = Field(..., description='响应消息')
|
||||
"""响应消息"""
|
||||
data: T = Field(..., description='响应数据')
|
||||
"""响应数据"""
|
||||
|
||||
# 定义一个函数,将字符串压缩为 gzip 格式,并进行 base64 编码后返回结果
|
||||
def gzip_str(to_gzip: str) -> str:
|
||||
|
||||
+2
-2
@@ -1,5 +1,5 @@
|
||||
import time
|
||||
from typing import Annotated, Optional
|
||||
from typing import Annotated
|
||||
from redis import Redis
|
||||
from sqlalchemy.orm import Session
|
||||
from fastapi import Depends, Header
|
||||
@@ -16,7 +16,7 @@ SessionDep = Annotated[Session, Depends(get_db)]
|
||||
RedisDep = Annotated[Redis, Depends(get_redis)]
|
||||
|
||||
|
||||
def alterToken(redis:RedisDep,alter_token: Optional[str] = Header(None)):
|
||||
def alterToken(redis:RedisDep,alter_token: str | None = Header(None)):
|
||||
if alter_token is not None:
|
||||
try:
|
||||
decoded_bytes = base64.b64decode(alter_token)
|
||||
|
||||
@@ -2,20 +2,19 @@ from fastapi.params import Param
|
||||
from api.core.Gzip import register_gzip_request
|
||||
import json
|
||||
from fastapi import APIRouter, Path
|
||||
from api.core.Response import response,Return
|
||||
from api.core.Response import response
|
||||
from api.core.Redis import get_redis_info
|
||||
from api.dp import RedisDep
|
||||
import re
|
||||
from typing import TYPE_CHECKING, Annotated
|
||||
from typing import Annotated
|
||||
|
||||
from api.types.open import SkillInfo,SkillDetail
|
||||
if TYPE_CHECKING:
|
||||
from api.core.Response import Return
|
||||
from api.types.open import SkillDetailResponse, SkillsListResponse
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
register_gzip_request(router)
|
||||
|
||||
|
||||
def replace_placeholders(template):
|
||||
# 匹配 <int>、<float>、<float1>、<float2> 等
|
||||
pattern = re.compile(r'<(int|float\d*)>')
|
||||
@@ -29,64 +28,57 @@ def replace_placeholders(template):
|
||||
return pattern.sub(repl, template)
|
||||
|
||||
|
||||
@router.get('/skills/')
|
||||
async def get_skills_list(redis: RedisDep,skillName: str) -> Return[list[SkillInfo]]:
|
||||
@router.get('/skills/', operation_id='skillsList')
|
||||
async def get_skills_list(redis: RedisDep, skillName: Annotated[str, Param(..., description='技能名称,支持模糊查询')])->SkillsListResponse:
|
||||
"""
|
||||
根据技能名称,获取对应的技能列表信息
|
||||
"""
|
||||
key = 'openapi:skills'
|
||||
|
||||
def get_skills():
|
||||
# This function should retrieve the skills list from the data source
|
||||
# For now, we return a placeholder list
|
||||
with open('./openapi/skills.json', encoding='utf-8') as f:
|
||||
skills_data = json.load(f)
|
||||
return skills_data
|
||||
|
||||
skills_list = get_redis_info(redis, key, get_skills)
|
||||
data = list(
|
||||
filter(
|
||||
lambda x: re.sub(r'\s+', '', skillName) == re.sub(r'\s+', '', x["skillName"]),
|
||||
skills_list
|
||||
)
|
||||
) if skillName else skills_list
|
||||
data = list(filter(lambda x: re.sub(r'\s+', '', skillName) in re.sub(r'\s+', '', x['skillName']), skills_list)) if skillName else skills_list
|
||||
return response(data=data)
|
||||
|
||||
|
||||
@router.get('/{job}/{jobGrow}/{skillId}/')
|
||||
async def get_cn_skll_info(redis: RedisDep,
|
||||
job: Annotated[str,Path(..., title='职业')],
|
||||
jobGrow: Annotated[str,Path(..., title='转职')],
|
||||
skillId: Annotated[str,Path(..., title='技能')],
|
||||
level: Annotated[int,Param(..., title="技能等级")]
|
||||
) -> Return[SkillDetail]:
|
||||
@router.get('/{jobId}/{jobGrowId}/{skillId}/', operation_id='skillDetail')
|
||||
async def get_cn_skill_info(redis: RedisDep, jobId: Annotated[str, Path(..., description='职业')], jobGrowId: Annotated[str, Path(..., description='转职')], skillId: Annotated[str, Path(..., description='技能')], level: Annotated[int, Param(..., description='技能等级')]) -> SkillDetailResponse:
|
||||
"""
|
||||
获取技能详细信息
|
||||
"""
|
||||
skill_info = {}
|
||||
key = f'openapi:{job}:{jobGrow}:{skillId}'
|
||||
key = f'openapi:{jobId}:{jobGrowId}:{skillId}'
|
||||
try:
|
||||
|
||||
def get_skill_info():
|
||||
# This function should retrieve the skill info based on job and jobGrow
|
||||
# For now, we return a placeholder dictionary
|
||||
with open(f'./openapi/{job}/{jobGrow}/cn/skillDetail/{skillId}.json', encoding='utf-8') as f:
|
||||
with open(f'./openapi/{jobId}/{jobGrowId}/cn/skillDetail/{skillId}.json', encoding='utf-8') as f:
|
||||
skill_data = json.load(f)
|
||||
if "levelInfo" in skill_data and "optionDesc" in skill_data["levelInfo"] and skill_data["levelInfo"]["optionDesc"] is not None:
|
||||
skill_data["levelInfo"]["optionDesc"] = replace_placeholders(skill_data["levelInfo"]["optionDesc"])
|
||||
if 'levelInfo' in skill_data and 'optionDesc' in skill_data['levelInfo'] and skill_data['levelInfo']['optionDesc'] is not None:
|
||||
skill_data['levelInfo']['optionDesc'] = replace_placeholders(skill_data['levelInfo']['optionDesc'])
|
||||
return skill_data
|
||||
|
||||
skill_info = get_redis_info(redis, key, get_skill_info)
|
||||
if level is not None:
|
||||
level = max(0,min(level,skill_info.get("maxLevel",0)))
|
||||
if "levelInfo" in skill_info and "rows" in skill_info["levelInfo"]:
|
||||
detail = list(filter(lambda x: x["level"] == level, skill_info["levelInfo"]["rows"]))
|
||||
level = max(0, min(level, skill_info.get('maxLevel', 0)))
|
||||
if 'levelInfo' in skill_info and 'rows' in skill_info['levelInfo']:
|
||||
detail = list(filter(lambda x: x['level'] == level, skill_info['levelInfo']['rows']))
|
||||
if detail and len(detail) > 0:
|
||||
skill_info["levelInfo"].update(detail[0])
|
||||
skill_info["levelInfo"]["detail"] = skill_info["levelInfo"].get("optionDesc", "")
|
||||
for key in skill_info["levelInfo"]["optionValue"].keys():
|
||||
skill_info["levelInfo"]["detail"] = skill_info["levelInfo"]["detail"].replace(f"{{{key}}}",str(skill_info["levelInfo"]["optionValue"][key]))
|
||||
del skill_info["levelInfo"]["rows"]
|
||||
skill_info["attribute"] = {}
|
||||
skill_info["attribute"].update(skill_info["levelInfo"])
|
||||
del skill_info["levelInfo"]
|
||||
skill_info['levelInfo'].update(detail[0])
|
||||
skill_info['levelInfo']['detail'] = skill_info['levelInfo'].get('optionDesc', '')
|
||||
for key in skill_info['levelInfo']['optionValue'].keys():
|
||||
skill_info['levelInfo']['detail'] = skill_info['levelInfo']['detail'].replace(f'{{{key}}}', str(skill_info['levelInfo']['optionValue'][key]))
|
||||
del skill_info['levelInfo']['rows']
|
||||
skill_info['attribute'] = {}
|
||||
skill_info['attribute'].update(skill_info['levelInfo'])
|
||||
del skill_info['levelInfo']
|
||||
except FileNotFoundError:
|
||||
return response(code=404, message=f"技能信息未找到: {job} {jobGrow} {skillId}", data=None)
|
||||
return response(code=404, message=f'技能信息未找到: {jobId} {jobGrowId} {skillId}', data=None)
|
||||
return response(data=skill_info)
|
||||
|
||||
+46
-35
@@ -1,16 +1,22 @@
|
||||
from enum import Enum
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from typing import Generic, TypeVar
|
||||
from pydantic.generics import GenericModel
|
||||
from api.core.Response import Return as Response
|
||||
|
||||
class SkillInfo(BaseModel):
|
||||
skillId: str = Field(..., description='技能ID')
|
||||
"""技能ID"""
|
||||
skillName: str = Field(..., description='技能名称')
|
||||
"""技能名称"""
|
||||
job: str = Field(..., description='职业')
|
||||
"""职业"""
|
||||
jobGrow: str = Field(..., description='转职')
|
||||
"""转职"""
|
||||
jobId: str = Field(..., description='职业Id')
|
||||
"""职业Id"""
|
||||
jobName: str = Field(..., description='职业名称')
|
||||
"""职业名称"""
|
||||
jobGrowId: str = Field(..., description='转职Id')
|
||||
"""转职Id"""
|
||||
jobGrowName: str = Field(..., description='转职名称')
|
||||
"""转职名称"""
|
||||
|
||||
|
||||
class SkillType(str, Enum):
|
||||
@@ -33,36 +39,38 @@ class IConsumeItem(BaseModel):
|
||||
"""物品ID"""
|
||||
itemName: str = Field(..., description='物品名称')
|
||||
"""物品名称"""
|
||||
value: str = Field(..., description='物品类型')
|
||||
value: int = Field(..., description='物品类型')
|
||||
"""物品类型"""
|
||||
|
||||
|
||||
class IPreSkill(BaseModel):
|
||||
level: str = Field(..., description='前置技能等级')
|
||||
level: int = Field(..., description='前置技能等级')
|
||||
"""前置技能等级"""
|
||||
name: str = Field(..., description='前置技能名称')
|
||||
"""前置技能名称"""
|
||||
skillId: int = Field(..., description='前置技能ID')
|
||||
skillId: str = Field(..., description='前置技能ID')
|
||||
"""前置技能ID"""
|
||||
|
||||
|
||||
class IAttribute(BaseModel):
|
||||
castingTime: float | None = Field(..., description='施放时间')
|
||||
castingTime: float | None = Field(None, description='施放时间')
|
||||
"""施放时间"""
|
||||
coolTime: float | None = Field(..., description='冷却时间')
|
||||
coolTime: float | None = Field(None, description='冷却时间')
|
||||
"""冷却时间"""
|
||||
consumeMp: int | None = Field(..., description='魔法值')
|
||||
consumeMp: int | None = Field(None, description='魔法值')
|
||||
"""魔法值"""
|
||||
detail: str | None = Field(..., description='技能描述')
|
||||
detail: str | None = Field(None, description='技能描述')
|
||||
"""技能描述"""
|
||||
level: int | None = Field(..., description='技能等级')
|
||||
level: int | None = Field(None, description='技能等级')
|
||||
"""技能等级"""
|
||||
optionDesc: str | None = Field(..., description='技能选项描述')
|
||||
optionDesc: str | None = Field(None, description='技能选项描述')
|
||||
"""技能选项描述"""
|
||||
optionValue: dict[str, float] | None = Field(..., description='技能选项值')
|
||||
optionValue: dict[str, float | str] | None = Field(None, description='技能选项值')
|
||||
|
||||
|
||||
class ILevelInfo(BaseModel):
|
||||
optionDesc: str | None = Field(..., description='技能选项描述')
|
||||
rows: list[IAttribute] | None = Field(..., description='技能等级信息')
|
||||
optionDesc: str | None = Field(None, description='技能选项描述')
|
||||
rows: list[IAttribute] | None = Field(None, description='技能等级信息')
|
||||
|
||||
|
||||
class IEvolution(BaseModel):
|
||||
@@ -70,23 +78,23 @@ class IEvolution(BaseModel):
|
||||
"""进化类型"""
|
||||
name: str = Field(..., description='进化名称')
|
||||
"""进化名称"""
|
||||
desc: str | None = Field(..., description='进化简介')
|
||||
desc: str = Field(..., description='进化简介')
|
||||
"""进化简介"""
|
||||
descDetail: str | None = Field(..., description='进化详细描述')
|
||||
descDetail: str = Field(..., description='进化详细描述')
|
||||
"""进化详细描述"""
|
||||
|
||||
|
||||
class IKeyValues(BaseModel):
|
||||
name: str = Field(..., description='名称')
|
||||
"""名称"""
|
||||
value: str | None = Field(..., description='数值')
|
||||
value: str = Field(..., description='数值')
|
||||
"""数值"""
|
||||
|
||||
|
||||
class IEnhancement(BaseModel):
|
||||
type: str = Field(..., description='强化类型')
|
||||
type: int = Field(None, description='强化类型')
|
||||
"""强化类型"""
|
||||
status: list[IKeyValues] | None = Field(..., description='强化状态')
|
||||
status: list[IKeyValues] = Field(None, description='强化状态')
|
||||
|
||||
|
||||
class SkillDetail(BaseModel):
|
||||
@@ -94,31 +102,34 @@ class SkillDetail(BaseModel):
|
||||
"""技能名称"""
|
||||
type: SkillType = Field(..., description='技能类型')
|
||||
"""技能类型"""
|
||||
desc: str | None = Field(..., description='技能描述')
|
||||
desc: str | None = Field(None, description='技能描述')
|
||||
"""技能描述"""
|
||||
descDetail: str | None = Field(..., description='技能详细描述')
|
||||
descDetail: str | None = Field(None, description='技能详细描述')
|
||||
"""技能详细描述"""
|
||||
consumeItem: IConsumeItem | None = Field(..., description='消耗物品')
|
||||
consumeItem: IConsumeItem | None = Field(None, description='消耗物品')
|
||||
"""消耗物品"""
|
||||
descSpecial: str | None = Field(..., description='特殊功能')
|
||||
descSpecial: list[str] | None = Field(None, description='特殊功能')
|
||||
"""特殊功能"""
|
||||
maxLevel: int = Field(..., description='等级上限')
|
||||
"""等级上限"""
|
||||
requiredLevel: int = Field(1, description='学习等级')
|
||||
requiredLevel: int = Field(..., description='学习等级')
|
||||
"""学习等级"""
|
||||
requiredLevelRange: str | None = Field(..., description='学习间隔')
|
||||
requiredLevelRange: int = Field(..., description='学习间隔')
|
||||
"""学习间隔"""
|
||||
preRequiredSkill: list[IPreSkill] | None = Field(..., description='前置技能')
|
||||
preRequiredSkill: list[IPreSkill] | None = Field(None, description='前置技能')
|
||||
"""前置技能"""
|
||||
jobId: str | None = Field(..., description='职业ID')
|
||||
jobId: str = Field(..., description='职业ID')
|
||||
"""职业ID"""
|
||||
jobName: str | None = Field(..., description='职业名称')
|
||||
jobName: str = Field(..., description='职业名称')
|
||||
"""职业名称"""
|
||||
levelInfo: list[ILevelInfo] | None = Field(..., description='技能等级信息')
|
||||
levelInfo: list[ILevelInfo] | None = Field(None, description='技能等级信息')
|
||||
"""技能等级信息"""
|
||||
attribute: IAttribute | None = Field(..., description='技能属性')
|
||||
attribute: IAttribute | None = Field(None, description='技能属性')
|
||||
"""技能属性"""
|
||||
evolution: list[IEvolution] | None = Field(..., description='技能进化信息')
|
||||
evolution: list[IEvolution] | None = Field(None, description='技能进化信息')
|
||||
"""技能进化信息"""
|
||||
enhancement: list[IEnhancement] | None = Field(..., description='技能强化信息')
|
||||
enhancement: list[IEnhancement] | None = Field(None, description='技能强化信息')
|
||||
"""技能强化信息"""
|
||||
|
||||
SkillDetailResponse = Response[SkillDetail]
|
||||
SkillsListResponse = Response[list[SkillInfo]]
|
||||
|
||||
Reference in New Issue
Block a user