This commit was merged in pull request #1.
This commit is contained in:
2026-06-02 20:56:07 +08:00
parent f824177aa5
commit bb3ad71d21
38 changed files with 1304 additions and 532 deletions
+3 -5
View File
@@ -1,9 +1,8 @@
import base64
import hmac
import json
import time
from core.basic.character import Character
from fastapi import Header
from fastapi import Header
from api.core.exception import ResponseException
@@ -36,7 +35,7 @@ def dict2obj(dictObj):
return d
def createToken(alter: str, equVersion: str, redis = None, expire=86400 / 24 * 2)->str:
def createToken(alter: str, equVersion: str, redis=None, expire=86400 / 24 * 2) -> str:
payload = json.dumps({"alter": alter, "equVersion": equVersion, "time": int(time.time() * 1000)})
token = base64.urlsafe_b64encode(payload.encode('utf-8')).decode('utf-8')
# character = createCharacter(alter)
@@ -51,7 +50,7 @@ def readToken(token: str, redis) -> AlterState:
info = dict2obj(json.loads(redis.get(f'token:{token}')))
except Exception as ex:
print(ex)
ResponseException('登录过期或无效Token,请刷新后重试')
raise ResponseException('登录过期或无效Token,请刷新后重试') from ex
if info is None:
raise ResponseException('登录过期或无效Token,请刷新后重试')
return info
@@ -67,4 +66,3 @@ def uid(access_uid: str | None = Header(None)):
return int(access_uid)
except Exception:
return -1
+1 -1
View File
@@ -47,4 +47,4 @@ def get_redis_info(redis, key, fun, expire = 0, without_gzip=False):
redis.set(key, gzip_zip(json.dumps(info)))
if expire > 0:
redis.expire(key, expire)
return info
return info
+4 -4
View File
@@ -6,7 +6,7 @@ from api.routes import openapi
from api.routes import colg
api_router = APIRouter()
api_router.include_router(info.router,prefix="/api",tags=["info"])
api_router.include_router(calc.router,prefix="/api",tags=["calc"])
api_router.include_router(openapi.router,prefix="/api/open",tags=["openapi"])
api_router.include_router(colg.router,prefix="/api/open",tags=["colg"])
api_router.include_router(info.router, prefix="/api", tags=["info"])
api_router.include_router(calc.router, prefix="/api", tags=["calc"])
api_router.include_router(openapi.router, prefix="/api/open", tags=["openapi"])
api_router.include_router(colg.router, prefix="/api/open", tags=["colg"])
+7 -9
View File
@@ -1,17 +1,15 @@
from api.core.Auth import createToken
from api.core.Gzip import register_gzip_request
import json
from fastapi import APIRouter, Body,Request
from api.core.Response import Return, response
from api.core.Redis import get_redis_info
from api.dp import RedisDep,AltersDep
from fastapi import APIRouter, Body
from api.core.Response import response
from api.dp import AltersDep
from core.basic.character import createCharacter
router = APIRouter()
register_gzip_request(router)
@router.post('/calc')
async def calc(request: Request,state:AltersDep,config=Body(None)):
character = createCharacter(state.alter, state.equVersion)
return response(data=character.calc(config))
async def calc(state: AltersDep, config=Body(None)):
character = createCharacter(state.alter, state.equVersion)
return response(data=character.calc(config))
+1 -1
View File
@@ -65,7 +65,7 @@ async def get_real_time_gold(redis: RedisDep) -> Response[list[RealTimeGold]]:
resp = response.json().get('data', {})
res.append({'updataTime': get_date_str(resp['x'][-1]), 'region': region_name, 'goldPrice': resp['y'][-1]})
return res
except Exception as e:
except Exception:
# print(f"File not found: {e}")
return response(data=[])
# 设置半小时过期一次
+6 -36
View File
@@ -1,11 +1,8 @@
import asyncio
import re
from api.core.Auth import createToken
from api.core.Gzip import register_gzip_request
import json
from fastapi import APIRouter, Request, Depends, Body
from typing import Annotated
from api.core.Response import Return, response
from fastapi import APIRouter, Body, Request
from api.core.Auth import createToken
from api.core.Response import response
from api.core.Redis import get_redis_info, get_redis_info_async
from api.dp import RedisDep, AltersDep
from core.basic.equipment import get_equipment
@@ -14,25 +11,13 @@ from core.basic.character import createCharacter
from core.open.helper import get_user_info_by_id, get_user_info_by_role_id, get_user_list_by_id, getRoleJobCode
from core.open.skillTree import get_skill_tree_info
from api.core.exception import ResponseException
from api.utils.skill_detail import normalize_skill_detail, skill_detail_for_level
router = APIRouter()
register_gzip_request(router)
def replace_placeholders(template):
# 匹配 <int>、<float>、<float1>、<float2> 等
pattern = re.compile(r'<(int|float\d*)>')
idx = 0
def repl(match):
nonlocal idx
idx += 1
return f'{{value{idx}}}'
return pattern.sub(repl, template)
@router.get('/adventure')
async def get_adventure_info(request: Request, redis: RedisDep, name: str = None):
adventure_info = get_redis_info(redis, 'dcalc:adventure', get_adv_list)
@@ -95,25 +80,10 @@ async def get_skill_info(state: AltersDep, skillId: str, level: int, redis: Redi
# For now, we return a placeholder dictionary
with open(f'./openapi/data/{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'])
return skill_data
return normalize_skill_detail(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']))
if detail and len(detail) > 0:
skill_info['levelInfo'].update(detail[0])
skill_info['levelInfo']['detail'] = skill_info['levelInfo'].get('optionDesc', '')
if skill_info['levelInfo']['optionValue'] is not None:
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 = skill_detail_for_level(skill_info, level)
except FileNotFoundError:
return response(code=404, message=f'技能信息未找到: {jobId} {jobGrowId} {skillId}', data=None)
return response(data=skill_info)
+3 -32
View File
@@ -4,6 +4,7 @@ from fastapi import APIRouter, Path, Query
from api.core.Response import response
from api.core.Redis import get_redis_info
from api.dp import RedisDep
from api.utils.skill_detail import normalize_skill_detail, skill_detail_for_level
import re
from typing import Annotated
@@ -23,19 +24,6 @@ router = APIRouter()
register_gzip_request(router)
def replace_placeholders(template):
# 匹配 <int>、<float>、<float1>、<float2> 等
pattern = re.compile(r'<(int|float\d*)>')
idx = 0
def repl(match):
nonlocal idx
idx += 1
return f'{{value{idx}}}'
return pattern.sub(repl, template)
@router.get('/skills/', operation_id='skillsList', response_model=SearchSkillResponse)
async def get_skills_list(
redis: RedisDep,
@@ -151,27 +139,10 @@ async def get_cn_skill_info(
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'])
return skill_data
return normalize_skill_detail(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']))
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 = skill_detail_for_level(skill_info, level)
except FileNotFoundError:
return response(
code=404,
-3
View File
@@ -1,7 +1,4 @@
from enum import Enum
from pydantic import BaseModel, Field
from typing import Generic, TypeVar
from api.core.Response import Return as Response
class RealTimeGold(BaseModel):
+60
View File
@@ -0,0 +1,60 @@
from copy import deepcopy
import re
from typing import Any
PLACEHOLDER_PATTERN = re.compile(r'<(?:int|float\d*)[^>]*>')
VALUE_PATTERN = re.compile(r'\{value\d+\}')
def replace_placeholders(template: str) -> str:
idx = 0
def repl(_: re.Match[str]) -> str:
nonlocal idx
idx += 1
return f'{{value{idx}}}'
return PLACEHOLDER_PATTERN.sub(repl, template)
def normalize_skill_detail(skill_data: dict[str, Any]) -> dict[str, Any]:
level_info = skill_data.get('levelInfo')
if not isinstance(level_info, dict):
return skill_data
option_desc = level_info.get('optionDesc')
if option_desc is not None:
level_info['optionDesc'] = replace_placeholders(option_desc)
return skill_data
def skill_detail_for_level(skill_data: dict[str, Any], level: int | None) -> dict[str, Any]:
result = deepcopy(skill_data)
if level is None:
return result
level = max(0, min(level, result.get('maxLevel', 0)))
level_info = result.get('levelInfo')
if not isinstance(level_info, dict):
return result
rows = level_info.get('rows')
if not isinstance(rows, list):
return result
detail = next((row for row in rows if row.get('level') == level), None)
if not detail:
return result
attribute = {**level_info, **detail}
attribute['detail'] = attribute.get('optionDesc', '')
for key, value in (attribute.get('optionValue') or {}).items():
attribute['detail'] = attribute['detail'].replace(f'{{{key}}}', str(value))
attribute['detail'] = VALUE_PATTERN.sub('-', attribute['detail'])
attribute.pop('rows', None)
result['attribute'] = attribute
result.pop('levelInfo', None)
return result