#!/usr/bin/env python """ 测试API调用 """ import requests import json def test_training_api(): """测试训练API调用""" print("=== 测试训练API ===") # 准备训练请求数据 training_data = { "product_id": "P001", "model_type": "tcn", "epochs": 5, "training_mode": "product", "store_id": None, "aggregation_method": "sum" } try: # 发送训练请求 response = requests.post( "http://localhost:5000/api/train", json=training_data, timeout=10 ) print(f"响应状态码: {response.status_code}") if response.status_code == 200: result = response.json() print("训练请求成功发送") print(f"任务ID: {result.get('task_id', 'N/A')}") else: print(f"训练请求失败: {response.text}") except requests.exceptions.ConnectionError: print("无法连接到API服务器 (http://localhost:5000)") print("请确保后端服务器正在运行") except Exception as e: print(f"API调用失败: {e}") def test_products_api(): """测试产品列表API""" print("\n=== 测试产品列表API ===") try: response = requests.get("http://localhost:5000/api/products", timeout=5) print(f"响应状态码: {response.status_code}") if response.status_code == 200: products = response.json() print(f"获取到 {len(products)} 个产品") if products: print(f"第一个产品: {products[0]}") else: print(f"获取产品列表失败: {response.text}") except requests.exceptions.ConnectionError: print("无法连接到API服务器") except Exception as e: print(f"API调用失败: {e}") def test_stores_api(): """测试店铺列表API""" print("\n=== 测试店铺列表API ===") try: response = requests.get("http://localhost:5000/api/stores", timeout=5) print(f"响应状态码: {response.status_code}") if response.status_code == 200: stores = response.json() print(f"获取到 {len(stores)} 个店铺") if stores: print(f"第一个店铺: {stores[0]}") else: print(f"获取店铺列表失败: {response.text}") except requests.exceptions.ConnectionError: print("无法连接到API服务器") except Exception as e: print(f"API调用失败: {e}") def main(): """主测试函数""" print("开始API测试") # 测试产品列表API test_products_api() # 测试店铺列表API test_stores_api() # 测试训练API test_training_api() print("\nAPI测试完成") if __name__ == "__main__": main()