Compare commits

..

2 Commits

Author SHA1 Message Date
xiaoxia 157ebd4ef2 fix: 修复PreviewVideoPanel JSX结构-将视频+Canvas叠加层移出!showTitlePreview块
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 1m0s
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 - Migration (alembic) (pull_request) Successful in 1m36s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m36s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m38s
AI Code Review / AI Code Review (pull_request) Successful in 1m39s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 50s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 51s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 1m55s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m59s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 4m42s
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 cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 39m21s
CI/CD Pipeline / CI Gate (pull_request) Failing after 14s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Has been cancelled
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 41s
- 预览成功+Canvas标题叠加层提升到顶层,仅在 hasPreview 条件下渲染
- 空状态/生成中/生成失败/预览信息添加 !showTitlePreview 守卫
- 修复 Canvas overlay 在 showTitlePreview=true 时永远不渲染的死代码问题
2026-08-16 00:46:25 +08:00
xiaoxia 3717787cc8 fix: 修复PR#1382引入的TypeScript编译错误
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 1m36s
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
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m42s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m44s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m59s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 2m7s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 42s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 1m31s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m13s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 4m9s
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 Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
AI Code Review / AI Code Review (pull_request) Successful in 4m45s
CI/CD Pipeline / PR Build Web Image (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
- PreviewVideoPanel: 删除重复的 showTitlePreview 声明,添加 TitlePreviewCanvas import
- Step4TitleSettings: 删除未使用的 TitlePreviewCanvas import
- GeneratePage: 补充 useQuery / getAssetsByKind / AssetItem import,修复回调参数类型标注

修复 CI Run#30879 阻断的 TS 编译错误
2026-08-16 00:40:29 +08:00
36 changed files with 852 additions and 1352 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(
-82
View File
@@ -1,82 +0,0 @@
/**
* 主动 Token 刷新模块
*
* 在 access_token 过期前主动刷新,避免 API 请求触发 401。
* JWT payload 是 base64 编码的 JSON,无需第三方库即可解码。
*/
import { useAuthStore } from "@/store/authStore"
import { refreshAccessToken } from "./login"
let refreshTimer: ReturnType<typeof setTimeout> | null = null
/** 提前刷新的缓冲时间(秒) */
const REFRESH_BUFFER_SECONDS = 60
/**
* 解码 JWT payload(不验签,仅读取 exp 字段)
*/
function decodeJwtPayload(token: string): { exp?: number } | null {
try {
const parts = token.split(".")
if (parts.length !== 3) return null
// JWT 使用 base64url 编码,需要转换为标准 base64
const payload = parts[1].replace(/-/g, "+").replace(/_/g, "/")
const padded = payload + "=".repeat((4 - (payload.length % 4)) % 4)
const decoded = atob(padded)
return JSON.parse(decoded)
} catch {
return null
}
}
/**
* 取消已调度的主动刷新
*/
export function cancelProactiveRefresh(): void {
if (refreshTimer) {
clearTimeout(refreshTimer)
refreshTimer = null
}
}
/**
* 调度主动刷新:在 token 过期前 REFRESH_BUFFER_SECONDS 秒自动刷新
*/
export function scheduleProactiveRefresh(): void {
cancelProactiveRefresh()
const accessToken = localStorage.getItem("access_token")
const refreshTokenValue = useAuthStore.getState().refreshToken
if (!accessToken || !refreshTokenValue) return
const payload = decodeJwtPayload(accessToken)
if (!payload?.exp) return
const now = Math.floor(Date.now() / 1000)
const secondsUntilExpiry = payload.exp - now
// 如果 token 已经过期或即将在缓冲时间内过期,立即刷新
const delaySeconds = Math.max(secondsUntilExpiry - REFRESH_BUFFER_SECONDS, 0)
refreshTimer = setTimeout(async () => {
try {
const data = await refreshAccessToken(refreshTokenValue)
const newAccessToken = data.access_token
const newRefreshToken = data.refresh_token ?? refreshTokenValue
// 更新 Zustand store + localStorage
useAuthStore
.getState()
.setAuth(useAuthStore.getState().user!, newAccessToken, newRefreshToken)
// 递归调度下一次刷新
scheduleProactiveRefresh()
} catch {
// 刷新失败 → 清除认证状态,跳转登录页
cancelProactiveRefresh()
useAuthStore.getState().clearAuth()
window.location.href = "/login"
}
}, delaySeconds * 1000)
}
-5
View File
@@ -6,7 +6,6 @@ import axios, { AxiosError, InternalAxiosRequestConfig } from "axios"
import { message } from "antd"
import { useAuthStore } from "@/store/authStore"
import { refreshAccessToken } from "./auth"
import { scheduleProactiveRefresh, cancelProactiveRefresh } from "./auth/tokenRefresh"
// 创建 Axios 实例
const apiClient = axios.create({
@@ -110,9 +109,6 @@ apiClient.interceptors.response.use(
// 处理排队的请求
processQueue(null, newAccessToken)
// 重新调度主动刷新(基于新 token 的过期时间)
scheduleProactiveRefresh()
// 重试原始请求
if (originalRequest.headers) {
originalRequest.headers.Authorization = `Bearer ${newAccessToken}`
@@ -120,7 +116,6 @@ apiClient.interceptors.response.use(
return apiClient(originalRequest)
} catch (refreshError) {
// 刷新失败 → 登出
cancelProactiveRefresh()
processQueue(refreshError, null)
useAuthStore.getState().clearAuth()
window.location.href = "/"
-8
View File
@@ -4,7 +4,6 @@
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"
import { useNavigate } from "react-router-dom"
import * as authApi from "@/api/auth"
import { scheduleProactiveRefresh, cancelProactiveRefresh } from "@/api/auth/tokenRefresh"
import { useAuthStore } from "@/store/authStore"
// 登录 Hook
@@ -32,9 +31,6 @@ export const useLogin = () => {
const user = await authApi.getCurrentUser()
setAuth(user, data.access_token, refreshToken)
// 启动主动 token 刷新,避免后续请求触发 401
scheduleProactiveRefresh()
// 跳转到登录前页面或仪表盘(与 Login.tsx onFinish 保持一致)
const redirect = localStorage.getItem("login_redirect") || "/app/dashboard"
localStorage.removeItem("login_redirect")
@@ -77,9 +73,6 @@ export const useWechatCallback = () => {
const user = await authApi.getCurrentUser()
setAuth(user, result.access_token, result.refresh_token)
// 启动主动 token 刷新
scheduleProactiveRefresh()
return { ...result, user }
}
@@ -128,7 +121,6 @@ export const useLogout = () => {
} catch (error) {
// 即使登出失败也清除本地状态
} finally {
cancelProactiveRefresh()
clearAuth()
queryClient.clear()
navigate("/")
-7
View File
@@ -9,13 +9,6 @@ import { QueryClient, QueryClientProvider } from "@tanstack/react-query"
import { ConfigProvider, App as AntApp } from "antd"
import zhCN from "antd/locale/zh_CN"
import router from "./router"
import { scheduleProactiveRefresh } from "./api/auth/tokenRefresh"
// 应用启动时,如果用户已登录,立即调度主动 token 刷新
// 这样可以在 token 过期前自动刷新,避免 API 请求触发 401
if (localStorage.getItem("access_token")) {
scheduleProactiveRefresh()
}
import "./index.css"
import "./styles/global.css"
+26 -1
View File
@@ -12,7 +12,10 @@
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"
@@ -71,6 +74,20 @@ 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()
@@ -218,6 +235,8 @@ const GeneratePage: React.FC = () => {
previewOverallError={step5Preview.previewError}
previewOverallProgress={step5Preview.progress}
previewAnyGenerating={step5Preview.anyGenerating}
previewTemplateName={step5Preview.templateName}
previewMaterialCount={step5Preview.materialCount}
onGeneratePreview={step5Preview.generatePreview}
onRegeneratePreview={step5Preview.regeneratePreview}
/>
@@ -235,7 +254,7 @@ const GeneratePage: React.FC = () => {
{/* ════ 右侧:预览 + 生成结果 ════ */}
<div className="xx-generate-right-col">
{/* 预览视频面板(Step4+ 显示 */}
{/* 预览视频面板(Step4+ 常驻,Step4 显示标题预览,Step5+ 显示预览视频 */}
{currentStep >= 4 && (
<PreviewVideoPanel
previewStatus={step5Preview.previewStatus}
@@ -244,8 +263,14 @@ const GeneratePage: React.FC = () => {
progress={step5Preview.progress}
videoRatio={videoRatio}
onRegenerate={step5Preview.regeneratePreview}
titleText={titleSettings.title}
titleSettings={titleSettings}
showTitlePreview={currentStep === 4}
sourceVideoUrl={sourceVideoUrl}
/>
)}
{/* 正式生成结果(Step6+ 才显示) */}
{currentStep >= 6 && (
<GenerateResultPanel
generated={generated}
@@ -74,6 +74,8 @@ export interface GenerateStepContentProps {
previewOverallError: string
previewOverallProgress: number
previewAnyGenerating: boolean
previewTemplateName: string
previewMaterialCount: string
onGeneratePreview: () => void
onRegeneratePreview: () => void
}
@@ -120,6 +122,8 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
previewOverallError,
previewOverallProgress,
previewAnyGenerating,
previewTemplateName,
previewMaterialCount,
onGeneratePreview,
onRegeneratePreview,
} = props
@@ -161,6 +165,9 @@ export const GenerateStepContent: React.FC<GenerateStepContentProps> = (props) =
case 5:
return (
<Step5GeneratePreview
templateName={previewTemplateName}
materialCount={previewMaterialCount}
duration={duration}
videoRatio={videoRatio}
previewCount={previewCount}
onPreviewCountChange={onPreviewCountChange}
@@ -1,16 +1,22 @@
/**
* 右侧预览视频面板
* Step4+: 显示预览视频面板
* Step5+: 显示后端生成的预览视频(标题已由 FFmpeg 烧录
* Step4 生成预览后常驻显示预览视频
* Step5+ 用 Canvas 绘制标题预览(替代 CSS overlay,与 ASS 渲染行为一致
*
* 设计说明:
* - Step4(标题设置页):右侧显示空状态提示,引导用户输入标题
* - Step5(预览生成页):显示后端返回的预览视频
* - Canvas 预览已删除(统一由后端 FFmpeg 渲染标题)
* 设计说明:标题预览仅在有视频时显示(叠加在视频画面上方)。
* 无视频状态(idle/loading/error)下不再单独显示标题预览,这是有意为之的设计简化。
*
* Canvas 居中修复说明:
* Canvas 的 CSS 位置和尺寸直接匹配视频实际渲染区域(通过 getBoundingClientRect),
* 绘制坐标系基于 Canvas 自身尺寸,x = w/2 即可实现水平居中,
* 避免容器与视频尺寸不一致时浏览器拉伸 Canvas 导致居中偏移。
*/
import React from "react"
import React, { useRef, useEffect, useCallback } from "react"
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
@@ -19,6 +25,14 @@ interface PreviewVideoPanelProps {
progress: number
videoRatio: string
onRegenerate: () => void
/** 标题文字 */
titleText?: string
/** 标题样式设置 */
titleSettings?: TitleSettings
/** Step4 标题预览模式 */
showTitlePreview?: boolean
/** 素材视频 URL(用于 Step4 标题预览背景) */
sourceVideoUrl?: string
}
/* ── 组件 ── */
@@ -30,32 +44,178 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
progress,
videoRatio,
onRegenerate,
titleText,
titleSettings,
showTitlePreview,
sourceVideoUrl,
}) => {
const hasPreview = previewStatus === "ready" && previewResult
const isLoading = previewStatus === "pending" || previewStatus === "generating"
const isError = previewStatus === "error"
const videoAspectStyle = { aspectRatio: (videoRatio || "16:9").replace(":", "/") }
// video 模式 refs
const canvasRef = useRef<HTMLCanvasElement>(null)
const containerRef = useRef<HTMLDivElement>(null)
const videoRef = useRef<HTMLVideoElement>(null)
// 字体加载状态(ref 供 draw 回调同步读取,无需 state 避免触发不必要的重渲染)
const fontLoadedRef = useRef(false)
/** 在 video canvas 上绘制标题 */
const drawVideoTitle = useCallback(() => {
// 通过 ref 读取字体状态,避免 fontLoaded 进入依赖数组
if (!fontLoadedRef.current) return
const canvas = canvasRef.current
const container = containerRef.current
if (!canvas || !container || !titleSettings) return
const ctx = canvas.getContext("2d")
if (!ctx) return
const containerRect = container.getBoundingClientRect()
if (containerRect.width <= 0 || containerRect.height <= 0) return
// 使用 video 元素的 getBoundingClientRect 获取实际渲染尺寸和位置
const videoEl = videoRef.current
let drawW = containerRect.width
let drawH = containerRect.height
let offsetX = 0
let offsetY = 0
if (videoEl && videoEl.clientWidth > 0 && videoEl.clientHeight > 0) {
const videoRect = videoEl.getBoundingClientRect()
drawW = videoRect.width
drawH = videoRect.height
offsetX = videoRect.left - containerRect.left
offsetY = videoRect.top - containerRect.top
}
// 更新 Canvas CSS 位置和尺寸,使其与视频实际渲染区域完全对齐
canvas.style.left = `${offsetX}px`
canvas.style.top = `${offsetY}px`
canvas.style.width = `${drawW}px`
canvas.style.height = `${drawH}px`
// 绘制时,坐标系基于 Canvas 自身尺寸,无需额外偏移
drawTitleOnCanvas(
ctx,
drawW,
drawH,
titleText || "",
titleSettings,
40,
titleSettings.position,
60,
)
}, [titleText, titleSettings])
// video 模式:ResizeObserver 监听容器尺寸变化 → 重绘
useEffect(() => {
if (!showTitlePreview || !hasPreview) return
const container = containerRef.current
if (!container) return
const observer = new ResizeObserver(() => {
drawVideoTitle()
})
observer.observe(container)
requestAnimationFrame(drawVideoTitle)
return () => observer.disconnect()
}, [showTitlePreview, hasPreview, drawVideoTitle])
// 字体加载检测:字体变更时重新检测,确保 measureText 使用正确字体
useEffect(() => {
if (!showTitlePreview || !titleSettings) {
fontLoadedRef.current = false
return
}
let cancelled = false
fontLoadedRef.current = false
const fontWeight = titleSettings.bold ? "bold" : ""
const fontStyle = titleSettings.italic ? "italic" : ""
const fontSpec =
`${fontStyle} ${fontWeight} ${titleSettings.size}px "${titleSettings.font}"`.trim()
const onFontReady = () => {
if (cancelled) return
fontLoadedRef.current = true
// ref 已同步更新,显式触发重绘(draw 内部通过 ref 检查字体状态)
requestAnimationFrame(() => {
if (!cancelled) {
drawVideoTitle()
}
})
}
if (document.fonts.check(fontSpec)) {
onFontReady()
return
}
document.fonts
.load(fontSpec)
.then(() => onFontReady())
.catch(() => {
document.fonts.ready.then(() => onFontReady())
})
return () => {
cancelled = true
}
}, [showTitlePreview, titleSettings, drawVideoTitle])
// video 加载完成后重绘
const handleVideoLoaded = useCallback(() => {
if (showTitlePreview) {
requestAnimationFrame(drawVideoTitle)
}
}, [showTitlePreview, drawVideoTitle])
return (
<div className="xx-generate-preview">
<div className="xx-preview-header">
<h3></h3>
{hasPreview && <span className="xx-preview-badge">480p </span>}
<h3>{showTitlePreview ? "标题预览" : "预览视频"}</h3>
{hasPreview && !showTitlePreview && <span className="xx-preview-badge">480p </span>}
</div>
{/* 空状态:还没生成预览 */}
{previewStatus === "idle" && (
{/* 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" && (
<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>
)}
{/* 生成中 */}
{isLoading && (
{/* 生成中(非 Step4 模式) */}
{!showTitlePreview && isLoading && (
<div className="xx-preview-loading-panel">
<div className="xx-preview-video" style={videoAspectStyle}>
<div className="xx-preview-loading-center">
@@ -71,8 +231,8 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
</div>
)}
{/* 生成失败 */}
{isError && (
{/* 生成失败(非 Step4 模式) */}
{!showTitlePreview && 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>
@@ -86,32 +246,54 @@ export const PreviewVideoPanel: React.FC<PreviewVideoPanelProps> = ({
</div>
)}
{/* 预览成功 */}
{/* 预览成功 + Canvas 标题叠加 */}
{hasPreview && (
<>
<div ref={containerRef} style={{ position: "relative" }}>
<div className="xx-preview-video" style={videoAspectStyle}>
<video src={previewResult.videoUrl} controls preload="metadata" />
<video
ref={videoRef}
src={previewResult.videoUrl}
controls
preload="metadata"
onLoadedMetadata={handleVideoLoaded}
/>
</div>
<div className="xx-preview-info">
<div className="xx-preview-info-row">
<span></span>
<span>
{(typeof previewResult.duration === "number" ? previewResult.duration : 0).toFixed(
1,
)}{" "}
</span>
</div>
<div className="xx-preview-info-row">
<span></span>
<span>{previewResult.clipCount} </span>
</div>
<div className="xx-preview-info-row">
<span></span>
<span>{videoRatio}</span>
</div>
{showTitlePreview && (
<canvas
ref={canvasRef}
style={{
position: "absolute",
top: 0,
left: 0,
width: "100%",
height: "100%",
zIndex: 1,
pointerEvents: "none",
}}
/>
)}
</div>
)}
{/* 预览信息(非 Step4 模式) */}
{!showTitlePreview && hasPreview && previewResult && (
<div className="xx-preview-info">
<div className="xx-preview-info-row">
<span></span>
<span>
{(typeof previewResult.duration === "number" ? previewResult.duration : 0).toFixed(1)}{" "}
</span>
</div>
</>
<div className="xx-preview-info-row">
<span></span>
<span>{previewResult.clipCount} </span>
</div>
<div className="xx-preview-info-row">
<span></span>
<span>{videoRatio}</span>
</div>
</div>
)}
</div>
)
@@ -15,6 +15,9 @@ import { InputNumber } from "antd"
import type { PreviewItem, PreviewStatus } from "../hooks/useStep5Preview"
interface Step5GeneratePreviewProps {
templateName: string
materialCount: string
duration: number
videoRatio: string
previewCount: number
onPreviewCountChange: (count: number) => void
@@ -37,6 +40,8 @@ const PREVIEW_COUNT_OPTIONS = [
]
const Step5GeneratePreview: React.FC<Step5GeneratePreviewProps> = ({
templateName: _templateName,
materialCount: _materialCount,
videoRatio,
previewCount,
onPreviewCountChange,
@@ -0,0 +1,188 @@
/**
* 标题实时预览 Canvas 组件
*
* 在 Step4 标题设置面板中嵌入,让用户实时看到标题文字、字体、大小、颜色、
* 位置、描边、阴影等样式的实际渲染效果(所见即所得)。
*
* 使用共享的 drawTitleOnCanvas 工具函数,与 PreviewVideoPanel 行为一致。
*/
import React, { useRef, useEffect } from "react"
import type { TitleSettings } from "../../types"
import { drawTitleOnCanvas } from "../../utils/drawTitleOnCanvas"
interface TitlePreviewCanvasProps {
/** 标题文字 */
titleText: string
/** 标题样式设置 */
titleSettings: TitleSettings
/** 视频比例,默认 "9:16"(竖屏) */
videoRatio?: string
/** 素材视频 URL(作为背景显示) */
sourceVideoUrl?: string
}
/**
* 解析 videoRatio 字符串为 aspect-ratio CSS 值
*/
function parseAspect(ratio: string): string {
return (ratio || "9:16").replace(":", "/")
}
const TitlePreviewCanvas: React.FC<TitlePreviewCanvasProps> = ({
titleText,
titleSettings,
videoRatio = "9:16",
sourceVideoUrl,
}) => {
const containerRef = useRef<HTMLDivElement>(null)
const canvasRef = useRef<HTMLCanvasElement>(null)
// 字体加载状态
const fontLoadedRef = useRef(false)
/** 在 Canvas 上绘制标题 */
const draw = () => {
const canvas = canvasRef.current
const container = containerRef.current
if (!canvas || !container) return
const ctx = canvas.getContext("2d")
if (!ctx) return
const rect = container.getBoundingClientRect()
if (rect.width <= 0 || rect.height <= 0) return
const w = rect.width
const h = rect.height
// 更新 Canvas CSS 尺寸匹配容器
canvas.style.width = `${w}px`
canvas.style.height = `${h}px`
drawTitleOnCanvas(ctx, w, h, titleText, titleSettings, 24, titleSettings.position, 40)
}
// 字体加载:确保 measureText 使用正确字体
useEffect(() => {
let cancelled = false
fontLoadedRef.current = false
const fontWeight = titleSettings.bold ? "bold" : ""
const fontStyle = titleSettings.italic ? "italic" : ""
const fontSpec =
`${fontStyle} ${fontWeight} ${titleSettings.size}px "${titleSettings.font}"`.trim()
const onFontReady = () => {
if (cancelled) return
fontLoadedRef.current = true
requestAnimationFrame(() => {
if (!cancelled) draw()
})
}
// 用 FontFace API 加载字体,失败则降级
try {
const fontFace = new FontFace(titleSettings.font, `local("${titleSettings.font}")`)
fontFace
.load()
.then(() => {
if (!cancelled) {
;(document.fonts as any).add(fontFace)
onFontReady()
}
})
.catch(() => {
// 字体加载失败,用默认字体继续
onFontReady()
})
} catch {
// FontFace 不可用,直接绘制
onFontReady()
}
// 同时检查 document.fonts 是否已有该字体
if (document.fonts.check(fontSpec)) {
onFontReady()
return
}
return () => {
cancelled = true
}
}, [titleSettings.font, titleSettings.size, titleSettings.bold, titleSettings.italic])
// props 变化时重绘
useEffect(() => {
requestAnimationFrame(draw)
}, [titleText, titleSettings])
// ResizeObserver 监听容器尺寸变化
useEffect(() => {
const container = containerRef.current
if (!container) return
const observer = new ResizeObserver(() => {
requestAnimationFrame(draw)
})
observer.observe(container)
return () => observer.disconnect()
}, [])
return (
<div>
<div
style={{
fontSize: 12,
color: "var(--text-tertiary, #999)",
marginBottom: 6,
fontWeight: 500,
}}
>
</div>
<div
ref={containerRef}
style={{
width: "100%",
aspectRatio: parseAspect(videoRatio),
background: sourceVideoUrl
? "#000"
: "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={{
position: "absolute",
top: 0,
left: 0,
width: "100%",
height: "100%",
}}
/>
</div>
</div>
)
}
export default TitlePreviewCanvas
@@ -0,0 +1,126 @@
/**
* Canvas 标题绘制工具函数(共享模块)
*
* 供 PreviewVideoPanel(预览视频标题叠加)和 TitlePreviewCanvas(标题设置实时预览)共用。
* 绘制行为与 ASS 字幕引擎一致:逐字换行、居中、描边/阴影。
*/
import type { TitleSettings } from "../types"
/**
* 将文本按 maxWidth 逐字换行,返回行数组。
* 与 ASS 字幕引擎的逐字换行行为一致。
*/
export function wrapText(ctx: CanvasRenderingContext2D, text: string, maxWidth: number): string[] {
const lines: string[] = []
let currentLine = ""
for (const char of text) {
const testLine = currentLine + char
if (ctx.measureText(testLine).width > maxWidth && currentLine) {
lines.push(currentLine)
currentLine = char
} else {
currentLine = testLine
}
}
if (currentLine) lines.push(currentLine)
return lines
}
/**
* 在 canvas 上绘制标题文字(含描边/阴影/多行居中)
*
* @param ctx canvas 上下文
* @param w canvas CSS 宽度
* @param h canvas CSS 高度
* @param text 标题文字
* @param settings 标题样式
* @param paddingX 左右边距(px),与 ASS 的 MarginL/MarginR 对应
* @param position "top" | "center" | "bottom"
* @param topOffset 顶部/底部偏移量
*/
export function drawTitleOnCanvas(
ctx: CanvasRenderingContext2D,
w: number,
h: number,
text: string,
settings: TitleSettings,
paddingX: number,
position: string,
topOffset: number,
) {
const dpr = window.devicePixelRatio || 1
// 设置 canvas 物理像素尺寸(高清屏适配)
ctx.canvas.width = Math.round(w * dpr)
ctx.canvas.height = Math.round(h * dpr)
ctx.scale(dpr, dpr)
// 清除
ctx.clearRect(0, 0, w, h)
// 可用宽度 = 总宽 - 左右边距
const availableWidth = w - paddingX * 2
if (availableWidth <= 0) return
// 字体设置
const fontSize = Math.round(Math.min(settings.size, 36))
const fontWeight = settings.bold ? "bold" : "normal"
const fontStyle = settings.italic ? "italic" : "normal"
ctx.font = `${fontStyle} ${fontWeight} ${fontSize}px "${settings.font}"`
// 文字属性
ctx.textAlign = "center"
ctx.textBaseline = "middle"
ctx.fillStyle = settings.color
const lineHeight = fontSize * 1.4
// 描边 & 阴影
if (settings.stroke) {
ctx.strokeStyle = "rgba(0,0,0,0.6)"
ctx.lineWidth = 2
ctx.lineJoin = "round"
}
if (settings.shadow) {
ctx.shadowColor = "rgba(0,0,0,0.7)"
ctx.shadowBlur = 4
ctx.shadowOffsetX = 2
ctx.shadowOffsetY = 2
}
// 换行
const displayText = text && text.trim() ? text : "请选择或输入标题"
const lines = wrapText(ctx, displayText, availableWidth)
// 起始 Y:根据 position 计算
const totalTextHeight = lines.length * lineHeight
let startY: number
switch (position) {
case "top":
startY = topOffset
break
case "center":
startY = (h - totalTextHeight) / 2 + lineHeight / 2
break
case "bottom":
default:
startY = h - topOffset - totalTextHeight + lineHeight / 2
break
}
// 居中 x = w/2
const x = w / 2
lines.forEach((line, i) => {
const y = startY + i * lineHeight
if (settings.stroke) ctx.strokeText(line, x, y)
ctx.fillText(line, x, y)
})
// 重置 shadow(避免影响后续绘制)
if (settings.shadow) {
ctx.shadowColor = "transparent"
ctx.shadowBlur = 0
ctx.shadowOffsetX = 0
ctx.shadowOffsetY = 0
}
}
+8 -21
View File
@@ -31,27 +31,14 @@ vi.mock("react-router-dom", async () => {
})
vi.mock("@/store/authStore", () => ({
useAuthStore: Object.assign(
(selector: any) =>
selector({
user: { id: "1", username: "testuser" },
token: "mock-token",
refreshToken: "mock-refresh-token",
isAuthenticated: true,
setAuth: mockSetAuth,
clearAuth: mockClearAuth,
}),
{
getState: () => ({
user: { id: "1", username: "testuser" },
token: "mock-token",
refreshToken: "mock-refresh-token",
isAuthenticated: true,
setAuth: mockSetAuth,
clearAuth: mockClearAuth,
}),
},
),
useAuthStore: (selector: any) =>
selector({
user: { id: "1", username: "testuser" },
token: "mock-token",
isAuthenticated: true,
setAuth: mockSetAuth,
clearAuth: mockClearAuth,
}),
}))
vi.mock("@tanstack/react-query", () => ({
+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):
@@ -13,16 +13,6 @@ import logging
from pathlib import Path
from typing import Any
from packages.domain.ass_subtitle_builder import (
TITLE_MARGIN_SIDE,
TITLE_MARGIN_TOP,
_wrap_title_text,
build_ass_style,
escape_ass_text,
format_ass_time,
hex_to_ass_color,
position_to_ass_alignment,
)
from packages.domain.subtitle import SubtitleTimeline
logger = logging.getLogger(__name__)
@@ -110,10 +100,7 @@ def generate_ass_from_timeline(
*,
video_width: int,
video_height: int,
video_duration: float = 0.0,
subtitle_config: dict[str, Any] | None = None,
title_text: str = "",
title_config: dict[str, Any] | None = None,
) -> Path:
"""从字幕时间轴生成 ASS 字幕文件。
@@ -173,76 +160,7 @@ def generate_ass_from_timeline(
events.append(f"Dialogue: 0,{start_time},{end_time},Default,,0,0,0,,{safe_text}")
# ── 标题样式与事件(叠加在 ASR 字幕之上)───────────────────────────
title_cfg = title_config or {}
if not isinstance(title_cfg, dict):
title_cfg = {}
title_enabled = title_cfg.get("enabled", True) and bool(title_text.strip())
title_style_line = ""
title_event_line = ""
if title_enabled:
# 兼容 boolean stroke/shadow → dict
_stroke_val = title_cfg.get("stroke")
if isinstance(_stroke_val, bool):
title_cfg["stroke"] = (
{"enabled": _stroke_val, "color": "#000000", "width": 2} if _stroke_val else {"enabled": False}
)
_shadow_val = title_cfg.get("shadow")
if isinstance(_shadow_val, bool):
title_cfg["shadow"] = (
{"enabled": _shadow_val, "color": "#000000", "blur": 4, "offset_x": 2, "offset_y": 2}
if _shadow_val
else {"enabled": False}
)
# 字段名归一化: font_size→size, font_color→color
if "font_size" in title_cfg and "size" not in title_cfg:
title_cfg["size"] = title_cfg["font_size"]
if "font_color" in title_cfg and "color" not in title_cfg:
title_cfg["color"] = title_cfg["font_color"]
t_color = hex_to_ass_color(title_cfg.get("color", "#ffffff"))
t_stroke = title_cfg.get("stroke", {}) or {}
t_shadow = title_cfg.get("shadow", {}) or {}
s_color = hex_to_ass_color(t_stroke.get("color", "#000000"))
s_width = float(t_stroke.get("width", 2)) if t_stroke.get("enabled", False) else 0.0
sh_blur = float(t_shadow.get("blur", 4)) if t_shadow.get("enabled", False) else 0.0
sh_offset = (
t_shadow.get("offset_x", 2) if t_shadow.get("enabled", False) else 0,
t_shadow.get("offset_y", 2) if t_shadow.get("enabled", False) else 0,
)
t_alignment = position_to_ass_alignment(title_cfg.get("position", "top"))
title_style_line = build_ass_style(
"TitleStyle",
font_name=title_cfg.get("font", "思源黑体"),
font_size=min(int(title_cfg.get("size", 36)), 36),
primary_color=t_color,
outline_color=s_color,
outline_width=s_width,
shadow_blur=sh_blur,
shadow_offset=sh_offset,
bold=bool(title_cfg.get("bold", True)),
italic=bool(title_cfg.get("italic", False)),
alignment=t_alignment,
margin_v=TITLE_MARGIN_TOP,
margin_l=TITLE_MARGIN_SIDE,
margin_r=TITLE_MARGIN_SIDE,
)
t_font_size = min(int(title_cfg.get("size", 36)), 36)
safe_raw = escape_ass_text(title_text.strip())
safe_wrapped = _wrap_title_text(safe_raw, video_width, t_font_size)
if video_duration > 0:
t_end_time = format_ass_time(video_duration)
else:
t_end_time = format_ass_time((timeline.segments[-1].end + 5.0) if timeline.segments else 60.0)
title_event_line = f"Dialogue: 0,0:00:00.00,{t_end_time},TitleStyle,,0,0,0,,{safe_wrapped}"
# 组装 ASS 文件
# 组装 ASS 文件
ass_content = f"""[Script Info]
ScriptType: v4.00+
PlayResX: {video_width}
@@ -252,12 +170,12 @@ WrapStyle: 2
Encoding: UTF-8
[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding # noqa: E501
{chr(10).join(filter(None, [title_style_line, style_line]))}
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
{style_line}
[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
{chr(10).join(filter(None, [title_event_line] + events))}
{chr(10).join(events)}
"""
output_path.parent.mkdir(parents=True, exist_ok=True)
@@ -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(
@@ -513,10 +589,7 @@ class UnifiedRenderService:
timeline,
video_width=self.output_width,
video_height=self.output_height,
video_duration=video_duration,
subtitle_config=subtitle_cfg,
title_text=title_text,
title_config=title_cfg,
)
logger.info(
"ASR自动字幕生成完成: plan_id=%s segments=%d duration=%.1fs",
@@ -524,14 +597,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
# 静态字幕模式(原有逻辑)
@@ -662,8 +792,6 @@ 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 = {}
@@ -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:
+93 -74
View File
@@ -1125,14 +1125,14 @@ def _render_video(
bgm_config: dict | None = None,
voice_ids: list[str] | None = None,
custom_title: str = "",
) -> tuple[Path, float]:
) -> 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}")
@@ -1157,33 +1157,10 @@ def _render_video(
list(template_config.keys()),
)
# ── 用户自定义标题覆盖模板标题配置 ──────────────────────────────────
if custom_title:
try:
user_title_cfg = json.loads(custom_title) if isinstance(custom_title, str) else custom_title
if isinstance(user_title_cfg, dict) and user_title_cfg.get("text", "").strip():
# 字段名归一化: 前端 font_size/font_color → 后端 size/color
if "font_size" in user_title_cfg and "size" not in user_title_cfg:
user_title_cfg["size"] = user_title_cfg["font_size"]
if "font_color" in user_title_cfg and "color" not in user_title_cfg:
user_title_cfg["color"] = user_title_cfg["font_color"]
plan_cfg = dict(virtual_plan.config or {})
plan_cfg["title"] = user_title_cfg
virtual_plan.config = plan_cfg
logger.info(
"[task_id=%s] [渲染] 用户自定义标题已注入: text=%s",
task_id,
user_title_cfg.get("text", "")[:30],
)
except (json.JSONDecodeError, TypeError):
logger.warning("[task_id=%s] custom_title JSON解析失败: %s", task_id, custom_title[:100])
# 用户自定义 BGM 覆盖模板 BGM(用户指定优先级最高)
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
@@ -1194,6 +1171,61 @@ 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 {})
_raw_title = plan_cfg.get("title", {}) or {}
title_cfg = dict(_raw_title) if isinstance(_raw_title, dict) else {}
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 + 低码率
@@ -1217,6 +1249,7 @@ def _render_video(
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(
@@ -1271,8 +1304,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(
@@ -1283,6 +1317,7 @@ def _upload_and_record(
editing_mode,
user_id: str = "",
video_name: str = "",
thumbnail_url: str = "",
) -> tuple[str, float, int, int]:
"""上传 OSS、创建视频记录并查重。
@@ -1543,7 +1578,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,
@@ -1563,6 +1598,35 @@ def generate_video(self, task_id: str) -> dict:
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 + 查重记录 ───────────────────────────────────────
@@ -1575,6 +1639,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:
@@ -1588,52 +1653,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
View File
@@ -67,9 +67,6 @@ RUN chmod +x /usr/local/bin/entrypoint-worker.sh
# 业务代码(变化最频繁,放最后)
COPY apps/worker/ /app/apps/worker/
# ---- Install CJK fonts for ASS subtitle rendering ----
RUN apt-get update && apt-get install -y --no-install-recommends fonts-noto-cjk && fc-cache -fv && rm -rf /var/lib/apt/lists/*
USER celery
# Worker 入口点
+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)
@@ -412,7 +412,7 @@ class TestBuildAssContent:
for line in result.split("\n"):
if line.startswith("Style: TitleStyle"):
parts = line.split(",")
assert parts[2] == "36"
assert parts[2] == "72"
break
def test_title_bold(self):
@@ -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,
-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 == ""
-256
View File
@@ -1,256 +0,0 @@
"""预览视频标题渲染修复测试 — 覆盖3个断点。
断点1: generate_video() → _render_video() 传递 custom_title
断点2: _render_video() 解析 custom_title 并注入 virtual_plan.config["title"]
断点3: generate_ass_from_timeline() ASR路径也渲染标题
"""
import json
import time
from pathlib import Path
from unittest.mock import MagicMock, PropertyMock, patch
import pytest
# ── 断点2: _render_video 标题注入 ─────────────────────────────────────────────
class TestRenderVideoCustomTitleInjection:
"""验证 _render_video 正确接收并注入 custom_title 到 virtual_plan.config['title']。"""
@pytest.fixture
def sample_custom_title(self):
"""模拟前端发送的 custom_title JSON(含 font_size/font_color)。"""
return json.dumps(
{
"text": "测试标题",
"font": "思源黑体",
"font_size": 30,
"font_color": "#FF0000",
"position": "top",
"bold": True,
"stroke": True,
"shadow": False,
},
ensure_ascii=False,
)
def _call_render_video_with_capture(self, custom_title, template_config=None, tmp_path=None):
"""调用 _render_video,在 RenderAdapter 处中断并捕获 virtual_plan.config。"""
from worker_app.tasks.generation import _render_video
captured_config = {}
class FakePlan:
def __init__(self):
self.config = {}
self.id = "test-plan"
fake_plan = FakePlan()
def capture_and_raise(*args, **kwargs):
# 此时 title 已注入到 fake_plan.config
captured_config.update(fake_plan.config or {})
raise RuntimeError("STOP_HERE")
with (
patch("worker_app.tasks.generation._build_plan_and_clips_from_task") as mock_build,
patch("worker_app.tasks.generation._load_template_plan_config", return_value=template_config),
patch("worker_app.tasks.generation.time.monotonic", side_effect=[0.0, 1.0]),
patch("video_processing.render_adapter.RenderAdapter") as mock_adapter_cls,
):
mock_build.return_value = (fake_plan, [], {})
mock_adapter_cls.side_effect = capture_and_raise
with pytest.raises(RuntimeError, match="STOP_HERE"):
_render_video(
task_id="test-task",
downloaded_videos=[tmp_path / "v1.mp4"] if tmp_path else [Path("/tmp/v1.mp4")],
voice_path=None,
editing_mode=MagicMock(value="one_take"),
project_id="proj-1",
template_id="tpl-1",
user_id="user-1",
temp_path=tmp_path or Path("/tmp"),
output_name="test_output",
resolution="1280x720",
bgm_config={},
voice_ids=[],
custom_title=custom_title,
)
return captured_config
def test_custom_title_injected_into_plan_config(self, sample_custom_title, tmp_path):
"""custom_title JSON 应被解析并注入 virtual_plan.config['title']。"""
config = self._call_render_video_with_capture(sample_custom_title, tmp_path=tmp_path)
assert "title" in config
title_cfg = config["title"]
assert title_cfg["text"] == "测试标题"
# 字段归一化: font_size → size
assert title_cfg["size"] == 30
# 字段归一化: font_color → color
assert title_cfg["color"] == "#FF0000"
def test_custom_title_overrides_template_title(self, sample_custom_title, tmp_path):
"""用户自定义标题应覆盖模板默认标题。"""
template_config = {"title": {"text": "模板默认标题", "size": 24}}
config = self._call_render_video_with_capture(
sample_custom_title, template_config=template_config, tmp_path=tmp_path
)
# 用户标题应覆盖模板标题
assert config["title"]["text"] == "测试标题"
assert config["title"]["size"] == 30
def test_empty_custom_title_no_injection(self, tmp_path):
"""空 custom_title 不应注入 title 字段。"""
config = self._call_render_video_with_capture("", tmp_path=tmp_path)
assert "title" not in config
def test_malformed_custom_title_gracefully_ignored(self, tmp_path):
"""非法 JSON 不应崩溃,应跳过注入。"""
config = self._call_render_video_with_capture("{invalid json!!!", tmp_path=tmp_path)
assert "title" not in config
# ── 断点3: generate_ass_from_timeline ASR路径支持标题 ──────────────────────────
class TestGenerateAssFromTimelineWithTitle:
"""验证 generate_ass_from_timeline 在有标题时生成包含 TitleStyle 的 ASS。"""
def test_title_included_in_ass_output(self, tmp_path):
"""有 title_text 时,ASS 输出应包含 TitleStyle 和标题事件。"""
from video_processing.subtitle_generator import generate_ass_from_timeline
from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
timeline = SubtitleTimeline(
segments=[
SubtitleSegment(start=0.0, end=2.0, text="你好世界"),
]
)
ass_path = tmp_path / "test.ass"
result = generate_ass_from_timeline(
ass_path,
timeline,
video_width=1280,
video_height=720,
video_duration=10.0,
subtitle_config={"font": "思源黑体", "size": 24},
title_text="我的标题",
title_config={"font": "思源黑体", "size": 36, "color": "#FFFFFF", "position": "top"},
)
content = result.read_text(encoding="utf-8")
# 应包含 TitleStyle
assert "TitleStyle" in content
# 应包含标题文本
assert "我的标题" in content
# 也应包含 ASR 字幕
assert "你好世界" in content
def test_no_title_no_title_style(self, tmp_path):
"""无标题时,ASS 输出不应包含 TitleStyle。"""
from video_processing.subtitle_generator import generate_ass_from_timeline
from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
timeline = SubtitleTimeline(
segments=[
SubtitleSegment(start=0.0, end=2.0, text="只有字幕"),
]
)
ass_path = tmp_path / "test.ass"
result = generate_ass_from_timeline(
ass_path,
timeline,
video_width=1280,
video_height=720,
video_duration=10.0,
subtitle_config={},
title_text="",
title_config={},
)
content = result.read_text(encoding="utf-8")
assert "TitleStyle" not in content
assert "只有字幕" in content
def test_title_field_normalization_in_ass(self, tmp_path):
"""前端字段名 font_size/font_color 应被正确归一化。"""
from video_processing.subtitle_generator import generate_ass_from_timeline
from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
timeline = SubtitleTimeline(segments=[SubtitleSegment(start=0.0, end=2.0, text="test")])
ass_path = tmp_path / "test.ass"
result = generate_ass_from_timeline(
ass_path,
timeline,
video_width=1280,
video_height=720,
video_duration=10.0,
subtitle_config={},
title_text="归一化测试",
title_config={
"font_size": 30, # 前端字段名
"font_color": "#FF0000", # 前端字段名
"position": "top",
},
)
content = result.read_text(encoding="utf-8")
assert "TitleStyle" in content
assert "归一化测试" in content
def test_title_boolean_stroke_shadow_compat(self, tmp_path):
"""boolean stroke/shadow 应被兼容处理。"""
from video_processing.subtitle_generator import generate_ass_from_timeline
from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
timeline = SubtitleTimeline(segments=[SubtitleSegment(start=0.0, end=2.0, text="test")])
ass_path = tmp_path / "test.ass"
result = generate_ass_from_timeline(
ass_path,
timeline,
video_width=1280,
video_height=720,
video_duration=10.0,
subtitle_config={},
title_text="描边测试",
title_config={
"size": 36,
"stroke": True, # boolean
"shadow": False, # boolean
},
)
content = result.read_text(encoding="utf-8")
assert "TitleStyle" in content
assert "描边测试" in content
# ── 断点1: _render_video 签名包含 custom_title ────────────────────────────────
class TestRenderVideoSignature:
"""验证 _render_video 函数签名正确。"""
def test_custom_title_parameter_exists(self):
"""_render_video 应有 custom_title 参数,默认空字符串。"""
import inspect
from worker_app.tasks.generation import _render_video
sig = inspect.signature(_render_video)
assert "custom_title" in sig.parameters
assert sig.parameters["custom_title"].default == ""
-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