refactor(ai-avatar): 删除封面标题叠加逻辑+废弃路由(封面一律从成片抽帧)
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 1s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 50s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 53s
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 2m8s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 43s
Preview Deploy / Deploy Preview Environment (pull_request) Failing after 3m12s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Successful in 1m17s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 3m56s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 4m51s
CI/CD Pipeline / Validate - Style (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Security (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
AI Code Review / AI Code Review (pull_request) Has been cancelled
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 86h23m45s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 86h23m57s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 86h23m24s
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Failing after 86h23m28s
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Failing after 86h23m29s
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Failing after 86h23m29s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 86h23m34s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 86h23m34s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 86h23m22s
CI/CD Pipeline / Check push changed paths (pull_request) Failing after 86h23m36s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 86h23m22s

- 删除前端 generateSmartCover(0处调用的死代码)
- 删除后端 POST /ai-avatar/render/smart-cover(对口型阶段抽封面入口,已废弃)
- 删除 SmartCoverRequest schema(无引用)
- 精简 ai_avatar_cover_service:删除 apply_title_to_cover / _overlay_title_png_on_image,
  封面一律从最终成片(已叠加标题/B-roll)抽帧,不再额外叠加
- 删除 persist_cover_to_oss / generate_smart_cover 的 title_config 参数
This commit is contained in:
xiaoxia
2026-09-13 00:58:02 +08:00
parent 577ec83636
commit 63fb0508be
4 changed files with 23 additions and 286 deletions
+1 -53
View File
@@ -18,7 +18,6 @@ from app.dependencies import get_db_session
from app.schemas.ai_avatar_render import (
AiAvatarRenderJobResponse,
CreateAiAvatarRenderRequest,
SmartCoverRequest,
SmartCoverResponse,
)
from app.services.ai_avatar_cover_service import generate_smart_cover
@@ -191,57 +190,6 @@ def retry_render_job(
return AiAvatarRenderJobResponse.model_validate(job)
# ── POST /smart-cover — 智能获取封面(MediaKit 抽帧 + 评分选帧)────────
@router.post("/smart-cover", response_model=SmartCoverResponse)
def generate_avatar_smart_cover(
body: SmartCoverRequest,
current_user: AuthenticatedUser = Depends(get_current_user),
) -> SmartCoverResponse:
"""智能获取数字人视频封面.
复用智能剪辑的 MediaKit 抽帧 + 质量评分选最佳帧逻辑(非 FFmpeg 简单截帧),
并将选中帧转存到自家 OSS,返回非临时的封面公网 URL。
前端「智能获取封面」按钮可直接调用本接口;不依赖渲染任务完成。
"""
video_url = (body.video_url or "").strip()
if not video_url.startswith(("http://", "https://")):
raise HTTPException(status_code=400, detail="video_url 必须是合法的 HTTP/HTTPS URL")
# 兼容:顶层 title_image_dataurl 透传进 title_config(前端可以放在任一位置)
title_config = getattr(body, "title_config", None)
top_dataurl = getattr(body, "title_image_dataurl", None)
if top_dataurl and isinstance(title_config, dict) and not title_config.get("title_image_dataurl"):
title_config = {**title_config, "title_image_dataurl": top_dataurl}
try:
cover_url = generate_smart_cover(
video_url,
max_frames=body.max_frames,
title_config=title_config,
)
except Exception as exc:
logger.error(
"智能封面生成异常: user=%s video_url=%s err=%s",
current_user.user.id,
video_url[:80],
exc,
exc_info=True,
)
cover_url = ""
if not cover_url:
return SmartCoverResponse(
cover_url="",
status="fallback_failed",
message="智能抽帧失败(MediaKit 不可用或抽帧异常),请稍后重试",
)
logger.info("智能封面生成成功: user=%s cover_url=%s", current_user.user.id, cover_url[:120])
return SmartCoverResponse(cover_url=cover_url, status="completed")
# ── POST /{job_id}/smart-cover — 从最终成片智能抽封面(步骤②)────────
@@ -269,7 +217,7 @@ def generate_render_smart_cover(
raise HTTPException(status_code=400, detail="渲染成片视频 URL 为空")
try:
# 成片已叠加标题,不传 title_config 避免双重叠加
# 从最终成片抽帧,帧本身已含标题/B-roll,直接转存 OSS
cover_url = generate_smart_cover(video_url, job_id=job_id, max_frames=5)
except Exception as exc:
logger.error(
+2 -22
View File
@@ -52,7 +52,7 @@ class CreateAiAvatarRenderRequest(BaseModel):
lipsync_job_id: str = Field(..., description="对口型任务 ID")
script_id: str = Field("", description="文案 ID(选自文案库时传;手动输入文案直生场景可留空)")
b_roll_segments: list[BRollSegment] = Field(default_factory=list, description="B-roll 片段列表")
title_config: dict[str, Any] = Field(default_factory=dict, description="标题配置")
title_config: dict[str, Any] = Field(default_factory=dict, description="标题配置(可含 title_image_dataurl:前端 Canvas 渲染的标题 PNG dataURL)")
cover_config: dict[str, Any] = Field(default_factory=dict, description="封面配置")
project_id: str = Field("", description="项目 ID")
@@ -67,7 +67,6 @@ class CreateAiAvatarRenderRequest(BaseModel):
@field_validator("script_id")
@classmethod
def validate_script_id(cls, v: str) -> str:
# script_id 可选:手动输入文案(TTS 直生)场景不关联文案库条目
return (v or "").strip()
@@ -109,27 +108,8 @@ class AiAvatarRenderProgressResponse(BaseModel):
error_message: str
class SmartCoverRequest(BaseModel):
"""智能封面请求 — MediaKit 抽帧 + 质量评分选最佳帧 + 可选标题叠加."""
video_url: str = Field(..., description="数字人视频 URL(对口型/渲染成片)")
max_frames: int = Field(5, ge=1, le=10, description="抽帧数量(默认 5)")
title_config: Optional[dict[str, Any]] = Field(
None,
description="标题配置;传入时在封面上叠加标题。"
"若 title_config 含 title_image_dataurl(前端 Canvas 渲染的 PNG dataURL),"
"后端用 overlay 叠加图片图层实现所见即所得;否则降级用 drawtext 重画文字。",
)
title_image_dataurl: Optional[str] = Field(
None,
description="前端 Canvas 渲染的标题 PNG dataURL(data:image/png;base64,...);"
"传入时后端用 overlay 叠加图片图层,不再用 drawtext 重画文字。"
"推荐方式:直接放在 title_config.title_image_dataurl 里透传,本字段为兼容保留。",
)
class SmartCoverResponse(BaseModel):
"""智能封面响应."""
"""智能封面响应(封面从最终成片抽帧,不再叠加标题)."""
cover_url: str = Field("", description="封面图公网 URL(OSS,非临时);失败为空")
status: str = Field("completed", description="completed / fallback_failed")
+20 -196
View File
@@ -1,20 +1,19 @@
"""AI 数字人封面服务 — 复用智能剪辑的 MediaKit 抽帧 + 质量评分选最佳帧.
"""AI 数字人封面服务 — MediaKit 抽帧 + 质量评分选最佳帧 + 转存 OSS.
与 generation_cover.py 的智能选帧能力对齐(不再用 FFmpeg 简单截帧):
1. MediaKit extract_frames 抽取多帧(默认 5 帧,SpecifiedFrames 策略)
2. cover_frame_scorer.score_frames 按清晰度/亮度/色彩评分选最佳
3. 下载最佳帧并转存 OSS,返回公网封面 URL
设计原则:封面一律从最终成片(已叠加标题/B-roll)抽帧,帧本身已含标题,
本服务**不再叠加标题**。对口型阶段的裸视频封面入口已删除(废弃)。
降级:MediaKit 不可用或抽帧失败时返回空字符串,由调用方决定回退策略。
"""
from __future__ import annotations
import base64
import binascii
import logging
import os
import subprocess
import tempfile
import uuid
from pathlib import Path
@@ -53,7 +52,6 @@ def _sign_video_url_for_mediakit(video_url: str) -> str:
own_host = urlparse(public_base).netloc.lower()
url_host = urlparse(video_url).netloc.lower()
if own_host and url_host == own_host:
# 是自家 OSS URL,重签 7 天有效期供 MediaKit 拉取
signed = storage.get_download_url(video_url, expires_seconds=MEDIAKIT_URL_TTL_SECONDS)
if signed:
logger.info("[数字人封面] video_url 已重签(自家 OSS 私有桶)")
@@ -64,19 +62,10 @@ def _sign_video_url_for_mediakit(video_url: str) -> str:
def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
"""从视频抽取多帧并评分选最佳帧,返回最佳帧的临时 URL.
Args:
video_url: 可公网访问的视频 URL
max_frames: 抽帧数量
Returns:
最佳帧图片 URL;失败返回空字符串
"""
"""从视频抽取多帧并评分选最佳帧,返回最佳帧的临时 URL."""
if not video_url:
return ""
# 确保 MediaKit 能访问 video_url(自家 OSS 私有桶需重签)
video_url = _sign_video_url_for_mediakit(video_url)
try:
@@ -89,11 +78,8 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
return ""
logger.info(
"[数字人封面] 开始抽帧: video_url=%s max_frames=%d poll_interval=%.1f max_poll=%d",
video_url[:80],
max_frames,
COVER_POLL_INTERVAL,
COVER_MAX_POLL_ATTEMPTS,
"[数字人封面] 开始抽帧: video_url=%s max_frames=%d",
video_url[:80], max_frames,
)
snapshots = mk.extract_frames(
@@ -111,7 +97,6 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
if len(snapshots) == 1:
return snapshots[0].get("image_url") or snapshots[0].get("url") or ""
# 使用连接池下载各帧(复用 TCP 连接,减少延迟)
import httpx
candidates = []
@@ -139,7 +124,6 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
best = scored[0] if scored else None
best_url = best.get("url", "") if best else ""
# 清理临时文件
for c in candidates:
p = c.get("image_path")
if p:
@@ -150,8 +134,7 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
logger.info(
"[数字人封面] 智能选帧完成: candidates=%d best_score=%s",
len(candidates),
best.get("score") if best else "n/a",
len(candidates), best.get("score") if best else "n/a",
)
return best_url
@@ -160,161 +143,19 @@ def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
return ""
def apply_title_to_cover(local_frame: str, *, title_config: dict | None) -> str:
"""在封面图上叠加标题,返回叠加后图片的本地路径.
优先使用前端 Canvas 渲染的 PNG 图层(overlay=0:0,所见即所得);
无 title_image_dataurl 时降级到 drawtext 重画文字。
ffmpeg 失败时回退返回原始 local_frame。竖屏封面按 720x1280 计算位置(drawtext 降级路径)。
"""
if not title_config or not isinstance(title_config, dict):
return local_frame
text = (
title_config.get("text")
or title_config.get("content")
or title_config.get("title")
or ""
).strip()
if not text:
return local_frame
enabled = title_config.get("enabled", True)
if not enabled:
return local_frame
# 优先:前端 Canvas 渲染的 PNG 图层
title_dataurl = title_config.get("title_image_dataurl")
if isinstance(title_dataurl, str) and title_dataurl.startswith("data:image/"):
out_path = _overlay_title_png_on_image(local_frame, title_dataurl)
if out_path:
return out_path
logger.warning("[数字人封面] PNG 叠加失败,回退 drawtext")
# 降级:drawtext 重画文字
try:
from packages.domain.video_filter_builder import build_title_drawtext_filter
drawtext_filter = build_title_drawtext_filter(
title_config,
output_width=720,
output_height=1280,
)
if not drawtext_filter:
return local_frame
base, ext = os.path.splitext(local_frame)
titled_path = f"{base}_titled{ext or '.jpg'}"
cmd = [
"ffmpeg",
"-i",
local_frame,
"-vf",
drawtext_filter,
"-y",
titled_path,
]
logger.info("[数字人封面] drawtext 叠加标题: text=%s", text[:30])
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=30,
)
if result.returncode != 0:
logger.warning(
"[数字人封面] drawtext 失败,回退无标题: exit=%s stderr=%s",
result.returncode,
(result.stderr or "")[-300:],
)
return local_frame
if not os.path.exists(titled_path) or os.path.getsize(titled_path) == 0:
logger.warning("[数字人封面] drawtext 输出为空,回退无标题")
return local_frame
return titled_path
except Exception as exc:
logger.warning("[数字人封面] 标题叠加异常,回退无标题: %s", exc, exc_info=True)
return local_frame
def _overlay_title_png_on_image(local_frame: str, dataurl: str) -> str | None:
"""解码 title PNG dataURL 并用 ffmpeg overlay 叠加到封面图上。
成功返回新文件路径;失败返回 None。
"""
try:
header, b64 = dataurl.split(",", 1)
if "base64" not in header:
return None
png_bytes = base64.b64decode(b64, validate=True)
if not png_bytes:
return None
base_dir = os.path.dirname(local_frame)
title_png = os.path.join(base_dir, f"title_{uuid.uuid4().hex[:8]}.png")
with open(title_png, "wb") as f:
f.write(png_bytes)
base, ext = os.path.splitext(local_frame)
titled_path = f"{base}_titled{ext or '.jpg'}"
cmd = [
"ffmpeg",
"-i",
local_frame,
"-i",
title_png,
"-filter_complex",
"[0:v][1:v]overlay=0:0",
"-y",
titled_path,
]
logger.info("[数字人封面] PNG overlay 叠加标题")
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=30,
)
# 清理临时 PNG
try:
os.unlink(title_png)
except OSError:
pass
if result.returncode != 0:
logger.warning(
"[数字人封面] PNG overlay 失败: exit=%s stderr=%s",
result.returncode,
(result.stderr or "")[-300:],
)
return None
if not os.path.exists(titled_path) or os.path.getsize(titled_path) == 0:
return None
return titled_path
except (binascii.Error, ValueError, OSError) as exc:
logger.warning("[数字人封面] PNG 解码/overlay 异常: %s", exc, exc_info=True)
return None
def persist_cover_to_oss(
frame_url: str,
*,
job_id: str = "",
prefix: str = "ai-avatar/covers",
title_config: dict | None = None,
) -> str:
"""下载帧图并转存到 OSS,返回公网封面 URL.
"""下载最佳帧图并转存到 OSS,返回公网封面 URL(预签名).
Args:
frame_url: MediaKit 返回的临时帧图 URL
job_id: 关联任务 ID(用于 OSS key 命名)
prefix: OSS key 前缀
title_config: 可选标题配置;传入时用 drawtext 叠加标题(竖屏 720x1280)
Returns:
OSS 公网 URL;失败回退原始 frame_url
封面来自最终成片抽帧,帧本身已含标题,本函数不再做任何文字/图片叠加。
"""
if not frame_url:
return ""
tmp_path: Optional[str] = None
titled_path: Optional[str] = None
try:
import httpx
@@ -335,21 +176,12 @@ def persist_cover_to_oss(
token = job_id or uuid.uuid4().hex[:12]
cover_key = f"{prefix}/{token}/cover_{uuid.uuid4().hex[:8]}.jpg"
upload_path = apply_title_to_cover(tmp_path, title_config=title_config)
if upload_path != tmp_path:
titled_path = upload_path
public_url = storage.upload_file(
file_or_path=upload_path,
file_or_path=tmp_path,
storage_key=cover_key,
content_type="image/jpeg",
)
logger.info(
"[数字人封面] 封面已转存 OSS: key=%s titled=%s",
cover_key,
bool(titled_path),
)
# 私有桶:返回预签名 URL(前端才能加载)
logger.info("[数字人封面] 封面已转存 OSS: key=%s", cover_key)
if public_url:
signed = storage.get_download_url(cover_key, expires_seconds=86400)
return signed
@@ -358,12 +190,11 @@ def persist_cover_to_oss(
logger.warning("[数字人封面] 封面转存 OSS 失败,返回原始 URL", exc_info=True)
return frame_url
finally:
for p in (tmp_path, titled_path):
if p:
try:
Path(p).unlink(missing_ok=True)
except Exception:
pass
if tmp_path:
try:
Path(tmp_path).unlink(missing_ok=True)
except Exception:
pass
def generate_smart_cover(
@@ -371,19 +202,12 @@ def generate_smart_cover(
*,
job_id: str = "",
max_frames: int = 5,
title_config: dict | None = None,
) -> str:
"""一站式:MediaKit 智能抽帧选最佳 → (可选)drawtext 叠加标题 → 转存 OSS.
"""一站式:MediaKit 智能抽帧选最佳 → 转存 OSS。失败返回空字符串。
供独立封面接口与渲染管线复用。失败返回空字符串。
Args:
video_url: 可公网访问的视频 URL
job_id: 关联任务 ID
max_frames: 抽帧数量
title_config: 可选标题配置;传入时在封面上叠加 drawtext 标题(竖屏 720x1280)
封面从最终成片抽帧,不再叠加任何标题(帧本身已含)。
"""
best_frame = select_best_cover_frame(video_url, max_frames=max_frames)
if not best_frame:
return ""
return persist_cover_to_oss(best_frame, job_id=job_id, title_config=title_config)
return persist_cover_to_oss(best_frame, job_id=job_id)
@@ -58,21 +58,6 @@ export const getLipsyncJob = async (id: string): Promise<LipsyncJob> => {
return response.data
}
/* ── 智能封面(MediaKit 抽帧 + 质量评分选最佳帧 + 可选标题 overlay/drawtext 叠加) ── */
export const generateSmartCover = async (
video_url: string,
title_config?: Record<string, unknown> | null,
max_frames = 5,
): Promise<{ cover_url: string; status: string; message: string }> => {
const response = await apiClient.post<{ cover_url: string; status: string; message: string }>(
"/ai-avatar/render/smart-cover",
{ video_url, max_frames, title_config: title_config ?? null },
// smart-cover 链路:下载视频+抽帧+overlay/drawtext 加标题+上传 OSS,需要较长时间,120s 超时
{ timeout: 120000 },
)
return response.data
}
/* ── 渲染 ── */
export const submitRender = async (data: {
lipsync_job_id: string