✨ Support File Cache Player Cards

modules/pay_log/log.py

@@ -67,7 +67,7 @@ class PayLog:
             return PayLogModel(info=BaseInfo(uid=uid), list=[]), False
         try:
             return PayLogModel.parse_obj(await self.load_json(file_path)), True
-        except jsonlib.decoder.JSONDecodeError:
+        except jsonlib.JSONDecodeError:
             return PayLogModel(info=BaseInfo(uid=uid), list=[]), False
 
     async def remove_history_info(

modules/playercards/file.py

@@ -0,0 +1,66 @@
+from pathlib import Path
+from typing import Optional, Dict, Union
+
+import aiofiles
+
+from utils.const import PROJECT_ROOT
+
+import ujson as jsonlib
+
+PLAYER_CARDS_PATH = PROJECT_ROOT.joinpath("data", "apihelper", "player_cards")
+PLAYER_CARDS_PATH.mkdir(parents=True, exist_ok=True)
+
+
+class PlayerCardsFile:
+    def __init__(self, player_cards_path: Path = PLAYER_CARDS_PATH):
+        self.player_cards_path = player_cards_path
+
+    @staticmethod
+    async def load_json(path):
+        async with aiofiles.open(path, "r", encoding="utf-8") as f:
+            return jsonlib.loads(await f.read())
+
+    @staticmethod
+    async def save_json(path, data: Dict):
+        async with aiofiles.open(path, "w", encoding="utf-8") as f:
+            return await f.write(jsonlib.dumps(data, ensure_ascii=False, indent=4))
+
+    def get_file_path(self, uid: Union[str, int]):
+        """获取文件路径
+        :param uid: UID
+        :return: 文件路径
+        """
+        return self.player_cards_path / f"{uid}.json"
+
+    async def load_history_info(
+        self,
+        uid: Union[str, int],
+    ) -> Optional[Dict]:
+        """读取历史记录数据
+        :param uid: uid
+        :return: 角色历史记录数据
+        """
+        file_path = self.get_file_path(uid)
+        if not file_path.exists():
+            return None
+        try:
+            return await self.load_json(file_path)
+        except jsonlib.JSONDecodeError:
+            return None
+
+    async def merge_info(
+        self,
+        uid: Union[str, int],
+        data: Dict,
+    ) -> Dict:
+        old_data = await self.load_history_info(uid)
+        if old_data is None:
+            await self.save_json(self.get_file_path(uid), data)
+            return data
+        data["avatarInfoList"] = data.get("avatarInfoList", [])
+        characters = [i.get("avatarId", 0) for i in data["avatarInfoList"]]
+        for i in old_data.get("avatarInfoList", []):
+            if i.get("avatarId", 0) not in characters:
+                data["avatarInfoList"].append(i)
+        await self.save_json(self.get_file_path(uid), data)
+        return data

plugins/genshin/player_cards.py

@@ -1,5 +1,6 @@
 from typing import Any, List, Tuple, Union, Optional
 
+import ujson
 from enkanetwork import (
     CharacterInfo,
     DigitType,
@@ -31,6 +32,7 @@ from core.template import TemplateService
 from core.user import UserService
 from core.user.error import UserNotFoundError
 from metadata.shortname import roleToName
+from modules.playercards.file import PlayerCardsFile
 from modules.playercards.helpers import ArtifactStatsTheory
 from utils.bot import get_args
 from utils.decorators.error import error_callable
@@ -47,24 +49,38 @@ class PlayerCards(Plugin, BasePlugin):
         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 = EnkaNetworkAPI(lang="chs", agent=config.enka_network_api_agent, cache=False)
-        self.client.set_cache(RedisCache(redis.client, key="plugin:player_cards:enka_network"))
+        self.cache = RedisCache(redis.client, key="plugin:player_cards:enka_network")
+        self.player_cards_file = PlayerCardsFile()
         self.template_service = template_service
         self.temp_photo: Optional[str] = None
 
     async def _fetch_user(self, uid) -> Union[EnkaNetworkResponse, str]:
         try:
-            return await self.client.fetch_user(uid)
+            data = await self.cache.get(uid)
+            if data is not None:
+                return EnkaNetworkResponse.parse_obj(data)
+            user = await self.client.http.fetch_user(uid)
+            data = user["content"].decode("utf-8", "surrogatepass")  # type: ignore
+            data = ujson.loads(data)
+            data = await self.player_cards_file.merge_info(uid, data)
+            await self.cache.set(uid, data)
+            return EnkaNetworkResponse.parse_obj(data)
         except EnkaServerError:
-            return "Enka.Network 服务请求错误，请稍后重试"
+            error = "Enka.Network 服务请求错误，请稍后重试"
         except Forbidden:
-            return "Enka.Network 服务请求被拒绝，请稍后重试"
+            error = "Enka.Network 服务请求被拒绝，请稍后重试"
         except AioHttpTimeoutException:
-            return "Enka.Network 服务请求超时，请稍后重试"
+            error = "Enka.Network 服务请求超时，请稍后重试"
         except HTTPException:
-            return "Enka.Network HTTP 服务请求错误，请稍后重试"
+            error = "Enka.Network HTTP 服务请求错误，请稍后重试"
         except (UIDNotFounded, VaildateUIDError):
-            return "UID 未找到，可能为服务器抽风，请稍后重试"
+            error = "UID 未找到，可能为服务器抽风，请稍后重试"
+        old_data = await self.player_cards_file.load_history_info(uid)
+        if old_data is not None:
+            logger.warning("UID %s | 角色卡片使用历史数据 | %s", uid, error)
+            return EnkaNetworkResponse.parse_obj(old_data)
+        return error
 
     @handler(CommandHandler, command="player_card", block=False)
     @handler(MessageHandler, filters=filters.Regex("^角色卡片查询(.*)"), block=False)
@@ -105,20 +121,15 @@ class PlayerCards(Plugin, BasePlugin):
             logger.info(f"用户 {user.full_name}[{user.id}] 角色卡片查询命令请求 || character_name[{character_name}] uid[{uid}]")
         else:
             logger.info(f"用户 {user.full_name}[{user.id}] 角色卡片查询命令请求")
-            buttons = []
+            buttons = [
-            temp = []
+                InlineKeyboardButton(
-            for index, value in enumerate(data.characters):
+                    value.name,
-                temp.append(
+                    callback_data=f"get_player_card|{user.id}|{uid}|{value.name}",
-                    InlineKeyboardButton(
-                        value.name,
-                        callback_data=f"get_player_card|{user.id}|{uid}|{value.name}",
-                    )
                 )
-                if index == 3:
+                for value in data.characters
-                    buttons.append(temp)
+                if value.name
-                    temp = []
+            ]
-            if len(temp) > 0:
+            buttons = [buttons[i : i + 4] for i in range(0, len(buttons), 4)]
-                buttons.append(temp)
             if isinstance(self.temp_photo, str):
                 photo = self.temp_photo
             else:

plugins/system/get_chat.py

@@ -1,6 +1,8 @@
 import contextlib
 import html
-from typing import List
+import os.path
+from datetime import datetime
+from typing import Tuple
 
 from telegram import Update, Chat, ChatMember, ChatMemberOwner, ChatMemberAdministrator
 from telegram.error import BadRequest, Forbidden
@@ -12,8 +14,10 @@ from core.plugin import Plugin, handler
 from core.sign import SignServices
 from core.user import UserService
 from core.user.error import UserNotFoundError
+from core.user.models import User
 from modules.gacha_log.log import GachaLog
 from modules.pay_log.log import PayLog
+from modules.playercards.file import PlayerCardsFile
 from utils.bot import get_args, get_chat as get_chat_with_cache
 from utils.decorators.admins import bot_admins_rights_check
 from utils.helpers import get_genshin_client
@@ -33,8 +37,9 @@ class GetChat(Plugin):
         self.sign_service = sign_service
         self.gacha_log = GachaLog()
         self.pay_log = PayLog()
+        self.player_cards_file = PlayerCardsFile()
 
-    async def parse_group_chat(self, chat: Chat, admins: List[ChatMember]) -> str:
+    async def parse_group_chat(self, chat: Chat, admins: Tuple[ChatMember]) -> str:
         text = f"群 ID：<code>{chat.id}</code>\n群名称：<code>{chat.title}</code>\n"
         if chat.username:
             text += f"群用户名：@{chat.username}\n"
@@ -61,6 +66,71 @@ class GetChat(Plugin):
                 text += "\n"
         return text
 
+    @staticmethod
+    async def parse_private_bind(user_info: User, chat_id: int) -> Tuple[str, int]:
+        if user_info.region == RegionEnum.HYPERION:
+            text = "米游社绑定："
+            uid = user_info.yuanshen_uid
+        else:
+            text = "原神绑定："
+            uid = user_info.genshin_uid
+        temp = "Cookie 绑定"
+        try:
+            await get_genshin_client(chat_id)
+        except CookiesNotFoundError:
+            temp = "UID 绑定"
+        return f"{text}<code>{temp}</code>\n游戏 ID：<code>{uid}</code>", uid
+
+    async def parse_private_sign(self, chat_id: int) -> str:
+        sign_info = await self.sign_service.get_by_user_id(chat_id)
+        if sign_info is not None:
+            text = (
+                f"\n自动签到：已开启"
+                f"\n推送会话：<code>{sign_info.chat_id}</code>"
+                f"\n开启时间：<code>{sign_info.time_created}</code>"
+                f"\n更新时间：<code>{sign_info.time_updated}</code>"
+                f"\n签到状态：<code>{sign_info.status.name}</code>"
+            )
+        else:
+            text = "\n自动签到：未开启"
+        return text
+
+    async def parse_private_gacha_log(self, chat_id: int, uid: int) -> str:
+        gacha_log, status = await self.gacha_log.load_history_info(str(chat_id), str(uid))
+        if status:
+            text = "\n抽卡记录："
+            for key, value in gacha_log.item_list.items():
+                text += f"\n   - {key}：{len(value)} 条"
+            text += f"\n   - 最后更新：{gacha_log.update_time.strftime('%Y-%m-%d %H:%M:%S')}"
+        else:
+            text = "\n抽卡记录：<code>未导入</code>"
+        return text
+
+    async def parse_private_pay_log(self, chat_id: int, uid: int) -> str:
+        pay_log, status = await self.pay_log.load_history_info(str(chat_id), str(uid))
+        return (
+            f"\n充值记录：\n   - 已导入 {len(pay_log.list)} 条\n   - 最后更新：{pay_log.info.export_time}"
+            if status
+            else "\n充值记录：<code>未导入</code>"
+        )
+
+    @staticmethod
+    def get_file_modify_time(path: str) -> datetime:
+        return datetime.fromtimestamp(os.path.getmtime(path))
+
+    async def parse_private_player_cards_file(self, uid: int) -> str:
+        player_cards = await self.player_cards_file.load_history_info(uid)
+        if player_cards is None:
+            text = "\n角色卡片：<code>未缓存</code>"
+        else:
+            time = self.get_file_modify_time(self.player_cards_file.get_file_path(uid))
+            text = (
+                f"\n角色卡片："
+                f"\n   - 已缓存 {len(player_cards.get('avatarInfoList', []))} 个角色"
+                f"\n   - 最后更新：{time.strftime('%Y-%m-%d %H:%M:%S')}"
+            )
+        return text
+
     async def parse_private_chat(self, chat: Chat) -> str:
         text = (
             f'<a href="tg://user?id={chat.id}">MENTION</a>\n'
@@ -74,44 +144,15 @@ class GetChat(Plugin):
         except UserNotFoundError:
             user_info = None
         if user_info is not None:
-            if user_info.region == RegionEnum.HYPERION:
+            temp, uid = await self.parse_private_bind(user_info, chat.id)
-                text += "米游社绑定："
+            text += temp
-                uid = user_info.yuanshen_uid
+            text += await self.parse_private_sign(chat.id)
-            else:
-                text += "原神绑定："
-                uid = user_info.genshin_uid
-            temp = "Cookie 绑定"
-            try:
-                await get_genshin_client(chat.id)
-            except CookiesNotFoundError:
-                temp = "UID 绑定"
-            text += f"<code>{temp}</code>\n游戏 ID：<code>{uid}</code>"
-            sign_info = await self.sign_service.get_by_user_id(chat.id)
-            if sign_info is not None:
-                text += (
-                    f"\n自动签到：已开启"
-                    f"\n推送会话：<code>{sign_info.chat_id}</code>"
-                    f"\n开启时间：<code>{sign_info.time_created}</code>"
-                    f"\n更新时间：<code>{sign_info.time_updated}</code>"
-                    f"\n签到状态：<code>{sign_info.status.name}</code>"
-                )
-            else:
-                text += "\n自动签到：未开启"
             with contextlib.suppress(Exception):
-                gacha_log, status = await self.gacha_log.load_history_info(str(chat.id), str(uid))
+                text += await self.parse_private_gacha_log(chat.id, uid)
-                if status:
-                    text += "\n抽卡记录："
-                    for key, value in gacha_log.item_list.items():
-                        text += f"\n   - {key}：{len(value)} 条"
-                    text += f"\n   - 最后更新：{gacha_log.update_time.strftime('%Y-%m-%d %H:%M:%S')}"
-                else:
-                    text += "\n抽卡记录：<code>未导入</code>"
             with contextlib.suppress(Exception):
-                pay_log, status = await self.pay_log.load_history_info(str(chat.id), str(uid))
+                text += await self.parse_private_pay_log(chat.id, uid)
-                if status:
+            with contextlib.suppress(Exception):
-                    text += f"\n充值记录：" f"\n   - {len(pay_log.list)} 条" f"\n   - 最后更新：{pay_log.info.export_time}"
+                text += await self.parse_private_player_cards_file(uid)
-                else:
-                    text += "\n充值记录：<code>未导入</code>"
         return text
 
     @handler(CommandHandler, command="get_chat", block=False)