apps/worker/celery_app.py

@@ -1,12 +0,0 @@
-import os
-from celery import Celery
-
-
-def create_celery_app() -> Celery:
-    app = Celery("xiaoxia_saas_worker")
-    app.conf.broker_url = os.getenv("CELERY_BROKER_URL", "redis://redis:6379/0")
-    app.conf.result_backend = os.getenv("CELERY_RESULT_BACKEND", "redis://redis:6379/1")
-    return app
-
-
-celery_app = create_celery_app()

apps/worker/video_processing/dedup.py

@@ -14,16 +14,12 @@ from celery import Task
 from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.generated_video_repository import SQLAlchemyGeneratedVideoRepository
-from packages.adapters.sqlalchemy_impl.session import SessionLocal, build_session_factory
-from packages.shared.config import get_shared_settings
 from packages.shared.storage import get_storage_service
 
-from apps.worker.celery_app import celery_app
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
 
 logger = logging.getLogger(__name__)
-settings = get_shared_settings()
-if SessionLocal is None:
-    build_session_factory(settings.database_url)
 
 
 def compute_phash(image: np.ndarray, hash_size: int = 8) -> str:

apps/worker/worker_app/celery_app.py

@@ -5,9 +5,12 @@ settings = get_settings()
 celery_app = Celery(settings.worker_name)
 celery_app.conf.broker_url = settings.broker_url
 celery_app.conf.result_backend = settings.result_backend
+celery_app.conf.broker_connection_retry_on_startup = True
 celery_app.conf.imports = (
     "worker_app.tasks.health",
     "worker_app.tasks.ingest",
     "worker_app.tasks.classification",
     "worker_app.tasks.generation",
+    "worker_app.tasks.voice_extraction",
+    "apps.worker.video_processing.dedup",
 )

apps/worker/worker_app/tasks/classification.py

@@ -13,12 +13,14 @@ from packages.domain import (
     ClassificationStatus,
 )
 
+from worker_app.celery_app import celery_app
 from .asset_analyzer import classify_asset_real
 
 logger = get_task_logger(__name__)
 
 
-def classify_asset(job_id: str) -> dict:
+@celery_app.task(bind=True, name="worker.classify_asset", max_retries=2)
+def classify_asset(self, job_id: str) -> dict:
     """
     Classify asset task.
 

apps/worker/worker_app/tasks/generation.py

@@ -14,6 +14,8 @@ from typing import Optional
 
 import oss2
 
+from worker_app.celery_app import celery_app
+
 OUTPUT_WIDTH = 1280
 OUTPUT_HEIGHT = 720
 OUTPUT_FPS = 25.0
@@ -214,29 +216,37 @@ def _process_with_editing_mode(
     )
 
 
-def generate_video(
-    task_id: str,
-    workspace_id: str,
-    project_id: str,
-    asset_library_id: str,
-    voice_library_id: str = "",
-    mode: str = "one_take",
-) -> dict:
+@celery_app.task(bind=True, name="worker.generate_video", max_retries=2)
+def generate_video(self, task_id: str) -> dict:
     """
     生成视频任务
 
     Args:
-        task_id: 任务 ID
-        workspace_id: 工作空间 ID
-        project_id: 项目 ID
-        asset_library_id: 素材库 ID
-        voice_library_id: 配音库 ID（可选）
-        mode: 剪辑模式，默认 one_take
+        task_id: 任务 ID（从数据库加载完整任务信息）
 
     Returns:
         生成结果字典
     """
     from packages.domain import GeneratedVideo, GenerationMode, GenerationTaskStatus
+    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
+        SQLAlchemyGenerationTaskRepository,
+    )
+    from worker_app.db import SessionLocal
+
+    # 从数据库加载任务信息
+    session = SessionLocal()
+    try:
+        task_repo = SQLAlchemyGenerationTaskRepository(session)
+        gen_task = task_repo.get(task_id)
+        if gen_task is None:
+            return {"status": "failed", "error": f"generation task {task_id} not found"}
+        workspace_id = gen_task.workspace_id
+        project_id = gen_task.project_id
+        asset_library_id = gen_task.asset_library_id
+        voice_library_id = gen_task.voice_library_id or ""
+        mode = gen_task.strategy_id or "one_take"
+    finally:
+        session.close()
 
     try:
         editing_mode = GenerationMode(mode)

apps/worker/worker_app/tasks/voice_extraction.py

@@ -10,16 +10,12 @@ from celery import Task
 from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
-from packages.adapters.sqlalchemy_impl.session import SessionLocal, build_session_factory
-from packages.shared.config import get_shared_settings
 from packages.shared.storage import get_storage_service
 
-from .celery_app import celery_app
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
 
 logger = logging.getLogger(__name__)
-settings = get_shared_settings()
-if SessionLocal is None:
-    build_session_factory(settings.database_url)
 
 
 class VoiceExtractor:

infra/docker/worker.Dockerfile

@@ -36,6 +36,13 @@ COPY migrations/ /app/migrations/
 ENV PYTHONPATH=/app
 ENV PYTHONUNBUFFERED=1
 
+# 创建非 root 用户运行 Worker
+RUN groupadd -r celery && useradd -r -g celery -d /app -s /sbin/nologin celery \
+    && chown -R celery:celery /app
+RUN mkdir -p /app/generated && chown celery:celery /app/generated
+
+USER celery
+
 # Worker 入口点
 WORKDIR /app/apps/worker
-CMD ["celery", "-A", "celery_app", "worker", "--loglevel=info", "--concurrency=2"]
+CMD ["celery", "-A", "worker_app.celery_app", "worker", "--loglevel=info", "--concurrency=2"]