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"