diff --git a/gsuid_core/data_store.py b/gsuid_core/data_store.py index 8357e9b..3d6512d 100644 --- a/gsuid_core/data_store.py +++ b/gsuid_core/data_store.py @@ -20,3 +20,4 @@ def get_res_path(_path: Optional[Union[str, List]] = None) -> Path: image_res = get_res_path('IMAGE_TEMP') +data_cache_path = get_res_path('DATA_CACHE_PATH') diff --git a/gsuid_core/utils/api/ambr/models.py b/gsuid_core/utils/api/ambr/models.py index 3e8f18d..f6ec9e2 100644 --- a/gsuid_core/utils/api/ambr/models.py +++ b/gsuid_core/utils/api/ambr/models.py @@ -276,7 +276,6 @@ class AmbrDungeon(TypedDict): class AmbrDaily(TypedDict): monday: List[AmbrDungeon] tuesday: List[AmbrDungeon] - tuesday: List[AmbrDungeon] thursday: List[AmbrDungeon] friday: List[AmbrDungeon] saturday: List[AmbrDungeon] diff --git a/gsuid_core/utils/api/ambr/request.py b/gsuid_core/utils/api/ambr/request.py index 2dbd6c3..e59436a 100644 --- a/gsuid_core/utils/api/ambr/request.py +++ b/gsuid_core/utils/api/ambr/request.py @@ -13,7 +13,7 @@ from PIL import Image from httpx import AsyncClient from ..types import AnyDict -from ..utils import _HEADER +from ..utils import _HEADER, cache_data from .models import ( AmbrBook, AmbrDaily, @@ -51,7 +51,11 @@ async def get_ambr_event_info() -> Optional[Dict[str, AmbrEvent]]: return None -async def get_ambr_char_data(id: Union[int, str]) -> Optional[AmbrCharacter]: +@cache_data +async def get_ambr_char_data( + id: Union[int, str], + cache_path: Optional[Path] = None, +) -> Optional[AmbrCharacter]: data = await _ambr_request(url=AMBR_CHAR_URL.format(id)) if isinstance(data, Dict) and data['response'] == 200: data = data['data'] @@ -59,7 +63,11 @@ async def get_ambr_char_data(id: Union[int, str]) -> Optional[AmbrCharacter]: return None -async def get_ambr_monster_data(id: Union[int, str]) -> Optional[AmbrMonster]: +@cache_data +async def get_ambr_monster_data( + id: Union[int, str], + cache_path: Optional[Path] = None, +) -> Optional[AmbrMonster]: data = await _ambr_request(url=AMBR_MONSTER_URL.format(id)) if isinstance(data, Dict) and data['response'] == 200: data = data['data'] @@ -67,7 +75,11 @@ async def get_ambr_monster_data(id: Union[int, str]) -> Optional[AmbrMonster]: return None -async def get_ambr_gcg_detail(id: Union[int, str]) -> Optional[AmbrGCGDetail]: +@cache_data +async def get_ambr_gcg_detail( + id: Union[int, str], + cache_path: Optional[Path] = None, +) -> Optional[AmbrGCGDetail]: data = await _ambr_request(url=AMBR_GCG_DETAIL.format(id)) if isinstance(data, Dict) and data['response'] == 200: data = data['data'] @@ -91,7 +103,11 @@ async def get_ambr_monster_list() -> Optional[AmbrMonsterList]: return None -async def get_ambr_weapon_data(id: Union[int, str]) -> Optional[AmbrWeapon]: +@cache_data +async def get_ambr_weapon_data( + id: Union[int, str], + cache_path: Optional[Path] = None, +) -> Optional[AmbrWeapon]: data = await _ambr_request(url=AMBR_WEAPON_URL.format(id)) if isinstance(data, Dict) and data['response'] == 200: data = data['data'] diff --git a/gsuid_core/utils/api/minigg/request.py b/gsuid_core/utils/api/minigg/request.py index ed1081f..8f4ca48 100644 --- a/gsuid_core/utils/api/minigg/request.py +++ b/gsuid_core/utils/api/minigg/request.py @@ -9,11 +9,13 @@ from __future__ import annotations import json import warnings from enum import Enum +from pathlib import Path from typing import Any, Dict, List, Union, Literal, Optional, cast, overload from httpx import AsyncClient from ..types import AnyDict +from ..utils import cache_data from .exception import MiniggNotFoundError from .models import ( Food, @@ -182,10 +184,12 @@ async def minigg_request( return data +@cache_data async def get_weapon_info( name: str, query_languages: APILanguages = APILanguages.CHS, result_languages: APILanguages = APILanguages.CHS, + cache_path: Optional[Path] = None, ) -> Union[Weapon, List[str], int]: '''获取武器信息 @@ -284,10 +288,12 @@ async def get_weapon_stats( return data +@cache_data async def get_character_info( name: str, query_languages: APILanguages = APILanguages.CHS, result_languages: APILanguages = APILanguages.CHS, + cache_path: Optional[Path] = None, ) -> Union[Character, List[str], int]: data = await minigg_request( '/characters', @@ -302,10 +308,12 @@ async def get_character_info( return data +@cache_data async def get_character_costs( name: str, query_languages: APILanguages = APILanguages.CHS, result_languages: APILanguages = APILanguages.CHS, + cache_path: Optional[Path] = None, ) -> Union[Costs, int]: data = await minigg_request( '/characters', @@ -370,10 +378,12 @@ async def get_constellation_info( return -1 +@cache_data async def get_talent_info( name: str, query_languages: APILanguages = APILanguages.CHS, result_languages: APILanguages = APILanguages.CHS, + cache_path: Optional[Path] = None, ) -> Union[CharacterTalents, int]: data = await minigg_request( '/talents', diff --git a/gsuid_core/utils/api/utils.py b/gsuid_core/utils/api/utils.py index 3c6e170..851c71a 100644 --- a/gsuid_core/utils/api/utils.py +++ b/gsuid_core/utils/api/utils.py @@ -1,3 +1,46 @@ -from gsuid_core.version import __version__ +import json +from pathlib import Path +from typing import TypeVar, Callable, Optional, Awaitable +import aiofiles + +from gsuid_core.version import __version__ +from gsuid_core.data_store import data_cache_path + +T = TypeVar('T') _HEADER = {'User-Agent': f'gsuid-utils/{__version__}'} + + +def cache_data(func: Callable[..., Awaitable[T]]): + async def wrapper(*args, **kwargs) -> Optional[T]: + id = ( + kwargs.get( + 'id', kwargs.get('name', args[0] if args else func.__name__) + ) + or func.__name__ + ) + cache_dir: Path = ( + kwargs.get('cache_path', data_cache_path) or data_cache_path + ) + cache_path = cache_dir / func.__name__ + cache_path.mkdir(parents=True, exist_ok=True) + + if cache_path and cache_path.exists() and cache_path.is_dir(): + cache_file = cache_path / f'{id}.json' + if cache_file.exists(): + async with aiofiles.open(cache_file, 'r') as file: + data = await file.read() + return json.loads(data) # 返回已缓存的数据 + + # 如果没有缓存,调用原始函数获取数据 + result = await func(*args, **kwargs) + + # 如果获取了数据,保存到缓存文件 + if cache_path and cache_path.exists() and cache_path.is_dir(): + cache_file = cache_path / f'{id}.json' + async with aiofiles.open(cache_file, 'w') as file: + await file.write(json.dumps(result)) + + return result + + return wrapper