✨ 利用缓存加快资源数据文件获取速度, 锁定依赖

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')

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]

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']

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',

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