xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 9 Pull Requests 2 Actions Packages Projects Releases 7 Wiki Activity

fix(worker): 修复素材下载——asset_id查assets表拿file_url + 统一OSS配置读取 #471

Merged
auto-approve-bot merged 3 commits from fix/worker-oss-settings-unified into develop 2026-07-17 15:37:41 +08:00
Conversation 2 Commits 3 Files Changed 5
+253 -186
5 changed files with 253 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/worker/video_processing/oss_helpers.py
+10 -4
View File
@@ -15,6 +15,8 @@ from urllib.parse import urlparse
import oss2
from packages.shared.config import get_shared_settings
logger = logging.getLogger(__name__)
# OSS 上传配置
@@ -31,14 +33,18 @@ OSS_MULTIPART_NUM_THREADS = 3 # 分片上传并发数
def oss_settings() -> tuple[str, str, str, str] | None:
"""获取 OSS 配置。
统一使用 SharedSettings 读取配置，与 SharedStorageService 保持一致，
支持从 .env 文件加载，避免两套配置路径不一致。
Returns:
(access_key_id, access_key_secret, endpoint, bucket_name) 元组，
配置缺失时返回 None。
"""
access_key_id = os.getenv("OSS_ACCESS_KEY_ID")
access_key_secret = os.getenv("OSS_ACCESS_KEY_SECRET")
endpoint = os.getenv("OSS_ENDPOINT")
bucket_name = os.getenv("OSS_BUCKET_NAME")
settings = get_shared_settings()
access_key_id = settings.oss_access_key_id
access_key_secret = settings.oss_access_key_secret
endpoint = settings.oss_endpoint
bucket_name = settings.oss_bucket_name
if not all([access_key_id, access_key_secret, endpoint, bucket_name]):
return None
return access_key_id, access_key_secret, endpoint, bucket_name
apps/worker/video_processing/render_adapter.py
+30 -5
View File
@@ -27,6 +27,7 @@ from video_processing.unified_render_service import UnifiedRenderService
from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import SQLAlchemyEditPlanClipRepository
from packages.adapters.sqlalchemy_impl.edit_plan_repository import SQLAlchemyEditPlanRepository
from packages.adapters.sqlalchemy_impl.models import AssetModel
from packages.domain.edit_plan import EditPlanStatus
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
@@ -294,10 +295,15 @@ class RenderAdapter:
except Exception:
logger.exception("进度回调失败")
@staticmethod
def _download_assets(clips: list[EditPlanClip], work_dir: Path) -> tuple[dict[str, Path], list[str], list[str]]:
def _download_assets(
self, clips: list[EditPlanClip], work_dir: Path
) -> tuple[dict[str, Path], list[str], list[str]]:
"""下载片段素材到本地。
先通过 asset_id 批量查询 assets 表获取 file_url（OSS存储路径），
再用 file_url 作为 OSS key 下载。asset_id 是 UUID 主键，
不能直接当作 OSS 存储路径使用。
Returns:
(asset_path_map, rendered_clip_ids, failed_clip_ids)
- asset_path_map: asset_id → local_path 映射（下载成功的）
@@ -312,6 +318,13 @@ class RenderAdapter:
failed_clip_ids: list[str] = []
seen_asset_ids: set[str] = set()
# 批量查询素材的 file_url（OSS 存储路径）
clip_asset_ids = [c.asset_id for c in clips if c.asset_id]
asset_storage_map: dict[str, str] = {}
if clip_asset_ids:
assets = self._db.query(AssetModel).filter(AssetModel.id.in_(clip_asset_ids)).all()
asset_storage_map = {a.id: a.file_url for a in assets if a.file_url}
for clip in clips:
asset_id = clip.asset_id
if not asset_id:
@@ -328,11 +341,23 @@ class RenderAdapter:
seen_asset_ids.add(asset_id)
# 生成安全的本地文件名
safe_name = f"clip_{clip.order:04d}_{abs(hash(asset_id)) % 100000:05d}.mp4"
# 从素材表获取 OSS 存储路径（file_url）
storage_key = asset_storage_map.get(asset_id)
if not storage_key:
logger.warning(
"素材无 file_url，跳过下载: clip_id=%s asset_id=%s",
clip.id,
asset_id,
)
failed_clip_ids.append(clip.id)
continue
# 生成安全的本地文件名（保留原始扩展名）
ext = Path(storage_key).suffix or ".mp4"
safe_name = f"clip_{clip.order:04d}_{abs(hash(asset_id)) % 100000:05d}{ext}"
local_path = asset_dir / safe_name
if download_asset(asset_id, local_path):
if download_asset(storage_key, local_path):
asset_path_map[asset_id] = local_path
rendered_clip_ids.append(clip.id)
logger.debug("素材下载成功: clip_id=%s asset_id=%s", clip.id, asset_id[:60])
tests/unit/test_oss_upload_crash_fix.py
+17 -43
View File
@@ -9,7 +9,6 @@
from __future__ import annotations
import os
import tempfile
import time
from pathlib import Path
@@ -27,14 +26,9 @@ class TestOSSBucketConnectTimeout:
mock_bucket_instance = MagicMock()
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
@@ -53,7 +47,7 @@ class TestOSSBucketConnectTimeout:
"""OSS 配置缺失时返回 None."""
from video_processing.oss_helpers import oss_bucket
with patch.dict(os.environ, {}, clear=True):
with patch("video_processing.oss_helpers.oss_settings", return_value=None):
bucket = oss_bucket()
assert bucket is None
@@ -80,14 +74,9 @@ class TestUploadToOSSMultipart:
mock_bucket = MagicMock()
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -113,14 +102,9 @@ class TestUploadToOSSMultipart:
mock_bucket = MagicMock()
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -167,14 +151,9 @@ class TestUploadToOSSTimeout:
try:
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -200,14 +179,9 @@ class TestUploadToOSSTimeout:
try:
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -227,7 +201,7 @@ class TestUploadToOSSTimeout:
file_path = Path(small_file.name)
try:
with patch.dict(os.environ, {}, clear=True):
with patch("video_processing.oss_helpers.oss_settings", return_value=None):
url = upload_to_oss(file_path, "test/noconfig.mp4")
assert url is None
finally:
tests/unit/test_p02_worker_oss_fix.py
+31 -86
View File
@@ -8,7 +8,7 @@
from __future__ import annotations
import os
from pathlib import Path
from unittest.mock import MagicMock, patch
# ── oss_bucket endpoint scheme 修复 ──────────────────────────────────────────
@@ -23,20 +23,13 @@ class TestOSSBucketEndpointScheme:
mock_bucket_instance = MagicMock()
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
):
# 清除缓存，确保重新创建
bucket = oss_bucket()
assert bucket is mock_bucket_instance
@@ -52,19 +45,13 @@ class TestOSSBucketEndpointScheme:
mock_bucket_instance = MagicMock()
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "https://oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
):
oss_bucket()
call_args = mock_bucket_cls.call_args
@@ -79,19 +66,13 @@ class TestOSSBucketEndpointScheme:
mock_bucket_instance = MagicMock()
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "http://oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "http://oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
):
oss_bucket()
call_args = mock_bucket_cls.call_args
@@ -102,17 +83,7 @@ class TestOSSBucketEndpointScheme:
"""凭证缺失时返回 None."""
from video_processing.oss_helpers import oss_bucket
with patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "",
"OSS_ACCESS_KEY_SECRET": "",
"OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
clear=True,
):
with patch("video_processing.oss_helpers.oss_settings", return_value=None):
bucket = oss_bucket()
assert bucket is None
@@ -128,17 +99,15 @@ class TestGetSignedDownloadUrl:
from video_processing.oss_helpers import get_signed_download_url
mock_bucket = MagicMock()
mock_bucket.sign_url.return_value = "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx"
mock_bucket.sign_url.return_value = (
"https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4"
"?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx"
)
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -159,14 +128,9 @@ class TestGetSignedDownloadUrl:
)
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -183,7 +147,7 @@ class TestGetSignedDownloadUrl:
"""bucket 为 None 时返回 None（不抛异常）."""
from video_processing.oss_helpers import get_signed_download_url
with patch.dict(os.environ, {}, clear=True):
with patch("video_processing.oss_helpers.oss_settings", return_value=None):
result = get_signed_download_url("generated/test.mp4")
assert result is None
@@ -195,14 +159,9 @@ class TestGetSignedDownloadUrl:
mock_bucket.sign_url.side_effect = Exception("sign failed")
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -224,17 +183,10 @@ class TestUploadToOSSReturnsHTTPS:
mock_bucket = MagicMock()
mock_bucket.put_object_from_file = MagicMock()
from pathlib import Path
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -252,17 +204,10 @@ class TestUploadToOSSReturnsHTTPS:
mock_bucket = MagicMock()
mock_bucket.put_object_from_file = MagicMock()
from pathlib import Path
with (
patch.dict(
os.environ,
{
"OSS_ACCESS_KEY_ID": "test-key",
"OSS_ACCESS_KEY_SECRET": "test-secret",
"OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
"OSS_BUCKET_NAME": "test-bucket",
},
patch(
"video_processing.oss_helpers.oss_settings",
return_value=("test-key", "test-secret", "https://oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
),
patch("video_processing.oss_helpers.oss2.Auth"),
patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
tests/unit/test_render_adapter.py
+165 -48
View File
@@ -70,13 +70,33 @@ def _make_clip(
def _make_adapter(
plan: FakePlan | None = None,
clips: list[FakeClip] | None = None,
asset_url_map: dict[str, str] | None = None,
) -> tuple[RenderAdapter, MagicMock, MagicMock]:
"""创建测试用的 RenderAdapter 及 mock repo。
Args:
plan: 模拟的剪辑计划
clips: 模拟的片段列表
asset_url_map: asset_id → file_url 映射，用于 mock assets 表查询
Returns:
(adapter, mock_plan_repo, mock_clip_repo)
"""
mock_db = MagicMock()
# mock assets 表查询：_download_assets 用
if asset_url_map is not None:
mock_query = MagicMock()
mock_assets = []
for aid, url in asset_url_map.items():
m = MagicMock()
m.id = aid
m.file_url = url
mock_assets.append(m)
mock_query.all.return_value = mock_assets
mock_query.filter.return_value = mock_query
mock_db.query.return_value = mock_query
adapter = RenderAdapter(mock_db)
# 替换内部 repo
@@ -229,7 +249,8 @@ class TestRenderPlan:
plan = FakePlan(id="plan_001", status="editing")
clips = [_make_clip("c1", order=0, duration=5.0)]
adapter, _, _ = _make_adapter(plan=plan, clips=clips)
asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
result = adapter.render_plan("plan_001")
@@ -243,7 +264,7 @@ class TestRenderPlan:
"""完整渲染流程成功。"""
# 素材下载成功
def _fake_download(asset_id, local_path):
def _fake_download(storage_key, local_path):
local_path.parent.mkdir(parents=True, exist_ok=True)
local_path.write_bytes(b"fake video data")
return True
@@ -269,7 +290,11 @@ class TestRenderPlan:
_make_clip("c1", order=0, duration=5.0),
_make_clip("c2", order=1, duration=5.0),
]
adapter, _, _ = _make_adapter(plan=plan, clips=clips)
asset_url_map = {
"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4",
"asset_c2.mp4": "https://test-bucket.oss.com/assets/asset_c2.mp4",
}
adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
result = adapter.render_plan(
"plan_001",
@@ -295,7 +320,7 @@ class TestRenderPlan:
def test_progress_callback(self, mock_download, tmp_path):
"""进度回调被正确触发。"""
def _fake_download(asset_id, local_path):
def _fake_download(storage_key, local_path):
local_path.parent.mkdir(parents=True, exist_ok=True)
local_path.write_bytes(b"fake data")
return True
@@ -310,7 +335,8 @@ class TestRenderPlan:
plan = FakePlan(id="plan_001", status="editing")
clips = [_make_clip("c1", order=0, duration=5.0)]
adapter, _, _ = _make_adapter(plan=plan, clips=clips)
asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
progress_values = []
@@ -333,8 +359,8 @@ class TestRenderPlan:
"""部分素材下载失败时，只使用成功的素材。"""
download_results = [True, False, True] # 3个素材中2个成功
def _fake_download(asset_id, local_path):
idx = hash(asset_id) % 3
def _fake_download(storage_key, local_path):
idx = hash(storage_key) % 3
if download_results[idx]:
local_path.parent.mkdir(parents=True, exist_ok=True)
local_path.write_bytes(b"fake data")
@@ -357,11 +383,16 @@ class TestRenderPlan:
with patch("video_processing.render_adapter.upload_to_oss", return_value="https://example.com/out.mp4"):
plan = FakePlan(id="plan_001", status="editing")
clips = [
_make_clip("c1", order=0, duration=3.0, asset_id="asset_001.mp4"),
_make_clip("c2", order=1, duration=3.0, asset_id="asset_002.mp4"),
_make_clip("c3", order=2, duration=3.0, asset_id="asset_003.mp4"),
_make_clip("c1", order=0, duration=3.0, asset_id="asset_001"),
_make_clip("c2", order=1, duration=3.0, asset_id="asset_002"),
_make_clip("c3", order=2, duration=3.0, asset_id="asset_003"),
]
adapter, _, _ = _make_adapter(plan=plan, clips=clips)
asset_url_map = {
"asset_001": "https://test-bucket.oss.com/assets/001.mp4",
"asset_002": "https://test-bucket.oss.com/assets/002.mp4",
"asset_003": "https://test-bucket.oss.com/assets/003.mp4",
}
adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
result = adapter.render_plan(
"plan_001",
@@ -377,38 +408,75 @@ class TestRenderPlan:
class TestDownloadAssets:
@staticmethod
def _make_mock_db(asset_url_map: dict[str, str]):
"""构造 mock db，根据 asset_id 返回对应的 AssetModel.file_url。"""
mock_db = MagicMock()
mock_query = MagicMock()
def _fake_filter(query):
# 模拟 .filter(AssetModel.id.in_([...])).all()
mock_assets = []
for asset_id, file_url in asset_url_map.items():
mock_asset = MagicMock()
mock_asset.id = asset_id
mock_asset.file_url = file_url
mock_assets.append(mock_asset)
mock_query.all.return_value = mock_assets
return mock_query
mock_query.filter.side_effect = _fake_filter
mock_db.query.return_value = mock_query
return mock_db
@patch("video_processing.render_adapter.download_asset")
def test_all_download_success(self, mock_download, tmp_path):
"""全部素材下载成功。"""
mock_download.return_value = True
asset_url_map = {
"asset_001": "https://bucket.oss.com/videos/key1.mp4",
"asset_002": "https://bucket.oss.com/videos/key2.mp4",
}
mock_db = self._make_mock_db(asset_url_map)
adapter = RenderAdapter(mock_db)
clips = [
_make_clip("c1", order=0, asset_id="key1.mp4"),
_make_clip("c2", order=1, asset_id="key2.mp4"),
_make_clip("c1", order=0, asset_id="asset_001"),
_make_clip("c2", order=1, asset_id="asset_002"),
]
asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
assert len(asset_path_map) == 2
assert "key1.mp4" in asset_path_map
assert "key2.mp4" in asset_path_map
assert "asset_001" in asset_path_map
assert "asset_002" in asset_path_map
assert len(rendered_ids) == 2
assert len(failed_ids) == 0
assert mock_download.call_count == 2
# 验证传给 download_asset 的是 file_url 而非 asset_id
call_keys = [call[0][0] for call in mock_download.call_args_list]
assert "https://bucket.oss.com/videos/key1.mp4" in call_keys
assert "https://bucket.oss.com/videos/key2.mp4" in call_keys
@patch("video_processing.render_adapter.download_asset")
def test_empty_asset_id_skipped(self, mock_download, tmp_path):
"""空 asset_id 的片段被跳过。"""
clips = [
_make_clip("c1", order=0, asset_id=""),
_make_clip("c2", order=1, asset_id="key2.mp4"),
]
mock_download.return_value = True
asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
asset_url_map = {"asset_002": "https://bucket.oss.com/videos/key2.mp4"}
mock_db = self._make_mock_db(asset_url_map)
adapter = RenderAdapter(mock_db)
clips = [
_make_clip("c1", order=0, asset_id=""),
_make_clip("c2", order=1, asset_id="asset_002"),
]
asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
assert len(asset_path_map) == 1
assert "key2.mp4" in asset_path_map
assert "asset_002" in asset_path_map
assert "c1" in failed_ids
assert "c2" in rendered_ids
assert mock_download.call_count == 1 # 只调用了一次下载
@@ -418,11 +486,15 @@ class TestDownloadAssets:
"""全部下载失败返回空字典。"""
mock_download.return_value = False
asset_url_map = {"asset_001": "https://bucket.oss.com/videos/key1.mp4"}
mock_db = self._make_mock_db(asset_url_map)
adapter = RenderAdapter(mock_db)
clips = [
_make_clip("c1", order=0, asset_id="key1.mp4"),
_make_clip("c1", order=0, asset_id="asset_001"),
]
asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
assert len(asset_path_map) == 0
assert len(rendered_ids) == 0
@@ -431,22 +503,34 @@ class TestDownloadAssets:
@patch("video_processing.render_adapter.download_asset")
def test_partial_download_failure(self, mock_download, tmp_path):
"""部分下载失败时正确区分成功/失败。"""
results = {"key1.mp4": True, "key2.mp4": False, "key3.mp4": True}
results = {
"https://bucket.oss.com/videos/key1.mp4": True,
"https://bucket.oss.com/videos/key2.mp4": False,
"https://bucket.oss.com/videos/key3.mp4": True,
}
def _fake_download(asset_id, local_path):
def _fake_download(storage_key, local_path):
local_path.parent.mkdir(parents=True, exist_ok=True)
local_path.write_bytes(b"fake")
return results.get(asset_id, False)
return results.get(storage_key, False)
mock_download.side_effect = _fake_download
asset_url_map = {
"asset_001": "https://bucket.oss.com/videos/key1.mp4",
"asset_002": "https://bucket.oss.com/videos/key2.mp4",
"asset_003": "https://bucket.oss.com/videos/key3.mp4",
}
mock_db = self._make_mock_db(asset_url_map)
adapter = RenderAdapter(mock_db)
clips = [
_make_clip("c1", order=0, asset_id="key1.mp4"),
_make_clip("c2", order=1, asset_id="key2.mp4"),
_make_clip("c3", order=2, asset_id="key3.mp4"),
_make_clip("c1", order=0, asset_id="asset_001"),
_make_clip("c2", order=1, asset_id="asset_002"),
_make_clip("c3", order=2, asset_id="asset_003"),
]
asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
assert len(asset_path_map) == 2
assert "c1" in rendered_ids
@@ -458,18 +542,44 @@ class TestDownloadAssets:
"""同一素材被多个 clip 引用时只下载一次。"""
mock_download.return_value = True
asset_url_map = {"asset_shared": "https://bucket.oss.com/videos/shared.mp4"}
mock_db = self._make_mock_db(asset_url_map)
adapter = RenderAdapter(mock_db)
clips = [
_make_clip("c1", order=0, asset_id="shared.mp4"),
_make_clip("c2", order=1, asset_id="shared.mp4"),
_make_clip("c1", order=0, asset_id="asset_shared"),
_make_clip("c2", order=1, asset_id="asset_shared"),
]
asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
assert len(asset_path_map) == 1
assert mock_download.call_count == 1
assert "c1" in rendered_ids
assert "c2" in rendered_ids
@patch("video_processing.render_adapter.download_asset")
def test_asset_without_file_url_skipped(self, mock_download, tmp_path):
"""素材在assets表中无file_url时跳过下载，标记为失败。"""
# 构造返回 asset 但 file_url 为空
mock_db = MagicMock()
mock_query = MagicMock()
mock_asset = MagicMock()
mock_asset.id = "asset_no_url"
mock_asset.file_url = ""
mock_query.all.return_value = [mock_asset]
mock_query.filter.return_value = mock_query
mock_db.query.return_value = mock_query
adapter = RenderAdapter(mock_db)
clips = [_make_clip("c1", order=0, asset_id="asset_no_url")]
asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
assert len(asset_path_map) == 0
assert "c1" in failed_ids
mock_download.assert_not_called()
# ── _prepare_bgm 测试 ────────────────────────────────────────────────────────
@@ -674,7 +784,7 @@ class TestRenderPlanWithBgmAsr:
def test_bgm_passed_to_render_service(self, mock_download, mock_render_cls, mock_upload, tmp_path):
"""BGM 路径被正确传递给 UnifiedRenderService。"""
def _fake_download(asset_id, local_path):
def _fake_download(storage_key, local_path):
local_path.parent.mkdir(parents=True, exist_ok=True)
local_path.write_bytes(b"fake video")
return True
@@ -702,8 +812,9 @@ class TestRenderPlanWithBgmAsr:
},
)
clips = [_make_clip("c1", order=0, duration=5.0)]
asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
adapter, _, _ = _make_adapter(plan=plan, clips=clips)
adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
# Mock BGM 下载
with patch.object(adapter, "_prepare_bgm", return_value=str(tmp_path / "bgm.mp3")):
@@ -724,7 +835,7 @@ class TestRenderPlanWithBgmAsr:
def test_asr_service_passed_to_render_service(self, mock_download, mock_render_cls, mock_upload, tmp_path):
"""ASR 服务被正确传递给 UnifiedRenderService。"""
def _fake_download(asset_id, local_path):
def _fake_download(storage_key, local_path):
local_path.parent.mkdir(parents=True, exist_ok=True)
local_path.write_bytes(b"fake video")
return True
@@ -745,8 +856,9 @@ class TestRenderPlanWithBgmAsr:
mock_asr = MagicMock()
plan = FakePlan(id="plan_asr")
clips = [_make_clip("c1", order=0, duration=5.0)]
asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
adapter, _, _ = _make_adapter(plan=plan, clips=clips)
adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
with patch.object(adapter, "_prepare_bgm", return_value=None):
with patch.object(adapter, "_get_asr_service", return_value=mock_asr):
@@ -764,16 +876,16 @@ class TestRenderPlanWithBgmAsr:
@patch("video_processing.render_adapter.download_asset")
def test_rendered_and_failed_clip_ids_in_result(self, mock_download, mock_render_cls, mock_upload, tmp_path):
"""渲染结果中包含成功和失败的 clip id 列表。"""
# 成功/失败映射，通过 asset_id 区分
# 成功/失败映射，通过 storage_key（file_url）区分
download_map = {
"good_001.mp4": True,
"bad_002.mp4": False,
"good_003.mp4": True,
"https://test-bucket.oss.com/assets/good_001.mp4": True,
"https://test-bucket.oss.com/assets/bad_002.mp4": False,
"https://test-bucket.oss.com/assets/good_003.mp4": True,
}
def _fake_download(asset_id, local_path):
def _fake_download(storage_key, local_path):
local_path.parent.mkdir(parents=True, exist_ok=True)
ok = download_map.get(asset_id, False)
ok = download_map.get(storage_key, False)
if ok:
local_path.write_bytes(b"fake video")
return ok
@@ -793,12 +905,17 @@ class TestRenderPlanWithBgmAsr:
plan = FakePlan(id="plan_mixed")
clips = [
_make_clip("c_good1", order=0, asset_id="good_001.mp4"),
_make_clip("c_bad", order=1, asset_id="bad_002.mp4"),
_make_clip("c_good2", order=2, asset_id="good_003.mp4"),
_make_clip("c_good1", order=0, asset_id="good_001"),
_make_clip("c_bad", order=1, asset_id="bad_002"),
_make_clip("c_good2", order=2, asset_id="good_003"),
]
asset_url_map = {
"good_001": "https://test-bucket.oss.com/assets/good_001.mp4",
"bad_002": "https://test-bucket.oss.com/assets/bad_002.mp4",
"good_003": "https://test-bucket.oss.com/assets/good_003.mp4",
}
adapter, _, _ = _make_adapter(plan=plan, clips=clips)
adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
result = adapter.render_plan(
"plan_mixed",