#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 完整流程测试: 选题生成 → 正文生成 测试 V2 API 的完整业务流程: 1. 生成选题 2. 基于选题生成正文 (原创模式) 3. 基于选题生成正文 (reference 模式) 4. 基于选题生成正文 (rewrite 模式) 使用方法: python tests/test_topic_to_content.py [--base-url http://localhost:8001] """ import argparse import json import requests import time from typing import Dict, Any, Optional class Colors: """终端颜色""" GREEN = '\033[92m' RED = '\033[91m' YELLOW = '\033[93m' BLUE = '\033[94m' CYAN = '\033[96m' RESET = '\033[0m' BOLD = '\033[1m' def print_header(text: str): print(f"\n{Colors.BOLD}{Colors.BLUE}{'='*60}{Colors.RESET}") print(f"{Colors.BOLD}{Colors.BLUE}{text}{Colors.RESET}") print(f"{Colors.BOLD}{Colors.BLUE}{'='*60}{Colors.RESET}") def print_step(step: int, text: str): print(f"\n{Colors.CYAN}[Step {step}] {text}{Colors.RESET}") def print_success(text: str): print(f"{Colors.GREEN}✅ {text}{Colors.RESET}") def print_error(text: str): print(f"{Colors.RED}❌ {text}{Colors.RESET}") def print_info(text: str): print(f"{Colors.YELLOW}ℹ️ {text}{Colors.RESET}") def call_api(base_url: str, endpoint: str, method: str = "GET", data: Dict = None) -> Dict: """调用 API""" url = f"{base_url}{endpoint}" try: if method == "GET": resp = requests.get(url, timeout=120) else: resp = requests.post(url, json=data, timeout=120) return resp.json() except requests.exceptions.Timeout: return {"success": False, "error": "请求超时"} except Exception as e: return {"success": False, "error": str(e)} def test_topic_generate(base_url: str, subject: Dict, style: Dict, audience: Dict) -> Optional[Dict]: """测试选题生成""" print_step(1, "生成选题") params = { "engine": "topic_generate", "params": { "month": "2025-02", "num_topics": 2, "prompt_version": "v2.0.0", "subject": subject, "style": style, "audience": audience, "hot_topics": { "events": [], "festivals": [ {"name": "春节", "date": "2025-01-29", "marketing_angle": "团圆、年味"} ], "trending": [] } } } print_info(f"请求参数: subject={subject['name']}, style={style['name']}, audience={audience['name']}") start_time = time.time() result = call_api(base_url, "/api/v2/aigc/execute", "POST", params) elapsed = time.time() - start_time if result.get("success"): topics = result.get("data", {}).get("topics", []) print_success(f"生成 {len(topics)} 个选题 (耗时 {elapsed:.1f}s)") for i, topic in enumerate(topics): print(f"\n {Colors.BOLD}选题 {i+1}:{Colors.RESET}") print(f" 标题: {topic.get('title', 'N/A')}") print(f" 日期: {topic.get('date', 'N/A')}") print(f" 风格: {topic.get('style', 'N/A')}") print(f" 人群: {topic.get('targetAudience', 'N/A')}") print(f" 逻辑: {topic.get('logic', 'N/A')[:50]}...") return topics[0] if topics else None else: print_error(f"选题生成失败: {result.get('error', 'Unknown error')}") return None def test_content_generate(base_url: str, topic: Dict, subject: Dict, style: Dict, audience: Dict, reference: Optional[Dict] = None, mode_name: str = "原创") -> Optional[Dict]: """测试正文生成""" print_step(2 if reference is None else (3 if reference.get('mode') == 'reference' else 4), f"生成正文 ({mode_name}模式)") params = { "engine": "content_generate", "params": { "prompt_version": "v2.0.0", "topic": topic, "subject": subject, "style": style, "audience": audience, "enable_judge": False } } if reference: params["params"]["reference"] = reference print_info(f"参考模式: {reference.get('mode')}") else: print_info("原创模式，无参考内容") start_time = time.time() result = call_api(base_url, "/api/v2/aigc/execute", "POST", params) elapsed = time.time() - start_time if result.get("success"): content = result.get("data", {}).get("content", {}) print_success(f"正文生成成功 (耗时 {elapsed:.1f}s)") print(f"\n {Colors.BOLD}标题:{Colors.RESET} {content.get('title', 'N/A')}") body = content.get('content', '') # 截取前 200 字符显示 preview = body[:200] + "..." if len(body) > 200 else body print(f"\n {Colors.BOLD}正文预览:{Colors.RESET}") for line in preview.split('\n')[:8]: print(f" {line}") print(f"\n {Colors.BOLD}TAG:{Colors.RESET} {content.get('tag', 'N/A')}") print(f"\n {Colors.BOLD}参考模式:{Colors.RESET} {content.get('referenceMode', 'none')}") return content else: print_error(f"正文生成失败: {result.get('error', 'Unknown error')}") return None def run_full_test(base_url: str): """运行完整测试""" print_header("TravelContentCreator 完整流程测试") print(f"API 地址: {base_url}") # 测试数据 subject = { "id": 1, "name": "北京环球影城", "type": "scenic_spot", "description": "亚洲最大的环球影城主题公园，拥有哈利波特魔法世界、变形金刚基地等七大主题景区", "location": "北京市通州区", "traffic_info": "地铁1号线/7号线环球度假区站", "highlights": ["哈利波特魔法世界", "变形金刚基地", "小黄人乐园", "侏罗纪世界"], "advantages": "全球顶级主题乐园，沉浸式体验", "products": [ { "id": 10, "name": "单日票", "price": 638, "original_price": 738, "sales_period": "2025-01-01 至 2025-03-31", "package_info": "含一次入园", "usage_rules": "入园当日有效，不可退改" } ] } style = {"id": "gonglue", "name": "攻略风"} audience = {"id": "qinzi", "name": "亲子向"} # 参考内容 (用于 reference 和 rewrite 测试) reference_content = { "title": "上海迪士尼遛娃天花板！一日游保姆级攻略🏰", "content": """带娃去迪士尼，这篇攻略你一定要收藏！ 🎯 必玩项目TOP5 1. 创极速光轮 - 全球最快迪士尼过山车 2. 加勒比海盗 - 沉浸式体验超震撼 3. 飞越地平线 - 裸眼4D环游世界 4. 小飞侠天空奇遇 - 适合小朋友 5. 七个小矮人矿山车 - 刺激又不吓人 ⏰ 时间规划 8:30 到达门口排队 9:00 开园直冲创极速光轮 12:00 午餐（建议自带） 14:00 花车巡游 20:00 烟花秀 💡 省钱tips - 门票提前买，便宜100+ - 自带水和零食 - 下载APP领快速通道""" } # Step 1: 生成选题 topic = test_topic_generate(base_url, subject, style, audience) if not topic: print_error("选题生成失败，终止测试") return False # Step 2: 原创模式生成正文 content1 = test_content_generate(base_url, topic, subject, style, audience, reference=None, mode_name="原创") # Step 3: reference 模式生成正文 (参考风格，原创内容) content2 = test_content_generate(base_url, topic, subject, style, audience, reference={"mode": "reference", **reference_content}, mode_name="reference 参考") # Step 4: rewrite 模式生成正文 (保留框架，换主体) content3 = test_content_generate(base_url, topic, subject, style, audience, reference={"mode": "rewrite", **reference_content}, mode_name="rewrite 改写") # 结果汇总 print_header("测试结果汇总") results = [ ("选题生成", topic is not None), ("原创模式正文", content1 is not None), ("reference 模式正文", content2 is not None), ("rewrite 模式正文", content3 is not None), ] all_passed = True for name, passed in results: if passed: print_success(f"{name}: 通过") else: print_error(f"{name}: 失败") all_passed = False if all_passed: print(f"\n{Colors.GREEN}{Colors.BOLD}🎉 所有测试通过！{Colors.RESET}") else: print(f"\n{Colors.RED}{Colors.BOLD}⚠️ 部分测试失败{Colors.RESET}") return all_passed def main(): parser = argparse.ArgumentParser(description="TravelContentCreator 完整流程测试") parser.add_argument("--base-url", default="http://localhost:8001", help="API 基础地址") args = parser.parse_args() # 检查服务是否可用 print_info(f"检查服务状态: {args.base_url}") try: resp = requests.get(f"{args.base_url}/", timeout=5) if resp.status_code == 200: print_success("服务运行正常") else: print_error(f"服务异常: HTTP {resp.status_code}") return except Exception as e: print_error(f"无法连接服务: {e}") print_info("请先启动服务: PYTHONPATH=. uvicorn api.main:app --port 8001") return # 运行测试 run_full_test(args.base_url) if __name__ == "__main__": main()