Tolan/backend/services/summary_service.py

"""
对话总结服务（SQLAlchemy ORM 版本）
"""
import uuid
from typing import Awaitable, Callable, List, Dict, Optional
import logging
from sqlalchemy import select
import redis.asyncio as aioredis

logger = logging.getLogger(__name__)


class SummaryService:
    """对话总结服务"""
    
    def __init__(self, db_manager):
        """
        初始化总结服务
        
        Args:
            db_manager: 数据库管理器实例
        """
        self.db = db_manager
        self.redis = None  # 🔥 延迟获取Redis
        self._redis_provider: Optional[Callable[[], Awaitable[aioredis.Redis]]] = (
            db_manager.get_redis if db_manager else None
        )
    
    async def _ensure_redis(self):
        """确保Redis连接可用"""
        if not self._redis_provider:
            if not self.db:
                raise RuntimeError("Redis 未初始化，缺少数据库管理器")
            self._redis_provider = self.db.get_redis
        redis_client = await self._redis_provider()
        if not redis_client:
            raise RuntimeError("Redis 客户端不可用")
        self.redis = redis_client
        return redis_client
    
    async def generate_summary(
        self,
        user_id: str,
        start_round: int,
        end_round: int
    ):
        """
        生成对话总结（异步执行）
        
        逻辑：
        - 如果是第一次总结（start_round=1）：只总结当前15轮消息
        - 如果是后续总结：获取之前的总结 + 当前15轮消息，进行累积总结
        
        Args:
            user_id: 用户ID
            start_round: 起始轮次
            end_round: 结束轮次
        """
        
        try:
            # 🔥 确保Redis连接可用
            await self._ensure_redis()
            
            logger.info(f"🔄 开始生成总结: user={user_id}, rounds={start_round}-{end_round}")
            
            # 0. 参数验证
            if end_round <= 0:
                logger.warning(f"⚠️ 无效的总结范围: user={user_id}, start_round={start_round}, end_round={end_round}")
                return
            
            if start_round > end_round:
                logger.warning(f"⚠️ 总结范围错误: user={user_id}, start_round={start_round}, end_round={end_round}")
                return
            
            # 使用 ORM 查询
            from database import db_manager
            from models_db import ConversationSummary, Message
            
            # 1. 检查是否已存在总结
            async with db_manager.get_session() as session:
                existing_query = await session.execute(
                    select(ConversationSummary)
                    .where(ConversationSummary.user_id == user_id)
                    .where(ConversationSummary.start_round == start_round)
                    .where(ConversationSummary.end_round == end_round)
                )
                existing = existing_query.scalar_one_or_none()
                
                if existing:
                    logger.info(f"⏭️ 总结已存在，跳过: user={user_id}, rounds={start_round}-{end_round}")
                    return
                
                # 2. 获取之前的总结（如果有）
                previous_summaries = []
                if start_round > 1:
                    previous_query = await session.execute(
                        select(ConversationSummary)
                        .where(ConversationSummary.user_id == user_id)
                        .where(ConversationSummary.end_round < start_round)
                        .order_by(ConversationSummary.end_round.asc())
                    )
                    previous_summaries_list = previous_query.scalars().all()
                    previous_summaries = [s.summary_text for s in previous_summaries_list]
                    logger.info(f"📚 获取到 {len(previous_summaries)} 个历史总结")
                
                # 3. 获取待总结的消息
                logger.info(f"🔍 [总结查询] user={user_id}, start_round={start_round}, end_round={end_round}")
                
                messages_query = await session.execute(
                    select(Message)
                    .where(Message.user_id == user_id)
                    .where(Message.round_number >= start_round)
                    .where(Message.round_number <= end_round)
                    .order_by(Message.round_number.asc(), Message.id.asc())
                )
                messages_list = messages_query.scalars().all()
                messages = [
                    {
                        "role": msg.role,
                        "content": msg.content,
                        "round_number": msg.round_number,
                    }
                    for msg in messages_list
                ]
            
            logger.info(f"📊 [总结查询结果] user={user_id}, 查询到 {len(messages)} 条消息")
            
            # 如果查询不到消息，检查数据库中实际有哪些轮次
            if not messages:
                # 诊断：查询该用户的所有消息轮次
                async with db_manager.get_session() as session:
                    from sqlalchemy import func
                    all_rounds_query = await session.execute(
                        select(Message.round_number, func.count(Message.id).label('count'))
                        .where(Message.user_id == user_id)
                        .group_by(Message.round_number)
                        .order_by(Message.round_number.asc())
                    )
                    all_rounds_list = all_rounds_query.all()
                    all_rounds = [
                        {"round_number": r.round_number, "count": r.count}
                        for r in all_rounds_list
                    ]
                
                # 过滤掉 round_number = 0 的记录（可能是历史错误数据）
                valid_rounds = [r['round_number'] for r in all_rounds if r['round_number'] > 0] if all_rounds else []
                
                logger.warning(
                    f"⚠️ 没有消息可总结: user={user_id}, "
                    f"查询范围={start_round}-{end_round}, "
                    f"数据库中实际轮次={valid_rounds if valid_rounds else '无消息'}, "
                    f"总记录数={len(all_rounds) if all_rounds else 0}"
                )
                return
            
            # 记录查询到的消息轮次范围
            if messages:
                actual_start = min(m['round_number'] for m in messages)
                actual_end = max(m['round_number'] for m in messages)
                logger.info(f"✅ [总结查询] 实际消息轮次范围: {actual_start}-{actual_end}, 共 {len(messages)} 条")
            
            # 4. 调用LLM生成总结（包含之前的总结）
            summary_text = await self._generate_summary_with_llm(messages, previous_summaries)
            
            if not summary_text:
                logger.warning(f"⚠️ 总结生成失败: user={user_id}")
                return
            
            # 4. 保存总结到MySQL（使用 ORM）
            summary_id = str(uuid.uuid4())
            
            async with db_manager.get_session() as session:
                summary = ConversationSummary(
                    summary_id=summary_id,
                    user_id=user_id,
                    start_round=start_round,
                    end_round=end_round,
                    summary_text=summary_text,
                    summary_type="auto",
                    message_count=len(messages),
                    token_count=len(summary_text) // 2
                )
                session.add(summary)
                await session.commit()
            
            logger.info(f"✅ 总结保存成功: user={user_id}, messages={len(messages)}")
            
            # 5. 更新Redis缓存
            await self._update_summary_cache(user_id, summary_text)
            
            # 6. 更新元数据
            meta_key = f"user:{user_id}:meta"
            await self.redis.hset(meta_key, 'last_summary_round', end_round)
            await self.redis.expire(meta_key, 86400)
            
            logger.info(f"🎉 总结完成: user={user_id}, rounds={start_round}-{end_round}")
            
        except Exception as e:
            logger.error(f"❌ 生成总结失败: user={user_id}, error={e}")
            import traceback
            traceback.print_exc()
    
    async def _generate_summary_with_llm(
        self, 
        messages: List[Dict], 
        previous_summaries: List[str] = None
    ) -> str:
        """
        调用LLM生成总结
        
        总结内容聚焦于：
        1. 用户个人信息（姓名、年龄、家庭住址等）- 用于后续精准回答用户问题
        2. 用户当前英语水平评估 - 用于逐步引导用户提高英语水平
        
        Args:
            messages: 消息列表
            previous_summaries: 之前的总结列表（用于累积总结）
        
        Returns:
            总结文本（包含用户信息和英语水平评估）
        """
        
        try:
            # 构建对话内容
            conversation_lines = []
            for msg in messages:
                role = msg['role']
                content = msg['content']
                round_num = msg.get('round_number', '')
                
                role_label = "用户" if role == "user" else "助手"
                conversation_lines.append(f"[第{round_num}轮] {role_label}: {content}")
            
            conversation = "\n".join(conversation_lines)
            
            # 构建总结Prompt
            if previous_summaries:
                # 有历史总结：进行累积总结
                previous_summaries_text = "\n\n".join([
                    f"【历史总结{i+1}】\n{summary}" 
                    for i, summary in enumerate(previous_summaries)
                ])
                
                prompt = f"""请基于以下历史总结和新的对话内容，生成一个更新的累积总结。

**历史总结：**
{previous_summaries_text}

**新的对话内容（第{messages[0].get('round_number', 1)}-{messages[-1].get('round_number', 15)}轮）：**
{conversation}

请整合历史总结和新对话内容，更新以下两类关键信息：

**第一类：用户个人信息**
请提取并整理用户在对话中透露的个人信息，包括但不限于：
- 姓名、昵称
- 年龄
- 家庭住址、所在城市
- 职业、工作
- 兴趣爱好
- 家庭成员信息
- 其他个人背景信息

如果某些信息未在对话中提及，请标注"未提及"。

**第二类：用户英语水平评估**
请评估用户当前的英语水平，包括：
- 词汇量水平（基础/中级/高级）
- 语法掌握程度（能使用哪些时态、句型）
- 口语流利度（是否流畅、是否有较多停顿）
- 发音准确性（是否有明显发音问题）
- 表达能力（能否清晰表达想法）
- 当前薄弱环节（需要重点提升的方面）

请按照以下格式输出总结（每部分100-150字）：

**用户个人信息：**
[整合后的用户个人信息，未提及的标注"未提及"]

**英语水平评估：**
[对用户当前英语水平的详细评估]"""
            else:
                # 第一次总结：只总结当前对话
                prompt = f"""请从以下英语口语练习对话中提炼两类关键信息：

**第一类：用户个人信息**
请提取并整理用户在对话中透露的个人信息，包括但不限于：
- 姓名、昵称
- 年龄
- 家庭住址、所在城市
- 职业、工作
- 兴趣爱好
- 家庭成员信息
- 其他个人背景信息

如果某些信息未在对话中提及，请标注"未提及"。

**第二类：用户英语水平评估**
请评估用户当前的英语水平，包括：
- 词汇量水平（基础/中级/高级）
- 语法掌握程度（能使用哪些时态、句型）
- 口语流利度（是否流畅、是否有较多停顿）
- 发音准确性（是否有明显发音问题）
- 表达能力（能否清晰表达想法）
- 当前薄弱环节（需要重点提升的方面）

对话内容：
{conversation}

请按照以下格式输出总结（每部分100-150字）：

**用户个人信息：**
[提取的用户个人信息，未提及的标注"未提及"]

**英语水平评估：**
[对用户当前英语水平的详细评估]"""
            
            # 调用AI服务生成总结
            from ai_service import AIService
            ai_service = AIService()
            
            # 使用非流式调用
            summary = await ai_service.generate_summary(prompt)
            
            return summary.strip()
            
        except Exception as e:
            logger.error(f"❌ LLM总结生成失败: {e}")
            
            # 降级方案：简单拼接
            return f"**用户个人信息：**\n未提及\n\n**英语水平评估：**\n对话总结（第{messages[0].get('round_number', 1)}-{messages[-1].get('round_number', 15)}轮）：共{len(messages)}条消息的英语口语练习对话。"
    
    async def _update_summary_cache(self, user_id: str, summary_text: str):
        """
        更新Redis中的总结缓存
        
        Args:
            user_id: 用户ID
            summary_text: 总结文本
        """
        try:
            redis = await self._ensure_redis()
            cache_key = f"user:{user_id}:latest_summary"
            await redis.setex(cache_key, 86400, summary_text)
            logger.debug(f"✅ 总结已缓存: user={user_id}")
        except Exception as e:
            logger.warning(f"⚠️ 缓存总结失败: user={user_id}, error={e}")
    
    async def get_latest_summary(self, user_id: str) -> str:
        """
        获取用户最新的总结
        
        Args:
            user_id: 用户ID
        
        Returns:
            总结文本或空字符串
        """
        try:
            redis = await self._ensure_redis()
            # 先查Redis
            cache_key = f"user:{user_id}:latest_summary"
            cached = await redis.get(cache_key)
            
            if cached:
                return cached
            
            # 再查MySQL（使用 ORM）
            from database import db_manager
            from models_db import ConversationSummary
            
            async with db_manager.get_session() as session:
                result_query = await session.execute(
                    select(ConversationSummary)
                    .where(ConversationSummary.user_id == user_id)
                    .order_by(ConversationSummary.end_round.desc())
                    .limit(1)
                )
                summary = result_query.scalar_one_or_none()
            
                if summary:
                    summary_text = summary.summary_text
                # 回填缓存
                await redis.setex(cache_key, 86400, summary_text)
                return summary_text
            
            return ""
            
        except Exception as e:
            logger.error(f"❌ 获取最新总结失败: user={user_id}, error={e}")
            return ""