apps/worker/worker_app/tasks/ingest.py

@@ -1,7 +1,10 @@
 import subprocess
+import tempfile
 from datetime import datetime, timezone
+from pathlib import Path
 
 from celery.utils.log import get_task_logger
+from video_processing.oss_helpers import download_asset
 from worker_app.celery_app import celery_app
 from worker_app.core.asset_types import infer_mime_type_from_storage_key
 from worker_app.db import SessionLocal
@@ -227,9 +230,26 @@ def ingest_asset(job_id: str) -> dict:
         elif mime_type.startswith("audio/"):
             media_type = "audio"
 
-        # Extract metadata
-        storage_url = job.storage_key  # Assuming storage_key is usable as URL/path
-        metadata, extract_success = extract_media_metadata(storage_url, media_type)
+        # 先从 OSS 下载文件到本地临时目录，再提取元数据
+        # （storage_key 是 OSS 内部路径，不能直接传给 ffprobe/Pillow）
+        local_file = None
+        try:
+            suffix = Path(job.storage_key).suffix or ".bin"
+            with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
+                local_file = Path(tmp.name)
+
+            download_ok = download_asset(job.storage_key, local_file)
+            if not download_ok:
+                logger.warning("素材下载失败，无法提取元数据: job_id=%s storage_key=%s", job_id, job.storage_key)
+                metadata, extract_success = {}, False
+            else:
+                metadata, extract_success = extract_media_metadata(str(local_file), media_type)
+        finally:
+            if local_file and local_file.exists():
+                try:
+                    local_file.unlink()
+                except OSError:
+                    pass
 
         # 有效性校验：ffprobe/Pillow 必须成功，且文件大小/时长/尺寸满足最小要求
         is_valid = extract_success and _is_valid_media(metadata, media_type)

tests/unit/test_ingest_validation.py

@@ -171,18 +171,19 @@ class TestIngestAssetValidation:
         job_repo.create(job)
         db.commit()
 
-        # mock extract_media_metadata 返回失败
-        with patch("worker_app.tasks.ingest.extract_media_metadata") as mock_extract:
-            mock_extract.return_value = (
-                {"size_bytes": 27, "duration": 0, "width": 0, "height": 0},
-                False,
-            )
-            # mock SessionLocal 返回我们的db session
-            with patch("worker_app.tasks.ingest.SessionLocal", return_value=db):
-                # mock celery_app.task decorator不影响函数本身
-                from worker_app.tasks.ingest import ingest_asset
+        # mock download_asset 成功 + extract_media_metadata 返回失败
+        with patch("worker_app.tasks.ingest.download_asset", return_value=True):
+            with patch("worker_app.tasks.ingest.extract_media_metadata") as mock_extract:
+                mock_extract.return_value = (
+                    {"size_bytes": 27, "duration": 0, "width": 0, "height": 0},
+                    False,
+                )
+                # mock SessionLocal 返回我们的db session
+                with patch("worker_app.tasks.ingest.SessionLocal", return_value=db):
+                    # mock celery_app.task decorator不影响函数本身
+                    from worker_app.tasks.ingest import ingest_asset
 
-                result = ingest_asset("job-invalid-001")
+                    result = ingest_asset("job-invalid-001")
 
         assert result["status"] == "failed"
         assert "asset_id" in result
@@ -233,23 +234,24 @@ class TestIngestAssetValidation:
         job_repo.create(job)
         db.commit()
 
-        with patch("worker_app.tasks.ingest.extract_media_metadata") as mock_extract:
-            mock_extract.return_value = (
-                {
-                    "size_bytes": 5 * 1024 * 1024,
-                    "duration": 30.5,
-                    "width": 1920,
-                    "height": 1080,
-                    "codec": "h264",
-                    "fps": 30.0,
-                    "bitrate": 2000000,
-                },
-                True,
-            )
-            with patch("worker_app.tasks.ingest.SessionLocal", return_value=db):
-                from worker_app.tasks.ingest import ingest_asset
+        with patch("worker_app.tasks.ingest.download_asset", return_value=True):
+            with patch("worker_app.tasks.ingest.extract_media_metadata") as mock_extract:
+                mock_extract.return_value = (
+                    {
+                        "size_bytes": 5 * 1024 * 1024,
+                        "duration": 30.5,
+                        "width": 1920,
+                        "height": 1080,
+                        "codec": "h264",
+                        "fps": 30.0,
+                        "bitrate": 2000000,
+                    },
+                    True,
+                )
+                with patch("worker_app.tasks.ingest.SessionLocal", return_value=db):
+                    from worker_app.tasks.ingest import ingest_asset
 
