#!/usr/bin/env python3 """Burn one ASS subtitle track into a high-quality H.264 MP4.""" from __future__ import annotations import argparse from collections import deque import hashlib import json import math import os import re import shutil import subprocess import sys import tempfile import textwrap from fractions import Fraction from pathlib import Path from typing import Any, Sequence DEFAULT_ENCODER = "libx264" PROGRESS_BAR_WIDTH = 20 PROGRESS_STEP_PERCENT = 5 MP4_COPY_AUDIO_CODECS = frozenset({"aac", "ac3", "alac", "eac3", "mp3"}) HDR_TRANSFERS = frozenset({"arib-std-b67", "smpte2084"}) HDR_SIDE_DATA = ( "content light level", "dolby vision", "dovi", "dynamic hdr", "hdr10+", "mastering display", ) FFMPEG_FULL_CANDIDATES = ( Path("/opt/homebrew/opt/ffmpeg-full/bin/ffmpeg"), Path("/usr/local/opt/ffmpeg-full/bin/ffmpeg"), ) ASS_WORD_JOINER = "\u2060" class BurnError(RuntimeError): """A user-actionable burn or verification failure.""" def _positive_crf(value: str) -> int: try: crf = int(value) except ValueError as exc: raise argparse.ArgumentTypeError("CRF must be an integer from 0 to 51") from exc if not 0 <= crf <= 51: raise argparse.ArgumentTypeError("CRF must be an integer from 0 to 51") return crf def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description=( "Burn an ASS subtitle file exactly once into an H.264/yuv420p MP4 " "while preserving the source dimensions and frame timing." ) ) parser.add_argument("video", type=Path, help="input video") parser.add_argument("subtitle", type=Path, help="input ASS subtitle file") parser.add_argument("output", type=Path, help="output MP4") parser.add_argument( "--force", action="store_true", help="replace OUTPUT if it already exists", ) parser.add_argument( "--crf", type=_positive_crf, default=18, help="H.264 constant-rate-factor quality (default: 18)", ) parser.add_argument( "--preset", default="slow", help="encoder preset (default: slow)", ) parser.add_argument( "--encoder", default=DEFAULT_ENCODER, help=f"FFmpeg H.264 encoder (default: {DEFAULT_ENCODER})", ) parser.add_argument( "--validation-report", type=Path, help="subtitle validation JSON (default: validation.json next to the ASS file)", ) parser.add_argument( "--allow-missing-font", action="store_true", help="continue with libass font substitution when the validated font is not installed", ) parser.add_argument( "--citation-file", type=Path, help="approved UTF-8 citation text to burn at the top left", ) return parser def _required_executables() -> tuple[str, str]: ffmpeg = shutil.which("ffmpeg") ffprobe = shutil.which("ffprobe") missing = [name for name, path in (("ffmpeg", ffmpeg), ("ffprobe", ffprobe)) if not path] if missing: raise BurnError(f"required executable not found in PATH: {', '.join(missing)}") assert ffmpeg is not None and ffprobe is not None ffmpeg = _select_libass_ffmpeg(ffmpeg) sibling_ffprobe = Path(ffmpeg).with_name("ffprobe") if sibling_ffprobe.is_file(): ffprobe = str(sibling_ffprobe) return ffmpeg, ffprobe def _ffmpeg_has_subtitles_filter(ffmpeg: str | Path) -> bool: result = subprocess.run( [str(ffmpeg), "-hide_banner", "-filters"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) return result.returncode == 0 and any( len(fields := line.split()) >= 2 and fields[1] == "subtitles" for line in result.stdout.splitlines() ) def _select_libass_ffmpeg( default: str, *, candidates: Sequence[Path] = FFMPEG_FULL_CANDIDATES, ) -> str: for candidate in (Path(default), *candidates): if candidate.is_file() and _ffmpeg_has_subtitles_filter(candidate): return str(candidate) return default def _require_libass_subtitles_filter(ffmpeg: str) -> None: if not _ffmpeg_has_subtitles_filter(ffmpeg): raise BurnError( "FFmpeg has no usable 'subtitles' filter; install an FFmpeg build " "with libass support" ) def _last_error_line(stderr: str) -> str: lines = [line.strip() for line in stderr.splitlines() if line.strip()] return f": {lines[-1]}" if lines else "" def _clock(seconds: float) -> str: total = max(0, int(seconds)) hours, remainder = divmod(total, 3600) minutes, seconds = divmod(remainder, 60) if hours: return f"{hours:02d}:{minutes:02d}:{seconds:02d}" return f"{minutes:02d}:{seconds:02d}" def _format_progress( percent: int, encoded_seconds: float, duration: float, speed: str, ) -> str: percent = max(0, min(100, int(percent))) filled = round(percent * PROGRESS_BAR_WIDTH / 100) bar = "█" * filled + "░" * (PROGRESS_BAR_WIDTH - filled) speed = speed.strip() or "--" return ( f"烧录 [{bar}] {percent:3d}% " f"{_clock(encoded_seconds)} / {_clock(duration)} {speed}" ) def _progress_seconds(values: dict[str, str]) -> float: raw = values.get("out_time_us") or values.get("out_time_ms") if raw: try: return max(0.0, int(raw) / 1_000_000) except ValueError: pass clock = values.get("out_time", "") try: hours, minutes, seconds = clock.split(":", 2) return max(0.0, int(hours) * 3600 + int(minutes) * 60 + float(seconds)) except (TypeError, ValueError): return 0.0 def _run_ffmpeg_with_progress(command: Sequence[str], duration: float) -> tuple[int, str]: process = subprocess.Popen( list(command), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", bufsize=1, ) if process.stdout is None: process.kill() raise BurnError("FFmpeg progress pipe was not available") values: dict[str, str] = {} diagnostics: deque[str] = deque(maxlen=12) last_bucket = 0 print(_format_progress(0, 0, duration, "--"), file=sys.stderr, flush=True) for raw_line in process.stdout: line = raw_line.strip() if not line: continue if "=" not in line: diagnostics.append(line) continue key, value = line.split("=", 1) values[key] = value if key != "progress": continue encoded_seconds = _progress_seconds(values) raw_percent = 100 * encoded_seconds / duration if duration > 0 else 0 bucket = min( 100, int(raw_percent // PROGRESS_STEP_PERCENT) * PROGRESS_STEP_PERCENT, ) if value == "end": bucket = 100 encoded_seconds = duration if bucket > last_bucket: print( _format_progress( bucket, encoded_seconds, duration, values.get("speed", "--"), ), file=sys.stderr, flush=True, ) last_bucket = bucket returncode = process.wait() process.stdout.close() return returncode, "\n".join(diagnostics) def _probe(ffprobe: str, path: Path) -> dict[str, Any]: result = subprocess.run( [ ffprobe, "-v", "error", "-show_format", "-show_streams", "-of", "json", str(path), ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) if result.returncode != 0: raise BurnError(f"ffprobe could not read {path}{_last_error_line(result.stderr)}") try: data = json.loads(result.stdout) except json.JSONDecodeError as exc: raise BurnError(f"ffprobe returned invalid JSON for {path}: {exc}") from exc if not isinstance(data, dict): raise BurnError(f"ffprobe returned an unexpected result for {path}") return data def _streams(probe: dict[str, Any], kind: str) -> list[dict[str, Any]]: streams = probe.get("streams", []) if not isinstance(streams, list): return [] return [ stream for stream in streams if isinstance(stream, dict) and stream.get("codec_type") == kind ] def _main_video_stream(probe: dict[str, Any]) -> dict[str, Any]: videos = _streams(probe, "video") if not videos: raise BurnError("input contains no video stream") return next( ( stream for stream in videos if not bool((stream.get("disposition") or {}).get("attached_pic")) ), videos[0], ) def _stream_dimensions(stream: dict[str, Any]) -> tuple[int, int]: try: width = int(stream["width"]) height = int(stream["height"]) except (KeyError, TypeError, ValueError) as exc: raise BurnError("video stream has no valid dimensions") from exc if width <= 0 or height <= 0: raise BurnError("video stream has no valid dimensions") return width, height def _duration(probe: dict[str, Any]) -> float: candidates: list[Any] = [] file_format = probe.get("format") if isinstance(file_format, dict): candidates.append(file_format.get("duration")) for stream in probe.get("streams", []): if isinstance(stream, dict): candidates.append(stream.get("duration")) durations: list[float] = [] for candidate in candidates: try: duration = float(candidate) except (TypeError, ValueError): continue if math.isfinite(duration) and duration > 0: durations.append(duration) return max(durations, default=0.0) def _frame_rate(stream: dict[str, Any]) -> Fraction | None: for key in ("avg_frame_rate", "r_frame_rate"): value = stream.get(key) if not isinstance(value, str): continue try: rate = Fraction(value) except (ValueError, ZeroDivisionError): continue if rate > 0: return rate return None def _is_hdr(stream: dict[str, Any]) -> bool: if str(stream.get("color_transfer", "")).lower() in HDR_TRANSFERS: return True if str(stream.get("color_primaries", "")).lower() == "bt2020": try: bit_depth = int(stream.get("bits_per_raw_sample", 0)) except (TypeError, ValueError): bit_depth = 0 pixel_format = str(stream.get("pix_fmt", "")).lower() if bit_depth >= 10 or re.search(r"(?:10|12|14|16)(?:le|be)?$", pixel_format): return True side_data = stream.get("side_data_list") if isinstance(side_data, list): for item in side_data: if not isinstance(item, dict): continue description = " ".join(str(value).lower() for value in item.values()) if any(marker in description for marker in HDR_SIDE_DATA): return True return False def _escape_filter_value(value: str) -> str: """Escape a value through FFmpeg's option and filtergraph parser layers.""" def escape(text: str, special: str) -> str: return "".join(f"\\{char}" if char in special else char for char in text) option_escaped = escape(value, "\\':") return escape(option_escaped, "\\'[],;") def _sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _citation_text(path: Path) -> str: try: value = path.read_text(encoding="utf-8") except (OSError, UnicodeDecodeError) as exc: raise BurnError(f"citation file is not readable UTF-8: {path}: {exc}") from exc lines = [" ".join(line.split()) for line in value.splitlines()] while lines and not lines[0]: lines.pop(0) while lines and not lines[-1]: lines.pop() normalized: list[str] = [] for line in lines: if line or not normalized or normalized[-1]: normalized.append(line) value = "\n".join(normalized) if not value: raise BurnError("citation file cannot be empty") if len(value) > 2000 or any(ord(character) < 32 and character != "\n" for character in value): raise BurnError("citation file contains unsupported or excessive text") return value def _ass_time(seconds: float) -> str: centiseconds = max(1, math.ceil(seconds * 100)) hours, remainder = divmod(centiseconds, 360000) minutes, remainder = divmod(remainder, 6000) whole_seconds, fraction = divmod(remainder, 100) return f"{hours}:{minutes:02d}:{whole_seconds:02d}.{fraction:02d}" def _ass_escape(value: str) -> str: """Losslessly encode untrusted visible text for an ASS Dialogue field.""" output: list[str] = [] for character in value: if character == "\\": output.append("\\" + ASS_WORD_JOINER) elif character == "{": output.append(r"\{{}") elif character == "\n": output.append(r"\N") else: output.append(character) return "".join(output) def _render_citation_ass(text: str, width: int, height: int, duration: float) -> str: font_size = max(16, min(28, round(height * 0.022))) notice_size = max(14, font_size - 4) margin_x = max(18, round(width * 0.0208)) margin_y = max(18, round(height * 0.0278)) padding_x = max(12, round(width * 0.0083)) padding_y = max(10, round(height * 0.0111)) portrait = height > width panel_width = min( width - 2 * margin_x, max( round(width * (0.90 if portrait else 0.375)), round(24 * font_size * 0.41) + 2 * padding_x, ), ) max_columns = max( 24, round((panel_width - 2 * padding_x) / (font_size * 0.41)), ) raw_lines = text.splitlines() notice: str | None = None if len(raw_lines) >= 3 and raw_lines[1] == "": notice = raw_lines[0] raw_lines = raw_lines[2:] wrapped_lines: list[str] = [] for paragraph in raw_lines: wrapped_lines.extend( textwrap.wrap( paragraph, width=max_columns, break_long_words=False, break_on_hyphens=False, ) or [""] ) citation_display = _ass_escape("\n".join(wrapped_lines)) if notice is not None: display = ( rf"{{\fs{notice_size}\1c&HCCCCCC&}}{_ass_escape(notice)}" rf"\N{{\fs8}}　\N{{\fs{font_size}\1c&HFFFFFF&}}{citation_display}" ) else: display = rf"{{\fs{font_size}\1c&HFFFFFF&}}{citation_display}" line_height = font_size * 1.05 panel_height = math.ceil( 2 * padding_y + len(wrapped_lines) * line_height + (notice_size * 1.05 + 8 if notice is not None else 0) + 3 ) text_x = margin_x + padding_x text_y = margin_y + padding_y panel = ( rf"{{\an7\pos({margin_x},{margin_y})\p1\1c&H000000&\1a&H78&}}" f"m 0 0 l {panel_width} 0 {panel_width} {panel_height} 0 {panel_height}" ) return ( "[Script Info]\n" "ScriptType: v4.00+\n" "WrapStyle: 2\n" "ScaledBorderAndShadow: yes\n" f"PlayResX: {width}\nPlayResY: {height}\n\n" "[V4+ Styles]\n" "Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, " "OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, " "ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, " "Alignment, MarginL, MarginR, MarginV, Encoding\n" f"Style: Citation,Arial,{font_size},&H10FFFFFF,&H10FFFFFF,&H00000000," "&H00000000,0,0,0,0,100,100,0,0,1,0,0,7,0,0,0,1\n\n" "[Events]\n" "Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\n" f"Dialogue: 9,0:00:00.00,{_ass_time(duration)},Citation,,0,0,0,,{panel}\n" f"Dialogue: 10,0:00:00.00,{_ass_time(duration)},Citation,,0,0,0,," rf"{{\an7\pos({text_x},{text_y})}}{display}\n" ) def _atomic_json(path: Path, value: dict[str, Any]) -> None: fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: with os.fdopen(fd, "w", encoding="utf-8") as handle: json.dump(value, handle, ensure_ascii=False, indent=2, sort_keys=True) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) except BaseException: try: os.close(fd) except OSError: pass Path(temporary).unlink(missing_ok=True) raise def _write_citation_receipt(output: Path, citation_file: Path) -> Path: receipt = output.with_suffix(output.suffix + ".citation.json") _atomic_json( receipt, { "schema_version": 1, "output_file": output.name, "output_sha256": _sha256_file(output), "citation_sha256": _sha256_file(citation_file), "position": "top-left", }, ) return receipt def _validate_validation_report(subtitle: Path, report_path: Path) -> dict[str, Any]: subtitle = subtitle.expanduser().resolve() report_path = report_path.expanduser().resolve() if not report_path.is_file(): raise BurnError(f"validation report does not exist or is not a file: {report_path}") try: report = json.loads(report_path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise BurnError(f"validation report is not valid UTF-8 JSON: {report_path}: {exc}") from exc if not isinstance(report, dict): raise BurnError("validation report root must be a JSON object") if report.get("structurally_valid") is not True: raise BurnError("validation report must declare structurally_valid=true") if report.get("validation_scope") != "structural_source_integrity": raise BurnError( "validation report scope must be structural_source_integrity" ) segment_count = report.get("segment_count") translation_count = report.get("translation_count") counts = (segment_count, translation_count) if any(isinstance(value, bool) or not isinstance(value, int) or value <= 0 for value in counts): raise BurnError( "validation report segment_count and translation_count must be positive integers" ) if segment_count != translation_count: raise BurnError( "validation report segment_count and translation_count must be equal" ) outputs = report.get("outputs") recorded_hash = outputs.get("bilingual.ass") if isinstance(outputs, dict) else None if not isinstance(recorded_hash, str) or not re.fullmatch( r"[0-9a-fA-F]{64}", recorded_hash ): raise BurnError( "validation report outputs['bilingual.ass'] must be a SHA-256 checksum" ) if _sha256_file(subtitle) != recorded_hash.lower(): raise BurnError("bilingual.ass SHA-256 does not match the validation report") return report _FONT_FILE_SUFFIXES = frozenset({".ttf", ".otf", ".ttc"}) _FONT_DIRECTORIES = ( "~/Library/Fonts", "/Library/Fonts", "/System/Library/Fonts", "~/.fonts", "~/.local/share/fonts", "/usr/share/fonts", "/usr/local/share/fonts", ) def _font_installed(family: str) -> bool | None: """Return True/False when detection is trustworthy, None when unavailable. libass silently substitutes another font when the requested family is missing, which would pass every later gate with the wrong deliverable. """ fc_list = shutil.which("fc-list") if fc_list: result = subprocess.run( [fc_list, ":", "family"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) if result.returncode == 0 and result.stdout.strip(): needle = family.casefold() return any( needle in entry.strip().casefold() for line in result.stdout.splitlines() for entry in line.split(",") ) token = re.sub(r"[\s_-]+", "", family).casefold() if not token: return None searched = False for directory in _FONT_DIRECTORIES: base = Path(directory).expanduser() if not base.is_dir(): continue searched = True for path in base.rglob("*"): if ( path.suffix.lower() in _FONT_FILE_SUFFIXES and token in re.sub(r"[\s_-]+", "", path.stem).casefold() ): return True return False if searched else None def _require_subtitle_font(report: dict[str, Any], *, allow_missing_font: bool) -> None: font = str(report.get("font") or "").strip() if not font: return installed = _font_installed(font) if installed is True: return if installed is None: print( f"warning: could not verify that font {font!r} is installed; " "libass substitutes missing fonts silently", file=sys.stderr, ) return message = ( f"font {font!r} required by the validated subtitles was not found; install it " "(MiSans: https://hyperos.mi.com/font/zh/download/)" ) if allow_missing_font: print(f"warning: {message}; continuing with libass substitution", file=sys.stderr) return raise BurnError(f"{message} or pass --allow-missing-font to accept substitution") def _audio_options(audio_streams: Sequence[dict[str, Any]]) -> tuple[list[str], list[str]]: if not audio_streams: return [], [] options = ["-c:a", "copy"] modes: list[str] = [] for output_index, stream in enumerate(audio_streams): codec = str(stream.get("codec_name", "")).lower() if codec in MP4_COPY_AUDIO_CODECS: modes.append(f"audio {output_index}: copied {codec}") continue options.extend( [ f"-c:a:{output_index}", "aac", f"-b:a:{output_index}", "256k", ] ) modes.append(f"audio {output_index}: {codec or 'unknown'} -> AAC") return options, modes def _encode_command( ffmpeg: str, video: Path, subtitle: Path, output: Path, video_stream: dict[str, Any], audio_streams: Sequence[dict[str, Any]], *, force: bool, crf: int, preset: str, encoder: str, citation_ass: Path | None = None, ) -> tuple[list[str], list[str]]: try: stream_index = int(video_stream["index"]) except (KeyError, TypeError, ValueError) as exc: raise BurnError("input video stream has no valid index") from exc audio_options, audio_modes = _audio_options(audio_streams) subtitle_filters = [f"subtitles=filename={_escape_filter_value(str(subtitle))}"] if citation_ass is not None: subtitle_filters.append( f"subtitles=filename={_escape_filter_value(str(citation_ass))}" ) command = [ ffmpeg, "-hide_banner", "-loglevel", "error", "-nostats", "-stats_period", "1", "-progress", "pipe:1", "-y" if force else "-n", "-i", str(video), "-map", f"0:{stream_index}", "-map", "0:a?", "-map_metadata", "0", "-map_chapters", "0", "-sn", "-dn", "-vf", ",".join(subtitle_filters), "-fps_mode:v:0", "passthrough", "-c:v", encoder, "-crf", str(crf), "-preset", preset, "-pix_fmt", "yuv420p", *audio_options, "-movflags", "+faststart", "-f", "mp4", str(output), ] return command, audio_modes def _verify_output( ffprobe: str, output: Path, input_video: dict[str, Any], input_had_audio: bool, *, input_duration: float, ) -> None: result = _probe(ffprobe, output) file_format = result.get("format") format_name = file_format.get("format_name", "") if isinstance(file_format, dict) else "" if "mp4" not in str(format_name).split(","): raise BurnError(f"output verification failed: container is not MP4 ({format_name or 'unknown'})") output_videos = _streams(result, "video") if not output_videos: raise BurnError("output verification failed: no video stream") output_video = output_videos[0] if output_video.get("codec_name") != "h264": raise BurnError( "output verification failed: video codec is " f"{output_video.get('codec_name', 'unknown')}, not H.264" ) output_duration = _duration(result) if output_duration <= 0: raise BurnError("output verification failed: duration is zero or unavailable") if input_duration <= 0: raise BurnError("output verification failed: input duration is zero or unavailable") duration_tolerance = max(0.5, input_duration * 0.01) if abs(output_duration - input_duration) > duration_tolerance: raise BurnError( "output verification failed: duration changed from " f"{input_duration:.3f}s to {output_duration:.3f}s " f"(allowed difference {duration_tolerance:.3f}s)" ) input_dimensions = _stream_dimensions(input_video) output_dimensions = _stream_dimensions(output_video) if output_dimensions != input_dimensions: raise BurnError( "output verification failed: dimensions changed from " f"{input_dimensions[0]}x{input_dimensions[1]} to " f"{output_dimensions[0]}x{output_dimensions[1]}" ) input_rate = _frame_rate(input_video) output_rate = _frame_rate(output_video) if input_rate is not None and output_rate is not None: relative_drift = abs(float(output_rate - input_rate)) / float(input_rate) if relative_drift > 0.005: raise BurnError( "output verification failed: frame rate changed from " f"{float(input_rate):.6g} to {float(output_rate):.6g} fps" ) if input_had_audio and not _streams(result, "audio"): raise BurnError("output verification failed: input audio is missing from output") def burn_subtitles( video: Path, subtitle: Path, output: Path, *, force: bool = False, crf: int = 18, preset: str = "slow", encoder: str = DEFAULT_ENCODER, validation_report: Path | None = None, allow_missing_font: bool = False, citation_file: Path | None = None, ) -> list[str]: video = video.expanduser().resolve() subtitle = subtitle.expanduser().resolve() output = output.expanduser().resolve() report_path = ( validation_report.expanduser().resolve() if validation_report is not None else subtitle.with_name("validation.json") ) citation_path = citation_file.expanduser().resolve() if citation_file is not None else None if not video.is_file(): raise BurnError(f"input video does not exist or is not a file: {video}") if not subtitle.is_file(): raise BurnError(f"ASS subtitle does not exist or is not a file: {subtitle}") if subtitle.suffix.lower() != ".ass": raise BurnError(f"subtitle must be an .ass file: {subtitle}") if output in (video, subtitle, report_path): raise BurnError("output must be different from all input files") if citation_path is not None and not citation_path.is_file(): raise BurnError(f"citation file does not exist or is not a file: {citation_path}") if citation_path is not None and output == citation_path: raise BurnError("output must be different from the citation file") if not output.parent.is_dir(): raise BurnError(f"output directory does not exist: {output.parent}") if output.exists() and not force: raise BurnError(f"output already exists (use --force to replace it): {output}") if output.exists() and not output.is_file(): raise BurnError(f"output exists and is not a regular file: {output}") if not preset.strip(): raise BurnError("encoder preset cannot be empty") if not encoder.strip(): raise BurnError("encoder cannot be empty") report = _validate_validation_report(subtitle, report_path) _require_subtitle_font(report, allow_missing_font=allow_missing_font) ffmpeg, ffprobe = _required_executables() _require_libass_subtitles_filter(ffmpeg) input_probe = _probe(ffprobe, video) input_video = _main_video_stream(input_probe) width, height = _stream_dimensions(input_video) input_duration = _duration(input_probe) if input_duration <= 0: raise BurnError("input duration is zero or unavailable") audio_streams = _streams(input_probe, "audio") if _is_hdr(input_video): print( "warning: HDR input detected. The compatibility H.264/yuv420p output is " "intended for SDR playback; HDR metadata and appearance may not be preserved.", file=sys.stderr, ) citation_ass: Path | None = None temporary_dir: tempfile.TemporaryDirectory[str] | None = None if citation_path is not None: temporary_dir = tempfile.TemporaryDirectory(prefix="materialsub-citation-") citation_ass = Path(temporary_dir.name) / "citation.ass" citation_ass.write_text( _render_citation_ass(_citation_text(citation_path), width, height, input_duration), encoding="utf-8", ) try: command, audio_modes = _encode_command( ffmpeg, video, subtitle, output, input_video, audio_streams, force=force, crf=crf, preset=preset, encoder=encoder, citation_ass=citation_ass, ) returncode, diagnostic = _run_ffmpeg_with_progress(command, input_duration) finally: if temporary_dir is not None: temporary_dir.cleanup() if returncode != 0: if output.is_file(): output.unlink() detail = _last_error_line(diagnostic) raise BurnError(f"FFmpeg subtitle burn failed with exit code {returncode}{detail}") try: _verify_output( ffprobe, output, input_video, bool(audio_streams), input_duration=input_duration, ) except BurnError: if output.is_file(): output.unlink() raise receipt = output.with_suffix(output.suffix + ".citation.json") if citation_path is not None: _write_citation_receipt(output, citation_path) elif receipt.is_file(): receipt.unlink() return audio_modes def main(argv: Sequence[str] | None = None) -> int: args = _parser().parse_args(argv) try: audio_modes = burn_subtitles( args.video, args.subtitle, args.output, force=args.force, crf=args.crf, preset=args.preset, encoder=args.encoder, validation_report=args.validation_report, allow_missing_font=args.allow_missing_font, citation_file=args.citation_file, ) except (BurnError, OSError) as exc: print(f"error: {exc}", file=sys.stderr) return 1 print( "validated ASS was burned into verified MP4: " f"{args.output.expanduser().resolve()}" ) for mode in audio_modes: print(mode) return 0 if __name__ == "__main__": raise SystemExit(main())