测试代码改为以test_开头

test_direct_api.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""直接测试API调用（不使用Celery）"""
+
+import requests
+from datetime import datetime
+from api_config import API_CONFIG, RETRY_CONFIG
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+def call_api_direct(api_key='main_api', custom_payload=None):
+    """直接调用API（复制invoke_call_api_task的核心逻辑）"""
+    current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+    logger.info(f"呼叫任务API开始执行 - 当前时间: {current_time}")
+    
+    # 获取重试配置
+    max_retries = RETRY_CONFIG['max_retries']
+    retry_delay = RETRY_CONFIG['retry_delay']
+    retry_on_status = RETRY_CONFIG['retry_on_status']
+    
+    # 获取呼叫任务API配置
+    call_config = API_CONFIG[api_key]
+    call_url = call_config['url']
+    method = call_config['method'].upper()
+    headers = call_config['headers']
+    timeout = call_config['timeout']
+    
+    # 构建呼叫参数
+    if custom_payload is None:
+        payload = call_config['body']
+    else:
+        payload = custom_payload
+    
+    try:
+        logger.info(f"正在调用呼叫任务API: {method} {call_url}")
+        
+        # 根据method选择请求方式
+        if method == 'POST':
+            response = requests.post(call_url, headers=headers, json=payload, timeout=timeout)
+        elif method == 'GET':
+            response = requests.get(call_url, headers=headers, params=payload, timeout=timeout)
+        elif method == 'PUT':
+            response = requests.put(call_url, headers=headers, json=payload, timeout=timeout)
+        elif method == 'DELETE':
+            response = requests.delete(call_url, headers=headers, timeout=timeout)
+        else:
+            raise ValueError(f"不支持的HTTP方法: {method}")
+        
+        # 检查响应状态
+        if response.status_code in [200, 201, 204]:
+            try:
+                result_data = response.json()
+            except:
+                result_data = {'response': response.text}
+            
+            logger.info(f"呼叫任务API调用成功，响应数据: {result_data}")
+            print(f"[{current_time}] ✅ 呼叫任务API成功 - {method} {call_url}")
+            return {
+                'status': 'success',
+                'message': f'呼叫任务API调用成功 ({method})',
+                'response': result_data,
+                'timestamp': current_time,
+                'api_key': api_key
+            }
+        else:
+            logger.error(f"呼叫任务API失败，状态码: {response.status_code}, 响应: {response.text}")
+            print(f"[{current_time}] ❌ 呼叫任务API失败 - 状态码: {response.status_code}")
+            return {
+                'status': 'failed',
+                'message': f'呼叫任务API失败，状态码: {response.status_code}',
+                'response': response.text,
+                'timestamp': current_time,
+                'api_key': api_key
+            }
+            
+    except requests.exceptions.RequestException as e:
+        logger.error(f"呼叫任务API异常: {str(e)}")
+        print(f"[{current_time}] ❌ 呼叫任务API异常 - {str(e)}")
+        return {
+            'status': 'error',
+            'message': f'呼叫任务API异常: {str(e)}',
+            'timestamp': current_time,
+            'api_key': api_key
+        }
+
+def main():
+    """遍历所有API配置并执行调用"""
+    print("🚀 开始遍历API配置列表执行直接调用")
+    print("=" * 60)
+    
+    results = []
+    
+    for api_key, config in API_CONFIG.items():
+        print(f"\n📞 正在调用API: {api_key}")
+        print(f"   URL: {config['url']}")
+        print(f"   方法: {config['method']}")
+        print(f"   超时: {config['timeout']}秒")
+        
+        # 直接调用API函数
+        result = call_api_direct(api_key=api_key)
+        
+        status_icon = "✅" if result['status'] == 'success' else "❌"
+        print(f"   结果: {status_icon} {result['status']}")
+        print(f"   消息: {result['message']}")
+        
+        results.append({
+            'api_key': api_key,
+            'status': result['status'],
+            'message': result['message'],
+            'result': result
+        })
+    
+    # 汇总结果
+    print("\n" + "=" * 60)
+    print("📊 任务执行汇总:")
+    
+    success_count = len([r for r in results if r['status'] == 'success'])
+    total_count = len(results)
+    
+    for result in results:
+        status_icon = "✅" if result['status'] == 'success' else "❌"
+        print(f"   {status_icon} {result['api_key']}: {result['status']}")
+    
+    print(f"\n🎯 总计: {success_count}/{total_count} 成功")
+    print("✨ 任务执行完成")
+
+if __name__ == "__main__":
+    main()