diff --git a/apps/api/app/services/ai_avatar_render_service.py b/apps/api/app/services/ai_avatar_render_service.py index 163a3d5df..bd65653f4 100644 --- a/apps/api/app/services/ai_avatar_render_service.py +++ b/apps/api/app/services/ai_avatar_render_service.py @@ -25,6 +25,7 @@ from packages.adapters.sqlalchemy_impl.models import ( ScriptModel, ) from packages.domain.video_filter_builder import ( + build_broll_overlay_filter, build_title_drawtext_filter, ) from packages.shared.storage import get_shared_storage_service @@ -228,27 +229,49 @@ class AiAvatarRenderService: self.db.commit() # 2. 构建 FFmpeg 滤镜链 (40%) - from packages.domain.video_filter_builder import build_broll_overlay_filter + # 用 ffprobe 探测输入视频分辨率，确保 B-roll 缩放与标题位置与实际输出一致。 + # AI 数字人对口型输出为 9:16 竖屏，默认兜底 720x1280；探测失败时使用默认值不阻断渲染。 + output_width, output_height = self._probe_video_resolution(input_video_path) + if output_width <= 0 or output_height <= 0: + output_width, output_height = 720, 1280 + logger.info( + "[数字人渲染] ffprobe 探测分辨率失败或无效，使用默认竖屏尺寸 %sx%s", + output_width, + output_height, + ) + else: + logger.info("[数字人渲染] 探测输入视频分辨率: %sx%s", output_width, output_height) - filter_complex = build_broll_overlay_filter( + broll_filter, broll_label = build_broll_overlay_filter( b_roll_segments=job.b_roll_segments, video_duration=lipsync_job.output_duration, + output_width=output_width, + output_height=output_height, ) - # 标题叠加 - title_filter = build_title_drawtext_filter(job.title_config) - if title_filter: - if filter_complex: - filter_complex += f"[vout]{title_filter}[vout_titled];" - else: - filter_complex = f"[0:v]{title_filter}[vout_titled];" + # 标题叠加（传入实际输出尺寸，保证位置计算正确） + title_filter = build_title_drawtext_filter( + job.title_config, + output_width=output_width, + output_height=output_height, + ) - # 清理末尾分号 - if filter_complex.endswith(";"): - filter_complex = filter_complex[:-1] - - # 最终输出标签 - final_label = "vout_titled" if title_filter else ("vout" if filter_complex else None) + 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 job.progress = 40 self.db.commit() @@ -427,6 +450,37 @@ class AiAvatarRenderService: os.unlink(tmp.name) raise + @staticmethod + def _probe_video_resolution(video_path: str) -> tuple[int, int]: + """用 ffprobe 探测视频分辨率，返回 (width, height)；失败返回 (0, 0)。""" + try: + result = subprocess.run( + [ + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=width,height", + "-of", + "csv=p=0:s=x", + video_path, + ], + capture_output=True, + text=True, + timeout=15, + ) + if result.returncode == 0 and result.stdout.strip(): + parts = result.stdout.strip().split("x") + if len(parts) == 2: + w, h = int(parts[0]), int(parts[1]) + if w > 0 and h > 0: + return w, h + except Exception as exc: + logger.warning("[数字人渲染] ffprobe 探测分辨率失败: %s", exc) + return 0, 0 + def _build_ffmpeg_command( self, *, @@ -450,7 +504,16 @@ class AiAvatarRenderService: cmd.extend(["-i", asset_url]) if filter_complex and final_label: - cmd.extend(["-filter_complex", filter_complex, "-map", f"[{final_label}]"]) + cmd.extend( + [ + "-filter_complex", + filter_complex, + "-map", + f"[{final_label}]", + "-map", + "0:a?", + ] + ) elif filter_complex: cmd.extend(["-filter_complex", filter_complex]) @@ -462,6 +525,10 @@ class AiAvatarRenderService: "veryfast", "-crf", "23", + "-c:a", + "aac", + "-b:a", + "128k", "-y", output_path, ] diff --git a/packages/domain/video_filter_builder.py b/packages/domain/video_filter_builder.py index f7563cb38..57523a545 100755 --- a/packages/domain/video_filter_builder.py +++ b/packages/domain/video_filter_builder.py @@ -377,26 +377,30 @@ def _append_audio_concat(parts: list[str], clip_chains: list[ClipFilterChain]) - # ── 标题 drawtext 滤镜构建（#1789）───────────────────────────────────────────── -# drawtext 字体搜索路径：按优先级列出常见安装位置 -# 服务器使用 Noto Sans SC（思源黑体）作为默认字体 +# drawtext 字体搜索路径：按优先级从高到低排列 +# 服务器使用 Noto Sans SC（思源黑体）作为默认字体。 +# - NotoSansSC-VF.ttf 是 worker-base.Dockerfile 中 COPY 的 VF 字体（含所有字重，无 Mono 变体），优先级最高 +# - .ttc 系列为 fonts-noto-cjk 包预装字体（Dockerfile 已删除含 Mono 变体的旧 .ttc，存在时作为 fallback） +# - DejaVuSans 仅含拉丁字符不支持中文，已移除 DRAWTEXT_FONT_SEARCH_PATHS: list[str] = [ + "/usr/share/fonts/opentype/noto/NotoSansSC-VF.ttf", "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc", "/usr/share/fonts/noto-cjk/NotoSansCJK-Regular.ttc", "/usr/share/fonts/google-noto-cjk/NotoSansCJK-Regular.ttc", "/usr/share/fonts/truetype/noto/NotoSansSC-Regular.ttf", "/usr/share/fonts/noto/NotoSansSC-Regular.ttf", - "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", ] -# 前端字体名 → drawtext 字体搜索关键字 +# 前端字体名 → drawtext 字体搜索关键字（匹配 DRAWTEXT_FONT_SEARCH_PATHS 中的文件名关键字） DRAWTEXT_FONT_MAP: dict[str, str] = { - "思源黑体": "NotoSansCJK", + "思源黑体": "NotoSansSC", "思源宋体": "NotoSerifCJK", - "苹方": "NotoSansCJK", - "PingFang": "NotoSansCJK", - "微软雅黑": "NotoSansCJK", + "苹方": "NotoSansSC", + "PingFang": "NotoSansSC", + "微软雅黑": "NotoSansSC", "楷体": "NotoSerifCJK", - "华康俪金黑": "NotoSansCJK", + "华康俪金黑": "NotoSansSC", } @@ -504,26 +508,30 @@ def build_title_drawtext_filter( params.append(f"fontsize={font_size}") params.append(f"fontcolor={font_color}") - # 粗体：bold 在 drawtext 中通过 font 的 Bold 变体实现 - # 若字体有 Bold 变体可用 fontfont=bold；否则通过 borderw 模拟 - if bold: - # 使用 font 参数尝试加载 Bold 变体（Noto Sans SC 有 Bold 变体文件） - params.append("font=bold") + # 粗体：drawtext 没有独立的 bold 参数，通过加大 borderw 模拟视觉粗体效果。 + # 注意：不能使用 `font=bold`——FFmpeg drawtext 的 font 参数需要 fontconfig 能解析的 + # 字体族名，而 "bold" 不是合法族名，会导致整个 filter_complex 解析失败（exit code 234）。 + # 当用户未显式配置描边宽度时，bold 模式自动将 borderw 提升到 3 以模拟粗体。 # 描边（borderw 需要 libfreetype 支持） + # 粗体无显式描边时，自动用 borderw=3 + 近色描边模拟粗体；显式 stroke 按用户配置走 + border_width = 0 + border_color = "000000" if stroke: if isinstance(stroke, bool): border_width = 2 - border_color = "black" + border_color = "000000" elif isinstance(stroke, dict): - border_width = int(stroke.get("width", 2)) if stroke.get("enabled", True) else 0 - border_color = (stroke.get("color") or "#000000").lstrip("#") - else: - border_width = 0 - border_color = "black" - if border_width > 0: - params.append(f"borderw={border_width}") - params.append(f"bordercolor={border_color}") + if stroke.get("enabled", True): + border_width = int(stroke.get("width", 2)) + border_color = (stroke.get("color") or "#000000").lstrip("#") + elif bold: + # 粗体模式且未配描边：加大描边宽度模拟粗体效果 + border_width = 3 + border_color = font_color # 用字体同色描边，视觉上加粗字形而非黑边 + if border_width > 0: + params.append(f"borderw={border_width}") + params.append(f"bordercolor={border_color}") # 阴影（shadowcolor + shadowx/y） if shadow: @@ -573,7 +581,7 @@ def build_broll_overlay_filter( video_duration: float, output_width: int = DEFAULT_OUTPUT_WIDTH, output_height: int = DEFAULT_OUTPUT_HEIGHT, -) -> str: +) -> tuple[str, str | None]: """构建 B-roll 叠加滤镜链。 支持两种模式: @@ -581,121 +589,181 @@ def build_broll_overlay_filter( - pip: 在对口型视频上叠加画中画 B-roll Args: - b_roll_segments: B-roll 片段配置列表 + b_roll_segments: B-roll 片段配置列表（原始顺序，决定 FFmpeg -i 输入顺序） video_duration: 对口型视频总时长(秒) - output_width: 输出宽度 - output_height: 输出高度 + output_width: 输出宽度（默认 1280；AI 数字人竖屏传 720） + output_height: 输出高度（默认 720；AI 数字人竖屏传 1280） Returns: - FFmpeg filter_complex 滤镜字符串片段 + (filter_complex_str, final_label) + - filter_complex_str: filter_complex 片段字符串（末尾无分号） + - final_label: 最终输出 pad 标签名，如 "vout"；无 B-roll 时返回 None """ if not b_roll_segments: - return "" + return "", None + + # 建立原始列表下标 → FFmpeg 输入下标的映射： + # cmd 中 [0:v] 是主视频，随后按 b_roll_segments 原始顺序追加 -i， + # 因此第 i 个 segment 的输入是 [{i+1}:v] + def _input_label(seg: dict[str, Any]) -> str: + # seg 必须来自 b_roll_segments；通过 id() 在原列表中查找 + for i, s in enumerate(b_roll_segments): + if s is seg: + return f"[{i + 1}:v]" + # fallback: 找不到时不应发生，保守返回 + return "[1:v]" parts: list[str] = [] sorted_segments = sorted(b_roll_segments, key=lambda s: s.get("start_time", 0)) - # 按模式分组处理 + # 按模式分组 fullscreen_segments = [s for s in sorted_segments if s.get("mode") == "fullscreen"] pip_segments = [s for s in sorted_segments if s.get("mode") == "pip"] + final_label = None + # ── fullscreen 模式: 切分 + concat ── if fullscreen_segments: - parts.append(_build_fullscreen_filters(fullscreen_segments, video_duration, output_width, output_height)) + fs_filter, fs_label = _build_fullscreen_filters( + fullscreen_segments, b_roll_segments, video_duration, output_width, output_height, _input_label + ) + parts.append(fs_filter) + final_label = fs_label + else: + fs_label = None # ── pip 模式: overlay 滤镜 ── if pip_segments: - for idx, seg in enumerate(pip_segments): - start = seg.get("start_time", 0) - end = seg.get("end_time", video_duration) - scale = seg.get("pip_scale", 0.3) - position = seg.get("pip_position", "bottom_right") - - pip_w = int(output_width * scale) - pip_h = int(output_height * scale) - - # 位置映射 - pos_map = { - "top_left": "10:10", - "top_right": "W-w-10:10", - "bottom_left": "10:H-h-10", - "bottom_right": "W-w-10:H-h-10", - "center": "(W-w)/2:(H-h)/2", - } - pos_expr = pos_map.get(position, pos_map["bottom_right"]) - - broll_input_idx = len(sorted_segments) # placeholder for input index - parts.append( - f"[{broll_input_idx + idx}:v]scale={pip_w}:{pip_h}," f"enable='between(t,{start},{end})'[pip{idx}];" - ) - # overlay onto main stream - if idx == 0: - base_label = "[vout]" if fullscreen_segments else "[0:v]" - else: - base_label = f"[pip{idx - 1}]" - parts.append(f"{base_label}[pip{idx}]overlay={pos_expr}:enable='between(t,{start},{end})'[vout{idx}];") + pip_filter, pip_label = _build_pip_filters( + pip_segments, output_width, output_height, _input_label, base_label=fs_label + ) + parts.append(pip_filter) + final_label = pip_label result = "".join(parts) # 清理末尾多余分号 if result.endswith(";"): result = result[:-1] - return result + return result, final_label def _build_fullscreen_filters( - segments: list[dict[str, Any]], + sorted_fs_segments: list[dict[str, Any]], + all_segments: list[dict[str, Any]], video_duration: float, output_width: int, output_height: int, -) -> str: - """构建 fullscreen 模式的切分 + concat 滤镜. + input_label_fn, +) -> tuple[str, str]: + """构建 fullscreen 模式的切分 + concat 滤镜。 - 将对口型视频按 B-roll 时间段切分，然后用 concat 拼接 B-roll 片段。 + 将主视频按 B-roll 时间段切分，然后用 concat 拼接主视频片段和 B-roll 片段。 + + Returns: + (filter_str, final_label) 其中 final_label 是 concat 输出的 pad 标签 """ parts: list[str] = [] prev_end = 0.0 - for idx, seg in enumerate(segments): + # 注意：这里的 idx 是 sorted_fs_segments 中的下标； + # 实际 FFmpeg 输入下标必须通过 input_label_fn 查询 + for idx, seg in enumerate(sorted_fs_segments): start = seg.get("start_time", 0) end = seg.get("end_time", video_duration) - # 保持原视频片段（B-roll 之前的部分） + # 主视频片段（B-roll 之前） if prev_end < start: parts.append(f"[0:v]trim=start={prev_end}:end={start},setpts=PTS-STARTPTS[main{idx}];") - # B-roll 片段：缩放至目标分辨率 + # B-roll 片段：缩放到输出分辨率并裁到对应时长 + in_lbl = input_label_fn(seg) parts.append( - f"[{idx + 1}:v]scale={output_width}:{output_height}" + f"{in_lbl}scale={output_width}:{output_height}" f":force_original_aspect_ratio=decrease," f"pad={output_width}:{output_height}:(ow-iw)/2:(oh-ih)/2," f"trim=start=0:end={end - start},setpts=PTS-STARTPTS[br{idx}];" ) prev_end = end - # 尾部片段 + # 尾部主视频片段 if prev_end < video_duration: - last_idx = len(segments) + last_idx = len(sorted_fs_segments) parts.append(f"[0:v]trim=start={prev_end}:end={video_duration},setpts=PTS-STARTPTS[main{last_idx}];") # concat 所有片段 - segment_labels = [] - for idx in range(len(segments)): - start = segments[idx].get("start_time", 0) - if (idx == 0 and segments[0].get("start_time", 0) > 0) or idx > 0: - prev_end_prev = segments[idx - 1].get("end_time", 0) if idx > 0 else 0 - if prev_end_prev < start: - segment_labels.append(f"[main{idx}]") + segment_labels: list[str] = [] + for idx, seg in enumerate(sorted_fs_segments): + start = seg.get("start_time", 0) + # 每段 B-roll 之前是否有主视频片段？ + has_main_before = (idx == 0 and start > 0) or (idx > 0 and sorted_fs_segments[idx - 1].get("end_time", 0) < start) + if has_main_before: + segment_labels.append(f"[main{idx}]") segment_labels.append(f"[br{idx}]") - if prev_end < video_duration: - segment_labels.append(f"[main{len(segments)}]") + segment_labels.append(f"[main{len(sorted_fs_segments)}]") + final_lbl = "vout_fs" n = len(segment_labels) if n > 0: concat_inputs = "".join(segment_labels) - parts.append(f"{concat_inputs}concat=n={n}:v=1:a=0[vout];") + parts.append(f"{concat_inputs}concat=n={n}:v=1:a=0[{final_lbl}];") + + return "".join(parts), final_lbl + + +def _build_pip_filters( + pip_segments: list[dict[str, Any]], + output_width: int, + output_height: int, + input_label_fn, + base_label: str | None, +) -> tuple[str, str]: + """构建 PIP（画中画）overlay 滤镜链。 + + Args: + pip_segments: 按时间排序的 pip 片段 + output_width: 输出宽度 + output_height: 输出高度 + input_label_fn: 片段 → 输入标签的映射函数 + base_label: 前序滤镜链输出的标签（如 fullscreen 的 vout_fs），为 None 则基于 [0:v] + + Returns: + (filter_str, final_label) + """ + parts: list[str] = [] + cur_label = base_label # 当前叠加到的标签 + + pos_map = { + "top_left": "10:10", + "top_right": "W-w-10:10", + "bottom_left": "10:H-h-10", + "bottom_right": "W-w-10:H-h-10", + "center": "(W-w)/2:(H-h)/2", + } + + for idx, seg in enumerate(pip_segments): + start = seg.get("start_time", 0) + end = seg.get("end_time", 0) + scale = seg.get("pip_scale", 0.3) + position = seg.get("pip_position", "bottom_right") + pos_expr = pos_map.get(position, pos_map["bottom_right"]) + + pip_w = max(1, int(output_width * scale)) + pip_h = max(1, int(output_height * scale)) + enable_expr = f"enable='between(t,{start},{end})'" + + in_lbl = input_label_fn(seg) + pip_scaled = f"pip{idx}" + parts.append(f"{in_lbl}scale={pip_w}:{pip_h},{enable_expr}[{pip_scaled}];") + + # overlay onto the current base + base = f"[{cur_label}]" if cur_label else "[0:v]" + out_lbl = f"vout_pip{idx}" if idx < len(pip_segments) - 1 else "vout" + parts.append(f"{base}[{pip_scaled}]overlay={pos_expr}:{enable_expr}[{out_lbl}];") + cur_label = out_lbl + + return "".join(parts), cur_label or "vout" - return "".join(parts) def build_cover_extract_command( diff --git a/tests/unit/test_ai_avatar_render_routes.py b/tests/unit/test_ai_avatar_render_routes.py index 028c4e273..f8b368c6b 100644 --- a/tests/unit/test_ai_avatar_render_routes.py +++ b/tests/unit/test_ai_avatar_render_routes.py @@ -258,8 +258,9 @@ class TestBrollOverlayFilter: def test_empty_segments_returns_empty(self): from packages.domain.video_filter_builder import build_broll_overlay_filter - result = build_broll_overlay_filter([], 30.0) + result, label = build_broll_overlay_filter([], 30.0) assert result == "" + assert label is None def test_pip_mode_generates_overlay(self): from packages.domain.video_filter_builder import build_broll_overlay_filter @@ -275,8 +276,9 @@ class TestBrollOverlayFilter: "pip_scale": 0.3, } ] - result = build_broll_overlay_filter(segments, 30.0) + result, label = build_broll_overlay_filter(segments, 30.0) assert "overlay" in result or "scale=" in result + assert label == "vout" def test_fullscreen_mode_generates_concat(self): from packages.domain.video_filter_builder import build_broll_overlay_filter @@ -290,8 +292,9 @@ class TestBrollOverlayFilter: "end_time": 10.0, } ] - result = build_broll_overlay_filter(segments, 30.0) + result, label = build_broll_overlay_filter(segments, 30.0) assert "trim" in result or "concat" in result + assert label == "vout_fs" def test_cover_extract_command(self): from packages.domain.video_filter_builder import build_cover_extract_command diff --git a/tests/unit/test_video_filter_builder.py b/tests/unit/test_video_filter_builder.py index 0832dffdb..45ef53c72 100755 --- a/tests/unit/test_video_filter_builder.py +++ b/tests/unit/test_video_filter_builder.py @@ -902,9 +902,10 @@ class TestResolveFontPath(unittest.TestCase): @patch("os.path.isfile") def test_unknown_font_fallback(self, mock_isfile): - mock_isfile.side_effect = lambda p: "DejaVu" in p + # DejaVuSans 已从 fallback 列表移除（不支持 CJK），用 VF 路径模拟 + mock_isfile.side_effect = lambda p: "NotoSansSC-VF" in p result = _resolve_font_path("UnknownFont") - self.assertIn("DejaVu", result) + self.assertIn("NotoSansSC-VF", result) @patch("os.path.isfile") def test_no_fonts_available(self, mock_isfile): @@ -927,9 +928,11 @@ class TestResolveFontPath(unittest.TestCase): @patch("os.path.isfile") def test_font_fallback_skips_nonexistent(self, mock_isfile): - mock_isfile.side_effect = lambda p: "DejaVu" in p + # 所有中文字体路径都不存在时，fallback 返回第一个存在的文件； + # DejaVuSans 已从列表移除（不支持 CJK），使用 VF 字体路径模拟存在文件 + mock_isfile.side_effect = lambda p: "NotoSansSC-VF" in p result = _resolve_font_path("不存在字体") - self.assertIn("DejaVu", result) + self.assertIn("NotoSansSC-VF", result) class TestDrawtextFontFileIncluded(unittest.TestCase): @@ -1028,6 +1031,14 @@ class TestDrawtextBoldFalse(unittest.TestCase): self.assertIsNotNone(result) self.assertNotIn("font=bold", result) + def test_bold_true_does_not_use_font_bold_param(self): + """粗体模式不得使用 `font=bold`——该参数无效，会导致 filter_complex 解析失败（exit 234）。""" + result = build_title_drawtext_filter({"text": "标题", "bold": True}) + self.assertIsNotNone(result) + self.assertNotIn("font=bold", result) + # 粗体应通过 borderw 实现 + self.assertIn("borderw=", result) + class TestDrawtextPositionBranches(unittest.TestCase): """位置相关分支覆盖。"""