diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index 78b34b42f..2b534a13d 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/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, } diff --git a/tests/unit/test_cover_title_overlay.py b/tests/unit/test_cover_title_overlay.py new file mode 100644 index 000000000..a9ae1f22c --- /dev/null +++ b/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"