daren/test_search_api.py

#!/usr/bin/env python3
"""
TikTok创作者数据库搜索API测试脚本
"""

import requests
import json

# 配置
BASE_URL = "http://localhost:8000"  # 根据您的服务器地址调整
TOKEN = "your_token_here"  # 替换为实际的认证token

# 请求头
HEADERS = {
    "Authorization": f"Token {TOKEN}",
    "Content-Type": "application/json"
}

def test_basic_search():
    """测试基础搜索功能"""
    print("=== 测试基础搜索功能 ===")
    
    # 测试1: 无关键词搜索（获取所有创作者）
    print("\n1. 测试无关键词搜索:")
    response = requests.get(
        f"{BASE_URL}/creators/search/",
        headers=HEADERS,
        params={"page": 1, "page_size": 5}
    )
    print(f"状态码: {response.status_code}")
    if response.status_code == 200:
        data = response.json()
        print(f"总数量: {data['data']['pagination']['total_count']}")
        print(f"返回数量: {len(data['data']['creators'])}")
    else:
        print(f"错误: {response.text}")
    
    # 测试2: 关键词搜索
    print("\n2. 测试关键词搜索 (搜索'beauty'):")
    response = requests.get(
        f"{BASE_URL}/creators/search/",
        headers=HEADERS,
        params={"q": "beauty", "page": 1, "page_size": 5}
    )
    print(f"状态码: {response.status_code}")
    if response.status_code == 200:
        data = response.json()
        print(f"搜索结果数量: {data['data']['search_info']['results_count']}")
        print(f"搜索关键词: {data['data']['search_info']['query']}")
        if data['data']['creators']:
            print(f"第一个结果: {data['data']['creators'][0]['Creator']['name']}")
    else:
        print(f"错误: {response.text}")

def test_advanced_search():
    """测试高级搜索功能"""
    print("\n=== 测试高级搜索功能 ===")
    
    # 测试高级搜索
    search_data = {
        "q": "beauty",
        "filters": {
            "category": ["Beauty & Personal Care", "Fashion"],
            "e_commerce_level": ["L2", "L3"],
            "gmv_range": ["$5k-$25k", "$25k-$50k"]
        }
    }
    
    print(f"\n搜索条件: {json.dumps(search_data, indent=2, ensure_ascii=False)}")
    
    response = requests.post(
        f"{BASE_URL}/creators/search/advanced/",
        headers=HEADERS,
        params={"page": 1, "page_size": 5},
        json=search_data
    )
    
    print(f"状态码: {response.status_code}")
    if response.status_code == 200:
        data = response.json()
        print(f"搜索结果数量: {data['data']['search_info']['results_count']}")
        print(f"应用的筛选条件: {data['data']['search_info']['filters_applied']}")
        if data['data']['creators']:
            creator = data['data']['creators'][0]
            print(f"第一个结果:")
            print(f"  - 名称: {creator['Creator']['name']}")
            print(f"  - 分类: {creator['Category']}")
            print(f"  - 电商等级: {creator['E-commerce Level']}")
            print(f"  - GMV: {creator['GMV']}")
    else:
        print(f"错误: {response.text}")

def test_search_suggestions():
    """测试搜索建议功能"""
    print("\n=== 测试搜索建议功能 ===")
    
    # 测试搜索建议
    test_queries = ["bea", "spo", "fas"]
    
    for query in test_queries:
        print(f"\n搜索建议 '{query}':")
        response = requests.get(
            f"{BASE_URL}/creators/search/suggestions/",
            headers=HEADERS,
            params={"q": query, "limit": 5}
        )
        
        print(f"状态码: {response.status_code}")
        if response.status_code == 200:
            data = response.json()
            suggestions = data['data']['suggestions']
            print(f"建议数量: {len(suggestions)}")
            for suggestion in suggestions:
                print(f"  - {suggestion['label']}")
        else:
            print(f"错误: {response.text}")

def test_pagination():
    """测试分页功能"""
    print("\n=== 测试分页功能 ===")
    
    # 测试分页
    response = requests.get(
        f"{BASE_URL}/creators/search/",
        headers=HEADERS,
        params={"page": 1, "page_size": 3}
    )
    
    print(f"状态码: {response.status_code}")
    if response.status_code == 200:
        data = response.json()
        pagination = data['data']['pagination']
        print(f"分页信息:")
        print(f"  - 当前页: {pagination['current_page']}")
        print(f"  - 总页数: {pagination['total_pages']}")
        print(f"  - 总数量: {pagination['total_count']}")
        print(f"  - 有下一页: {pagination['has_next']}")
        print(f"  - 有上一页: {pagination['has_prev']}")
        print(f"  - 每页数量: {pagination['page_size']}")
    else:
        print(f"错误: {response.text}")

def main():
    """主测试函数"""
    print("TikTok创作者数据库搜索API测试")
    print("=" * 50)
    
    try:
        # 运行所有测试
        test_basic_search()
        test_advanced_search()
        test_search_suggestions()
        test_pagination()
        
        print("\n" + "=" * 50)
        print("所有测试完成！")
        
    except requests.exceptions.ConnectionError:
        print("错误: 无法连接到服务器，请确保服务器正在运行")
    except Exception as e:
        print(f"测试过程中发生错误: {e}")

if __name__ == "__main__":
    print("请在运行测试前确保:")
    print("1. 服务器正在运行")
    print("2. 已更新BASE_URL和TOKEN配置")
    print("3. 数据库中有测试数据")
    print()
    
    confirm = input("是否继续运行测试? (y/n): ")
    if confirm.lower() == 'y':
        main()
    else:
        print("测试已取消")