feat: 新增原片字幕支持并优化视频合并流程

- 为VideoClipParams新增原字幕路径配置字段，支持单条/多条字幕路径
- 完善webui参数获取逻辑，处理字幕路径兼容性并对接前端选择
- 重构后端字幕处理流程，支持自动匹配视频对应原字幕，合并原声字幕
- 优化视频合并逻辑，新增ffmpeg无损copy合并判断，自动回退重编码提升效率
- 新增ffmpeg快速素材合并路径，支持自定义字幕样式与多音轨混合
- 新增多个单元测试覆盖字幕匹配、合并及视频合并场景

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="视频语言")
 

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"]]

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 ""

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,

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()

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()

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()

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,

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', '')
     }