✨ Use Redis to Cache EnkaNetwork Data

plugins/genshin/avatar_list.py

@@ -13,6 +13,7 @@ from telegram.ext import CallbackContext, filters
 from telegram.helpers import create_deep_linked_url
 
 from core.base.assets import AssetsService
+from core.base.redisdb import RedisDB
 from core.baseplugin import BasePlugin
 from core.config import config
 from core.cookies.error import CookiesNotFoundError
@@ -25,6 +26,7 @@ from metadata.genshin import AVATAR_DATA, NAMECARD_DATA
 from modules.wiki.base import Model
 from utils.decorators.error import error_callable
 from utils.decorators.restricts import restricts
+from utils.enkanetwork import RedisCache
 from utils.helpers import get_genshin_client
 from utils.log import logger
 from utils.patch.aiohttp import AioHttpTimeoutException
@@ -32,12 +34,17 @@ from utils.patch.aiohttp import AioHttpTimeoutException
 
 class AvatarListPlugin(Plugin, BasePlugin):
     def __init__(
-        self, cookies_service: CookiesService, assets_service: AssetsService, template_service: TemplateService
+        self,
+        cookies_service: CookiesService = None,
+        assets_service: AssetsService = None,
+        template_service: TemplateService = None,
+        redis: RedisDB = None,
     ) -> None:
         self.cookies_service = cookies_service
         self.assets_service = assets_service
         self.template_service = template_service
         self.enka_client = EnkaNetworkAPI(lang="chs", agent=config.enka_network_api_agent)
+        self.enka_client.set_cache(RedisCache(redis.client, key="plugin:avatar_list:enka_network", ttl=60 * 60 * 3))
         self.enka_assets = EnkaAssets(lang="chs")
 
     async def get_user_client(self, user: User, message: Message, context: CallbackContext) -> Optional[Client]:
@@ -224,7 +231,7 @@ class AvatarListPlugin(Plugin, BasePlugin):
         try:
             name_card, avatar, nickname, rarity = await self.get_final_data(client, characters, update)
         except Exception as exc:
-            logger.error("卡片信息请求失败", exc_info=exc)
+            logger.error("卡片信息请求失败 %s", str(exc))
             name_card, avatar, nickname, rarity = await self.get_default_final_data(characters, update)
 
         render_data = {

plugins/genshin/player_cards.py

@@ -23,6 +23,7 @@ from telegram.ext import CallbackContext, CallbackQueryHandler, CommandHandler,
 from telegram.helpers import create_deep_linked_url
 
 from core.base.assets import DEFAULT_EnkaAssets
+from core.base.redisdb import RedisDB
 from core.baseplugin import BasePlugin
 from core.config import config
 from core.plugin import Plugin, handler
@@ -34,6 +35,7 @@ from modules.playercards.helpers import ArtifactStatsTheory
 from utils.bot import get_args
 from utils.decorators.error import error_callable
 from utils.decorators.restricts import restricts
+from utils.enkanetwork import RedisCache
 from utils.helpers import url_to_file
 from utils.log import logger
 from utils.models.base import RegionEnum
@@ -41,9 +43,12 @@ from utils.patch.aiohttp import AioHttpTimeoutException
 
 
 class PlayerCards(Plugin, BasePlugin):
-    def __init__(self, user_service: UserService = None, template_service: TemplateService = None):
+    def __init__(
+        self, user_service: UserService = None, template_service: TemplateService = None, redis: RedisDB = None
+    ):
         self.user_service = user_service
         self.client = EnkaNetworkAPI(lang="chs", agent=config.enka_network_api_agent)
+        self.client.set_cache(RedisCache(redis.client, key="plugin:player_cards:enka_network"))
         self.template_service = template_service
         self.temp_photo: Optional[str] = None
 

utils/enkanetwork.py

@@ -0,0 +1,34 @@
+import json
+from typing import Dict, Any, Optional
+
+from enkanetwork import Cache
+from redis import asyncio as aioredis
+
+__all__ = ("RedisCache",)
+
+
+class RedisCache(Cache):
+    def __init__(self, redis: aioredis.Redis, key: Optional[str] = None, ttl: int = 60 * 3) -> None:
+        self.redis = redis
+        self.ttl = ttl
+        self.key = key
+        super().__init__(1024, 60 * 3)
+
+    def get_qname(self, key):
+        if self.key:
+            return f"{self.key}:{key}"
+        else:
+            return f"enka_network:{key}"
+
+    async def get(self, key) -> Optional[Dict[str, Any]]:
+        qname = self.get_qname(key)
+        data = await self.redis.get(qname)
+        if data:
+            json_data = str(data, encoding="utf-8")
+            return json.loads(json_data)
+        return None
+
+    async def set(self, key, value) -> None:
+        qname = self.get_qname(key)
+        data = json.dumps(value)
+        await self.redis.set(qname, data, ex=self.ttl)