Maas-Model-platform/backend/app/services/cache_service.py

"""
缓存服务

提供Redis缓存操作，用于存储API密钥、模型信息和用户权限等数据，减轻数据库压力。
"""
from typing import Optional, List, Dict, Any
import json
import logging
from app.core.redis import redis_manager

logger = logging.getLogger(__name__)


class CacheService:
    """缓存服务类"""
    
    # 缓存键前缀
    API_KEY_PREFIX = "api_key:"
    MODEL_PREFIX = "model:"
    USER_PERMISSION_PREFIX = "user_permission:"
    USER_LOCAL_MODELS_PREFIX = "user_local_models:"
    
    # 缓存过期时间（秒）
    API_KEY_EXPIRE = 3600  # 1小时
    MODEL_EXPIRE = 7200  # 2小时
    PERMISSION_EXPIRE = 1800  # 30分钟
    USER_LOCAL_MODELS_EXPIRE = 1800  # 30分钟
    
    @classmethod
    async def get_api_key(cls, key_id: int) -> Optional[Dict[str, Any]]:
        """获取API密钥缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return None
        
        try:
            data = await redis.get(f"{cls.API_KEY_PREFIX}{key_id}")
            return json.loads(data) if data else None
        except Exception as e:
            logger.warning(f"获取API密钥缓存失败: {e}")
            return None
    
    @classmethod
    async def set_api_key(cls, key_id: int, data: Dict[str, Any]):
        """设置API密钥缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return
        
        try:
            await redis.setex(
                f"{cls.API_KEY_PREFIX}{key_id}",
                cls.API_KEY_EXPIRE,
                json.dumps(data)
            )
        except Exception as e:
            logger.warning(f"设置API密钥缓存失败: {e}")
    
    @classmethod
    async def get_model(cls, model_id: int) -> Optional[Dict[str, Any]]:
        """获取模型缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return None
        
        try:
            data = await redis.get(f"{cls.MODEL_PREFIX}{model_id}")
            return json.loads(data) if data else None
        except Exception as e:
            logger.warning(f"获取模型缓存失败: {e}")
            return None
    
    @classmethod
    async def set_model(cls, model_id: int, data: Dict[str, Any]):
        """设置模型缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return
        
        try:
            await redis.setex(
                f"{cls.MODEL_PREFIX}{model_id}",
                cls.MODEL_EXPIRE,
                json.dumps(data)
            )
        except Exception as e:
            logger.warning(f"设置模型缓存失败: {e}")
    
    @classmethod
    async def get_user_permission(cls, user_id: str, model_id: int) -> Optional[bool]:
        """获取用户权限缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return None
        
        try:
            data = await redis.get(f"{cls.USER_PERMISSION_PREFIX}{user_id}:{model_id}")
            return data == "1" if data else None
        except Exception as e:
            logger.warning(f"获取用户权限缓存失败: {e}")
            return None
    
    @classmethod
    async def set_user_permission(cls, user_id: str, model_id: int, has_access: bool):
        """设置用户权限缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return
        
        try:
            await redis.setex(
                f"{cls.USER_PERMISSION_PREFIX}{user_id}:{model_id}",
                cls.PERMISSION_EXPIRE,
                "1" if has_access else "0"
            )
        except Exception as e:
            logger.warning(f"设置用户权限缓存失败: {e}")
    
    @classmethod
    async def get_user_local_models(cls, user_id: str) -> Optional[List[int]]:
        """获取用户本地模型列表缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return None
        
        try:
            data = await redis.get(f"{cls.USER_LOCAL_MODELS_PREFIX}{user_id}")
            return json.loads(data) if data else None
        except Exception as e:
            logger.warning(f"获取用户本地模型列表缓存失败: {e}")
            return None
    
    @classmethod
    async def set_user_local_models(cls, user_id: str, model_ids: List[int]):
        """设置用户本地模型列表缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return
        
        try:
            await redis.setex(
                f"{cls.USER_LOCAL_MODELS_PREFIX}{user_id}",
                cls.USER_LOCAL_MODELS_EXPIRE,
                json.dumps(model_ids)
            )
        except Exception as e:
            logger.warning(f"设置用户本地模型列表缓存失败: {e}")
    
    @classmethod
    async def delete_api_key(cls, key_id: int):
        """删除API密钥缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return
        
        try:
            await redis.delete(f"{cls.API_KEY_PREFIX}{key_id}")
        except Exception as e:
            logger.warning(f"删除API密钥缓存失败: {e}")
    
    @classmethod
    async def delete_model(cls, model_id: int):
        """删除模型缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return
        
        try:
            await redis.delete(f"{cls.MODEL_PREFIX}{model_id}")
        except Exception as e:
            logger.warning(f"删除模型缓存失败: {e}")
    
    @classmethod
    async def delete_user_permission(cls, user_id: str, model_id: int):
        """删除用户权限缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return
        
        try:
            await redis.delete(f"{cls.USER_PERMISSION_PREFIX}{user_id}:{model_id}")
        except Exception as e:
            logger.warning(f"删除用户权限缓存失败: {e}")
    
    @classmethod
    async def delete_user_local_models(cls, user_id: str):
        """删除用户本地模型列表缓存"""
        redis = redis_manager.get_client()
        if not redis:
            return
        
        try:
            await redis.delete(f"{cls.USER_LOCAL_MODELS_PREFIX}{user_id}")
        except Exception as e:
            logger.warning(f"删除用户本地模型列表缓存失败: {e}")