Tolan/backend/services/redis_cache_service.py

"""
Redis缓存管理服务
用于高速缓存对话上下文，减少MySQL查询
"""
import json
import time
from typing import Awaitable, Callable, Dict, List, Optional
import logging
import redis.asyncio as aioredis

logger = logging.getLogger(__name__)


class RedisCacheService:
    """Redis缓存服务"""
    
    def __init__(self, redis_provider: Callable[[], Awaitable[aioredis.Redis]]):
        """
        初始化Redis缓存服务
        
        Args:
            redis_provider: 一个可调用对象，返回 redis.asyncio.Redis 实例
        """
        if not callable(redis_provider):
            raise ValueError("redis_provider 必须可调用")
        self._redis_provider = redis_provider
        
        # 缓存Key设计
        self.KEY_USER_CONTEXT = "context:{user_id}"           # 用户上下文
        self.KEY_USER_MESSAGES = "messages:{user_id}"         # 用户消息列表
        self.KEY_USER_ROUND = "round:{user_id}"               # 当前轮次
        self.KEY_USER_SUMMARY = "summary:{user_id}:{end_round}" # 历史总结
        
        # TTL设置
        self.TTL_CONTEXT = 86400      # 24小时
        self.TTL_MESSAGES = 86400     # 24小时
        self.TTL_SUMMARY = 604800     # 7天
    
    async def _get_redis(self) -> aioredis.Redis:
        """获取最新的Redis客户端，确保连接有效"""
        redis_client = await self._redis_provider()
        if not redis_client:
            raise RuntimeError("Redis 客户端不可用")
        return redis_client
    
    async def get_context(self, user_id: str) -> Optional[Dict]:
        """
        从Redis获取用户上下文（零延迟）
        
        Returns:
            {
                "summary": str | None,  # 最新的一个总结（已包含所有历史信息）
                "recent_messages": List[Dict],
                "current_round": int,
                "total_tokens": int
            }
        """
        try:
            redis = await self._get_redis()
            key = self.KEY_USER_CONTEXT.format(user_id=user_id)
            cached = await redis.get(key)
            
            if cached:
                context = json.loads(cached)
                logger.info(f"⚡ Redis命中: user={user_id}, round={context.get('current_round')}")
                return context
            
            logger.info(f"⚠️ Redis未命中: user={user_id}")
            return None
            
        except Exception as e:
            logger.error(f"❌ Redis读取失败: {e}")
            return None
    
    async def set_context(
        self, 
        user_id: str, 
        context: Dict,
        ttl: int = None
    ) -> bool:
        """
        更新Redis上下文缓存（超快写入）
        
        Args:
            user_id: 用户ID
            context: 上下文数据
            ttl: 过期时间（秒），默认24小时
        """
        try:
            redis = await self._get_redis()
            key = self.KEY_USER_CONTEXT.format(user_id=user_id)
            ttl = ttl or self.TTL_CONTEXT
            
            await redis.setex(
                key,
                ttl,
                json.dumps(context, ensure_ascii=False)
            )
            
            logger.info(f"✅ Redis更新: user={user_id}")
            return True
            
        except Exception as e:
            logger.error(f"❌ Redis写入失败: {e}")
            return False
    
    async def append_message(
        self, 
        user_id: str, 
        role: str, 
        content: str,
        round_number: int
    ) -> bool:
        """
        追加消息到Redis（实时更新）
        
        使用Redis List结构，支持高效追加和范围查询
        """
        try:
            redis = await self._get_redis()
            key = self.KEY_USER_MESSAGES.format(user_id=user_id)
            
            message = {
                'role': role,
                'content': content,
                'round_number': round_number,
                'timestamp': time.time()
            }
            
            # 追加到List尾部
            await redis.rpush(key, json.dumps(message, ensure_ascii=False))
            
            # 设置过期时间
            await redis.expire(key, self.TTL_MESSAGES)
            
            # 限制List长度（最多保留200条消息）
            await redis.ltrim(key, -200, -1)
            
            logger.debug(f"✅ Redis追加消息: user={user_id}, round={round_number}, role={role}")
            return True
            
        except Exception as e:
            logger.error(f"❌ Redis追加失败: {e}")
            return False
    
    async def get_recent_messages(
        self, 
        user_id: str, 
        count: int = 30
    ) -> List[Dict]:
        """
        获取最近N条消息（从Redis List）
        
        Args:
            user_id: 用户ID
            count: 消息数量（默认30条，即15轮）
        """
        try:
            redis = await self._get_redis()
            key = self.KEY_USER_MESSAGES.format(user_id=user_id)
            
            # 获取最后N条消息
            messages_raw = await redis.lrange(key, -count, -1)
            
            messages = []
            for msg_bytes in messages_raw:
                try:
                    # 解码并解析JSON
                    msg_str = msg_bytes.decode('utf-8') if isinstance(msg_bytes, bytes) else msg_bytes
                    messages.append(json.loads(msg_str))
                except Exception as e:
                    logger.warning(f"⚠️ 解析消息失败: {e}")
                    continue
            
            logger.debug(f"⚡ Redis获取消息: user={user_id}, count={len(messages)}")
            return messages
            
        except Exception as e:
            logger.error(f"❌ Redis读取消息失败: {e}")
            return []
    
    async def get_current_round(self, user_id: str) -> int:
        """获取当前轮次"""
        try:
            redis = await self._get_redis()
            key = self.KEY_USER_ROUND.format(user_id=user_id)
            round_bytes = await redis.get(key)
            
            if round_bytes:
                round_str = round_bytes.decode('utf-8') if isinstance(round_bytes, bytes) else round_bytes
                return int(round_str)
            return 0
            
        except Exception as e:
            logger.error(f"❌ Redis读取轮次失败: {e}")
            return 0
    
    async def set_current_round(self, user_id: str, round_number: int) -> bool:
        """设置当前轮次"""
        try:
            redis = await self._get_redis()
            key = self.KEY_USER_ROUND.format(user_id=user_id)
            await redis.setex(key, self.TTL_CONTEXT, str(round_number))
            return True
        except Exception as e:
            logger.error(f"❌ Redis设置轮次失败: {e}")
            return False
    
    async def increment_round(self, user_id: str) -> int:
        """轮次自增"""
        try:
            redis = await self._get_redis()
            key = self.KEY_USER_ROUND.format(user_id=user_id)
            new_round = await redis.incr(key)
            await redis.expire(key, self.TTL_CONTEXT)
            
            # 同时更新 user:{user_id}:meta 中的 current_round（保持一致性）
            meta_key = f"user:{user_id}:meta"
            await redis.hset(meta_key, 'current_round', new_round)
            await redis.expire(meta_key, self.TTL_CONTEXT)
            
            logger.debug(f"✅ Redis轮次自增: user={user_id}, new_round={new_round}")
            return new_round
            
        except Exception as e:
            logger.error(f"❌ Redis轮次自增失败: {e}")
            return 0
    
    async def get_summary(self, user_id: str, end_round: int) -> Optional[str]:
        """获取历史总结"""
        try:
            redis = await self._get_redis()
            key = self.KEY_USER_SUMMARY.format(user_id=user_id, end_round=end_round)
            summary_bytes = await redis.get(key)
            
            if summary_bytes:
                summary = summary_bytes.decode('utf-8') if isinstance(summary_bytes, bytes) else summary_bytes
                logger.info(f"⚡ Redis获取总结: user={user_id}, end_round={end_round}")
                return summary
            
            return None
            
        except Exception as e:
            logger.error(f"❌ Redis读取总结失败: {e}")
            return None
    
    async def set_summary(
        self, 
        user_id: str, 
        end_round: int, 
        summary: str
    ) -> bool:
        """保存历史总结"""
        try:
            redis = await self._get_redis()
            key = self.KEY_USER_SUMMARY.format(user_id=user_id, end_round=end_round)
            await redis.setex(key, self.TTL_SUMMARY, summary)
            
            logger.info(f"✅ Redis保存总结: user={user_id}, end_round={end_round}, 长度={len(summary)}字")
            return True
            
        except Exception as e:
            logger.error(f"❌ Redis保存总结失败: {e}")
            return False
    
    async def build_context_from_cache(
        self, 
        user_id: str, 
        current_round: int,
        db_manager=None
    ) -> Dict:
        """
        从Redis构建完整上下文（超快）
        
        逻辑：
        - 如果 current_round <= 15: 返回所有消息
        - 如果 current_round > 15: 返回所有历史总结 + 最近15轮
        
        Args:
            user_id: 用户ID
            current_round: 当前轮次
            db_manager: 数据库管理器（用于获取所有总结）
        """
        try:
            if current_round <= 15:
                # 前15轮：直接返回所有消息
                messages = await self.get_recent_messages(user_id, count=current_round * 2)
                
                return {
                    "summary": None,
                    "recent_messages": messages,
                    "current_round": current_round,
                    "total_tokens": self._estimate_tokens(messages=messages)
                }
            else:
                # 超过15轮：最新总结 + 最近15轮
                # 🔥 修复：查找 end_round < current_round 的最新总结
                # 因为总结是累积的，最新的总结包含了所有历史信息
                summary_end_round = current_round - 1
                
                logger.info(f"🔍 [Redis] 查询总结: user={user_id}, current_round={current_round}, summary_end_round={summary_end_round}")
                
                # 获取消息
                messages = await self.get_recent_messages(user_id, count=30)  # 15轮=30条消息
                
                # 从数据库获取最新的一个总结（已包含所有历史信息，使用 ORM）
                summary = None
                if db_manager:
                    try:
                        from models_db import ConversationSummary
                        from sqlalchemy import select, func
                        
                        async with db_manager.get_session() as session:
                        # 先查询所有总结，用于调试
                            all_summaries_result = await session.execute(
                                select(
                                    ConversationSummary.end_round,
                                    ConversationSummary.start_round,
                                    ConversationSummary.created_at
                                )
                                .where(ConversationSummary.user_id == user_id)
                                .order_by(ConversationSummary.end_round.desc())
                                .limit(10)
                            )
                            all_summaries = [
                                {
                                    'end_round': row.end_round,
                                    'start_round': row.start_round,
                                    'created_at': row.created_at
                                }
                                for row in all_summaries_result.all()
                            ]
                            
                            if all_summaries:
                                logger.info(f"📊 [Redis] 数据库中的总结: user={user_id}, 总数={len(all_summaries)}, 列表={[(s['start_round'], s['end_round']) for s in all_summaries]}")
                            else:
                                logger.warning(f"⚠️ [Redis] 数据库中没有任何总结: user={user_id}")
                                
                                # 调试：查询数据库中所有不同的 user_id
                                try:
                                    all_user_ids_result = await session.execute(
                                        select(
                                            ConversationSummary.user_id,
                                            func.count(ConversationSummary.id).label('summary_count')
                                        )
                                        .group_by(ConversationSummary.user_id)
                                        .limit(10)
                                    )
                                    all_user_ids = [
                                        {
                                            'user_id': row.user_id,
                                            'summary_count': row.summary_count
                                        }
                                        for row in all_user_ids_result.all()
                                    ]
                                    if all_user_ids:
                                        logger.info(f"🔍 [Redis] 数据库中存在的其他 user_id: {[(u['user_id'], u['summary_count']) for u in all_user_ids]}")
                                except Exception as e:
                                    logger.debug(f"查询所有 user_id 失败: {e}")
                            
                            # 查询符合条件的总结
                            summary_result = await session.execute(
                                select(
                                    ConversationSummary.summary_text,
                                    ConversationSummary.end_round
                                )
                                .where(
                                    ConversationSummary.user_id == user_id,
                                    ConversationSummary.end_round <= summary_end_round
                                )
                                .order_by(ConversationSummary.end_round.desc())
                                .limit(1)
                            )
                            summary_row = summary_result.first()
                            if summary_row:
                                summary = summary_row.summary_text
                                end_round = summary_row.end_round
                                logger.info(f"✅ [Redis] 获取最新总结: user={user_id}, end_round={end_round}, summary_length={len(summary)}")
                            else:
                                logger.warning(f"⚠️ [Redis] 未找到总结: user={user_id}, up_to_round={summary_end_round}, 数据库中总结数={len(all_summaries) if all_summaries else 0}")
                    except Exception as e:
                        logger.warning(f"⚠️ 从数据库获取总结失败: {e}")
                        import traceback
                        logger.error(traceback.format_exc())
                
                return {
                    "summary": summary,
                    "recent_messages": messages,
                    "current_round": current_round,
                    "total_tokens": self._estimate_tokens(summary=summary, messages=messages)
                }
                
        except Exception as e:
            logger.error(f"❌ 从Redis构建上下文失败: {e}")
            return {
                "summary": None,
                "recent_messages": [],
                "current_round": current_round,
                "total_tokens": 0
            }
    
    def _estimate_tokens(self, summary: str = None, messages: List[Dict] = None) -> int:
        """估算token数量"""
        total = 0
        if summary:
            total += len(summary) // 2
        if messages:
            for msg in messages:
                total += len(msg.get('content', '')) // 2
        return total
    
    async def invalidate_context(self, user_id: str) -> bool:
        """清除用户上下文缓存（强制刷新）"""
        try:
            redis = await self._get_redis()
            key = self.KEY_USER_CONTEXT.format(user_id=user_id)
            await redis.delete(key)
            logger.info(f"🗑️ Redis清除缓存: user={user_id}")
            return True
        except Exception as e:
            logger.error(f"❌ Redis清除失败: {e}")
            return False
    
    async def sync_from_db(self, user_id: str, messages: List[Dict], current_round: int) -> bool:
        """
        从数据库同步数据到Redis（初始化缓存）
        
        Args:
            user_id: 用户ID
            messages: 消息列表
            current_round: 当前轮次
        """
        try:
            redis = await self._get_redis()
            # 清空旧数据
            messages_key = self.KEY_USER_MESSAGES.format(user_id=user_id)
            await redis.delete(messages_key)
            
            # 批量写入消息
            if messages:
                pipeline = redis.pipeline()
                for msg in messages:
                    pipeline.rpush(
                        messages_key,
                        json.dumps(msg, ensure_ascii=False)
                    )
                pipeline.expire(messages_key, self.TTL_MESSAGES)
                await pipeline.execute()
            
            # 设置轮次
            await self.set_current_round(user_id, current_round)
            
            logger.info(f"✅ Redis同步完成: user={user_id}, messages={len(messages)}, round={current_round}")
            return True
            
        except Exception as e:
            logger.error(f"❌ Redis同步失败: {e}")
            return False
    
    async def health_check(self) -> bool:
        """健康检查"""
        try:
            redis = await self._get_redis()
            await redis.ping()
            return True
        except Exception as e:
            logger.error(f"❌ Redis健康检查失败: {e}")
            return False