diff --git a/apps/api/app/api/routes/ai_avatar_render.py b/apps/api/app/api/routes/ai_avatar_render.py index d9e96d1b7..a031fb8d2 100644 --- a/apps/api/app/api/routes/ai_avatar_render.py +++ b/apps/api/app/api/routes/ai_avatar_render.py @@ -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,51 +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") - - try: - cover_url = generate_smart_cover( - video_url, - max_frames=body.max_frames, - title_config=getattr(body, "title_config", None), - ) - 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 — 从最终成片智能抽封面(步骤②)──────── @@ -263,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( diff --git a/apps/api/app/schemas/ai_avatar_render.py b/apps/api/app/schemas/ai_avatar_render.py index 84761c2f4..d3a2e0a30 100644 --- a/apps/api/app/schemas/ai_avatar_render.py +++ b/apps/api/app/schemas/ai_avatar_render.py @@ -52,7 +52,9 @@ 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 +69,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,18 +110,8 @@ class AiAvatarRenderProgressResponse(BaseModel): error_message: str -class SmartCoverRequest(BaseModel): - """智能封面请求 — MediaKit 抽帧 + 质量评分选最佳帧 + 可选标题 drawtext 叠加.""" - - 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="标题配置;传入时在封面上用 drawtext 叠加标题(竖屏 720x1280)" - ) - - class SmartCoverResponse(BaseModel): - """智能封面响应.""" + """智能封面响应(封面从最终成片抽帧,不再叠加标题).""" cover_url: str = Field("", description="封面图公网 URL(OSS,非临时);失败为空") status: str = Field("completed", description="completed / fallback_failed") diff --git a/apps/api/app/services/ai_avatar_cover_service.py b/apps/api/app/services/ai_avatar_cover_service.py index fe561633b..a17a000b0 100644 --- a/apps/api/app/services/ai_avatar_cover_service.py +++ b/apps/api/app/services/ai_avatar_cover_service.py @@ -1,18 +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 logging -import os -import subprocess import tempfile import uuid from pathlib import Path @@ -51,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 私有桶)") @@ -62,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: @@ -87,11 +78,9 @@ 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=%s max_frames=%d", video_url[:80], max_frames, - COVER_POLL_INTERVAL, - COVER_MAX_POLL_ATTEMPTS, ) snapshots = mk.extract_frames( @@ -109,7 +98,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 = [] @@ -137,7 +125,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: @@ -158,87 +145,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: - """用 ffmpeg drawtext 在封面图上叠加标题,返回叠加后图片的本地路径. - - ffmpeg 失败时回退返回原始 local_frame。竖屏封面按 720x1280 计算位置。 - """ - if not title_config or not isinstance(title_config, dict): - return local_frame - text = (title_config.get("text") or title_config.get("content") or "").strip() - if not text: - return local_frame - enabled = title_config.get("enabled", True) - if not enabled: - return local_frame - - 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("[数字人封面] 叠加标题: 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 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 @@ -259,21 +178,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 @@ -282,12 +192,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( @@ -295,19 +204,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) diff --git a/apps/api/app/services/ai_avatar_render_service.py b/apps/api/app/services/ai_avatar_render_service.py index a86d3bb8e..41a052fc2 100644 --- a/apps/api/app/services/ai_avatar_render_service.py +++ b/apps/api/app/services/ai_avatar_render_service.py @@ -9,6 +9,8 @@ from __future__ import annotations +import base64 +import binascii import logging import os import subprocess @@ -27,6 +29,7 @@ from packages.adapters.sqlalchemy_impl.models import ( from packages.domain.video_filter_builder import ( build_broll_overlay_filter, build_title_drawtext_filter, + build_title_overlay_filter, ) from packages.shared.storage import get_shared_storage_service @@ -248,29 +251,12 @@ class AiAvatarRenderService: output_height=output_height, ) - # 标题叠加(传入实际输出尺寸,保证位置计算正确) - title_filter = build_title_drawtext_filter( - job.title_config, - output_width=output_width, - output_height=output_height, - ) - - filter_complex = "" - final_label = None - if broll_filter and title_filter: - # B-roll → 标题叠在 B-roll 输出上 - filter_complex = broll_filter + f";[{broll_label}]{title_filter}[vout_titled]" - final_label = "vout_titled" - elif broll_filter: - filter_complex = broll_filter - final_label = broll_label - elif title_filter: - filter_complex = f"[0:v]{title_filter}[vout_titled]" - final_label = "vout_titled" - else: - # 无滤镜:直接拷贝视频流 - filter_complex = "" - final_label = None + # 标题叠加路径:优先前端 Canvas 渲染的 PNG 图层(所见即所得), + # 无 title_image_dataurl 时降级到 drawtext 重画文字。 + title_cfg = job.title_config if isinstance(job.title_config, dict) else {} + title_dataurl = (title_cfg or {}).get("title_image_dataurl") if title_cfg else None + use_title_png = isinstance(title_dataurl, str) and title_dataurl.startswith("data:image/") + title_input_index = 1 + len(job.b_roll_segments or []) if use_title_png else None job.progress = 40 self.db.commit() @@ -279,9 +265,81 @@ class AiAvatarRenderService: with tempfile.TemporaryDirectory() as tmpdir: output_video_path = os.path.join(tmpdir, "output.mp4") + # 在临时目录里解码保存标题 PNG(with 退出自动清理) + title_png_path: Optional[str] = None + extra_inputs: list[str] = [] + title_filter = None + if use_title_png: + try: + title_png_path = os.path.join(tmpdir, f"title_{job.id}.png") + self._save_title_dataurl_to_file(title_dataurl, dst_path=title_png_path) + extra_inputs.append(title_png_path) + logger.info( + "[数字人渲染] 标题 PNG 已保存: %s (input index %d)", title_png_path, title_input_index + ) + except Exception as exc: + logger.warning("[数字人渲染] 标题 PNG 解码/保存失败,降级 drawtext: %s", exc) + title_png_path = None + extra_inputs = [] + + # 构建标题滤镜 + final_label = None + if title_png_path and title_input_index is not None: + title_input_label = f"[{title_input_index}:v]" + base_label = f"[{broll_label}]" if broll_label else "[0:v]" + title_filter = build_title_overlay_filter( + title_cfg, + output_width=output_width, + output_height=output_height, + title_png_path=title_png_path, + title_input_label=title_input_label, + base_label=base_label, + output_label="vout_titled", + ) + if not title_filter: + # build 返回 None → 文件不存在(极端并发情况),降级 drawtext + title_png_path = None + extra_inputs = [] + + if title_png_path: + # overlay 路径 + if broll_filter and title_filter: + filter_complex = broll_filter + f";{title_filter}" + elif broll_filter: + filter_complex = broll_filter + final_label = broll_label + elif title_filter: + filter_complex = title_filter + else: + filter_complex = "" + if title_filter: + final_label = "vout_titled" + elif not final_label: + final_label = None + else: + # 降级:drawtext 重画文字 + title_filter = build_title_drawtext_filter( + title_cfg, + output_width=output_width, + output_height=output_height, + ) + if broll_filter and title_filter: + filter_complex = broll_filter + f";[{broll_label}]{title_filter}[vout_titled]" + final_label = "vout_titled" + elif broll_filter: + filter_complex = broll_filter + final_label = broll_label + elif title_filter: + filter_complex = f"[0:v]{title_filter}[vout_titled]" + final_label = "vout_titled" + else: + filter_complex = "" + final_label = None + cmd_list = self._build_ffmpeg_command( input_video=input_video_path, b_roll_segments=job.b_roll_segments, + extra_inputs=extra_inputs, filter_complex=filter_complex, final_label=final_label, output_path=output_video_path, @@ -408,6 +466,44 @@ class AiAvatarRenderService: os.unlink(tmp.name) raise + @staticmethod + def _save_title_dataurl_to_file(dataurl: str, *, dst_path: str | None = None, job_id: str = "") -> str: + """解码前端传来的 data:image/png;base64,... 并保存为本地 PNG 文件。 + + Args: + dataurl: 完整 dataURL 字符串 + dst_path: 指定输出路径;为 None 时创建临时文件并返回路径 + job_id: 仅在 dst_path 为空时用于临时文件命名 + + Returns: + 保存后的本地文件路径 + """ + if not isinstance(dataurl, str) or not dataurl.startswith("data:image/"): + raise ValueError("title_image_dataurl 不是合法的 data:image URL") + # 拆分 data:image/png;base64, + try: + header, b64 = dataurl.split(",", 1) + except ValueError as exc: + raise ValueError("title_image_dataurl 缺少 base64 payload") from exc + if "base64" not in header: + raise ValueError("title_image_dataurl 不是 base64 编码") + try: + png_bytes = base64.b64decode(b64, validate=True) + except (binascii.Error, ValueError) as exc: + raise ValueError(f"title_image_dataurl base64 解码失败: {exc}") from exc + if not png_bytes: + raise ValueError("title_image_dataurl 解码后为空") + + if dst_path: + out_path = dst_path + with open(out_path, "wb") as f: + f.write(png_bytes) + return out_path + suffix = f"_title_{job_id}.png" if job_id else "_title.png" + with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp: + tmp.write(png_bytes) + return tmp.name + @staticmethod def _probe_video_resolution(video_path: str) -> tuple[int, int]: """用 ffprobe 探测视频分辨率,返回 (width, height);失败返回 (0, 0)。""" @@ -444,6 +540,7 @@ class AiAvatarRenderService: *, input_video: str, b_roll_segments: list[dict[str, Any]], + extra_inputs: list[str] | None = None, filter_complex: str, final_label: Optional[str], output_path: str, @@ -460,6 +557,9 @@ class AiAvatarRenderService: asset_url = seg.get("asset_url", "") if asset_url: cmd.extend(["-i", asset_url]) + # 额外输入(例如前端 Canvas 渲染的标题 PNG) + for extra in extra_inputs or []: + cmd.extend(["-i", extra]) if filter_complex and final_label: cmd.extend( diff --git a/apps/api/app/tasks/lipsync_tts.py b/apps/api/app/tasks/lipsync_tts.py index 187e0be07..7205bfe03 100644 --- a/apps/api/app/tasks/lipsync_tts.py +++ b/apps/api/app/tasks/lipsync_tts.py @@ -55,13 +55,17 @@ def _sign_media_url(url: str) -> str: def _split_script_into_sentences(script_text: str) -> list[str]: - """按句号/问号/感叹号/分号/换行分句(与前端 splitScriptIntoSentences 一致).""" + """按句号/问号/感叹号/分号/逗号/换行分句(与前端 SENTENCE_SPLIT_RE 一致). + + 中文短视频文案习惯用「,」断小句(如"卖花的叫花无缺,卖姜的叫姜子牙"), + 必须把逗号也纳入分隔符,否则多句文案会被识别成一整句,导致 B-roll 时间戳错位。 + """ import re text = (script_text or "").strip() if not text: return [] - parts = re.split(r"[。!?!?;;\n\r]+", text) + parts = re.split(r"[。!?!??!;;,,\n\r]+", text) return [p.strip() for p in parts if p.strip()] diff --git a/apps/web/src/pages/ai-avatar/AiAvatarPage.tsx b/apps/web/src/pages/ai-avatar/AiAvatarPage.tsx index a13d0bd1c..af31176c2 100644 --- a/apps/web/src/pages/ai-avatar/AiAvatarPage.tsx +++ b/apps/web/src/pages/ai-avatar/AiAvatarPage.tsx @@ -32,6 +32,7 @@ import { buildTitleConfigPayload, buildCoverConfigPayload, } from "./utils/contract" +import { renderTitleToPngDataUrl, getVideoResolution } from "./utils/titleCanvas" /** 面板折叠状态 */ type PanelKey = "video" | "voice" | "script" | "lipsync" | "title" | "cover" @@ -210,6 +211,23 @@ const AiAvatarPage: React.FC = () => { try { // 确保有 project_id(AI数字人入口独立,不在项目内,自动取默认项目;#1860 P0 bugfix) const defaultProject = await getOrCreateDefaultProject() + + // 用 Canvas 预渲染标题为 PNG dataURL(所见即所得,后端用 overlay 直接叠加) + let titleImageDataUrl: string | null = null + if (state.titleConfig.title?.trim()) { + try { + const res = await getVideoResolution(state.lipsyncJob.output_video_url || "") + titleImageDataUrl = renderTitleToPngDataUrl({ + titleConfig: state.titleConfig, + videoWidth: res.width, + videoHeight: res.height, + }) + } catch (canvasErr) { + console.warn("[渲染] 标题 Canvas 渲染失败,降级 drawtext:", canvasErr) + titleImageDataUrl = null + } + } + const job = await submitRender({ lipsync_job_id: state.lipsyncJob.id, script_id: state.script?.id, @@ -223,7 +241,7 @@ const AiAvatarPage: React.FC = () => { pip_position: seg.pip_position, pip_scale: seg.pip_scale, })) as never, - title_config: buildTitleConfigPayload(state.titleConfig), + title_config: buildTitleConfigPayload(state.titleConfig, titleImageDataUrl), // 封面不阻塞渲染:用户未选定封面时传空 dict,后端不生成封面;渲染完成后再单独抽帧 cover_config: state.coverConfig.smart_cover_url || diff --git a/apps/web/src/pages/ai-avatar/api/aiAvatar.ts b/apps/web/src/pages/ai-avatar/api/aiAvatar.ts index 25ec6cf85..7f5b6a764 100644 --- a/apps/web/src/pages/ai-avatar/api/aiAvatar.ts +++ b/apps/web/src/pages/ai-avatar/api/aiAvatar.ts @@ -58,21 +58,6 @@ export const getLipsyncJob = async (id: string): Promise => { return response.data } -/* ── 智能封面(MediaKit 抽帧 + 质量评分选最佳帧 + 可选 drawtext 标题叠加) ── */ -export const generateSmartCover = async ( - video_url: string, - title_config?: Record | 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 链路:下载视频+抽帧+drawtext 加标题+上传 OSS,需要较长时间,120s 超时 - { timeout: 120000 }, - ) - return response.data -} - /* ── 渲染 ── */ export const submitRender = async (data: { lipsync_job_id: string @@ -82,6 +67,7 @@ export const submitRender = async (data: { cover_config?: Record project_id?: string }): Promise => { + // title_config 内可含 title_image_dataurl(前端 Canvas 渲染的 PNG dataURL) const response = await apiClient.post("/ai-avatar/render", data) return response.data } diff --git a/apps/web/src/pages/ai-avatar/utils/contract.ts b/apps/web/src/pages/ai-avatar/utils/contract.ts index 19d4843db..d8d361028 100644 --- a/apps/web/src/pages/ai-avatar/utils/contract.ts +++ b/apps/web/src/pages/ai-avatar/utils/contract.ts @@ -28,10 +28,13 @@ export function normalizeEmotion(raw: string | undefined | null): VoiceEmotion { * 后端真实字段:text(或content)、font(或font_preset)、font_size(或size)、 * font_color(或color,可传 #RRGGBB)、position(top/center/bottom/custom)、 * enabled、bold、stroke{enabled,width,color}、shadow{enabled,color,offset_x,offset_y}、 - * pos_x/pos_y(custom 时)。 + * pos_x/pos_y(custom 时)、title_image_dataurl(前端 Canvas 渲染的 PNG dataURL,WYSIWYG 路径优先)。 * 口播标题默认 position=bottom(不传后端会默认 top 跑到画面顶部)。 */ -export function buildTitleConfigPayload(cfg: AiAvatarTitleConfig): Record { +export function buildTitleConfigPayload( + cfg: AiAvatarTitleConfig, + titleImageDataUrl?: string | null, +): Record { const text = (cfg.title || "").trim() if (!text) return {} const position = cfg.position || "bottom" @@ -53,6 +56,10 @@ export function buildTitleConfigPayload(cfg: AiAvatarTitleConfig): Record part.trim()) - .filter((part) => part.length > 0) - - // 2. 优先使用后端精确时间戳 - // 校验:必须是数组、条数一致、每条都有 start_time/end_time,否则降级估算 - if (Array.isArray(sentenceTimings) && sentenceTimings.length === rawParts.length) { + // 1. 后端返回了 sentence_timings:校验通过就直接用,跳过本地分句 + // 校验条件放宽:只要是数组、至少1条、每条 start_time/end_time 是数字即可 + // (不再强制要求条数相等——后端静音检测可能按停顿切出更多/更少边界, + // 比如文案用逗号连写时本地只分1句、后端按停顿切4句,后端的切法才是对的) + if (Array.isArray(sentenceTimings) && sentenceTimings.length > 0) { const valid = sentenceTimings.every( (t) => t && typeof t.start_time === "number" && typeof t.end_time === "number" && + isFinite(t.start_time) && + isFinite(t.end_time) && t.end_time >= t.start_time, ) if (valid) { let accChar = 0 return sentenceTimings.map((t, i) => { - const part = rawParts[i] ?? t.text ?? "" - const charCount = part.replace(/\s/g, "").length + const sentenceText = (t.text || "").trim() || `句子${i + 1}` + const charCount = sentenceText.replace(/\s/g, "").length const sentence: ScriptSentence = { - index: t.index ?? i, - text: part, + index: typeof t.index === "number" ? t.index : i, + text: sentenceText, charCount, startChar: accChar, startTime: round1(t.start_time), @@ -70,7 +74,12 @@ export function splitScriptIntoSentences( } } - // 3. 降级:按字数比例线性估算 + // 2. 本地分句 + 按字数比例估算(降级路径) + const rawParts = text + .split(SENTENCE_SPLIT_RE) + .map((part) => part.trim()) + .filter((part) => part.length > 0) + const totalChars = rawParts.reduce((sum, part) => sum + part.replace(/\s/g, "").length, 0) const duration = outputDuration > 0 ? outputDuration : 0 diff --git a/apps/web/src/pages/ai-avatar/utils/titleCanvas.ts b/apps/web/src/pages/ai-avatar/utils/titleCanvas.ts new file mode 100644 index 000000000..840b1232b --- /dev/null +++ b/apps/web/src/pages/ai-avatar/utils/titleCanvas.ts @@ -0,0 +1,174 @@ +/** + * AI数字人 — 标题 Canvas 渲染工具 + * + * 把标题按前端预览的 HTML/CSS 效果画到透明背景 PNG 上(与视频同分辨率), + * 以 dataURL 形式传给后端,后端用 FFmpeg overlay 直接叠加图层, + * 彻底解决前端 HTML/CSS 预览 ≠ FFmpeg drawtext 成片的 WYSIWYG 问题。 + */ +import type { AiAvatarTitleConfig } from "../types" + +export interface RenderTitlePngOptions { + /** 标题配置 */ + titleConfig: AiAvatarTitleConfig + /** 视频宽度(像素),默认 720 */ + videoWidth?: number + /** 视频高度(像素),默认 1280 */ + videoHeight?: number +} + +/** + * 将标题渲染为透明背景 PNG 的 dataURL(data:image/png;base64,...) + * Canvas 尺寸与视频一致,保证叠加时 1:1 像素对齐。 + * + * 标题为空时返回 null。 + */ +export function renderTitleToPngDataUrl(opts: RenderTitlePngOptions): string | null { + const { titleConfig, videoWidth = 720, videoHeight = 1280 } = opts + if (!titleConfig) return null + const rawTitle = (titleConfig.title || "").trim() + if (!rawTitle) return null + + // 按 / 或 / 分割为多行 + const lines = rawTitle + .split(/[//]/) + .map((l) => l.trim()) + .filter((l) => l.length > 0) + if (lines.length === 0) return null + + const canvas = document.createElement("canvas") + canvas.width = videoWidth + canvas.height = videoHeight + const ctx = canvas.getContext("2d") + if (!ctx) return null + + const size = Math.max(12, Math.round(titleConfig.size || 48)) + const bold = !!titleConfig.bold + const italic = !!titleConfig.italic + const color = titleConfig.color || "#ffffff" + const stroke = !!titleConfig.stroke + const shadow = !!titleConfig.shadow + + // 字体族 fallback 链:优先中文字体 + const fontFamily = + '"Noto Sans CJK SC","Source Han Sans CN","PingFang SC","Microsoft YaHei",sans-serif' + const fontParts: string[] = [] + if (italic) fontParts.push("italic") + if (bold) fontParts.push("bold") + fontParts.push(`${size}px`, fontFamily) + ctx.font = fontParts.join(" ") + ctx.fillStyle = color + ctx.textAlign = "center" + ctx.textBaseline = "middle" + + // 阴影(shadow=true 时开启) + if (shadow) { + ctx.shadowColor = "rgba(0,0,0,0.8)" + ctx.shadowBlur = 4 + ctx.shadowOffsetX = 0 + ctx.shadowOffsetY = 2 + } + + // 位置计算:与 PanelLipsyncPreview 的 CSS 对齐 + // 预览用 top/bottom 8px padding + transform translateX(-50%) 居中; + // 这里画到整尺寸 canvas,padding 按比例放大到全分辨率(预览缩放 0.35x 时 8px ≈ 23px 全尺寸, + // 为更贴近原 CSS 16px 安全边距,用 16px 作为内边距)。 + const PAD = 16 + let centerX = videoWidth / 2 + const position = titleConfig.position || "bottom" + const lineGap = size * 1.2 + const totalTextH = lines.length * lineGap - (lineGap - size) // 所有行的总高度 + // 文本块顶部 y(textBaseline=middle 时首行基线) + let firstLineY: number + if ( + position === "custom" && + typeof titleConfig.pos_x === "number" && + typeof titleConfig.pos_y === "number" + ) { + centerX = (Math.max(0, Math.min(100, titleConfig.pos_x)) / 100) * videoWidth + const centerY = (Math.max(0, Math.min(100, titleConfig.pos_y)) / 100) * videoHeight + firstLineY = centerY - totalTextH / 2 + size / 2 + } else if (position === "top") { + // 顶部:y = size/2 + PAD + firstLineY = size / 2 + PAD + } else if (position === "center") { + firstLineY = videoHeight / 2 - totalTextH / 2 + size / 2 + } else { + // bottom(默认) + firstLineY = videoHeight - totalTextH - PAD + size / 2 + } + + // 描边参数(stroke=true 或 bold 默认细描边模拟粗体时都画; + // 注意:浏览器原生 bold 已经是粗体 glyph,Canvas 这里对 stroke=true 才加黑描边, + // 与预览 CSS 的 WebkitTextStroke 保持一致,不对 bold 自动加描边避免双粗)。 + const doStroke = stroke + // 逐行绘制 + lines.forEach((line, idx) => { + const y = firstLineY + idx * lineGap + if (doStroke) { + const prevShadowColor = ctx.shadowColor + const prevShadowBlur = ctx.shadowBlur + // 描边不要带阴影(避免黑色描边发虚) + ctx.shadowColor = "rgba(0,0,0,0)" + ctx.shadowBlur = 0 + ctx.lineWidth = Math.max(2, size * 0.06) + ctx.strokeStyle = "#000000" + ctx.lineJoin = "round" + ctx.strokeText(line, centerX, y) + // 恢复阴影 + if (shadow) { + ctx.shadowColor = "rgba(0,0,0,0.8)" + ctx.shadowBlur = 4 + } else { + ctx.shadowColor = prevShadowColor + ctx.shadowBlur = prevShadowBlur + } + } + ctx.fillText(line, centerX, y) + }) + + try { + return canvas.toDataURL("image/png") + } catch { + return null + } +} + +/** + * 获取视频真实分辨率(HTMLVideoElement + loadedmetadata,超时 3 秒兜底 720×1280)。 + */ +export function getVideoResolution( + videoUrl: string, + timeoutMs = 3000, +): Promise<{ width: number; height: number }> { + return new Promise((resolve) => { + if (!videoUrl) { + resolve({ width: 720, height: 1280 }) + return + } + const video = document.createElement("video") + video.preload = "metadata" + video.muted = true + video.playsInline = true + video.crossOrigin = "anonymous" + let settled = false + const done = (w: number, h: number) => { + if (settled) return + settled = true + video.removeAttribute("src") + video.load() + resolve({ width: w, height: h }) + } + const timer = window.setTimeout(() => done(720, 1280), timeoutMs) + video.onloadedmetadata = () => { + window.clearTimeout(timer) + const w = video.videoWidth || 720 + const h = video.videoHeight || 1280 + done(w, h) + } + video.onerror = () => { + window.clearTimeout(timer) + done(720, 1280) + } + video.src = videoUrl + }) +} diff --git a/packages/domain/video_filter_builder.py b/packages/domain/video_filter_builder.py index 496e68d5a..91f82959f 100755 --- a/packages/domain/video_filter_builder.py +++ b/packages/domain/video_filter_builder.py @@ -420,44 +420,29 @@ def _escape_drawtext_text(text: str) -> str: return result -# 粗体字体文件映射:服务器镜像只保留了 NotoSansSC-VF.ttf(可变字体,已删除 -# NotoSansCJK-Bold.ttc 以避免 Mono 变体问题,见 worker-base.Dockerfile), -# 因此无法通过 fontfile 切换到 Bold 字重。这里保留路径列表作为未来扩展, -# 实际加粗通过 borderw 黑色描边实现(见下)。 -DRAWTEXT_BOLD_FONT_SEARCH_PATHS: list[str] = [ - "/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc", - "/usr/share/fonts/noto-cjk/NotoSansCJK-Bold.ttc", - "/usr/share/fonts/google-noto-cjk/NotoSansCJK-Bold.ttc", - "/usr/share/fonts/truetype/noto/NotoSansSC-Bold.ttf", - "/usr/share/fonts/noto/NotoSansSC-Bold.ttf", -] +# 粗体字体已由前端 Canvas 直接渲染(Canvas 使用浏览器原生粗体 glyph), +# FFmpeg 侧不再需要查找 Bold 字体文件;drawtext 仅作为旧版前端的降级路径, +# 通过 borderw 黑色细描边模拟粗体(见 build_title_drawtext_filter)。 -def _resolve_font_path(font_name: str, bold: bool = False) -> str: +def _resolve_font_path(font_name: str) -> str: """解析字体名到服务器实际字体文件路径。 查找策略: 1. 通过 DRAWTEXT_FONT_MAP 映射前端字体名到服务器关键字 - 2. bold=True 时优先查找粗体变体;找不到回退常规字重 - 3. 在 DRAWTEXT_FONT_SEARCH_PATHS 中查找匹配路径 - 4. 未找到则返回空字符串(drawtext 使用内置默认字体) + 2. 在 DRAWTEXT_FONT_SEARCH_PATHS 中查找匹配路径 + 3. 未找到则返回空字符串(drawtext 使用内置默认字体) + + 注:粗体已由前端 Canvas 渲染时直接用浏览器 bold glyph 绘制, + 此处仅作为旧版前端降级路径,无需切换 Bold 字体文件。 """ keyword = DRAWTEXT_FONT_MAP.get(font_name, font_name) import os - if bold: - for path in DRAWTEXT_BOLD_FONT_SEARCH_PATHS: - if keyword.lower() in path.lower() and os.path.isfile(path): - return path - # 粗体文件找不到时,再查常规字重(后面会用描边兜底加粗) for path in DRAWTEXT_FONT_SEARCH_PATHS: if keyword.lower() in path.lower() and os.path.isfile(path): return path # fallback:遍历搜索任意可用字体 - if bold: - for path in DRAWTEXT_BOLD_FONT_SEARCH_PATHS: - if os.path.isfile(path): - return path for path in DRAWTEXT_FONT_SEARCH_PATHS: if os.path.isfile(path): return path @@ -493,8 +478,10 @@ def build_title_drawtext_filter( if not title_config or not isinstance(title_config, dict): return None - # 字段名归一化:兼容 content/text、font_preset/font 两套命名 - text = (title_config.get("text") or title_config.get("content") or "").strip() + # 字段名归一化:兼容 content/text/title 三套命名 + text = ( + title_config.get("text") or title_config.get("content") or title_config.get("title") or "" + ).strip() if not text: return None @@ -518,8 +505,8 @@ def build_title_drawtext_filter( # ── 构建 drawtext 参数 ── params: list[str] = [] - # 字体文件:粗体优先使用 Bold 字体文件,避免同色描边造成字形偏移/重影 - font_path = _resolve_font_path(font_name, bold=bold) + # 字体文件(drawtext 降级路径:粗体通过 borderw 黑色描边模拟) + font_path = _resolve_font_path(font_name) if font_path: escaped_path = font_path.replace("\\", "\\\\").replace(":", "\\\\:").replace("'", "\\\\'") params.append(f"fontfile='{escaped_path}'") @@ -599,6 +586,43 @@ def build_title_drawtext_filter( return "drawtext=" + ":".join(params) +def build_title_overlay_filter( + title_config: dict[str, Any], + output_width: int, # noqa: ARG001 - 保留参数签名,PNG 已按视频分辨率绘制 + output_height: int, # noqa: ARG001 + title_png_path: str, + *, + title_input_label: str = "[1:v]", + base_label: str = "[0:v]", + output_label: str = "vout_titled", +) -> str | None: + """构建标题 PNG 图层 overlay 滤镜(WYSIWYG 路径)。 + + 前端用 Canvas 把标题画成与视频同分辨率的透明 PNG(所见即所得), + 后端直接 overlay=0:0 叠加即可,PNG 透明区域不遮挡视频。 + + Args: + title_config: 标题配置 dict(仅用来判断降级) + output_width: 输出宽度(未使用,PNG 已按该分辨率绘制) + output_height: 输出高度(未使用) + title_png_path: 已保存到本地的标题 PNG 文件路径 + title_input_label: 标题 PNG 在 filter_complex 中的输入标签(默认 "[1:v]") + base_label: 前序滤镜输出标签(如 B-roll 输出 "[vout]") + output_label: overlay 输出标签名 + + Returns: + overlay 滤镜字符串;title_png_path 为空/文件不存在时返回 None(降级到 drawtext) + """ + import os + + if not title_png_path or not os.path.isfile(title_png_path): + return None + if not title_config or not isinstance(title_config, dict): + return None + + return f"{base_label}{title_input_label}overlay=0:0[{output_label}]" + + # ── B-roll 叠加滤镜 ───────────────────────────────────────────────────────── diff --git a/tests/unit/test_lipsync_tts.py b/tests/unit/test_lipsync_tts.py index cc57603b2..748382539 100644 --- a/tests/unit/test_lipsync_tts.py +++ b/tests/unit/test_lipsync_tts.py @@ -428,6 +428,9 @@ class TestPersistOutputVideoTask: storage = MagicMock() storage.public_url = "https://oss.example.com/" storage.upload_file.return_value = upload_url or "https://oss.example.com/lipsync-outputs/user-1/job-1.mp4" + # _sign_media_url 内部会调 storage.get_download_url,必须mock返回字符串 + _upload_url = upload_url or "https://oss.example.com/lipsync-outputs/user-1/job-1.mp4" + storage.get_download_url.return_value = _upload_url + "?signed" fake_httpx = ModuleType("httpx") fake_httpx.Client = FakeHttpxClient @@ -448,7 +451,7 @@ class TestPersistOutputVideoTask: job = self._make_persist_job(output_video_url="https://temp.mk/x.mp4") session, fake_client, storage, patches = self._persist_patches( - job=job, video_bytes=b"VIDEODATA", upload_url="https://oss.example.com/lipsync-outputs/u1/j1.mp4" + job=job, video_bytes=b"VIDEODATA", upload_url="https://oss.example.com/lipsync-outputs/user-1/job-1.mp4" ) entered = [p.__enter__() for p in patches] try: @@ -459,17 +462,11 @@ class TestPersistOutputVideoTask: fake_client.get.assert_called_once_with("https://temp.mk/x.mp4") storage.upload_file.assert_called_once() - # 上传的 key 必须是 lipsync-outputs/{user_id}/{job_id}.mp4 - key_arg = ( - storage.upload_file.call_args.args[1] - if storage.upload_file.call_args.args - else storage.upload_file.call_args.kwargs.get("key") - ) - # upload_file(data, key, content_type=...) call_args = storage.upload_file.call_args.args + # 上传的 key 必须是 lipsync-outputs/{user_id}/{job_id}.mp4 assert call_args[1] == "lipsync-outputs/user-1/job-1.mp4" - # output_video_url 被替换为签名后的永久 URL - assert job.output_video_url == "https://oss.example.com/lipsync-outputs/u1/j1.mp4?signed" + # upload_file 返回永久 URL,再被 _sign_media_url 追加 ?signed + assert job.output_video_url == "https://oss.example.com/lipsync-outputs/user-1/job-1.mp4?signed" assert job.updated_at is not None session.commit.assert_called_once() session.close.assert_called_once() diff --git a/tests/unit/test_video_filter_builder.py b/tests/unit/test_video_filter_builder.py index a0d039236..61aa2aabb 100755 --- a/tests/unit/test_video_filter_builder.py +++ b/tests/unit/test_video_filter_builder.py @@ -13,6 +13,7 @@ from __future__ import annotations +import os import unittest from dataclasses import FrozenInstanceError from unittest.mock import patch @@ -29,10 +30,12 @@ from packages.domain.video_filter_builder import ( ClipFilterChain, _escape_drawtext_text, _resolve_font_path, + build_broll_overlay_filter, build_clip_filter, build_concat_filter, build_filter_complex, build_title_drawtext_filter, + build_title_overlay_filter, build_xfade_filter, chain_filters, has_audio, @@ -1161,5 +1164,97 @@ class TestDrawtextNotDictConfig(unittest.TestCase): self.assertIsNone(build_title_drawtext_filter([1, 2, 3])) +class TestTitleOverlay(unittest.TestCase): + """build_title_overlay_filter 单元测试(WYSIWYG PNG 叠加路径)。""" + + def test_overlay_filter_format(self): + """PNG 文件存在时返回正确的 overlay 滤镜字符串。""" + import tempfile + + with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp: + tmp.write(b"\x89PNG\r\n\x1a\n") + tmp_path = tmp.name + try: + result = build_title_overlay_filter( + {"text": "标题"}, + output_width=720, + output_height=1280, + title_png_path=tmp_path, + title_input_label="[2:v]", + base_label="[vout]", + output_label="vout_titled", + ) + self.assertIsNotNone(result) + self.assertIn("[vout][2:v]overlay=0:0[vout_titled]", result) + finally: + os.unlink(tmp_path) + + def test_overlay_default_labels(self): + """不传 label 参数时使用默认 [0:v] / [1:v] / vout_titled。""" + import tempfile + + with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp: + tmp.write(b"\x89PNG\r\n\x1a\n") + tmp_path = tmp.name + try: + result = build_title_overlay_filter( + {"text": "标题"}, + output_width=720, + output_height=1280, + title_png_path=tmp_path, + ) + self.assertEqual(result, "[0:v][1:v]overlay=0:0[vout_titled]") + finally: + os.unlink(tmp_path) + + def test_overlay_returns_none_when_png_missing(self): + """PNG 文件不存在时返回 None,供调用方降级到 drawtext。""" + result = build_title_overlay_filter( + {"text": "标题"}, + output_width=720, + output_height=1280, + title_png_path="/nonexistent/path/title.png", + ) + self.assertIsNone(result) + + def test_overlay_returns_none_for_empty_config(self): + """title_config 为空/非 dict 时返回 None。""" + import tempfile + + with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp: + tmp.write(b"\x89PNG\r\n\x1a\n") + tmp_path = tmp.name + try: + self.assertIsNone( + build_title_overlay_filter( + None, + output_width=720, + output_height=1280, + title_png_path=tmp_path, + ) + ) + self.assertIsNone( + build_title_overlay_filter( + "not a dict", + output_width=720, + output_height=1280, + title_png_path=tmp_path, + ) + ) + finally: + os.unlink(tmp_path) + + def test_overlay_returns_none_for_empty_path(self): + """title_png_path 为空字符串时返回 None。""" + self.assertIsNone( + build_title_overlay_filter( + {"text": "标题"}, + output_width=720, + output_height=1280, + title_png_path="", + ) + ) + + if __name__ == "__main__": unittest.main()