mark.tian/task-executer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

调试完成

Browse Source
This commit is contained in:
mark.tian
2025-12-09 17:21:32 +08:00
parent 5b26207405
commit 0af7eebed8
23 changed files with 25 additions and 1142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -176,3 +176,6 @@ cython_debug/
.vscode
/celerybeat-schedule.bak
/celerybeat-schedule.dat
/celerybeat-schedule.dir

14
README.md
View File

@@ -74,10 +74,9 @@ python simulate_failure.py
# 测试不同HTTP方法的API调用
python test_methods.py
# 使用通用API任务
from generic_api_task import generic_api_call
result = generic_api_call.delay('get_api') # 调用GET API
result = generic_api_call.delay('put_api', {'key': 'value'}) # 调用PUT API
# 使用呼叫API任务
from call_api_task import execute_call_api_task
result = execute_call_api_task.delay() # 调用所有配置的API
```
### 4. 测试Body配置
@@ -85,10 +84,9 @@ result = generic_api_call.delay('put_api', {'key': 'value'}) # 调用PUT API
# 测试不同API配置的body参数
python test_body_config.py
# 使用自定义body参数
from generic_api_task import generic_api_call
custom_body = {'title': '自定义标题', 'content': '自定义内容'}
result = generic_api_call.delay('main_api', custom_body)
# 使用呼叫API任务（自动遍历所有配置）
from call_api_task import execute_call_api_task
result = execute_call_api_task.delay() # 自动调用所有配置的API
```
### 3. 手动触发定时任务

4
invoke_call_api_task.py → call_api_task.py
View File

@@ -7,8 +7,8 @@ from api_config import API_CONFIG, RETRY_CONFIG
logger = logging.getLogger(__name__)
@app.task(bind=True, max_retries=None)
def invoke_call_api_task(self):
@app.task(bind=True, name='call_api', max_retries=None)
def execute_call_api_task(self):
"""遍历API配置文件中的所有API信息并调用接口"""
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
logger.info(f"开始遍历API配置文件执行调用任务 - 当前时间: {current_time}")

19
celery_app.py
View File

