Compare commits

...

6 Commits

Author SHA1 Message Date
CI Bot 4a36b0a70d fix(#1341): 删除孤立测试文件 test_cover_service.py(cover_service 已在 887e66b0 中删除) 2026-08-12 22:44:44 +08:00
CI Bot b1a4067101 style: Prettier 格式化修复(useStep4Preview/useStep6Cover/EditorDrawers) 2026-08-12 22:33:27 +08:00
CI Bot cada58e9ea fix(#1341): 修复3个ESLint warning,确保CI Frontend Lint通过
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 27s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m19s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m58s
AI Code Review / AI Code Review (pull_request) Failing after 1m59s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 1m57s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 2m22s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 2m52s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 2m36s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 3m32s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 4m7s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build API Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
- useGenerateVideo.ts: 移除useCallback多余的selectedTemplate依赖
- useStep4Preview.ts: 删除不必要的as any类型断言(EditPlanConfig已有rendered_storage_key字段)
- useStep4Preview.ts: useCallback deps补充selectedTemplate
2026-08-12 22:21:47 +08:00
xiaoxia 7c1235c018 perf(#1341): 优化封面生成性能,改用 SceneChange 策略
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 45s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m28s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m47s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 1m32s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 1m49s
AI Code Review / AI Code Review (pull_request) Failing after 2m34s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 2m46s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 2m20s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 3m22s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m53s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m53s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
- 抽帧策略: TimeInterval → SceneChange(自动识别画面变化,选最佳帧)
- 抽帧数量: 5 → 3(平衡速度和质量)
- 轮询间隔: 3.0s → 2.0s(加快轮询)

预期效果:
- 封面质量更高(SceneChange 自动选最佳画面)
- 生成速度更快(3帧 vs 5帧)
- 减少超时风险

配合 PR #1347 的错误日志增强,便于定位问题
2026-08-12 22:13:57 +08:00
CI Bot b81d61b614 fix(#1341): 修复 ESLint TS 类型错误,删除 cover.py.bak
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 44s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Frontend Lint (pull_request) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build API Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Web Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
AI Code Review / AI Code Review (pull_request) Has been cancelled
PR Automation / Auto Approve on CI Green (pull_request) Has been cancelled
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been cancelled
Preview Deploy / Deploy Preview Environment (pull_request) Has been cancelled
- useStep6Cover.ts: catch(err) 用 as 做类型断言,避免 strict 模式下 unknown 属性访问
- 删除误提交的 cover.py.bak 文件
2026-08-12 22:12:32 +08:00
xiaoxia f64a4b4213 fix(#1341): 增强封面生成错误日志和前端提示
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 35s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 39s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 50s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m37s
Preview Deploy / Deploy Preview Environment (pull_request) Failing after 1m47s
AI Code Review / AI Code Review (pull_request) Failing after 2m1s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 3m8s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m3s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 3m4s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m55s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 4m7s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
- 后端 cover.py: 在 3 步查找预览视频的每个步骤增加详细日志
- 后端 cover.py: 在调用 MediaKit 前后增加日志
- 前端 useStep6Cover.ts: catch 块中解析后端返回的详细错误信息
- 前端: 区分后端错误、网络错误、超时错误,显示具体原因

目的:让封面生成失败时能看到具体卡在哪一步,便于定位问题
2026-08-12 22:07:19 +08:00
7 changed files with 121 additions and 488 deletions
@@ -52,11 +52,13 @@ def editor_generate_cover(
# ── 3 步查找预览视频 URL ──────────────────────────────────────────
# 第一步:从 plan.config 读取
logger.info("[封面生成] 步骤1: 从 plan.config 查找 rendered_storage_key: plan_id=%s", plan_id)
rendered_storage_key = (plan.config or {}).get("rendered_storage_key", "")
# 第二步:如果还没有,通过 generation_task_id 查找预览任务的产物
if not rendered_storage_key:
generation_task_id = (plan.config or {}).get("generation_task_id", "")
logger.info("[封面生成] 步骤2: 通过 generation_task_id 查找: plan_id=%s task_id=%s", plan_id, generation_task_id)
if generation_task_id:
try:
gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
@@ -68,7 +70,12 @@ def editor_generate_cover(
if videos:
rendered_storage_key = getattr(videos[0], "file_url", "") or ""
logger.info(
"封面生成: 通过 generation_task_id 找到视频: plan_id=%s task_id=%s",
"[封面生成] ✅ 步骤2找到视频: plan_id=%s task_id=%s url=%s",
plan_id,
generation_task_id,
rendered_storage_key[:80],
)
logger.info(
plan_id,
generation_task_id,
)
@@ -83,6 +90,7 @@ def editor_generate_cover(
if not rendered_storage_key:
try:
gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
logger.info("[封面生成] 步骤3: 通过 user+template 查找: plan_id=%s template_id=%s", plan_id, template_id)
preview_tasks = gen_task_repo.list_latest_completed_preview(
user_id=str(current_user.user.id),
template_id=template_id,
@@ -110,6 +118,7 @@ def editor_generate_cover(
# 仍然找不到才报 400
if not rendered_storage_key:
logger.error("[封面生成] ❌ 找不到预览视频: plan_id=%s", plan_id)
raise HTTPException(
status_code=400,
detail="请先生成预览视频,再生成封面",
@@ -142,6 +151,7 @@ def editor_generate_cover(
from packages.shared.ai_service import run_generate_cover
try:
logger.info("[封面生成] 开始调用 AI 封面生成服务: plan_id=%s", plan_id)
cover_data = run_generate_cover(
plan_id=plan_id,
asset_ids=body.asset_ids,
@@ -239,7 +239,6 @@ const EditorDrawers: React.FC<EditorDrawersProps> = ({
onChange={onStickerChange}
totalDuration={totalDuration}
/>
</>
)
}
@@ -78,7 +78,7 @@ export function useGenerateVideo(props: UseGenerateVideoProps) {
setGenerateError(finalMsg)
message.error(finalMsg)
}
}, [props, selectedTemplate, clearTimer, startPolling])
}, [props, clearTimer, startPolling])
/* 重新生成(失败后重试) */
const retry = useCallback(() => {
@@ -212,99 +212,102 @@ export function useStep4Preview({
}, [clearPollTimer])
/** 轮询单个预览任务状态 */
const pollPreviewStatus = useCallback((index: number, taskId: string) => {
const poll = async () => {
// 竞态检查
if (taskIdsRef.current.get(index) !== taskId) return
// 超时检查
if (Date.now() - startTimeRef.current > POLL_TIMEOUT_MS) {
setItems((prev) =>
prev.map((it) =>
it.index === index ? { ...it, status: "error", error: "预览生成超时,请重试" } : it,
),
)
return
}
try {
const data: PreviewTaskResponse = await getPreviewStatus(taskId)
const pollPreviewStatus = useCallback(
(index: number, taskId: string) => {
const poll = async () => {
// 竞态检查
if (taskIdsRef.current.get(index) !== taskId) return
const status = data.status as ApiPreviewStatus
// 超时检查
if (Date.now() - startTimeRef.current > POLL_TIMEOUT_MS) {
setItems((prev) =>
prev.map((it) =>
it.index === index ? { ...it, status: "error", error: "预览生成超时,请重试" } : it,
),
)
return
}
if (status === "completed") {
const result: PreviewResult = {
taskId: safeString(data.task_id, ""),
videoUrl: safeString(data.video_url, ""),
clipCount: safeNumber(data.clip_count),
transitionCount: safeNumber(data.transition_count),
materialUsage: safeNumber(data.material_usage),
duration: safeNumber(data.duration),
fileSize: safeNumber(data.file_size),
generateDuration: safeNumber(data.generate_duration),
progress: 100,
try {
const data: PreviewTaskResponse = await getPreviewStatus(taskId)
if (taskIdsRef.current.get(index) !== taskId) return
const status = data.status as ApiPreviewStatus
if (status === "completed") {
const result: PreviewResult = {
taskId: safeString(data.task_id, ""),
videoUrl: safeString(data.video_url, ""),
clipCount: safeNumber(data.clip_count),
transitionCount: safeNumber(data.transition_count),
materialUsage: safeNumber(data.material_usage),
duration: safeNumber(data.duration),
fileSize: safeNumber(data.file_size),
generateDuration: safeNumber(data.generate_duration),
progress: 100,
}
setItems((prev) =>
prev.map((it) =>
it.index === index ? { ...it, status: "ready", result, progress: 100 } : it,
),
)
// 保存预览视频 URL 到 plan config,供封面生成使用
if (result.videoUrl && selectedTemplate) {
updateEditPlan(selectedTemplate, {
config: { rendered_storage_key: result.videoUrl },
}).catch((err) => {
console.warn("[Step4] 保存预览视频URL到plan config失败:", err)
})
}
return
}
setItems((prev) =>
prev.map((it) =>
it.index === index ? { ...it, status: "ready", result, progress: 100 } : it,
),
)
// 保存预览视频 URL 到 plan config,供封面生成使用
if (result.videoUrl && selectedTemplate) {
updateEditPlan(selectedTemplate, {
config: { rendered_storage_key: result.videoUrl } as any,
}).catch((err) => {
console.warn("[Step4] 保存预览视频URL到plan config失败:", err)
})
if (status === "failed") {
setItems((prev) =>
prev.map((it) =>
it.index === index
? {
...it,
status: "error",
error: safeString(data.error_message, "预览生成失败,请重试"),
}
: it,
),
)
return
}
return
}
if (status === "failed") {
if (status === "cancelled") {
setItems((prev) =>
prev.map((it) =>
it.index === index ? { ...it, status: "error", error: "预览任务已取消" } : it,
),
)
return
}
// pending / generating 状态继续轮询
const prog = safeNumber(data.progress)
const nextStatus: PreviewStatus = status === "pending" ? "pending" : "generating"
setItems((prev) =>
prev.map((it) =>
it.index === index
? {
...it,
status: "error",
error: safeString(data.error_message, "预览生成失败,请重试"),
}
: it,
it.index === index ? { ...it, status: nextStatus, progress: prog } : it,
),
)
return
const delay = status === "pending" ? 5000 : 2000
pollTimersRef.current.set(index, setTimeout(poll, delay))
} catch {
if (taskIdsRef.current.get(index) !== taskId) return
pollTimersRef.current.set(index, setTimeout(poll, 3000))
}
if (status === "cancelled") {
setItems((prev) =>
prev.map((it) =>
it.index === index ? { ...it, status: "error", error: "预览任务已取消" } : it,
),
)
return
}
// pending / generating 状态继续轮询
const prog = safeNumber(data.progress)
const nextStatus: PreviewStatus = status === "pending" ? "pending" : "generating"
setItems((prev) =>
prev.map((it) =>
it.index === index ? { ...it, status: nextStatus, progress: prog } : it,
),
)
const delay = status === "pending" ? 5000 : 2000
pollTimersRef.current.set(index, setTimeout(poll, delay))
} catch {
if (taskIdsRef.current.get(index) !== taskId) return
pollTimersRef.current.set(index, setTimeout(poll, 3000))
}
}
pollTimersRef.current.set(index, setTimeout(poll, 1000))
}, [])
pollTimersRef.current.set(index, setTimeout(poll, 1000))
},
[selectedTemplate],
)
/** 生成所有预览 */
const generatePreview = useCallback(async () => {
@@ -106,8 +106,29 @@ export function useStep6Cover({
}
} catch (err) {
clearTimeout(timeoutId)
console.warn("[Step6] 智能封面生成失败:", err)
message.error("封面生成失败,请稍后重试")
console.error("[Step6] 智能封面生成失败:", err)
// 提取详细错误信息
let errorMsg = "封面生成失败"
const e = err as {
response?: { data?: { detail?: string; message?: string }; status?: number }
request?: unknown
message?: string
}
if (e.response) {
// 后端返回错误
const detail = e.response.data?.detail || e.response.data?.message || ""
errorMsg = detail || `后端错误 (${e.response.status})`
console.error("[Step6] 后端返回:", e.response.data)
} else if (e.request) {
// 请求已发送但无响应
errorMsg = "服务器无响应,请检查网络连接"
console.error("[Step6] 请求无响应:", e.request)
} else if (e.message) {
errorMsg = e.message
}
message.error(errorMsg)
} finally {
clearTimeout(timeoutId)
setGenerating(false)
+4 -4
View File
@@ -402,10 +402,10 @@ def _call_ai_cover_service(
logger.info("调用 MediaKit 抽帧: plan_id=%s video=%s", plan_id, primary_video_url[:80])
frames = client.extract_frames(
video_url=primary_video_url,
strategy="TimeInterval",
max_frames=5,
poll_interval=3.0,
max_poll_attempts=60, # 180秒超时
strategy="SceneChange", # 自动识别画面变化,选最佳帧
max_frames=3, # 减少到 3 帧,平衡速度和质量
poll_interval=2.0, # 缩短轮询间隔
max_poll_attempts=60, # 120秒超时
)
if frames and len(frames) > 0:
-400
View File
@@ -1,400 +0,0 @@
"""
封面管理服务单元测试
"""
import subprocess
from pathlib import Path
from unittest.mock import Mock, patch
import pytest
from apps.api.app.services.cover_service import (
COVER_STORAGE_PREFIX,
DEFAULT_COVER_HEIGHT,
DEFAULT_COVER_QUALITY,
DEFAULT_COVER_WIDTH,
CoverService,
)
class TestGetCoverConfig:
"""get_cover_config 静态方法测试"""
def test_get_cover_config_default(self):
"""测试默认封面配置"""
config = {}
result = CoverService.get_cover_config(config)
assert result["type"] == "ai_frame"
assert result["image_url"] == ""
assert result["frame_time"] is None
def test_get_cover_config_with_custom_values(self):
"""测试自定义封面配置"""
config = {
"cover": {
"type": "manual",
"image_url": "https://example.com/cover.jpg",
"frame_time": 5.5,
}
}
result = CoverService.get_cover_config(config)
assert result["type"] == "manual"
assert result["image_url"] == "https://example.com/cover.jpg"
assert result["frame_time"] == 5.5
def test_get_cover_config_cover_not_dict(self):
"""测试 cover 不是 dict 时返回默认值"""
config = {"cover": "not-a-dict"}
result = CoverService.get_cover_config(config)
assert result["type"] == "ai_frame"
assert result["image_url"] == ""
assert result["frame_time"] is None
def test_get_cover_config_partial_fields(self):
"""测试部分字段存在时,其余字段用默认值"""
config = {"cover": {"type": "custom"}}
result = CoverService.get_cover_config(config)
assert result["type"] == "custom"
assert result["image_url"] == ""
assert result["frame_time"] is None
def test_get_cover_config_empty_cover_dict(self):
"""测试空的 cover dict"""
config = {"cover": {}}
result = CoverService.get_cover_config(config)
assert result["type"] == "ai_frame"
assert result["image_url"] == ""
class TestExtractCoverFromClip:
"""extract_cover_from_clip 测试"""
@pytest.fixture
def mock_storage(self):
storage = Mock()
storage.download_file = Mock()
storage.upload_file = Mock()
storage.get_url = Mock(return_value="https://oss.example.com/covers/plan1/cover_1000.jpg")
return storage
@pytest.fixture
def mock_asset_repo(self):
repo = Mock()
repo.get = Mock(return_value=None)
return repo
@pytest.fixture
def video_asset(self):
asset = Mock()
asset.storage_key = "videos/test-video.mp4"
asset.mime_type = "video/mp4"
return asset
@pytest.fixture
def service(self, mock_storage, mock_asset_repo):
return CoverService(storage_service=mock_storage, asset_repository=mock_asset_repo)
def test_extract_cover_asset_not_found(self, service, mock_asset_repo):
"""测试素材不存在时报错"""
mock_asset_repo.get.return_value = None
with pytest.raises(ValueError, match="素材不存在"):
service.extract_cover_from_clip(plan_id="plan-1", asset_id="nonexistent")
def test_extract_cover_asset_no_storage_key(self, service, mock_asset_repo):
"""测试素材没有文件时报错"""
asset = Mock()
asset.storage_key = ""
asset.mime_type = "video/mp4"
mock_asset_repo.get.return_value = asset
with pytest.raises(ValueError, match="素材没有文件"):
service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-no-file")
def test_extract_cover_asset_not_video(self, service, mock_asset_repo):
"""测试非视频素材报错"""
asset = Mock()
asset.storage_key = "images/photo.jpg"
asset.mime_type = "image/jpeg"
mock_asset_repo.get.return_value = asset
with pytest.raises(ValueError, match="素材不是视频类型"):
service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-img")
def test_extract_cover_download_failure(self, service, mock_asset_repo, mock_storage, video_asset):
"""测试下载素材失败"""
mock_asset_repo.get.return_value = video_asset
mock_storage.download_file.side_effect = Exception("网络错误")
with pytest.raises(RuntimeError, match="下载素材失败"):
service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-1")
def test_extract_cover_upload_failure(self, service, mock_asset_repo, mock_storage, video_asset):
"""测试上传封面失败"""
mock_asset_repo.get.return_value = video_asset
def fake_download(storage_key, local_path):
# 创建一个假的视频文件
Path(local_path).parent.mkdir(parents=True, exist_ok=True)
with open(local_path, "wb") as f:
f.write(b"fake video data")
mock_storage.download_file.side_effect = fake_download
mock_storage.upload_file.side_effect = Exception("上传失败")
# mock _extract_frame 避免真的调 ffmpeg
with patch.object(CoverService, "_extract_frame") as mock_extract:
def fake_extract(video_path, output_path, **kwargs):
# 创建假的封面文件
with open(output_path, "wb") as f:
f.write(b"\xff\xd8\xff\xe0fake jpeg data")
mock_extract.side_effect = fake_extract
with pytest.raises(RuntimeError, match="上传封面失败"):
service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-1")
def test_extract_cover_get_url_falls_back_to_key(self, service, mock_asset_repo, mock_storage, video_asset):
"""测试获取 URL 失败时降级为 storage_key"""
mock_asset_repo.get.return_value = video_asset
def fake_download(storage_key, local_path):
Path(local_path).parent.mkdir(parents=True, exist_ok=True)
with open(local_path, "wb") as f:
f.write(b"fake video data")
mock_storage.download_file.side_effect = fake_download
mock_storage.get_url.side_effect = Exception("URL服务不可用")
with patch.object(CoverService, "_extract_frame") as mock_extract:
def fake_extract(video_path, output_path, **kwargs):
with open(output_path, "wb") as f:
f.write(b"\xff\xd8\xff\xe0fake jpeg")
mock_extract.side_effect = fake_extract
result = service.extract_cover_from_clip(plan_id="plan-abc", asset_id="asset-xyz", frame_time=2.5)
assert result["type"] == "manual"
assert result["frame_time"] == 2.5
# URL 失败时返回 storage_key
assert COVER_STORAGE_PREFIX in result["image_url"]
assert "plan-abc" in result["image_url"]
def test_extract_cover_success(self, service, mock_asset_repo, mock_storage, video_asset):
"""测试抽帧成功完整流程"""
mock_asset_repo.get.return_value = video_asset
def fake_download(storage_key, local_path):
Path(local_path).parent.mkdir(parents=True, exist_ok=True)
with open(local_path, "wb") as f:
f.write(b"fake video data for testing")
mock_storage.download_file.side_effect = fake_download
with patch.object(CoverService, "_extract_frame") as mock_extract:
def fake_extract(video_path, output_path, **kwargs):
with open(output_path, "wb") as f:
f.write(b"\xff\xd8\xff\xe0fake jpeg image data")
mock_extract.side_effect = fake_extract
result = service.extract_cover_from_clip(
plan_id="plan-123",
asset_id="asset-456",
frame_time=3.0,
width=720,
height=1280,
quality=3,
)
assert result["type"] == "manual"
assert result["image_url"] == "https://oss.example.com/covers/plan1/cover_1000.jpg"
assert result["frame_time"] == 3.0
# 验证上传被调用
mock_storage.upload_file.assert_called_once()
upload_args = mock_storage.upload_file.call_args[1]
assert upload_args["content_type"] == "image/jpeg"
assert "plan-123" in upload_args["storage_key"]
assert "3000" in upload_args["storage_key"] # frame_time * 1000
# 验证 _extract_frame 被调用且参数正确
mock_extract.assert_called_once()
extract_kwargs = mock_extract.call_args[1]
assert extract_kwargs["time_sec"] == 3.0
assert extract_kwargs["width"] == 720
assert extract_kwargs["height"] == 1280
assert extract_kwargs["quality"] == 3
class TestGenerateSmartCover:
"""generate_smart_cover 测试"""
@pytest.fixture
def service(self):
return CoverService(storage_service=Mock(), asset_repository=Mock())
def test_generate_smart_cover_calls_extract_with_default_time(self, service):
"""测试智能封面调用 extract_cover_from_clip 并设置 type 为 ai_frame"""
fake_result = {"type": "manual", "image_url": "test.jpg", "frame_time": 3.0}
with patch.object(service, "extract_cover_from_clip", return_value=fake_result) as mock_extract:
result = service.generate_smart_cover(plan_id="plan-1", asset_id="asset-1")
mock_extract.assert_called_once()
call_kwargs = mock_extract.call_args[1]
assert call_kwargs["plan_id"] == "plan-1"
assert call_kwargs["asset_id"] == "asset-1"
assert call_kwargs["frame_time"] == 3.0 # 默认第3秒
assert result["type"] == "ai_frame"
assert result["image_url"] == "test.jpg"
def test_generate_smart_cover_passes_dimensions(self, service):
"""测试智能封面传递尺寸和质量参数"""
fake_result = {"type": "manual", "image_url": "test.jpg", "frame_time": 3.0}
with patch.object(service, "extract_cover_from_clip", return_value=fake_result) as mock_extract:
service.generate_smart_cover(
plan_id="plan-1",
asset_id="asset-1",
width=1080,
height=1920,
quality=5,
)
call_kwargs = mock_extract.call_args[1]
assert call_kwargs["width"] == 1080
assert call_kwargs["height"] == 1920
assert call_kwargs["quality"] == 5
class TestExtractFrame:
"""_extract_frame 静态方法测试(mock subprocess"""
@pytest.fixture
def video_path(self, tmp_path):
path = tmp_path / "test_video.mp4"
path.write_bytes(b"fake video")
return path
@pytest.fixture
def output_path(self, tmp_path):
return tmp_path / "cover.jpg"
def test_extract_frame_success(self, video_path, output_path):
"""测试 FFmpeg 抽帧成功"""
fake_result = Mock()
fake_result.returncode = 0
with patch("subprocess.run", return_value=fake_result) as mock_run:
CoverService._extract_frame(
video_path=video_path,
output_path=output_path,
time_sec=2.5,
width=1080,
height=1920,
quality=5,
)
assert mock_run.call_count == 1
cmd = mock_run.call_args[0][0]
assert cmd[0] == "ffmpeg"
assert "-ss" in cmd
assert "2.500" in cmd
assert "-vframes" in cmd
# 验证 scale+crop 滤镜存在
vf_index = cmd.index("-vf") + 1
assert "scale=" in cmd[vf_index]
assert "crop=" in cmd[vf_index]
def test_extract_frame_fallback_to_simple_command(self, video_path, output_path):
"""测试主命令失败时回退到简化命令"""
fail_result = Mock()
fail_result.returncode = 1
fail_result.stderr = "Filter graph error"
success_result = Mock()
success_result.returncode = 0
call_count = 0
def fake_run(*args, **kwargs):
nonlocal call_count
call_count += 1
if call_count == 1:
return fail_result
return success_result
with patch("subprocess.run", side_effect=fake_run) as mock_run:
CoverService._extract_frame(
video_path=video_path,
output_path=output_path,
time_sec=1.0,
width=1080,
height=1920,
quality=5,
)
assert mock_run.call_count == 2
# 第二次是简化命令(没有 -vf 参数)
second_cmd = mock_run.call_args_list[1][0][0]
assert "-vf" not in second_cmd
def test_extract_frame_both_commands_fail(self, video_path, output_path):
"""测试两个命令都失败时报错"""
fail_result = Mock()
fail_result.returncode = 1
fail_result.stderr = "Invalid data found when processing input"
with patch("subprocess.run", return_value=fail_result):
with pytest.raises(RuntimeError, match="FFmpeg 抽帧失败"):
CoverService._extract_frame(
video_path=video_path,
output_path=output_path,
time_sec=1.0,
width=1080,
height=1920,
quality=5,
)
def test_extract_frame_timeout(self, video_path, output_path):
"""测试 FFmpeg 抽帧超时"""
with patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd="ffmpeg", timeout=60)):
with pytest.raises(RuntimeError, match="FFmpeg 抽帧超时"):
CoverService._extract_frame(
video_path=video_path,
output_path=output_path,
time_sec=1.0,
width=1080,
height=1920,
quality=5,
)
def test_extract_frame_ffmpeg_not_found(self, video_path, output_path):
"""测试 FFmpeg 不可用"""
with patch("subprocess.run", side_effect=FileNotFoundError("ffmpeg not found")):
with pytest.raises(RuntimeError, match="FFmpeg 不可用"):
CoverService._extract_frame(
video_path=video_path,
output_path=output_path,
time_sec=1.0,
width=1080,
height=1920,
quality=5,
)
class TestDefaults:
"""默认常量测试"""
def test_default_dimensions(self):
"""测试默认尺寸常量"""
assert DEFAULT_COVER_WIDTH == 1080
assert DEFAULT_COVER_HEIGHT == 1920
assert DEFAULT_COVER_QUALITY == 5
assert COVER_STORAGE_PREFIX == "covers"