Celery Worker Fix Agent
8210806632
1. 修复 Worker 启动入口: celery_app → worker_app.celery_app - 旧 celery_app.py 不导入任何任务模块，导致 Worker 注册 0 个任务 - 删除旧版 apps/worker/celery_app.py，统一使用 worker_app/celery_app.py 2. 为缺少装饰器的任务补充 @celery_app.task: - classification.py: classify_asset() 添加装饰器 - generation.py: generate_video() 添加装饰器，修正签名匹配 API 调用方式 3. 确保 voice_extraction 任务被正确注册: - 添加 voice_extraction 到 celery_app imports - 修复 voice_extraction.py 中错误的相对导入 (.celery_app → worker_app.celery_app) - 修复 dedup.py 中指向已删除模块的导入 4. 修复 worker_app/celery_app.py: - 添加 broker_connection_retry_on_startup=True - imports 中添加 voice_extraction 和 dedup 模块 5. 修复 Dockerfile: - CMD 改为 celery -A worker_app.celery_app - 添加非 root 用户 celery 运行 Worker 6. 新建 packages/shared/config.py 和 storage.py 兼容层 - 为 worker 任务模块提供统一的 config/storage 访问入口
112 lines
3.5 KiB
Python
Executable File
112 lines
3.5 KiB
Python
Executable File
from celery import Task
|
|
from celery.utils.log import get_task_logger
|
|
|
|
from packages.adapters.sqlalchemy_impl.classification_job_repository import (
|
|
SQLAlchemyClassificationJobRepository,
|
|
)
|
|
from packages.adapters.sqlalchemy_impl.asset_repository import (
|
|
SQLAlchemyAssetRepository,
|
|
)
|
|
from packages.domain import (
|
|
ClassificationJob,
|
|
ClassificationJobStatus,
|
|
ClassificationStatus,
|
|
)
|
|
|
|
from worker_app.celery_app import celery_app
|
|
from .asset_analyzer import classify_asset_real
|
|
|
|
logger = get_task_logger(__name__)
|
|
|
|
|
|
@celery_app.task(bind=True, name="worker.classify_asset", max_retries=2)
|
|
def classify_asset(self, job_id: str) -> dict:
|
|
"""
|
|
Classify asset task.
|
|
|
|
Steps:
|
|
1. Fetch ClassificationJob from repository
|
|
2. Fetch Asset from repository
|
|
3. Run real classification based on video features
|
|
4. Update ClassificationJob with result
|
|
5. Update Asset with classification and status
|
|
6. Return result
|
|
"""
|
|
from worker_app.db import SessionLocal
|
|
|
|
# 创建数据库 session 和 repository
|
|
session = SessionLocal()
|
|
try:
|
|
job_repo = SQLAlchemyClassificationJobRepository(session)
|
|
asset_repo = SQLAlchemyAssetRepository(session)
|
|
|
|
job = job_repo.get(job_id)
|
|
if job is None:
|
|
return {"status": "failed", "error": "job not found"}
|
|
|
|
try:
|
|
# Update job status to PROCESSING
|
|
job.status = ClassificationJobStatus.PROCESSING
|
|
job_repo.update(job)
|
|
session.commit()
|
|
|
|
# Get the asset to find the video path
|
|
asset = asset_repo.get(job.asset_id)
|
|
if asset is None:
|
|
raise ValueError(f"Asset not found: {job.asset_id}")
|
|
|
|
# Determine media path from storage_key
|
|
# In production, this would be a full URL/path to the media file
|
|
video_path = asset.storage_key
|
|
|
|
# Run real classification
|
|
classification, confidence = classify_asset_real(video_path)
|
|
|
|
# Update job with classification result
|
|
job.status = ClassificationJobStatus.COMPLETED
|
|
job.classification = classification
|
|
job.confidence = confidence
|
|
job_repo.update(job)
|
|
|
|
# Update asset with classification status and result
|
|
asset.classification_status = ClassificationStatus.COMPLETED
|
|
asset_repo.update(asset)
|
|
|
|
session.commit()
|
|
|
|
logger.info(
|
|
f"Classification completed for asset {asset.id}: "
|
|
f"category={classification}, confidence={confidence}"
|
|
)
|
|
|
|
return {
|
|
"status": "completed",
|
|
"job_id": job.id,
|
|
"classification": classification,
|
|
"confidence": confidence,
|
|
}
|
|
except Exception as e:
|
|
session.rollback()
|
|
logger.error(f"Classification failed for job {job_id}: {e}")
|
|
|
|
# Update job status to FAILED
|
|
job.status = ClassificationJobStatus.FAILED
|
|
job.error_message = str(e)
|
|
job_repo.update(job)
|
|
|
|
# Update asset classification status to FAILED
|
|
asset = asset_repo.get(job.asset_id)
|
|
if asset:
|
|
asset.classification_status = ClassificationStatus.FAILED
|
|
asset_repo.update(asset)
|
|
|
|
session.commit()
|
|
|
|
return {
|
|
"status": "failed",
|
|
"job_id": job.id,
|
|
"error": str(e),
|
|
}
|
|
finally:
|
|
session.close()
|