life-echo/api/test_conversation.py

#!/usr/bin/env python3
"""
多轮对话测试脚本
测试对话引导 Agent 和回忆录整理功能
"""
import asyncio
import json
import uuid
import httpx
import websockets
from datetime import datetime

# 配置
BASE_URL = "http://localhost:8000"
WS_URL = "ws://localhost:8000"

# 测试用户信息
TEST_PHONE = f"138{uuid.uuid4().hex[:8]}"  # 随机手机号避免冲突
TEST_PASSWORD = "test123456"
TEST_NICKNAME = "测试用户"

# 模拟用户的多轮对话内容（关于童年和教育阶段）
CONVERSATION_MESSAGES = [
    # 童年阶段
    "我出生在南方一个小镇，小时候跟奶奶住在一起。",
    "奶奶家有个小院子，夏天的时候我们经常坐在院子里乘凉，她给我讲故事。",
    "那段时光真的很美好，我记得奶奶总是给我做红烧肉，那是我最爱吃的菜。",
    "小时候最开心的事就是过年，可以放鞭炮，还能收到压岁钱。",
    
    # 教育阶段
    "后来我去城里上学了，那是我第一次离开家，心里特别害怕。",
    "初中的时候遇到了一个很好的语文老师，她鼓励我多读书，对我影响很大。",
    "高考那年压力特别大，但最后还是考上了理想的大学。",
]


