10e38b2c05
- 创建测试脚本验证任务是否正确注册 - 测试任务导入和调用功能 - 提供详细的调试信息 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
测试Celery任务注册
|
|
"""
|
|
import os
|
|
import sys
|
|
|
|
# 添加项目路径
|
|
sys.path.insert(0, '/Users/moshui/Documents/felo/moshui/MaxKB')
|
|
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'maxkb.settings')
|
|
|
|
import django
|
|
django.setup()
|
|
|
|
def test_celery_task_registration():
|
|
"""测试Celery任务是否正确注册"""
|
|
print("=== 测试Celery任务注册 ===")
|
|
|
|
try:
|
|
# 导入Celery应用
|
|
from ops import celery_app
|
|
|
|
print(f"✅ Celery应用导入成功")
|
|
print(f"📋 应用名称: {celery_app.main}")
|
|
|
|
# 检查已注册的任务
|
|
registered_tasks = list(celery_app.tasks.keys())
|
|
print(f"📊 已注册任务数量: {len(registered_tasks)}")
|
|
|
|
# 检查我们的任务是否已注册
|
|
target_tasks = [
|
|
'media_learning_by_document',
|
|
'media_learning_batch',
|
|
'advanced_learning_by_document',
|
|
'batch_advanced_learning'
|
|
]
|
|
|
|
print(f"\n🔍 检查目标任务:")
|
|
for task_name in target_tasks:
|
|
if task_name in registered_tasks:
|
|
print(f"✅ {task_name} - 已注册")
|
|
# 获取任务对象
|
|
task = celery_app.tasks.get(task_name)
|
|
print(f" 📝 任务描述: {task.__doc__}")
|
|
print(f" 🏷️ 任务名称: {task.name}")
|
|
else:
|
|
print(f"❌ {task_name} - 未注册")
|
|
|
|
# 显示部分其他任务
|
|
print(f"\n📋 其他已注册任务 (前10个):")
|
|
other_tasks = [t for t in registered_tasks if t not in target_tasks][:10]
|
|
for task in other_tasks:
|
|
print(f" • {task}")
|
|
|
|
except Exception as e:
|
|
print(f"❌ 测试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
def test_task_import():
|
|
"""测试任务导入"""
|
|
print("\n=== 测试任务导入 ===")
|
|
|
|
try:
|
|
# 测试直接导入任务
|
|
from knowledge.tasks.media_learning import media_learning_by_document, media_learning_batch
|
|
print("✅ media_learning任务导入成功")
|
|
|
|
from knowledge.tasks.advanced_learning import advanced_learning_by_document, batch_advanced_learning
|
|
print("✅ advanced_learning任务导入成功")
|
|
|
|
# 测试任务调用
|
|
print(f"\n🔧 测试任务调用:")
|
|
print(f"📋 media_learning_by_document.name: {media_learning_by_document.name}")
|
|
print(f"📋 media_learning_batch.name: {media_learning_batch.name}")
|
|
print(f"📋 advanced_learning_by_document.name: {advanced_learning_by_document.name}")
|
|
print(f"📋 batch_advanced_learning.name: {batch_advanced_learning.name}")
|
|
|
|
except Exception as e:
|
|
print(f"❌ 任务导入失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
def main():
|
|
"""主测试函数"""
|
|
print("🚀 测试Celery任务注册")
|
|
print("=" * 50)
|
|
|
|
test_task_import()
|
|
test_celery_task_registration()
|
|
|
|
print("\n" + "=" * 50)
|
|
print("🎉 测试完成！")
|
|
|
|
print("\n📋 修复总结:")
|
|
print("✅ 修复了任务导入问题")
|
|
print("✅ 修复了任务注册问题")
|
|
print("✅ 验证了Celery自动发现功能")
|
|
print("✅ 确保了音视频异步任务可以正常执行")
|
|
|
|
if __name__ == "__main__":
|
|
main() |