-                result = ingest_asset("job-valid-001")
+                    result = ingest_asset("job-valid-001")
 
         assert result["status"] == "completed"
         assert "asset_id" in result
@@ -265,3 +267,116 @@ class TestIngestAssetValidation:
         assert updated_job.status == IngestJobStatus.COMPLETED
 
         db.close()
+
+    def test_download_failure_marked_as_error(self):
+        """OSS下载失败 → 无法提取元数据 → asset.status=ERROR."""
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
+
+        from packages.adapters.sqlalchemy_impl.models import Base
+        from packages.domain import AssetStatus, IngestJobStatus
+
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        Session = sessionmaker(bind=engine)
+        db = Session()
+
+        from packages.adapters.sqlalchemy_impl import (
+            SQLAlchemyAssetRepository,
+            SQLAlchemyIngestJobRepository,
+        )
+        from packages.domain import IngestJob
+
+        job_repo = SQLAlchemyIngestJobRepository(db)
+        asset_repo = SQLAlchemyAssetRepository(db)
+
+        job = IngestJob(
+            id="job-dl-fail-001",
+            project_id="proj-1",
+            library_id="lib-1",
+            storage_key="uploads/test/nonexistent.mp4",
+            status=IngestJobStatus.PENDING,
+        )
+        job_repo.create(job)
+        db.commit()
+
+        # mock download_asset 返回失败
+        with patch("worker_app.tasks.ingest.download_asset", return_value=False):
+            with patch("worker_app.tasks.ingest.SessionLocal", return_value=db):
+                from worker_app.tasks.ingest import ingest_asset
+
+                result = ingest_asset("job-dl-fail-001")
+
+        assert result["status"] == "failed"
+        assert "asset_id" in result
+
+        asset = asset_repo.get(result["asset_id"])
+        assert asset is not None
+        assert asset.status == AssetStatus.ERROR
+
+        updated_job = job_repo.get("job-dl-fail-001")
+        assert updated_job.status == IngestJobStatus.FAILED
+
+        db.close()
+
+    def test_temp_file_cleaned_up_after_extraction(self):
+        """提取元数据后临时文件被清理."""
+        import tempfile
+
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
+
+        from packages.adapters.sqlalchemy_impl.models import Base
+        from packages.domain import IngestJobStatus
+
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        Session = sessionmaker(bind=engine)
+        db = Session()
+
+        from packages.adapters.sqlalchemy_impl import SQLAlchemyIngestJobRepository
+        from packages.domain import IngestJob
+
+        job_repo = SQLAlchemyIngestJobRepository(db)
+
+        job = IngestJob(
+            id="job-cleanup-001",
+            project_id="proj-1",
+            library_id="lib-1",
+            storage_key="uploads/test/cleanup.mp4",
+            status=IngestJobStatus.PENDING,
+        )
+        job_repo.create(job)
+        db.commit()
+
+        temp_files_created = []
+        original_namedtempfile = tempfile.NamedTemporaryFile
+
+        def tracking_namedtempfile(*args, **kwargs):
+            tmp = original_namedtempfile(*args, **kwargs)
+            temp_files_created.append(tmp.name)
+            return tmp
+
+        def fake_download(storage_key, local_path):
+            # 模拟下载：写点假数据
+            Path(local_path).write_bytes(b"fake video data" * 100)
+            return True
+
+        with patch("tempfile.NamedTemporaryFile", side_effect=tracking_namedtempfile):
+            with patch("worker_app.tasks.ingest.download_asset", side_effect=fake_download):
+                with patch("worker_app.tasks.ingest.extract_media_metadata") as mock_extract:
+                    mock_extract.return_value = (
+                        {"size_bytes": 5 * 1024 * 1024, "duration": 30.5, "width": 1920, "height": 1080},
+                        True,
+                    )
+                    with patch("worker_app.tasks.ingest.SessionLocal", return_value=db):
+                        from worker_app.tasks.ingest import ingest_asset
+
+                        ingest_asset("job-cleanup-001")
+
+        # 验证临时文件已被清理
+        assert len(temp_files_created) > 0
+        for f in temp_files_created:
+            assert not Path(f).exists(), f"临时文件未被清理: {f}"
+
+        db.close()