from api.core.Gzip import register_gzip_request import json 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 SKILL_DETAIL_NORMALIZE_VERSION, normalize_skill_detail, skill_detail_for_level from api.utils.openapi_data import OPENAPI_SKILLS_PATH, OPENAPI_SUMMARY_DIR, openapi_locale_path from api.utils.skill_tree import ( SKILL_TREE_WITH_DESC_SPECIAL_VERSION, filter_skill_tree_by_desc_special, load_skill_tree_with_desc_special, ) import re from typing import Annotated from api.types.open import ( JobInfoResponse, SkillDataSummaryResponse, SkillDetailResponse, SkillsListResponse, SearchSkillResponse, ) import os from core.character.adventure import get_adv_list router = APIRouter() register_gzip_request(router) @router.get('/skills/', operation_id='skillsList', response_model=SearchSkillResponse) async def get_skills_list( redis: RedisDep, skillName: Annotated[str | None, Query(description='技能名称，支持模糊查询')] = None, ) -> SearchSkillResponse: """ 根据技能名称，获取对应的技能列表信息 """ 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 OPENAPI_SKILLS_PATH.open(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) in re.sub(r'\s+', '', x['skillName']), skills_list, ) ) if skillName else skills_list ) return response(data=data) @router.get('/jobInfo/', operation_id='jobsInfo', response_model=JobInfoResponse) async def get_jobs_info( redis: RedisDep, jobName: Annotated[str, Query(..., description='职业名称，支持模糊查询')] = None, ) -> JobInfoResponse: """ 根据职业名称，获取对应的职业转职信息 """ adventure_info = get_redis_info(redis, 'dcalc:adventure', get_adv_list) jobName = jobName.replace('（', '(').replace('）', ')') result = [] for adv in adventure_info: if '(男)' in jobName or '(女)' in jobName: gender = jobName[-3:] toSearch = jobName[:-3] if gender not in adv.get('title', ''): continue matched_children = [child for child in adv.get('children', []) if (toSearch in child.get('title', '') or toSearch in child.get('name', ''))] else: matched_children = [child for child in adv.get('children', []) if jobName in child.get('title', '') or jobName in child.get('name', '')] if matched_children: parent = {k: v for k, v in adv.items() if k != 'children'} for child in matched_children: entry = parent.copy() entry['child'] = child result.append( { 'jobId': entry.get('name', ''), 'jobName': entry.get('title', ''), 'jobGrowId': child.get('name', ''), 'jobGrowName': child.get('title', ''), } ) return response(data=result) @router.get('/{jobId}/{jobGrowId}/skills/', operation_id='skillsByJob', response_model=SkillsListResponse) async def get_cn_skills_by_job( redis: RedisDep, jobId: Annotated[str, Path(..., description='职业')], jobGrowId: Annotated[str, Path(..., description='转职')], descSpecial: Annotated[str, Query(description='技能特殊功能')] = '', ) -> SkillsListResponse: """ 获取职业技能列表，职业ID和转职ID可通过jobsInfo获取 Args: jobId (str): 职业ID，必填参数 jobGrowId (str): 转职ID，必填参数 descSpecial (str, optional): 技能特殊功能，默认为''，可选参数，支持传入以下功能关键词以过滤技能列表：压制、无敌、追踪、聚集敌人、强控敌人 """ key = f'openapi:{SKILL_TREE_WITH_DESC_SPECIAL_VERSION}:{jobId}:{jobGrowId}:skills' def get_skills_by_job(): return load_skill_tree_with_desc_special(jobId, jobGrowId) skills_list = get_redis_info(redis, key, get_skills_by_job) return response(data=filter_skill_tree_by_desc_special(skills_list, descSpecial)) @router.get('/{jobId}/{jobGrowId}/{skillId}/', operation_id='skillDetail', response_model=SkillDetailResponse) 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 | None, Query(description='技能等级')] = None, ) -> SkillDetailResponse: """ 获取技能详细信息 Args: jobId (str): 职业ID，必填参数 jobGrowId (str): 转职ID，必填参数 skillId (str): 技能ID，必填参数 level (int, optional): 技能等级，默认为None """ skill_info = {} key = f'openapi:{SKILL_DETAIL_NORMALIZE_VERSION}:{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( openapi_locale_path('cn', jobId, jobGrowId, 'skillDetail', f'{skillId}.json'), encoding='utf-8', ) as f: skill_data = json.load(f) return normalize_skill_detail(skill_data) skill_info = get_redis_info(redis, key, get_skill_info) 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) @router.get('/skillData/summary/', operation_id='skillDataSummary', response_model=SkillDataSummaryResponse) async def get_skill_data_summary( redis: RedisDep, job: Annotated[str, Query(description='职业')], skills: Annotated[ str, Query( description='技能列表字符串,逗号拼接技能名称,不传递或者传递为空时,查询所有', ), ] = '', weapons: Annotated[ str, Query( description='武器类型列表字符串,逗号拼接武器类型名称,不传递或者传递为空时,查询所有', ), ] = '', ) -> SkillDataSummaryResponse: """ 获取技能数据汇总信息 Args: job (str): 职业名称，必填参数 skills (str, optional): 技能列表字符串，逗号拼接技能名称，不传递或者传递为空时，查询所有技能，默认为'' weapons (str, optional): 武器类型列表字符串，逗号拼接武器类型名称，不传递或者传递为空时，查询所有武器类型，默认为'' Response: { "code": int, "message": str, "data": [ { "技能名称": str, "武器类型": str, "职业": str, "技能描述": str, "技能等级": int, "冷却时间": float, "消耗": str, "伤害": float, # 已学等级 "learnedLevel": int | str, "基础百分比": float | str, "等效百分比": float | str, "原始冷却": float | str, "实际冷却": float | str, "理论秒伤": float | str, "SP": float | None | str, # 所在等级 "unlockAtRoleLevel": str | None, } ] } """ job = job.replace('（', '(').replace('）', ')') key = f'openapi:skillData:summary:{job}' data = [] skills = skills.split(',') if skills else [] weapons = weapons.split(',') if weapons else [] try: def get_skill_data_summary(): # This function should retrieve the skill data summary based on job # For now, we return a placeholder dictionary json_files = [f for f in os.listdir(OPENAPI_SUMMARY_DIR) if f.endswith('.json')] res = [] for i in json_files: if job in i: k = re.search(r'^\d+(?:-\d+)?(.*?)_', i) with (OPENAPI_SUMMARY_DIR / i).open(encoding='utf-8') as f: skill_data = json.load(f) weapon = i.split('_')[-1].replace('.json', '') res.append({'weapon': weapon if weapon else '通用', 'skills': skill_data, 'job': k.group(1) if k else ''}) return res job_summary = get_redis_info(redis, key, get_skill_data_summary) for item in job_summary: if len(weapons) == 0 or item['weapon'] in weapons: items = list(filter(lambda x: x['技能名称'] in skills, item['skills'])) if len(skills) > 0 else item['skills'] for skill in items: skill['武器类型'] = item['weapon'] skill['职业'] = item['job'] data.extend(items or []) except FileNotFoundError as e: print(f'File not found: {e}') return response(data=[]) return response(data=data)