添加缓存预热

backend/api/core/RedisPreheat.py

@@ -0,0 +1,133 @@
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor, as_completed
+import json
+import uuid
+
+import redis
+
+from api.core.Gzip import gzip_zip
+from core.basic.character import createCharacter
+from core.character.adventure import adv, get_adv_list
+from database.connect import RedisPool
+
+
+PreheatLoader = Callable[[], object]
+PreheatTask = tuple[str, PreheatLoader]
+PREHEAT_LOCK_KEY = 'dcalc:preheat:lock'
+PREHEAT_LOCK_EXPIRE_SECONDS = 15 * 60
+
+
+def _load_skills():
+    with open('./openapi/data/skills.json', encoding='utf-8') as file:
+        return json.load(file)
+
+
+def _iter_open_alters() -> list[str]:
+    alters: list[str] = []
+    for parent in adv:
+        for child in parent['children']:
+            alter = child.get('class')
+            if child.get('open') and alter:
+                alters.append(alter)
+    return alters
+
+
+def _build_character_preheat_tasks(equ_version: str = '0') -> list[PreheatTask]:
+    tasks: list[PreheatTask] = []
+    for alter in _iter_open_alters():
+        tasks.append(
+            (
+                f'dcalc:character:{alter}:{equ_version}',
+                lambda alter=alter: createCharacter(alter, equ_version).getInfo(),
+            )
+        )
+    return tasks
+
+
+def _write_preheated_value(redis_conn: redis.Redis, key: str, value: object, expire: int = 0, without_gzip: bool = False):
+    if value is None:
+        return
+    if without_gzip:
+        redis_conn.set(key, json.dumps(value))
+    else:
+        redis_conn.set(key, gzip_zip(json.dumps(value)))
+    if expire > 0:
+        redis_conn.expire(key, expire)
+
+
+def _run_preheat_task(key: str, loader: PreheatLoader) -> tuple[str, object]:
+    return key, loader()
+
+
+def delete_keys_with_prefix(redis_conn: redis.Redis, prefix: str):
+    keys = redis_conn.keys(f'{prefix}*')
+    if keys:
+        redis_conn.delete(*keys)
+
+
+def _acquire_preheat_lock(redis_conn: redis.Redis) -> str | None:
+    lock_token = str(uuid.uuid4())
+    locked = redis_conn.set(PREHEAT_LOCK_KEY, lock_token, nx=True, ex=PREHEAT_LOCK_EXPIRE_SECONDS)
+    if not locked:
+        return None
+    return lock_token
+
+
+def _release_preheat_lock(redis_conn: redis.Redis, lock_token: str):
+    redis_conn.eval(
+        """
+        if redis.call('GET', KEYS[1]) == ARGV[1] then
+            return redis.call('DEL', KEYS[1])
+        end
+        return 0
+        """,
+        1,
+        PREHEAT_LOCK_KEY,
+        lock_token,
+    )
+
+
+def preheat_redis_cache() -> list[str]:
+    # if config.DEBUG_MODE:
+    #     print('Skip redis preheat in debug mode.')
+    #     return []
+    base_preheat_tasks: list[PreheatTask] = [
+        ('dcalc:adventure', get_adv_list),
+        ('openapi:skills', _load_skills),
+    ]
+    character_preheat_tasks = _build_character_preheat_tasks()
+    warmed_keys: list[str] = []
+
+    with redis.Redis(connection_pool=RedisPool) as redis_conn:
+        lock_token = _acquire_preheat_lock(redis_conn)
+        if not lock_token:
+            print('Skip redis preheat: another instance is warming cache.')
+            return []
+
+        try:
+            delete_keys_with_prefix(redis_conn, 'dcalc:')
+            delete_keys_with_prefix(redis_conn, 'openapi:')
+
+            for key, loader in base_preheat_tasks:
+                try:
+                    _write_preheated_value(redis_conn, key, loader())
+                    warmed_keys.append(key)
+                except Exception as exc:
+                    print(f'Redis preheat failed for {key}: {exc}')
+
+            max_workers = min(8, len(character_preheat_tasks))
+            if max_workers > 0:
+                with ThreadPoolExecutor(max_workers=max_workers) as executor:
+                    future_map = {executor.submit(_run_preheat_task, key, loader): key for key, loader in character_preheat_tasks}
+                    for future in as_completed(future_map):
+                        key = future_map[future]
+                        try:
+                            _, value = future.result()
+                            _write_preheated_value(redis_conn, key, value)
+                            warmed_keys.append(key)
+                        except Exception as exc:
+                            print(f'Redis preheat failed for {key}: {exc}')
+        finally:
+            _release_preheat_lock(redis_conn, lock_token)
+
+    return warmed_keys

backend/api/routes/colg.py

@@ -1,12 +1,11 @@
 import asyncio
 
 from api.types.colg import HistoryGold, OfficialGold, RealTimeGold
-from fastapi.params import Param
 from api.core.Gzip import register_gzip_request
 import json
-from fastapi import APIRouter, Path, Query, Depends
+from fastapi import APIRouter, Query
 from api.core.Response import response
