Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 720b040710 | |||
| 6090412d26 |
@@ -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,
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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),
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user