fix: add title_text support to cover generation endpoint with image overlay

apps/api/app/api/routes/generation_cover.py

@@ -47,6 +47,10 @@ class GenerateCoverRequest(BaseModel):
         ge=0.0,
         description="手动选帧时间点（秒），仅 cover_type=manual 时有效",
     )
+    title_text: Optional[str] = Field(
+        default=None,
+        description="用户选择的标题文字，叠加到封面图上",
+    )
 
 
 class GenerateCoverResponse(BaseModel):
@@ -59,6 +63,74 @@ class GenerateCoverResponse(BaseModel):
 # ── Route ────────────────────────────────────────────────────────────────
 
 
+
+# ── 封面标题叠加工具函数 ──────────────────────────────────────────────
+
+
+def _apply_title_to_cover_image(
+    image_url: str,
+    title_text: str,
+    plan_id: str,
+) -> str:
+    """下载封面图、叠加标题文字、重新上传，返回新 URL。
+
+    Args:
+        image_url: 原始封面图片 URL
+        title_text: 要叠加的标题文字
+        plan_id: 用于日志和 OSS 路径
+
+    Returns:
+        叠加标题后的新封面图片 URL，失败时返回原 URL
+    """
+    import tempfile as _tempfile
+    import os as _os
+    from pathlib import Path as _Path
+
+    if not title_text or not title_text.strip():
+        return image_url
+
+    # 下载原图
+    import urllib.request as _urllib_req
+    tmp_dir = _tempfile.mkdtemp(prefix=f"cover_title_{plan_id[:8]}_")
+    src_path = _os.path.join(tmp_dir, "cover_src.jpg")
+    try:
+        _urllib_req.urlretrieve(image_url, src_path)
+    except Exception:
+        logger.warning("[封面标题] 下载封面图失败: url=%s", image_url[:80])
+        return image_url
+
+    # 调用 thumbnail_generator 的 _overlay_title_on_image
+    try:
+        from video_processing.thumbnail_generator import _overlay_title_on_image
+        _overlay_title_on_image(src_path, title_text)
+    except Exception:
+        logger.warning("[封面标题] FFmpeg drawtext 失败: plan_id=%s", plan_id, exc_info=True)
+        return image_url
+
+    # 重新上传到 OSS
+    try:
+        from packages.shared.storage import get_shared_storage_service
+        storage_svc = get_shared_storage_service()
+        storage_key = f"covers/{plan_id}/titled_cover.jpg"
+        with open(src_path, "rb") as f:
+            file_data = f.read()
+        new_url = storage_svc.upload_bytes(file_data, storage_key, content_type="image/jpeg")
+        if new_url:
+            return new_url
+        logger.warning("[封面标题] OSS 上传返回空: plan_id=%s", plan_id)
+        return image_url
+    except Exception:
+        logger.warning("[封面标题] OSS 上传失败: plan_id=%s", plan_id, exc_info=True)
+        return image_url
+    finally:
+        # 清理临时文件
+        try:
+            _Path(src_path).unlink(missing_ok=True)
+            _os.rmdir(tmp_dir)
+        except Exception:
+            pass
+
+
 @router.post("/generate-cover", response_model=GenerateCoverResponse)
 def generate_cover(
     body: GenerateCoverRequest,
@@ -201,6 +273,24 @@ def generate_cover(
             "confidence": 0.9,
         }
         if cover_data["image_url"]:
+            # 如果用户指定了标题且候选帧未叠加标题，则叠加标题到封面图上
+            title_text = (body.title_text or "").strip()
+            if title_text:
+                try:
+                    cover_data["image_url"] = _apply_title_to_cover_image(
+                        cover_data["image_url"], title_text, plan_id
+                    )
+                    logger.info(
+                        "[封面生成] 标题已叠加到封面候选帧: plan_id=%s title=%s",
+                        plan_id,
+                        title_text[:30],
+                    )
+                except Exception:
+                    logger.warning(
+                        "[封面生成] 标题叠加失败（使用原图）: plan_id=%s",
+                        plan_id,
+                        exc_info=True,
+                    )
             current_config = dict(plan.config) if plan.config else {}
             current_config["cover"] = cover_data
             normalized = normalize_plan_config(current_config)
@@ -221,6 +311,25 @@ def generate_cover(
     except RuntimeError as e:
         raise HTTPException(status_code=500, detail=str(e)) from e
 
+    # 如果用户指定了标题，叠加到 AI 生成的封面图上
+    title_text = (body.title_text or "").strip()
+    if title_text and cover_data.get("image_url"):
+        try:
+            cover_data["image_url"] = _apply_title_to_cover_image(
+                cover_data["image_url"], title_text, plan_id
+            )
+            logger.info(
+                "[封面生成] 标题已叠加到 AI 封面: plan_id=%s title=%s",
+                plan_id,
+                title_text[:30],
+            )
+        except Exception:
+            logger.warning(
+                "[封面生成] AI 封面标题叠加失败（使用原图）: plan_id=%s",
+                plan_id,
+                exc_info=True,
+            )
+
     current_config = dict(plan.config) if plan.config else {}
     current_config["cover"] = cover_data
     normalized = normalize_plan_config(current_config)