148 lines
7.1 KiB
Python
148 lines
7.1 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import requests
|
|
import json
|
|
from utils.file_io import process_llm_json_text
|
|
|
|
def test_final_integration():
|
|
"""最终整合测试 - 展示完整的纯搜索整合功能"""
|
|
|
|
url = "http://localhost:2714/api/v1/content-integration/integrate"
|
|
|
|
# 测试数据 - 使用平级字段结构
|
|
test_data = {
|
|
"keywords": ["馥桂萌宠园攻略"],
|
|
"cookies": "abRequestId=873258ba-49fe-530b-9c07-529e7871508d; webBuild=4.72.0; xsecappid=xhs-pc-web; a1=19808118ec0gxnuh5mtpv8o8aepgp2m65bpbtiizn30000193947; webId=3f4bd516682e25b71cdd139a11b4d896; web_session=040069b295652fcb5b6e1389413a4be4606547",
|
|
# 搜索配置参数（平级）
|
|
"sort_type": 2, # 最多点赞
|
|
"note_type": 2, # 普通笔记
|
|
"note_time": 3, # 半年内
|
|
"note_range": 0, # 不限范围
|
|
"pos_distance": 0, # 不限位置
|
|
"query_num": 20 # 20条笔记快速测试
|
|
}
|
|
|
|
print("🎯 旅游内容整合系统 - 最终测试")
|
|
print("=" * 50)
|
|
print(f"📍 目标关键词: {', '.join(test_data['keywords'])}")
|
|
print(f"📊 搜索配置: 最多点赞排序，{test_data['query_num']}条笔记")
|
|
print(f"🔄 模式: 纯搜索整合（无文档上传）")
|
|
print()
|
|
|
|
try:
|
|
print("⏳ 开始内容整合...")
|
|
response = requests.post(
|
|
url,
|
|
json=test_data,
|
|
headers={"Content-Type": "application/json"},
|
|
timeout=120
|
|
)
|
|
|
|
print(f"📊 响应状态: {response.status_code}")
|
|
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
print("✅ API调用成功!")
|
|
|
|
if result.get('success'):
|
|
print("🎉 内容整合成功!")
|
|
|
|
# 显示处理统计
|
|
processing_time = result.get('processing_time', 'N/A')
|
|
timestamp = result.get('timestamp', 'N/A')
|
|
print(f"⏱️ 处理时间: {processing_time}")
|
|
print(f"🕒 完成时间: {timestamp}")
|
|
|
|
# 显示输入统计
|
|
input_summary = result.get('input_summary', {})
|
|
print(f"📄 处理文档: {input_summary.get('document_count', 0)} 个")
|
|
print(f"📝 XHS笔记: {input_summary.get('xhs_notes_count', 0)} 条")
|
|
|
|
# 显示XHS信息
|
|
xhs_info = result.get('xhs_info', {})
|
|
total_interactions = xhs_info.get('total_interactions', 0)
|
|
authors = xhs_info.get('authors', [])
|
|
print(f"👥 涉及作者: {len(authors)} 位")
|
|
print(f"💬 总互动数: {total_interactions} (点赞+评论+分享)")
|
|
|
|
# 解析整合内容
|
|
content = result.get('integrated_content', '')
|
|
if content:
|
|
print("\n🔍 内容解析结果:")
|
|
try:
|
|
# 使用file_io模块解析JSON
|
|
parsed_content = process_llm_json_text(content)
|
|
|
|
if parsed_content and isinstance(parsed_content, dict):
|
|
attractions = parsed_content.get('attractions', [])
|
|
print(f"✅ 成功解析JSON格式内容")
|
|
print(f"🏞️ 识别景区数量: {len(attractions)}")
|
|
|
|
# 显示详细信息
|
|
for i, attraction in enumerate(attractions, 1):
|
|
name = attraction.get('name', 'N/A')
|
|
products = attraction.get('products', [])
|
|
print(f"\n📍 景区 {i}: {name}")
|
|
print(f" 📦 产品数量: {len(products)}")
|
|
|
|
for j, product in enumerate(products, 1):
|
|
print(f"\n 🎫 产品 {j}: {product.get('product_name', 'N/A')}")
|
|
print(f" 💰 价格: {product.get('price', 'N/A')}")
|
|
|
|
# 显示优势
|
|
advantages = product.get('key_advantages', '')
|
|
if advantages:
|
|
preview = advantages[:80] + "..." if len(advantages) > 80 else advantages
|
|
print(f" 🎯 核心优势: {preview}")
|
|
|
|
# 显示交通信息
|
|
transport = product.get('transportation', {})
|
|
if isinstance(transport, dict):
|
|
address = transport.get('address', 'N/A')
|
|
guide = transport.get('guide', 'N/A')
|
|
print(f" 📍 地址: {address}")
|
|
if guide and guide != 'N/A':
|
|
guide_preview = guide[:60] + "..." if len(guide) > 60 else guide
|
|
print(f" 🚗 交通: {guide_preview}")
|
|
|
|
# 显示详细描述数量
|
|
descriptions = product.get('detailed_description', [])
|
|
if descriptions:
|
|
print(f" 📝 详细说明: {len(descriptions)} 条")
|
|
else:
|
|
print("❌ 内容解析失败或格式不正确")
|
|
print(f"📄 原始内容预览: {content[:200]}...")
|
|
|
|
except Exception as e:
|
|
print(f"❌ 内容解析异常: {e}")
|
|
print(f"📄 原始内容预览: {content[:200]}...")
|
|
|
|
# 显示输出文件
|
|
output_file = result.get('output_file')
|
|
if output_file:
|
|
print(f"\n💾 结果已保存至: {output_file}")
|
|
|
|
else:
|
|
error_msg = result.get('error_message', 'Unknown error')
|
|
print(f"❌ 内容整合失败: {error_msg}")
|
|
|
|
else:
|
|
print("❌ API调用失败")
|
|
try:
|
|
error_info = response.json()
|
|
print(f"📄 错误详情: {json.dumps(error_info, ensure_ascii=False, indent=2)}")
|
|
except:
|
|
print(f"📄 错误响应: {response.text}")
|
|
|
|
except requests.exceptions.Timeout:
|
|
print("⏰ 请求超时 - 内容整合需要较长时间，请稍候")
|
|
|
|
except Exception as e:
|
|
print(f"❌ 测试异常: {e}")
|
|
|
|
print("\n" + "=" * 50)
|
|
print("🏁 测试完成")
|
|
|
|
if __name__ == "__main__":
|
|
test_final_integration() |