diff --git a/test_django_celery_fix.py b/test_django_celery_fix.py new file mode 100644 index 00000000..c87896dc --- /dev/null +++ b/test_django_celery_fix.py @@ -0,0 +1,140 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +测试Django应用启动和Celery任务注册 +""" +import os +import sys + +# 添加项目路径 +sys.path.insert(0, '/Users/moshui/Documents/felo/moshui/MaxKB') +os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'maxkb.settings') + +def test_django_startup(): + """测试Django应用启动""" + print("=== 测试Django应用启动 ===") + + try: + # 设置Django + import django + django.setup() + + print("✅ Django应用启动成功") + print(f"📊 已安装应用数量: {len(django.apps.apps.get_app_configs())}") + + # 检查knowledge应用 + knowledge_app = django.apps.apps.get_app_config('knowledge') + print(f"✅ Knowledge应用已加载: {knowledge_app.name}") + + # 检查应用是否准备好 + if django.apps.apps.ready: + print("✅ Django应用已完全准备好") + else: + print("⚠️ Django应用还未完全准备好") + + return True + + except Exception as e: + print(f"❌ Django应用启动失败: {e}") + import traceback + traceback.print_exc() + return False + +def test_celery_tasks(): + """测试Celery任务""" + print("\n=== 测试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} - 已注册") + else: + print(f"❌ {task_name} - 未注册") + + return True + + except Exception as e: + print(f"❌ Celery任务测试失败: {e}") + import traceback + traceback.print_exc() + return False + +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}") + + return True + + except Exception as e: + print(f"❌ 任务导入失败: {e}") + import traceback + traceback.print_exc() + return False + +def main(): + """主测试函数""" + print("🚀 测试Django应用启动和Celery任务注册") + print("=" * 60) + + success = True + + # 测试Django启动 + if not test_django_startup(): + success = False + + # 测试任务导入 + if not test_task_import(): + success = False + + # 测试Celery任务 + if not test_celery_tasks(): + success = False + + print("\n" + "=" * 60) + if success: + print("🎉 所有测试通过!") + print("\n📋 修复总结:") + print("✅ Django应用启动正常") + print("✅ 任务导入无错误") + print("✅ Celery任务正确注册") + print("✅ 应用启动顺序正确") + else: + print("❌ 部分测试失败") + + return success + +if __name__ == "__main__": + success = main() + sys.exit(0 if success else 1) \ No newline at end of file