-from api.core.Redis import get_redis_info,get_redis_info_async
+from api.core.Redis import get_redis_info, get_redis_info_async
 from api.dp import RedisDep
 from typing import Annotated
 import requests
@@ -47,7 +46,7 @@ def get_date_str(timestamp: int) -> str:
     return beijing_time.strftime('%Y-%m-%d %H:%M:%S')
 
 
-@router.get('/gold/realtime/', operation_id='realtimeGold',response_model=Response[list[RealTimeGold]])
+@router.get('/gold/realtime/', operation_id='realtimeGold', response_model=Response[list[RealTimeGold]])
 async def get_real_time_gold(redis: RedisDep) -> Response[list[RealTimeGold]]:
     """
     返回所有跨区的实时金价信息
@@ -74,7 +73,7 @@ async def get_real_time_gold(redis: RedisDep) -> Response[list[RealTimeGold]]:
     return response(data=data)
 
 
-@router.get('/gold/history/', operation_id='historyGold',response_model=Response[HistoryGold] | Response[None])
+@router.get('/gold/history/', operation_id='historyGold', response_model=Response[HistoryGold] | Response[None])
 async def get_history_gold(region: Annotated[int, Query(..., description='跨区id')], timeRange: Annotated[int, Query(..., description='时间跨度天数')] = 7) -> Response[HistoryGold] | Response[None]:
     """
     获取历史金价数据
@@ -146,7 +145,7 @@ async def get_history_gold(region: Annotated[int, Query(..., description='跨区
     return response(data={'region': region_name, 'data': data, 'echartsData': echartsData})
 
 
-@router.get('/gold/official/', operation_id='officialGold',response_model=Response[list[OfficialGold]])
+@router.get('/gold/official/', operation_id='officialGold', response_model=Response[list[OfficialGold]])
 async def get_official_gold(redis: RedisDep) -> Response[list[OfficialGold]]:
     """
     获取DNF官方金币寄售实时数据
@@ -162,6 +161,7 @@ async def get_official_gold(redis: RedisDep) -> Response[list[OfficialGold]]:
     key = 'openapi:colg:real_time_gold_officical'
     list = ['1', '2', '3A', '3B', '4', '5', '6', '7']
     try:
+
         async def get_data():
             datas = []
             async with aiohttp.ClientSession() as session:

backend/api/routes/info.py

@@ -124,12 +124,12 @@ async def get_dnfhelper_list(uid: str, redis: RedisDep):
 
 @router.get('/dnfhelper/{uid}/')
 async def get_info_by_dnfhelper(uid: str, state: AltersDep, redis: RedisDep):
-    # try:
-    result = await get_user_info_by_id(uid, state)
-    return response(data=result)
-    # except Exception as e:
-    #     if isinstance(e, ResponseException):
-    #         return response(code=500, message=str(e), data=None)
+    try:
+        result = await get_user_info_by_id(uid, state)
+        return response(data=result)
+    except Exception as e:
+        if isinstance(e, ResponseException):
+            return response(code=500, message=str(e), data=None)
     return response(code=500, message='Failed to fetch DNFHelper info', data=None)
 
 

backend/api/routes/openapi.py

@@ -1,7 +1,6 @@
-from fastapi.params import Param
 from api.core.Gzip import register_gzip_request
 import json
-from fastapi import APIRouter, Path, Query, Depends
+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
@@ -40,7 +39,7 @@ def replace_placeholders(template):
 @router.get('/skills/', operation_id='skillsList', response_model=SearchSkillResponse)
 async def get_skills_list(
     redis: RedisDep,
-    skillName: Annotated[str, Param(..., description='技能名称，支持模糊查询')] = None,
+    skillName: Annotated[str | None, Query(description='技能名称，支持模糊查询')] = None,
 ) -> SearchSkillResponse:
     """
     根据技能名称，获取对应的技能列表信息
@@ -129,7 +128,7 @@ async def get_cn_skill_info(
     jobId: Annotated[str, Path(..., description='职业')],
     jobGrowId: Annotated[str, Path(..., description='转职')],
     skillId: Annotated[str, Path(..., description='技能')],
-    level: Annotated[int, Param(..., description='技能等级')] = None,
+    level: Annotated[int | None, Query(description='技能等级')] = None,
 ) -> SkillDetailResponse:
     """
     获取技能详细信息
@@ -185,18 +184,16 @@ async def get_cn_skill_info(
 @router.get('/skillData/summary/', operation_id='skillDataSummary', response_model=SkillDataSummaryResponse)
 async def get_skill_data_summary(
     redis: RedisDep,
-    job: Annotated[str, Param(..., description='职业')],
+    job: Annotated[str, Query(description='职业')],
     skills: Annotated[
         str,
-        Param(
-            ...,
+        Query(
             description='技能列表字符串,逗号拼接技能名称,不传递或者传递为空时,查询所有',
         ),
     ] = '',
     weapons: Annotated[
         str,
-        Param(
-            ...,
+        Query(
             description='武器类型列表字符串,逗号拼接武器类型名称,不传递或者传递为空时,查询所有',
         ),
     ] = '',