fix(preview): 预览渲染支持标题配置,解决封面标题时序问题 #1378
@@ -32,143 +32,6 @@ router = APIRouter(tags=["Generation"])
|
||||
|
||||
|
||||
|
||||
# ── Helper: 封面标题叠加 ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _escape_drawtext_text(text: str) -> str:
|
||||
"""转义 ffmpeg drawtext 的特殊字符."""
|
||||
# ffmpeg drawtext 需要转义: ' → \' , : → \: , % → %%
|
||||
escaped = text.replace("'", "'\\''")
|
||||
escaped = escaped.replace(":", "\\:")
|
||||
escaped = escaped.replace("%", "%%")
|
||||
return escaped
|
||||
|
||||
|
||||
def _overlay_title_on_cover_image(
|
||||
cover_image_url: str,
|
||||
title_text: str,
|
||||
title_config: dict | None = None,
|
||||
plan_id: str = "",
|
||||
) -> str | None:
|
||||
"""用 ffmpeg drawtext 在封面图片上叠加标题文字.
|
||||
|
||||
轻量方案:下载现有封面图片 → ffmpeg drawtext 叠加标题 → 上传 OSS → 返回新 URL。
|
||||
失败时返回 None,调用方应降级使用原 cover_url。
|
||||
|
||||
Args:
|
||||
cover_image_url: 原始封面图片 URL
|
||||
title_text: 标题文本
|
||||
title_config: 标题样式配置(font_size, font_color, position 等)
|
||||
plan_id: 计划 ID(用于日志)
|
||||
|
||||
Returns:
|
||||
叠加标题后的新封面 URL,失败返回 None
|
||||
"""
|
||||
import hashlib
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from packages.shared.ffmpeg_utils import FFMPEG_BIN
|
||||
from packages.shared.storage import get_shared_storage_service
|
||||
|
||||
logger.info(
|
||||
"[封面标题叠加] 开始: plan_id=%s title=%s",
|
||||
plan_id,
|
||||
title_text[:30] if title_text else "",
|
||||
)
|
||||
|
||||
title_config = title_config or {}
|
||||
|
||||
# 下载原始封面图片
|
||||
tmp_dir = tempfile.mkdtemp(prefix="cover_title_")
|
||||
input_path = Path(tmp_dir) / "input.jpg"
|
||||
output_path = Path(tmp_dir) / "output.jpg"
|
||||
|
||||
try:
|
||||
# 下载封面图片
|
||||
import requests
|
||||
resp = requests.get(cover_image_url, stream=True, timeout=30)
|
||||
resp.raise_for_status()
|
||||
with open(input_path, "wb") as f:
|
||||
for chunk in resp.iter_content(chunk_size=8 * 1024):
|
||||
if chunk:
|
||||
f.write(chunk)
|
||||
|
||||
if not input_path.exists() or input_path.stat().st_size == 0:
|
||||
logger.warning("[封面标题叠加] 下载封面图片失败: %s", cover_image_url[:60])
|
||||
return None
|
||||
|
||||
# 构建 ffmpeg drawtext filter
|
||||
escaped_title = _escape_drawtext_text(title_text)
|
||||
|
||||
# 样式配置
|
||||
font_size = int(title_config.get("font_size", 48))
|
||||
font_color = title_config.get("font_color", "#ffffff") or "#ffffff"
|
||||
# 去掉 # 前缀,ffmpeg 用 0xRRGGBB 格式
|
||||
ffmpeg_color = font_color.replace("#", "0x")
|
||||
|
||||
# drawtext filter: 文字 + 黑色描边 + 阴影,居中偏下
|
||||
drawtext_filter = (
|
||||
f"drawtext=text='{escaped_title}'"
|
||||
f":fontsize={font_size}"
|
||||
f":fontcolor={ffmpeg_color}"
|
||||
f":shadowcolor=0x000000:shadowx=2:shadowy=2"
|
||||
f":borderw=2:bordercolor=black"
|
||||
f":x=(w-text_w)/2:y=h*0.85-text_h"
|
||||
)
|
||||
|
||||
# 执行 ffmpeg
|
||||
cmd = [
|
||||
FFMPEG_BIN,
|
||||
"-y",
|
||||
"-i", str(input_path),
|
||||
"-vf", drawtext_filter,
|
||||
"-q:v", "2",
|
||||
str(output_path),
|
||||
]
|
||||
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
stderr_text = result.stderr.decode(errors="replace")[:200] if result.stderr else ""
|
||||
logger.error(
|
||||
"[封面标题叠加] ffmpeg 失败: rc=%d stderr=%s",
|
||||
result.returncode,
|
||||
stderr_text,
|
||||
)
|
||||
return None
|
||||
|
||||
if not output_path.exists() or output_path.stat().st_size == 0:
|
||||
logger.warning("[封面标题叠加] ffmpeg 输出文件为空")
|
||||
return None
|
||||
|
||||
# 上传到 OSS
|
||||
url_hash = hashlib.sha256(cover_image_url.encode()).hexdigest()[:12]
|
||||
storage_key = f"covers/{url_hash}_titled.jpg"
|
||||
|
||||
storage_svc = get_shared_storage_service()
|
||||
new_url = storage_svc.upload_file(str(output_path), storage_key, content_type="image/jpeg")
|
||||
|
||||
logger.info(
|
||||
"[封面标题叠加] ✅ 成功: plan_id=%s url=%s",
|
||||
plan_id,
|
||||
new_url[:60] if new_url else "",
|
||||
)
|
||||
return new_url
|
||||
|
||||
except Exception:
|
||||
logger.exception("[封面标题叠加] 异常,降级使用原封面")
|
||||
return None
|
||||
finally:
|
||||
# 清理临时文件
|
||||
shutil.rmtree(tmp_dir, ignore_errors=True)
|
||||
|
||||
|
||||
# ── Schemas ──────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -421,33 +284,10 @@ def generate_cover(
|
||||
)
|
||||
|
||||
if cover_url_from_task:
|
||||
# 检查是否有标题需要叠加(用户在选封面前已选标题)
|
||||
final_cover_url = cover_url_from_task
|
||||
title_text = ((plan.config or {}).get("title", {}) or {}).get("text", "") or ""
|
||||
if title_text and title_text.strip():
|
||||
title_config = (plan.config or {}).get("title", {}) or {}
|
||||
titled_url = _overlay_title_on_cover_image(
|
||||
cover_image_url=cover_url_from_task,
|
||||
title_text=title_text,
|
||||
title_config=title_config,
|
||||
plan_id=plan_id,
|
||||
)
|
||||
if titled_url:
|
||||
final_cover_url = titled_url
|
||||
logger.info(
|
||||
"[封面生成] 标题叠加成功: plan_id=%s title=%s",
|
||||
plan_id,
|
||||
title_text[:30],
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
"[封面生成] 标题叠加失败,降级使用原封面: plan_id=%s",
|
||||
plan_id,
|
||||
)
|
||||
|
||||
# 标题已在预览视频渲染时烧录(ASS字幕),封面帧自然包含标题
|
||||
cover_data = {
|
||||
"type": "ai_frame",
|
||||
"image_url": final_cover_url,
|
||||
"image_url": cover_url_from_task,
|
||||
"frame_time": 0.0,
|
||||
"confidence": 0.95,
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
|
||||
from app.auth import AuthenticatedUser, get_current_user
|
||||
@@ -267,6 +268,20 @@ def create_preview_generation_task(
|
||||
# 从模板读取 editing_mode / mode 作为 strategy_id(渲染 pipeline 的 mode 参数)
|
||||
strategy_id = _resolve_strategy_id_from_template(request.template_id, db, user_id)
|
||||
|
||||
# 处理标题配置:如果有标题文本,序列化到 custom_title 字段传递给 worker
|
||||
title_config = request.title_config or {}
|
||||
title_text = (title_config.get("text") or "").strip()
|
||||
custom_title_value = ""
|
||||
if title_text:
|
||||
# 将标题文本和样式配置序列化为 JSON 存入 custom_title
|
||||
# Worker 端会解析 JSON 获取完整标题配置
|
||||
custom_title_value = json.dumps(title_config, ensure_ascii=False)
|
||||
logger.info(
|
||||
"[预览生成] 标题配置: text=%s, config_keys=%s",
|
||||
title_text[:30],
|
||||
list(title_config.keys()),
|
||||
)
|
||||
|
||||
use_case = CreateGenerationTaskUseCase(generation_task_repository)
|
||||
|
||||
try:
|
||||
@@ -290,6 +305,7 @@ def create_preview_generation_task(
|
||||
auto_retry_enabled=False,
|
||||
auto_retry_max=0,
|
||||
is_preview=True,
|
||||
custom_title=custom_title_value,
|
||||
)
|
||||
)
|
||||
except ValueError as e:
|
||||
|
||||
@@ -181,6 +181,10 @@ class CreatePreviewGenerationTaskRequest(BaseModel):
|
||||
default="",
|
||||
description="关联的编辑计划ID(可选),用于确认生成时复用预览产物",
|
||||
)
|
||||
title_config: dict = Field(
|
||||
default_factory=dict,
|
||||
description="标题配置(可选),渲染时烧录到预览视频中。支持字段: text/font/font_size/font_color/position/bold/stroke/shadow",
|
||||
)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _check_template_id(self) -> "CreatePreviewGenerationTaskRequest":
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import tempfile
|
||||
@@ -1171,18 +1172,58 @@ def _render_video(
|
||||
)
|
||||
|
||||
# 用户自定义标题覆盖模板标题(用户指定优先级最高)
|
||||
# 支持两种格式:
|
||||
# 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 {})
|
||||
title_cfg["text"] = custom_title.strip()
|
||||
title_cfg["enabled"] = True
|
||||
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
|
||||
logger.info(
|
||||
"[task_id=%s] [渲染] 用户自定义标题已注入: title=%s",
|
||||
task_id,
|
||||
custom_title[:50],
|
||||
)
|
||||
|
||||
# 确保输出分辨率配置存在
|
||||
# 优先级:用户指定 > 模板配置 > 默认 1280x720
|
||||
|
||||
@@ -32,6 +32,7 @@ class CreateGenerationTaskCommand:
|
||||
output_height: int = 720
|
||||
cover_url: str = ""
|
||||
custom_title: str = ""
|
||||
title_config: dict = field(default_factory=dict)
|
||||
|
||||
|
||||
class CreateGenerationTaskUseCase:
|
||||
|
||||
@@ -1,219 +0,0 @@
|
||||
"""Tests for cover title overlay functionality in generation_cover.py."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from apps.api.app.api.routes.generation_cover import _escape_drawtext_text
|
||||
|
||||
|
||||
class TestEscapeDrawtextText:
|
||||
"""Tests for _escape_drawtext_text helper."""
|
||||
|
||||
def test_single_quote_escaped(self):
|
||||
"""单引号应被转义."""
|
||||
result = _escape_drawtext_text("it's a test")
|
||||
assert "'\\''" in result
|
||||
|
||||
def test_colon_escaped(self):
|
||||
"""冒号应被转义."""
|
||||
result = _escape_drawtext_text("test: value")
|
||||
assert "\\:" in result
|
||||
|
||||
def test_percent_escaped(self):
|
||||
"""百分号应被转义."""
|
||||
result = _escape_drawtext_text("100% done")
|
||||
assert "%%" in result
|
||||
|
||||
def test_normal_text_unchanged(self):
|
||||
"""普通文本不应被修改."""
|
||||
result = _escape_drawtext_text("普通标题")
|
||||
assert result == "普通标题"
|
||||
|
||||
def test_multiple_special_chars(self):
|
||||
"""多个特殊字符都应被转义."""
|
||||
result = _escape_drawtext_text("it's 50%: done")
|
||||
assert "\\:" in result
|
||||
assert "%%" in result
|
||||
|
||||
|
||||
class TestOverlayTitleOnCoverImage:
|
||||
"""Tests for _overlay_title_on_cover_image function."""
|
||||
|
||||
@patch("packages.shared.storage.get_shared_storage_service")
|
||||
@patch("packages.shared.ffmpeg_utils.FFMPEG_BIN", "ffmpeg")
|
||||
@patch("subprocess.run")
|
||||
@patch("requests.get")
|
||||
def test_successful_overlay(
|
||||
self,
|
||||
mock_requests_get: MagicMock,
|
||||
mock_subprocess_run: MagicMock,
|
||||
mock_storage_svc: MagicMock,
|
||||
tmp_path,
|
||||
):
|
||||
"""成功叠加标题."""
|
||||
from apps.api.app.api.routes.generation_cover import _overlay_title_on_cover_image
|
||||
|
||||
# Mock HTTP download
|
||||
mock_response = MagicMock()
|
||||
mock_response.iter_content.return_value = [b"fake image data"]
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_requests_get.return_value = mock_response
|
||||
|
||||
# Mock ffmpeg success
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 0
|
||||
mock_subprocess_run.return_value = mock_result
|
||||
|
||||
# Mock storage upload
|
||||
mock_storage = MagicMock()
|
||||
mock_storage.upload_file.return_value = "https://example.com/covers/test_titled.jpg"
|
||||
mock_storage_svc.return_value = mock_storage
|
||||
|
||||
# Need to mock Path operations
|
||||
with (
|
||||
patch("pathlib.Path.exists", return_value=True),
|
||||
patch("pathlib.Path.stat") as mock_stat,
|
||||
patch("pathlib.Path.mkdir"),
|
||||
patch("pathlib.Path.unlink"),
|
||||
):
|
||||
mock_stat.return_value.st_size = 1000
|
||||
|
||||
result = _overlay_title_on_cover_image(
|
||||
cover_image_url="https://example.com/cover.jpg",
|
||||
title_text="测试标题",
|
||||
title_config={"font_size": 48},
|
||||
plan_id="test-plan-123",
|
||||
)
|
||||
|
||||
assert result == "https://example.com/covers/test_titled.jpg"
|
||||
|
||||
@patch("requests.get")
|
||||
def test_download_failure_returns_none(self, mock_requests_get: MagicMock):
|
||||
"""下载失败时返回 None."""
|
||||
from apps.api.app.api.routes.generation_cover import _overlay_title_on_cover_image
|
||||
|
||||
mock_requests_get.side_effect = Exception("Network error")
|
||||
|
||||
result = _overlay_title_on_cover_image(
|
||||
cover_image_url="https://example.com/cover.jpg",
|
||||
title_text="测试标题",
|
||||
plan_id="test-plan-123",
|
||||
)
|
||||
|
||||
assert result is None
|
||||
|
||||
@patch("subprocess.run")
|
||||
@patch("requests.get")
|
||||
def test_ffmpeg_failure_returns_none(
|
||||
self,
|
||||
mock_requests_get: MagicMock,
|
||||
mock_subprocess_run: MagicMock,
|
||||
):
|
||||
"""ffmpeg 失败时返回 None."""
|
||||
from apps.api.app.api.routes.generation_cover import _overlay_title_on_cover_image
|
||||
|
||||
# Mock successful download
|
||||
mock_response = MagicMock()
|
||||
mock_response.iter_content.return_value = [b"fake image data"]
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_requests_get.return_value = mock_response
|
||||
|
||||
# Mock ffmpeg failure
|
||||
mock_result = MagicMock()
|
||||
mock_result.returncode = 1
|
||||
mock_result.stderr = b"ffmpeg error"
|
||||
mock_subprocess_run.return_value = mock_result
|
||||
|
||||
with (
|
||||
patch("pathlib.Path.exists", return_value=True),
|
||||
patch("pathlib.Path.stat") as mock_stat,
|
||||
patch("pathlib.Path.mkdir"),
|
||||
patch("pathlib.Path.unlink"),
|
||||
):
|
||||
mock_stat.return_value.st_size = 1000
|
||||
|
||||
result = _overlay_title_on_cover_image(
|
||||
cover_image_url="https://example.com/cover.jpg",
|
||||
title_text="测试标题",
|
||||
plan_id="test-plan-123",
|
||||
)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestCoverRouteWithTitle:
|
||||
"""Integration tests for cover route with title overlay."""
|
||||
|
||||
@patch("apps.api.app.api.routes.generation_cover._overlay_title_on_cover_image")
|
||||
def test_cover_url_with_title_overlay(self, mock_overlay: MagicMock):
|
||||
"""有标题时应调用叠加函数."""
|
||||
mock_overlay.return_value = "https://example.com/covers/titled.jpg"
|
||||
|
||||
# Simulate plan.config with title
|
||||
plan_config = {
|
||||
"title": {"text": "测试标题", "font_size": 48},
|
||||
"generation_task_id": "task-123",
|
||||
}
|
||||
|
||||
title_text = (plan_config.get("title", {}) or {}).get("text", "") or ""
|
||||
assert title_text == "测试标题"
|
||||
|
||||
# Should call overlay
|
||||
if title_text and title_text.strip():
|
||||
result = mock_overlay(
|
||||
cover_image_url="https://example.com/cover.jpg",
|
||||
title_text=title_text,
|
||||
title_config=plan_config.get("title", {}),
|
||||
plan_id="test-plan",
|
||||
)
|
||||
assert result == "https://example.com/covers/titled.jpg"
|
||||
|
||||
def test_cover_url_without_title_no_overlay(self):
|
||||
"""无标题时不应调用叠加函数."""
|
||||
plan_config = {
|
||||
"title": {"text": "", "font_size": 48},
|
||||
"generation_task_id": "task-123",
|
||||
}
|
||||
|
||||
title_text = (plan_config.get("title", {}) or {}).get("text", "") or ""
|
||||
assert title_text == ""
|
||||
|
||||
# Should NOT call overlay - title is empty
|
||||
assert not (title_text and title_text.strip())
|
||||
|
||||
def test_cover_url_with_title_config_none(self):
|
||||
"""title_config 为 None 时应正常处理."""
|
||||
plan_config = {
|
||||
"generation_task_id": "task-123",
|
||||
}
|
||||
|
||||
title_text = (plan_config.get("title", {}) or {}).get("text", "") or ""
|
||||
assert title_text == ""
|
||||
|
||||
@patch("apps.api.app.api.routes.generation_cover._overlay_title_on_cover_image")
|
||||
def test_overlay_failure_fallback_to_original(self, mock_overlay: MagicMock):
|
||||
"""叠加失败时应降级使用原封面."""
|
||||
mock_overlay.return_value = None # 叠加失败
|
||||
|
||||
# Simulate the fallback logic
|
||||
cover_url_from_task = "https://example.com/original.jpg"
|
||||
title_text = "测试标题"
|
||||
title_config = {"font_size": 48}
|
||||
|
||||
titled_url = mock_overlay(
|
||||
cover_image_url=cover_url_from_task,
|
||||
title_text=title_text,
|
||||
title_config=title_config,
|
||||
plan_id="test-plan",
|
||||
)
|
||||
|
||||
# 叠加失败时应该降级使用原封面
|
||||
if titled_url:
|
||||
final_cover_url = titled_url
|
||||
else:
|
||||
final_cover_url = cover_url_from_task
|
||||
|
||||
assert final_cover_url == "https://example.com/original.jpg"
|
||||
@@ -0,0 +1,190 @@
|
||||
"""Tests for preview title_config feature.
|
||||
|
||||
验证预览 API 的 title_config 字段和 Worker 的标题配置解析逻辑。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class TestPreviewTitleConfigSchema:
|
||||
"""测试 CreatePreviewGenerationTaskRequest 的 title_config 字段."""
|
||||
|
||||
def test_title_config_default_empty(self):
|
||||
"""title_config 默认为空 dict."""
|
||||
from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
|
||||
|
||||
req = CreatePreviewGenerationTaskRequest(
|
||||
template_id="test_template",
|
||||
asset_ids=["asset1"],
|
||||
)
|
||||
assert req.title_config == {}
|
||||
|
||||
def test_title_config_with_text(self):
|
||||
"""传入标题文本."""
|
||||
from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
|
||||
|
||||
req = CreatePreviewGenerationTaskRequest(
|
||||
template_id="test_template",
|
||||
asset_ids=["asset1"],
|
||||
title_config={"text": "测试标题"},
|
||||
)
|
||||
assert req.title_config["text"] == "测试标题"
|
||||
|
||||
def test_title_config_with_full_style(self):
|
||||
"""传入完整标题样式配置."""
|
||||
from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
|
||||
|
||||
config = {
|
||||
"text": "我的视频标题",
|
||||
"font": "思源黑体",
|
||||
"font_size": 48,
|
||||
"font_color": "#ffffff",
|
||||
"position": "top",
|
||||
"bold": True,
|
||||
"stroke": 2,
|
||||
"shadow": True,
|
||||
}
|
||||
req = CreatePreviewGenerationTaskRequest(
|
||||
template_id="test_template",
|
||||
asset_ids=["asset1"],
|
||||
title_config=config,
|
||||
)
|
||||
assert req.title_config["text"] == "我的视频标题"
|
||||
assert req.title_config["font_size"] == 48
|
||||
assert req.title_config["position"] == "top"
|
||||
|
||||
|
||||
class TestCommandTitleConfig:
|
||||
"""测试 CreateGenerationTaskCommand 的 title_config 字段."""
|
||||
|
||||
def test_command_has_title_config(self):
|
||||
"""Command 包含 title_config 字段."""
|
||||
from packages.application.generation_tasks import CreateGenerationTaskCommand
|
||||
|
||||
cmd = CreateGenerationTaskCommand(
|
||||
title_config={"text": "hello", "font_size": 32},
|
||||
)
|
||||
assert cmd.title_config["text"] == "hello"
|
||||
assert cmd.title_config["font_size"] == 32
|
||||
|
||||
def test_command_title_config_default_empty(self):
|
||||
"""Command 的 title_config 默认为空 dict."""
|
||||
from packages.application.generation_tasks import CreateGenerationTaskCommand
|
||||
|
||||
cmd = CreateGenerationTaskCommand()
|
||||
assert cmd.title_config == {}
|
||||
|
||||
|
||||
class TestWorkerTitleConfigParsing:
|
||||
"""测试 Worker 渲染时的标题配置解析逻辑."""
|
||||
|
||||
def test_json_format_parsing(self):
|
||||
"""JSON 格式的 custom_title 能正确解析."""
|
||||
config = {"text": "测试标题", "font_size": 48, "font_color": "#ff0000"}
|
||||
custom_title = json.dumps(config, ensure_ascii=False)
|
||||
|
||||
ct_stripped = custom_title.strip()
|
||||
parsed = None
|
||||
if ct_stripped.startswith("{"):
|
||||
try:
|
||||
parsed = json.loads(ct_stripped)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
parsed = None
|
||||
|
||||
assert parsed is not None
|
||||
assert parsed["text"] == "测试标题"
|
||||
assert parsed["font_size"] == 48
|
||||
|
||||
def test_plain_text_fallback(self):
|
||||
"""纯文本的 custom_title 不触发 JSON 解析."""
|
||||
custom_title = "简单的标题文字"
|
||||
|
||||
ct_stripped = custom_title.strip()
|
||||
parsed = None
|
||||
if ct_stripped.startswith("{"):
|
||||
try:
|
||||
parsed = json.loads(ct_stripped)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
parsed = None
|
||||
|
||||
assert parsed is None
|
||||
|
||||
def test_invalid_json_fallback(self):
|
||||
"""无效 JSON 的 custom_title 降级为纯文本."""
|
||||
custom_title = "{invalid json"
|
||||
|
||||
ct_stripped = custom_title.strip()
|
||||
parsed = None
|
||||
if ct_stripped.startswith("{"):
|
||||
try:
|
||||
parsed = json.loads(ct_stripped)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
parsed = None
|
||||
|
||||
assert parsed is None
|
||||
|
||||
def test_json_without_text_skipped(self):
|
||||
"""JSON 格式但缺少 text 字段时,跳过标题注入."""
|
||||
config = {"font_size": 48}
|
||||
custom_title = json.dumps(config, ensure_ascii=False)
|
||||
|
||||
ct_stripped = custom_title.strip()
|
||||
parsed = json.loads(ct_stripped)
|
||||
title_text = (parsed.get("text") or "").strip()
|
||||
|
||||
assert title_text == ""
|
||||
|
||||
def test_style_key_mapping(self):
|
||||
"""前端字段名正确映射到 ASS 字段名."""
|
||||
config = {
|
||||
"text": "标题",
|
||||
"font_size": 48,
|
||||
"font_color": "#ffffff",
|
||||
"font_preset": "思源黑体",
|
||||
}
|
||||
|
||||
style_keys = ["font", "font_size", "font_color", "position", "bold", "stroke", "shadow", "font_preset"]
|
||||
title_cfg = {}
|
||||
for key in style_keys:
|
||||
if key in config and config[key] is not None:
|
||||
mapped_key = {
|
||||
"font_size": "size",
|
||||
"font_color": "color",
|
||||
"font_preset": "font",
|
||||
}.get(key, key)
|
||||
title_cfg[mapped_key] = config[key]
|
||||
|
||||
assert title_cfg["size"] == 48
|
||||
assert title_cfg["color"] == "#ffffff"
|
||||
assert title_cfg["font"] == "思源黑体"
|
||||
|
||||
|
||||
class TestPreviewRouteTitleConfigPassing:
|
||||
"""测试预览路由正确序列化 title_config 到 custom_title."""
|
||||
|
||||
def test_title_config_serialization(self):
|
||||
"""title_config 序列化为 JSON 字符串."""
|
||||
title_config = {
|
||||
"text": "我的标题",
|
||||
"font_size": 32,
|
||||
"font_color": "#d4a843",
|
||||
}
|
||||
serialized = json.dumps(title_config, ensure_ascii=False)
|
||||
|
||||
parsed = json.loads(serialized)
|
||||
assert parsed["text"] == "我的标题"
|
||||
assert parsed["font_size"] == 32
|
||||
|
||||
def test_empty_title_config_produces_empty_string(self):
|
||||
"""空 title_config 时 custom_title 为空字符串."""
|
||||
title_config = {}
|
||||
title_text = (title_config.get("text") or "").strip()
|
||||
custom_title_value = ""
|
||||
if title_text:
|
||||
custom_title_value = json.dumps(title_config, ensure_ascii=False)
|
||||
|
||||
assert custom_title_value == ""
|
||||
Reference in New Issue
Block a user