From dc12f390bb83ff0afe5badee52787b58ae38ff91 Mon Sep 17 00:00:00 2001 From: viccy Date: Mon, 8 Jun 2026 13:05:30 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E5=8E=9F=E7=89=87?= =?UTF-8?q?=E5=AD=97=E5=B9=95=E6=94=AF=E6=8C=81=E5=B9=B6=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E8=A7=86=E9=A2=91=E5=90=88=E5=B9=B6=E6=B5=81=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 为VideoClipParams新增原字幕路径配置字段，支持单条/多条字幕路径 - 完善webui参数获取逻辑，处理字幕路径兼容性并对接前端选择 - 重构后端字幕处理流程，支持自动匹配视频对应原字幕，合并原声字幕 - 优化视频合并逻辑，新增ffmpeg无损copy合并判断，自动回退重编码提升效率 - 新增ffmpeg快速素材合并路径，支持自定义字幕样式与多音轨混合 - 新增多个单元测试覆盖字幕匹配、合并及视频合并场景 --- app/models/schema.py | 2 + app/services/generate_video.py | 1047 ++++++++++++++++- app/services/merger_video.py | 200 +++- app/services/script_subtitle.py | 252 +++- app/services/task.py | 158 ++- .../test_merger_video_concat_unittest.py | 120 ++ app/services/test_script_subtitle_unittest.py | 98 ++ .../test_task_subtitle_resolution_unittest.py | 46 + webui.py | 11 + webui/components/script_settings.py | 3 + 10 files changed, 1863 insertions(+), 74 deletions(-) create mode 100644 app/services/test_merger_video_concat_unittest.py create mode 100644 app/services/test_task_subtitle_resolution_unittest.py diff --git a/app/models/schema.py b/app/models/schema.py index b492bb1..5b16143 100644 --- a/app/models/schema.py +++ b/app/models/schema.py @@ -165,6 +165,8 @@ class VideoClipParams(BaseModel): video_clip_json_path: Optional[str] = Field(default="", description="LLM 生成的视频剪辑脚本路径") video_origin_path: Optional[str] = Field(default="", description="原视频路径") video_origin_paths: Optional[List[str]] = Field(default=[], description="原视频路径列表") + original_subtitle_path: Optional[str] = Field(default="", description="原视频字幕路径") + original_subtitle_paths: Optional[List[str]] = Field(default=[], description="原视频字幕路径列表") video_aspect: Optional[VideoAspect] = Field(default=VideoAspect.portrait.value, description="视频比例") video_language: Optional[str] = Field(default="zh-CN", description="视频语言") diff --git a/app/services/generate_video.py b/app/services/generate_video.py index d66166b..cca0c04 100644 --- a/app/services/generate_video.py +++ b/app/services/generate_video.py @@ -9,6 +9,10 @@ ''' import os +import json +import re +import shlex +import subprocess import traceback import tempfile from typing import Optional, Dict, Any @@ -49,6 +53,9 @@ SUBTITLE_MASK_DEFAULTS = { }, } +_FFMPEG_FILTER_CACHE: Dict[tuple[str, str], bool] = {} +_FFMPEG_ENCODER_CACHE: Dict[tuple[str, str], bool] = {} + def _clamp(value, minimum, maximum): return min(max(value, minimum), maximum) @@ -266,6 +273,924 @@ def is_valid_subtitle_file(subtitle_path: str) -> bool: return False +def _has_existing_file(file_path: Optional[str]) -> bool: + return bool(file_path and os.path.exists(file_path)) + + +def _get_ffmpeg_binary() -> str: + for env_name in ("NARRATO_FFMPEG_EXE", "IMAGEIO_FFMPEG_EXE"): + candidate = os.environ.get(env_name, "").strip() + if candidate and os.path.isfile(candidate): + return candidate + + try: + import imageio_ffmpeg + + candidate = imageio_ffmpeg.get_ffmpeg_exe() + if candidate and os.path.isfile(candidate): + return candidate + except Exception as e: + logger.debug(f"未找到 imageio-ffmpeg 二进制: {e}") + + return "ffmpeg" + + +def _get_ffprobe_binary(ffmpeg_binary: Optional[str] = None) -> str: + for env_name in ("NARRATO_FFPROBE_EXE", "IMAGEIO_FFPROBE_EXE"): + candidate = os.environ.get(env_name, "").strip() + if candidate and os.path.isfile(candidate): + return candidate + + if ffmpeg_binary: + sibling = os.path.join(os.path.dirname(ffmpeg_binary), "ffprobe") + if os.path.isfile(sibling): + return sibling + + return "ffprobe" + + +def _check_ffmpeg_binary(ffmpeg_binary: str) -> bool: + try: + subprocess.run( + [ffmpeg_binary, "-version"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=True, + ) + return True + except (subprocess.SubprocessError, FileNotFoundError) as e: + logger.error(f"ffmpeg 不可用: {ffmpeg_binary}, {e}") + return False + + +def _format_ffmpeg_float(value: float) -> str: + return f"{float(value):.3f}".rstrip("0").rstrip(".") + + +def _quote_filter_value(value: str) -> str: + escaped = str(value).replace("\\", "\\\\").replace("'", "\\'") + return f"'{escaped}'" + + +def _probe_video(video_path: str) -> Dict[str, Any]: + ffmpeg_binary = _get_ffmpeg_binary() + ffprobe_binary = _get_ffprobe_binary(ffmpeg_binary) + cmd = [ + ffprobe_binary, + "-v", + "error", + "-print_format", + "json", + "-show_streams", + "-show_format", + video_path, + ] + result = subprocess.run( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + if result.returncode != 0: + raise RuntimeError(f"ffprobe 读取视频失败: {result.stderr.strip()}") + + data = json.loads(result.stdout or "{}") + streams = data.get("streams", []) + video_stream = next((stream for stream in streams if stream.get("codec_type") == "video"), None) + if not video_stream: + raise RuntimeError("ffprobe 未找到视频流") + + duration = ( + video_stream.get("duration") + or data.get("format", {}).get("duration") + or 0 + ) + duration = float(duration) + if duration <= 0: + raise RuntimeError("ffprobe 未获取到有效视频时长") + + return { + "width": int(video_stream["width"]), + "height": int(video_stream["height"]), + "duration": duration, + "has_audio": any(stream.get("codec_type") == "audio" for stream in streams), + } + + +def _ffmpeg_filter_available(filter_name: str) -> bool: + ffmpeg_binary = _get_ffmpeg_binary() + cache_key = (ffmpeg_binary, filter_name) + if cache_key in _FFMPEG_FILTER_CACHE: + return _FFMPEG_FILTER_CACHE[cache_key] + + try: + result = subprocess.run( + [ffmpeg_binary, "-hide_banner", "-filters"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + available = False + if result.returncode == 0: + for line in result.stdout.splitlines(): + parts = line.split() + if len(parts) >= 2 and parts[1] == filter_name: + available = True + break + _FFMPEG_FILTER_CACHE[cache_key] = available + return available + except Exception: + _FFMPEG_FILTER_CACHE[cache_key] = False + return False + + +def _ffmpeg_encoder_available(encoder_name: str) -> bool: + ffmpeg_binary = _get_ffmpeg_binary() + cache_key = (ffmpeg_binary, encoder_name) + if cache_key in _FFMPEG_ENCODER_CACHE: + return _FFMPEG_ENCODER_CACHE[cache_key] + + try: + result = subprocess.run( + [ffmpeg_binary, "-hide_banner", "-encoders"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + available = result.returncode == 0 and encoder_name in result.stdout + _FFMPEG_ENCODER_CACHE[cache_key] = available + return available + except Exception: + _FFMPEG_ENCODER_CACHE[cache_key] = False + return False + + +def _select_compatible_encoder(preferred_encoder: str) -> str: + if _ffmpeg_encoder_available(preferred_encoder): + return preferred_encoder + logger.warning(f"当前 ffmpeg 二进制不支持编码器 {preferred_encoder}，回退 libx264") + return "libx264" + + +def _srt_timestamp_to_seconds(timestamp: str) -> float: + match = re.match( + r"(?P\d{2}):(?P\d{2}):(?P\d{2}),(?P\d{3})", + timestamp.strip(), + ) + if not match: + raise ValueError(f"无效 SRT 时间戳: {timestamp}") + parts = {key: int(value) for key, value in match.groupdict().items()} + return ( + parts["hours"] * 3600 + + parts["minutes"] * 60 + + parts["seconds"] + + parts["millis"] / 1000 + ) + + +def _parse_srt_subtitles(subtitle_path: str) -> list[tuple[float, float, str]]: + with open(subtitle_path, "r", encoding="utf-8-sig") as file: + content = file.read().strip() + + if not content: + return [] + + subtitles = [] + blocks = re.split(r"\n\s*\n", content) + time_pattern = re.compile( + r"(?P\d{2}:\d{2}:\d{2},\d{3})\s*-->\s*" + r"(?P\d{2}:\d{2}:\d{2},\d{3})" + ) + for block in blocks: + lines = [line.strip("\ufeff") for line in block.splitlines() if line.strip()] + if not lines: + continue + + time_index = next( + (index for index, line in enumerate(lines) if time_pattern.search(line)), + None, + ) + if time_index is None: + continue + + match = time_pattern.search(lines[time_index]) + if not match: + continue + + text = "\n".join(lines[time_index + 1:]).strip() + if not text: + continue + + subtitles.append( + ( + _srt_timestamp_to_seconds(match.group("start")), + _srt_timestamp_to_seconds(match.group("end")), + text, + ) + ) + return subtitles + + +def _normalize_hex_color(color: Optional[str], default: str) -> str: + color_names = { + "white": "#FFFFFF", + "black": "#000000", + "red": "#FF0000", + "green": "#008000", + "blue": "#0000FF", + "yellow": "#FFFF00", + "cyan": "#00FFFF", + "magenta": "#FF00FF", + } + value = (color or default or "").strip() + value = color_names.get(value.lower(), value) + + if not value.startswith("#"): + return default + value = value[1:] + if len(value) == 3: + value = "".join(char * 2 for char in value) + if len(value) != 6: + return default + try: + int(value, 16) + except ValueError: + return default + return f"#{value.upper()}" + + +def _css_color_to_ass(color: Optional[str], default: str) -> str: + hex_color = _normalize_hex_color(color, default)[1:] + red = int(hex_color[0:2], 16) + green = int(hex_color[2:4], 16) + blue = int(hex_color[4:6], 16) + return f"&H00{blue:02X}{green:02X}{red:02X}" + + +def _resolve_font_path(subtitle_font: str) -> Optional[str]: + if subtitle_font and os.path.isabs(subtitle_font) and os.path.exists(subtitle_font): + return subtitle_font + + if subtitle_font: + font_path = os.path.join(utils.font_dir(), subtitle_font) + if os.path.exists(font_path): + return font_path + + for candidate in [ + os.path.join(utils.font_dir(), "SourceHanSansCN-Regular.otf"), + os.path.join(utils.font_dir(), "SimHei.ttf"), + "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", + "/System/Library/Fonts/STHeiti Medium.ttc", + "/System/Library/Fonts/Hiragino Sans GB.ttc", + ]: + if os.path.exists(candidate): + return candidate + return None + + +def _resolve_font_family(font_path: Optional[str], subtitle_font: str) -> str: + if font_path: + try: + return ImageFont.truetype(font_path, 12).getname()[0] + except Exception: + pass + if subtitle_font: + return os.path.splitext(os.path.basename(subtitle_font))[0] + return "Arial" + + +def _estimate_subtitle_margin( + video_height: int, + font_size: int, + subtitle_position: str, + custom_position: float, + orientation_subtitle_y_percent: Optional[float], +) -> tuple[int, int]: + if subtitle_position == "top": + return 8, max(10, round(video_height * 0.05)) + if subtitle_position == "center": + return 5, 10 + + y_percent = orientation_subtitle_y_percent + if y_percent is None and subtitle_position == "custom": + y_percent = custom_position + + if y_percent is not None: + estimated_text_height = max(24, round(font_size * 1.35)) + y = (video_height - estimated_text_height) * (y_percent / 100) + margin = video_height - y - estimated_text_height + return 2, max(10, round(margin)) + + return 2, max(10, round(video_height * 0.05)) + + +def _build_subtitle_filter( + subtitle_path: str, + font_path: Optional[str], + subtitle_font: str, + subtitle_font_size: int, + subtitle_color: str, + stroke_color: str, + stroke_width: float, + video_width: int, + video_height: int, + subtitle_position: str, + custom_position: float, + orientation_subtitle_y_percent: Optional[float], +) -> str: + font_family = _resolve_font_family(font_path, subtitle_font) + alignment, margin_v = _estimate_subtitle_margin( + video_height=video_height, + font_size=subtitle_font_size, + subtitle_position=subtitle_position, + custom_position=custom_position, + orientation_subtitle_y_percent=orientation_subtitle_y_percent, + ) + force_style = ",".join( + [ + f"Fontname={font_family}", + f"Fontsize={subtitle_font_size}", + f"PrimaryColour={_css_color_to_ass(subtitle_color, '#FFFFFF')}", + f"OutlineColour={_css_color_to_ass(stroke_color, '#000000')}", + "BorderStyle=1", + f"Outline={stroke_width}", + "Shadow=0", + f"Alignment={alignment}", + f"MarginV={margin_v}", + ] + ) + + args = [f"filename={_quote_filter_value(subtitle_path)}"] + args.append(f"original_size={video_width}x{video_height}") + if font_path: + args.append(f"fontsdir={_quote_filter_value(os.path.dirname(font_path))}") + args.append(f"force_style={_quote_filter_value(force_style)}") + return f"subtitles={':'.join(args)}" + + +def _css_color_to_drawtext(color: Optional[str], default: str) -> str: + return f"0x{_normalize_hex_color(color, default)[1:]}" + + +def _escape_drawtext_text(text: str) -> str: + return ( + text.replace("\\", "\\\\") + .replace("%", "\\%") + .replace("\r\n", "\n") + .replace("\r", "\n") + .replace("\n", "\\n") + ) + + +def _resolve_drawtext_y_expression( + subtitle_position: str, + custom_position: float, + orientation_subtitle_y_percent: Optional[float], +) -> str: + if subtitle_position == "top": + return "h*0.05" + if subtitle_position == "center": + return "(h-text_h)/2" + + y_percent = orientation_subtitle_y_percent + if y_percent is None and subtitle_position == "custom": + y_percent = custom_position + + if y_percent is not None: + return f"(h-text_h)*{_format_ffmpeg_float(y_percent / 100)}" + return "h*0.95-text_h" + + +def _build_drawtext_filters( + subtitle_path: str, + font_path: Optional[str], + subtitle_font_size: int, + subtitle_color: str, + stroke_color: str, + stroke_width: float, + subtitle_position: str, + custom_position: float, + orientation_subtitle_y_percent: Optional[float], + video_width: int, +) -> list[str]: + subtitles = _parse_srt_subtitles(subtitle_path) + if not subtitles: + raise RuntimeError("SRT 字幕解析结果为空，无法使用 drawtext 快路径") + + y_expr = _resolve_drawtext_y_expression( + subtitle_position=subtitle_position, + custom_position=custom_position, + orientation_subtitle_y_percent=orientation_subtitle_y_percent, + ) + max_width = video_width * 0.9 + drawtext_filters = [] + + for start, end, text in subtitles: + wrapped_text = text + if font_path: + wrapped_text, _ = wrap_text( + text, + max_width=max_width, + font=font_path, + fontsize=subtitle_font_size, + ) + + args = [] + if font_path: + args.append(f"fontfile={_quote_filter_value(font_path)}") + args.extend( + [ + f"text={_quote_filter_value(_escape_drawtext_text(wrapped_text))}", + f"fontcolor={_css_color_to_drawtext(subtitle_color, '#FFFFFF')}", + f"fontsize={subtitle_font_size}", + f"borderw={stroke_width}", + f"bordercolor={_css_color_to_drawtext(stroke_color, '#000000')}", + "x=(w-text_w)/2", + f"y={y_expr}", + ( + "enable=" + f"{_quote_filter_value(f'between(t,{_format_ffmpeg_float(start)},{_format_ffmpeg_float(end)})')}" + ), + ] + ) + drawtext_filters.append(f"drawtext={':'.join(args)}") + + return drawtext_filters + + +def _hex_to_rgba(color: Optional[str], default: str, alpha: int = 255) -> tuple[int, int, int, int]: + hex_color = _normalize_hex_color(color, default)[1:] + return ( + int(hex_color[0:2], 16), + int(hex_color[2:4], 16), + int(hex_color[4:6], 16), + alpha, + ) + + +def _create_subtitle_png_file( + text: str, + font_path: Optional[str], + subtitle_font_size: int, + subtitle_color: str, + stroke_color: str, + stroke_width: float, + video_width: int, + output_dir: str, +) -> str: + font = ImageFont.truetype(font_path, subtitle_font_size) if font_path else ImageFont.load_default() + wrapped_text, _ = wrap_text( + text, + max_width=video_width * 0.9, + font=font_path or "Arial", + fontsize=subtitle_font_size, + ) + stroke_width_px = max(0, int(round(float(stroke_width)))) + padding = max(8, stroke_width_px * 3 + 6) + + probe = Image.new("RGBA", (1, 1), (0, 0, 0, 0)) + draw = ImageDraw.Draw(probe) + bbox = draw.multiline_textbbox( + (0, 0), + wrapped_text, + font=font, + spacing=4, + stroke_width=stroke_width_px, + align="center", + ) + text_width = max(1, bbox[2] - bbox[0]) + text_height = max(1, bbox[3] - bbox[1]) + image = Image.new( + "RGBA", + (text_width + padding * 2, text_height + padding * 2), + (0, 0, 0, 0), + ) + draw = ImageDraw.Draw(image) + draw.multiline_text( + (image.width / 2, padding - bbox[1]), + wrapped_text, + font=font, + fill=_hex_to_rgba(subtitle_color, "#FFFFFF"), + anchor="ma", + spacing=4, + align="center", + stroke_width=stroke_width_px, + stroke_fill=_hex_to_rgba(stroke_color, "#000000"), + ) + + temp_file = tempfile.NamedTemporaryFile( + suffix=".png", + prefix="subtitle_text_", + dir=output_dir, + delete=False, + ) + temp_file.close() + image.save(temp_file.name) + return temp_file.name + + +def _resolve_overlay_y_expression( + subtitle_position: str, + custom_position: float, + orientation_subtitle_y_percent: Optional[float], +) -> str: + if subtitle_position == "top": + return "main_h*0.05" + if subtitle_position == "center": + return "(main_h-overlay_h)/2" + + y_percent = orientation_subtitle_y_percent + if y_percent is None and subtitle_position == "custom": + y_percent = custom_position + + if y_percent is not None: + return f"(main_h-overlay_h)*{_format_ffmpeg_float(y_percent / 100)}" + return "main_h*0.95-overlay_h" + + +def _create_subtitle_mask_alpha_file(region: Dict[str, Any], output_dir: str) -> str: + alpha = _build_subtitle_mask_alpha(region) + temp_file = tempfile.NamedTemporaryFile( + suffix=".png", + prefix="subtitle_mask_", + dir=output_dir, + delete=False, + ) + temp_file.close() + alpha.save(temp_file.name) + return temp_file.name + + +def _build_mask_filter( + input_label: str, + mask_input_index: int, + region: Dict[str, Any], + output_label: str, +) -> list[str]: + blur_sigma = ( + max(4, round(region["blur_radius"] * (0.9 + region["opacity"] * 0.35))) + if region["blur_radius"] > 0 + else 0 + ) + brightness = 1.0 + 0.03 + region["opacity"] * 0.04 + contrast = 0.975 - region["opacity"] * 0.035 + saturation = 1.0 + region["opacity"] * 0.03 + obliterate_width = max(24, round(region["padded_width"] * 0.12)) + obliterate_height = max(12, round(region["padded_height"] * 0.18)) + + blur_chain = ( + f"[masksrc]crop={region['padded_width']}:{region['padded_height']}:" + f"{region['padded_x']}:{region['padded_y']}," + f"scale={obliterate_width}:{obliterate_height}:flags=bicubic," + f"scale={region['padded_width']}:{region['padded_height']}:flags=lanczos" + ) + if blur_sigma > 0: + blur_chain += f",gblur=sigma={blur_sigma}" + blur_chain += ( + ",boxblur=4," + f"eq=brightness={brightness - 1.0:.3f}:" + f"contrast={contrast:.3f}:saturation={saturation:.3f}," + "format=rgba[maskblur]" + ) + + return [ + f"{input_label}split[maskbase][masksrc]", + blur_chain, + ( + f"[{mask_input_index}:v]format=gray," + f"scale={region['padded_width']}:{region['padded_height']}[maskalpha]" + ), + "[maskblur][maskalpha]alphamerge[masked]", + ( + f"[maskbase][masked]overlay={region['padded_x']}:{region['padded_y']}:" + f"format=auto{output_label}" + ), + ] + + +def _build_video_encoder_args(encoder: str, threads: int) -> list[str]: + if encoder == "h264_vaapi": + logger.warning("当前合成滤镜链暂不使用 VAAPI 编码，回退到 libx264") + encoder = "libx264" + + args = ["-c:v", encoder] + if encoder == "h264_nvenc": + args.extend(["-preset", "fast", "-cq", "23"]) + elif encoder == "h264_videotoolbox": + args.extend(["-q:v", "65"]) + elif encoder == "h264_qsv": + args.extend(["-preset", "veryfast", "-global_quality", "23"]) + elif encoder == "h264_amf": + args.extend(["-quality", "speed", "-qp_i", "23", "-qp_p", "23"]) + else: + args.extend(["-preset", "veryfast", "-crf", "23", "-threads", str(threads)]) + return args + + +def _build_moviepy_encoder_options() -> tuple[str, list[str]]: + from app.utils import ffmpeg_utils + + encoder = _select_compatible_encoder(ffmpeg_utils.get_optimal_ffmpeg_encoder()) + if encoder == "h264_vaapi": + logger.warning("MoviePy 兼容路径暂不使用 VAAPI 编码，回退到 libx264") + encoder = "libx264" + + if encoder == "h264_nvenc": + return encoder, ["-preset", "fast", "-cq", "23", "-pix_fmt", "yuv420p"] + if encoder == "h264_videotoolbox": + return encoder, ["-q:v", "65", "-pix_fmt", "yuv420p"] + if encoder == "h264_qsv": + return encoder, ["-preset", "veryfast", "-global_quality", "23", "-pix_fmt", "yuv420p"] + if encoder == "h264_amf": + return encoder, ["-quality", "speed", "-qp_i", "23", "-qp_p", "23", "-pix_fmt", "yuv420p"] + return "libx264", ["-preset", "veryfast", "-crf", "23", "-pix_fmt", "yuv420p"] + + +def _build_ffmpeg_merge_command( + video_path: str, + audio_path: str, + output_path: str, + subtitle_path: Optional[str], + bgm_path: Optional[str], + options: Dict[str, Any], +) -> tuple[list[str], list[str]]: + from app.utils import ffmpeg_utils + + video_meta = _probe_video(video_path) + output_dir = os.path.dirname(output_path) + duration = float(video_meta["duration"]) + duration_arg = _format_ffmpeg_float(duration) + video_width = int(video_meta["width"]) + video_height = int(video_meta["height"]) + + voice_volume = options.get("voice_volume", AudioVolumeDefaults.VOICE_VOLUME) + bgm_volume = options.get("bgm_volume", AudioVolumeDefaults.BGM_VOLUME) + original_audio_volume = options.get("original_audio_volume", AudioVolumeDefaults.ORIGINAL_VOLUME) + keep_original_audio = options.get("keep_original_audio", True) + subtitle_font = options.get("subtitle_font", "") + subtitle_font_size = int(options.get("subtitle_font_size", 40)) + subtitle_color = options.get("subtitle_color", "#FFFFFF") + subtitle_position = options.get("subtitle_position", "bottom") + custom_position = float(options.get("custom_position", 70)) + stroke_color = options.get("stroke_color", "#000000") + stroke_width = options.get("stroke_width", 1) + threads = int(options.get("threads", 2)) + fps = options.get("fps", 30) + subtitle_enabled = options.get("subtitle_enabled", True) + subtitle_mask_enabled = bool(options.get("subtitle_mask_enabled", False)) + + input_args = ["-i", video_path] + next_input_index = 1 + audio_filters = [] + audio_labels = [] + temp_files = [] + + if keep_original_audio and original_audio_volume > 0 and video_meta["has_audio"]: + label = f"a{len(audio_labels)}" + audio_filters.append( + f"[0:a]volume={original_audio_volume},atrim=0:{duration_arg}," + f"asetpts=PTS-STARTPTS[{label}]" + ) + audio_labels.append(f"[{label}]") + + if _has_existing_file(audio_path): + voice_input_index = next_input_index + next_input_index += 1 + input_args.extend(["-i", audio_path]) + label = f"a{len(audio_labels)}" + audio_filters.append( + f"[{voice_input_index}:a]volume={voice_volume},atrim=0:{duration_arg}," + f"asetpts=PTS-STARTPTS[{label}]" + ) + audio_labels.append(f"[{label}]") + + if _has_existing_file(bgm_path) and bgm_volume > 0: + bgm_input_index = next_input_index + next_input_index += 1 + input_args.extend(["-stream_loop", "-1", "-i", bgm_path]) + fade_start = max(0.0, duration - 3.0) + label = f"a{len(audio_labels)}" + audio_filters.append( + f"[{bgm_input_index}:a]volume={bgm_volume},atrim=0:{duration_arg}," + f"afade=t=out:st={_format_ffmpeg_float(fade_start)}:d=3," + f"asetpts=PTS-STARTPTS[{label}]" + ) + audio_labels.append(f"[{label}]") + + if len(audio_labels) == 1: + audio_filters.append( + f"{audio_labels[0]}atrim=0:{duration_arg},asetpts=PTS-STARTPTS[aout]" + ) + elif len(audio_labels) > 1: + audio_filters.append( + f"{''.join(audio_labels)}amix=inputs={len(audio_labels)}:" + f"duration=longest:dropout_transition=0:normalize=0," + f"atrim=0:{duration_arg},asetpts=PTS-STARTPTS[aout]" + ) + + valid_subtitle = bool( + subtitle_enabled + and subtitle_path + and is_valid_subtitle_file(subtitle_path) + ) + has_subtitles_filter = _ffmpeg_filter_available("subtitles") if valid_subtitle else False + has_drawtext_filter = _ffmpeg_filter_available("drawtext") if valid_subtitle else False + if valid_subtitle and not has_subtitles_filter and not has_drawtext_filter: + if not _ffmpeg_filter_available("overlay"): + raise RuntimeError("当前 ffmpeg 缺少 subtitles/drawtext/overlay 字幕处理滤镜") + logger.warning("当前 ffmpeg 缺少 subtitles/drawtext，改用 PNG 字幕叠加快路径") + + video_filters = [] + current_video_label = "[0:v]" + + if subtitle_enabled and subtitle_mask_enabled: + region = _resolve_subtitle_mask_region(video_width, video_height, options) + mask_path = _create_subtitle_mask_alpha_file(region, output_dir) + temp_files.append(mask_path) + mask_input_index = next_input_index + next_input_index += 1 + input_args.extend(["-loop", "1", "-t", duration_arg, "-i", mask_path]) + logger.info( + "ffmpeg 字幕遮罩已启用: " + f"{region['orientation']} x={region['x']} y={region['y']} " + f"w={region['width']} h={region['height']} blur={region['blur_radius']}" + ) + video_filters.extend( + _build_mask_filter( + input_label=current_video_label, + mask_input_index=mask_input_index, + region=region, + output_label="[v_masked]", + ) + ) + current_video_label = "[v_masked]" + + if valid_subtitle: + font_path = _resolve_font_path(subtitle_font) + if font_path: + logger.info(f"ffmpeg 使用字幕字体: {font_path}") + orientation_subtitle_y_percent = _resolve_orientation_subtitle_y_percent( + video_width, + video_height, + options, + ) + if has_drawtext_filter: + drawtext_filters = _build_drawtext_filters( + subtitle_path=subtitle_path, + font_path=font_path, + subtitle_font_size=subtitle_font_size, + subtitle_color=subtitle_color, + stroke_color=stroke_color, + stroke_width=stroke_width, + subtitle_position=subtitle_position, + custom_position=custom_position, + orientation_subtitle_y_percent=orientation_subtitle_y_percent, + video_width=video_width, + ) + for index, drawtext_filter in enumerate(drawtext_filters): + next_label = f"[v_drawtext_{index}]" + video_filters.append(f"{current_video_label}{drawtext_filter}{next_label}") + current_video_label = next_label + elif has_subtitles_filter: + subtitle_filter = _build_subtitle_filter( + subtitle_path=subtitle_path, + font_path=font_path, + subtitle_font=subtitle_font, + subtitle_font_size=subtitle_font_size, + subtitle_color=subtitle_color, + stroke_color=stroke_color, + stroke_width=stroke_width, + video_width=video_width, + video_height=video_height, + subtitle_position=subtitle_position, + custom_position=custom_position, + orientation_subtitle_y_percent=orientation_subtitle_y_percent, + ) + video_filters.append(f"{current_video_label}{subtitle_filter}[v_subtitled]") + current_video_label = "[v_subtitled]" + else: + y_expr = _resolve_overlay_y_expression( + subtitle_position=subtitle_position, + custom_position=custom_position, + orientation_subtitle_y_percent=orientation_subtitle_y_percent, + ) + for index, (start, end, text) in enumerate(_parse_srt_subtitles(subtitle_path)): + png_path = _create_subtitle_png_file( + text=text, + font_path=font_path, + subtitle_font_size=subtitle_font_size, + subtitle_color=subtitle_color, + stroke_color=stroke_color, + stroke_width=stroke_width, + video_width=video_width, + output_dir=output_dir, + ) + temp_files.append(png_path) + subtitle_input_index = next_input_index + next_input_index += 1 + input_args.extend(["-loop", "1", "-t", duration_arg, "-i", png_path]) + next_label = f"[v_subtitle_png_{index}]" + enable_expr = ( + f"between(t,{_format_ffmpeg_float(start)},{_format_ffmpeg_float(end)})" + ) + video_filters.append( + f"{current_video_label}[{subtitle_input_index}:v]" + f"overlay=x=(main_w-overlay_w)/2:y={y_expr}:" + f"enable={_quote_filter_value(enable_expr)}:format=auto{next_label}" + ) + current_video_label = next_label + elif subtitle_enabled and subtitle_path: + logger.warning(f"字幕文件无效或为空: {subtitle_path}，ffmpeg 快路径跳过字幕") + + has_video_filter = bool(video_filters) + if has_video_filter: + final_video_filters = [] + if fps: + final_video_filters.append(f"fps={fps}") + final_video_filters.append("format=yuv420p") + video_filters.append( + f"{current_video_label}{','.join(final_video_filters)}[vout]" + ) + + filter_parts = [*video_filters, *audio_filters] + ffmpeg_binary = _get_ffmpeg_binary() + cmd = [ffmpeg_binary, "-y", "-hide_banner", "-loglevel", "error", *input_args] + if filter_parts: + cmd.extend(["-filter_complex", ";".join(filter_parts)]) + + if has_video_filter: + encoder = _select_compatible_encoder(ffmpeg_utils.get_optimal_ffmpeg_encoder()) + cmd.extend(["-map", "[vout]", *_build_video_encoder_args(encoder, threads)]) + else: + cmd.extend(["-map", "0:v:0", "-c:v", "copy"]) + + if audio_labels: + cmd.extend(["-map", "[aout]", "-c:a", "aac", "-b:a", "192k"]) + else: + cmd.append("-an") + + cmd.extend(["-t", duration_arg, "-movflags", "+faststart", output_path]) + return cmd, temp_files + + +def _merge_materials_with_ffmpeg( + video_path: str, + audio_path: str, + output_path: str, + subtitle_path: Optional[str] = None, + bgm_path: Optional[str] = None, + options: Optional[Dict[str, Any]] = None, +) -> bool: + ffmpeg_binary = _get_ffmpeg_binary() + if not _check_ffmpeg_binary(ffmpeg_binary): + return False + + options = options or {} + temp_files = [] + try: + cmd, temp_files = _build_ffmpeg_merge_command( + video_path=video_path, + audio_path=audio_path, + output_path=output_path, + subtitle_path=subtitle_path, + bgm_path=bgm_path, + options=options, + ) + logger.info(f"使用 ffmpeg 快速合并素材: {shlex.join(cmd)}") + result = subprocess.run( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + if result.returncode != 0: + logger.warning(f"ffmpeg 快速合并失败，将回退 MoviePy: {result.stderr[-3000:]}") + if os.path.exists(output_path): + try: + os.remove(output_path) + except OSError: + pass + return False + + logger.success(f"ffmpeg 素材合并完成: {output_path}") + return True + except Exception as e: + logger.warning(f"ffmpeg 快速合并不可用，将回退 MoviePy: {e}") + return False + finally: + for temp_file in temp_files: + try: + if os.path.exists(temp_file): + os.remove(temp_file) + except OSError: + pass + + def merge_materials( video_path: str, audio_path: str, @@ -364,6 +1289,41 @@ def merge_materials( if bgm_path: logger.info(f" ④ 背景音乐: {bgm_path}") logger.info(f" ⑤ 输出: {output_path}") + + merge_engine = str(options.get("merge_engine", "ffmpeg")).lower() + use_ffmpeg_merge = bool(options.get("use_ffmpeg_merge", True)) + if use_ffmpeg_merge and merge_engine != "moviepy": + ffmpeg_options = dict(options) + ffmpeg_options.update( + { + "voice_volume": voice_volume, + "bgm_volume": bgm_volume, + "original_audio_volume": original_audio_volume, + "keep_original_audio": keep_original_audio, + "subtitle_font": subtitle_font, + "subtitle_font_size": subtitle_font_size, + "subtitle_color": subtitle_color, + "subtitle_bg_color": subtitle_bg_color, + "subtitle_position": subtitle_position, + "custom_position": custom_position, + "stroke_color": stroke_color, + "stroke_width": stroke_width, + "threads": threads, + "fps": fps, + "subtitle_enabled": subtitle_enabled, + "subtitle_mask_enabled": subtitle_mask_enabled, + } + ) + if _merge_materials_with_ffmpeg( + video_path=video_path, + audio_path=audio_path, + output_path=output_path, + subtitle_path=subtitle_path, + bgm_path=bgm_path, + options=ffmpeg_options, + ): + return output_path + logger.warning("ffmpeg 快速合并失败，继续使用 MoviePy 兼容路径") # 加载视频 try: @@ -406,7 +1366,7 @@ def merge_materials( temp_original_path = os.path.join(temp_dir, "temp_original.wav") # 保存原声到临时文件进行分析 - original_audio.write_audiofile(temp_original_path, verbose=False, logger=None) + original_audio.write_audiofile(temp_original_path, logger=None) # 计算智能音量调整 tts_adjustment, original_adjustment = normalizer.calculate_volume_adjustment( @@ -475,9 +1435,8 @@ def merge_materials( logger.warning("没有可用的音频轨道，输出视频将没有声音") # 处理字体路径 - font_path = None - if subtitle_path and subtitle_font: - font_path = os.path.join(utils.font_dir(), subtitle_font) + font_path = _resolve_font_path(subtitle_font) if subtitle_path else None + if font_path: if os.name == "nt": font_path = font_path.replace("\\", "/") logger.info(f"使用字体: {font_path}") @@ -508,24 +1467,28 @@ def merge_materials( # 创建文本片段 try: - _clip = TextClip( - text=wrapped_txt, - font=font_path, - font_size=subtitle_font_size, - color=subtitle_color, - bg_color=subtitle_bg_color, # 这里已经在前面处理过，None表示透明 - stroke_color=stroke_color, - stroke_width=stroke_width, - ) + text_clip_kwargs = { + "text": wrapped_txt, + "font_size": subtitle_font_size, + "color": subtitle_color, + "bg_color": subtitle_bg_color, # 这里已经在前面处理过，None表示透明 + "stroke_color": stroke_color, + "stroke_width": stroke_width, + } + if font_path: + text_clip_kwargs["font"] = font_path + _clip = TextClip(**text_clip_kwargs) except Exception as e: logger.error(f"创建字幕片段失败: {str(e)}, 使用简化参数重试") # 如果上面的方法失败，尝试使用更简单的参数 - _clip = TextClip( - text=wrapped_txt, - font=font_path, - font_size=subtitle_font_size, - color=subtitle_color, - ) + fallback_kwargs = { + "text": wrapped_txt, + "font_size": subtitle_font_size, + "color": subtitle_color, + } + if font_path: + fallback_kwargs["font"] = font_path + _clip = TextClip(**fallback_kwargs) # 设置字幕时间 duration = subtitle_item[0][1] - subtitle_item[0][0] @@ -561,12 +1524,14 @@ def merge_materials( # 创建TextClip工厂函数 def make_textclip(text): - return TextClip( - text=text, - font=font_path, - font_size=subtitle_font_size, - color=subtitle_color, - ) + text_clip_kwargs = { + "text": text, + "font_size": subtitle_font_size, + "color": subtitle_color, + } + if font_path: + text_clip_kwargs["font"] = font_path + return TextClip(**text_clip_kwargs) # 处理字幕 - 修复字幕开关bug和空字幕文件问题 if subtitle_enabled and subtitle_path: @@ -601,13 +1566,31 @@ def merge_materials( # 导出最终视频 try: - video_clip.write_videofile( - output_path, - audio_codec="aac", - temp_audiofile_path=output_dir, - threads=threads, - fps=fps, - ) + encoder, ffmpeg_params = _build_moviepy_encoder_options() + logger.info(f"MoviePy 导出编码器: {encoder}, 参数: {ffmpeg_params}") + try: + video_clip.write_videofile( + output_path, + codec=encoder, + audio_codec="aac", + temp_audiofile_path=output_dir, + threads=threads, + fps=fps, + ffmpeg_params=ffmpeg_params, + ) + except Exception: + if encoder == "libx264": + raise + logger.warning(f"MoviePy 使用 {encoder} 导出失败，回退 libx264: {traceback.format_exc()}") + video_clip.write_videofile( + output_path, + codec="libx264", + audio_codec="aac", + temp_audiofile_path=output_dir, + threads=threads, + fps=fps, + ffmpeg_params=["-preset", "veryfast", "-crf", "23", "-pix_fmt", "yuv420p"], + ) logger.success(f"素材合并完成: {output_path}") except Exception as e: logger.error(f"导出视频失败: {str(e)}") diff --git a/app/services/merger_video.py b/app/services/merger_video.py index c6ef84d..ddb9f64 100644 --- a/app/services/merger_video.py +++ b/app/services/merger_video.py @@ -9,6 +9,7 @@ ''' import os +import json import shutil import subprocess from enum import Enum @@ -127,6 +128,188 @@ def create_ffmpeg_concat_file(video_paths: List[str], concat_file_path: str) -> return concat_file_path +def _get_video_stream_signature(video_path: str) -> Optional[dict]: + """ + 获取用于判断 concat copy 是否安全的视频流关键参数。 + """ + probe_cmd = [ + 'ffprobe', '-v', 'error', + '-select_streams', 'v:0', + '-show_entries', + 'stream=codec_name,profile,width,height,pix_fmt,r_frame_rate,avg_frame_rate,time_base,sample_aspect_ratio', + '-of', 'json', + video_path + ] + + try: + result = subprocess.run( + probe_cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=True + ) + streams = json.loads(result.stdout or "{}").get("streams", []) + if not streams: + logger.warning(f"视频没有可用的视频流，不能使用 copy 合并: {video_path}") + return None + + stream = streams[0] + return { + "codec_name": stream.get("codec_name"), + "profile": stream.get("profile"), + "width": stream.get("width"), + "height": stream.get("height"), + "pix_fmt": stream.get("pix_fmt"), + "r_frame_rate": stream.get("r_frame_rate"), + "avg_frame_rate": stream.get("avg_frame_rate"), + "time_base": stream.get("time_base"), + "sample_aspect_ratio": stream.get("sample_aspect_ratio", "1:1"), + } + except Exception as e: + logger.warning(f"探测视频流参数失败，不能使用 copy 合并: {video_path}, 错误: {str(e)}") + return None + + +def _can_concat_video_copy(video_paths: List[str]) -> bool: + """ + 判断所有片段的视频流参数是否一致，避免 concat copy 造成时间轴或封装异常。 + """ + if not video_paths: + return False + + signatures = [] + for video_path in video_paths: + signature = _get_video_stream_signature(video_path) + if not signature: + return False + signatures.append(signature) + + base_signature = signatures[0] + for video_path, signature in zip(video_paths[1:], signatures[1:]): + if signature != base_signature: + logger.warning( + "视频片段参数不一致，跳过 copy 合并并回退重编码: " + f"{video_path}, 基准={base_signature}, 当前={signature}" + ) + return False + + return True + + +def _get_media_duration(video_path: str) -> Optional[float]: + probe_cmd = [ + 'ffprobe', '-v', 'error', + '-show_entries', 'format=duration', + '-of', 'csv=p=0', + video_path + ] + + try: + result = subprocess.run( + probe_cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=True + ) + return float(result.stdout.strip()) + except Exception as e: + logger.warning(f"探测视频时长失败: {video_path}, 错误: {str(e)}") + return None + + +def _concat_duration_matches(video_paths: List[str], output_path: str) -> bool: + input_durations = [] + for video_path in video_paths: + duration = _get_media_duration(video_path) + if duration is None: + return False + input_durations.append(duration) + + output_duration = _get_media_duration(output_path) + if output_duration is None: + return False + + expected_duration = sum(input_durations) + diff = abs(expected_duration - output_duration) + tolerance = max(0.5, len(video_paths) * 0.04) + if diff > tolerance: + logger.warning( + "视频流 copy 合并后的时长偏差过大，将回退重编码: " + f"期望={expected_duration:.3f}s, 实际={output_duration:.3f}s, 偏差={diff:.3f}s" + ) + return False + + logger.info( + "视频流 copy 合并时长校验通过: " + f"期望={expected_duration:.3f}s, 实际={output_duration:.3f}s" + ) + return True + + +def _build_concat_video_copy_cmd(concat_file: str, output_path: str) -> List[str]: + return [ + 'ffmpeg', '-y', + '-f', 'concat', + '-safe', '0', + '-i', concat_file, + '-c:v', 'copy', + '-an', + '-movflags', '+faststart', + '-avoid_negative_ts', 'make_zero', + output_path + ] + + +def _build_concat_video_reencode_cmd(concat_file: str, output_path: str, threads: int) -> List[str]: + return [ + 'ffmpeg', '-y', + '-f', 'concat', + '-safe', '0', + '-i', concat_file, + '-c:v', 'libx264', + '-preset', 'medium', + '-profile:v', 'high', + '-an', + '-threads', str(threads), + output_path + ] + + +def _concat_video_streams( + video_paths: List[str], + concat_file: str, + output_path: str, + threads: int +) -> None: + """ + 优先使用无损 copy 合并视频流，失败时回退到原来的重编码合并。 + """ + if _can_concat_video_copy(video_paths): + copy_cmd = _build_concat_video_copy_cmd(concat_file, output_path) + try: + subprocess.run(copy_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + if _concat_duration_matches(video_paths, output_path): + logger.info("视频流 copy 合并完成") + return + + if os.path.exists(output_path): + try: + os.remove(output_path) + except OSError as e: + logger.warning(f"删除 copy 合并临时结果失败，将继续尝试重编码覆盖: {str(e)}") + except subprocess.CalledProcessError as e: + error_msg = e.stderr.decode() if e.stderr else str(e) + logger.warning(f"视频流 copy 合并失败，将回退重编码合并: {error_msg}") + else: + logger.info("视频流不满足 copy 合并条件，将使用重编码合并") + + reencode_cmd = _build_concat_video_reencode_cmd(concat_file, output_path, threads) + subprocess.run(reencode_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + logger.info("视频流重编码合并完成") + + def process_single_video( input_path: str, output_path: str, @@ -474,22 +657,7 @@ def combine_clip_videos( concat_file = os.path.join(temp_dir, "concat_list.txt") create_ffmpeg_concat_file(video_paths_only, concat_file) - # 合并所有视频流，但不包含音频 - concat_cmd = [ - 'ffmpeg', '-y', - '-f', 'concat', - '-safe', '0', - '-i', concat_file, - '-c:v', 'libx264', - '-preset', 'medium', - '-profile:v', 'high', - '-an', # 不包含音频 - '-threads', str(threads), - video_concat_path - ] - - subprocess.run(concat_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - logger.info("视频流合并完成") + _concat_video_streams(video_paths_only, concat_file, video_concat_path, threads) # 2. 提取并合并有音频的片段 audio_segments = [video for video in processed_videos if video["keep_audio"]] diff --git a/app/services/script_subtitle.py b/app/services/script_subtitle.py index 2259580..0be21f6 100644 --- a/app/services/script_subtitle.py +++ b/app/services/script_subtitle.py @@ -5,12 +5,16 @@ from typing import Iterable, List, Optional, Sequence, Tuple from loguru import logger +from app.services.short_drama_narration_validation import build_subtitle_index +from app.services.subtitle_text import read_subtitle_text from app.utils import utils DEFAULT_SUBTITLE_OST_TYPES = (0, 2) +DEFAULT_ORIGINAL_SUBTITLE_OST_TYPES = (1,) DEFAULT_MAX_CHARS_PER_SUBTITLE = 12 SENTENCE_PART_RE = re.compile(r"[^。！？!?；;，,、\n]+[。！？!?；;，,、]?") +SubtitleEntry = Tuple[float, float, str] def _normalize_text(text: str) -> str: @@ -116,7 +120,95 @@ def _safe_ost_value(value) -> Optional[int]: return None +def _coerce_positive_int(value) -> Optional[int]: + try: + number = int(value) + except (TypeError, ValueError): + return None + return number if number > 0 else None + + +def _normalize_paths(paths) -> List[str]: + if isinstance(paths, str): + paths = [paths] + if not paths: + return [] + + normalized_paths = [] + seen = set() + for item in paths: + if not isinstance(item, str): + continue + item = item.strip() + if not item or item in seen: + continue + normalized_paths.append(item) + seen.add(item) + return normalized_paths + + +def _resolve_script_video_id(item: dict, video_origin_paths: Sequence[str]) -> int: + video_id = _coerce_positive_int(item.get("video_id") or item.get("video_index")) + if video_id is not None: + return video_id + + video_name = os.path.basename( + str(item.get("video_name") or item.get("source_video") or "").strip() + ) + if video_name: + for index, video_path in enumerate(video_origin_paths, start=1): + if os.path.basename(video_path) == video_name: + return index + + return 1 + + +def _read_subtitle_file(subtitle_path: str) -> str: + try: + return read_subtitle_text(subtitle_path).text + except Exception as e: + logger.warning(f"读取原片字幕失败: {subtitle_path}, {e}") + return "" + + +def _build_combined_original_subtitle_content( + original_subtitle_paths, + video_origin_paths=None, +) -> str: + subtitle_paths = _normalize_paths(original_subtitle_paths) + video_paths = _normalize_paths(video_origin_paths) + sections = [] + + for index, subtitle_path in enumerate(subtitle_paths, start=1): + if not os.path.exists(subtitle_path): + logger.warning(f"原片字幕文件不存在，跳过: {subtitle_path}") + continue + + content = _read_subtitle_file(subtitle_path) + if not content: + logger.warning(f"原片字幕文件为空，跳过: {subtitle_path}") + continue + + video_path = video_paths[index - 1] if index <= len(video_paths) else "" + if video_path: + header = ( + f"# 视频 {index}: {os.path.basename(video_path)}\n" + f"字幕文件: {os.path.basename(subtitle_path)}" + ) + else: + header = f"# 视频 {index}\n字幕文件: {os.path.basename(subtitle_path)}" + sections.append(f"{header}\n{content}".strip()) + + return "\n\n".join(sections) + + def _resolve_item_time_range(item: dict, current_time: float) -> Tuple[Optional[Tuple[float, float]], float]: + duration = float(item.get("duration", 0.0) or 0.0) + if duration > 0: + start = current_time + end = current_time + duration + return (start, end), end + edited_time_range = item.get("editedTimeRange") if edited_time_range: try: @@ -125,23 +217,16 @@ def _resolve_item_time_range(item: dict, current_time: float) -> Tuple[Optional[ except ValueError as e: logger.warning(f"解析 editedTimeRange 失败，将尝试使用 duration: {e}") - duration = float(item.get("duration", 0.0) or 0.0) - if duration <= 0: - return None, current_time - - start = current_time - end = current_time + duration - return (start, end), end + return None, current_time -def _build_srt_blocks( +def _build_narration_subtitle_entries( list_script: Sequence[dict], include_ost: Iterable[int], max_chars: int, -) -> List[str]: +) -> List[SubtitleEntry]: include_ost_set = {int(item) for item in include_ost} - blocks = [] - subtitle_index = 1 + entries: List[SubtitleEntry] = [] current_time = 0.0 for item in list_script: @@ -166,35 +251,158 @@ def _build_srt_blocks( for chunk_index, chunk in enumerate(chunks): chunk_start = start + chunk_duration * chunk_index chunk_end = end if chunk_index == len(chunks) - 1 else start + chunk_duration * (chunk_index + 1) - blocks.append( - "\n".join( - [ - str(subtitle_index), - f"{format_srt_time(chunk_start)} --> {format_srt_time(chunk_end)}", - chunk, - ] - ) + entries.append((chunk_start, chunk_end, chunk)) + + return entries + + +def _build_original_subtitle_entries( + list_script: Sequence[dict], + original_subtitle_paths=None, + video_origin_paths=None, + include_ost: Iterable[int] = DEFAULT_ORIGINAL_SUBTITLE_OST_TYPES, +) -> List[SubtitleEntry]: + original_subtitle_content = _build_combined_original_subtitle_content( + original_subtitle_paths, + video_origin_paths, + ) + if not original_subtitle_content: + return [] + + video_paths = _normalize_paths(video_origin_paths) + subtitle_index = build_subtitle_index(original_subtitle_content, video_paths) + if not subtitle_index: + logger.warning("原片字幕索引为空，无法为原声片段生成字幕") + return [] + + cues_by_video = {} + for cue in subtitle_index: + cues_by_video.setdefault(cue.video_id, []).append(cue) + + include_ost_set = {int(item) for item in include_ost} + entries: List[SubtitleEntry] = [] + current_time = 0.0 + + for item in list_script: + time_range, current_time = _resolve_item_time_range(item, current_time) + if not time_range: + continue + + ost = _safe_ost_value(item.get("OST")) + if ost not in include_ost_set: + continue + + source_time_range = item.get("sourceTimeRange") or item.get("timestamp") + try: + source_start, source_end = parse_time_range(source_time_range) + except ValueError as e: + logger.warning(f"解析原声片段源时间失败，跳过原片字幕: {e}") + continue + + target_start, target_end = time_range + source_duration = source_end - source_start + target_duration = target_end - target_start + if source_duration <= 0 or target_duration <= 0: + continue + + video_id = _resolve_script_video_id(item, video_paths) + video_cues = cues_by_video.get(video_id, []) + if not video_cues: + logger.warning(f"视频 {video_id} 未找到可用原片字幕，片段 {item.get('_id')} 跳过") + continue + + for cue in video_cues: + cue_start = cue.start_ms / 1000 + cue_end = cue.end_ms / 1000 + overlap_start = max(source_start, cue_start) + overlap_end = min(source_end, cue_end) + if overlap_end <= overlap_start: + continue + + text = clean_subtitle_text(cue.text) + if not text: + continue + + mapped_start = target_start + (overlap_start - source_start) + mapped_end = target_start + (overlap_end - source_start) + mapped_start = max(target_start, min(mapped_start, target_end)) + mapped_end = max(target_start, min(mapped_end, target_end)) + if mapped_end <= mapped_start: + continue + + entries.append((mapped_start, mapped_end, text)) + + return entries + + +def _subtitle_entries_to_blocks(entries: Sequence[SubtitleEntry]) -> List[str]: + blocks = [] + sorted_entries = sorted( + entries, + key=lambda entry: (entry[0], entry[1], entry[2]), + ) + + for subtitle_index, (start, end, text) in enumerate(sorted_entries, start=1): + blocks.append( + "\n".join( + [ + str(subtitle_index), + f"{format_srt_time(start)} --> {format_srt_time(end)}", + text, + ] ) - subtitle_index += 1 + ) return blocks +def _build_srt_blocks( + list_script: Sequence[dict], + include_ost: Iterable[int], + max_chars: int, +) -> List[str]: + entries = _build_narration_subtitle_entries( + list_script, + include_ost=include_ost, + max_chars=max_chars, + ) + return _subtitle_entries_to_blocks(entries) + + def create_script_subtitle_file( task_id: str, list_script: Sequence[dict], output_file: Optional[str] = None, include_ost: Optional[Iterable[int]] = None, max_chars: int = DEFAULT_MAX_CHARS_PER_SUBTITLE, + original_subtitle_paths=None, + video_origin_paths=None, + include_original_ost: Optional[Iterable[int]] = None, ) -> str: - """Create a full SRT file from script narration and edited timeline ranges.""" + """Create a full SRT file from script narration plus original-audio subtitles.""" if not list_script: return "" if include_ost is None: include_ost = DEFAULT_SUBTITLE_OST_TYPES + if include_original_ost is None: + include_original_ost = DEFAULT_ORIGINAL_SUBTITLE_OST_TYPES - blocks = _build_srt_blocks(list_script, include_ost=include_ost, max_chars=max_chars) + entries = _build_narration_subtitle_entries( + list_script, + include_ost=include_ost, + max_chars=max_chars, + ) + entries.extend( + _build_original_subtitle_entries( + list_script, + original_subtitle_paths=original_subtitle_paths, + video_origin_paths=video_origin_paths, + include_ost=include_original_ost, + ) + ) + + blocks = _subtitle_entries_to_blocks(entries) if not blocks: logger.warning("程序化字幕未生成内容") return "" diff --git a/app/services/task.py b/app/services/task.py index f5f60ce..d7aa1c9 100644 --- a/app/services/task.py +++ b/app/services/task.py @@ -38,6 +38,125 @@ def _get_auto_transcription_backend(params: VideoClipParams) -> str: return backend +def _get_original_subtitle_paths(params: VideoClipParams) -> list[str]: + subtitle_paths = getattr(params, "original_subtitle_paths", []) or [] + if isinstance(subtitle_paths, str): + subtitle_paths = [subtitle_paths] + + normalized_paths = [] + seen = set() + for subtitle_path in subtitle_paths: + if not isinstance(subtitle_path, str): + continue + subtitle_path = subtitle_path.strip() + if subtitle_path and subtitle_path not in seen: + normalized_paths.append(subtitle_path) + seen.add(subtitle_path) + + single_subtitle_path = str(getattr(params, "original_subtitle_path", "") or "").strip() + if single_subtitle_path and single_subtitle_path not in seen: + normalized_paths.insert(0, single_subtitle_path) + + if not normalized_paths: + normalized_paths = _find_original_subtitle_paths_for_videos(_get_video_origin_paths(params)) + + return normalized_paths + + +def _get_video_origin_paths(params: VideoClipParams) -> list[str]: + video_paths = getattr(params, "video_origin_paths", []) or [] + if isinstance(video_paths, str): + video_paths = [video_paths] + + normalized_paths = [] + seen = set() + for video_path in video_paths: + if not isinstance(video_path, str): + continue + video_path = video_path.strip() + if video_path and video_path not in seen: + normalized_paths.append(video_path) + seen.add(video_path) + + single_video_path = str(getattr(params, "video_origin_path", "") or "").strip() + if single_video_path and single_video_path not in seen: + normalized_paths.insert(0, single_video_path) + + return normalized_paths + + +def _video_stem_candidates(video_path: str) -> list[str]: + stem = path.splitext(path.basename(str(video_path or "").strip()))[0] + if not stem: + return [] + + candidates = [stem] + timestamp_stripped = re.sub(r"_[0-9]{14}$", "", stem) + if timestamp_stripped and timestamp_stripped not in candidates: + candidates.append(timestamp_stripped) + return candidates + + +def _find_original_subtitle_paths_for_videos(video_paths: list[str]) -> list[str]: + subtitle_dir = utils.subtitle_dir() + if not path.isdir(subtitle_dir): + return [] + + subtitle_files = [ + path.join(subtitle_dir, filename) + for filename in os.listdir(subtitle_dir) + if filename.lower().endswith(".srt") + ] + if not subtitle_files: + return [] + + resolved_paths = [] + seen = set() + for video_path in video_paths: + candidates = _video_stem_candidates(video_path) + if not candidates: + continue + + matches = [] + for subtitle_path in subtitle_files: + subtitle_stem = path.splitext(path.basename(subtitle_path))[0] + for candidate in candidates: + if subtitle_stem == candidate or subtitle_stem.startswith(f"{candidate}_"): + matches.append(subtitle_path) + break + + if not matches: + continue + + matches.sort(key=lambda item: path.getmtime(item), reverse=True) + selected_path = matches[0] + if selected_path not in seen: + resolved_paths.append(selected_path) + seen.add(selected_path) + + if resolved_paths: + logger.info(f"未从参数获取原片字幕，已按视频文件名自动匹配: {resolved_paths}") + return resolved_paths + + +def _create_programmatic_subtitle_file( + task_id: str, + list_script: list[dict], + params: VideoClipParams, +) -> str: + if not getattr(params, "subtitle_enabled", True): + return "" + + original_subtitle_paths = _get_original_subtitle_paths(params) + logger.info(f"程序化字幕使用原片字幕路径: {original_subtitle_paths or '未提供'}") + return script_subtitle.create_script_subtitle_file( + task_id=task_id, + list_script=list_script, + original_subtitle_paths=original_subtitle_paths, + video_origin_paths=_get_video_origin_paths(params), + ) + + def _build_subtitle_mask_options(params: VideoClipParams, enabled=None) -> dict: mask_configured = bool( getattr(params, "subtitle_enabled", True) @@ -279,7 +398,19 @@ def start_subclip(task_id: str, params: VideoClipParams, subclip_path_videos: di logger.info(f"音频文件合并成功->{merged_audio_path}") # 合并字幕文件 - merged_subtitle_path = subtitle_merger.merge_subtitle_files(new_script_list) + merged_subtitle_path = "" + if getattr(params, "subtitle_enabled", True): + try: + merged_subtitle_path = _create_programmatic_subtitle_file( + task_id, + new_script_list, + params, + ) + except Exception as e: + logger.warning(f"程序化字幕生成失败，将尝试合并TTS字幕: {e}") + + if not merged_subtitle_path and getattr(params, "subtitle_enabled", True): + merged_subtitle_path = subtitle_merger.merge_subtitle_files(new_script_list) if merged_subtitle_path: logger.info(f"字幕文件合并成功->{merged_subtitle_path}") else: @@ -296,6 +427,15 @@ def start_subclip(task_id: str, params: VideoClipParams, subclip_path_videos: di logger.warning("没有需要合并的音频/字幕") merged_audio_path = "" merged_subtitle_path = "" + if getattr(params, "subtitle_enabled", True): + try: + merged_subtitle_path = _create_programmatic_subtitle_file( + task_id, + new_script_list, + params, + ) + except Exception as e: + logger.warning(f"程序化字幕生成失败: {e}") """ 5. 合并视频 @@ -574,9 +714,10 @@ def start_subclip_unified(task_id: str, params: VideoClipParams): merged_subtitle_path = "" if getattr(params, "subtitle_enabled", True): try: - merged_subtitle_path = script_subtitle.create_script_subtitle_file( - task_id=task_id, - list_script=new_script_list, + merged_subtitle_path = _create_programmatic_subtitle_file( + task_id, + new_script_list, + params, ) except Exception as e: logger.warning(f"程序化字幕生成失败，将尝试合并TTS字幕: {e}") @@ -600,6 +741,15 @@ def start_subclip_unified(task_id: str, params: VideoClipParams): logger.warning("没有需要合并的音频/字幕") merged_audio_path = "" merged_subtitle_path = "" + if getattr(params, "subtitle_enabled", True): + try: + merged_subtitle_path = _create_programmatic_subtitle_file( + task_id, + new_script_list, + params, + ) + except Exception as e: + logger.warning(f"程序化字幕生成失败: {e}") sm.state.update_task( task_id, state=const.TASK_STATE_PROCESSING, diff --git a/app/services/test_merger_video_concat_unittest.py b/app/services/test_merger_video_concat_unittest.py new file mode 100644 index 0000000..1297f76 --- /dev/null +++ b/app/services/test_merger_video_concat_unittest.py @@ -0,0 +1,120 @@ +import subprocess +import unittest +from unittest import mock + +from app.services import merger_video + + +class MergerVideoConcatTests(unittest.TestCase): + def test_can_concat_video_copy_when_signatures_match(self): + signature = { + "codec_name": "h264", + "profile": "High", + "width": 1080, + "height": 1920, + "pix_fmt": "yuv420p", + "r_frame_rate": "30/1", + "avg_frame_rate": "30/1", + "time_base": "1/15360", + "sample_aspect_ratio": "1:1", + } + + with mock.patch.object( + merger_video, + "_get_video_stream_signature", + side_effect=[signature, dict(signature)], + ): + self.assertTrue(merger_video._can_concat_video_copy(["1.mp4", "2.mp4"])) + + def test_can_concat_video_copy_rejects_mismatched_signature(self): + base_signature = { + "codec_name": "h264", + "profile": "High", + "width": 1080, + "height": 1920, + "pix_fmt": "yuv420p", + "r_frame_rate": "30/1", + "avg_frame_rate": "30/1", + "time_base": "1/15360", + "sample_aspect_ratio": "1:1", + } + mismatch_signature = dict(base_signature, r_frame_rate="24000/1001") + + with mock.patch.object( + merger_video, + "_get_video_stream_signature", + side_effect=[base_signature, mismatch_signature], + ): + self.assertFalse(merger_video._can_concat_video_copy(["1.mp4", "2.mp4"])) + + def test_concat_video_streams_prefers_copy_when_compatible(self): + completed = subprocess.CompletedProcess(args=["ffmpeg"], returncode=0) + + with ( + mock.patch.object(merger_video, "_can_concat_video_copy", return_value=True), + mock.patch.object(merger_video, "_concat_duration_matches", return_value=True), + mock.patch.object(merger_video.subprocess, "run", return_value=completed) as run_mock, + ): + merger_video._concat_video_streams( + ["1.mp4", "2.mp4"], + "concat.txt", + "video_concat.mp4", + threads=4, + ) + + cmd = run_mock.call_args.args[0] + self.assertEqual("copy", cmd[cmd.index("-c:v") + 1]) + self.assertNotIn("libx264", cmd) + + def test_concat_video_streams_falls_back_when_copy_duration_mismatches(self): + completed = subprocess.CompletedProcess(args=["ffmpeg"], returncode=0) + + with ( + mock.patch.object(merger_video, "_can_concat_video_copy", return_value=True), + mock.patch.object(merger_video, "_concat_duration_matches", return_value=False), + mock.patch.object(merger_video.os.path, "exists", return_value=False), + mock.patch.object(merger_video.subprocess, "run", return_value=completed) as run_mock, + ): + merger_video._concat_video_streams( + ["1.mp4", "2.mp4"], + "concat.txt", + "video_concat.mp4", + threads=6, + ) + + self.assertEqual(2, run_mock.call_count) + fallback_cmd = run_mock.call_args_list[1].args[0] + self.assertEqual("libx264", fallback_cmd[fallback_cmd.index("-c:v") + 1]) + self.assertEqual("6", fallback_cmd[fallback_cmd.index("-threads") + 1]) + + def test_concat_video_streams_falls_back_to_reencode_when_copy_fails(self): + copy_error = subprocess.CalledProcessError( + returncode=1, + cmd=["ffmpeg"], + stderr=b"copy failed", + ) + completed = subprocess.CompletedProcess(args=["ffmpeg"], returncode=0) + + with ( + mock.patch.object(merger_video, "_can_concat_video_copy", return_value=True), + mock.patch.object( + merger_video.subprocess, + "run", + side_effect=[copy_error, completed], + ) as run_mock, + ): + merger_video._concat_video_streams( + ["1.mp4", "2.mp4"], + "concat.txt", + "video_concat.mp4", + threads=8, + ) + + self.assertEqual(2, run_mock.call_count) + fallback_cmd = run_mock.call_args_list[1].args[0] + self.assertEqual("libx264", fallback_cmd[fallback_cmd.index("-c:v") + 1]) + self.assertEqual("8", fallback_cmd[fallback_cmd.index("-threads") + 1]) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/services/test_script_subtitle_unittest.py b/app/services/test_script_subtitle_unittest.py index a4eed37..7b8bdaf 100644 --- a/app/services/test_script_subtitle_unittest.py +++ b/app/services/test_script_subtitle_unittest.py @@ -89,6 +89,104 @@ class ScriptSubtitleTests(unittest.TestCase): self.assertIn("00:00:00,000 -->", content) self.assertIn("--> 00:00:03,000", content) + def test_create_script_subtitle_file_includes_original_audio_subtitles(self): + list_script = [ + { + "_id": 1, + "OST": 0, + "narration": "前情解说。", + "editedTimeRange": "00:00:00-00:00:02", + "duration": 2, + }, + { + "_id": 2, + "video_id": 1, + "video_name": "source.mp4", + "OST": 1, + "narration": "播放原片2", + "timestamp": "00:00:10,000-00:00:14,000", + "sourceTimeRange": "00:00:10,000-00:00:14,000", + "editedTimeRange": "00:00:02-00:00:06", + "duration": 4, + }, + ] + original_srt = """1 +00:00:09,000 --> 00:00:11,000 +开头会被裁掉一秒。 + +2 +00:00:11,500 --> 00:00:13,000 +这句原声对白应该出现！ + +3 +00:00:13,500 --> 00:00:15,000 +结尾只保留半秒。 +""" + + with tempfile.TemporaryDirectory() as temp_dir: + subtitle_file = Path(temp_dir) / "source.srt" + subtitle_file.write_text(original_srt, encoding="utf-8") + output_file = Path(temp_dir) / "script_subtitles.srt" + script_subtitle.create_script_subtitle_file( + task_id="test", + list_script=list_script, + output_file=str(output_file), + original_subtitle_paths=[str(subtitle_file)], + video_origin_paths=["source.mp4"], + max_chars=16, + ) + content = output_file.read_text(encoding="utf-8") + + self.assertIn("前情解说", content) + self.assertIn("开头会被裁掉一秒", content) + self.assertIn("这句原声对白应该出现", content) + self.assertIn("结尾只保留半秒", content) + self.assertIn("00:00:02,000 --> 00:00:03,000", content) + self.assertIn("00:00:03,500 --> 00:00:05,000", content) + self.assertIn("00:00:05,500 --> 00:00:06,000", content) + self.assertNotIn("播放原片2", content) + + def test_create_script_subtitle_file_uses_matching_video_id_for_original_subtitles(self): + list_script = [ + { + "_id": 1, + "video_id": 2, + "video_name": "second.mp4", + "OST": 1, + "narration": "播放原片1", + "timestamp": "00:00:01,000-00:00:03,000", + "sourceTimeRange": "00:00:01,000-00:00:03,000", + "editedTimeRange": "00:00:00-00:00:02", + "duration": 2, + }, + ] + first_srt = """1 +00:00:01,000 --> 00:00:03,000 +第一个视频的字幕不应该出现。 +""" + second_srt = """1 +00:00:01,000 --> 00:00:03,000 +第二个视频的字幕应该出现。 +""" + + with tempfile.TemporaryDirectory() as temp_dir: + first_file = Path(temp_dir) / "first.srt" + second_file = Path(temp_dir) / "second.srt" + output_file = Path(temp_dir) / "script_subtitles.srt" + first_file.write_text(first_srt, encoding="utf-8") + second_file.write_text(second_srt, encoding="utf-8") + script_subtitle.create_script_subtitle_file( + task_id="test", + list_script=list_script, + output_file=str(output_file), + original_subtitle_paths=[str(first_file), str(second_file)], + video_origin_paths=["first.mp4", "second.mp4"], + ) + content = output_file.read_text(encoding="utf-8") + + self.assertIn("第二个视频的字幕应该出现", content) + self.assertNotIn("第一个视频的字幕不应该出现", content) + if __name__ == "__main__": unittest.main() diff --git a/app/services/test_task_subtitle_resolution_unittest.py b/app/services/test_task_subtitle_resolution_unittest.py new file mode 100644 index 0000000..b7744b2 --- /dev/null +++ b/app/services/test_task_subtitle_resolution_unittest.py @@ -0,0 +1,46 @@ +import tempfile +import time +import unittest +from pathlib import Path + +from app.models.schema import VideoClipParams +from app.services import task + + +class TaskSubtitleResolutionTests(unittest.TestCase): + def test_get_original_subtitle_paths_falls_back_to_matching_video_name(self): + original_subtitle_dir = task.utils.subtitle_dir + + with tempfile.TemporaryDirectory() as temp_dir: + temp_path = Path(temp_dir) + older = temp_path / "01_1080p_fun_asr.srt" + newer = temp_path / "01_1080p_fun_asr_20260608010240.srt" + unrelated = temp_path / "other_fun_asr.srt" + older.write_text("older", encoding="utf-8") + unrelated.write_text("other", encoding="utf-8") + time.sleep(0.01) + newer.write_text("newer", encoding="utf-8") + + task.utils.subtitle_dir = lambda: str(temp_path) + params = VideoClipParams( + video_origin_path="/tmp/01_1080p_20260608113314.mp4", + ) + + try: + subtitle_paths = task._get_original_subtitle_paths(params) + finally: + task.utils.subtitle_dir = original_subtitle_dir + + self.assertEqual([str(newer)], subtitle_paths) + + def test_get_original_subtitle_paths_keeps_explicit_params(self): + params = VideoClipParams( + video_origin_path="/tmp/01_1080p_20260608113314.mp4", + original_subtitle_paths=["/tmp/provided.srt"], + ) + + self.assertEqual(["/tmp/provided.srt"], task._get_original_subtitle_paths(params)) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui.py b/webui.py index 57f8eb7..bf9dd71 100644 --- a/webui.py +++ b/webui.py @@ -328,11 +328,22 @@ def get_jianying_export_params(draft_name=None) -> VideoClipParams: voice_name = get_voice_name_for_tts_engine(tts_engine) voice_rate = st.session_state.get('voice_rate', 1.0) voice_pitch = st.session_state.get('voice_pitch', 1.0) + subtitle_paths = st.session_state.get('subtitle_paths', []) + if isinstance(subtitle_paths, str): + subtitle_paths = [subtitle_paths] + subtitle_paths = [ + path for path in subtitle_paths + if isinstance(path, str) and path.strip() + ] + if not subtitle_paths and st.session_state.get('subtitle_path'): + subtitle_paths = [st.session_state.get('subtitle_path')] return VideoClipParams( video_clip_json_path=st.session_state['video_clip_json_path'], video_origin_path=st.session_state['video_origin_path'], video_origin_paths=st.session_state.get('video_origin_paths', []), + original_subtitle_path=subtitle_paths[0] if subtitle_paths else "", + original_subtitle_paths=subtitle_paths, tts_engine=tts_engine, voice_name=voice_name, voice_rate=voice_rate, diff --git a/webui/components/script_settings.py b/webui/components/script_settings.py index e57c42d..5d68dd6 100644 --- a/webui/components/script_settings.py +++ b/webui/components/script_settings.py @@ -1718,11 +1718,14 @@ def save_script_with_validation(tr, video_clip_json_details): def get_script_params(): """获取脚本参数""" + subtitle_paths = _selected_subtitle_paths() return { 'video_language': st.session_state.get('video_language', ''), 'video_clip_json_path': st.session_state.get('video_clip_json_path', ''), 'video_origin_path': st.session_state.get('video_origin_path', ''), 'video_origin_paths': _selected_video_paths(), + 'original_subtitle_path': subtitle_paths[0] if subtitle_paths else '', + 'original_subtitle_paths': subtitle_paths, 'video_name': st.session_state.get('video_name', ''), 'video_plot': st.session_state.get('video_plot', '') }