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配置"""
|
||
print("🚀 开始遍历API配置列表执行调用")
|
||
print("=" * 50)
|
||
|
||
results = []
|
||
|
||
for api_key, config in API_CONFIG.items():
|
||
print(f"\n📞 正在调用API: {api_key}")
|
||
print(f" URL: {config['url']}")
|
||
print(f" 方法: {config['method']}")
|
||
|
||
try:
|
||
# 使用异步方式执行Celery任务
|
||
from celery import current_app
|
||
|
||
# 提交异步任务
|
||
async_result = invoke_call_api_task.delay(api_key=api_key)
|
||
|
||
# 等待任务完成
|
||
result = async_result.get(timeout=60)
|
||
|
||
print(f" 状态: {result['status']}")
|
||
print(f" 消息: {result['message']}")
|
||
|
||
status_icon = "✅" if result['status'] == 'success' else "❌"
|
||
print(f" 结果: {status_icon} {result['status']}")
|
||
|
||
results.append({
|
||
'api_key': api_key,
|
||
'status': result['status'],
|
||
'message': result['message']
|
||
})
|
||
|
||
except Exception as e:
|
||
print(f" ❌ 异常: {str(e)}")
|
||
results.append({
|
||
'api_key': api_key,
|
||
'status': 'error',
|
||
'message': str(e)
|
||
})
|
||
|
||
# 短暂延迟
|
||
time.sleep(1)
|
||
|
||
# 汇总
|
||
print("\n" + "=" * 50)
|
||
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['message']}")
|
||
|
||
print(f"\n🎯 总计: {success_count}/{total_count} 成功")
|
||
|
||
if __name__ == "__main__":
|
||
main() |