Tolan/backend/services/summary_service.py

"""
对话总结服务
"""
import uuid
from typing import List, Dict
import logging

logger = logging.getLogger(__name__)


class SummaryService:
    """对话总结服务"""
    
    def __init__(self, db_manager):
        """
        初始化总结服务
        
        Args:
            db_manager: 数据库管理器实例
        """
        self.db = db_manager
        self.redis = db_manager.get_redis()
    
    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:
            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
            
            # 1. 检查是否已存在总结
            existing = await self.db.execute_one(
                """
                SELECT id FROM conversation_summaries
                WHERE user_id = %(user_id)s 
                  AND start_round = %(start_round)s 
                  AND end_round = %(end_round)s
                """,
                {"user_id": user_id, "start_round": start_round, "end_round": end_round}
            )
            
            if existing:
                logger.info(f"⏭️ 总结已存在，跳过: user={user_id}, rounds={start_round}-{end_round}")
                return
            
            # 2. 获取之前的总结（如果有）
            previous_summaries = []
            if start_round > 1:
                previous_summaries_result = await self.db.execute_query(
                    """
                    SELECT summary_text, end_round
                    FROM conversation_summaries
                    WHERE user_id = %(user_id)s AND end_round < %(start_round)s
                    ORDER BY end_round ASC
                    """,
                    {"user_id": user_id, "start_round": start_round}
                )
                previous_summaries = [s['summary_text'] for s in previous_summaries_result]
                logger.info(f"📚 获取到 {len(previous_summaries)} 个历史总结")
            
            # 3. 获取待总结的消息
            logger.info(f"🔍 [总结查询] user={user_id}, start_round={start_round}, end_round={end_round}")
            
            messages = await self.db.execute_query(
                """
                SELECT role, content, round_number
                FROM messages
                WHERE user_id = %(user_id)s 
                  AND round_number >= %(start_round)s 
                  AND round_number <= %(end_round)s
                ORDER BY round_number ASC, id ASC
                """,
                {"user_id": user_id, "start_round": start_round, "end_round": end_round}
            )
            
            logger.info(f"📊 [总结查询结果] user={user_id}, 查询到 {len(messages)} 条消息")
            
            # 如果查询不到消息，检查数据库中实际有哪些轮次
            if not messages:
                # 诊断：查询该用户的所有消息轮次
                all_rounds = await self.db.execute_query(
                    """
                    SELECT DISTINCT round_number, COUNT(*) as count
                    FROM messages
                    WHERE user_id = %(user_id)s
                    GROUP BY round_number
                    ORDER BY round_number ASC
                    """,
                    {"user_id": user_id}
                )
                
                # 过滤掉 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
            summary_id = str(uuid.uuid4())
            
            query = """
                INSERT INTO conversation_summaries
                (summary_id, user_id, start_round, end_round, summary_text, 
                 summary_type, message_count, token_count)
                VALUES (%(summary_id)s, %(user_id)s, %(start_round)s, %(end_round)s, 
                        %(summary_text)s, %(summary_type)s, %(message_count)s, %(token_count)s)
            """
            
            await self.db.execute_update(query, {
                "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
            })
            
            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:
            cache_key = f"user:{user_id}:latest_summary"
            await self.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
            cache_key = f"user:{user_id}:latest_summary"
            cached = await self.redis.get(cache_key)
            
            if cached:
                return cached
            
            # 再查MySQL
            result = await self.db.execute_one(
                """
                SELECT summary_text
                FROM conversation_summaries
                WHERE user_id = %(user_id)s
                ORDER BY end_round DESC
                LIMIT 1
                """,
                {"user_id": user_id}
            )
            
            if result:
                summary_text = result['summary_text']
                # 回填缓存
                await self.redis.setex(cache_key, 86400, summary_text)
                return summary_text
            
            return ""
            
        except Exception as e:
            logger.error(f"❌ 获取最新总结失败: user={user_id}, error={e}")
            return ""