diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index d3370d308..ef4400615 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/apps/api/app/api/routes/generation_cover.py @@ -63,7 +63,15 @@ class GenerateCoverResponse(BaseModel): # ── Route ──────────────────────────────────────────────────────────────── -def _persist_cover_frame(frame_url: str, plan_id: str, title_text: str = "") -> str: +def _persist_cover_frame( + frame_url: str, + plan_id: str, + title_text: str = "", + *, + title_color: str = "#ffffff", + title_position: str = "bottom", + title_font_size: int | None = None, +) -> str: """下载 MediaKit 返回的临时帧图,可选叠加标题后转存到 OSS covers/ 路径。 Args: @@ -71,6 +79,9 @@ def _persist_cover_frame(frame_url: str, plan_id: str, title_text: str = "") -> plan_id: 剪辑计划 ID(生成 OSS key) title_text: 非空时用 Pillow 在帧上叠加标题(用于 E2 从源素材抽帧, 因为源素材本身没有烧录标题) + title_color: 标题字体颜色(#RRGGBB) + title_position: 标题位置 top/center/bottom + title_font_size: 标题字号,None 时自动计算 """ import tempfile import uuid @@ -94,7 +105,13 @@ def _persist_cover_frame(frame_url: str, plan_id: str, title_text: str = "") -> try: from packages.shared.title_overlay import apply_title_to_image - applied = apply_title_to_image(tmp_path, title_text) + applied = apply_title_to_image( + tmp_path, + title_text, + color=title_color, + position=title_position, + font_size=title_font_size, + ) if applied: logger.info("[封面生成] E2 帧图已叠加标题: plan_id=%s", plan_id) except Exception: @@ -418,11 +435,15 @@ def generate_cover( asset_repo = SQLAlchemyAssetRepository(db) storage_svc = get_shared_storage_service() mk_client = get_mediakit_client() - # 从 plan.config 读取标题,E2 从源素材抽帧时叠加(源素材本身无标题) + # 从 plan.config 读取完整标题样式,E2 从源素材抽帧时叠加(源素材本身无标题) _e2_title_cfg = (plan.config or {}).get("title", {}) or {} if not isinstance(_e2_title_cfg, dict): _e2_title_cfg = {} _e2_title_text = (_e2_title_cfg.get("text", "") or "").strip() if _e2_title_cfg.get("enabled", True) else "" + # 读取标题样式:前端可能传 color 或 font_color,都兼容 + _e2_title_color = _e2_title_cfg.get("color") or _e2_title_cfg.get("font_color") or "#ffffff" + _e2_title_position = _e2_title_cfg.get("position", "bottom") or "bottom" + _e2_title_font_size = _e2_title_cfg.get("font_size") or _e2_title_cfg.get("size") if mk_client.is_available: for aid in body.asset_ids: try: @@ -452,7 +473,14 @@ def generate_cover( if snapshots: raw = snapshots[0].get("image_url") or snapshots[0].get("url") or "" if raw: - cover_url_from_task = _persist_cover_frame(raw, plan_id, title_text=_e2_title_text) + cover_url_from_task = _persist_cover_frame( + raw, + plan_id, + title_text=_e2_title_text, + title_color=_e2_title_color, + title_position=_e2_title_position, + title_font_size=_e2_title_font_size, + ) logger.info( "[封面生成] 统一管道封面(步骤E-source-asset): plan_id=%s url=%s", plan_id, diff --git a/apps/worker/video_processing/thumbnail_generator.py b/apps/worker/video_processing/thumbnail_generator.py index 8f8edf49f..4549a2d93 100755 --- a/apps/worker/video_processing/thumbnail_generator.py +++ b/apps/worker/video_processing/thumbnail_generator.py @@ -21,12 +21,13 @@ def apply_title_overlay( image_path: str, title_text: str, *, + color: str = "#ffffff", position: str = "bottom", font_size: int | None = None, margin_ratio: float = 0.06, stroke_width_ratio: float = 0.04, ) -> str: - """在图片上绘制标题文字(白色 + 黑色描边/阴影)。 + """在图片上绘制标题文字(指定颜色 + 黑色描边/阴影)。 委托给 packages.shared.title_overlay.apply_title_to_image, 保持 Worker 内调用方式不变。title_text 为空时直接返回原路径。 @@ -38,6 +39,7 @@ def apply_title_overlay( result = apply_title_to_image( image_path, title_text, + color=color, position=position, font_size=font_size, margin_ratio=margin_ratio, @@ -208,6 +210,9 @@ def extract_and_upload_cover_frames( *, num_frames: int = 3, title_text: str = "", + title_color: str = "#ffffff", + title_position: str = "bottom", + title_font_size: int | None = None, ) -> list[dict]: """从视频中抽取多帧作为封面候选,上传到 OSS。 @@ -215,8 +220,11 @@ def extract_and_upload_cover_frames( video_path: 视频文件路径 plan_id: 编辑计划 ID(用于生成 storage key) num_frames: 抽取帧数(默认 3) - title_text: 标题文字;非空时用 Pillow 叠加到每帧(白色 + 黑色描边)。 + title_text: 标题文字;非空时用 Pillow 叠加到每帧。 从已渲染视频抽帧时通常传空(标题已烧录);从源素材抽帧时传标题。 + title_color: 标题字体颜色(#RRGGBB) + title_position: 标题位置 top/center/bottom + title_font_size: 标题字号,None 时自动计算 Returns: 封面候选列表,每项包含 {"url": str, "position": float} @@ -244,7 +252,13 @@ def extract_and_upload_cover_frames( ) # 从源素材抽帧时叠加标题文字;已渲染视频标题已烧录时传空字符串跳过 if title_text and title_text.strip(): - apply_title_overlay(frame_path, title_text) + apply_title_overlay( + frame_path, + title_text, + color=title_color, + position=title_position, + font_size=title_font_size, + ) storage_key = f"covers/{plan_id}/frame_{i}.jpg" url = upload_to_oss(frame_path, storage_key) if url: diff --git a/packages/shared/title_overlay.py b/packages/shared/title_overlay.py index f7ee4ec22..30ccd4728 100644 --- a/packages/shared/title_overlay.py +++ b/packages/shared/title_overlay.py @@ -40,6 +40,21 @@ def find_title_font(size: int): return ImageFont.load_default() +def _parse_hex_color(color: str, fallback=(255, 255, 255)) -> tuple[int, int, int]: + "将 #RRGGBB / #RGB 解析为 RGB 元组,失败返回 fallback。" + if not color or not isinstance(color, str): + return fallback + c = color.strip().lstrip("#") + try: + if len(c) == 6: + return (int(c[0:2], 16), int(c[2:4], 16), int(c[4:6], 16)) + if len(c) == 3: + return (int(c[0] * 2, 16), int(c[1] * 2, 16), int(c[2] * 2, 16)) + except (ValueError, IndexError): + pass + return fallback + + def wrap_title_text(text: str, font, max_width: int) -> list[str]: """按像素宽度对中英文混合文本自动换行,支持显式 \\n。""" lines: list[str] = [] @@ -71,6 +86,7 @@ def apply_title_to_image( image_path: str, title_text: str, *, + color: str = "#ffffff", position: str = "bottom", font_size: Optional[int] = None, margin_ratio: float = 0.06, @@ -81,6 +97,7 @@ def apply_title_to_image( Args: image_path: 图片路径(处理结果覆盖写回) title_text: 标题文字;为空直接返回 None 表示跳过 + color: 字体颜色(#RRGGBB),默认白色 position: top / center / bottom font_size: 字号,None 时按图片宽度自动计算 margin_ratio: 边缘留白占短边比例 @@ -109,6 +126,7 @@ def apply_title_to_image( if font is None: return None + text_rgb = _parse_hex_color(color) stroke_width = max(2, int(font_size * stroke_width_ratio)) margin = int(min(img_w, img_h) * margin_ratio) max_text_width = img_w - 2 * margin @@ -139,12 +157,12 @@ def apply_title_to_image( y = y_start + i * (line_height + line_gap) # 阴影 draw.text((x + 2, y + 2), ln, font=font, fill=(0, 0, 0)) - # 白色文字 + 黑色描边 + # 文字(颜色由 color 参数控制)+ 黑色描边 draw.text( (x, y), ln, font=font, - fill=(255, 255, 255), + fill=text_rgb, stroke_width=stroke_width, stroke_fill=(0, 0, 0), ) diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index 62c8245d2..1250c33fb 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -987,6 +987,159 @@ class TestUploadCoverType: # 标题文字必须透传给持久化函数(用于源素材帧叠加标题) assert mock_persist.call_args.kwargs.get("title_text") == "我的视频标题" + def test_e2_passes_full_title_style_to_persist(self): + """步骤E2:plan.config.title 包含完整样式时,color/position/font_size 都传给 _persist_cover_frame。""" + from unittest.mock import MagicMock, patch + + from app.api.routes.generation_cover import GenerateCoverRequest + + mock_plan = MagicMock() + mock_plan.config = { + "title": { + "enabled": True, + "text": "样式标题", + "color": "#00ff00", + "position": "top", + "font_size": 42, + } + } + + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + mock_template_svc = MagicMock() + mock_db = MagicMock() + + mock_asset = MagicMock() + mock_asset.file_type = "video" + mock_asset.storage_key = "uploads/src.mp4" + mock_asset_repo = MagicMock() + mock_asset_repo.get.return_value = mock_asset + + mock_mk = MagicMock() + mock_mk.is_available = True + mock_mk.extract_frames.return_value = [{"image_url": "https://mk/frame.jpg"}] + + mock_storage = MagicMock() + mock_storage.get_url.return_value = "https://oss.example.com/uploads/src.mp4" + + body = GenerateCoverRequest(cover_type="ai_frame", asset_ids=["a1"]) + + with ( + patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch( + "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository", + return_value=mock_asset_repo, + ), + patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk), + patch("packages.shared.storage.get_shared_storage_service", return_value=mock_storage), + patch( + "app.api.routes.generation_cover._persist_cover_frame", + return_value="https://oss.example.com/covers/styled.jpg", + ) as mock_persist, + patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize, + ): + mock_repo = MagicMock() + mock_repo.get.return_value = None + mock_repo.list_by_source_edit_plan.return_value = [] + mock_repo.list_latest_completed_preview.return_value = [] + mock_repo_cls.return_value = mock_repo + mock_normalize.return_value = { + "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/styled.jpg"} + } + + from app.api.routes.generation_cover import generate_cover + + result = generate_cover( + body=body, + template_id="tpl", + plan_id="plan-style", + services=(mock_template_svc, mock_plan_svc), + db=mock_db, + current_user=MagicMock(), + ) + + assert result.cover["image_url"] == "https://oss.example.com/covers/styled.jpg" + kwargs = mock_persist.call_args.kwargs + assert kwargs["title_text"] == "样式标题" + assert kwargs["title_color"] == "#00ff00" + assert kwargs["title_position"] == "top" + assert kwargs["title_font_size"] == 42 + + def test_e2_title_style_fallback_font_color(self): + """步骤E2:前端传 font_color 时能正确兼容读取。""" + from unittest.mock import MagicMock, patch + + from app.api.routes.generation_cover import GenerateCoverRequest + + mock_plan = MagicMock() + mock_plan.config = { + "title": { + "enabled": True, + "text": "兼容标题", + "font_color": "#123456", + "position": "center", + } + } + + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + mock_template_svc = MagicMock() + mock_db = MagicMock() + + mock_asset = MagicMock() + mock_asset.file_type = "video" + mock_asset.storage_key = "uploads/src.mp4" + mock_asset_repo = MagicMock() + mock_asset_repo.get.return_value = mock_asset + + mock_mk = MagicMock() + mock_mk.is_available = True + mock_mk.extract_frames.return_value = [{"image_url": "https://mk/frame.jpg"}] + + mock_storage = MagicMock() + mock_storage.get_url.return_value = "https://oss.example.com/uploads/src.mp4" + + body = GenerateCoverRequest(cover_type="ai_frame", asset_ids=["a1"]) + + with ( + patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch( + "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository", + return_value=mock_asset_repo, + ), + patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk), + patch("packages.shared.storage.get_shared_storage_service", return_value=mock_storage), + patch( + "app.api.routes.generation_cover._persist_cover_frame", + return_value="https://oss.example.com/covers/compat.jpg", + ) as mock_persist, + patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize, + ): + mock_repo = MagicMock() + mock_repo.get.return_value = None + mock_repo.list_by_source_edit_plan.return_value = [] + mock_repo.list_latest_completed_preview.return_value = [] + mock_repo_cls.return_value = mock_repo + mock_normalize.return_value = { + "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/compat.jpg"} + } + + from app.api.routes.generation_cover import generate_cover + + generate_cover( + body=body, + template_id="tpl", + plan_id="plan-compat", + services=(mock_template_svc, mock_plan_svc), + db=mock_db, + current_user=MagicMock(), + ) + + kwargs = mock_persist.call_args.kwargs + assert kwargs["title_color"] == "#123456" + assert kwargs["title_position"] == "center" + assert kwargs["title_font_size"] is None + def test_step_e_skips_non_video_assets(self): """步骤E2:asset_ids 里只有图片素材时,不调用 MediaKit 并返回 400。""" from unittest.mock import MagicMock, patch diff --git a/tests/unit/test_title_overlay.py b/tests/unit/test_title_overlay.py index 75c7eac1a..a4e648d6d 100644 --- a/tests/unit/test_title_overlay.py +++ b/tests/unit/test_title_overlay.py @@ -56,3 +56,34 @@ def test_wrap_title_text_respects_explicit_newline(): font = ImageFont.load_default() lines = wrap_title_text("第一行\n第二行", font, max_width=10000) assert lines == ["第一行", "第二行"] + + +def test_apply_title_to_image_custom_color(sample_image): + """自定义颜色参数能正常生成图片。""" + result = apply_title_to_image(sample_image, "彩色标题", color="#ff0000") + assert result == sample_image + assert Path(sample_image).stat().st_size > 0 + + +def test_apply_title_to_image_short_hex_color(sample_image): + """3 位缩写 hex 颜色也能正常解析。""" + result = apply_title_to_image(sample_image, "短色", color="#f00") + assert result == sample_image + + +def test_apply_title_to_image_invalid_color_fallback(sample_image): + """无效颜色字符串 fallback 到白色,不报错。""" + result = apply_title_to_image(sample_image, "异常色", color="not-a-color") + assert result == sample_image + + +def test_parse_hex_color(): + from packages.shared.title_overlay import _parse_hex_color + + assert _parse_hex_color("#ffffff") == (255, 255, 255) + assert _parse_hex_color("#000000") == (0, 0, 0) + assert _parse_hex_color("#ff0000") == (255, 0, 0) + assert _parse_hex_color("#f00") == (255, 0, 0) + assert _parse_hex_color("") == (255, 255, 255) + assert _parse_hex_color("invalid") == (255, 255, 255) + assert _parse_hex_color("#gggggg") == (255, 255, 255)