TravelContentCreator/tests/test_full_flow.py

#!/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()