From a5d1dda65f2b58b59cd4e7899170701ccc6595f4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E6=9C=B1=E6=BD=AE?= <zhuchaowe@users.noreply.github.com>
Date: Sun, 31 Aug 2025 01:25:29 +0800
Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0Django=E5=BA=94=E7=94=A8?=
 =?UTF-8?q?=E5=90=AF=E5=8A=A8=E5=92=8CCelery=E4=BB=BB=E5=8A=A1=E6=B3=A8?=
 =?UTF-8?q?=E5=86=8C=E6=B5=8B=E8=AF=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 创建测试脚本验证修复效果
- 测试Django应用启动、任务导入和注册
- 提供详细的错误诊断信息

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 test_django_celery_fix.py | 140 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 140 insertions(+)
 create mode 100644 test_django_celery_fix.py

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