资料整合模块测试脚本

tests/test_integration_simple.py

@@ -0,0 +1,688 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""
+简化版离线整合测试脚本
+使用已有数据测试integration模块的不同效果
+调用真实AI进行内容整合
+"""
+
+import os
+import json
+import asyncio
+from typing import Dict, List, Optional, Any
+from datetime import datetime
+from pathlib import Path
+from dataclasses import dataclass
+
+# 添加项目根目录到路径
+import sys
+project_root = Path(__file__).parent
+sys.path.insert(0, str(project_root))
+
+# 导入核心模块
+from core.ai.ai_agent import AIAgent
+from core.config import ConfigManager, AIModelConfig
+from utils.prompts import PromptTemplate
+
+@dataclass
+class SimpleNote:
+    """简化的笔记数据模型"""
+    note_id: str
+    title: str
+    content: str
+    author: str
+    likes: int
+    comments: int
+    shares: int
+    note_url: str
+    images: List[str]
+    
+@dataclass 
+class SimpleSearchResult:
+    """简化的搜索结果"""
+    keyword: str
+    notes: List[SimpleNote]
+    total_count: int
+    success: bool
+
+class OfflineIntegrationTester:
+    """离线整合测试器"""
+    
+    def __init__(self, data_file: str):
+        """
+        初始化测试器
+        
+        Args:
+            data_file: 离线数据文件路径
+        """
+        self.data_file = data_file
+        self.raw_data = None
+        self.processed_notes = []
+        
+        # 初始化配置管理器和AI代理
+        self.config_manager = ConfigManager()
+        self.config_manager.load_from_directory("config", server_mode=True)
+        
+        # 获取AI配置并初始化AI代理
+        ai_config = self.config_manager.get_config('ai_model', AIModelConfig)
+        self.ai_agent = AIAgent(ai_config)
+        
+        # 加载integration prompt模板
+        self.prompt_template = PromptTemplate(
+            system_prompt_path="resource/prompt/integration/system.txt",
+            user_prompt_path="resource/prompt/integration/user.txt"
+        )
+        
+        # 加载数据
+        self._load_data()
+        
+        print(f"✅ 离线测试器初始化完成，加载了 {len(self.processed_notes)} 条笔记数据")
+        print(f"🤖 AI代理已初始化，模型: {ai_config.model}")
+    
+    def _load_data(self):
+        """加载离线数据"""
+        try:
+            with open(self.data_file, 'r', encoding='utf-8') as f:
+                self.raw_data = json.load(f)
+            
+            # 转换为SimpleNote对象
+            if 'results' in self.raw_data and self.raw_data['results']:
+                result = self.raw_data['results'][0]  # 取第一个搜索结果
+                
+                for note_data in result.get('notes', []):
+                    note = SimpleNote(
+                        note_id=note_data.get('note_id', ''),
+                        title=note_data.get('title', ''),
+                        content=note_data.get('content', ''),
+                        author=note_data.get('author', ''),
+                        likes=note_data.get('like_count', 0),
+                        comments=note_data.get('comment_count', 0),
+                        shares=note_data.get('share_count', 0),
+                        note_url=note_data.get('note_url', ''),
+                        images=note_data.get('images', [])
+                    )
+                    self.processed_notes.append(note)
+                    
+        except Exception as e:
+            print(f"❌ 数据加载失败: {e}")
+            self.processed_notes = []
+    
+    def get_search_result(self, keyword: str = None) -> SimpleSearchResult:
+        """获取搜索结果"""
+        if not keyword:
+            keyword = self.raw_data['results'][0]['keyword'] if self.raw_data else '测试关键词'
+        
+        return SimpleSearchResult(
+            keyword=keyword,
+            notes=self.processed_notes,
+            total_count=len(self.processed_notes),
+            success=True
+        )
+    
+    def create_mock_documents(self, content_type: str = "travel_guide") -> Dict[str, str]:
+        """创建模拟文档内容"""
+        
+        mock_docs = {
+            "travel_guide": {
+                "上海馥桂萌宠园游玩攻略.txt": """上海馥桂萌宠园游玩攻略
+
+一、园区简介
+馥桂萌宠园位于上海市嘉定区，是一个集萌宠互动、户外休闲、亲子娱乐为一体的综合性主题公园。
+
+二、开放信息
+- 开放时间：9:00-17:00（周一至周日）
+- 门票价格：成人票80元，儿童票50元（1.2米以下免费）
+- 地址：上海市嘉定区朱桥镇
+- 交通：地铁11号线嘉定西站转公交
+
+三、主要区域
+1. 萌宠互动区：可与羊驼、袋鼠、小兔子等动物亲密接触
+2. 户外游乐区：秋千、滑梯等儿童设施
+3. 餐饮休息区：提供简餐和饮品
+4. 桂花林区：春秋季节桂花飘香，适合拍照
+
+四、游玩建议
+- 建议游玩时间：3-4小时
+- 最佳游玩季节：春秋两季
+- 携带物品：防晒用品、湿巾、零食
+- 注意事项：爱护动物，文明游园""",
+                
+                "萌宠园交通指南.md": """# 馥桂萌宠园交通指南
+
+## 公共交通
+### 地铁+公交
+1. 地铁11号线至嘉定西站
+2. 转乘嘉定69路公交至朱桥站
+3. 步行约10分钟即可到达
+
+## 自驾路线
+从市区出发：
+1. 走外环高速(A20)
+2. 转嘉定环线(A5)
+3. 在朱桥出口下高速
+4. 沿指示牌行驶约5分钟
+
+## 停车信息
+- 园区提供免费停车场
+- 停车位约200个
+- 节假日建议早到，避免停车位紧张"""
+            },
+            
+            "product_info": {
+                "门票套餐信息.txt": """门票套餐信息
+
+基础门票：
+- 成人票：80元/人
+- 儿童票：50元/人（3-12岁）
+- 老年票：60元/人（65岁以上）
+- 免票：1.2米以下儿童
+
+套餐选择：
+1. 家庭套票（2大1小）：180元
+2. 亲子套票（1大1小）：120元  
+3. 团体票（10人以上）：70元/人
+
+增值服务：
+- 动物喂食包：20元/份
+- 拍照服务：50元/次
+- 导览服务：100元/团"""
+            },
+            
+            "user_reviews": {
+                "游客评价汇总.txt": """游客评价汇总
+
+正面评价：
+✅ 动物种类丰富，互动性强
+✅ 环境优美，适合拍照
+✅ 工作人员服务态度好
+✅ 停车方便，交通便利
+✅ 价格合理，性价比高
+
+负面反馈：
+❌ 节假日人流量大
+❌ 餐饮选择相对有限
+❌ 部分设施需要维护
+❌ 雨天游玩体验一般
+
+综合评分：4.2/5.0"""
+            }
+        }
+        
+        return mock_docs.get(content_type, mock_docs["travel_guide"])
+    
+    def _format_xhs_notes(self, notes: List[SimpleNote]) -> str:
+        """格式化小红书笔记内容"""
+        if not notes:
+            return "暂无相关笔记内容"
+        
+        formatted_notes = []
+        for i, note in enumerate(notes, 1):
+            note_content = f"""第{i}条笔记：
+标题：{note.title}
+作者：{note.author}
+内容：{note.content if note.content else '（无文字内容）'}
+互动数据：👍{note.likes} 💬{note.comments} 🔄{note.shares}
+笔记链接：{note.note_url}
+"""
+            if note.images:
+                note_content += f"图片数量：{len(note.images)}张\n"
+            
+            formatted_notes.append(note_content)
+        
+        return "\n".join(formatted_notes)
+    
+    def _format_document_content(self, documents: Dict[str, str]) -> str:
+        """格式化文档内容"""
+        if not documents:
+            return "暂无文档内容"
+        
+        formatted_docs = []
+        for filename, content in documents.items():
+            doc_content = f"""文档：{filename}
+内容：
+{content}
+"""
+            formatted_docs.append(doc_content)
+        
+        return "\n".join(formatted_docs)
+    
+    def test_basic_integration(self):
+        """测试基础整合功能"""
+        print("\n" + "="*60)
+        print("🔄 测试基础整合功能")
+        print("="*60)
+        
+        # 获取数据
+        search_result = self.get_search_result("上海馥桂萌宠园攻略")
+        documents = self.create_mock_documents("travel_guide")
+        
+        print(f"📚 文档处理结果：")
+        print(f"   - 文档数量：{len(documents)}")
+        total_doc_length = sum(len(content) for content in documents.values())
+        print(f"   - 总内容长度：{total_doc_length}")
+        
+        print(f"\n📱 小红书数据：")
+        print(f"   - 笔记数量：{len(search_result.notes)}")
+        print(f"   - 总点赞数：{sum(note.likes for note in search_result.notes)}")
+        print(f"   - 总评论数：{sum(note.comments for note in search_result.notes)}")
+        
+        # 显示部分内容示例
+        print(f"\n📝 内容示例：")
+        first_doc = list(documents.values())[0]
+        print(f"   文档摘要：{first_doc[:100]}...")
+        
+        if search_result.notes:
+            popular_note = max(search_result.notes, key=lambda x: x.likes)
+            print(f"   热门笔记：{popular_note.title} (👍{popular_note.likes})")
+        
+        return {
+            "search_result": search_result,
+            "documents": documents
+        }
+    
+    def test_different_document_types(self):
+        """测试不同文档类型的整合效果"""
+        print("\n" + "="*60)
+        print("📊 测试不同文档类型的整合效果")
+        print("="*60)
+        
+        doc_types = ["travel_guide", "product_info", "user_reviews"]
+        results = {}
+        
+        for doc_type in doc_types:
+            print(f"\n🔸 处理文档类型：{doc_type}")
+            
+            documents = self.create_mock_documents(doc_type)
+            total_length = sum(len(content) for content in documents.values())
+            
+            results[doc_type] = {
+                "document_count": len(documents),
+                "content_length": total_length,
+                "files": list(documents.keys())
+            }
+            
+            print(f"   ✅ 文档数量：{results[doc_type]['document_count']}")
+            print(f"   ✅ 内容长度：{results[doc_type]['content_length']}")
+            print(f"   ✅ 文件列表：{results[doc_type]['files']}")
+        
+        return results
+    
+    def test_content_filtering(self):
+        """测试内容过滤和筛选"""
+        print("\n" + "="*60)
+        print("🔍 测试内容过滤和筛选")
+        print("="*60)
+        
+        search_result = self.get_search_result()
+        
+        # 按点赞数筛选
+        high_quality_notes = [note for note in search_result.notes if note.likes >= 50]
+        print(f"📈 高质量笔记（点赞≥50）：{len(high_quality_notes)} 条")
+        
+        # 按评论数筛选
+        interactive_notes = [note for note in search_result.notes if note.comments >= 10]
+        print(f"💬 高互动笔记（评论≥10）：{len(interactive_notes)} 条")
+        
+        # 按标题关键词筛选
+        guide_notes = [note for note in search_result.notes if '攻略' in note.title or '指南' in note.title]
+        print(f"📋 攻略类笔记：{len(guide_notes)} 条")
+        
+        # 显示统计信息
+        if search_result.notes:
+            avg_likes = sum(note.likes for note in search_result.notes) / len(search_result.notes)
+            avg_comments = sum(note.comments for note in search_result.notes) / len(search_result.notes)
+            
+            print(f"\n📊 统计信息：")
+            print(f"   平均点赞数：{avg_likes:.1f}")
+            print(f"   平均评论数：{avg_comments:.1f}")
+            
+            # 找出最受欢迎的笔记
+            most_liked = max(search_result.notes, key=lambda x: x.likes)
+            most_commented = max(search_result.notes, key=lambda x: x.comments)
+            
+            print(f"\n🏆 最受欢迎：")
+            print(f"   最多点赞：《{most_liked.title}》- {most_liked.likes} 赞")
+            print(f"   最多评论：《{most_commented.title}》- {most_commented.comments} 评论")
+        
+        return {
+            "total_notes": len(search_result.notes),
+            "high_quality_notes": len(high_quality_notes),
+            "interactive_notes": len(interactive_notes), 
+            "guide_notes": len(guide_notes)
+        }
+    
+    async def test_real_ai_integration(self):
+        """测试真实AI整合功能"""
+        print("\n" + "="*60)
+        print("🤖 测试真实AI整合功能")
+        print("="*60)
+        
+        try:
+            # 准备数据
+            search_result = self.get_search_result()
+            documents = self.create_mock_documents("travel_guide")
+            
+            print("🔄 正在准备数据...")
+            
+            # 格式化文档内容
+            document_content = self._format_document_content(documents)
+            
+            # 格式化小红书笔记内容
+            xhs_content = self._format_xhs_notes(search_result.notes)
+            
+            # 构建prompt
+            system_prompt = self.prompt_template.get_system_prompt()
+            user_prompt = self.prompt_template.build_user_prompt(
+                keywords=search_result.keyword or "馥桂萌宠园",
+                document_content=document_content or "暂无文档内容",
+                xhs_notes_content=xhs_content or "暂无笔记内容"
+            )
+            
+            print(f"📝 System Prompt长度: {len(system_prompt)} 字符")
+            print(f"📝 User Prompt长度: {len(user_prompt)} 字符")
+            
+            # 调用真实AI
+            print("🔄 正在调用AI进行内容整合...")
+            start_time = datetime.now()
+            
+            response_text, input_tokens, output_tokens, time_cost = await self.ai_agent.generate_text(
+                system_prompt=system_prompt,
+                user_prompt=user_prompt,
+                use_stream=True,
+                stage="content_integration",
+            )
+            
+            end_time = datetime.now()
+            
+            print("✅ AI整合完成")
+            print(f"📄 生成内容长度：{len(response_text)} 字符")
+            print(f"🎯 输入Token数：{input_tokens}")
+            print(f"🎯 输出Token数：{output_tokens}")
+            print(f"⏱️ 处理时间：{time_cost:.2f} 秒")
+            
+            # 尝试解析JSON
+            try:
+                # 使用file_io模块的JSON处理功能
+                from utils.file_io import process_llm_json_text
+                parsed_json = process_llm_json_text(response_text)
+                
+                if parsed_json:
+                    print("✅ JSON解析成功")
+                    print(f"📊 解析结果包含 {len(parsed_json)} 个顶级字段")
+                    
+                    # 显示解析后的结构
+                    if isinstance(parsed_json, dict):
+                        print("🏗️ 数据结构：")
+                        for key, value in parsed_json.items():
+                            if isinstance(value, dict):
+                                print(f"   {key}: 包含 {len(value)} 个子字段")
+                            elif isinstance(value, list):
+                                print(f"   {key}: 列表，包含 {len(value)} 个项目")
+                            else:
+                                print(f"   {key}: {type(value).__name__}")
+                    
+                    formatted_response = json.dumps(parsed_json, ensure_ascii=False, indent=2)
+                else:
+                    print("⚠️ JSON解析失败，使用原始响应")
+                    formatted_response = response_text
+                    
+            except Exception as parse_error:
+                print(f"⚠️ JSON解析出错: {parse_error}")
+                formatted_response = response_text
+            
+            print(f"\n📝 AI响应预览：")
+            preview_text = formatted_response[:500] + "..." if len(formatted_response) > 500 else formatted_response
+            print(preview_text)
+            
+            return {
+                "ai_response": formatted_response,
+                "content_length": len(response_text),
+                "input_tokens": input_tokens,
+                "output_tokens": output_tokens,
+                "processing_time": time_cost,
+                "source_notes": len(search_result.notes),
+                "source_docs": len(documents),
+                "json_parsed": parsed_json is not None
+            }
+            
+        except Exception as e:
+            print(f"❌ AI整合测试失败：{e}")
+            import traceback
+            traceback.print_exc()
+            return None
+    
+    def test_export_formats(self):
+        """测试不同导出格式"""
+        print("\n" + "="*60)
+        print("📤 测试不同导出格式")
+        print("="*60)
+        
+        search_result = self.get_search_result()
+        documents = self.create_mock_documents("travel_guide")
+        
+        # 模拟不同格式的导出
+        formats = {
+            "summary": "简要摘要格式",
+            "blog_post": "博客文章格式", 
+            "travel_guide": "旅游攻略格式",
+            "product_sales": "产品销售格式",
+            "attraction_standard": "景点标准格式"
+        }
+        
+        export_results = {}
+        
+        for format_type, description in formats.items():
+            print(f"🔸 生成{description}...")
+            
+            # 模拟不同格式的内容生成
+            mock_content = self._generate_format_content(format_type, search_result, documents)
+            
+            export_results[format_type] = {
+                "content": mock_content,
+                "length": len(mock_content),
+                "format": format_type
+            }
+            
+            print(f"   ✅ 长度：{len(mock_content)} 字符")
+        
+        return export_results
+    
+    def _generate_format_content(self, format_type: str, search_result: SimpleSearchResult, documents: Dict[str, str]) -> str:
+        """根据格式类型生成相应内容"""
+        
+        base_info = f"基于{len(search_result.notes)}篇小红书游记和官方资料"
+        
+        if format_type == "summary":
+            return f"【简要摘要】{base_info}，馥桂萌宠园是嘉定区优质亲子游目的地，特色萌宠互动体验，适合周末家庭出游。"
+        
+        elif format_type == "blog_post":
+            return f"""# 周末带娃新选择 | 上海馥桂萌宠园深度体验
+
+今天想和大家分享一个超棒的亲子游目的地——位于嘉定的馥桂萌宠园！{base_info}，这里真的是遛娃神器！
+
+## 为什么推荐这里？
+🦘 可以近距离接触羊驼、袋鼠等萌宠
+🌸 环境优美，特别是桂花季节
+👨‍👩‍👧‍👦 设施完善，非常适合家庭游
+
+详细攻略请看下文..."""
+
+        elif format_type == "travel_guide":
+            return f"""## 上海馥桂萌宠园游玩攻略
+
+### 基本信息
+- 地址：上海市嘉定区朱桥镇
+- 门票：成人80元，儿童50元
+- 开放时间：9:00-17:00
+
+### 交通指南
+地铁11号线嘉定西站转公交，或自驾直达
+
+### 游玩亮点
+{base_info}，园区主要特色包括萌宠互动、户外游乐、休闲拍照等。
+
+### 注意事项
+建议游玩3-4小时，春秋季节体验最佳。"""
+
+        elif format_type == "product_sales":
+            return f"""🔥【限时特惠】上海馥桂萌宠园门票
+
+⭐ 产品亮点：
+• {base_info}，口碑爆棚！
+• 🦘 独特萌宠互动体验
+• 🌸 网红拍照打卡圣地  
+• 👨‍👩‍👧‍👦 亲子游首选目的地
+
+💰 特价套餐：
+原价130元，现价只要99元！
+包含：成人门票1张 + 动物喂食包1份
+
+📞 立即预订：xxx-xxxx-xxxx"""
+
+        else:  # attraction_standard
+            return f"""景点名称：上海馥桂萌宠园
+景点类型：主题公园/动物园
+地理位置：上海市嘉定区朱桥镇
+开放状态：正常营业
+门票价格：80-50元
+适宜人群：亲子家庭
+推荐指数：4.2/5.0
+游玩时长：3-4小时
+最佳季节：春秋两季
+
+{base_info}，具有良好的口碑和游客体验。"""
+    
+    def test_parameter_effects(self):
+        """测试不同参数的效果"""
+        print("\n" + "="*60)
+        print("⚙️ 测试不同参数的效果")
+        print("="*60)
+        
+        search_result = self.get_search_result()
+        
+        # 测试不同的过滤阈值
+        thresholds = [
+            {"likes": 20, "comments": 5},
+            {"likes": 50, "comments": 10},
+            {"likes": 100, "comments": 20}
+        ]
+        
+        print("🔧 测试不同过滤阈值的效果：")
+        for i, threshold in enumerate(thresholds, 1):
+            high_likes = [note for note in search_result.notes if note.likes >= threshold["likes"]]
+            high_comments = [note for note in search_result.notes if note.comments >= threshold["comments"]]
+            
+            print(f"   阈值{i}（点赞≥{threshold['likes']}, 评论≥{threshold['comments']}）:")
+            print(f"     - 符合点赞条件：{len(high_likes)} 条")
+            print(f"     - 符合评论条件：{len(high_comments)} 条")
+        
+        # 测试不同的内容长度限制
+        length_limits = [100, 200, 500]
+        
+        print(f"\n📏 测试不同内容长度限制的效果：")
+        for limit in length_limits:
+            documents = self.create_mock_documents("travel_guide")
+            truncated_docs = {}
+            for filename, content in documents.items():
+                truncated_docs[filename] = content[:limit] + "..." if len(content) > limit else content
+            
+            total_length = sum(len(content) for content in truncated_docs.values())
+            print(f"   限制{limit}字符：总长度 {total_length} 字符")
+        
+        return {
+            "threshold_tests": len(thresholds),
+            "length_tests": len(length_limits)
+        }
+
+async def main():
+    """主函数"""
+    print("🚀 启动真实AI离线整合测试...")
+    
+    # 检查数据文件
+    data_file = "batch_search_20250717_104407.json"
+    if not os.path.exists(data_file):
+        print(f"❌ 数据文件不存在：{data_file}")
+        return
+    
+    # 初始化测试器
+    try:
+        tester = OfflineIntegrationTester(data_file)
+    except Exception as e:
+        print(f"❌ 测试器初始化失败：{e}")
+        import traceback
+        traceback.print_exc()
+        return
+    
+    try:
+        # 执行各种测试
+        print(f"\n{'='*60}")
+        print("🧪 开始执行真实AI离线整合测试套件")
+        print(f"{'='*60}")
+        
+        # 1. 基础整合测试
+        basic_result = tester.test_basic_integration()
+        
+        # 2. 不同文档类型测试
+        doc_type_results = tester.test_different_document_types()
+        
+        # 3. 内容过滤测试
+        filter_results = tester.test_content_filtering()
+        
+        # 4. 真实AI整合测试（异步）
+        ai_results = await tester.test_real_ai_integration()
+        with open("ai_results.json", "w") as f:
+            json.dump(ai_results, f, ensure_ascii=False, indent=2)
+        # 5. 导出格式测试
+        export_results = tester.test_export_formats()
+        
+        # 6. 参数效果测试
+        param_results = tester.test_parameter_effects()
+        
+        # 输出测试总结
+        print(f"\n{'='*60}")
+        print("📋 测试结果总结")
+        print(f"{'='*60}")
+        
+        print(f"✅ 基础整合测试：成功")
+        print(f"✅ 文档类型测试：处理了 {len(doc_type_results)} 种类型")
+        print(f"✅ 内容过滤测试：识别出 {filter_results['high_quality_notes']} 条高质量笔记")
+        
+        if ai_results:
+            print(f"✅ 真实AI整合测试：成功")
+            print(f"   - 处理时间：{ai_results['processing_time']:.2f}秒")
+            print(f"   - 输入Token：{ai_results['input_tokens']}")
+            print(f"   - 输出Token：{ai_results['output_tokens']}")
+            print(f"   - JSON解析：{'成功' if ai_results['json_parsed'] else '失败'}")
+        else:
+            print(f"❌ 真实AI整合测试：失败")
+        
+        # print(f"✅ 导出格式测试：生成了 {len(export_results)} 种格式")
+        # print(f"✅ 参数效果测试：测试了 {param_results['threshold_tests']} 种阈值配置")
+        
+        # print(f"\n🎉 所有测试完成！")
+        
+        # # 提供测试建议
+        # print(f"\n💡 测试建议：")
+        # print(f"   1. 可以调整AI模型参数（temperature、top_p等）来优化输出质量")
+        # print(f"   2. 可以修改integration prompt模板来改进整合效果")
+        # print(f"   3. 可以测试不同的文档类型组合")
+        # print(f"   4. 可以调整内容过滤的阈值")
+        # print(f"   5. 可以测试不同的导出格式效果")
+        
+        # # 显示原始数据概览
+        # if tester.raw_data:
+        #     summary = tester.raw_data.get('summary', {})
+        #     print(f"\n📊 原始数据概览：")
+        #     print(f"   - 总关键词数：{summary.get('total_keywords', 0)}")
+        #     print(f"   - 成功搜索数：{summary.get('successful_searches', 0)}")
+        #     print(f"   - 总笔记数：{summary.get('total_notes', 0)}")
+        #     print(f"   - 总互动数：{summary.get('total_interactions', 0)}")
+        
+    except Exception as e:
+        print(f"❌ 测试过程中发生错误：{e}")
+        import traceback
+        traceback.print_exc()
+
+if __name__ == "__main__":
+    asyncio.run(main())