67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""测试遍历API配置列表执行调用任务"""
|
|
|
|
from invoke_call_api_task import invoke_call_api_task
|
|
from api_config import API_CONFIG
|
|
import time
|
|
|
|
def main():
|
|
"""遍历API配置文件中的所有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调用任务
|
|
try:
|
|
result = invoke_call_api_task.delay(api_key=api_key)
|
|
|
|
# 等待任务完成并获取结果
|
|
task_result = result.get(timeout=config['timeout'] + 10)
|
|
|
|
print(f" 结果: {task_result['status']}")
|
|
print(f" 消息: {task_result['message']}")
|
|
if task_result['status'] == 'success':
|
|
print(f" ✅ 成功")
|
|
else:
|
|
print(f" ❌ 失败")
|
|
|
|
results.append({
|
|
'api_key': api_key,
|
|
'status': task_result['status'],
|
|
'result': task_result
|
|
})
|
|
|
|
except Exception as e:
|
|
print(f" ❌ 异常: {str(e)}")
|
|
results.append({
|
|
'api_key': api_key,
|
|
'status': 'error',
|
|
'error': str(e)
|
|
})
|
|
|
|
# 短暂延迟避免请求过于频繁
|
|
time.sleep(1)
|
|
|
|
# 汇总结果
|
|
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() |