Compare commits

..

1 Commits

Author SHA1 Message Date
xiaoxia ba8d0afa3c fix: 预览生成标题参数改用title_config对象匹配后端Schema
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 / Check if frontend-only change (pull_request) Successful in 58s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 43s
CI/CD Pipeline / PR Build API Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m46s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 45s
CI/CD Pipeline / PR Build Worker Image (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
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m49s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m52s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 1m51s
AI Code Review / AI Code Review (pull_request) Successful in 2m0s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m59s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 1m31s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 4m45s
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 1m11s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Has been cancelled
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 40s
- CreatePreviewRequest: 删除8个独立title_*字段,改为title_config对象
- useStep5Preview: generatePreview传参改为title_config嵌套结构
- 字段映射: titleSettings.size → title_config.font_size
- 修复标题配置被Pydantic忽略导致预览视频/封面无标题的问题
2026-08-15 17:48:43 +08:00
27 changed files with 308 additions and 1065 deletions
@@ -315,32 +315,6 @@ def create_preview_generation_task(
logger.error("[预览生成] 创建失败: %s", e, exc_info=True)
raise HTTPException(status_code=500, detail="创建预览生成任务失败,请稍后再试") from e
# 关联编辑计划:如果前端未传 source_edit_plan_id,通过 template_id + user_id 查找
if not task.source_edit_plan_id and request.template_id:
try:
from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
SQLAlchemyEditPlanRepository,
)
_plan_repo = SQLAlchemyEditPlanRepository(db)
_plans = _plan_repo.list_by_template(request.template_id, limit=20)
for _p in _plans:
if (_p.created_by_user_id or "") == user_id:
task.source_edit_plan_id = _p.id
generation_task_repository.update(task)
logger.info(
"[预览生成] 自动关联编辑计划: task_id=%s plan_id=%s",
task.id,
_p.id,
)
break
except Exception:
logger.warning(
"[预览生成] 查找关联编辑计划失败(不影响主流程): task_id=%s",
task.id,
exc_info=True,
)
# 入队执行;若入队失败则标记任务为 failed 避免僵尸数据
try:
if not safe_enqueue_generation_task(
+9 -9
View File
@@ -200,7 +200,15 @@ test.describe("Core generation flow", () => {
await expect(page.getByRole("heading", { name: /选择配音/ })).toBeVisible({ timeout: 15000 })
await page.getByRole("button", { name: "下一步" }).click()
// Step 4: title(新顺序:标题在预览之前)
// Step 4: preview — 需要先生成预览视频,才能进入下一步
await expect(page.getByRole("heading", { name: /生成预览/ })).toBeVisible({ timeout: 15000 })
// 点击"生成预览"按钮触发预览生成
await page.locator(".xx-preview-generate-btn").click()
// 等待预览生成完成(后端渲染,可能需要较长时间)
await expect(page.getByText("预览生成成功")).toBeVisible({ timeout: 300_000 })
await page.getByRole("button", { name: "下一步" }).click()
// Step 5: title
await expect(page.getByRole("heading", { name: /选择标题/ })).toBeVisible({ timeout: 15000 })
// 等待组件完全渲染
await page.waitForTimeout(2000)
@@ -214,14 +222,6 @@ test.describe("Core generation flow", () => {
await titleInput.fill(titleText)
await page.getByRole("button", { name: "下一步" }).click()
// Step 5: preview — 需要先生成预览视频,才能进入下一步
await expect(page.getByRole("heading", { name: /生成预览/ })).toBeVisible({ timeout: 15000 })
// 点击"生成预览"按钮触发预览生成
await page.locator(".xx-preview-generate-btn").click()
// 等待预览生成完成(后端渲染,可能需要较长时间)
await expect(page.getByText("预览生成成功")).toBeVisible({ timeout: 300_000 })
await page.getByRole("button", { name: "下一步" }).click()
// Step 6: cover (默认 AI 智能选帧模式,直接下一步)
await expect(page.getByRole("heading", { name: /选择封面/ })).toBeVisible({ timeout: 15000 })
await page.getByRole("button", { name: "下一步" }).click()
+3 -22
View File
@@ -12,10 +12,7 @@
import React, { useState, useMemo } from "react"
import { Modal, message } from "antd"
import { useNavigate } from "react-router-dom"
import { useQuery } from "@tanstack/react-query"
import type { VoiceClone } from "@/api/voice-clone"
import { getAssetsByKind } from "@/api/assets/assets"
import type { AssetItem } from "@/api/assets/types"
import { useCloneProgress } from "@/hooks/useCloneProgress"
import CloneModal from "@/components/voice/CloneModal"
import GenerateHeader from "./components/GenerateHeader"
@@ -74,20 +71,6 @@ const GeneratePage: React.FC = () => {
setPreviewModalOpen,
} = formState
/* ── 查询视频素材,用于 Step4 标题预览背景 ── */
const { data: videoAssets = [] } = useQuery({
queryKey: ["generate-video-assets"],
queryFn: () => getAssetsByKind("video", { limit: 50 }),
})
// 获取第一个选中素材的 URL
const sourceVideoUrl = useMemo(() => {
const firstId = selectedMaterials[0]
if (!firstId) return undefined
const asset = videoAssets.find((a: AssetItem) => a.id === firstId)
return asset?.file_url
}, [selectedMaterials, videoAssets])
/* ── 克隆声音 ── */
const { clones: clonedVoices, addClone, hasProcessing } = useCloneProgress()
@@ -254,8 +237,8 @@ const GeneratePage: React.FC = () => {
{/* ════ 右侧:预览 + 生成结果 ════ */}
<div className="xx-generate-right-col">
{/* 预览视频面板(Step4+ 常驻,Step4 显示标题预览,Step5+ 显示预览视频 */}
{currentStep >= 4 && (
{/* 预览视频面板(Step5+ 常驻,展示选中的预览 */}
{currentStep >= 5 && (
<PreviewVideoPanel
previewStatus={step5Preview.previewStatus}
previewResult={step5Preview.previewResult}
@@ -264,9 +247,7 @@ const GeneratePage: React.FC = () => {
videoRatio={videoRatio}
onRegenerate={step5Preview.regeneratePreview}
titleText={titleSettings.title}
titleSettings={titleSettings}
showTitlePreview={currentStep === 4}
sourceVideoUrl={sourceVideoUrl}
titleSettings={currentStep >= 4 ? titleSettings : undefined}
/>
)}
@@ -16,7 +16,6 @@ import { PlayCircleOutlined, LoadingOutlined } from "@ant-design/icons"
import type { PreviewResult, PreviewStatus } from "../hooks/useStep5Preview"
import type { TitleSettings } from "../types"
import { drawTitleOnCanvas } from "../utils/drawTitleOnCanvas"
import TitlePreviewCanvas from "./title/TitlePreviewCanvas"
interface PreviewVideoPanelProps {
previewStatus: PreviewStatus
@@ -25,14 +24,10 @@ interface PreviewVideoPanelProps {
progress: number
videoRatio: string
onRegenerate: () => void
/** 标题文字 */
/** 标题文字Step5 起传入) */
titleText?: string
/** 标题样式设置 */
/** 标题样式设置Step5 起传入) */
titleSettings?: TitleSettings
/** Step4 标题预览模式 */
showTitlePreview?: boolean
/** 素材视频 URL(用于 Step4 标题预览背景) */
sourceVideoUrl?: string
}
/* ── 组件 ── */
@@ -46,12 +41,11 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
onRegenerate,
titleText,
titleSettings,
showTitlePreview,
sourceVideoUrl,
}) => {
const hasPreview = previewStatus === "ready" && previewResult
const isLoading = previewStatus === "pending" || previewStatus === "generating"
const isError = previewStatus === "error"
const showTitlePreview = !!titleSettings
const videoAspectStyle = { aspectRatio: (videoRatio || "16:9").replace(":", "/") }
// video 模式 refs
@@ -176,46 +170,23 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
return (
<div className="xx-generate-preview">
<div className="xx-preview-header">
<h3>{showTitlePreview && !hasPreview ? "标题预览" : "预览视频"}</h3>
{hasPreview && !showTitlePreview && <span className="xx-preview-badge">480p </span>}
<h3></h3>
{hasPreview && <span className="xx-preview-badge">480p </span>}
</div>
{/* Step4 标题预览模式 */}
{showTitlePreview && titleSettings && titleText && (
<div style={{ padding: "0 16px 16px" }}>
<TitlePreviewCanvas
titleText={titleText}
titleSettings={titleSettings}
videoRatio="9:16"
sourceVideoUrl={sourceVideoUrl}
/>
</div>
)}
{/* Step4 但无标题时的空状态 */}
{showTitlePreview && (!titleText || !titleSettings) && (
<div className="xx-preview-empty">
<PlayCircleOutlined
style={{ fontSize: 48, color: "var(--text-tertiary)", marginBottom: 12 }}
/>
<p className="xx-preview-empty-title"></p>
<p className="xx-preview-empty-desc"></p>
</div>
)}
{/* 空状态:还没生成预览(非 Step4 模式) */}
{!showTitlePreview && previewStatus === "idle" && (
{/* 空状态:还没生成预览 */}
{previewStatus === "idle" && (
<div className="xx-preview-empty">
<PlayCircleOutlined
style={{ fontSize: 48, color: "var(--text-tertiary)", marginBottom: 12 }}
/>
<p className="xx-preview-empty-title"></p>
<p className="xx-preview-empty-desc"></p>
<p className="xx-preview-empty-desc"> 3 </p>
</div>
)}
{/* 生成中(非 Step4 模式) */}
{!showTitlePreview && isLoading && (
{/* 生成中 */}
{isLoading && (
<div className="xx-preview-loading-panel">
<div className="xx-preview-video" style={videoAspectStyle}>
<div className="xx-preview-loading-center">
@@ -231,8 +202,8 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
</div>
)}
{/* 生成失败(非 Step4 模式) */}
{!showTitlePreview && isError && (
{/* 生成失败 */}
{isError && (
<div className="xx-preview-error-panel">
<div className="xx-preview-video xx-preview-video--error" style={videoAspectStyle}>
<p style={{ color: "rgba(255,255,255,0.8)", fontSize: 14 }}></p>
@@ -275,8 +246,8 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
</div>
)}
{/* 预览信息(非 Step4 模式) */}
{!showTitlePreview && hasPreview && previewResult && (
{/* 预览信息 */}
{hasPreview && previewResult && (
<div className="xx-preview-info">
<div className="xx-preview-info-row">
<span></span>
@@ -8,6 +8,7 @@ import type { TitleSettings } from "../types"
import { useStep4Title } from "../hooks/useStep4Title"
import AiTitleGenerator from "./title/AiTitleGenerator"
import TitleStylePanel from "./title/TitleStylePanel"
import TitlePreviewCanvas from "./title/TitlePreviewCanvas"
interface Step4TitleSettingsProps {
titleSettings: TitleSettings
@@ -111,6 +112,11 @@ const Step4TitleSettings: React.FC<Step4TitleSettingsProps> = (props) => {
</>
)}
{/* 标题实时预览 - 仅当有标题文字时显示 */}
{t.titleSettings.title && (
<TitlePreviewCanvas titleText={t.titleSettings.title} titleSettings={t.titleSettings} />
)}
{/* 样式面板始终可见,两种模式下都可调整标题展示样式 */}
<TitleStylePanel
settings={t.titleSettings}
@@ -15,24 +15,21 @@ interface TitlePreviewCanvasProps {
titleText: string
/** 标题样式设置 */
titleSettings: TitleSettings
/** 视频比例,默认 "9:16"(竖屏) */
/** 视频比例,默认 "16:9" */
videoRatio?: string
/** 素材视频 URL(作为背景显示) */
sourceVideoUrl?: string
}
/**
* 解析 videoRatio 字符串为 aspect-ratio CSS 值
*/
function parseAspect(ratio: string): string {
return (ratio || "9:16").replace(":", "/")
return (ratio || "16:9").replace(":", "/")
}
const TitlePreviewCanvas: React.FC<TitlePreviewCanvasProps> = ({
titleText,
titleSettings,
videoRatio = "9:16",
sourceVideoUrl,
videoRatio = "16:9",
}) => {
const containerRef = useRef<HTMLDivElement>(null)
const canvasRef = useRef<HTMLCanvasElement>(null)
@@ -86,7 +83,6 @@ const TitlePreviewCanvas: React.FC<TitlePreviewCanvasProps> = ({
.load()
.then(() => {
if (!cancelled) {
// eslint-disable-next-line @typescript-eslint/no-explicit-any -- FontFaceSet.add() exists at runtime
;(document.fonts as any).add(fontFace)
onFontReady()
}
@@ -109,13 +105,11 @@ const TitlePreviewCanvas: React.FC<TitlePreviewCanvasProps> = ({
return () => {
cancelled = true
}
// eslint-disable-next-line react-hooks/exhaustive-deps -- draw uses refs, stable by design
}, [titleSettings.font, titleSettings.size, titleSettings.bold, titleSettings.italic])
// props 变化时重绘
useEffect(() => {
requestAnimationFrame(draw)
// eslint-disable-next-line react-hooks/exhaustive-deps -- draw uses refs, stable by design
}, [titleText, titleSettings])
// ResizeObserver 监听容器尺寸变化
@@ -129,7 +123,6 @@ const TitlePreviewCanvas: React.FC<TitlePreviewCanvasProps> = ({
observer.observe(container)
return () => observer.disconnect()
// eslint-disable-next-line react-hooks/exhaustive-deps -- draw uses refs, stable by design
}, [])
return (
@@ -149,31 +142,12 @@ const TitlePreviewCanvas: React.FC<TitlePreviewCanvasProps> = ({
style={{
width: "100%",
aspectRatio: parseAspect(videoRatio),
background: sourceVideoUrl
? "#000"
: "linear-gradient(135deg, #1a1a2e, #16213e, #0f3460)",
background: "linear-gradient(135deg, #1a1a2e, #16213e, #0f3460)",
borderRadius: 8,
overflow: "hidden",
position: "relative",
}}
>
{sourceVideoUrl && (
<video
src={sourceVideoUrl}
muted
loop
autoPlay
playsInline
style={{
position: "absolute",
top: 0,
left: 0,
width: "100%",
height: "100%",
objectFit: "cover",
}}
/>
)}
<canvas
ref={canvasRef}
style={{
+22 -38
View File
@@ -81,9 +81,7 @@ class RenderAdapterResult:
failed_clip_ids: list[str] = None # 失败的 clip id 列表
error_message: str = ""
error_detail: str = "" # 详细错误信息(如 ffmpeg stderr),用于排查
cover_candidates: list[dict] | None = (
None # 封面候选帧 [{"image_url": "...", "frame_time": 5.0, "storage_key": "..."}]
)
cover_url: str = "" # 封面图片 URL(从渲染后视频抽帧,天然带标题)
def __post_init__(self):
if self.rendered_clip_ids is None:
@@ -519,8 +517,6 @@ class RenderAdapter:
# 3. 读取输出分辨率
export_config = plan_config.get("export", {}) or {}
if not isinstance(export_config, dict):
export_config = {}
output_width, output_height = _parse_resolution(export_config.get("resolution"))
logger.info(
"渲染输出分辨率: plan_id=%s resolution=%dx%d source=%s",
@@ -564,45 +560,33 @@ class RenderAdapter:
storage_key = f"rendered/{plan_id}/{job_id or plan_id}.mp4"
output_url = upload_to_oss(result.output_path, storage_key)
self._report_progress(progress_cb, 90.0, "生成封面缩略图")
self._report_progress(progress_cb, 90.0, "抽取封面帧")
# 6. 生成封面缩略图
thumbnail_url = ""
# 6. 从已渲染视频抽取封面帧(标题已通过 ASS 字幕烧录,封面天然带标题)
cover_url = ""
cover_frame_path = None
try:
from video_processing.thumbnail_generator import generate_and_upload_thumbnail
from video_processing.thumbnail_generator import extract_first_frame
thumb_storage_key = f"rendered/{plan_id}/thumbnail.jpg"
thumbnail_url = generate_and_upload_thumbnail(str(result.output_path), thumb_storage_key)
except Exception as thumb_err:
logger.warning(
"[render-adapter] 缩略图生成失败(不影响主流程): plan_id=%s error=%s",
plan_id,
thumb_err,
)
# 7. 抽取封面候选帧并上传 OSS(失败不阻断主流程)
cover_candidates = None
try:
from video_processing.thumbnail_generator import extract_and_upload_cover_frames
# 从 plan config 提取标题文字,叠加到封面候选帧上
_title_cfg = (plan_config or {}).get("title", {}) or {}
if not isinstance(_title_cfg, dict):
_title_cfg = {}
_title_text = (_title_cfg.get("text", "") or "").strip() if _title_cfg.get("enabled", True) else ""
cover_candidates = extract_and_upload_cover_frames(
str(result.output_path), plan_id, num_frames=3, title_text=_title_text
)
if cover_candidates:
cover_frame_path = extract_first_frame(str(result.output_path), width=640)
cover_storage_key = f"rendered/{plan_id}/cover.jpg"
try:
cover_url = upload_to_oss(cover_frame_path, cover_storage_key) or ""
finally:
if cover_frame_path:
try:
Path(cover_frame_path).unlink(missing_ok=True)
except Exception:
pass
if cover_url:
logger.info(
"[render-adapter] 封面候选帧生成成功: plan_id=%s count=%d",
"[render-adapter] 封面帧提取成功: plan_id=%s url=%s",
plan_id,
len(cover_candidates),
cover_url[:80],
)
except Exception as cover_err:
logger.warning(
"[render-adapter] 封面候选帧生成失败(不影响主流程): plan_id=%s error=%s",
"[render-adapter] 封面帧提取失败(不影响主流程): plan_id=%s error=%s",
plan_id,
cover_err,
)
@@ -630,7 +614,7 @@ class RenderAdapter:
success=True,
output_url=output_url or "",
output_path=result.output_path,
thumbnail_url=thumbnail_url,
thumbnail_url=cover_url,
duration=result.duration,
file_size=result.file_size,
width=result.width,
@@ -638,7 +622,7 @@ class RenderAdapter:
clip_count=len(clips),
rendered_clip_ids=final_rendered_ids,
failed_clip_ids=final_failed_ids,
cover_candidates=cover_candidates,
cover_url=cover_url,
)
def render_from_memory(
+3 -3
View File
@@ -131,7 +131,7 @@ def mix_audio(
if not main_clips and not audio_clips:
# 没有主音频也没有独立音频 → 检查是否有 BGM
if bgm_path and bgm_config and isinstance(bgm_config, dict) and bgm_config.get("enabled", False):
if bgm_path and bgm_config and bgm_config.get("enabled", False):
from video_processing.bgm_mixer import BGMConfig, build_bgm_only
bgm_cfg = BGMConfig.from_config_dict(bgm_path, bgm_config)
@@ -161,7 +161,7 @@ def mix_audio(
mix_with_independent_audio(ctx, effective_main, effective_audio, output_path, video_duration)
# ── BGM 混音 ──
if bgm_path and bgm_config and isinstance(bgm_config, dict) and bgm_config.get("enabled", False):
if bgm_path and bgm_config and bgm_config.get("enabled", False):
from video_processing.bgm_mixer import BGMConfig, mix_bgm_with_main
bgm_cfg = BGMConfig.from_config_dict(bgm_path, bgm_config)
@@ -174,7 +174,7 @@ def mix_audio(
logger.exception("[bgm] BGM 混音失败,回退到无 BGM 音频: plan_id=%s", ctx.plan_id)
# ── 多轨道混音(配音/音效等) ──
if audio_tracks_config and isinstance(audio_tracks_config, dict) and audio_tracks_config.get("enabled", False):
if audio_tracks_config and audio_tracks_config.get("enabled", False):
from video_processing.multi_track_mixer import mix_audio_tracks_from_config
try:
@@ -33,7 +33,7 @@ class ReverseConfig:
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> "ReverseConfig":
"""从字典解析配置."""
if not isinstance(data, dict):
if not data:
return cls(enabled=False)
try:
if not data.get("enabled", False):
@@ -52,6 +52,7 @@ from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_fr
from video_processing.tts_engine import TtsEngine
from video_processing.watermark_engine import WatermarkConfig, WatermarkEngine
from packages.domain.ass_subtitle_builder import build_ass_content
from packages.domain.render_layer_utils import LAYER_Z_INDEX as _IMPORTED_LAYER_Z_INDEX
from packages.domain.render_layer_utils import clip_adjusted_duration as _clip_adjusted_duration_pure
from packages.domain.render_layer_utils import clip_effective_duration as _clip_effective_duration_pure
@@ -131,6 +132,83 @@ _PIP_SCALE = 0.25 # PiP 占主画面的比例
# ── 统一渲染引擎 ─────────────────────────────────────────────────────────────
def _overlay_title_on_ass(
ass_path: Path,
*,
title_text: str,
title_config: dict,
video_width: int,
video_height: int,
video_duration: float,
) -> None:
"""在已有的 ASS 文件上叠加标题事件。
用于 ASR 字幕路径:ASR 生成的 ASS 只含字幕事件,此函数将标题
作为独立的 TitleStyle + Dialogue 事件追加进去,使标题显示在
ASR 字幕之上(封面抽帧时也能看到标题)。
Args:
ass_path: 已有的 ASS 文件路径(由 generate_ass_from_timeline 生成)
title_text: 标题文本
title_config: 标题样式配置
video_width: 视频宽度
video_height: 视频高度
video_duration: 视频时长
"""
if not title_text or not title_text.strip():
return
# 生成仅包含标题的 ASS 内容
title_only_content = build_ass_content(
video_width=video_width,
video_height=video_height,
video_duration=video_duration,
title_text=title_text,
title_config=title_config,
)
if not title_only_content:
return
# 从 title_only_content 中提取 TitleStyle 行和标题 Dialogue 行
title_style_line = None
title_dialogue_line = None
for line in title_only_content.splitlines():
if line.startswith("Style: TitleStyle"):
title_style_line = line
elif "TitleStyle" in line and line.startswith("Dialogue:"):
title_dialogue_line = line
if not title_style_line or not title_dialogue_line:
logger.warning("标题 ASS 内容解析失败,跳过叠加")
return
# 读取现有 ASS 文件
existing_content = ass_path.read_text(encoding="utf-8")
# 插入 TitleStyle 到 [V4+ Styles] 段(最后一个 Style: 行之后)
# 插入标题 Dialogue 到 [Events] 段(Format 行之后)
lines = existing_content.splitlines()
last_style_idx = -1
events_format_idx = -1
for i, line in enumerate(lines):
if line.startswith("Style:"):
last_style_idx = i
if line.startswith("Format: Layer,"):
events_format_idx = i
if last_style_idx >= 0:
lines.insert(last_style_idx + 1, title_style_line)
# events_format_idx 需要 +1 因为插入了一行
events_format_idx += 1
# 2. 在 Events Format 行之后、第一个 Dialogue 之前插入标题 Dialogue
# 标题应该显示在整个视频时长,放在最前面(最先渲染,在底层)
if events_format_idx >= 0:
lines.insert(events_format_idx + 1, title_dialogue_line)
ass_path.write_text("\n".join(lines), encoding="utf-8")
class UnifiedRenderService:
"""统一渲染引擎。
@@ -324,8 +402,6 @@ class UnifiedRenderService:
else:
config = self.plan.config or {}
bgm_config = config.get("bgm", {}) or {}
if not isinstance(bgm_config, dict):
bgm_config = {}
audio_tracks_config = config.get("audio_tracks") or {}
noise_reduction_config = config.get("audio_noise_reduction")
ctx = RenderContext(
@@ -482,11 +558,7 @@ class UnifiedRenderService:
"""
config = self.plan.config or {}
title_cfg = config.get("title", {}) or {}
if not isinstance(title_cfg, dict):
title_cfg = {}
subtitle_cfg = config.get("subtitle", {}) or {}
if not isinstance(subtitle_cfg, dict):
subtitle_cfg = {}
title_enabled = title_cfg.get("enabled", True)
subtitle_enabled = subtitle_cfg.get("enabled", True)
@@ -521,14 +593,71 @@ class UnifiedRenderService:
timeline.segment_count,
video_duration,
)
# ASR 路径也需要叠加标题(标题作为独立 ASS Event 追加到 ASR 字幕之上)
# 用独立 try-except 包裹,避免叠加失败时覆盖已生成的 ASR 数据
if has_title:
try:
_overlay_title_on_ass(
ass_path,
title_text=title_text,
title_config=title_cfg,
video_width=self.output_width,
video_height=self.output_height,
video_duration=video_duration,
)
logger.info(
"ASR字幕叠加标题: plan_id=%s title=%s",
self.plan.id,
title_text[:30],
)
except Exception:
logger.warning(
"ASR字幕叠加标题失败,保留纯ASR字幕: plan_id=%s",
self.plan.id,
exc_info=True,
)
return ass_path
else:
# ASR 无结果,不生成字幕
# ASR 无结果:如果有标题,仍然生成标题 ASS
if has_title:
generate_ass_subtitles(
ass_path,
video_width=self.output_width,
video_height=self.output_height,
video_duration=video_duration,
title_text=title_text,
title_config=title_cfg,
)
logger.info(
"ASR无结果但生成标题: plan_id=%s title=%s",
self.plan.id,
title_text[:30],
)
return ass_path
logger.info("ASR自动字幕无识别结果,跳过字幕: plan_id=%s", self.plan.id)
return None
except Exception:
# ASR 失败降级:不生成字幕,不阻断主流程
logger.warning("ASR自动字幕生成失败,跳过字幕", exc_info=True)
# ASR 失败降级:如果有标题,仍然生成标题 ASS
if has_title:
try:
generate_ass_subtitles(
ass_path,
video_width=self.output_width,
video_height=self.output_height,
video_duration=video_duration,
title_text=title_text,
title_config=title_cfg,
)
logger.info(
"ASR失败但生成标题: plan_id=%s title=%s",
self.plan.id,
title_text[:30],
)
return ass_path
except Exception:
logger.warning("ASR失败后标题生成也失败", exc_info=True)
else:
logger.warning("ASR自动字幕生成失败,跳过字幕", exc_info=True)
return None
# 静态字幕模式(原有逻辑)
@@ -659,11 +788,7 @@ class UnifiedRenderService:
"""
config = self.plan.config or {}
tts_cfg = config.get("tts", {}) or {}
if not isinstance(tts_cfg, dict):
tts_cfg = {}
subtitle_cfg = config.get("subtitle", {}) or {}
if not isinstance(subtitle_cfg, dict):
subtitle_cfg = {}
use_subtitle_align = False # 是否使用字幕对齐模式
# 兼容前端顶层字段:voice_id / custom_text / voice_clone_profile_id
@@ -107,8 +107,6 @@ def _finalize_render_success(
# 从 plan.config.title.text 读取视频名称
plan_config = plan.config or {}
title_cfg = plan_config.get("title", {}) or {}
if not isinstance(title_cfg, dict):
title_cfg = {}
video_name = (title_cfg.get("text") or "").strip() or f"generated-{generation_task_id[:8]}.mp4"
if generation_task_id:
try:
+98 -58
View File
@@ -13,6 +13,7 @@
from __future__ import annotations
import json
import logging
import os
import tempfile
@@ -983,9 +984,9 @@ def _load_template_plan_config(template_id: str) -> dict:
# 从独立字段组装成 plan.config 格式
plan_config: dict[str, Any] = {}
title_cfg = template.title_config if isinstance(template.title_config, dict) else {}
subtitle_cfg = template.subtitle_config if isinstance(template.subtitle_config, dict) else {}
bgm_cfg = template.bgm_config if isinstance(template.bgm_config, dict) else {}
title_cfg = template.title_config or {}
subtitle_cfg = template.subtitle_config or {}
bgm_cfg = template.bgm_config or {}
if title_cfg:
plan_config["title"] = title_cfg
@@ -1123,14 +1124,15 @@ def _render_video(
resolution: str = "",
bgm_config: dict | None = None,
voice_ids: list[str] | None = None,
) -> tuple[Path, float]:
custom_title: str = "",
) -> tuple[Path, float, str]:
"""渲染视频(含配音混音)。
使用 RenderAdapter 统一渲染入口,复用 BGM/ASR/分辨率/缩略图逻辑。
使用 RenderAdapter 统一渲染入口,复用 BGM/ASR/分辨率/封面抽取逻辑。
Args:
Returns:
(output_path, render_duration)
(output_path, render_duration, cover_url)
"""
if not downloaded_videos:
raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
@@ -1159,8 +1161,6 @@ def _render_video(
if bgm_config:
plan_cfg = virtual_plan.config or {}
template_bgm = plan_cfg.get("bgm", {}) or {}
if not isinstance(template_bgm, dict):
template_bgm = {}
merged_bgm = merge_bgm_config(template_bgm, bgm_config)
plan_cfg["bgm"] = merged_bgm
virtual_plan.config = plan_cfg
@@ -1171,6 +1171,60 @@ def _render_video(
merged_bgm.get("source", ""),
)
# 用户自定义标题覆盖模板标题(用户指定优先级最高)
# 支持两种格式:
# 1. JSON 格式(新):{"text": "xxx", "font_size": 32, ...} — 包含标题文本和样式
# 2. 纯文本格式(旧):直接作为标题文本使用
if custom_title and custom_title.strip():
plan_cfg = dict(virtual_plan.config or {})
title_cfg = dict(plan_cfg.get("title", {}) or {})
ct_stripped = custom_title.strip()
parsed_config = None
if ct_stripped.startswith("{"):
try:
parsed_config = json.loads(ct_stripped)
except (json.JSONDecodeError, ValueError):
parsed_config = None
if parsed_config and isinstance(parsed_config, dict):
# JSON 格式:合并完整标题配置(文本 + 样式)
title_text = (parsed_config.get("text") or "").strip()
if title_text:
title_cfg["text"] = title_text
title_cfg["enabled"] = True
# 合并样式字段(用户指定 > 模板默认)
style_keys = ["font", "font_size", "font_color", "position", "bold", "stroke", "shadow", "font_preset"]
for key in style_keys:
if key in parsed_config and parsed_config[key] is not None:
# 前端字段名映射到 ASS 字段名
mapped_key = {
"font_size": "size",
"font_color": "color",
"font_preset": "font",
}.get(key, key)
title_cfg[mapped_key] = parsed_config[key]
logger.info(
"[task_id=%s] [渲染] 用户标题配置(JSON)已注入: text=%s, style_keys=%s",
task_id,
title_text[:50],
[k for k in style_keys if k in parsed_config],
)
else:
logger.warning(
"[task_id=%s] [渲染] JSON标题缺少text字段,跳过",
task_id,
)
else:
# 纯文本格式:仅设置文本
title_cfg["text"] = ct_stripped
title_cfg["enabled"] = True
logger.info(
"[task_id=%s] [渲染] 用户自定义标题已注入: title=%s",
task_id,
ct_stripped[:50],
)
plan_cfg["title"] = title_cfg
virtual_plan.config = plan_cfg
# 确保输出分辨率配置存在
# 优先级:用户指定 > 模板配置 > 默认 1280x720
# 预览模式:强制 854x480 + 低码率
@@ -1191,9 +1245,8 @@ def _render_video(
plan_cfg = dict(virtual_plan.config or {})
plan_cfg["voice_id"] = voice_ids[0]
subtitle_cfg = plan_cfg.get("subtitle", {}) or {}
if not isinstance(subtitle_cfg, dict):
subtitle_cfg = {}
subtitle_cfg["auto_generated"] = True
subtitle_cfg["enabled"] = True # 确保 ASR 字幕路径被触发,标题叠加也依赖此路径
plan_cfg["subtitle"] = subtitle_cfg
virtual_plan.config = plan_cfg
logger.info(
@@ -1248,8 +1301,9 @@ def _render_video(
# 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理
output_path = render_output_path
cover_url = getattr(render_result, "cover_url", "") or ""
return output_path, render_duration
return output_path, render_duration, cover_url
def _upload_and_record(
@@ -1260,6 +1314,7 @@ def _upload_and_record(
editing_mode,
user_id: str = "",
video_name: str = "",
thumbnail_url: str = "",
) -> tuple[str, float, int, int]:
"""上传 OSS、创建视频记录并查重。
@@ -1520,7 +1575,7 @@ def generate_video(self, task_id: str) -> dict:
else:
_resolved_resolution = task_info.get("resolution", "")
output_path, render_duration = _render_video(
output_path, render_duration, cover_url = _render_video(
task_id=task_id,
downloaded_videos=downloaded_videos,
voice_path=audio_path,
@@ -1533,12 +1588,42 @@ def generate_video(self, task_id: str) -> dict:
resolution=_resolved_resolution,
bgm_config=task_info.get("bgm_config", {}),
voice_ids=task_info.get("voice_ids", []),
custom_title=task_info.get("custom_title", ""),
)
if gen_task:
gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s")
_flush_logs(task_id, gen_task)
# 持久化封面 URL 到 GenerationTask(统一封面管道:从渲染后视频抽帧)
if cover_url:
_cover_session = None
try:
_cover_session = SessionLocal()
try:
from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
_cover_model = (
_cover_session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first()
)
if _cover_model:
_cover_model.cover_url = cover_url
_cover_session.commit()
logger.info(
"[task_id=%s] 封面URL已持久化: %s",
task_id,
cover_url[:80],
)
finally:
if _cover_session:
_cover_session.close()
except Exception as cover_err:
logger.warning(
"[task_id=%s] 封面URL持久化失败(不影响主流程): %s",
task_id,
cover_err,
)
_update_task_progress(task_id, 80, "渲染完成")
# ── 4. 上传 OSS + 查重记录 ───────────────────────────────────────
@@ -1551,6 +1636,7 @@ def generate_video(self, task_id: str) -> dict:
editing_mode=editing_mode,
user_id=user_id,
video_name=task_info.get("video_title", ""),
thumbnail_url=cover_url,
)
if gen_task:
@@ -1564,52 +1650,6 @@ def generate_video(self, task_id: str) -> dict:
_update_task_progress(task_id, 95, "上传完成")
# ── 4.5 封面抽帧 ────────────────────────────────────────────────
# 预览视频上传完成后,提取封面帧写入 gen_task.cover_url
# 这样封面路由(generation_cover.py 步骤A)可以通过 generation_task_id 直接找到
try:
from packages.shared.mediakit_client import get_mediakit_client
mk_client = get_mediakit_client()
if mk_client.is_available:
_update_task_progress(task_id, 96, "提取封面帧")
snapshots = mk_client.extract_frames(
video_url=file_url,
strategy="SpecifiedFrames",
max_frames=1,
)
if snapshots and len(snapshots) > 0:
cover_frame_url = snapshots[0].get("image_url", "")
if cover_frame_url and gen_task:
# 通过独立 session 持久化 cover_url
_cover_session = SessionLocal()
try:
from packages.adapters.sqlalchemy_impl.models import (
GenerationTaskModel,
)
_cover_model = (
_cover_session.query(GenerationTaskModel)
.filter(GenerationTaskModel.id == task_id)
.first()
)
if _cover_model:
_cover_model.cover_url = cover_frame_url
_cover_session.commit()
logger.info(
"[task_id=%s] 封面帧提取成功: %s",
task_id,
cover_frame_url[:80],
)
finally:
_cover_session.close()
else:
logger.warning("[task_id=%s] 封面帧提取返回空结果", task_id)
else:
logger.warning("[task_id=%s] MediaKit 未配置,跳过封面帧提取", task_id)
except Exception:
logger.warning("[task_id=%s] 封面帧提取失败(不影响主流程)", task_id, exc_info=True)
# ── 5. 标记完成 ──────────────────────────────────────────────────
_update_task_status(task_id, "mark_completed", result_count=video_count)
+3 -24
View File
@@ -247,27 +247,6 @@ def build_ass_content(
title_config = title_config or {}
subtitle_config = subtitle_config or {}
# ── 兼容前端简化格式:stroke/shadow 为 boolean 时,转换为标准 dict ──
# 前端 TitleSettings 发送 stroke=true/false, shadow=true/false
# 后端 build_ass_style 期望 stroke={enabled, color, width}, shadow={enabled, blur, offset_x, offset_y}
if title_config:
_stroke_val = title_config.get("stroke")
if isinstance(_stroke_val, bool):
title_config["stroke"] = {
"enabled": _stroke_val,
"color": "#000000",
"width": 2,
} if _stroke_val else {"enabled": False}
_shadow_val = title_config.get("shadow")
if isinstance(_shadow_val, bool):
title_config["shadow"] = {
"enabled": _shadow_val,
"color": "#000000",
"blur": 4,
"offset_x": 2,
"offset_y": 2,
} if _shadow_val else {"enabled": False}
title_enabled = title_config.get("enabled", True) and bool(title_text.strip())
subtitle_enabled = subtitle_config.get("enabled", True) and bool(subtitle_text.strip())
@@ -283,7 +262,7 @@ def build_ass_content(
title_stroke = title_config.get("stroke", {}) or {}
title_shadow = title_config.get("shadow", {}) or {}
stroke_color = hex_to_ass_color(title_stroke.get("color", "#000000"))
stroke_width = float(title_stroke.get("width", 2)) if title_stroke.get("enabled", False) else 0.0
stroke_width = float(title_stroke.get("width", 1)) if title_stroke.get("enabled", False) else 0.0
shadow_blur = float(title_shadow.get("blur", 4)) if title_shadow.get("enabled", False) else 0.0
shadow_offset = (
title_shadow.get("offset_x", 2) if title_shadow.get("enabled", False) else 0,
@@ -296,7 +275,7 @@ def build_ass_content(
build_ass_style(
"TitleStyle",
font_name=title_config.get("font", "思源黑体"),
font_size=min(int(title_config.get("size", 36)), 36),
font_size=int(title_config.get("size", 48)),
primary_color=title_color,
outline_color=stroke_color,
outline_width=stroke_width,
@@ -313,7 +292,7 @@ def build_ass_content(
# 根据视频宽度和字号自动换行标题,防止超出画面
# 先 escape 特殊字符,再插入换行符 \N,避免顺序颠倒导致 \N 被转义
title_font_size = min(int(title_config.get("size", 36)), 36)
title_font_size = int(title_config.get("size", 48))
safe_title_text_raw = escape_ass_text(title_text)
safe_title_text = _wrap_title_text(safe_title_text_raw, video_width, title_font_size)
+1 -1
View File
@@ -95,7 +95,7 @@ class ChromaKeyConfig:
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> ChromaKeyConfig:
"""从字典解析配置,参数越界自动钳制."""
if not isinstance(data, dict) or not data.get("enabled", False):
if not data or not data.get("enabled", False):
return cls(enabled=False)
key_color = str(data.get("key_color", DEFAULT_KEY_COLOR)).strip()
+1 -1
View File
@@ -196,7 +196,7 @@ class ColorGradeConfig:
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> "ColorGradeConfig":
"""从字典解析配置."""
if not isinstance(data, dict) or not data.get("enabled", False):
if not data or not data.get("enabled", False):
return cls(enabled=False)
preset = data.get("preset", "")
+1 -1
View File
@@ -76,7 +76,7 @@ class IntroOutroConfig:
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> "IntroOutroConfig":
"""从字典构造."""
if not isinstance(data, dict):
if not data:
return cls()
enabled = data.get("enabled", False)
+1 -1
View File
@@ -81,7 +81,7 @@ class NoiseReductionConfig:
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> NoiseReductionConfig:
"""从字典解析配置,参数越界自动钳制."""
if not isinstance(data, dict) or not data.get("enabled", False):
if not data or not data.get("enabled", False):
return cls(enabled=False)
level_str = str(data.get("level", "medium")).lower()
+1 -1
View File
@@ -136,7 +136,7 @@ class PiPConfig:
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> "PiPConfig":
"""从字典解析配置."""
if not isinstance(data, dict) or not data.get("enabled", False):
if not data or not data.get("enabled", False):
return cls(enabled=False)
layers_data = data.get("layers", [])
+1 -1
View File
@@ -86,7 +86,7 @@ class WatermarkConfig:
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> WatermarkConfig | None:
"""从字典构造,空配置返回 None(不加水印)."""
if not isinstance(data, dict):
if not data:
return None
enabled = data.get("enabled", False)
@@ -158,7 +158,7 @@ class TestRenderVideoVoiceInjection:
from packages.domain import EditingMode
output_path, render_duration = _render_video(
output_path, render_duration, _cover_url = _render_video(
task_id="test_task_123",
downloaded_videos=[Path("/tmp/video1.mp4")],
voice_path=None,
-130
View File
@@ -1,130 +0,0 @@
"""
测试:模板 config 字段存储了非 dict 值(如 True / False / str)时,
渲染链路不会崩溃('bool' object has no attribute 'get')。
覆盖两个关键文件:
1. generation.py — _load_template_plan_config 旧系统路径
2. unified_render_service.py — _maybe_generate_ass
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
# Add worker app to path
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
class TestLoadTemplatePlanConfigBoolDefense:
"""_load_template_plan_config 旧系统路径对非 dict 值的防护。"""
def _call_old_path(self, title_cfg, subtitle_cfg, bgm_cfg):
"""通过 mock 新模板系统返回 None,强制走旧模板系统 fallback 路径。"""
from worker_app.tasks.generation import _load_template_plan_config
mock_old_template = MagicMock()
mock_old_template.title_config = title_cfg
mock_old_template.subtitle_config = subtitle_cfg
mock_old_template.bgm_config = bgm_cfg
mock_session = MagicMock()
# 旧系统 query 返回 mock template
mock_session.query.return_value.filter.return_value.first.return_value = mock_old_template
# Mock 新模板系统 repo.get() 返回 None(强制走 fallback
mock_repo_cls = MagicMock()
mock_repo_cls.return_value.get.return_value = None
with (
patch("worker_app.tasks.generation.SessionLocal", return_value=mock_session),
patch("packages.adapters.sqlalchemy_impl.SQLAlchemyEditTemplateRepository", mock_repo_cls),
patch("packages.adapters.sqlalchemy_impl.SQLAlchemyTemplateClipConfigRepository", MagicMock()),
):
return _load_template_plan_config("fake-id")
def test_bool_values_return_empty(self):
"""title_config=True / subtitle_config=False / bgm_config='str' → 全部过滤掉"""
result = self._call_old_path(True, False, "not_a_dict")
assert isinstance(result, dict)
assert "title" not in result
assert "subtitle" not in result
assert "bgm" not in result
def test_valid_dict_passes_through(self):
"""正常 dict 正常传递"""
result = self._call_old_path(
{"text": "标题", "enabled": True},
{"text": "副标题"},
{"enabled": True, "source": "test.mp3"},
)
assert result["title"] == {"text": "标题", "enabled": True}
assert result["subtitle"] == {"text": "副标题"}
assert result["bgm"] == {"enabled": True, "source": "test.mp3"}
def test_none_returns_empty(self):
"""None → 空 dict"""
result = self._call_old_path(None, None, None)
assert result == {}
def test_mixed_valid_and_invalid(self):
"""部分有效、部分无效时只保留有效的"""
result = self._call_old_path({"text": "OK"}, True, None)
assert "title" in result
assert "subtitle" not in result
assert "bgm" not in result
def test_int_and_list_also_filtered(self):
"""int / list 类型也被过滤"""
result = self._call_old_path(42, [1, 2, 3], 0)
assert result == {}
class TestUnifiedRenderBoolConfigDefense:
"""_maybe_generate_ass 对 plan.config 中非 dict title/subtitle 的防护。"""
def _make_service(self, config):
from video_processing.unified_render_service import UnifiedRenderService
service = UnifiedRenderService.__new__(UnifiedRenderService)
mock_plan = MagicMock()
mock_plan.config = config
service.plan = mock_plan
service.task_id = "test-task"
return service
def test_bool_title_does_not_crash(self):
"""config['title']=True → 不崩溃,返回 None"""
service = self._make_service({"title": True, "subtitle": {}})
result = service._maybe_generate_ass(video_duration=10.0)
assert result is None
def test_bool_subtitle_does_not_crash(self):
"""config['subtitle']=False → 不崩溃,返回 None"""
service = self._make_service({"title": {}, "subtitle": False})
result = service._maybe_generate_ass(video_duration=10.0)
assert result is None
def test_str_title_does_not_crash(self):
"""config['title']='plain string' → 不崩溃"""
service = self._make_service({"title": "plain string", "subtitle": {}})
result = service._maybe_generate_ass(video_duration=10.0)
assert result is None
def test_none_config_does_not_crash(self):
"""config=None → 不崩溃"""
service = self._make_service(None)
result = service._maybe_generate_ass(video_duration=10.0)
assert result is None
def test_int_title_does_not_crash(self):
"""config['title']=42 → 不崩溃"""
service = self._make_service({"title": 42, "subtitle": 0})
result = service._maybe_generate_ass(video_duration=10.0)
assert result is None
-142
View File
@@ -1,142 +0,0 @@
# -*- coding: utf-8 -*-
"""测试 Step6 封面生成 400 修复:
1. Worker 渲染完成后提取封面帧写入 cover_url
2. API 创建预览任务时自动关联 source_edit_plan_id
"""
import pytest
from packages.application.generation_tasks import CreateGenerationTaskCommand
from packages.domain.generation_task import GenerationTask, GenerationTaskStatus
def _make_task(**kwargs):
return GenerationTask(
id=kwargs.get("id", "task-001"),
project_id=kwargs.get("project_id", ""),
asset_library_id=kwargs.get("asset_library_id", ""),
template_id=kwargs.get("template_id", "tpl-001"),
created_by_user_id=kwargs.get("user_id", "user-001"),
asset_ids=kwargs.get("asset_ids", ["asset-1"]),
status=kwargs.get("status", GenerationTaskStatus.RUNNING),
source_edit_plan_id=kwargs.get("source_edit_plan_id", ""),
cover_url=kwargs.get("cover_url", ""),
is_preview=kwargs.get("is_preview", True),
)
class TestWorkerCoverFrameExtraction:
"""Worker 端:渲染完成后提取封面帧写入 cover_url"""
def test_cover_url_set_after_frame_extraction(self):
"""extract_frames 返回结果时,cover_url 应被设置"""
task = _make_task()
assert task.cover_url == ""
mock_frame_url = "https://oss.example.com/frames/frame_001.jpg"
task.cover_url = mock_frame_url
assert task.cover_url == mock_frame_url
def test_cover_url_empty_when_no_frames(self):
"""extract_frames 返回空时,cover_url 应保持为空"""
task = _make_task()
assert task.cover_url == ""
def test_cover_url_preserved_on_extraction_failure(self):
"""extract_frames 异常时,cover_url 保持原值"""
task = _make_task(cover_url="")
try:
raise RuntimeError("MediaKit timeout")
except RuntimeError:
pass
assert task.cover_url == ""
def test_cover_url_first_frame_used(self):
"""多帧结果应使用第一帧"""
frames = [
{"image_url": "https://oss.example.com/frame_001.jpg", "timestamp": 0.0},
{"image_url": "https://oss.example.com/frame_002.jpg", "timestamp": 1.5},
]
task = _make_task()
task.cover_url = frames[0]["image_url"]
assert task.cover_url == "https://oss.example.com/frame_001.jpg"
def test_cover_url_not_set_when_empty_image_url(self):
"""帧的 image_url 为空时不应设置 cover_url"""
frames = [{"image_url": "", "timestamp": 0.0}]
task = _make_task()
frame_url = frames[0].get("image_url", "")
if frame_url:
task.cover_url = frame_url
assert task.cover_url == ""
class TestPreviewSourceEditPlanId:
"""API 端:预览任务自动关联 source_edit_plan_id"""
def test_source_edit_plan_id_set_when_provided(self):
"""前端传入 source_edit_plan_id 时应直接使用"""
cmd = CreateGenerationTaskCommand(
project_id="",
asset_library_id="",
strategy_id="one-take",
template_id="tpl-001",
asset_ids=["asset-1"],
created_by_user_id="user-001",
source_edit_plan_id="plan-xyz",
)
assert cmd.source_edit_plan_id == "plan-xyz"
def test_source_edit_plan_id_empty_when_not_provided(self):
"""前端未传入时 source_edit_plan_id 默认为空"""
cmd = CreateGenerationTaskCommand(
project_id="",
asset_library_id="",
strategy_id="one-take",
template_id="tpl-001",
asset_ids=["asset-1"],
created_by_user_id="user-001",
)
assert cmd.source_edit_plan_id == ""
def test_task_preserves_source_edit_plan_id(self):
"""GenerationTask 应保持 source_edit_plan_id"""
task = _make_task(source_edit_plan_id="plan-abc")
assert task.source_edit_plan_id == "plan-abc"
class TestCoverRouteStepB:
"""封面路由步骤 B:通过 source_edit_plan_id 查找"""
def test_step_b_finds_preview_task_by_source_plan(self):
"""步骤 B 应找到 source_edit_plan_id 匹配的已完成预览任务"""
task = _make_task(
source_edit_plan_id="plan-abc",
cover_url="https://oss.example.com/cover.jpg",
status=GenerationTaskStatus.COMPLETED,
)
is_valid = (
task.source_edit_plan_id == "plan-abc"
and task.status == GenerationTaskStatus.COMPLETED
and bool(task.cover_url)
)
assert is_valid is True
def test_step_b_skips_non_completed_tasks(self):
"""步骤 B 应跳过非 completed 状态的任务"""
task = _make_task(
source_edit_plan_id="plan-abc",
cover_url="https://oss.example.com/cover.jpg",
status=GenerationTaskStatus.FAILED,
)
is_valid = task.status == GenerationTaskStatus.COMPLETED and bool(task.cover_url)
assert is_valid is False
def test_step_b_skips_tasks_without_cover_url(self):
"""步骤 B 应跳过没有 cover_url 的任务"""
task = _make_task(
source_edit_plan_id="plan-abc",
cover_url="",
status=GenerationTaskStatus.COMPLETED,
)
is_valid = task.status == GenerationTaskStatus.COMPLETED and bool(task.cover_url)
assert is_valid is False
@@ -127,11 +127,6 @@ class TestExtractFirstFrame(unittest.TestCase):
extract_first_frame(video.name)
import pytest
pytest.skip("RenderAdapterResult.cover_url 已被 cover_candidates 替代,测试待更新", allow_module_level=True)
class TestRenderAdapterCoverUrl(unittest.TestCase):
"""RenderAdapterResult.cover_url 字段测试."""
-4
View File
@@ -7,10 +7,6 @@
4. ASR 失败但有标题时,降级生成标题 ASS
"""
import pytest
pytest.skip("_overlay_title_on_ass 函数已被移除,测试待更新", allow_module_level=True)
from pathlib import Path
from unittest.mock import MagicMock, patch
@@ -1,291 +0,0 @@
# -*- coding: utf-8 -*-
"""测试预览任务自动关联 edit_plangeneration_preview.py 增量覆盖率补充)。
覆盖 generation_preview.py 中的 edit_plan 自动关联逻辑:
- 前端未传 source_edit_plan_id 时,通过 template_id + user_id 自动查找
- 找到匹配 plan 后设置 task.source_edit_plan_id 并持久化
- 查找失败时不影响主流程
- 前端已传 source_edit_plan_id 时跳过自动关联
"""
from __future__ import annotations
import os
import sys
from dataclasses import dataclass, field
from typing import Any, Optional
from unittest.mock import MagicMock, patch
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
from packages.domain.generation_task import GenerationTask, GenerationTaskStatus
# ── Stub Repository ──────────────────────────────────────────────────────────
class StubGenerationTaskRepository:
"""内存中模拟 GenerationTask 仓储"""
def __init__(self) -> None:
self._store: dict[str, Any] = {}
def create(self, task: Any) -> Any:
self._store[task.id] = task
return task
def get(self, task_id: str) -> Optional[Any]:
return self._store.get(task_id)
def update(self, task: Any) -> Any:
if task.id not in self._store:
raise ValueError(f"GenerationTask {task.id} not found")
self._store[task.id] = task
return task
def count_pending_by_user(self, user_id: str) -> int:
return 0
def count_pending_total(self) -> int:
return 0
# ── Fake Edit Plan ──────────────────────────────────────────────────────────
@dataclass
class FakeEditPlan:
id: str = "plan-001"
created_by_user_id: str = "user-001"
template_id: str = "tpl-001"
class FakeEditPlanRepository:
def __init__(self, plans: list[FakeEditPlan] | None = None):
self._plans = plans or []
def list_by_template(self, template_id: str, limit: int = 20) -> list:
return [p for p in self._plans if p.template_id == template_id]
# ── Auth Fakes ──────────────────────────────────────────────────────────────
@dataclass
class FakeUser:
id: str = "user-001"
email: str = "test@example.com"
@dataclass
class FakeAuthenticatedUser:
user: FakeUser = field(default_factory=FakeUser)
session_id: str | None = None
token_type: str | None = None
# ── Fixtures ─────────────────────────────────────────────────────────────────
@pytest.fixture
def gen_task_repo() -> StubGenerationTaskRepository:
return StubGenerationTaskRepository()
@pytest.fixture
def mock_db() -> MagicMock:
return MagicMock()
@pytest.fixture
def app(gen_task_repo: StubGenerationTaskRepository, mock_db: MagicMock) -> FastAPI:
"""构建测试 FastAPI 应用,注入 Stub"""
from app.api.routes.generation_preview import router
from app.auth import get_current_user
from app.dependencies import (
get_asset_repository,
get_db_session,
get_generated_video_repository,
get_generation_task_repository,
)
test_app = FastAPI()
test_app.include_router(router, prefix="/api/v1/generation")
test_app.dependency_overrides[get_current_user] = lambda: FakeAuthenticatedUser()
test_app.dependency_overrides[get_generation_task_repository] = lambda: gen_task_repo
test_app.dependency_overrides[get_db_session] = lambda: mock_db
test_app.dependency_overrides[get_asset_repository] = lambda: MagicMock()
test_app.dependency_overrides[get_generated_video_repository] = lambda: MagicMock()
yield test_app
test_app.dependency_overrides.clear()
@pytest.fixture
def client(app: FastAPI) -> TestClient:
return TestClient(app)
def _make_request_body(**kwargs: Any) -> dict:
defaults = dict(
template_id="tpl-001",
asset_ids=["asset-1"],
title_ids=[],
voice_ids=[],
preview_count=1,
video_ratio="",
source_edit_plan_id="",
video_title="",
bgm_config={},
)
defaults.update(kwargs)
return defaults
# ── Tests ────────────────────────────────────────────────────────────────────
class TestPreviewEditPlanAutoAssociation:
"""预览任务创建后自动关联 edit_plan"""
@patch(
"app.api.routes.generation_preview._resolve_strategy_id_from_template",
return_value="one_take",
)
@patch(
"app.api.routes.generation_preview._infer_video_ratio_from_template",
return_value="9:16",
)
@patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True)
def test_auto_associate_when_source_plan_empty(
self,
mock_enqueue,
mock_ratio,
mock_strategy,
client: TestClient,
gen_task_repo: StubGenerationTaskRepository,
):
"""前端未传 source_edit_plan_id 时,应通过 template_id+user_id 自动查找并关联"""
fake_plan = FakeEditPlan(id="plan-auto-001", created_by_user_id="user-001", template_id="tpl-001")
fake_plan_repo = FakeEditPlanRepository(plans=[fake_plan])
with patch(
"packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository",
return_value=fake_plan_repo,
):
resp = client.post(
"/api/v1/generation/preview",
json=_make_request_body(source_edit_plan_id=""),
)
assert resp.status_code == 201
# 找到 store 中的 task 并验证 source_edit_plan_id 被设置
tasks = list(gen_task_repo._store.values())
assert len(tasks) == 1
task = tasks[0]
assert task.source_edit_plan_id == "plan-auto-001"
@patch(
"app.api.routes.generation_preview._resolve_strategy_id_from_template",
return_value="one_take",
)
@patch(
"app.api.routes.generation_preview._infer_video_ratio_from_template",
return_value="9:16",
)
@patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True)
def test_skip_associate_when_source_plan_provided(
self,
mock_enqueue,
mock_ratio,
mock_strategy,
client: TestClient,
gen_task_repo: StubGenerationTaskRepository,
):
"""前端已传 source_edit_plan_id 时,不应触发自动关联"""
resp = client.post(
"/api/v1/generation/preview",
json=_make_request_body(source_edit_plan_id="plan-explicit-001"),
)
assert resp.status_code == 201
tasks = list(gen_task_repo._store.values())
assert len(tasks) == 1
assert tasks[0].source_edit_plan_id == "plan-explicit-001"
@patch(
"app.api.routes.generation_preview._resolve_strategy_id_from_template",
return_value="one_take",
)
@patch(
"app.api.routes.generation_preview._infer_video_ratio_from_template",
return_value="9:16",
)
@patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True)
def test_association_failure_does_not_break_main_flow(
self,
mock_enqueue,
mock_ratio,
mock_strategy,
client: TestClient,
gen_task_repo: StubGenerationTaskRepository,
):
"""edit_plan 查找异常时不影响任务创建和入队"""
with patch(
"packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository",
side_effect=RuntimeError("DB connection lost"),
):
resp = client.post(
"/api/v1/generation/preview",
json=_make_request_body(source_edit_plan_id=""),
)
# 任务仍然创建成功
assert resp.status_code == 201
tasks = list(gen_task_repo._store.values())
assert len(tasks) == 1
# source_edit_plan_id 保持为空(关联失败)
assert tasks[0].source_edit_plan_id == ""
@patch(
"app.api.routes.generation_preview._resolve_strategy_id_from_template",
return_value="one_take",
)
@patch(
"app.api.routes.generation_preview._infer_video_ratio_from_template",
return_value="9:16",
)
@patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True)
def test_auto_associate_skips_when_no_matching_user(
self,
mock_enqueue,
mock_ratio,
mock_strategy,
client: TestClient,
gen_task_repo: StubGenerationTaskRepository,
):
"""模板下有 plan 但 created_by_user_id 不匹配时,不关联"""
fake_plan = FakeEditPlan(id="plan-other-user", created_by_user_id="user-999", template_id="tpl-001")
fake_plan_repo = FakeEditPlanRepository(plans=[fake_plan])
with patch(
"packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository",
return_value=fake_plan_repo,
):
resp = client.post(
"/api/v1/generation/preview",
json=_make_request_body(source_edit_plan_id=""),
)
assert resp.status_code == 201
tasks = list(gen_task_repo._store.values())
assert len(tasks) == 1
# user 不匹配,source_edit_plan_id 保持为空
assert tasks[0].source_edit_plan_id == ""
-2
View File
@@ -327,7 +327,6 @@ class TestRenderPlan:
@patch("video_processing.render_adapter.upload_to_oss")
@patch("video_processing.render_adapter.UnifiedRenderService")
@patch("video_processing.render_adapter.download_asset")
@pytest.mark.skip(reason="thumbnail_url mock 与当前代码不匹配,待更新")
def test_thumbnail_generated_on_success(self, mock_download, mock_render_cls, mock_upload, tmp_path):
"""渲染成功后生成缩略图,thumbnail_url 正确返回。"""
@@ -379,7 +378,6 @@ class TestRenderPlan:
@patch("video_processing.render_adapter.upload_to_oss")
@patch("video_processing.render_adapter.UnifiedRenderService")
@patch("video_processing.render_adapter.download_asset")
@pytest.mark.skip(reason="thumbnail_url mock 与当前代码不匹配,待更新")
def test_thumbnail_failure_does_not_block(self, mock_download, mock_render_cls, mock_upload, tmp_path):
"""缩略图生成失败不影响主流程,thumbnail_url 为空串。"""
-215
View File
@@ -1,215 +0,0 @@
"""标题渲染前后端一致性测试。
验证 build_ass_content 生成的 ASS 样式参数与前端 drawTitleOnCanvas.ts 一致:
- 字号上限 36px
- 描边宽度 2px
- 阴影 blur=4, offset=2
- boolean stroke/shadow 自动转换
"""
import sys
from pathlib import Path
# Ensure packages is importable
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "packages"))
from domain.ass_subtitle_builder import build_ass_content, build_ass_style
class TestFontSizeCap:
"""字号上限应与前端 Math.min(settings.size, 36) 一致。"""
def test_default_font_size_is_36(self):
"""无 size 字段时,默认字号应为 36。"""
config = {"text": "test"}
content = build_ass_content(
video_width=1080,
video_height=1920,
video_duration=10.0,
title_text="测试标题",
title_config=config,
)
assert ",36," in content, f"默认字号应为36,实际内容: {content}"
def test_size_32_preserved(self):
"""size=32 应原样使用。"""
config = {"size": 32}
content = build_ass_content(
video_width=1080,
video_height=1920,
video_duration=10.0,
title_text="测试标题",
title_config=config,
)
assert ",32," in content
def test_size_60_capped_at_36(self):
"""size=60 应被 cap 到 36。"""
config = {"size": 60}
content = build_ass_content(
video_width=1080,
video_height=1920,
video_duration=10.0,
title_text="测试标题",
title_config=config,
)
# 解析 Style 行的 Fontsize 字段(第3个字段,索引2)
style_line = [line for line in content.splitlines() if line.startswith("Style: TitleStyle")][0]
fields = [f.strip() for f in style_line.split(",")]
font_size = int(fields[2])
assert font_size == 36, f"字号60应被cap到36, 实际={font_size}"
def test_size_24_preserved(self):
"""size=24 应原样使用(小于36,不cap)。"""
config = {"size": 24}
content = build_ass_content(
video_width=1080,
video_height=1920,
video_duration=10.0,
title_text="测试标题",
title_config=config,
)
assert ",24," in content
class TestBooleanStrokeNormalization:
"""前端 stroke=true/false 应自动转换为标准 dict。"""
def test_stroke_true_enables_outline(self):
"""stroke=true 应生成 outline_width=2 的样式。"""
config = {"stroke": True}
content = build_ass_content(
video_width=1080,
video_height=1920,
video_duration=10.0,
title_text="测试标题",
title_config=config,
)
# 解析 Style 行的 Outline 字段(第17个字段,索引16)
style_line = [line for line in content.splitlines() if line.startswith("Style: TitleStyle")][0]
fields = [f.strip() for f in style_line.split(",")]
outline_width = float(fields[16])
assert outline_width == 2.0, f"stroke=true 应产生 outline_width=2, 实际={outline_width}"
def test_stroke_false_no_outline(self):
"""stroke=false 应生成 outline_width=0 的样式。"""
config = {"stroke": False}
content = build_ass_content(
video_width=1080,
video_height=1920,
video_duration=10.0,
title_text="测试标题",
title_config=config,
)
style_line = [line for line in content.splitlines() if line.startswith("Style: TitleStyle")][0]
fields = [f.strip() for f in style_line.split(",")]
outline_width = float(fields[16])
assert outline_width == 0.0, f"stroke=false 应产生 outline_width=0, 实际={outline_width}"
def test_stroke_dict_still_works(self):
"""stroke={enabled:true, width:3} 仍应正常工作。"""
config = {"stroke": {"enabled": True, "width": 3, "color": "#FF0000"}}
content = build_ass_content(
video_width=1080,
video_height=1920,
video_duration=10.0,
title_text="测试标题",
title_config=config,
)
style_line = [line for line in content.splitlines() if line.startswith("Style: TitleStyle")][0]
fields = [f.strip() for f in style_line.split(",")]
outline_width = float(fields[16])
assert outline_width == 3.0, f"自定义stroke width=3 应保留, 实际={outline_width}"
class TestBooleanShadowNormalization:
"""前端 shadow=true/false 应自动转换为标准 dict。"""
def test_shadow_true_enables_shadow(self):
"""shadow=true 应生成 shadow_depth=2 的样式。"""
config = {"shadow": True}
content = build_ass_content(
video_width=1080,
video_height=1920,
video_duration=10.0,
title_text="测试标题",
title_config=config,
)
style_line = [line for line in content.splitlines() if line.startswith("Style: TitleStyle")][0]
fields = [f.strip() for f in style_line.split(",")]
# Shadow 字段是第18个(索引17
shadow_depth = int(fields[17])
assert shadow_depth == 2, f"shadow=true 应产生 shadow_depth=2, 实际={shadow_depth}"
def test_shadow_false_no_shadow(self):
"""shadow=false 应生成 shadow_depth=0 的样式。"""
config = {"shadow": False}
content = build_ass_content(
video_width=1080,
video_height=1920,
video_duration=10.0,
title_text="测试标题",
title_config=config,
)
style_line = [line for line in content.splitlines() if line.startswith("Style: TitleStyle")][0]
fields = [f.strip() for f in style_line.split(",")]
shadow_depth = int(fields[17])
assert shadow_depth == 0, f"shadow=false 应产生 shadow_depth=0, 实际={shadow_depth}"
def test_shadow_dict_still_works(self):
"""shadow={enabled:true, blur:8} 仍应正常工作。"""
config = {"shadow": {"enabled": True, "blur": 8, "offset_x": 3, "offset_y": 3}}
content = build_ass_content(
video_width=1080,
video_height=1920,
video_duration=10.0,
title_text="测试标题",
title_config=config,
)
style_line = [line for line in content.splitlines() if line.startswith("Style: TitleStyle")][0]
fields = [f.strip() for f in style_line.split(",")]
shadow_depth = int(fields[17])
assert shadow_depth == 3, f"自定义shadow offset_y=3 应保留, 实际={shadow_depth}"
class TestFullStyleConsistency:
"""完整样式参数一致性测试。"""
def test_frontend_default_style_matches_backend(self):
"""前端默认样式参数应在后端产生一致的 ASS 输出。
前端默认:font_size=24(或用户设置), bold=false, stroke=true, shadow=true, color=#FFFFFF
"""
config = {
"text": "标题文本",
"font": "思源黑体",
"size": 28,
"color": "#FFFFFF",
"bold": True,
"italic": False,
"stroke": True,
"shadow": True,
"position": "top",
}
content = build_ass_content(
video_width=1080,
video_height=1920,
video_duration=10.0,
title_text="标题文本",
title_config=config,
)
style_line = [line for line in content.splitlines() if line.startswith("Style: TitleStyle")][0]
fields = [f.strip() for f in style_line.split(",")]
# Fontname
assert fields[1] == "思源黑体"
# Fontsize = 28 (小于36,不cap)
assert fields[2] == "28"
# Bold = -1 (True)
assert fields[7] == "-1"
# Outline width = 2 (前端默认 stroke width)
assert float(fields[16]) == 2.0
# Shadow depth = 2 (offset_y)
assert int(fields[17]) == 2
# Alignment = 8 (top)
assert int(fields[18]) == 8