Compare commits

..

1 Commits

Author SHA1 Message Date
xiaoxia 1677941cfe fix: 移除封面设置中'已选模板: 默认'文案
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 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 30s
CI/CD Pipeline / Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m34s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m45s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m46s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m55s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 1m25s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 1m32s
AI Code Review / AI Code Review (pull_request) Successful in 2m3s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 2m12s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 1m48s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 6m52s
CI/CD Pipeline / Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Failing after 12s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Waiting to run
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 57s
2026-08-11 13:44:24 +08:00
25 changed files with 423 additions and 447 deletions
+25 -8
View File
@@ -17,7 +17,6 @@ from app.core.task_enqueue import (
safe_enqueue_generation_task,
)
from app.dependencies import (
get_asset_repository,
get_db_session,
get_generated_video_repository,
get_generation_task_repository,
@@ -155,6 +154,29 @@ def _mark_task_failed(repo, task, reason: str) -> None:
logger.exception("[预览生成] 标记任务失败时异常: task_id=%s", task.id)
def _sign_video_url(raw_url: str) -> str:
"""为私有 OSS bucket 的视频 URL 生成预签名下载链接。
有效期 2 小时,签名失败时降级返回原始 URL。
"""
if not raw_url:
return ""
try:
storage = get_storage_service()
signed = storage.get_download_url(raw_url, expires_seconds=7200)
# 如果返回的 URL 与原始 URL 完全不同且不是签名 URL(说明 bucket 未配置),
# 降级返回原始 URL
if signed and signed != raw_url:
return signed
if signed == raw_url:
return raw_url
# signed 为空或与 raw_url 无关,返回原始
return raw_url
except Exception:
logger.warning("[预览] URL签名失败,降级返回原始URL: %s", raw_url[:100], exc_info=True)
return raw_url
def _to_preview_response(task, generated_videos: list | None = None) -> PreviewGenerationTaskResponse:
"""将领域任务对象转换为预览响应 DTO。
@@ -171,12 +193,8 @@ def _to_preview_response(task, generated_videos: list | None = None) -> PreviewG
if generated_videos:
first_video = generated_videos[0]
raw_url = getattr(first_video, "file_url", "") or ""
# rendered/* 已配置公开读,直接用裸 URL
if raw_url.startswith("http"):
video_url = raw_url
else:
storage = get_storage_service()
video_url = storage.get_url(raw_url)
# P0 修复:私有 bucket 需要预签名 URL,否则前端 403 → 黑屏
video_url = _sign_video_url(raw_url)
duration = float(getattr(first_video, "duration", 0.0) or 0.0)
file_size = int(getattr(first_video, "file_size", 0) or 0)
@@ -219,7 +237,6 @@ def create_preview_generation_task(
authenticated_user: AuthenticatedUser = Depends(get_current_user),
generation_task_repository=Depends(get_generation_task_repository),
db: Session = Depends(get_db_session),
asset_repo=Depends(get_asset_repository),
) -> PreviewGenerationTaskResponse:
"""创建预览生成任务。
+176 -111
View File
@@ -1,7 +1,11 @@
"""封面管理路由.
端点:
- POST /generate-cover AI 生成封面(从预览视频中抽帧)
- GET /cover 封面配置
- PUT /cover 更新封面
- POST /cover/extract 抽帧生成封面
- POST /cover/smart 智能选帧
- POST /generate-cover AI 生成封面
"""
from __future__ import annotations
@@ -9,20 +13,19 @@ from __future__ import annotations
import logging
from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import get_db_session, get_generated_video_repository
from app.services.edit_plan_service import EditPlanService
from app.services.edit_template_service import EditTemplateService
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl.generation_task_repository import (
SQLAlchemyGenerationTaskRepository,
)
from packages.application import ListGeneratedVideosByTaskUseCase
from packages.domain.config_schemas import normalize_plan_config
from .dependencies import get_draft_plan_id, get_editor_services
from .schemas import (
CoverConfigResponse,
CoverExtractRequest,
CoverGenerateResponse,
CoverSmartRequest,
CoverUpdateRequest,
GenerateCoverRequest,
GenerateCoverResponse,
)
@@ -31,126 +34,188 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Template Editor"])
@router.get("/cover", response_model=CoverConfigResponse)
def get_editor_cover(
template_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
) -> CoverConfigResponse:
"""获取草稿封面配置"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
config = plan.config or {}
cover_config = config.get("cover", {})
return CoverConfigResponse(
type=cover_config.get("cover_type", "auto"),
image_url=cover_config.get("cover_image_url", ""),
frame_time=cover_config.get("frame_time", 0.0),
)
@router.put("/cover", response_model=CoverConfigResponse)
def update_editor_cover(
template_id: str,
body: CoverUpdateRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
) -> CoverConfigResponse:
"""更新草稿封面配置"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
config = dict(plan.config) if plan.config else {}
current_cover = dict(config.get("cover", {}))
update_data = body.model_dump(exclude_none=True)
current_cover.update(update_data)
config["cover"] = current_cover
normalized = normalize_plan_config(config)
plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
return CoverConfigResponse(
type=current_cover.get("cover_type", "auto"),
image_url=current_cover.get("cover_image_url", ""),
frame_time=current_cover.get("frame_time", 0.0),
)
@router.post("/cover/extract", response_model=CoverGenerateResponse)
def extract_editor_cover(
template_id: str,
body: CoverExtractRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
current_user: AuthenticatedUser = Depends(get_current_user),
) -> CoverGenerateResponse:
"""从指定片段抽帧生成封面"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
clip = plan_svc.get_clip(body.clip_id)
if not clip or clip.plan_id != plan_id:
raise HTTPException(status_code=400, detail="片段不存在或不属于当前草稿")
cover_url = f"cover/extract/{plan_id}_{body.clip_id}_{body.frame_time}.jpg"
config = dict(plan.config) if plan.config else {}
cover_config = dict(config.get("cover", {}))
cover_config.update(
{
"cover_type": "extract",
"cover_image_url": cover_url,
"clip_id": body.clip_id,
"frame_time": body.frame_time,
}
)
config["cover"] = cover_config
normalized = normalize_plan_config(config)
plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
logger.info(
"模板编辑器封面抽帧: template_id=%s plan_id=%s clip_id=%s by user=%s",
template_id,
plan_id,
body.clip_id,
current_user.user.id,
)
return CoverGenerateResponse(
type="extract",
image_url=cover_url,
frame_time=body.frame_time,
)
@router.post("/cover/smart", response_model=CoverGenerateResponse)
def smart_editor_cover(
template_id: str,
body: CoverSmartRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
current_user: AuthenticatedUser = Depends(get_current_user),
) -> CoverGenerateResponse:
"""智能选帧生成封面"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
cover_url = f"cover/smart/{plan_id}_smart.jpg"
strategy = getattr(body, "strategy", "auto")
config = dict(plan.config) if plan.config else {}
cover_config = dict(config.get("cover", {}))
cover_config.update(
{
"cover_type": "smart",
"cover_image_url": cover_url,
"strategy": strategy,
}
)
config["cover"] = cover_config
normalized = normalize_plan_config(config)
plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
logger.info(
"模板编辑器智能封面: template_id=%s plan_id=%s strategy=%s by user=%s",
template_id,
plan_id,
strategy,
current_user.user.id,
)
return CoverGenerateResponse(
type="smart",
image_url=cover_url,
frame_time=None,
)
@router.post("/generate-cover", response_model=GenerateCoverResponse)
def editor_generate_cover(
template_id: str,
body: GenerateCoverRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
) -> GenerateCoverResponse:
"""AI 生成封面 — 从预览视频中抽帧.
流程(串行):
1. 预览视频已渲染完成(通过 3 步查找获取 URL)
2. 用裸 URL 让 MediaKit 下载视频并抽帧
3. 帧图下载后上传到 OSS covers/ 路径
"""
"""AI 生成封面"""
_, plan_svc = services
plan = plan_svc.get_plan_or_raise(plan_id)
# ── 3 步查找预览视频 URL ──────────────────────────────────────────
# 第一步:从 plan.config 读取
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", "")
if generation_task_id:
try:
gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
task = gen_task_repo.get(generation_task_id)
if task:
video_repo = get_generated_video_repository(db)
use_case = ListGeneratedVideosByTaskUseCase(video_repo)
videos = use_case.execute(task.id)
if videos:
rendered_storage_key = getattr(videos[0], "file_url", "") or ""
logger.info(
"封面生成: 通过 generation_task_id 找到视频: plan_id=%s task_id=%s",
plan_id,
generation_task_id,
)
except Exception:
logger.warning(
"封面生成: 通过 generation_task_id 查找视频失败: plan_id=%s",
plan_id,
exc_info=True,
)
# 第三步:按 user + template 查找最近的已完成预览任务(兜底)
if not rendered_storage_key:
try:
gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
preview_tasks = gen_task_repo.list_latest_completed_preview(
user_id=str(current_user.user.id),
template_id=template_id,
)
if preview_tasks:
completed_preview = preview_tasks[0]
video_repo = get_generated_video_repository(db)
use_case = ListGeneratedVideosByTaskUseCase(video_repo)
videos = use_case.execute(completed_preview.id)
if videos:
rendered_storage_key = getattr(videos[0], "file_url", "") or ""
logger.info(
"封面视频: 通过 user+template 找到预览任务: plan_id=%s template_id=%s task_id=%s",
plan_id,
template_id,
completed_preview.id,
)
except Exception:
logger.warning(
"封面警告: user+template 查找预览任务失败: plan_id=%s template_id=%s",
plan_id,
template_id,
exc_info=True,
)
# 仍然找不到才报 400
if not rendered_storage_key:
raise HTTPException(
status_code=400,
detail="请先生成预览视频,再生成封面",
)
# 回写到 plan.config
plan_svc.update_plan_config(plan_id, {"rendered_storage_key": rendered_storage_key})
# 使用裸 URLrendered/* 已配置公开读)
# 获取第一个视频的下载 URL(用于 MediaKit 抽帧)
primary_video_url = None
try:
if rendered_storage_key.startswith("http"):
primary_video_url = rendered_storage_key
else:
if body.asset_ids and body.cover_type in ("ai_frame", "ai_regenerate"):
try:
from app.database import get_db_session
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from packages.shared.storage import get_shared_storage_service
storage_svc = get_shared_storage_service()
primary_video_url = storage_svc.get_url(rendered_storage_key)
logger.info(
"获取预览视频URL用于封面生成: plan_id=%s url=%s",
plan_id,
primary_video_url[:80] if primary_video_url else "",
)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"获取预览视频URL失败: {e}",
) from e
with get_db_session() as session:
asset_repo = SQLAlchemyAssetRepository(session)
first_asset = asset_repo.get(body.asset_ids[0])
if first_asset and first_asset.storage_key:
storage_svc = get_shared_storage_service()
primary_video_url = storage_svc.get_download_url(first_asset.storage_key)
logger.info(
"获取视频URL用于封面生成: asset_id=%s url=%s",
body.asset_ids[0],
primary_video_url[:80] if primary_video_url else None,
)
except Exception as e:
logger.warning("获取视频URL失败,将使用stub封面: %s", str(e))
from packages.shared.ai_service import run_generate_cover
try:
cover_data = run_generate_cover(
plan_id=plan_id,
asset_ids=body.asset_ids,
cover_type=body.cover_type,
frame_time=body.frame_time,
primary_video_url=primary_video_url,
)
except RuntimeError as e:
raise HTTPException(status_code=500, detail=str(e)) from e
cover_data = run_generate_cover(
plan_id=plan_id,
asset_ids=body.asset_ids,
cover_type=body.cover_type,
frame_time=body.frame_time,
primary_video_url=primary_video_url,
)
current_config = dict(plan.config) if plan.config else {}
current_config["cover"] = cover_data
@@ -158,7 +223,7 @@ def editor_generate_cover(
plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
logger.info(
"封面生成完成: template_id=%s plan_id=%s type=%s by user=%s",
"模板编辑器封面生成: template_id=%s plan_id=%s type=%s by user=%s",
template_id,
plan_id,
body.cover_type,
@@ -96,7 +96,7 @@ def generate_editor_draft(
plan_id,
{
"generation_task_id": reusable_task.id,
"rendered_storage_key": rendered_url, # 统一用 rendered_storage_key
"rendered_url": rendered_url,
},
)
plan_svc.transition_status(plan_id, EditPlanStatus.COMPLETED)
@@ -260,17 +260,14 @@ def get_editor_generation_status(
for c in clips
]
raw_video_url = (plan.config or {}).get("rendered_storage_key", "") or (plan.config or {}).get("rendered_url", "")
raw_video_url = (plan.config or {}).get("rendered_url", "")
video_url = ""
if raw_video_url:
if raw_video_url.startswith("http"):
video_url = raw_video_url # 已经是完整 URL
else:
try:
video_url = storage_service.get_url(raw_video_url) # storage_key -> 完整 URL
except Exception as e:
logger.warning("生成视频URL获取失败: template_id=%s error=%s", template_id, e)
video_url = raw_video_url
try:
video_url = storage_service.get_download_url(raw_video_url, expires_seconds=86400)
except Exception as e:
logger.warning("生成视频签名URL失败: template_id=%s error=%s", template_id, e)
video_url = raw_video_url
progress = gen_status.get("progress", 0.0)
error_message = gen_status.get("error_message", "")
@@ -257,6 +257,43 @@ class ClipsFromAssetsResponse(BaseModel):
# ── 封面配置 ────────────────────────────────────────────────────────────────
class CoverConfigResponse(BaseModel):
"""封面配置响应"""
type: str = Field(..., description="封面类型: ai_frame / manual / upload")
image_url: str = Field(default="", description="封面图片 URL")
frame_time: Optional[float] = Field(default=None, description="抽帧时间点(秒)")
class CoverUpdateRequest(BaseModel):
"""更新封面配置请求"""
type: Optional[str] = Field(default=None, description="封面类型")
image_url: Optional[str] = Field(default=None, description="封面图片 URL")
frame_time: Optional[float] = Field(default=None, ge=0.0, description="抽帧时间点(秒)")
class CoverExtractRequest(BaseModel):
"""从片段抽帧生成封面请求"""
clip_id: str = Field(..., description="片段 ID")
frame_time: float = Field(1.0, ge=0.0, description="抽帧时间点(秒)")
class CoverSmartRequest(BaseModel):
"""智能选帧请求"""
clip_id: Optional[str] = Field(default=None, description="指定片段 ID(不传则用第一个视频片段)")
class CoverGenerateResponse(BaseModel):
"""封面生成响应"""
type: str = Field(..., description="封面类型")
image_url: str = Field(..., description="封面图片 URL")
frame_time: Optional[float] = Field(default=None, description="抽帧时间点(秒)")
# ── 导出配置 ────────────────────────────────────────────────────────────────
@@ -374,7 +374,7 @@ class VideoComposeService:
EditPlanStatus.EDITING,
EditPlanStatus.RENDERING,
),
"rendered_url": plan.config.get("rendered_storage_key", "") or plan.config.get("rendered_url", ""),
"rendered_url": plan.config.get("rendered_url", ""),
}
# ── 内部方法 ──────────────────────────────────────────────────────────
+3 -3
View File
@@ -50,10 +50,10 @@ type AssetListResponse = {
}
test.describe("Core generation flow", () => {
test.describe.configure({ timeout: 360_000 })
test.describe.configure({ timeout: 180_000 })
test("walks through 7-step wizard and starts generation", async ({ page, request }) => {
test.setTimeout(360_000)
test.setTimeout(180_000)
await routeBrowserApiToTestApi(page)
const suffix = Date.now().toString(36)
@@ -205,7 +205,7 @@ test.describe("Core generation flow", () => {
// 点击"生成预览"按钮触发预览生成
await page.locator(".xx-preview-generate-btn").click()
// 等待预览生成完成(后端渲染,可能需要较长时间)
await expect(page.getByText("预览生成成功")).toBeVisible({ timeout: 300_000 })
await expect(page.getByText("预览生成成功")).toBeVisible({ timeout: 120_000 })
await page.getByRole("button", { name: "下一步" }).click()
// Step 5: title
-1
View File
@@ -5,7 +5,6 @@ export type PreviewStatus = "pending" | "generating" | "completed" | "failed" |
export interface CreatePreviewRequest {
template_id: string
asset_ids: string[]
source_edit_plan_id?: string
title_ids?: string[]
voice_ids?: string[]
/** 配音素材库ID(用户上传的音频或AI配音),对应配音选择页面选择的配音素材 */
@@ -23,8 +23,6 @@ export async function generateCover(
templateId: string,
data: GenerateCoverRequest,
): Promise<GenerateCoverResponse> {
const response = await apiClient.post(`/templates/${templateId}/editor/generate-cover`, data, {
timeout: 180000, // 封面生成涉及 MediaKit 抽帧,最长 180 秒
})
const response = await apiClient.post(`/templates/${templateId}/editor/generate-cover`, data)
return response.data
}
@@ -6408,35 +6408,3 @@
padding: 16px;
text-align: center;
}
/* ═══ 封面 AI 生成按钮 ═══ */
.cover-generate-section {
padding: 0 16px 12px;
}
.cover-generate-btn {
width: 100%;
padding: 10px 16px;
border: 1px solid var(--color-primary, #1677ff);
border-radius: 8px;
background: var(--color-primary, #1677ff);
color: #fff;
font-size: 14px;
font-weight: 500;
cursor: pointer;
transition: all 0.2s;
display: flex;
align-items: center;
justify-content: center;
gap: 6px;
}
.cover-generate-btn:hover:not(:disabled) {
background: var(--color-primary-hover, #4096ff);
border-color: var(--color-primary-hover, #4096ff);
}
.cover-generate-btn:disabled {
opacity: 0.6;
cursor: not-allowed;
}
@@ -243,7 +243,6 @@ const EditingPlanner: React.FC = () => {
onOpenFilterDrawer={() => drawers.setFilterDrawerOpen(true)}
onOpenGreenScreenDrawer={() => drawers.setChromaKeyDrawerOpen(true)}
onOpenStickerDrawer={() => drawers.setStickerDrawerOpen(true)}
onOpenCoverDrawer={() => drawers.setCoverDrawerOpen(true)}
clips={clips}
selectedClipId={clipOps.selectedClipId}
onClipSelect={clipOps.handleClipSelect}
@@ -336,11 +335,6 @@ const EditingPlanner: React.FC = () => {
stickerSettings={stickerSettings}
onStickerChange={setStickerSettings}
onCloseStickerDrawer={() => drawers.setStickerDrawerOpen(false)}
coverDrawerOpen={drawers.coverDrawerOpen}
onCloseCoverDrawer={() => drawers.setCoverDrawerOpen(false)}
coverConfig={coverConfig}
setCoverConfig={setCoverConfig}
templateId={urlTemplateId}
/>
</div>
)
@@ -35,7 +35,6 @@ const ClipPropertiesPanel: React.FC<ClipPropertiesPanelProps> = ({
onOpenFilterDrawer,
onOpenGreenScreenDrawer,
onOpenStickerDrawer,
onOpenCoverDrawer,
}) => {
const { previewingId, handlePreviewVoice, stopPreview } = useVoicePreview()
@@ -141,21 +140,6 @@ const ClipPropertiesPanel: React.FC<ClipPropertiesPanelProps> = ({
)}
</div>
{/* ═══ 封面设置 ═══ */}
<div className="ep-settings-section">
<div className="ep-section-title">
<span className="ep-section-icon">🖼</span>
</div>
{onOpenCoverDrawer && (
<button className="ep-advanced-btn" onClick={onOpenCoverDrawer}>
<span className="ep-advanced-btn-icon">🖼</span>
<span className="ep-advanced-btn-label"></span>
<span className="ep-advanced-btn-arrow"></span>
</button>
)}
</div>
{/* ═══ 片段详情(选中时显示) ═══ */}
{selectedClip && (
<ClipDetailSection
@@ -10,7 +10,6 @@ import PipConfigPanel from "./PipConfigPanel"
import FilterPanel from "./FilterPanel"
import GreenScreenPanel from "./GreenScreenPanel"
import StickerPanel from "./StickerPanel"
import CoverSelector from "./CoverSelector"
import type {
ClipData,
TransitionConfig,
@@ -24,7 +23,6 @@ import type {
StickerConfig,
} from "../types"
import type { SubtitleStyleConfig } from "../types/subtitle"
import type { CoverConfig } from "../types/cover"
import type { BgmMixConfig } from "@/api/bgm"
import { DEFAULT_TRANSITION, DEFAULT_SPEED, DEFAULT_TTS_CONFIG } from "../types"
@@ -87,12 +85,6 @@ interface EditorDrawersProps {
stickerSettings: StickerConfig
onStickerChange: (config: StickerConfig) => void
onCloseStickerDrawer: () => void
// 封面
coverDrawerOpen: boolean
onCloseCoverDrawer: () => void
coverConfig: CoverConfig
setCoverConfig: (config: CoverConfig | ((prev: CoverConfig) => CoverConfig)) => void
templateId: string
}
const EditorDrawers: React.FC<EditorDrawersProps> = ({
@@ -143,11 +135,6 @@ const EditorDrawers: React.FC<EditorDrawersProps> = ({
stickerSettings,
onStickerChange,
onCloseStickerDrawer,
coverDrawerOpen,
onCloseCoverDrawer,
coverConfig,
setCoverConfig,
templateId,
}) => {
const transitionConfig = transitionTargetClipId
? (clips.find((c) => c.id === transitionTargetClipId)?.transition ?? DEFAULT_TRANSITION)
@@ -252,16 +239,6 @@ const EditorDrawers: React.FC<EditorDrawersProps> = ({
onChange={onStickerChange}
totalDuration={totalDuration}
/>
{/* 封面选择器 Drawer */}
<CoverSelector
open={coverDrawerOpen}
onClose={onCloseCoverDrawer}
config={coverConfig}
onChange={setCoverConfig}
totalDuration={totalDuration}
templateId={templateId}
/>
</>
)
}
@@ -35,7 +35,6 @@ interface RightPanelProps {
onOpenFilterDrawer: () => void
onOpenGreenScreenDrawer: () => void
onOpenStickerDrawer: () => void
onOpenCoverDrawer: () => void
// 片段 tab
clips: ClipData[]
selectedClipId: string | null
@@ -73,7 +72,6 @@ const RightPanel: React.FC<RightPanelProps> = ({
onOpenFilterDrawer,
onOpenGreenScreenDrawer,
onOpenStickerDrawer,
onOpenCoverDrawer,
clips,
selectedClipId,
onClipSelect,
@@ -144,7 +142,6 @@ const RightPanel: React.FC<RightPanelProps> = ({
onOpenFilterDrawer={onOpenFilterDrawer}
onOpenGreenScreenDrawer={onOpenGreenScreenDrawer}
onOpenStickerDrawer={onOpenStickerDrawer}
onOpenCoverDrawer={onOpenCoverDrawer}
/>
)
})()}
@@ -2,9 +2,8 @@
* 封面选择器
* 抽帧选封面 + 上传自定义封面 + 智能封面推荐
*/
import React, { useState } from "react"
import { Drawer, Modal, Spin, message } from "antd"
import { generateCover } from "@/api/template-editor/aiFeatures"
import React from "react"
import { Drawer } from "antd"
import type { CoverConfig, CoverMode } from "../../types"
import { useCoverSelector, MODE_LABELS, MODE_ICONS } from "./useCoverSelector"
import { CoverAutoMode, CoverFrameMode, CoverUploadMode } from "./CoverModePanels"
@@ -15,7 +14,6 @@ interface CoverSelectorProps {
config: CoverConfig
onChange: (config: CoverConfig) => void
totalDuration: number
templateId: string
}
const CoverSelector: React.FC<CoverSelectorProps> = ({
@@ -24,7 +22,6 @@ const CoverSelector: React.FC<CoverSelectorProps> = ({
config,
onChange,
totalDuration,
templateId,
}) => {
const {
fileInputRef,
@@ -39,31 +36,6 @@ const CoverSelector: React.FC<CoverSelectorProps> = ({
formatTime,
} = useCoverSelector({ config, onChange })
const [generating, setGenerating] = useState(false)
const handleGenerateCover = async () => {
if (!templateId) {
message.error("请先保存模板")
return
}
setGenerating(true)
try {
const res = await generateCover(templateId, {
asset_ids: [],
cover_type: "ai_frame",
})
const imageUrl = res.cover?.image_url || res.cover?.thumbnail_url || ""
if (imageUrl) {
update({ upload_url: imageUrl, thumbnail_url: imageUrl })
}
message.success("封面生成成功")
} catch (err: any) {
message.error("封面生成失败: " + (err?.message || "未知错误"))
} finally {
setGenerating(false)
}
}
return (
<Drawer
title="封面选择"
@@ -161,33 +133,12 @@ const CoverSelector: React.FC<CoverSelectorProps> = ({
</div>
</div>
{/* AI 生成封面按钮 */}
{config.enabled && config.mode === "auto" && (
<div className="cover-generate-section">
<button
className="cover-generate-btn"
onClick={handleGenerateCover}
disabled={generating}
>
{generating ? "生成中..." : "🤖 AI 生成封面"}
</button>
</div>
)}
{/* 底部 */}
<div className="cover-footer">
<button className="cover-reset-btn" onClick={handleReset}>
</button>
</div>
{/* 生成进度弹窗 */}
<Modal open={generating} closable={false} footer={null} centered>
<div style={{ textAlign: "center", padding: "24px 0" }}>
<Spin size="large" />
<p style={{ marginTop: 16, fontSize: 14, color: "#666" }}>AI ...</p>
</div>
</Modal>
</Drawer>
)
}
@@ -17,7 +17,6 @@ export const useEditorDrawers = () => {
const [filterDrawerOpen, setFilterDrawerOpen] = useState(false)
const [chromaKeyDrawerOpen, setChromaKeyDrawerOpen] = useState(false)
const [stickerDrawerOpen, setStickerDrawerOpen] = useState(false)
const [coverDrawerOpen, setCoverDrawerOpen] = useState(false)
/* ── 目标片段 ID ── */
/** 当前正在编辑转场的片段 ID(null = 全局默认转场) */
@@ -67,8 +66,6 @@ export const useEditorDrawers = () => {
setChromaKeyDrawerOpen,
stickerDrawerOpen,
setStickerDrawerOpen,
coverDrawerOpen,
setCoverDrawerOpen,
// 目标 ID
transitionTargetClipId,
speedTargetClipId,
@@ -68,5 +68,4 @@ export interface ClipPropertiesPanelProps {
/** 打开贴纸面板 Drawer */
onOpenStickerDrawer?: () => void
/** 打开封面选择器 Drawer */
onOpenCoverDrawer?: () => void
}
@@ -1,5 +1,5 @@
import { afterEach, describe, expect, it, vi } from "vitest"
import { render, cleanup, act } from "@testing-library/react"
import { render, cleanup } from "@testing-library/react"
import TtsPanel from "@/pages/editing-planner/components/TtsPanel"
import { DEFAULT_TTS_CONFIG } from "@/pages/editing-planner/types"
@@ -16,28 +16,17 @@ const defaultProps = {
}
describe("TtsPanel", () => {
afterEach(async () => {
// Flush pending React updates before cleanup to avoid
// "window is not defined" errors after jsdom teardown
await act(async () => {})
afterEach(() => {
cleanup()
})
it("should render without crashing", async () => {
let container: HTMLElement
await act(async () => {
const result = render(<TtsPanel {...defaultProps} />)
container = result.container
})
expect(container!).toBeTruthy()
it("should render without crashing", () => {
const { container } = render(<TtsPanel {...defaultProps} />)
expect(container).toBeTruthy()
})
it("should render when closed", async () => {
let container: HTMLElement
await act(async () => {
const result = render(<TtsPanel {...defaultProps} open={false} />)
container = result.container
})
expect(container!).toBeTruthy()
it("should render when closed", () => {
const { container } = render(<TtsPanel {...defaultProps} open={false} />)
expect(container).toBeTruthy()
})
})
@@ -1436,64 +1436,6 @@ def generate_video(self, task_id: str) -> dict:
_update_task_progress(task_id, 30, "素材下载完成")
# ── 2.5 MediaKit 视频理解(渲染前分析素材内容)─────────────────
asset_analyses = {}
if task_asset_ids:
try:
from packages.shared.mediakit_client import get_mediakit_client
from packages.shared.storage import get_shared_storage_service
mk_client = get_mediakit_client()
if mk_client.is_available:
storage_svc = get_shared_storage_service()
asset_urls = []
valid_asset_ids = []
# 获取素材 URL(从 downloaded_videos 获取本地路径或从 asset 表获取 OSS URL
for i, asset_id in enumerate(task_asset_ids[:5]):
try:
# 优先使用已下载的本地文件
if i < len(downloaded_videos) and downloaded_videos[i]:
# 本地文件路径,需要上传或直接用 OSS URL
from packages.adapters.sqlalchemy_impl.asset_repository import (
SQLAlchemyAssetRepository,
)
_s = SessionLocal()
try:
_ar = SQLAlchemyAssetRepository(_s)
asset = _ar.get(asset_id)
if asset and getattr(asset, "storage_key", ""):
asset_urls.append(storage_svc.get_url(asset.storage_key))
valid_asset_ids.append(asset_id)
finally:
_s.close()
except Exception:
continue
if asset_urls:
_update_task_progress(task_id, 35, "MediaKit 视频理解中...")
analyses = mk_client.analyze_videos(
prompt="分析这个视频的主要内容,描述场景、物体、人物动作和主题",
video_urls=asset_urls,
level="Economy",
poll_interval=2.0,
max_poll_attempts=15,
)
if analyses:
for i, content in enumerate(analyses):
if i < len(valid_asset_ids) and content:
asset_analyses[valid_asset_ids[i]] = content
logger.info("[task_id=%s] MediaKit 视频理解完成: %d 个素材", task_id, len(asset_urls))
# 保存分析结果到 extra_meta
if asset_analyses and gen_task:
gen_task.extra_meta = {**(gen_task.extra_meta or {}), "asset_analyses": asset_analyses}
_repo.update(gen_task)
_flush_logs(task_id, gen_task)
except Exception:
logger.warning("[task_id=%s] MediaKit 视频理解失败,继续渲染", task_id, exc_info=True)
# ── 3. 渲染 + 混音 ───────────────────────────────────────────────
_update_task_progress(task_id, 40, "开始渲染")
# 动态分辨率:优先使用 output_width/output_height,其次 resolution 字符串
@@ -148,22 +148,6 @@ class SQLAlchemyGenerationTaskRepository:
)
return [_to_domain(m) for m in models]
def list_latest_completed_preview(self, user_id: str, template_id: str, limit: int = 1) -> list[GenerationTask]:
"""按用户+模板查找最近已完成的预览任务。"""
models = (
self.session.query(GenerationTaskModel)
.filter(
GenerationTaskModel.created_by_user_id == user_id,
GenerationTaskModel.template_id == template_id,
GenerationTaskModel.is_preview,
GenerationTaskModel.status == "completed",
)
.order_by(GenerationTaskModel.created_at.desc())
.limit(limit)
.all()
)
return [_to_domain(m) for m in models]
def list_by_source_edit_plan(self, plan_id: str) -> list[GenerationTask]:
models = (
self.session.query(GenerationTaskModel)
+18 -7
View File
@@ -362,7 +362,7 @@ def _call_ai_cover_service(
"""调用 AI 封面生成服务.
当 cover_type 为 ai_frame 或 ai_regenerate 时,调用 MediaKit 视频截帧。
失败时抛出 RuntimeError
失败或未配置时降级为 stub 行为
Args:
plan_id: 剪辑计划 ID
@@ -404,8 +404,6 @@ def _call_ai_cover_service(
video_url=primary_video_url,
strategy="TimeInterval",
max_frames=5,
poll_interval=3.0,
max_poll_attempts=60, # 180秒超时
)
if frames and len(frames) > 0:
@@ -434,12 +432,25 @@ def _call_ai_cover_service(
logger.warning("MediaKit 返回的帧无 image_url")
except Exception as e:
logger.exception("MediaKit 抽帧失败: %s", str(e))
logger.exception("MediaKit 抽帧失败,降级到 stub: %s", str(e))
# 封面生成失败 - 不再降级到 stub,直接报错
raise RuntimeError(
f"封面生成失败: plan_id={plan_id}, MediaKit 不可用或抽帧失败。" f"请检查 primary_video_url 是否可访问。"
# 降级:stub 行为 - 返回 SVG data URI 占位图
logger.info("使用 stub 封面: plan_id=%s", plan_id)
time.sleep(0.3)
svg_placeholder = (
"data:image/svg+xml,"
"<svg xmlns='http://www.w3.org/2000/svg' width='1080' height='1920'>"
"<rect width='1080' height='1920' fill='#1a1a2e'/>"
"<text x='540' y='920' text-anchor='middle' fill='#e0e0e0' font-size='48' font-family='sans-serif'>封面生成中</text>"
"<text x='540' y='1000' text-anchor='middle' fill='#888888' font-size='32' font-family='sans-serif'>请配置 MediaKit API Key</text>"
"</svg>"
)
return {
"type": "ai_frame",
"image_url": svg_placeholder,
"frame_time": round(random.uniform(1.0, 10.0), 1),
"confidence": round(random.uniform(0.80, 0.98), 2),
}
# ── 公共入口 ────────────────────────────────────────────────────────────────
+11 -9
View File
@@ -399,16 +399,18 @@ class TestRunAIRecommend(unittest.TestCase):
class TestGenerateCover(unittest.TestCase):
"""封面生成测试."""
"""封面生成测试(降级路径)."""
def test_ai_frame_type_raises_without_mediakit(self):
"""AI封面模式在MediaKit不可用时抛出RuntimeError."""
with self.assertRaises(RuntimeError):
run_generate_cover(
plan_id="plan-1",
asset_ids=["a1"],
cover_type="ai_frame",
)
def test_ai_frame_type(self):
"""AI封面模式返回预期结构."""
result = run_generate_cover(
plan_id="plan-1",
asset_ids=["a1"],
cover_type="ai_frame",
)
self.assertIn("type", result)
self.assertEqual(result["type"], "ai_frame")
self.assertIn("image_url", result)
def test_manual_type(self):
"""手动选帧模式."""
@@ -231,18 +231,18 @@ class TestAIRunTasks:
# 即使没有素材,也应该有 intro + outro
assert len(result["clips"]) >= 2
def test_run_generate_cover_ai_frame_raises_without_mediakit(self):
"""ai_frame cover raises RuntimeError when MediaKit is unavailable."""
import pytest
def test_run_generate_cover_ai_frame(self):
from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover
with pytest.raises(RuntimeError, match="MediaKit"):
run_generate_cover(
plan_id="plan-001",
asset_ids=["asset-1"],
cover_type="ai_frame",
)
result = run_generate_cover(
plan_id="plan-001",
asset_ids=["asset-1"],
cover_type="ai_frame",
)
assert result["type"] == "ai_frame"
assert "image_url" in result
assert "frame_time" in result
assert "confidence" in result
def test_run_generate_cover_manual(self):
from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover
+53 -5
View File
@@ -630,7 +630,7 @@ class TestToPreviewResponse:
assert resp.file_size == 0
def test_completed_task_with_videos(self):
"""已完成任务,带视频结果(URL直接返回,rendered/*已公开读"""
"""已完成任务,带视频结果(URL签名后返回"""
task = _make_task(
status=GenerationTaskStatus.COMPLETED,
progress=100.0,
@@ -639,8 +639,12 @@ class TestToPreviewResponse:
video.file_url = "https://cdn.example.com/preview.mp4"
video.duration = 30.5
video.file_size = 1024000
resp = _to_preview_response(task, generated_videos=[video])
assert resp.video_url == "https://cdn.example.com/preview.mp4"
# Mock storage service to return a signed URL
mock_storage = MagicMock()
mock_storage.get_download_url.return_value = "https://cdn.example.com/preview.mp4?sig=test123"
with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
resp = _to_preview_response(task, generated_videos=[video])
assert resp.video_url == "https://cdn.example.com/preview.mp4?sig=test123"
assert resp.duration == 30.5
assert resp.file_size == 1024000
@@ -982,7 +986,7 @@ class TestGetPreviewRoute:
# Mock URL 签名(返回带签名的 URL)
mock_storage = MagicMock()
mock_storage.get_download_url.return_value = "https://cdn.example.com/preview_final.mp4"
mock_storage.get_download_url.return_value = "https://cdn.example.com/preview_final.mp4?sig=abc123"
with patch("app.api.routes.generation_preview.GetGenerationTaskUseCase") as MockGet:
MockGet.return_value.execute.return_value = task
@@ -995,7 +999,7 @@ class TestGetPreviewRoute:
generation_task_repository=repo,
generated_video_repository=vid_repo,
)
assert resp.video_url == "https://cdn.example.com/preview_final.mp4"
assert resp.video_url == "https://cdn.example.com/preview_final.mp4?sig=abc123"
assert resp.duration == 25.0
@@ -1084,6 +1088,50 @@ class TestWorkerPreviewResolution:
# ═══════════════════════════════════════════════════════════════════════════════
class TestSignVideoUrl:
"""_sign_video_url 预签名 URL 测试。"""
def test_empty_url_returns_empty(self):
"""空 URL 直接返回空字符串。"""
from app.api.routes.generation_preview import _sign_video_url
assert _sign_video_url("") == ""
def test_signs_oss_url(self):
"""OSS URL 应被签名。"""
from app.api.routes.generation_preview import _sign_video_url
mock_storage = MagicMock()
mock_storage.get_download_url.return_value = "https://signed.example.com/video.mp4?sig=abc"
with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
result = _sign_video_url("https://bucket.oss.example.com/video.mp4")
assert result == "https://signed.example.com/video.mp4?sig=abc"
mock_storage.get_download_url.assert_called_once()
def test_fallback_on_sign_failure(self):
"""签名失败时降级返回原始 URL。"""
from app.api.routes.generation_preview import _sign_video_url
with patch("app.api.routes.generation_preview.get_storage_service", side_effect=RuntimeError("no storage")):
result = _sign_video_url("https://bucket.oss.example.com/video.mp4")
assert result == "https://bucket.oss.example.com/video.mp4"
def test_sign_returns_none_fallback(self):
"""get_download_url 返回 None 时降级返回原始 URL。"""
from app.api.routes.generation_preview import _sign_video_url
mock_storage = MagicMock()
mock_storage.get_download_url.return_value = None
with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
result = _sign_video_url("https://bucket.oss.example.com/video.mp4")
assert result == "https://bucket.oss.example.com/video.mp4"
class TestInferVideoRatioFromTemplate:
"""_infer_video_ratio_from_template 单元测试。"""
+44 -36
View File
@@ -167,8 +167,8 @@ class TestAICoverService:
mock_client.extract_frames.assert_called_once()
@patch("packages.shared.mediakit_client.get_mediakit_client")
def test_call_ai_cover_with_mediakit_failure_raises(self, mock_get_client):
"""MediaKit 失败时抛出 RuntimeError(不再降级到 stub."""
def test_call_ai_cover_with_mediakit_failure_fallback(self, mock_get_client):
"""MediaKit 失败时降级到 stub."""
mock_client = Mock()
mock_client.is_available = True
mock_client.extract_frames.side_effect = Exception("API error")
@@ -176,26 +176,30 @@ class TestAICoverService:
from packages.shared.ai_service import _call_ai_cover_service
with pytest.raises(RuntimeError, match="MediaKit"):
_call_ai_cover_service(
plan_id="plan-123",
asset_ids=["asset-1"],
cover_type="ai_frame",
primary_video_url="https://example.com/video.mp4",
)
result = _call_ai_cover_service(
plan_id="plan-123",
asset_ids=["asset-1"],
cover_type="ai_frame",
primary_video_url="https://example.com/video.mp4",
)
def test_call_ai_cover_without_video_url_raises(self):
"""没有视频 URL 时抛出 RuntimeError(不再降级到 stub."""
# 应该降级到 stub
assert result["type"] == "ai_frame"
assert result["image_url"].startswith("data:image/svg+xml,")
def test_call_ai_cover_without_video_url_fallback(self):
"""没有视频 URL 时使用 stub."""
from packages.shared.ai_service import _call_ai_cover_service
with pytest.raises(RuntimeError, match="MediaKit"):
_call_ai_cover_service(
plan_id="plan-123",
asset_ids=["asset-1"],
cover_type="ai_frame",
primary_video_url=None,
)
result = _call_ai_cover_service(
plan_id="plan-123",
asset_ids=["asset-1"],
cover_type="ai_frame",
primary_video_url=None,
)
assert result["type"] == "ai_frame"
assert result["image_url"].startswith("data:image/svg+xml,")
def test_call_ai_cover_upload_type(self):
"""upload 类型直接返回."""
@@ -227,25 +231,27 @@ class TestAICoverService:
assert result["frame_time"] == 5.0
@patch("packages.shared.mediakit_client.get_mediakit_client")
def test_call_ai_cover_mediakit_not_available_raises(self, mock_get_client):
"""MediaKit 未配置时抛出 RuntimeError(不再降级到 stub."""
def test_call_ai_cover_mediakit_not_available(self, mock_get_client):
"""MediaKit 未配置时使用 stub."""
mock_client = Mock()
mock_client.is_available = False
mock_get_client.return_value = mock_client
from packages.shared.ai_service import _call_ai_cover_service
with pytest.raises(RuntimeError, match="MediaKit"):
_call_ai_cover_service(
plan_id="plan-123",
asset_ids=["asset-1"],
cover_type="ai_frame",
primary_video_url="https://example.com/video.mp4",
)
result = _call_ai_cover_service(
plan_id="plan-123",
asset_ids=["asset-1"],
cover_type="ai_frame",
primary_video_url="https://example.com/video.mp4",
)
assert result["type"] == "ai_frame"
assert result["image_url"].startswith("data:image/svg+xml,")
@patch("packages.shared.mediakit_client.get_mediakit_client")
def test_call_ai_cover_empty_frames_raises(self, mock_get_client):
"""MediaKit 返回空帧列表时抛出 RuntimeError(不再降级."""
def test_call_ai_cover_empty_frames_fallback(self, mock_get_client):
"""MediaKit 返回空帧列表时降级."""
mock_client = Mock()
mock_client.is_available = True
mock_client.extract_frames.return_value = []
@@ -253,13 +259,15 @@ class TestAICoverService:
from packages.shared.ai_service import _call_ai_cover_service
with pytest.raises(RuntimeError, match="MediaKit"):
_call_ai_cover_service(
plan_id="plan-123",
asset_ids=["asset-1"],
cover_type="ai_frame",
primary_video_url="https://example.com/video.mp4",
)
result = _call_ai_cover_service(
plan_id="plan-123",
asset_ids=["asset-1"],
cover_type="ai_frame",
primary_video_url="https://example.com/video.mp4",
)
assert result["type"] == "ai_frame"
assert result["image_url"].startswith("data:image/svg+xml,")
class TestGenerateCover:
+27 -15
View File
@@ -434,22 +434,34 @@ class TestAiCoverService:
assert result["image_url"].startswith("data:image/svg+xml,")
assert "手动选帧" in result["image_url"]
def test_cover_type_ai_frame_raises_without_mediakit(self):
"""ai_frame mode raises RuntimeError when MediaKit is unavailable."""
with pytest.raises(RuntimeError, match="MediaKit"):
_call_ai_cover_service("plan1", ["a1"], "ai_frame")
def test_cover_type_ai_regenerate_raises_without_mediakit(self):
"""ai_regenerate mode raises RuntimeError when MediaKit is unavailable."""
with pytest.raises(RuntimeError, match="MediaKit"):
_call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
def test_cover_type_manual_still_works(self):
"""manual mode does not require MediaKit and still returns stub."""
def test_cover_type_ai_frame(self):
with patch("shared.ai_service.time.sleep"):
result = _call_ai_cover_service("plan1", ["a1"], "manual", frame_time=5.5)
assert result["type"] == "manual"
assert result["frame_time"] == 5.5
with patch("shared.ai_service.random.uniform", side_effect=[5.0, 0.9]):
result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
assert result["type"] == "ai_frame"
assert result["frame_time"] == 5.0
assert result["confidence"] == 0.9
assert result["image_url"].startswith("data:image/svg+xml,")
assert "封面生成中" in result["image_url"]
def test_cover_type_ai_regenerate(self):
with patch("shared.ai_service.time.sleep"):
result = _call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
assert result["type"] == "ai_frame"
def test_cover_frame_time_in_range(self):
with patch("shared.ai_service.time.sleep"):
result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
assert 1.0 <= result["frame_time"] <= 10.0
def test_stub_returns_svg_data_uri(self):
"""stub 降级返回 SVG data URI,不含任何后端 API 路径."""
with patch("shared.ai_service.time.sleep"):
result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
assert result["image_url"].startswith("data:image/svg+xml,")
assert "/api/v1/" not in result["image_url"]
assert "1080" in result["image_url"]
assert "1920" in result["image_url"]
def test_manual_stub_returns_svg_data_uri(self):
"""manual 模式返回 SVG data URI 占位图."""