from __future__ import annotations import json import os import shutil import subprocess import tempfile import uuid from pathlib import Path import gradio as gr from shared.utils.audio_video import get_hdr_video_encode_args, get_video_encode_args from shared.utils.video_codecs import normalize_video_container, validate_video_output_settings from shared.utils.utils import get_video_info_details from shared.utils.video_decode import resolve_media_binary from shared.utils.video_metadata import DEFAULT_RESERVED_VIDEO_METADATA_BYTES, write_reserved_video_ffmetadata from . import constants class ContinuationMergeOutputLockedError(PermissionError): def __init__(self, output_path: str) -> None: self.output_path = str(output_path or "") super().__init__(f"Unable to replace locked output file: {self.output_path}") def probe_resume_frame_count(ffprobe_path: str, output_path: str, fps_float: float) -> tuple[int, str]: probe = subprocess.run([ffprobe_path, "-v", "error", "-select_streams", "v:0", "-count_packets", "-show_entries", "stream=nb_read_packets", "-of", "json", output_path], capture_output=True, text=True, encoding="utf-8", errors="ignore", check=False) if probe.returncode == 0: try: frame_count = int((((json.loads(probe.stdout).get("streams") or [{}])[0]).get("nb_read_packets")) or 0) except (TypeError, ValueError, json.JSONDecodeError, IndexError): frame_count = 0 if frame_count > 0: return frame_count, "" probe = subprocess.run([ffprobe_path, "-v", "error", "-select_streams", "v:0", "-count_frames", "-show_entries", "stream=nb_read_frames", "-of", "json", output_path], capture_output=True, text=True, encoding="utf-8", errors="ignore", check=False) if probe.returncode == 0: try: frame_count = int((((json.loads(probe.stdout).get("streams") or [{}])[0]).get("nb_read_frames")) or 0) except (TypeError, ValueError, json.JSONDecodeError, IndexError): frame_count = 0 if frame_count > 0: return frame_count, "" metadata = get_video_info_details(output_path) frame_count = int(metadata.get("frame_count") or 0) if frame_count > 0: return frame_count, "" duration = float(metadata.get("duration") or 0.0) if duration > 0 and fps_float > 0: return int(round(duration * fps_float)), "" stderr = (probe.stderr or "").strip() return 0, stderr or "existing output contains no readable frame count or duration metadata" def normalize_container_name(video_container: str | None) -> str: return normalize_video_container(video_container) SUPPORTED_OUTPUT_CONTAINERS = constants.SUPPORTED_OUTPUT_CONTAINERS LIVE_AUDIO_LOOP_PADDING_SECONDS = 5.0 def is_supported_output_container(video_container: str | None) -> bool: return normalize_container_name(video_container) in SUPPORTED_OUTPUT_CONTAINERS def is_quicktime_output_container(video_container: str | None) -> bool: return normalize_container_name(video_container) in {"mp4", "mov"} def _get_live_mux_output_args(video_container: str | None) -> list[str]: video_container = normalize_container_name(video_container) if video_container == "mkv": return ["-fflags", "+flush_packets", "-flush_packets", "1", "-f", "matroska", "-live", "1"] if is_quicktime_output_container(video_container): return ["-movflags", "+frag_keyframe+empty_moov+default_base_moof"] return [] def _probe_media_duration(ffprobe_path: str, media_path: str) -> float: result = subprocess.run([ffprobe_path, "-v", "error", "-show_entries", "format=duration", "-of", "json", media_path], capture_output=True, text=True, encoding="utf-8", errors="ignore", check=False) if result.returncode != 0: return 0.0 try: return max(0.0, float((((json.loads(result.stdout) or {}).get("format") or {}).get("duration")) or 0.0)) except (TypeError, ValueError, json.JSONDecodeError): return 0.0 def _probe_audio_end_time(ffprobe_path: str, media_path: str, audio_index: int) -> float: stream_selector = f"a:{max(0, int(audio_index))}" result = subprocess.run([ffprobe_path, "-v", "error", "-select_streams", stream_selector, "-show_entries", "stream=start_time,duration", "-of", "json", media_path], capture_output=True, text=True, encoding="utf-8", errors="ignore", check=False) if result.returncode == 0: try: stream = (((json.loads(result.stdout) or {}).get("streams") or [{}])[0]) start_time = 0.0 if stream.get("start_time") in (None, "", "N/A") else max(0.0, float(stream.get("start_time"))) duration_time = 0.0 if stream.get("duration") in (None, "", "N/A") else max(0.0, float(stream.get("duration"))) if duration_time > 0.0: return start_time + duration_time except (TypeError, ValueError, json.JSONDecodeError, IndexError): pass approximate_end = _probe_media_duration(ffprobe_path, media_path) if approximate_end <= 0.0: return 0.0 for window_seconds in (2.0, 8.0, 32.0, 128.0): probe_start = max(0.0, approximate_end - window_seconds) probe_span = max(0.25, approximate_end - probe_start + 0.25) result = subprocess.run( [ffprobe_path, "-v", "error", "-select_streams", stream_selector, "-read_intervals", f"{probe_start:.6f}%+{probe_span:.6f}", "-show_packets", "-show_entries", "packet=pts_time,duration_time", "-of", "csv=p=0", media_path], capture_output=True, text=True, encoding="utf-8", errors="ignore", check=False, ) if result.returncode != 0: continue last_end = 0.0 for raw_line in result.stdout.splitlines(): fields = [field.strip() for field in str(raw_line or "").strip().split(",")] if len(fields) <= 0 or len(fields[0]) == 0: continue try: pts_time = float(fields[0]) except (TypeError, ValueError): continue try: duration_time = float(fields[1]) if len(fields) > 1 and len(fields[1]) > 0 else 0.0 except (TypeError, ValueError): duration_time = 0.0 last_end = max(last_end, pts_time, pts_time + max(0.0, duration_time)) if last_end > 0.0: return last_end return approximate_end def _probe_selected_audio_end_time(ffprobe_path: str, media_path: str, audio_track_no: int | None) -> float: _, audio_stream_count = _probe_media_stream_layout(ffprobe_path, media_path) if audio_stream_count <= 0: return 0.0 if audio_track_no is None: audio_indices = range(audio_stream_count) else: audio_indices = [max(0, min(audio_stream_count - 1, int(audio_track_no) - 1))] return max((_probe_audio_end_time(ffprobe_path, media_path, audio_index) for audio_index in audio_indices), default=0.0) def probe_selected_audio_overhang(ffprobe_path: str, media_path: str, audio_track_no: int | None, visible_duration_seconds: float) -> float: visible_duration_seconds = max(0.0, float(visible_duration_seconds or 0.0)) _, audio_stream_count = _probe_media_stream_layout(ffprobe_path, media_path) if audio_stream_count <= 0: return 0.0 if audio_track_no is None: audio_indices = range(audio_stream_count) else: audio_indices = [max(0, min(audio_stream_count - 1, int(audio_track_no) - 1))] video_end_seconds = _probe_primary_video_start_time(ffprobe_path, media_path) + visible_duration_seconds seam_start = max(0.0, video_end_seconds - 1.0) for probe_span in (2.0, 8.0, 32.0, 128.0): last_end = 0.0 for audio_index in audio_indices: result = subprocess.run( [ ffprobe_path, "-v", "error", "-select_streams", f"a:{int(audio_index)}", "-read_intervals", f"{seam_start:.6f}%+{probe_span:.6f}", "-show_packets", "-show_entries", "packet=pts_time,duration_time", "-of", "csv=p=0", media_path, ], capture_output=True, text=True, encoding="utf-8", errors="ignore", check=False, ) if result.returncode != 0: continue for raw_line in result.stdout.splitlines(): fields = [field.strip() for field in str(raw_line or "").strip().split(",")] if len(fields) <= 0 or len(fields[0]) == 0: continue try: pts_time = float(fields[0]) except (TypeError, ValueError): continue try: duration_time = float(fields[1]) if len(fields) > 1 and len(fields[1]) > 0 else 0.0 except (TypeError, ValueError): duration_time = 0.0 last_end = max(last_end, pts_time, pts_time + max(0.0, duration_time)) if last_end > video_end_seconds: return max(0.0, last_end - video_end_seconds) return max(0.0, _probe_selected_audio_end_time(ffprobe_path, media_path, audio_track_no) - video_end_seconds) def _probe_audio_stream_codecs(ffprobe_path: str, media_path: str) -> list[str]: result = subprocess.run([ffprobe_path, "-v", "error", "-select_streams", "a", "-show_entries", "stream=index,codec_name", "-of", "json", media_path], capture_output=True, text=True, encoding="utf-8", errors="ignore", check=False) if result.returncode != 0: raise gr.Error((result.stderr or result.stdout or f"ffprobe failed for {media_path}").strip()) try: streams = list((json.loads(result.stdout) or {}).get("streams") or []) except json.JSONDecodeError as exc: raise gr.Error(f"Unable to read audio codecs for {media_path}") from exc return [str(stream.get("codec_name") or "").strip().lower() for stream in streams] def validate_audio_copy_container(ffprobe_path: str, source_path: str, video_container: str, audio_track_no: int | None) -> None: video_container = normalize_container_name(video_container) if video_container not in {"mp4", "mov"}: return supported_codecs = {"aac", "ac3", "alac", "eac3", "mp3", "opus"} if video_container == "mp4" else {"aac", "ac3", "alac", "eac3", "mp3", "pcm_s16le", "pcm_s24le", "pcm_s32le"} audio_codecs = _probe_audio_stream_codecs(ffprobe_path, source_path) if audio_track_no is None: selected_codecs = [codec for codec in audio_codecs if len(codec) > 0] else: selected_index = max(0, int(audio_track_no) - 1) selected_codecs = [audio_codecs[selected_index]] if selected_index < len(audio_codecs) and len(audio_codecs[selected_index]) > 0 else [] incompatible_codecs = [codec for codec in selected_codecs if codec not in supported_codecs] if len(incompatible_codecs) > 0: track_label = f"audio track {int(audio_track_no)}" if audio_track_no is not None else "the selected audio tracks" codecs_text = ", ".join(sorted(set(incompatible_codecs))) raise gr.Error(f"{video_container.upper()} output cannot packet-copy {track_label} with codec(s): {codecs_text}. Use an .mkv output path or choose a compatible track.") def validate_output_codec_container(video_codec: str, video_container: str, audio_codec: str | None = None, width: int | None = None, height: int | None = None, output_path: str | None = None) -> None: error = validate_video_output_settings(video_codec, video_container, audio_codec=audio_codec, width=width, height=height, allowed_containers=SUPPORTED_OUTPUT_CONTAINERS) if error is None: return suffix = Path(output_path).suffix if output_path else f".{normalize_container_name(video_container)}" raise gr.Error(f"Output file extension '{suffix}' is incompatible with the current video codec/container settings. {error} Choose a compatible output extension or update the Configuration plugin settings.") def start_video_mux_process(ffmpeg_path: str, output_path: str, width: int, height: int, fps_float: float, video_codec: str, video_container: str, reserved_metadata_path: str | None = None) -> subprocess.Popen: validate_output_codec_container(video_codec, video_container, width=width, height=height, output_path=output_path) command = [ ffmpeg_path, "-y", "-v", "error", "-f", "rawvideo", "-pix_fmt", "rgb24", "-video_size", f"{width}x{height}", "-framerate", f"{float(fps_float):.12g}", "-i", "pipe:0", ] metadata_input_index = None if reserved_metadata_path and os.path.isfile(reserved_metadata_path): command += ["-f", "ffmetadata", "-i", reserved_metadata_path] metadata_input_index = 1 if metadata_input_index is not None: command += ["-map_metadata", str(metadata_input_index)] command += ["-map", "0:v:0"] command += get_video_encode_args(video_codec, video_container) command += _get_live_mux_output_args(video_container) command += [output_path] return subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, bufsize=0) def start_av_mux_process(ffmpeg_path: str, output_path: str, width: int, height: int, fps_float: float, video_codec: str, video_container: str, source_path: str, start_seconds: float, audio_track_no: int | None, reserved_metadata_path: str | None = None, source_audio_duration_seconds: float | None = None) -> subprocess.Popen: validate_output_codec_container(video_codec, video_container, width=width, height=height, output_path=output_path) command = [ ffmpeg_path, "-y", "-v", "error", "-f", "rawvideo", "-pix_fmt", "rgb24", "-video_size", f"{width}x{height}", "-framerate", f"{float(fps_float):.12g}", "-i", "pipe:0", "-stream_loop", "-1", "-ss", f"{max(0.0, float(start_seconds)):.12g}", ] if source_audio_duration_seconds is not None and source_audio_duration_seconds > 0.0: command += ["-t", f"{max(0.0, float(source_audio_duration_seconds) + LIVE_AUDIO_LOOP_PADDING_SECONDS):.12g}"] command += ["-i", source_path] metadata_input_index = None if reserved_metadata_path and os.path.isfile(reserved_metadata_path): command += ["-f", "ffmetadata", "-i", reserved_metadata_path] metadata_input_index = 2 else: metadata_input_index = 1 command += ["-map_metadata", str(metadata_input_index), "-map", "0:v:0"] if audio_track_no is None: command += ["-map", "1:a?"] else: command += ["-map", f"1:a:{max(0, int(audio_track_no) - 1)}?"] command += get_video_encode_args(video_codec, video_container) + ["-c:a", "copy", "-shortest"] command += _get_live_mux_output_args(video_container) command += [output_path] return subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, bufsize=0) def start_hdr_video_mux_process(ffmpeg_path: str, output_path: str, width: int, height: int, fps_float: float, video_codec: str, video_container: str, reserved_metadata_path: str | None = None) -> subprocess.Popen: command = [ ffmpeg_path, "-y", "-v", "error", "-f", "rawvideo", "-pix_fmt", "gbrpf32le", "-video_size", f"{width}x{height}", "-framerate", f"{float(fps_float):.12g}", "-i", "pipe:0", ] metadata_input_index = None if reserved_metadata_path and os.path.isfile(reserved_metadata_path): command += ["-f", "ffmetadata", "-i", reserved_metadata_path] metadata_input_index = 1 if metadata_input_index is not None: command += ["-map_metadata", str(metadata_input_index)] command += ["-map", "0:v:0"] command += get_hdr_video_encode_args(video_codec, video_container) command += _get_live_mux_output_args(video_container) command += [output_path] return subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, bufsize=0) def start_hdr_av_mux_process(ffmpeg_path: str, output_path: str, width: int, height: int, fps_float: float, video_codec: str, video_container: str, source_path: str, start_seconds: float, audio_track_no: int | None, reserved_metadata_path: str | None = None, source_audio_duration_seconds: float | None = None) -> subprocess.Popen: command = [ ffmpeg_path, "-y", "-v", "error", "-f", "rawvideo", "-pix_fmt", "gbrpf32le", "-video_size", f"{width}x{height}", "-framerate", f"{float(fps_float):.12g}", "-i", "pipe:0", "-stream_loop", "-1", "-ss", f"{max(0.0, float(start_seconds)):.12g}", ] if source_audio_duration_seconds is not None and source_audio_duration_seconds > 0.0: command += ["-t", f"{max(0.0, float(source_audio_duration_seconds) + LIVE_AUDIO_LOOP_PADDING_SECONDS):.12g}"] command += ["-i", source_path] if reserved_metadata_path and os.path.isfile(reserved_metadata_path): command += ["-f", "ffmetadata", "-i", reserved_metadata_path, "-map_metadata", "2"] else: command += ["-map_metadata", "1"] command += ["-map", "0:v:0"] if audio_track_no is None: command += ["-map", "1:a?"] else: command += ["-map", f"1:a:{max(0, int(audio_track_no) - 1)}?"] command += get_hdr_video_encode_args(video_codec, video_container) + ["-c:a", "copy", "-shortest"] command += _get_live_mux_output_args(video_container) command += [output_path] return subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, bufsize=0) def finalize_mux_process(process: subprocess.Popen, *, timeout_seconds: float = 30.0) -> tuple[int, str, bool]: if process.stdin is not None and not process.stdin.closed: try: process.stdin.close() except OSError: pass forced_termination = False try: return_code = process.wait(timeout=timeout_seconds) except subprocess.TimeoutExpired: forced_termination = True process.kill() return_code = process.wait(timeout=5) stderr = process.stderr.read().decode("utf-8", errors="ignore").strip() if process.stderr is not None else "" return return_code, stderr, forced_termination def _reserve_sidecar_path(preferred_path: Path, fallback_path_factory) -> str: for attempt in range(1000): candidate = preferred_path if attempt == 0 else fallback_path_factory(uuid.uuid4().hex[:8]) try: handle = os.open(str(candidate), os.O_CREAT | os.O_EXCL | os.O_WRONLY) except FileExistsError: continue try: os.close(handle) except OSError: pass return str(candidate) raise gr.Error(f"Unable to reserve a scratch filename near {preferred_path.parent}") def reserve_video_only_output_path(output_path: str) -> str: output = Path(output_path).resolve() preferred = output.with_name(f"{output.stem}_videoonly{output.suffix}") return _reserve_sidecar_path(preferred, lambda token: output.with_name(f"{output.stem}_videoonly_{token}{output.suffix}")) def create_reserved_metadata_file(output_path: str) -> str: output = Path(output_path).resolve() preferred = output.with_name(f"{output.name}.ffmeta") reserved_metadata_path = _reserve_sidecar_path(preferred, lambda token: output.with_name(f"{output.name}.{token}.ffmeta")) write_reserved_video_ffmetadata(reserved_metadata_path, DEFAULT_RESERVED_VIDEO_METADATA_BYTES) return reserved_metadata_path def delete_file_if_exists(file_path: str | None, *, label: str) -> None: if not isinstance(file_path, str) or not os.path.isfile(file_path): return try: os.remove(file_path) except OSError as exc: print(f"[Process Full Video] Warning: failed to delete {label} {file_path}: {exc}") def delete_released_chunk_outputs(state: dict, chunk_output_paths: list[str], *, preserve_paths: list[str] | None = None) -> list[str]: if not isinstance(state, dict): return chunk_output_paths gen = state.get("gen", {}) if not isinstance(gen, dict): return chunk_output_paths referenced_paths = { str(Path(path).resolve()) for path in list(gen.get("file_list", []) or []) + list(gen.get("audio_file_list", []) or []) if isinstance(path, str) and len(path.strip()) > 0 } referenced_paths.update( str(Path(path).resolve()) for path in list(preserve_paths or []) if isinstance(path, str) and len(path.strip()) > 0 ) kept_paths: list[str] = [] for path in chunk_output_paths: resolved = str(Path(path).resolve()) if resolved in referenced_paths: kept_paths.append(resolved) continue if os.path.isfile(resolved): try: os.remove(resolved) except OSError: kept_paths.append(resolved) return kept_paths def get_last_generated_video_path(paths: list[str]) -> str | None: video_paths = [str(Path(path).resolve()) for path in paths if isinstance(path, str) and os.path.isfile(path) and str(Path(path).suffix).lower() in {".mp4", ".mkv", ".mov"}] return video_paths[-1] if len(video_paths) > 0 else None def _make_output_temp_dir(output_path: str, prefix: str) -> str: return tempfile.mkdtemp(prefix=prefix, dir=str(Path(output_path).resolve().parent)) def _probe_media_stream_layout(ffprobe_path: str, media_path: str) -> tuple[int, int]: result = subprocess.run([ffprobe_path, "-v", "error", "-show_entries", "stream=codec_type", "-of", "json", media_path], capture_output=True, text=True) if result.returncode != 0: raise gr.Error((result.stderr or result.stdout or f"ffprobe failed for {media_path}").strip()) try: streams = list((json.loads(result.stdout) or {}).get("streams") or []) except json.JSONDecodeError as exc: raise gr.Error(f"Unable to read media stream layout for {media_path}") from exc video_count = sum(1 for stream in streams if str(stream.get("codec_type") or "").lower() == "video") audio_count = sum(1 for stream in streams if str(stream.get("codec_type") or "").lower() == "audio") return video_count, audio_count def _probe_primary_video_codec(ffprobe_path: str, media_path: str) -> str: result = subprocess.run([ffprobe_path, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=codec_name", "-of", "json", media_path], capture_output=True, text=True, encoding="utf-8", errors="ignore", check=False) if result.returncode != 0: raise gr.Error((result.stderr or result.stdout or f"ffprobe failed for {media_path}").strip()) try: codec_name = str((((json.loads(result.stdout) or {}).get("streams") or [{}])[0]).get("codec_name") or "").strip().lower() except (TypeError, ValueError, json.JSONDecodeError, IndexError): codec_name = "" if len(codec_name) == 0: raise gr.Error(f"Unable to detect the video codec of {media_path}") return codec_name def _probe_primary_video_start_time(ffprobe_path: str, media_path: str) -> float: result = subprocess.run([ffprobe_path, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=start_time", "-of", "json", media_path], capture_output=True, text=True, encoding="utf-8", errors="ignore", check=False) if result.returncode == 0: try: stream = (((json.loads(result.stdout) or {}).get("streams") or [{}])[0]) start_time = stream.get("start_time") if start_time not in (None, "", "N/A"): return max(0.0, float(start_time)) except (TypeError, ValueError, json.JSONDecodeError, IndexError): pass packet_times = _probe_video_packet_times(ffprobe_path, media_path, start_seconds=0.0, duration_seconds=1.0) return max(0.0, min(packet_times)) if len(packet_times) > 0 else 0.0 def _probe_video_packet_times(ffprobe_path: str, media_path: str, *, start_seconds: float | None = None, duration_seconds: float | None = None) -> list[float]: command = [ffprobe_path, "-v", "error", "-select_streams", "v:0"] if start_seconds is not None and duration_seconds is not None and float(duration_seconds) > 0.0: command += ["-read_intervals", f"{max(0.0, float(start_seconds)):.6f}%+{max(0.05, float(duration_seconds)):.6f}"] command += ["-show_packets", "-show_entries", "packet=dts_time,pts_time", "-of", "json", media_path] result = subprocess.run(command, capture_output=True, text=True, encoding="utf-8", errors="ignore", check=False) if result.returncode != 0: raise gr.Error((result.stderr or result.stdout or f"ffprobe failed for {media_path}").strip()) try: return sorted( float(packet.get("dts_time") if packet.get("dts_time") is not None else packet.get("pts_time")) for packet in ((json.loads(result.stdout) or {}).get("packets") or []) if packet.get("dts_time") is not None or packet.get("pts_time") is not None ) except (TypeError, ValueError, json.JSONDecodeError): return [] def _probe_video_frame_gap(ffprobe_path: str, media_path: str, fps_float: float, *, near_time: float | None = None, window_seconds: float = 4.0) -> tuple[float, float, float] | None: fps_value = float(fps_float or 0.0) if fps_value <= 0.0 or not os.path.isfile(media_path): return None max_delta = max(1.0 / fps_value * 1.5, 0.05) if near_time is None: packet_times = _probe_video_packet_times(ffprobe_path, media_path) if len(packet_times) < 2: return None for current_pts, next_pts in zip(packet_times, packet_times[1:]): delta = float(next_pts) - float(current_pts) if delta > max_delta: return float(current_pts), float(next_pts), float(delta) return None seam_time = max(0.0, float(near_time)) local_window = max(1.0, float(window_seconds)) seam_margin = max_delta * 2.0 total_duration = max(seam_time, _probe_media_duration(ffprobe_path, media_path)) probe_start = max(0.0, seam_time - max(0.25, seam_margin)) probe_duration = min(max(local_window, 4.0), max(0.05, total_duration - probe_start)) while probe_duration > 0.0: packet_times = _probe_video_packet_times(ffprobe_path, media_path, start_seconds=probe_start, duration_seconds=probe_duration) prev_candidates = [packet_time for packet_time in packet_times if packet_time <= seam_time + seam_margin] if len(prev_candidates) > 0: prev_time = prev_candidates[-1] next_candidates = [packet_time for packet_time in packet_times if packet_time > prev_time + 1e-9] if len(next_candidates) > 0: next_time = next_candidates[0] delta = float(next_time) - float(prev_time) return None if delta <= max_delta else (float(prev_time), float(next_time), float(delta)) if probe_start + probe_duration >= total_duration - 1e-6: return None probe_duration = min(max(2.0 * probe_duration, local_window), max(0.05, total_duration - probe_start)) return None def _write_concat_list(list_path: str, media_paths: list[str]) -> None: with open(list_path, "w", encoding="utf-8") as handle: for media_path in media_paths: escaped_path = str(media_path).replace("'", "'\\''") handle.write(f"file '{escaped_path}'\n") def _split_sparse_fine_seek(seconds: float) -> tuple[float, float]: seconds = max(0.0, float(seconds or 0.0)) coarse_seconds = max(0.0, seconds - 1.0) return coarse_seconds, seconds - coarse_seconds def _build_video_reconstruct_bsf(fps_float: float) -> str: fps_value = max(float(fps_float or 0.0), 1.0) frame_duration_expr = f"1/({fps_value:.15g}*TB)" return ( "setts=" f"pts='if(eq(N,0),PTS,PREV_OUTPTS+(PTS-PREV_INPTS)-(PREV_INDURATION-DURATION))':" f"dts='if(eq(N,0),DTS,PREV_OUTDTS+(DTS-PREV_INDTS)-(PREV_INDURATION-DURATION))':" f"duration='if(eq(N,0),{frame_duration_expr},DURATION)'" ) def _concat_audio_segments(ffmpeg_path: str, segment_paths: list[str], output_path: str, work_dir: str, *, segment_trim_seconds: list[float] | None = None, segment_duration_seconds: list[float | None] | None = None, audio_stream_indices: list[int] | None = None) -> None: extracted_paths: list[str] = [] list_path = os.path.join(work_dir, "audio.txt") try: for segment_no, segment_path in enumerate(segment_paths, start=1): extracted_path = os.path.join(work_dir, f"segment_{segment_no}_audio.mka") trim_seconds = max(0.0, float(segment_trim_seconds[segment_no - 1])) if segment_trim_seconds is not None and segment_no - 1 < len(segment_trim_seconds) else 0.0 duration_seconds = None if segment_duration_seconds is not None and segment_no - 1 < len(segment_duration_seconds) and segment_duration_seconds[segment_no - 1] is not None: duration_seconds = max(0.0, float(segment_duration_seconds[segment_no - 1])) audio_stream_index = max(0, int(audio_stream_indices[segment_no - 1])) if audio_stream_indices is not None and segment_no - 1 < len(audio_stream_indices) else 0 command = [ffmpeg_path, "-y", "-v", "error"] coarse_seek_seconds, fine_seek_seconds = _split_sparse_fine_seek(trim_seconds) if coarse_seek_seconds > 0.0: command += ["-ss", f"{coarse_seek_seconds:.12g}"] command += ["-i", segment_path] if fine_seek_seconds > 0.0: command += ["-ss", f"{fine_seek_seconds:.12g}"] if duration_seconds is not None and duration_seconds > 0.0: command += ["-t", f"{duration_seconds:.12g}"] command += ["-map", f"0:a:{audio_stream_index}?", "-c", "copy", "-avoid_negative_ts", "make_zero", extracted_path] result = subprocess.run(command, capture_output=True, text=True) if result.returncode != 0 or not os.path.isfile(extracted_path): raise gr.Error((result.stderr or result.stdout or f"ffmpeg failed to extract audio from {segment_path}").strip()) extracted_paths.append(extracted_path) _write_concat_list(list_path, extracted_paths) result = subprocess.run([ffmpeg_path, "-y", "-v", "error", "-f", "concat", "-safe", "0", "-i", list_path, "-c", "copy", output_path], capture_output=True, text=True) if result.returncode != 0 or not os.path.isfile(output_path): raise gr.Error((result.stderr or result.stdout or "ffmpeg failed to concatenate audio streams").strip()) finally: for extracted_path in extracted_paths: if os.path.isfile(extracted_path): os.remove(extracted_path) if os.path.isfile(list_path): os.remove(list_path) def concat_video_segments( ffmpeg_path: str, segment_paths: list[str], output_path: str, video_codec: str, video_container: str, audio_codec_key: str, *, segment_audio_trim_seconds: list[float] | None = None, segment_audio_duration_seconds: list[float | None] | None = None, fps_float: float | None = None, selected_audio_track_no: int | None = None, reserved_metadata_path: str | None = None, source_audio_path: str | None = None, source_audio_start_seconds: float | None = None, source_audio_duration_seconds: float | None = None, source_audio_track_no: int | None = None, ) -> None: normalized_segment_paths: list[str] = [] for path in segment_paths: segment_path = str(Path(str(path).strip()).resolve()) if len(segment_path) == 0 or not os.path.isfile(segment_path): raise gr.Error(f"Output segment is missing and cannot be merged: {path}") normalized_segment_paths.append(segment_path) segment_paths = normalized_segment_paths if len(segment_paths) == 0: raise gr.Error("No output segments available to merge.") if len(segment_paths) == 1: if str(Path(segment_paths[0]).resolve()) != str(Path(output_path).resolve()): try: os.replace(segment_paths[0], output_path) except OSError as exc: raise ContinuationMergeOutputLockedError(output_path) from exc return ffprobe_path = resolve_media_binary("ffprobe") layouts = [_probe_media_stream_layout(ffprobe_path, path) for path in segment_paths] if any(video_count != 1 for video_count, _ in layouts): raise gr.Error("All continuation segments must contain exactly one video stream.") use_source_audio_merge = isinstance(source_audio_path, str) and len(str(source_audio_path).strip()) > 0 if use_source_audio_merge: validate_audio_copy_container(ffprobe_path, str(source_audio_path), video_container, source_audio_track_no) audio_stream_counts = [audio_count for _, audio_count in layouts] has_audio = use_source_audio_merge or any(audio_count > 0 for audio_count in audio_stream_counts) if not use_source_audio_merge and has_audio and any(audio_count <= 0 for audio_count in audio_stream_counts): raise gr.Error("All continuation segments must expose an audio stream.") fps_value = float(fps_float or 0.0) concat_dir = _make_output_temp_dir(output_path, "wangp_process_full_video_concat_") merged_video_path = os.path.join(concat_dir, "merged_video.mkv") temp_output_path = os.path.join(concat_dir, f"{Path(output_path).stem}_merged{Path(output_path).suffix}") try: video_codec_name = _probe_primary_video_codec(ffprobe_path, segment_paths[0]) video_bsf = "h264_mp4toannexb" if video_codec_name == "h264" else "hevc_mp4toannexb" if video_codec_name in ("hevc", "h265") else "" if len(video_bsf) == 0: raise gr.Error(f"Unsupported continuation video codec for no-reencode merge: {video_codec_name}") reconstructed_video_bsf = f"{_build_video_reconstruct_bsf(fps_value)},{video_bsf}" if use_source_audio_merge: command = [ffmpeg_path, "-y", "-v", "error", "-f", "mpegts", "-i", "pipe:0", "-ss", f"{max(0.0, float(source_audio_start_seconds or 0.0)):.12g}", "-t", f"{max(0.0, float(source_audio_duration_seconds or 0.0)):.12g}", "-i", str(source_audio_path)] if reserved_metadata_path and os.path.isfile(reserved_metadata_path): command += ["-f", "ffmetadata", "-i", reserved_metadata_path, "-map_metadata", "2"] else: command += ["-map_metadata", "1"] command += ["-map", "0:v:0"] if source_audio_track_no is None: command += ["-map", "1:a?"] else: command += ["-map", f"1:a:{max(0, int(source_audio_track_no) - 1)}?"] command += ["-c", "copy"] if is_quicktime_output_container(video_container): command += ["-movflags", "+faststart"] command += [temp_output_path] mux_process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, bufsize=0) try: if mux_process.stdin is None: raise gr.Error("ffmpeg source-audio merge did not expose a writable video pipe.") for segment_path in segment_paths: segment_process = subprocess.Popen([ffmpeg_path, "-v", "error", "-i", segment_path, "-map", "0:v:0", "-c", "copy", "-bsf:v", reconstructed_video_bsf, "-f", "mpegts", "-"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0) try: if segment_process.stdout is None: raise gr.Error(f"ffmpeg failed to expose the TS stream for {segment_path}.") shutil.copyfileobj(segment_process.stdout, mux_process.stdin, length=1024 * 1024) except Exception: segment_process.kill() segment_process.wait(timeout=5) raise finally: if segment_process.stdout is not None: segment_process.stdout.close() segment_returncode = segment_process.wait() segment_stderr = segment_process.stderr.read().decode("utf-8", errors="ignore").strip() if segment_process.stderr is not None else "" if segment_returncode != 0: raise gr.Error((segment_stderr or f"ffmpeg failed to stream {segment_path} for concat").strip()) mux_returncode, mux_stderr, _ = finalize_mux_process(mux_process) except Exception: try: if mux_process.stdin is not None and not mux_process.stdin.closed: mux_process.stdin.close() except OSError: pass mux_process.kill() mux_process.wait(timeout=5) raise if mux_returncode != 0 or not os.path.isfile(temp_output_path): raise gr.Error((mux_stderr or "ffmpeg source-audio merge failed").strip()) else: concat_ts_path = os.path.join(concat_dir, "segments.ts") ts_paths: list[str] = [] for index, segment_path in enumerate(segment_paths, start=1): ts_path = os.path.join(concat_dir, f"segment_{index}.ts") result = subprocess.run([ffmpeg_path, "-y", "-v", "error", "-i", segment_path, "-map", "0:v:0", "-c", "copy", "-bsf:v", reconstructed_video_bsf, "-f", "mpegts", ts_path], capture_output=True, text=True) if result.returncode != 0 or not os.path.isfile(ts_path): raise gr.Error((result.stderr or result.stdout or f"ffmpeg failed to prepare {segment_path} for concat").strip()) ts_paths.append(ts_path) with open(concat_ts_path, "wb") as handle: for ts_path in ts_paths: with open(ts_path, "rb") as ts_file: shutil.copyfileobj(ts_file, handle, length=1024 * 1024) result = subprocess.run([ffmpeg_path, "-y", "-v", "error", "-i", concat_ts_path, "-map", "0:v:0", "-c", "copy", merged_video_path], capture_output=True, text=True) if result.returncode != 0 or not os.path.isfile(merged_video_path): raise gr.Error((result.stderr or result.stdout or "ffmpeg failed to concatenate video stream").strip()) command = [ffmpeg_path, "-y", "-v", "error", "-i", merged_video_path] if has_audio: merged_audio_path = os.path.join(concat_dir, "merged_audio.mka") selected_audio_index = max(0, int(selected_audio_track_no or 1) - 1) audio_stream_indices = [0 if audio_count <= 1 else min(audio_count - 1, selected_audio_index) for audio_count in audio_stream_counts] resolved_audio_duration_seconds = [None] * len(segment_paths) if segment_audio_duration_seconds is not None: for segment_index, duration_seconds in enumerate(segment_audio_duration_seconds[:len(segment_paths)]): if duration_seconds is not None: resolved_audio_duration_seconds[segment_index] = max(0.0, float(duration_seconds)) elif fps_value > 0.0: for segment_index, segment_path in enumerate(segment_paths): segment_frame_count, _ = probe_resume_frame_count(ffprobe_path, segment_path, fps_value) if segment_frame_count > 0: resolved_audio_duration_seconds[segment_index] = float(segment_frame_count) / fps_value _concat_audio_segments(ffmpeg_path, segment_paths, merged_audio_path, concat_dir, segment_trim_seconds=segment_audio_trim_seconds, segment_duration_seconds=resolved_audio_duration_seconds, audio_stream_indices=audio_stream_indices) command += ["-i", merged_audio_path] if reserved_metadata_path and os.path.isfile(reserved_metadata_path): command += ["-f", "ffmetadata", "-i", reserved_metadata_path] command += ["-map_metadata", "2" if has_audio else "1"] else: command += ["-map_metadata", "0"] command += ["-map", "0:v:0"] if has_audio: command += ["-map", "1:a:0"] command += ["-c", "copy"] if is_quicktime_output_container(video_container): command += ["-movflags", "+faststart"] command += [temp_output_path] result = subprocess.run(command, capture_output=True, text=True) if result.returncode != 0 or not os.path.isfile(temp_output_path): raise gr.Error((result.stderr or result.stdout or "ffmpeg concat failed").strip()) timeline_gap = _probe_video_frame_gap(ffprobe_path, temp_output_path, fps_value, near_time=_probe_media_duration(ffprobe_path, segment_paths[0])) if timeline_gap is not None: gap_start, gap_end, gap_seconds = timeline_gap raise gr.Error(f"Merged video timeline contains a {gap_seconds:.6f}s gap near {gap_start:.3f}s -> {gap_end:.3f}s.") try: os.replace(temp_output_path, output_path) except OSError as exc: raise ContinuationMergeOutputLockedError(output_path) from exc finally: shutil.rmtree(concat_dir, ignore_errors=True)