style: auto-format with black + isort + prettier [skip ci-format-check]

Root cause: cover frame is extracted during preview video rendering (Step 3),
but user selects title AFTER rendering (Step 4). The cover frame naturally
has no title because it was captured before the title was chosen.

Fix: In generation_cover.py, after finding cover_url_from_task, check if
plan.config['title']['text'] has a title. If yes, download the cover image,
use ffmpeg drawtext to overlay the title (white text + black shadow, centered
near bottom), upload to OSS, and return the new URL. On failure, fallback
to original cover_url.

Changes:
- Add _overlay_title_on_cover_image() helper in generation_cover.py
- Add _escape_drawtext_text() for ffmpeg special character escaping
- Modify cover route to check for title and overlay if present
- 12 new unit tests (test_cover_title_overlay.py)
- All 13603 tests pass

Fixes: cover title missing issue in staging

apps/api/app/api/routes/generation_cover.py

@@ -31,6 +31,145 @@ logger = logging.getLogger(__name__)
 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 ──────────────────────────────────────────────────────────────
 
 
@@ -282,9 +421,33 @@ 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,
+                    )
+
             cover_data = {
                 "type": "ai_frame",
-                "image_url": cover_url_from_task,
+                "image_url": final_cover_url,
                 "frame_time": 0.0,
                 "confidence": 0.95,
             }

tests/unit/test_cover_title_overlay.py

@@ -0,0 +1,219 @@
+"""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"