class ConversationTester:
    """对话测试器"""
    
    def __init__(self):
        self.token = None
        self.user_id = None
        self.conversation_id = str(uuid.uuid4())
    
    async def register_or_login(self):
        """注册或登录用户"""
        async with httpx.AsyncClient(timeout=30.0) as client:
            # 先尝试注册
            print(f"\n📝 注册用户: {TEST_PHONE}")
            resp = await client.post(
                f"{BASE_URL}/api/auth/register",
                json={
                    "phone": TEST_PHONE,
                    "password": TEST_PASSWORD,
                    "nickname": TEST_NICKNAME
                }
            )
            
            if resp.status_code == 201:
                data = resp.json()
                self.token = data["access_token"]
                print(f"✅ 注册成功!")
            elif resp.status_code == 400 and "已被注册" in resp.text:
                # 已注册，尝试登录
                print(f"ℹ️  用户已存在，尝试登录...")
                resp = await client.post(
                    f"{BASE_URL}/api/auth/login",
                    json={
                        "phone": TEST_PHONE,
                        "password": TEST_PASSWORD
                    }
                )
                if resp.status_code == 200:
                    data = resp.json()
                    self.token = data["access_token"]
                    print(f"✅ 登录成功!")
                else:
                    raise Exception(f"登录失败: {resp.text}")
            else:
                raise Exception(f"注册失败: {resp.text}")
            
            print(f"🔑 Token: {self.token[:30]}...")
    
    async def get_memoir_state(self):
        """获取回忆录状态"""
        async with httpx.AsyncClient(timeout=60.0) as client:
            resp = await client.get(
                f"{BASE_URL}/api/memoir-state",
                headers={"Authorization": f"Bearer {self.token}"}
            )
            if resp.status_code != 200:
                print(f"  ⚠️ 状态API返回 {resp.status_code}: {resp.text[:200]}")
                return {"current_stage": "unknown", "covered_stages": [], "slots": {}}
            return resp.json()
    
    async def get_chapters(self):
        """获取章节列表"""
        async with httpx.AsyncClient(timeout=60.0) as client:
            resp = await client.get(
                f"{BASE_URL}/api/chapters",
                headers={"Authorization": f"Bearer {self.token}"}
            )
            if resp.status_code != 200:
                print(f"  ⚠️ 章节API返回 {resp.status_code}: {resp.text[:200]}")
                return []
            return resp.json()
    
    async def get_book(self):
        """获取回忆录信息"""
        async with httpx.AsyncClient(timeout=60.0) as client:
            resp = await client.get(
                f"{BASE_URL}/api/books/current",
                headers={"Authorization": f"Bearer {self.token}"}
            )
            if resp.status_code != 200:
                print(f"  ⚠️ 回忆录API返回 {resp.status_code}: {resp.text[:200]}")
                return {"message": "获取失败"}
            return resp.json()
    
    async def get_tasks_status(self):
        """获取任务状态"""
        async with httpx.AsyncClient(timeout=60.0) as client:
            resp = await client.get(
                f"{BASE_URL}/api/tasks/status",
                headers={"Authorization": f"Bearer {self.token}"}
            )
            if resp.status_code != 200:
                return {"total": 0, "all_completed": True, "tasks": []}
            return resp.json()
    
    async def clear_tasks(self):
        """清除任务记录"""
        async with httpx.AsyncClient(timeout=60.0) as client:
            await client.delete(
                f"{BASE_URL}/api/tasks/clear",
                headers={"Authorization": f"Bearer {self.token}"}
            )
    
    async def run_conversation(self):
        """运行多轮对话"""
        print(f"\n🔗 连接 WebSocket: {self.conversation_id}")
        
        ws_url = f"{WS_URL}/ws/conversation/{self.conversation_id}?token={self.token}"
        
        async with websockets.connect(ws_url) as ws:
            # 接收连接确认
            msg = await ws.recv()
            data = json.loads(msg)
            print(f"✅ 连接成功: {data['type']}")
            
            # 多轮对话
            for i, user_message in enumerate(CONVERSATION_MESSAGES, 1):
                print(f"\n{'='*60}")
                print(f"📤 第 {i} 轮对话")
                print(f"{'='*60}")
                print(f"👤 用户: {user_message}")
                
                # 发送消息
                await ws.send(json.dumps({
                    "type": "text",
                    "data": {"text": user_message}
                }))
                
                # 接收 Agent 回复（可能是多条消息）
                try:
                    while True:
                        msg = await asyncio.wait_for(ws.recv(), timeout=30)
                        data = json.loads(msg)
                        if data["type"] == "agent_response":
                            msg_data = data['data']
                            total = msg_data.get('total', 1)
                            index = msg_data.get('index', 0)
                            print(f"🤖 Agent: {msg_data['text']}")
                            # 如果是最后一条消息，退出循环
                            if index >= total - 1:
                                break
                        elif data["type"] == "error":
                            print(f"❌ 错误: {data['data']['message']}")
                            break
                        else:
                            break
                except asyncio.TimeoutError:
                    print("⏰ 等待响应超时")
                
                # 短暂等待，模拟真实对话节奏
                await asyncio.sleep(1)
            
            # 结束对话
            print(f"\n{'='*60}")
            print("📭 结束对话")
            print(f"{'='*60}")
            
            await ws.send(json.dumps({
                "type": "end_conversation",
                "conversation_id": self.conversation_id
            }))
            
            try:
                # 结束时会触发 process_conversation_segments，可能需要更长时间
                msg = await asyncio.wait_for(ws.recv(), timeout=60)
                data = json.loads(msg)
                if data['type'] == 'error':
                    print(f"❌ 结束对话错误: {data['data'].get('message', 'unknown')}")
                else:
                    print(f"✅ 对话结束: {data['type']}")
            except asyncio.TimeoutError:
                print("⏰ 等待结束确认超时（但后台处理可能仍在进行）")
    
    async def wait_for_processing(self, max_wait_seconds: int = 300, check_interval: int = 3):
        """
        等待后台处理完成
        通过查询 Celery 任务状态来判断处理是否完成
        
        Args:
            max_wait_seconds: 最大等待时间（秒），默认 5 分钟
            check_interval: 检查间隔（秒）
        
        Returns:
            是否在超时前完成
        """
        print(f"\n⏳ 等待后台任务完成（最多 {max_wait_seconds} 秒）...")
        print("   提示: 通过 Celery 任务状态 API 追踪任务进度")
        
        start_time = asyncio.get_event_loop().time()
        
        while True:
            elapsed = asyncio.get_event_loop().time() - start_time
            
            if elapsed >= max_wait_seconds:
                print(f"\n⚠️  已等待 {max_wait_seconds} 秒，超时退出")
                return False
            
            # 检查任务状态
            tasks_status = await self.get_tasks_status()
            total = tasks_status.get("total", 0)
            pending = tasks_status.get("pending", 0)
            running = tasks_status.get("running", 0)
            success = tasks_status.get("success", 0)
            failure = tasks_status.get("failure", 0)
            all_completed = tasks_status.get("all_completed", False)
            
            # 同时检查章节内容
            chapters = await self.get_chapters()
            chapter_count = len(chapters)
            total_content_length = sum(len(ch.get('content', '')) for ch in chapters)
            
            status_str = f"📊 总:{total} 等待:{pending} 运行:{running} 成功:{success} 失败:{failure}"
            content_str = f"📚 章节:{chapter_count} 内容:{total_content_length}字符"
            print(f"   [{int(elapsed):3d}s] {status_str} | {content_str}")
            
            # 判断是否完成：
            # 1. 有任务且全部完成
            # 2. 或者没有任务但有章节内容（兼容旧逻辑）
            if total > 0 and all_completed:
                print(f"\n✅ 所有任务已完成！共 {total} 个任务，等待 {int(elapsed)} 秒")
                return True
            
            # 如果没有任务记录，等待一会儿任务提交
            if total == 0 and elapsed < 15:
                await asyncio.sleep(check_interval)
                continue
            
            # 如果长时间没有任务但有内容，也认为完成
            if total == 0 and chapter_count > 0 and elapsed > 30:
                print(f"\n✅ 无待处理任务，已有 {chapter_count} 个章节。等待 {int(elapsed)} 秒")
                return True
            
            await asyncio.sleep(check_interval)
    
    async def check_results(self):
        """检查回忆录生成结果"""
        print(f"\n{'='*60}")
        print("📊 检查结果")
        print(f"{'='*60}")
        
        # 等待后台处理完成（使用智能轮询）
        await self.wait_for_processing(max_wait_seconds=180, check_interval=5)
        
        # 获取回忆录状态
        print("\n📋 回忆录状态:")
        state = await self.get_memoir_state()
        print(f"  当前阶段: {state.get('current_stage', 'N/A')}")
        print(f"  已完成阶段: {state.get('covered_stages', [])}")
        
        # 显示已填充的 slots
        slots = state.get('slots', {})
        for stage, stage_slots in slots.items():
            filled = [k for k, v in stage_slots.items() if v.get('snippet')]
            if filled:
                print(f"  {stage} 已填充: {filled}")
                for slot_name in filled:
                    snippet = stage_slots[slot_name].get('snippet', '')
                    if snippet:
                        print(f"    - {slot_name}: {snippet[:50]}...")
        
        # 获取章节
        print("\n📚 生成的章节:")
        chapters = await self.get_chapters()
        if chapters:
            for ch in chapters:
                is_new = "🆕" if ch.get("is_new") else ""
                content_len = len(ch.get('content', ''))
                print(f"  {is_new} [{ch.get('category', 'N/A')}] {ch.get('title', 'N/A')} ({content_len} 字符)")
        else:
            print("  (暂无章节)")
        
        # 获取回忆录
        print("\n📖 回忆录信息:")
        book = await self.get_book()
        if "message" not in book:
            print(f"  标题: {book.get('title', 'N/A')}")
            print(f"  总字数: {book.get('total_words', 0)}")
            print(f"  有更新: {'是' if book.get('has_update') else '否'}")
        else:
            print(f"  {book.get('message', 'N/A')}")
        
        # 显示回忆录完整内容
        if chapters:
            print(f"\n{'='*60}")
            print("📜 回忆录完整内容")
            print(f"{'='*60}")
            for ch in chapters:
                category = ch.get('category', 'N/A')
                title = ch.get('title', '未命名章节')
                content = ch.get('content', '')
                
                print(f"\n{'─'*60}")
                print(f"【{title}】({category})")
                print(f"{'─'*60}")
                if content:
                    print(content)
                else:
                    print("(暂无内容)")
            print(f"\n{'='*60}")


async def main():
    """主函数"""
    print("=" * 60)
    print("🎭 Life Echo 多轮对话测试")
    print(f"⏰ 开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 60)
    
    tester = ConversationTester()
    
    try:
        # 1. 注册/登录
        await tester.register_or_login()
        
        # 2. 清除旧的任务记录
        await tester.clear_tasks()
        print("\n🧹 已清除旧的任务记录")
        
        # 3. 查看初始状态
        print("\n📋 初始回忆录状态:")
        state = await tester.get_memoir_state()
        print(f"  当前阶段: {state.get('current_stage', 'N/A')}")
        
        # 4. 运行多轮对话
        await tester.run_conversation()
        
        # 5. 检查结果
        await tester.check_results()
        
    except Exception as e:
        print(f"\n❌ 测试失败: {e}")
        import traceback
        traceback.print_exc()
    
    print("\n" + "=" * 60)
    print(f"⏰ 结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 60)


if __name__ == "__main__":
    asyncio.run(main())