@@ -7,24 +7,33 @@ logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 创建Celery应用
app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')
app = Celery(
'tasks',
broker='redis://localhost:6379/0',
backend='redis://localhost:6379/0',
include=['call_api_task']
)
# 配置Celery
app.conf.update(
timezone='Asia/Shanghai',
enable_utc=True,
enable_utc=False,
broker_url='redis://localhost:6379/0',
result_backend='redis://localhost:6379/0',
task_serializer='json',
accept_content=['json'],
result_serializer='json',
ask_track_started=True,
# Beat 调度配置
# crontab(hour=9, minute=0), # 每天上午9点执行
beat_schedule={
'daily-morning-task': {
'task': 'tasks.daily_morning_task',
'schedule': 60.0 * 2.0, # 每2分钟执行一次
'options': {'queue': 'default'}
'task': 'call_api',
'schedule': 60.0 * 2.0
},
},
# Flower监控配置
worker_send_task_events=True,
task_send_sent_event=True,

136
tasks.py
View File

@@ -1,136 +0,0 @@
from celery_app import app
from datetime import datetime
import logging
import requests
from api_config import API_CONFIG, RETRY_CONFIG
logger = logging.getLogger(__name__)
@app.task(bind=True, max_retries=None)
def daily_morning_task(self):
"""每天定时运行配置中的API接口并支持重试机制"""
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
logger.info(f"开始执行，当前时间: {current_time}")
# 获取重试配置
max_retries = RETRY_CONFIG['max_retries']
retry_delay = RETRY_CONFIG['retry_delay']
retry_on_status = RETRY_CONFIG['retry_on_status']
# 遍历配置文件中的所有API接口
results = []
for api_name, api_config in API_CONFIG.items():
logger.info(f"开始处理API配置: {api_name}")
api_url = api_config['url']
headers = api_config['headers']
timeout = api_config['timeout']
payload = api_config['body']
method = api_config['method'].upper()
try:
logger.info(f"正在调用API接口: {method} {api_url}")
# 根据method选择请求方式
if method == 'POST':
response = requests.post(api_url, headers=headers, json=payload, timeout=timeout)
elif method == 'GET':
response = requests.get(api_url, headers=headers, params=payload, timeout=timeout)
elif method == 'PUT':
response = requests.put(api_url, headers=headers, json=payload, timeout=timeout)
elif method == 'DELETE':
response = requests.delete(api_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}")
# 构建成功消息
if response.status_code == 201:
print(f"[{current_time}] ✅ API调用成功 ({api_name}) - ID: {result_data.get('id')}")
message = 'API调用成功'
response_id = result_data.get('id')
else:
print(f"[{current_time}] ✅ API调用成功 ({api_name}) - {method}")
message = f'API调用成功 ({method})'
response_id = None
results.append({
'api_name': api_name,
'status': 'success',
'message': message,
'response': result_data,
'response_id': response_id,
'timestamp': current_time,
'retry_count': getattr(self.request, 'retries', 0)
})
elif response.status_code in retry_on_status:
# 需要重试的状态码
current_retry = getattr(self.request, 'retries', 0)
if current_retry < max_retries:
logger.warning(f"API调用需要重试，当前重试次数: {current_retry + 1}/{max_retries}")
print(f"[{current_time}] 🔄 API调用失败，正在重试 ({current_retry + 1}/{max_retries}) - {api_name} - 状态码: {response.status_code}")
raise self.retry(countdown=retry_delay, exc=Exception(f"API调用失败，状态码: {response.status_code}"))
else:
logger.error(f"API调用失败，已达到最大重试次数: {max_retries}")
print(f"[{current_time}] ❌ API调用失败，已达到最大重试次数 - {api_name} - 状态码: {response.status_code}")
results.append({
'api_name': api_name,
'status': 'failed',
'message': f'API调用失败，已达到最大重试次数: {max_retries}',
'response': response.text,
'timestamp': current_time,
'retry_count': current_retry + 1
})
else:
# 其他失败状态码，不重试
logger.error(f"API调用失败，状态码: {response.status_code}, 响应: {response.text}")
print(f"[{current_time}] ❌ API调用失败 - {api_name} - 状态码: {response.status_code}")
results.append({
'api_name': api_name,
'status': 'failed',
'message': f'API调用失败，状态码: {response.status_code}',
'response': response.text,
'timestamp': current_time,
'retry_count': getattr(self.request, 'retries', 0)
})
except requests.exceptions.RequestException as e:
# 网络异常，需要重试
current_retry = getattr(self.request, 'retries', 0)
if current_retry < max_retries:
logger.warning(f"API调用异常，需要重试，当前重试次数: {current_retry + 1}/{max_retries}")
print(f"[{current_time}] 🔄 API调用异常，正在重试 ({current_retry + 1}/{max_retries}) - {api_name} - 异常: {str(e)}")
raise self.retry(countdown=retry_delay, exc=e)
else:
logger.error(f"API调用异常，已达到最大重试次数: {max_retries}")
print(f"[{current_time}] ❌ API调用异常，已达到最大重试次数 - {api_name} - 异常: {str(e)}")
results.append({
'api_name': api_name,
'status': 'error',
'message': f'API调用异常，已达到最大重试次数: {max_retries}',
'error': str(e),
'timestamp': current_time,
'retry_count': current_retry + 1
})
# 汇总结果
success_count = len([r for r in results if r['status'] == 'success'])
total_count = len(results)
logger.info(f"定时任务执行完成 - 成功: {success_count}/{total_count}")
print(f"[{current_time}] 📊 任务汇总 - 成功: {success_count}/{total_count}")
return {
'status': 'completed',
'total_apis': total_count,
'success_count': success_count,
'failed_count': total_count - success_count,
'results': results,
'timestamp': current_time
}

43
test_api.py
View File

@@ -1,43 +0,0 @@
#!/usr/bin/env python3
"""测试API调用脚本"""
import requests
from datetime import datetime
import json
def test_api_call():
"""测试API调用"""
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
api_url = "https://jsonplaceholder.typicode.com/posts"
headers = {
'Content-Type': 'application/json',
'User-Agent': 'Celery-Task/1.0'
}
payload = {
'title': f'API测试 - {current_time}',
'body': f'这是测试API调用的数据',
'userId': 1
}
try:
print(f"正在测试API调用: {api_url}")
response = requests.post(api_url, headers=headers, json=payload, timeout=30)
if response.status_code == 201:
result_data = response.json()
print(f"✅ API调用成功")
print(f"响应数据: {json.dumps(result_data, indent=2, ensure_ascii=False)}")
return True
else:
print(f"❌ API调用失败 - 状态码: {response.status_code}")
print(f"响应内容: {response.text}")
return False
except requests.exceptions.RequestException as e:
print(f"❌ API调用异常: {str(e)}")
return False
if __name__ == '__main__':
print("=== API调用测试 ===")
success = test_api_call()
print(f"\n测试结果: {'成功' if success else '失败'}")

37
test_api_list.py
View File

@@ -1,37 +0,0 @@
#!/usr/bin/env python3
"""测试遍历API配置列表功能"""
import os
import sys
from datetime import datetime
from tasks import daily_morning_task
def test_api_list_execution():
"""测试遍历API配置列表的执行"""
try:
print("开始测试遍历API配置列表功能...")
# 直接调用任务函数进行测试
result = daily_morning_task()
print("\n" + "=" * 40)
print("测试结果汇总:")
print("=" * 40)
print(f"总API数量: {result['total_apis']}")
print(f"成功数量: {result['success_count']}")
print(f"失败数量: {result['failed_count']}")
print(f"总体状态: {result['status']}")
print("\n各API执行状态:")
for api_result in result['results']:
status_icon = "✅" if api_result['status'] == 'success' else "❌"
print(f"{status_icon} {api_result['api_name']}")
return result
except Exception as e:
print(f"❌ 测试失败: {str(e)}")
return None
if __name__ == "__main__":
test_api_list_execution()

67
test_api_list_execution.py
View File

@@ -1,67 +0,0 @@
#!/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()

53
test_beat_schedule.py
View File

@@ -1,53 +0,0 @@
#!/usr/bin/env python3
"""测试新的调度配置 - 运行2次观察执行情况"""
import subprocess
import time
import sys
import threading
from datetime import datetime
def run_beat():
"""运行Celery Beat调度器"""
try:
print("🚀 启动Celery Beat调度器...")
process = subprocess.Popen([
sys.executable, '-m', 'celery', '-A', 'celery_app', 'beat', '--loglevel=info'
], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
# 实时输出日志
for line in iter(process.stdout.readline, ''):
print(f"[BEAT] {line.strip()}")
except KeyboardInterrupt:
print("\n🛑 停止Beat调度器")
process.terminate()
except Exception as e:
print(f"❌ Beat启动失败: {e}")
def test_schedule():
"""测试新的调度配置"""
print("=" * 60)
print(f"测试新的调度配置 - 每2分钟执行一次")
print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 60)
print()
print("⚠️ 注意: 这是测试模式，将运行大约4分钟来观察2次任务执行")
print("💡 如果Redis没有运行，请先启动Redis服务")
print()
# 启动Beat调度器
beat_thread = threading.Thread(target=run_beat)
beat_thread.daemon = True
beat_thread.start()
# 等待4分钟观察执行
try:
time.sleep(240) # 4分钟，应该能看到2次任务执行
print("\n✅ 测试完成！")
except KeyboardInterrupt:
print("\n🛑 手动停止测试")
if __name__ == "__main__":
test_schedule()

84
test_body_config.py
View File

@@ -1,84 +0,0 @@
#!/usr/bin/env python3
"""测试body配置功能"""
from tasks import daily_morning_task
from invoke_call_api_task import generic_api_call
import api_config
import time
import json
def test_body_configuration():
"""测试不同API配置的body参数"""
print("=== 测试Body配置功能 ===")
# 测试不同的API配置
test_apis = ['main_api', 'backup_api', 'get_api', 'put_api']
for api_key in test_apis:
print(f"\n--- 测试 {api_key} 配置 ---")
# 显示配置信息
config = api_config.API_CONFIG[api_key]
print(f"URL: {config['url']}")
print(f"Method: {config['method']}")
print(f"Body: {json.dumps(config.get('body', {}), indent=2, ensure_ascii=False)}")
# 执行任务
result = generic_api_call.delay(api_key)
print(f"任务ID: {result.id}")
# 等待完成
for i in range(10):
try:
if result.ready():
print(f"任务状态: {result.status}")
if result.successful():
task_result = result.get()
print(f"✅ 任务成功: {task_result.get('message')}")
print(f"发送的Body: {json.dumps(api_config.get('body', {}), indent=2, ensure_ascii=False)}")
else:
print(f"❌ 任务失败: {result.result}")
break
time.sleep(1)
except Exception as e:
print(f"检查任务状态时出错: {e}")
time.sleep(1)
time.sleep(2) # 等待一下再进行下一个测试
print("\n✅ 所有body配置测试完成")
def test_custom_body():
"""测试自定义body参数"""
print("\n=== 测试自定义Body参数 ===")
custom_payload = {
'title': '自定义任务',
'body': '这是自定义的请求体数据',
'userId': 999,
'extra_field': '额外字段'
}
print(f"自定义Body: {json.dumps(custom_payload, indent=2, ensure_ascii=False)}")
result = generic_api_call.delay('main_api', custom_payload)
print(f"任务ID: {result.id}")
# 等待完成
for i in range(10):
try:
if result.ready():
print(f"任务状态: {result.status}")
if result.successful():
task_result = result.get()
print(f"✅ 自定义任务成功: {task_result.get('message')}")
else:
print(f"❌ 自定义任务失败: {result.result}")
break
time.sleep(1)
except Exception as e:
print(f"检查任务状态时出错: {e}")
time.sleep(1)
if __name__ == '__main__':
test_body_configuration()
test_custom_body()

47
test_call_api.py
View File

@@ -1,47 +0,0 @@
#!/usr/bin/env python3
"""测试呼叫任务API功能"""
import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from invoke_call_api_task import generic_api_call
from datetime import datetime
def test_call_api_task():
"""测试呼叫任务API功能"""
print("=" * 60)
print(f"测试呼叫任务API功能 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 60)
try:
# 测试默认呼叫任务API
print("🔧 测试默认呼叫任务API (main_api):")
result1 = generic_api_call()
print(f"✅ 默认呼叫结果: {result1['status']} - {result1['message']}")
print("\n🔧 测试备用呼叫任务API (backup_api):")
result2 = generic_api_call(api_key='backup_api')
print(f"✅ 备用呼叫结果: {result2['status']} - {result2['message']}")
print("\n🔧 测试自定义呼叫参数:")
custom_payload = {
'title': '呼叫任务测试',
'body': '这是自定义的呼叫任务参数',
'userId': 888
}
result3 = generic_api_call(api_key='main_api', custom_payload=custom_payload)
print(f"✅ 自定义呼叫结果: {result3['status']} - {result3['message']}")
print("\n" + "=" * 60)
print("📊 呼叫任务API测试完成！")
print("=" * 60)
return [result1, result2, result3]
except Exception as e:
print(f"❌ 测试失败: {str(e)}")
return None
if __name__ == "__main__":
test_call_api_task()

73
test_celery_api_list.py
View File

@@ -1,73 +0,0 @@
#!/usr/bin/env python3
"""使用Celery任务遍历API配置列表执行调用"""
from invoke_call_api_task import invoke_call_api_task
from api_config import API_CONFIG
def test_celery_api_list():
"""使用Celery异步任务遍历所有API配置"""
print("🚀 使用Celery任务遍历API配置列表")
print("=" * 50)
results = []
task_ids = []
# 提交所有任务
for api_key in API_CONFIG.keys():
print(f"📤 提交任务: {api_key}")
# 提交异步任务
task_result = invoke_call_api_task.delay(api_key=api_key)
task_ids.append((api_key, task_result))
results.append({
'api_key': api_key,
'task_id': task_result.id,
'status': 'submitted'
})
print(f"\n⏳ 等待 {len(task_ids)} 个任务完成...")
# 等待所有任务完成
for api_key, task_result in task_ids:
try:
# 等待任务结果
result = task_result.get(timeout=60)
# 更新结果
for r in results:
if r['api_key'] == api_key:
r['status'] = result['status']
r['message'] = result['message']
r['task_result'] = result
break
status_icon = "✅" if result['status'] == 'success' else "❌"
print(f" {status_icon} {api_key}: {result['status']}")
except Exception as e:
for r in results:
if r['api_key'] == api_key:
r['status'] = 'error'
r['message'] = str(e)
break
print(f" ❌ {api_key}: 错误 - {str(e)[:30]}")
# 汇总
print("\n" + "=" * 50)
print("📊 Celery任务执行汇总:")
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("✨ 所有任务执行完成")
return results
if __name__ == "__main__":
test_celery_api_list()

129
test_direct_api.py
View File

@@ -1,129 +0,0 @@
#!/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()

77
test_merged_status.py
View File

@@ -1,77 +0,0 @@
#!/usr/bin/env python3
"""测试合并状态码功能"""
import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from datetime import datetime
from api_config import API_CONFIG, RETRY_CONFIG
import requests
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def test_status_code_handling():
"""测试不同状态码的处理"""
print("测试合并状态码 200, 201, 204 的处理")
print("=" * 40)
current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# 测试几个主要的API
test_apis = ['main_api', 'health_check', 'get_api']
for api_name in test_apis:
if api_name in API_CONFIG:
api_config = API_CONFIG[api_name]
api_url = api_config['url']
headers = api_config['headers']
timeout = api_config['timeout']
payload = api_config['body']
method = api_config['method'].upper()
try:
print(f"\n🔧 测试 {api_name}: {method} {api_url}")
# 根据method选择请求方式
if method == 'POST':
response = requests.post(api_url, headers=headers, json=payload, timeout=timeout)
elif method == 'GET':
response = requests.get(api_url, headers=headers, params=payload, timeout=timeout)
elif method == 'PUT':
response = requests.put(api_url, headers=headers, json=payload, timeout=timeout)
elif method == 'DELETE':
response = requests.delete(api_url, headers=headers, timeout=timeout)
print(f"📊 状态码: {response.status_code}")
# 使用新的合并状态码逻辑
if response.status_code in [200, 201, 204]: # 成功状态码
try:
result_data = response.json()
except:
result_data = {'response': response.text}
# 构建成功消息
if response.status_code == 201:
print(f"✅ 成功 (创建) - ID: {result_data.get('id')}")
elif response.status_code == 200:
print(f"✅ 成功 (OK) - 数据长度: {len(str(result_data))}")
elif response.status_code == 204:
print(f"✅ 成功 (无内容)")
else:
print(f"❌ 失败 - 状态码: {response.status_code}")
except requests.exceptions.RequestException as e:
print(f"❌ 网络异常: {str(e)}")
continue # 继续测试下一个API
except Exception as e:
print(f"❌ 其他异常: {str(e)}")
continue # 继续测试下一个API
if __name__ == "__main__":
test_status_code_handling()

53
test_methods.py
View File

@@ -1,53 +0,0 @@
#!/usr/bin/env python3
"""测试不同HTTP方法的API调用"""
from tasks import daily_morning_task
import api_config
import time
def test_different_methods():
"""测试不同HTTP方法的API调用"""
print("=== 测试不同HTTP方法 ===")
# 保存原始配置
original_config = api_config.API_CONFIG['main_api'].copy()
test_cases = [
('POST', 'https://jsonplaceholder.typicode.com/posts'),
('GET', 'https://jsonplaceholder.typicode.com/posts/1'),
('PUT', 'https://jsonplaceholder.typicode.com/posts/1'),
]
for method, url in test_cases:
print(f"\n--- 测试 {method} 方法 ---")
# 临时修改配置
api_config.API_CONFIG['main_api']['method'] = method
api_config.API_CONFIG['main_api']['url'] = url
# 执行任务
result = daily_morning_task.delay()
print(f"任务ID: {result.id}")
# 等待完成
for i in range(10):
try:
if result.ready():
print(f"任务状态: {result.status}")
if result.successful():
print(f"任务结果: {result.get()}")
else:
print(f"任务失败: {result.result}")
break
time.sleep(1)
except Exception as e:
print(f"检查任务状态时出错: {e}")
time.sleep(1)
time.sleep(2) # 等待一下再进行下一个测试
# 恢复原始配置
api_config.API_CONFIG['main_api'] = original_config
print("\n✅ 所有测试完成，配置已恢复")
if __name__ == '__main__':
test_different_methods()

45
test_quick_api.py
View File

@@ -1,45 +0,0 @@
#!/usr/bin/env python3
"""快速测试API列表调用"""
import requests
from api_config import API_CONFIG
from datetime import datetime
def quick_test():
print("🚀 快速测试API列表调用")
print("=" * 40)
success = 0
total = 0
for api_key, config in API_CONFIG.items():
total += 1
print(f"\n📞 {api_key} ({config['method']})")
try:
if config['method'] == 'GET':
resp = requests.get(config['url'], headers=config['headers'],
params=config['body'], timeout=config['timeout'])
elif config['method'] == 'POST':
resp = requests.post(config['url'], headers=config['headers'],
json=config['body'], timeout=config['timeout'])
elif config['method'] == 'PUT':
resp = requests.put(config['url'], headers=config['headers'],
json=config['body'], timeout=config['timeout'])
else:
resp = requests.delete(config['url'], headers=config['headers'],
timeout=config['timeout'])
if resp.status_code in [200, 201, 204]:
print(f" ✅ 成功 - 状态码: {resp.status_code}")
success += 1
else:
print(f" ❌ 失败 - 状态码: {resp.status_code}")
except Exception as e:
print(f" ❌ 异常: {str(e)[:50]}")
print(f"\n📊 结果: {success}/{total} 成功")
if __name__ == "__main__":
quick_test()

32
test_quick_updated.py
View File

@@ -1,32 +0,0 @@
#!/usr/bin/env python3
"""快速测试更新后的invoke_call_api_task方法"""
import logging
logging.basicConfig(level=logging.INFO)
from invoke_call_api_task import invoke_call_api_task
from api_config import API_CONFIG
def quick_test():
print("🚀 快速测试更新后的invoke_call_api_task")
print("=" * 40)
# 创建任务实例
task_instance = invoke_call_api_task()
# 执行任务
result = task_instance.run()
print(f"\n📊 执行汇总:")
print(f" 状态: {result['status']}")
print(f" 总数: {result['total_apis']}")
print(f" 成功: {result['success_count']}")
print(f" 失败: {result['failed_count']}")
print(f"\n📋 API调用结果:")
for api_result in result['results']:
icon = "✅" if api_result['status'] == 'success' else "❌"
print(f" {icon} {api_result['call_api_key']} - {api_result['status']}")
if __name__ == "__main__":
quick_test()

38
test_retry.py
View File

@@ -1,38 +0,0 @@
#!/usr/bin/env python3
"""测试重试机制脚本"""
from tasks import daily_morning_task
import time
def test_retry_mechanism():
"""测试重试机制"""
print("=== 测试API重试机制 ===")
print("启动任务测试...")
# 手动触发任务
result = daily_morning_task.delay()
print(f"任务ID: {result.id}")
print("等待任务执行完成...")
# 监控任务状态
for i in range(30): # 最多等待30秒
try:
status = result.status
if status in ['SUCCESS', 'FAILURE']:
print(f"\n任务最终状态: {status}")
if status == 'SUCCESS':
print(f"任务结果: {result.get()}")
else:
print(f"任务失败信息: {result.result}")
break
else:
print(f"任务状态: {status}... (等待中)")
time.sleep(1)
except Exception as e:
print(f"检查任务状态时出错: {e}")
time.sleep(1)
print("测试完成")
if __name__ == '__main__':
test_retry_mechanism()

39
test_schedule.py
View File

@@ -1,39 +0,0 @@
#!/usr/bin/env python3
"""测试调度配置"""
import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from celery_app import app
from datetime import datetime
def check_schedule_config():
"""检查当前的调度配置"""
print("当前Celery调度配置:")
print("=" * 40)
beat_schedule = app.conf.beat_schedule
for task_name, task_config in beat_schedule.items():
schedule_seconds = task_config['schedule']
task_path = task_config['task']
# 转换为可读格式
if schedule_seconds < 60:
interval = f"{schedule_seconds}秒"
elif schedule_seconds < 3600:
minutes = schedule_seconds / 60
interval = f"{minutes:.1f}分钟"
else:
hours = schedule_seconds / 3600
interval = f"{hours:.1f}小时"
print(f"🕐 任务名称: {task_name}")
print(f"📝 任务路径: {task_path}")
print(f"⏰ 执行间隔: {interval} ({schedule_seconds}秒)")
print(f"🔧 队列: {task_config.get('options', {}).get('queue', 'default')}")
print()
if __name__ == "__main__":
check_schedule_config()

67
test_simple_api_list.py
View File

@@ -1,67 +0,0 @@
#!/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()

58
test_simulate_failure.py
View File

@@ -1,58 +0,0 @@
#!/usr/bin/env python3
"""模拟API失败情况，测试重试机制"""
def modify_config_for_test():
"""临时修改配置以测试重试机制"""
import api_config
# 保存原始配置
original_url = api_config.API_CONFIG['main_api']['url']
# 修改为无效URL以模拟失败
api_config.API_CONFIG['main_api']['url'] = 'https://nonexistent-api.example.com/posts'
print(f"已临时修改API URL为: {api_config.API_CONFIG['main_api']['url']}")
print("现在运行测试任务将触发重试机制")
print("测试完成后会恢复原始配置")
return original_url
def restore_config(original_url):
"""恢复原始配置"""
import api_config
api_config.API_CONFIG['main_api']['url'] = original_url
print(f"已恢复原始API URL: {original_url}")
def test_retry_with_failure():
"""测试失败情况下的重试机制"""
original_url = modify_config_for_test()
try:
from tasks import daily_morning_task
import time
print("\n=== 启动失败重试测试 ===")
result = daily_morning_task.delay()
print(f"任务ID: {result.id}")
# 监控重试过程
for i in range(60): # 等待足够长时间观察重试
try:
status = result.status
if status in ['SUCCESS', 'FAILURE']:
print(f"\n任务最终状态: {status}")
print(f"任务结果: {result.result}")
break
else:
print(f"任务状态: {status}... (重试中)")
time.sleep(2)
except Exception as e:
print(f"检查任务状态时出错: {e}")
time.sleep(2)
finally:
restore_config(original_url)
print("\n配置已恢复，测试完成")
if __name__ == '__main__':
test_retry_with_failure()

10
test_task.py
View File

@@ -1,10 +0,0 @@
#!/usr/bin/env python3
"""测试脚本 - 手动触发任务"""
from tasks import daily_morning_task
if __name__ == '__main__':
print("手动触发定时任务进行测试...")
result = daily_morning_task.delay()
print(f"任务已提交，任务ID: {result.id}")
print("等待任务执行结果...")
print(f"任务结果: {result.get(timeout=10)}")

39
test_updated_invoke_call.py
View File

@@ -1,39 +0,0 @@
#!/usr/bin/env python3
"""测试修改后的invoke_call_api_task方法"""
import logging
logging.basicConfig(level=logging.INFO)
from invoke_call_api_task import invoke_call_api_task
def test_updated_method():
"""测试更新后的invoke_call_api_task方法"""
print("🚀 测试修改后的invoke_call_api_task方法")
print("=" * 50)
try:
# 直接调用任务实例来测试逻辑
task_instance = invoke_call_api_task()
result = task_instance.run()
print(f"\n📊 执行结果:")
print(f" 总体状态: {result['status']}")
print(f" 总API数量: {result['total_apis']}")
print(f" 成功数量: {result['success_count']}")
print(f" 失败数量: {result['failed_count']}")
print(f" 时间戳: {result['timestamp']}")
print(f"\n📋 详细结果:")
for api_result in result['results']:
status_icon = "✅" if api_result['status'] == 'success' else "❌"
print(f" {status_icon} {api_result['call_api_key']}: {api_result['status']}")
print(f"\n🎯 任务执行完成")
return result
except Exception as e:
print(f"❌ 测试失败: {str(e)}")
return None
if __name__ == "__main__":
test_updated_method()