fix(preview): 预览渲染支持标题配置,解决封面标题时序问题 #1378

Merged
auto-approve-bot merged 2 commits from fix/preview-title-before-cover into develop 2026-08-15 17:06:00 +08:00
7 changed files with 261 additions and 388 deletions
+2 -162
View File
@@ -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:
+4
View File
@@ -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":
+48 -7
View File
@@ -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
+1
View File
@@ -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:
-219
View File
@@ -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"
+190
View File
@@ -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 == ""