zhouxiaoka--autoclip/backend/core/celery_simple_fixed.py

"""
修复的简化Celery应用配置
解决任务路由和状态更新问题
"""

import os
from celery import Celery

# 创建Celery应用
celery_app = Celery('autoclip')

# 基本配置
celery_app.conf.update(
    # 序列化格式
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    
    # Redis配置
    broker_url='redis://localhost:6379/0',
    result_backend='redis://localhost:6379/0',
    
    # Broker配置
    broker_transport='redis',
    broker_transport_options={},
    
    # 队列配置
    task_default_queue='processing',
    task_default_exchange='processing',
    task_default_routing_key='processing',
    
    # 时区
    timezone='Asia/Shanghai',
    enable_utc=True,
    
    # 任务配置
    task_always_eager=False,
    task_eager_propagates=True,
    
    # 工作进程配置
    worker_prefetch_multiplier=1,
    worker_max_tasks_per_child=1000,
    worker_disable_rate_limits=True,
    
    # 结果配置
    result_expires=3600,
    task_ignore_result=False,
    
    # 任务路由配置
    task_routes={
        'backend.tasks.processing.*': {'queue': 'processing'},
        'backend.tasks.video.*': {'queue': 'upload'},
        'backend.tasks.notification.*': {'queue': 'notification'},
        'backend.tasks.maintenance.*': {'queue': 'maintenance'},
        'backend.tasks.upload.*': {'queue': 'upload'},
    },
    
    # 禁用自动发现，手动注册任务
    autodiscover_tasks=False,
)

# 手动注册任务，避免自动发现
@celery_app.task(bind=True, name='tasks.processing.process_video_pipeline')
def process_video_pipeline(self, project_id: str, input_video_path: str, input_srt_path: str, *args, **kwargs):
    """视频处理流水线任务"""
    # 直接调用有进度更新服务的版本
    return backend_process_video_pipeline(self, project_id, input_video_path, input_srt_path, *args, **kwargs)

@celery_app.task(bind=True, name='tasks.processing.process_single_step')
def process_single_step(self, project_id: str, step: str, config: dict, *args, **kwargs):
    """单个步骤处理任务"""
    print(f"🔧 开始处理项目 {project_id} 的步骤: {step}")
    if args:
        print(f"⚠️  额外位置参数: {args}")
    if kwargs:
        print(f"⚠️  额外关键字参数: {kwargs}")
    
    # 模拟处理过程
    import time
    time.sleep(3)
    
    print(f"✅ 步骤 {step} 处理完成")
    return {
        "success": True,
        "project_id": project_id,
        "step": step,
        "message": f"步骤 {step} 处理完成"
    }

# 兼容性任务名称
@celery_app.task(bind=True, name='backend.tasks.processing.process_video_pipeline')
def backend_process_video_pipeline(self, project_id: str, input_video_path: str, input_srt_path: str, *args, **kwargs):
    """后端视频处理流水线任务（兼容性）"""
    # 直接实现任务逻辑，避免函数引用问题
    print(f"🎬 开始处理项目: {project_id}")
    print(f"📹 视频路径: {input_video_path}")
    print(f"📝 字幕路径: {input_srt_path}")
    if args:
        print(f"⚠️  额外位置参数: {args}")
    if kwargs:
        print(f"⚠️  额外关键字参数: {kwargs}")
    
    # 获取任务ID
    task_id = self.request.id
    print(f"🔑 Celery任务ID: {task_id}")
    
    # 模拟处理过程
    import time
    steps = [
        "大纲提取",
        "时间定位", 
        "内容评分",
        "标题生成",
        "主题聚类",
        "视频切割"
    ]
    
    for i, step in enumerate(steps):
        progress = (i + 1) * 16  # 每步16%
        print(f"📊 步骤 {i+1}/6: {step} - {progress}%")
        
        # 更新任务状态
        try:
            self.update_state(
                state='PROGRESS',
                meta={
                    'current': i + 1,
                    'total': 6,
                    'status': f'正在执行: {step}',
                    'progress': progress
                }
            )
        except Exception as e:
            print(f"⚠️  更新任务状态失败: {e}")
        
        time.sleep(2)  # 模拟处理时间
    
    print(f"✅ 项目 {project_id} 处理完成")
    
    # 尝试更新数据库中的任务和项目状态
    try:
        from ..core.database import SessionLocal
        from ..models.task import Task, TaskStatus
        from ..models.project import Project, ProjectStatus
        from datetime import datetime
        
        # 直接更新数据库，避免异步调用问题
        db = SessionLocal()
        try:
            # 更新任务状态
            task = db.query(Task).filter(Task.id == task_id).first()
            if task:
                task.status = TaskStatus.COMPLETED
                task.progress = 100.0
                task.current_step = '完成'
                task.completed_at = datetime.utcnow()
                task.updated_at = datetime.utcnow()
                print(f"✅ 任务状态已更新到数据库")
            else:
                print(f"⚠️  找不到任务: {task_id}")
            
            # 更新项目状态
            project = db.query(Project).filter(Project.id == project_id).first()
            if project:
                project.status = ProjectStatus.COMPLETED
                project.completed_at = datetime.utcnow()
                project.updated_at = datetime.utcnow()
                print(f"✅ 项目状态已更新为已完成: {project_id}")
            else:
                print(f"⚠️  找不到项目: {project_id}")
            
            db.commit()
            
        finally:
            db.close()
            
    except Exception as e:
        print(f"⚠️  更新数据库状态失败: {e}")
    
    return {
        "success": True,
        "project_id": project_id,
        "message": "视频处理完成",
        "steps": steps
    }

@celery_app.task(bind=True, name='backend.tasks.processing.process_single_step')
def backend_process_single_step(self, project_id: str, step: str, config: dict, *args, **kwargs):
    """后端单个步骤处理任务（兼容性）"""
    return process_single_step(self, project_id, step, config, *args, **kwargs)

if __name__ == '__main__':
    celery_app.start()