#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 完整业务流程测试 测试选题生成 → 内容生成 → 海报生成的完整链路 """ import requests import json import time from typing import Dict, Any, List BASE_URL = "http://localhost:8001" # ========== 测试数据 ========== SCENIC_SPOT = { "id": 1, "name": "天津冒险湾", "description": "天津最大的水上乐园，拥有多种刺激水上项目和亲子设施，包括水上过山车、漂流河、儿童戏水区、海浪池等。园区占地面积大，设施齐全，是夏季避暑戏水的绝佳去处。", "address": "天津市滨海新区海滨大道168号", "location": "天津市", "traffic_info": "地铁9号线直达，自驾可走津滨高速", "highlights": ["水上过山车", "儿童戏水区", "漂流河", "海浪池", "温泉SPA"], "opening_hours": "09:00-18:00 (夏季延长至21:00)", "ticket_info": "成人票 199 元，儿童票 99 元", "tips": "建议自带泳衣，园区内也有售卖；建议避开周末高峰；可提前在线购票享优惠" } PRODUCT = { "id": 10, "name": "家庭套票", "price": 299, "original_price": 399, "description": "含2大1小门票，赠送储物柜一个，适合三口之家出游", "includes": ["2张成人票", "1张儿童票", "储物柜1个", "免费停车"], "valid_period": "2025-01-01 至 2025-03-31", "usage_rules": "需提前1天预约，入园当日有效，不可退改" } STYLE = { "id": "gonglue", "name": "攻略风" } AUDIENCE = { "id": "qinzi", "name": "亲子向" } def print_section(title: str): print("\n" + "=" * 60) print(f" {title}") print("=" * 60) def print_json(data: Any, title: str = None): if title: print(f"\n{title}:") print(json.dumps(data, ensure_ascii=False, indent=2)) def test_health(): """测试服务健康状态""" print_section("1. 健康检查") resp = requests.get(f"{BASE_URL}/") print(f"状态码: {resp.status_code}") print(f"响应: {resp.json()}") return resp.status_code == 200 def test_config_api(): """测试配置 API""" print_section("2. 配置 API") # 风格 resp = requests.get(f"{BASE_URL}/api/v2/aigc/config/styles") styles = resp.json() print(f"风格列表: {[s['name'] for s in styles['styles']]}") # 人群 resp = requests.get(f"{BASE_URL}/api/v2/aigc/config/audiences") audiences = resp.json() print(f"人群列表: {[a['name'] for a in audiences['audiences']]}") return True def test_topic_generate(): """测试选题生成""" print_section("3. 选题生成") request_body = { "engine": "topic_generate", "params": { "month": "2025-01", "num_topics": 3, "scenic_spot": SCENIC_SPOT, "product": PRODUCT, "style": STYLE, "audience": AUDIENCE }, "async_mode": False } print("\n请求参数:") print(f" 景区: {SCENIC_SPOT['name']}") print(f" 产品: {PRODUCT['name']}") print(f" 风格: {STYLE['name']}") print(f" 人群: {AUDIENCE['name']}") print(f" 数量: 3") print("\n正在调用 LLM 生成选题... (可能需要 30-60 秒)") start_time = time.time() resp = requests.post( f"{BASE_URL}/api/v2/aigc/execute", json=request_body, timeout=120 ) elapsed = time.time() - start_time result = resp.json() print(f"\n耗时: {elapsed:.1f} 秒") print(f"成功: {result.get('success')}") if result.get('success'): data = result.get('data', {}) topics = data.get('topics', []) print(f"生成选题数: {len(topics)}") for i, topic in enumerate(topics, 1): print(f"\n选题 {i}:") print(f" 日期: {topic.get('date', 'N/A')}") print(f" 标题: {topic.get('title', topic.get('object', 'N/A'))}") print(f" 逻辑: {topic.get('logic', 'N/A')[:50]}...") return topics else: print(f"错误: {result.get('error')}") return None def test_content_generate(topic: Dict[str, Any]): """测试内容生成""" print_section("4. 内容生成") request_body = { "engine": "content_generate", "params": { "topic": topic, "scenic_spot": SCENIC_SPOT, "product": PRODUCT, "style": STYLE, "audience": AUDIENCE, "enable_judge": False # 先不审核，加快测试 }, "async_mode": False } print(f"\n选题: {topic.get('title', topic.get('object', 'N/A'))}") print("\n正在调用 LLM 生成内容... (可能需要 30-60 秒)") start_time = time.time() resp = requests.post( f"{BASE_URL}/api/v2/aigc/execute", json=request_body, timeout=180 ) elapsed = time.time() - start_time result = resp.json() print(f"\n耗时: {elapsed:.1f} 秒") print(f"成功: {result.get('success')}") if result.get('success'): data = result.get('data', {}) content = data.get('content', {}) print(f"\n生成内容:") print(f" 标题: {content.get('title', 'N/A')}") print(f" 正文长度: {len(content.get('content', ''))} 字符") print(f" 标签: {content.get('tag', 'N/A')}") # 显示正文预览 body = content.get('content', '') if body: print(f"\n正文预览:") print(f" {body[:200]}...") return content else: print(f"错误: {result.get('error')}") return None def test_poster_generate(content: Dict[str, Any]): """测试海报生成""" print_section("5. 海报生成") request_body = { "engine": "poster_generate", "params": { "template_id": "vibrant", "content": { "title": content.get('title', '寒假特惠'), "content": content.get('content', ''), "tag": content.get('tag', ''), "subtitle": f"{SCENIC_SPOT['name']} {PRODUCT['name']}", "price": str(PRODUCT['price']), "original_price": str(PRODUCT['original_price']), "slogan": "限时优惠，错过等一年！" }, "scenic_spot": SCENIC_SPOT, "product": PRODUCT, "image_urls": [] # 暂无图片 }, "async_mode": False } print(f"\n海报内容:") print(f" 标题: {request_body['params']['content']['title']}") print(f" 副标题: {request_body['params']['content']['subtitle']}") print(f" 价格: {request_body['params']['content']['price']}") print("\n正在生成海报...") start_time = time.time() resp = requests.post( f"{BASE_URL}/api/v2/aigc/execute", json=request_body, timeout=180 ) elapsed = time.time() - start_time result = resp.json() print(f"\n耗时: {elapsed:.1f} 秒") print(f"成功: {result.get('success')}") if result.get('success'): data = result.get('data', {}) print(f"海报路径: {data.get('poster_path', 'N/A')}") return data else: print(f"错误: {result.get('error')}") return None def main(): print("\n" + "=" * 60) print(" TravelContentCreator 完整业务流程测试") print("=" * 60) print(f" 服务地址: {BASE_URL}") print(f" 时间: {time.strftime('%Y-%m-%d %H:%M:%S')}") # 1. 健康检查 if not test_health(): print("\n❌ 服务不可用，请先启动服务") return # 2. 配置 API test_config_api() # 3. 选题生成 topics = test_topic_generate() if not topics: print("\n❌ 选题生成失败，终止测试") return # 4. 内容生成 (使用第一个选题) content = test_content_generate(topics[0]) if not content: print("\n⚠️ 内容生成失败，跳过海报生成") else: # 5. 海报生成 test_poster_generate(content) # 汇总 print_section("测试结果汇总") print(f" ✅ 健康检查: 通过") print(f" ✅ 配置 API: 通过") print(f" {'✅' if topics else '❌'} 选题生成: {'通过' if topics else '失败'}") print(f" {'✅' if content else '❌'} 内容生成: {'通过' if content else '失败'}") print(f"\n完整流程测试完成！") if __name__ == "__main__": main()