import asyncio import random import re from datetime import datetime, timedelta, time from typing import Optional, Tuple, Union, TYPE_CHECKING import genshin from genshin.errors import GenshinException from genshin.models import BaseCharacter from genshin.models import CalculatorCharacterDetails from pydantic import ValidationError from sqlalchemy.exc import SQLAlchemyError from sqlmodel import SQLModel, Field, String, Column, Integer, BigInteger, select, DateTime, func, delete, Index from telegram.ext import ContextTypes from core.basemodel import RegionEnum from core.config import config from core.dependence.mysql import MySQL from core.dependence.redisdb import RedisDB from core.error import ServiceNotFoundError from core.plugin import Plugin from core.services.cookies.services import CookiesService, PublicCookiesService from core.services.players.services import PlayersService from core.services.users.services import UserService from core.sqlmodel.session import AsyncSession from utils.const import REGION_MAP from utils.log import logger if TYPE_CHECKING: from sqlalchemy import Table from genshin import Client as GenshinClient __all__ = ("GenshinHelper", "PlayerNotFoundError", "CookiesNotFoundError") class PlayerNotFoundError(Exception): def __init__(self, user_id): super().__init__(f"User not found, user_id: {user_id}") class CookiesNotFoundError(Exception): def __init__(self, user_id): super().__init__(f"{user_id} cookies not found") class CharacterDetailsSQLModel(SQLModel, table=True): __tablename__ = "character_details" __table_args__ = ( Index("index_player_character", "player_id", "character_id", unique=True), dict(mysql_charset="utf8mb4", mysql_collate="utf8mb4_general_ci"), ) id: Optional[int] = Field(default=None, sa_column=Column(Integer, primary_key=True, autoincrement=True)) player_id: int = Field(sa_column=Column(BigInteger())) character_id: int = Field(sa_column=Column(BigInteger())) data: Optional[str] = Field(sa_column=Column(String(length=4096))) time_updated: Optional[datetime] = Field(sa_column=Column(DateTime, onupdate=func.now())) # pylint: disable=E1102 class CharacterDetails(Plugin): def __init__( self, mysql: MySQL, redis: RedisDB, ) -> None: self.mysql = mysql self.redis = redis.client self.ttl = 60 * 60 * 6 async def initialize(self) -> None: def fetch_and_update_objects(connection): if not self.mysql.engine.dialect.has_table(connection, table_name="character_details"): logger.info("正在创建角色详细信息表") table: "Table" = SQLModel.metadata.tables["character_details"] table.create(connection) logger.success("创建角色详细信息表成功") async with self.mysql.engine.begin() as conn: await conn.run_sync(fetch_and_update_objects) asyncio.create_task(self.save_character_details_task(max_ttl=None)) self.application.job_queue.run_daily(self.del_old_data_job, time(hour=3, minute=0)) self.application.job_queue.run_repeating(self.save_character_details_job, timedelta(hours=1)) async def save_character_details_job(self, _: ContextTypes.DEFAULT_TYPE): await self.save_character_details() async def del_old_data_job(self, _: ContextTypes.DEFAULT_TYPE): await self.del_old_data(timedelta(days=7)) async def del_old_data(self, expiration_time: timedelta): expire_time = datetime.now() - expiration_time statement = delete(CharacterDetailsSQLModel).where(CharacterDetailsSQLModel.time_updated <= expire_time) async with AsyncSession(self.mysql.engine) as session: await session.execute(statement) async def save_character_details_task(self, max_ttl: Optional[int] = 60 * 60): logger.info("正在从Redis中保存角色详细信息") try: await self.save_character_details(max_ttl) except SQLAlchemyError as exc: logger.error("写入到数据库失败 code[%s]", exc.code) logger.debug("写入到数据库失败", exc_info=exc) except Exception as exc: logger.error("save_character_details 执行失败", exc_info=exc) else: logger.success("从Redis中保存角色详细信息成功") async def save_character_details(self, max_ttl: Optional[int] = 60 * 60): keys = await self.redis.keys("plugins:character_details:*") for key in keys: key = str(key, encoding="utf-8") ttl = await self.redis.ttl(key) if max_ttl is None or (0 <= ttl <= max_ttl): try: player_id, character_id = re.findall(r"\d+", key) except ValueError: logger.warning("非法Key %s", key) continue data = await self.redis.get(key) if data is None: logger.warning("Redis key[%s] 数据未找到", key) # 如果未找到可能因为处理过程中已经过期,导致该数据未回写到 MySQL continue str_data = str(data, encoding="utf-8") async with AsyncSession(self.mysql.engine) as session: statement = ( select(CharacterDetailsSQLModel) .where(CharacterDetailsSQLModel.player_id == player_id) .where(CharacterDetailsSQLModel.character_id == character_id) ) results = await session.exec(statement) sql_data = results.first() if sql_data is None: sql_data = CharacterDetailsSQLModel( player_id=player_id, character_id=character_id, data=str_data, time_updated=datetime.now() ) async with AsyncSession(self.mysql.engine) as session: session.add(sql_data) await session.commit() else: if sql_data.time_updated <= datetime.now() - timedelta(hours=2): sql_data.data = str_data sql_data.time_updated = datetime.now() async with AsyncSession(self.mysql.engine) as session: session.add(sql_data) await session.commit() @staticmethod def get_qname(uid: int, character: int): return f"plugins:character_details:{uid}:{character}" async def get_character_details_for_redis( self, uid: int, character_id: int, ) -> Optional["CalculatorCharacterDetails"]: name = self.get_qname(uid, character_id) data = await self.redis.get(name) if data is None: return None json_data = str(data, encoding="utf-8") return CalculatorCharacterDetails.parse_raw(json_data) async def set_character_details_for_redis(self, uid: int, character_id: int, detail: "CalculatorCharacterDetails"): randint = random.randint(1, 30) # nosec await self.redis.set( self.get_qname(uid, character_id), detail.json(), ex=self.ttl + randint * 60 # 使用随机数防止缓存雪崩 ) async def get_character_details_for_mysql( self, uid: int, character_id: int, ) -> Optional["CalculatorCharacterDetails"]: async with AsyncSession(self.mysql.engine) as session: statement = ( select(CharacterDetailsSQLModel) .where(CharacterDetailsSQLModel.player_id == uid) .where(CharacterDetailsSQLModel.character_id == character_id) ) results = await session.exec(statement) data = results.first() if data is not None: try: return CalculatorCharacterDetails.parse_raw(data.data) except ValidationError as exc: logger.error("解析数据出现异常 ValidationError", exc_info=exc) await session.delete(data) await session.commit() except ValueError as exc: logger.error("解析数据出现异常 ValueError", exc_info=exc) await session.delete(data) await session.commit() return None async def get_character_details( self, client: "GenshinClient", character: "Union[int,BaseCharacter]" ) -> Optional["CalculatorCharacterDetails"]: """缓存 character_details 并定时对其进行数据存储 当遇到 Too Many Requests 可以获取以前的数据 :param client: genshin.py :param character: :return: """ uid = client.uid if uid is not None: if isinstance(character, BaseCharacter): character_id = character.id else: character_id = character detail = await self.get_character_details_for_redis(uid, character_id) if detail is not None: return detail try: detail = await client.get_character_details(character) except GenshinException as exc: if "Too Many Requests" in exc.msg: return await self.get_character_details_for_mysql(uid, character_id) raise exc await self.set_character_details_for_redis(uid, character_id, detail) return detail try: return await client.get_character_details(character) except GenshinException as exc: if "Too Many Requests" in exc.msg: logger.warning("Too Many Requests") else: raise exc return None class GenshinHelper(Plugin): def __init__( self, cookies: CookiesService, public_cookies: PublicCookiesService, user: UserService, redis: RedisDB, player: PlayersService, ) -> None: self.cookies_service = cookies self.public_cookies_service = public_cookies self.user_service = user self.redis_db = redis self.players_service = player if self.redis_db and config.genshin_ttl: self.genshin_cache = genshin.RedisCache(self.redis_db.client, ttl=config.genshin_ttl) else: self.genshin_cache = None if None in (temp := [self.user_service, self.cookies_service, self.players_service]): raise ServiceNotFoundError(*filter(lambda x: x is None, temp)) @staticmethod def region_server(uid: Union[int, str]) -> RegionEnum: if isinstance(uid, (int, str)): region = REGION_MAP.get(str(uid)[0]) else: raise TypeError("UID variable type error") if region: return region raise ValueError(f"UID {uid} isn't associated with any region.") async def get_genshin_client( self, user_id: int, region: Optional[RegionEnum] = None, need_cookie: bool = True ) -> Optional[genshin.Client]: """通过 user_id 和 region 获取私有的 `genshin.Client`""" player = await self.players_service.get_player(user_id, region) if player is None: raise PlayerNotFoundError(user_id) cookies = None if need_cookie: if player.account_id is None: raise CookiesNotFoundError(user_id) cookie_model = await self.cookies_service.get(player.user_id, player.account_id, player.region) if cookie_model is None: raise CookiesNotFoundError(user_id) cookies = cookie_model.data uid = player.player_id region = player.region if region == RegionEnum.HYPERION: # 国服 game_region = genshin.types.Region.CHINESE elif region == RegionEnum.HOYOLAB: # 国际服 game_region = genshin.types.Region.OVERSEAS else: raise TypeError("Region is not None") client = genshin.Client( cookies, lang="zh-cn", game=genshin.types.Game.GENSHIN, region=game_region, uid=uid, hoyolab_id=player.account_id, ) if self.genshin_cache is not None: client.cache = self.genshin_cache return client async def get_public_genshin_client(self, user_id: int) -> Tuple[genshin.Client, int]: """通过 user_id 获取公共的 `genshin.Client`""" player = await self.players_service.get_player(user_id) region = player.region cookies = await self.public_cookies_service.get_cookies(user_id, region) uid = player.player_id if region is RegionEnum.HYPERION: game_region = genshin.types.Region.CHINESE elif region is RegionEnum.HOYOLAB: game_region = genshin.types.Region.OVERSEAS else: raise TypeError("Region is not `RegionEnum.NULL`") client = genshin.Client( cookies.data, region=game_region, uid=uid, game=genshin.types.Game.GENSHIN, lang="zh-cn" ) if self.genshin_cache is not None: client.cache = self.genshin_cache return client, uid