12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879 |
- import json
- import requests
- from typing import Dict, Any
- import os
- def load_json_file(file_path: str) -> Dict[str, Any]:
- """从JSON文件加载数据"""
- if not os.path.exists(file_path):
- raise FileNotFoundError(f"文件不存在: {file_path}")
- with open(file_path, 'r', encoding='utf-8') as f:
- try:
- return json.load(f)
- except json.JSONDecodeError as e:
- raise ValueError(f"JSON解析错误: {str(e)}")
- def send_to_server(api_url: str, payload: Dict[str, Any]) -> None:
- """发送数据到服务器并处理响应"""
- headers = {'Content-Type': 'application/json'}
- try:
- response = requests.post(
- api_url,
- json=payload,
- headers=headers,
- timeout=10
- )
- # 解析响应
- resp_data = response.json()
- print("\n服务器响应:")
- print(f"状态码: {response.status_code}")
- print(f"响应内容: {resp_data}")
- # 检查响应格式
- if not isinstance(resp_data, dict) or 'status' not in resp_data or 'msg' not in resp_data:
- print("警告: 服务器返回了非标准响应格式")
- if response.status_code == 200 and resp_data.get('status') == 200:
- print("请求处理成功!")
- else:
- print("请求处理失败!")
- except requests.exceptions.RequestException as e:
- print(f"\n请求失败: {str(e)}")
- if hasattr(e, 'response') and e.response is not None:
- print(f"错误响应: {e.response.text}")
- def main():
- # 配置参数
- config = {
- "server_url": "http://127.0.0.1:8081/process_brdf/",
- "json_file": "test_data.json"
- }
- print("=== BRDF 测试客户端 ===")
- print(f"服务器地址: {config['server_url']}")
- print(f"测试数据文件: {config['json_file']}")
- try:
- # 1. 加载测试数据
- print("\n正在加载测试数据...")
- payload = load_json_file(config['json_file'])
- print("测试数据加载成功!")
- print(json.dumps(payload, indent=2, ensure_ascii=False))
- # 2. 发送到服务器
- print("\n正在发送数据到服务器...")
- send_to_server(config['server_url'], payload)
- except Exception as e:
- print(f"\n发生错误: {str(e)}")
- if __name__ == "__main__":
- main()
|