From f6b50b49ece3b941093bd716889b64ad7cc58159 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 23 Aug 2026 10:28:52 +0800 Subject: [PATCH] fix(cover): worker local ffmpeg cover frames + title overlay for source-asset fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题： 1. 之前 Worker 用 MediaKit 网络抽帧写 cover_url，因状态字符串 bug（#1461 已修） 一直失败；且在 API 请求内同步抽帧易超时。 2. 封面从源素材抽取时无标题（源素材未烧录标题）。 修复： 1. Worker 渲染完成后直接复用 RenderAdapter 已用本地 ffmpeg 抽出的 cover_candidates： - _render_video 返回值改为 (output_path, render_duration, cover_candidates) - 第一帧写入 gen_task.cover_url，完整列表写入 metadata.cover_candidates - 删除步骤 4.5 的 MediaKit 抽帧代码（不依赖网络，不阻塞） 2. 标题叠加： - 新增 packages/shared/title_overlay.py，Pillow 绘制白色文字+黑色描边/阴影， 支持 CJK 字体和自动换行（API/Worker 共用） - thumbnail_generator.extract_and_upload_cover_frames 的 title_text 参数真正生效 - render_adapter 从已渲染视频抽帧时传空 title_text（标题已 ASS 烧录，避免重影） - 封面 API E2 从源素材抽帧后，从 plan.config.title.text 读取标题并叠加 3. API 基础镜像增加 Pillow 和 fonts-noto-cjk（E2 标题叠加依赖） 测试：新增 test_title_overlay.py（5）、E2 标题透传测试，更新 1294 解包； 相关 51 passed。 --- apps/api/app/api/routes/generation_cover.py | 33 +++- .../worker/video_processing/render_adapter.py | 11 +- .../video_processing/thumbnail_generator.py | 45 +++++- apps/worker/worker_app/tasks/generation.py | 86 +++++----- infra/docker/api-base.Dockerfile | 2 + packages/shared/title_overlay.py | 153 ++++++++++++++++++ requirements-base.txt | 1 + .../unit/test_1294_preview_voice_injection.py | 2 +- tests/unit/test_generation_cover.py | 67 ++++++++ tests/unit/test_title_overlay.py | 58 +++++++ 10 files changed, 399 insertions(+), 59 deletions(-) create mode 100644 packages/shared/title_overlay.py create mode 100644 tests/unit/test_title_overlay.py diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index a0ff342aa..d3370d308 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/apps/api/app/api/routes/generation_cover.py @@ -63,8 +63,15 @@ class GenerateCoverResponse(BaseModel): # ── Route ──────────────────────────────────────────────────────────────── -def _persist_cover_frame(frame_url: str, plan_id: str) -> str: - """下载 MediaKit 返回的临时帧图并转存到 OSS covers/ 路径。""" +def _persist_cover_frame(frame_url: str, plan_id: str, title_text: str = "") -> str: + """下载 MediaKit 返回的临时帧图，可选叠加标题后转存到 OSS covers/ 路径。 + + Args: + frame_url: MediaKit 返回的临时帧图 URL + plan_id: 剪辑计划 ID（生成 OSS key） + title_text: 非空时用 Pillow 在帧上叠加标题（用于 E2 从源素材抽帧， + 因为源素材本身没有烧录标题） + """ import tempfile import uuid from pathlib import Path @@ -82,6 +89,21 @@ def _persist_cover_frame(frame_url: str, plan_id: str) -> str: tmp.write(resp.content) tmp_path = tmp.name + # E2 从源素材抽帧时，源素材无标题，叠加标题文字 + if title_text and title_text.strip(): + try: + from packages.shared.title_overlay import apply_title_to_image + + applied = apply_title_to_image(tmp_path, title_text) + if applied: + logger.info("[封面生成] E2 帧图已叠加标题: plan_id=%s", plan_id) + except Exception: + logger.warning( + "[封面生成] E2 标题叠加失败（返回无标题帧）: plan_id=%s", + plan_id, + exc_info=True, + ) + from packages.shared.storage import get_shared_storage_service storage = get_shared_storage_service() @@ -396,6 +418,11 @@ def generate_cover( asset_repo = SQLAlchemyAssetRepository(db) storage_svc = get_shared_storage_service() mk_client = get_mediakit_client() + # 从 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 "" if mk_client.is_available: for aid in body.asset_ids: try: @@ -425,7 +452,7 @@ 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) + cover_url_from_task = _persist_cover_frame(raw, plan_id, title_text=_e2_title_text) logger.info( "[封面生成] 统一管道封面(步骤E-source-asset): plan_id=%s url=%s", plan_id, diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index c9782214d..a8c2301f5 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -585,14 +585,11 @@ class RenderAdapter: try: from video_processing.thumbnail_generator import extract_and_upload_cover_frames - # 从 plan config 提取标题文字，叠加到封面候选帧上 - _title_cfg = (plan_config or {}).get("title", {}) or {} - if not isinstance(_title_cfg, dict): - _title_cfg = {} - _title_text = (_title_cfg.get("text", "") or "").strip() if _title_cfg.get("enabled", True) else "" - + # 已渲染视频在统一渲染阶段已通过 ASS 字幕把标题烧录进画面， + # 抽帧天然带标题，因此这里传空字符串，避免 Pillow 二次叠加导致重影。 + # Pillow 叠加仅用于 API 从源素材抽帧（源素材本身无标题）的兜底场景。 cover_candidates = extract_and_upload_cover_frames( - str(result.output_path), plan_id, num_frames=3, title_text=_title_text + str(result.output_path), plan_id, num_frames=3, title_text="" ) if cover_candidates: logger.info( diff --git a/apps/worker/video_processing/thumbnail_generator.py b/apps/worker/video_processing/thumbnail_generator.py index ad6e2fd6f..8f8edf49f 100755 --- a/apps/worker/video_processing/thumbnail_generator.py +++ b/apps/worker/video_processing/thumbnail_generator.py @@ -1,7 +1,8 @@ -"""视频封面抽帧工具 — 从已渲染视频中抽取帧作为封面。 +"""视频封面抽帧工具 — 从视频中抽取帧作为封面，支持标题文字叠加。 -统一封面管道：视频渲染时标题已通过 ASS 字幕烧进视频， -渲染完成后直接从此视频抽帧，封面天然带标题，无需额外叠加逻辑。 +统一封面管道： +- 从已渲染视频抽帧：标题已通过 ASS 字幕烧进视频，帧天然带标题，无需再叠加。 +- 从源素材抽帧（API E2 兜底）：源素材无标题，通过 Pillow 在帧上绘制标题文字。 """ from __future__ import annotations @@ -12,6 +13,38 @@ from pathlib import Path logger = logging.getLogger(__name__) +# ── 标题叠加（Pillow）────────────────────────────────────────────────────── +# 实现统一放在 packages/shared/title_overlay.py，API 和 Worker 共用。 + + +def apply_title_overlay( + image_path: str, + title_text: str, + *, + 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 为空时直接返回原路径。 + """ + from packages.shared.title_overlay import apply_title_to_image + + if not title_text or not title_text.strip(): + return image_path + result = apply_title_to_image( + image_path, + title_text, + position=position, + font_size=font_size, + margin_ratio=margin_ratio, + stroke_width_ratio=stroke_width_ratio, + ) + return result or image_path + def extract_first_frame( video_path: str, @@ -182,7 +215,8 @@ def extract_and_upload_cover_frames( video_path: 视频文件路径 plan_id: 编辑计划 ID（用于生成 storage key） num_frames: 抽取帧数（默认 3） - title_text: 标题文字（当前版本未叠加，预留参数） + title_text: 标题文字；非空时用 Pillow 叠加到每帧（白色 + 黑色描边）。 + 从已渲染视频抽帧时通常传空（标题已烧录）；从源素材抽帧时传标题。 Returns: 封面候选列表，每项包含 {"url": str, "position": float} @@ -208,6 +242,9 @@ def extract_and_upload_cover_frames( seek_ratio=ratio, min_seek_seconds=0.5, ) + # 从源素材抽帧时叠加标题文字；已渲染视频标题已烧录时传空字符串跳过 + if title_text and title_text.strip(): + apply_title_overlay(frame_path, title_text) storage_key = f"covers/{plan_id}/frame_{i}.jpg" url = upload_to_oss(frame_path, storage_key) if url: diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index dfef315d2..49b0dd53d 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -967,14 +967,14 @@ def _render_video( bgm_config: dict | None = None, voice_ids: list[str] | None = None, custom_title: str = "", -) -> tuple[Path, float]: +) -> tuple[Path, float, list[dict] | None]: """渲染视频（含配音混音）。 使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。 Args: Returns: - (output_path, render_duration) + (output_path, render_duration, cover_candidates) """ if not downloaded_videos: raise RuntimeError(f"素材下载结果为空: task_id={task_id}") @@ -1113,8 +1113,10 @@ def _render_video( # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理 output_path = render_output_path + # RenderAdapter 在渲染完成后用本地 ffmpeg 抽取的封面候选帧（已上传 OSS） + cover_candidates = getattr(render_result, "cover_candidates", None) - return output_path, render_duration + return output_path, render_duration, cover_candidates def _upload_and_record( @@ -1385,7 +1387,7 @@ def generate_video(self, task_id: str) -> dict: else: _resolved_resolution = task_info.get("resolution", "") - output_path, render_duration = _render_video( + output_path, render_duration, cover_candidates = _render_video( task_id=task_id, downloaded_videos=downloaded_videos, voice_path=audio_path, @@ -1430,51 +1432,47 @@ def generate_video(self, task_id: str) -> dict: _update_task_progress(task_id, 95, "上传完成") - # ── 4.5 封面抽帧 ──────────────────────────────────────────────── - # 预览视频上传完成后，提取封面帧写入 gen_task.cover_url - # 这样封面路由（generation_cover.py 步骤A）可以通过 generation_task_id 直接找到 + # ── 4.5 封面帧持久化 ──────────────────────────────────────────── + # RenderAdapter 在渲染完成后已用本地 ffmpeg 从 output_path 抽帧 + # （标题通过 ASS 烧录，帧天然带标题），并上传 OSS 返回 cover_candidates。 + # 这里把第一帧写入 gen_task.cover_url，完整列表写入 metadata， + # 封面路由（generation_cover.py）的 A/B/C/D 步骤即可直接命中。 try: - from packages.shared.mediakit_client import get_mediakit_client + if cover_candidates: + first = cover_candidates[0] + # 候选帧字段兼容：RenderAdapter 用 image_url，thumbnail_generator 用 url + cover_frame_url = first.get("image_url") or first.get("url") or "" + if cover_frame_url: + _cover_session = SessionLocal() + try: + from packages.adapters.sqlalchemy_impl.models import ( + GenerationTaskModel, + ) - mk_client = get_mediakit_client() - if mk_client.is_available: - _update_task_progress(task_id, 96, "提取封面帧") - snapshots = mk_client.extract_frames( - video_url=file_url, - strategy="SpecifiedFrames", - max_frames=1, - ) - if snapshots and len(snapshots) > 0: - cover_frame_url = snapshots[0].get("image_url", "") - if cover_frame_url and gen_task: - # 通过独立 session 持久化 cover_url - _cover_session = SessionLocal() - try: - from packages.adapters.sqlalchemy_impl.models import ( - GenerationTaskModel, + _cover_model = ( + _cover_session.query(GenerationTaskModel) + .filter(GenerationTaskModel.id == task_id) + .first() + ) + if _cover_model: + _cover_model.cover_url = cover_frame_url + # 持久化完整候选列表到 metadata + meta = dict(_cover_model.metadata or {}) + meta["cover_candidates"] = cover_candidates + _cover_model.metadata = meta + _cover_session.commit() + logger.info( + "[task_id=%s] 封面帧已持久化(ffmpeg本地抽帧): cover_url=%s candidates=%d", + task_id, + cover_frame_url[:80], + len(cover_candidates), ) - - _cover_model = ( - _cover_session.query(GenerationTaskModel) - .filter(GenerationTaskModel.id == task_id) - .first() - ) - if _cover_model: - _cover_model.cover_url = cover_frame_url - _cover_session.commit() - logger.info( - "[task_id=%s] 封面帧提取成功: %s", - task_id, - cover_frame_url[:80], - ) - finally: - _cover_session.close() - else: - logger.warning("[task_id=%s] 封面帧提取返回空结果", task_id) + finally: + _cover_session.close() else: - logger.warning("[task_id=%s] MediaKit 未配置，跳过封面帧提取", task_id) + logger.warning("[task_id=%s] 渲染未产出 cover_candidates，封面将依赖 API 兜底", task_id) except Exception: - logger.warning("[task_id=%s] 封面帧提取失败（不影响主流程）", task_id, exc_info=True) + logger.warning("[task_id=%s] 封面帧持久化失败（不影响主流程）", task_id, exc_info=True) # ── 5. 标记完成 ────────────────────────────────────────────────── _update_task_status(task_id, "mark_completed", result_count=video_count) diff --git a/infra/docker/api-base.Dockerfile b/infra/docker/api-base.Dockerfile index 9108a2df9..72117a8f1 100644 --- a/infra/docker/api-base.Dockerfile +++ b/infra/docker/api-base.Dockerfile @@ -17,6 +17,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ libpq-dev \ libpq5 \ ffmpeg \ + fonts-noto-cjk \ + fontconfig \ && rm -rf /var/lib/apt/lists/* # 创建虚拟环境 diff --git a/packages/shared/title_overlay.py b/packages/shared/title_overlay.py new file mode 100644 index 000000000..f7ee4ec22 --- /dev/null +++ b/packages/shared/title_overlay.py @@ -0,0 +1,153 @@ +"""封面标题文字叠加（Pillow）— API / Worker 共用。 + +在封面帧上绘制白色标题文字 + 黑色描边/阴影，支持 CJK 字体和自动换行。 +从已渲染视频抽帧时通常不需要调用（标题已烧录）； +从源素材抽帧（API E2 兜底）时调用，保证封面带标题。 +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Optional + +logger = logging.getLogger(__name__) + +# 按优先级查找 CJK 字体（Debian/Ubuntu fonts-noto-cjk 安装路径） +_FONT_CANDIDATES = ( + "/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc", + "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/truetype/noto/NotoSansCJK-Bold.ttc", + "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc", +) + + +def find_title_font(size: int): + """查找可用的 CJK 字体并返回 PIL ImageFont，找不到返回 None。""" + try: + from PIL import ImageFont + except ImportError: + return None + + for fp in _FONT_CANDIDATES: + if Path(fp).exists(): + try: + return ImageFont.truetype(fp, size=size) + except Exception: + continue + logger.warning("未找到 CJK 字体，标题叠加将使用 PIL 默认字体（中文可能显示为方块）") + return ImageFont.load_default() + + +def wrap_title_text(text: str, font, max_width: int) -> list[str]: + """按像素宽度对中英文混合文本自动换行，支持显式 \\n。""" + lines: list[str] = [] + current = "" + for ch in text: + if ch == "\n": + if current: + lines.append(current) + current = "" + continue + trial = current + ch + try: + bbox = font.getbbox(trial) + width = bbox[2] - bbox[0] + except Exception: + width = len(trial) * (font.size // 2) + if width <= max_width: + current = trial + else: + if current: + lines.append(current) + current = ch + if current: + lines.append(current) + return lines + + +def apply_title_to_image( + image_path: str, + title_text: str, + *, + position: str = "bottom", + font_size: Optional[int] = None, + margin_ratio: float = 0.06, + stroke_width_ratio: float = 0.04, +) -> Optional[str]: + """在图片上绘制标题文字并覆盖保存。 + + Args: + image_path: 图片路径（处理结果覆盖写回） + title_text: 标题文字；为空直接返回 None 表示跳过 + position: top / center / bottom + font_size: 字号，None 时按图片宽度自动计算 + margin_ratio: 边缘留白占短边比例 + stroke_width_ratio: 描边宽度占字号比例 + + Returns: + 成功返回 image_path；标题为空或 PIL 不可用返回 None。 + """ + if not title_text or not title_text.strip(): + return None + + try: + from PIL import Image, ImageDraw + except ImportError: + logger.warning("Pillow 未安装，跳过标题叠加: image=%s", image_path) + return None + + img = Image.open(image_path).convert("RGB") + draw = ImageDraw.Draw(img) + img_w, img_h = img.size + + if font_size is None: + font_size = max(28, min(72, img_w // 16)) + + font = find_title_font(font_size) + if font is None: + return None + + 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 + + lines = wrap_title_text(title_text.strip(), font, max_text_width) + if not lines: + return None + + line_heights = [] + for ln in lines: + bbox = font.getbbox(ln) + line_heights.append(bbox[3] - bbox[1]) + line_height = max(line_heights) if line_heights else font_size + line_gap = int(line_height * 0.3) + total_height = len(lines) * line_height + (len(lines) - 1) * line_gap + + if position == "top": + y_start = margin + elif position == "center": + y_start = (img_h - total_height) // 2 + else: + y_start = img_h - total_height - margin + + for i, ln in enumerate(lines): + bbox = font.getbbox(ln) + line_w = bbox[2] - bbox[0] + x = (img_w - line_w) // 2 + y = y_start + i * (line_height + line_gap) + # 阴影 + draw.text((x + 2, y + 2), ln, font=font, fill=(0, 0, 0)) + # 白色文字 + 黑色描边 + draw.text( + (x, y), + ln, + font=font, + fill=(255, 255, 255), + stroke_width=stroke_width, + stroke_fill=(0, 0, 0), + ) + + img.save(image_path, "JPEG", quality=92) + return image_path diff --git a/requirements-base.txt b/requirements-base.txt index 637ae9f2e..d29b06545 100644 --- a/requirements-base.txt +++ b/requirements-base.txt @@ -29,3 +29,4 @@ httpx==0.27.2 # Prometheus monitoring prometheus-client==0.21.1 +Pillow==10.4.0 diff --git a/tests/unit/test_1294_preview_voice_injection.py b/tests/unit/test_1294_preview_voice_injection.py index 7f3c75709..3b5c11363 100644 --- a/tests/unit/test_1294_preview_voice_injection.py +++ b/tests/unit/test_1294_preview_voice_injection.py @@ -158,7 +158,7 @@ class TestRenderVideoVoiceInjection: from packages.domain import EditingMode - output_path, render_duration = _render_video( + output_path, render_duration, cover_candidates = _render_video( task_id="test_task_123", downloaded_videos=[Path("/tmp/video1.mp4")], voice_path=None, diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index dc6a05747..62c8245d2 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -920,6 +920,73 @@ class TestUploadCoverType: call_kwargs = mock_mk.extract_frames.call_args.kwargs assert "source-clip.mp4" in call_kwargs["video_url"] + def test_e2_passes_plan_title_to_persist_for_overlay(self): + """步骤E2：plan.config.title.text 存在时，作为 title_text 传给 _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": "我的视频标题"}} + + 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/final.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/final.jpg"} + } + + from app.api.routes.generation_cover import generate_cover + + result = generate_cover( + body=body, + template_id="tpl", + plan_id="plan-title", + services=(mock_template_svc, mock_plan_svc), + db=mock_db, + current_user=MagicMock(), + ) + + assert result.cover["image_url"] == "https://oss.example.com/covers/final.jpg" + # 标题文字必须透传给持久化函数（用于源素材帧叠加标题） + assert mock_persist.call_args.kwargs.get("title_text") == "我的视频标题" + 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 new file mode 100644 index 000000000..75c7eac1a --- /dev/null +++ b/tests/unit/test_title_overlay.py @@ -0,0 +1,58 @@ +"""packages.shared.title_overlay 单元测试。""" + +from __future__ import annotations + +import tempfile +from pathlib import Path + +import pytest + +from packages.shared.title_overlay import apply_title_to_image, wrap_title_text + + +@pytest.fixture +def sample_image(): + """生成一张 640x360 的纯黑测试图片。""" + from PIL import Image + + tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) + tmp.close() + img = Image.new("RGB", (640, 360), color=(0, 0, 0)) + img.save(tmp.name, "JPEG") + yield tmp.name + Path(tmp.name).unlink(missing_ok=True) + + +def test_apply_title_to_image_empty_text_returns_none(sample_image): + assert apply_title_to_image(sample_image, "") is None + assert apply_title_to_image(sample_image, " ") is None + + +def test_apply_title_to_image_draws_title(sample_image): + result = apply_title_to_image(sample_image, "测试标题") + assert result == sample_image + assert Path(sample_image).exists() + assert Path(sample_image).stat().st_size > 0 + + +def test_apply_title_to_image_respects_position(sample_image): + for pos in ("top", "center", "bottom"): + result = apply_title_to_image(sample_image, "位置测试", position=pos) + assert result == sample_image + + +def test_wrap_title_text_supports_long_text(): + from PIL import ImageFont + + font = ImageFont.load_default() + lines = wrap_title_text("这是一个比较长的标题需要自动换行处理ABC", font, max_width=40) + assert isinstance(lines, list) + assert len(lines) >= 1 + + +def test_wrap_title_text_respects_explicit_newline(): + from PIL import ImageFont + + font = ImageFont.load_default() + lines = wrap_title_text("第一行\n第二行", font, max_width=10000) + assert lines == ["第一行", "第二行"] -- 2.54.0