bazarr/libs/ffsubsync/ffsubsync.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import argparse
from datetime import datetime
import logging
import os
import shutil
import subprocess
import sys
from typing import cast, Any, Callable, Dict, List, Optional, Tuple, Union

import numpy as np

from ffsubsync.aligners import FFTAligner, MaxScoreAligner
from ffsubsync.constants import (
    DEFAULT_APPLY_OFFSET_SECONDS,
    DEFAULT_FRAME_RATE,
    DEFAULT_MAX_OFFSET_SECONDS,
    DEFAULT_MAX_SUBTITLE_SECONDS,
    DEFAULT_NON_SPEECH_LABEL,
    DEFAULT_START_SECONDS,
    DEFAULT_VAD,
    DEFAULT_ENCODING,
    FRAMERATE_RATIOS,
    SAMPLE_RATE,
    SUBTITLE_EXTENSIONS,
    DEFAULT_MIN_SCORE,
    DEFAULT_QUALITY_MAX_OFFSET_SECONDS,
    DEFAULT_MAX_FRAMERATE_DEVIATION,
    is_remote_url,
)
from ffsubsync.ffmpeg_utils import ffmpeg_bin_path
from ffsubsync.sklearn_shim import Pipeline, TransformerMixin
from ffsubsync.speech_transformers import (
    VideoSpeechTransformer,
    MultiSegmentVideoSpeechTransformer,
    DeserializeSpeechTransformer,
    PGSSpeechTransformer,
    ProgressInfo,
    make_subtitle_speech_pipeline,
)
from ffsubsync.subtitle_parser import make_subtitle_parser
from ffsubsync.subtitle_transformers import SubtitleMerger, SubtitleShifter
from ffsubsync.version import get_version


logger: logging.Logger = logging.getLogger(__name__)


def override(args: argparse.Namespace, **kwargs: Any) -> Dict[str, Any]:
    args_dict = dict(args.__dict__)
    args_dict.update(kwargs)
    return args_dict


def _ref_format(ref_fname: Optional[str]) -> Optional[str]:
    if ref_fname is None:
        return None
    return ref_fname[-3:]


def make_test_case(
    args: argparse.Namespace, npy_savename: Optional[str], sync_was_successful: bool
) -> int:
    if npy_savename is None:
        raise ValueError("need non-null npy_savename")
    tar_dir = "{}.{}".format(
        args.reference, datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
    )
    logger.info("creating test archive {}.tar.gz...".format(tar_dir))
    os.mkdir(tar_dir)
    try:
        log_path = "ffsubsync.log"
        if args.log_dir_path is not None and os.path.isdir(args.log_dir_path):
            log_path = os.path.join(args.log_dir_path, log_path)
        shutil.copy(log_path, tar_dir)
        shutil.copy(args.srtin[0], tar_dir)
        if sync_was_successful:
            shutil.move(args.srtout, tar_dir)
        if _ref_format(args.reference) in SUBTITLE_EXTENSIONS:
            shutil.copy(args.reference, tar_dir)
        elif args.serialize_speech or args.reference == npy_savename:
            shutil.copy(npy_savename, tar_dir)
        else:
            shutil.move(npy_savename, tar_dir)
        supported_formats = set(list(zip(*shutil.get_archive_formats()))[0])
        preferred_formats = ["gztar", "bztar", "xztar", "zip", "tar"]
        for archive_format in preferred_formats:
            if archive_format in supported_formats:
                shutil.make_archive(tar_dir, archive_format, os.curdir, tar_dir)
                break
        else:
            logger.error(
                "failed to create test archive; no formats supported "
                "(this should not happen)"
            )
            return 1
        logger.info("...done")
    finally:
        shutil.rmtree(tar_dir)
    return 0


def _resolve_srtout(args: argparse.Namespace, srtin: Optional[str]) -> Optional[str]:
    """Decide where the synced subtitles for ``srtin`` should be written.

    Overwriting the input takes precedence; otherwise, when subtitles were
    auto-detected from the reference, write a `<name>.synced.srt` next to each
    input; otherwise fall back to the explicit (possibly ``None``) output.
    """
    if args.overwrite_input:
        return srtin
    if getattr(args, "auto_srtout", False) and srtin is not None:
        return "{}.synced.srt".format(os.path.splitext(srtin)[0])
    return args.srtout


def get_srt_pipe_maker(
    args: argparse.Namespace, srtin: Optional[str]
) -> Callable[[Optional[float]], Union[Pipeline, Callable[[float], Pipeline]]]:
    if srtin is None:
        srtin_format = "srt"
    else:
        srtin_format = os.path.splitext(srtin)[-1][1:]
    parser = make_subtitle_parser(fmt=srtin_format, caching=True, **args.__dict__)
    return lambda scale_factor: make_subtitle_speech_pipeline(
        **override(args, scale_factor=scale_factor, parser=parser)
    )


def get_framerate_ratios_to_try(args: argparse.Namespace) -> List[Optional[float]]:
    if args.no_fix_framerate:
        return []
    else:
        framerate_ratios = list(
            np.concatenate(
                [np.array(FRAMERATE_RATIOS), 1.0 / np.array(FRAMERATE_RATIOS)]
            )
        )
        if args.gss:
            framerate_ratios.append(None)
        return framerate_ratios


def assess_alignment_quality(
    best_score: float,
    offset_seconds: float,
    scale_factor: float,
    *,
    min_score: float,
    max_offset_seconds: float,
    max_framerate_deviation: float,
) -> List[str]:
    """Return reasons an alignment looks too low-quality to trust (empty = trust it).

    Used by --skip-sync-on-low-quality to decide whether to leave the subtitles
    unmodified rather than apply a probably-wrong shift. A negative score means the
    best alignment is anti-correlated; an implausibly large offset or framerate
    scale suggests a spurious match.
    """
    reasons: List[str] = []
    if best_score < min_score:
        reasons.append("score %.1f < %.1f" % (best_score, min_score))
    if abs(offset_seconds) > max_offset_seconds:
        reasons.append(
            "|offset| %.1fs > %.1fs" % (abs(offset_seconds), max_offset_seconds)
        )
    framerate_deviation = abs(scale_factor - 1.0)
    if framerate_deviation > max_framerate_deviation:
        reasons.append(
            "framerate deviation %.3f > %.3f"
            % (framerate_deviation, max_framerate_deviation)
        )
    return reasons


def try_sync(
    args: argparse.Namespace, reference_pipe: Optional[Pipeline], result: Dict[str, Any]
) -> bool:
    result["sync_was_successful"] = False
    sync_was_successful = True
    logger.info(
        "extracting speech segments from %s...",
        "stdin" if not args.srtin else "subtitles file(s) {}".format(args.srtin),
    )
    if not args.srtin:
        args.srtin = [None]
    for srtin in args.srtin:
        try:
            skip_sync = args.skip_sync or reference_pipe is None
            skip_infer_framerate_ratio = (
                args.skip_infer_framerate_ratio or reference_pipe is None
            )
            srtout = _resolve_srtout(args, srtin)
            srt_pipe_maker = get_srt_pipe_maker(args, srtin)
            framerate_ratios = get_framerate_ratios_to_try(args)
            srt_pipes = [srt_pipe_maker(1.0)] + [
                srt_pipe_maker(rat) for rat in framerate_ratios
            ]
            for srt_pipe in srt_pipes:
                if callable(srt_pipe):
                    continue
                else:
                    srt_pipe.fit(srtin)
            if (
                not skip_infer_framerate_ratio
                and hasattr(reference_pipe[-1], "num_frames")
                and reference_pipe[-1].num_frames is not None
            ):
                inferred_framerate_ratio_from_length = (
                    float(reference_pipe[-1].num_frames)
                    / cast(Pipeline, srt_pipes[0])[-1].num_frames
                )
                logger.info(
                    "inferred frameratio ratio: %.3f"
                    % inferred_framerate_ratio_from_length
                )
                srt_pipes.append(
                    cast(
                        Pipeline, srt_pipe_maker(inferred_framerate_ratio_from_length)
                    ).fit(srtin)
                )
                logger.info("...done")
            logger.info("computing alignments...")
            if skip_sync:
                best_score = 0.0
                best_srt_pipe = cast(Pipeline, srt_pipes[0])
                offset_samples = 0
            else:
                (best_score, offset_samples), best_srt_pipe = MaxScoreAligner(
                    FFTAligner, srtin, SAMPLE_RATE, args.max_offset_seconds
                ).fit_transform(
                    reference_pipe.transform(args.reference),
                    srt_pipes,
                )
            if best_score < 0:
                sync_was_successful = False
            logger.info("...done")
            offset_seconds = (
                offset_samples / float(SAMPLE_RATE) + args.apply_offset_seconds
            )
            scale_step = best_srt_pipe.named_steps["scale"]
            logger.info("score: %.3f", best_score)
            logger.info("offset seconds: %.3f", offset_seconds)
            logger.info("framerate scale factor: %.3f", scale_step.scale_factor)
            low_quality_reasons: List[str] = []
            if getattr(args, "skip_sync_on_low_quality", False):
                low_quality_reasons = assess_alignment_quality(
                    best_score,
                    offset_seconds,
                    scale_step.scale_factor,
                    min_score=args.min_score,
                    max_offset_seconds=args.quality_max_offset_seconds,
                    max_framerate_deviation=args.max_framerate_deviation,
                )
            if low_quality_reasons:
                logger.warning(
                    "low-quality alignment (%s); leaving subtitles unmodified",
                    "; ".join(low_quality_reasons),
                )
                sync_was_successful = False
                # write the original (unscaled, unshifted) subtitles unchanged
                original_subs = best_srt_pipe.named_steps["parse"].subs_
                out_subs = original_subs.clone_props_for_subs(list(original_subs))
                if args.output_encoding != "same":
                    out_subs = out_subs.set_encoding(args.output_encoding)
                logger.info(
                    "writing original (unsynced) output to {}".format(srtout or "stdout")
                )
                out_subs.write_file(srtout)
                continue
            output_steps: List[Tuple[str, TransformerMixin]] = [
                ("shift", SubtitleShifter(offset_seconds))
            ]
            if args.merge_with_reference:
                output_steps.append(
                    ("merge", SubtitleMerger(reference_pipe.named_steps["parse"].subs_))
                )
            output_pipe = Pipeline(output_steps)
            out_subs = output_pipe.fit_transform(scale_step.subs_)
            if args.output_encoding != "same":
                out_subs = out_subs.set_encoding(args.output_encoding)
            suppress_output_thresh = args.suppress_output_if_offset_less_than
            if offset_seconds >= (suppress_output_thresh or float("-inf")):
                logger.info("writing output to {}".format(srtout or "stdout"))
                out_subs.write_file(srtout)
            else:
                logger.warning(
                    "suppressing output because offset %s was less than suppression threshold %s",
                    offset_seconds,
                    args.suppress_output_if_offset_less_than,
                )
        except Exception:
            sync_was_successful = False
            logger.exception("failed to sync %s", srtin)
        else:
            result["offset_seconds"] = offset_seconds
            result["framerate_scale_factor"] = scale_step.scale_factor
    result["sync_was_successful"] = sync_was_successful
    return sync_was_successful


def make_reference_pipe(
    args: argparse.Namespace,
    progress_handler: Optional[Callable[[ProgressInfo], None]] = None,
) -> Pipeline:
    pgs_stream = getattr(args, "pgs_ref_stream", None)
    if pgs_stream is not None:
        # "auto" (bare --pgs-ref-stream flag) → let PGSSpeechTransformer auto-detect
        resolved_stream: Optional[str] = None if pgs_stream == "auto" else pgs_stream
        if resolved_stream is not None and not resolved_stream.startswith("0:"):
            resolved_stream = "0:" + resolved_stream
        return Pipeline(
            [
                (
                    "speech_extract",
                    PGSSpeechTransformer(
                        sample_rate=SAMPLE_RATE,
                        start_seconds=args.start_seconds,
                        ffmpeg_path=args.ffmpeg_path,
                        ref_stream=resolved_stream,
                        gui_mode=args.gui_mode,
                    ),
                ),
            ]
        )
    ref_format = _ref_format(args.reference)
    if ref_format in SUBTITLE_EXTENSIONS:
        if args.vad is not None:
            logger.warning("Vad specified, but reference was not a movie")
        return cast(
            Pipeline,
            make_subtitle_speech_pipeline(
                fmt=ref_format,
                **override(args, encoding=args.reference_encoding or DEFAULT_ENCODING),
            ),
        )
    elif ref_format in ("npy", "npz"):
        if args.vad is not None:
            logger.warning("Vad specified, but reference was not a movie")
        return Pipeline(
            [("deserialize", DeserializeSpeechTransformer(args.non_speech_label))]
        )
    else:
        vad = args.vad or DEFAULT_VAD
        if args.reference_encoding is not None:
            logger.warning(
                "Reference srt encoding specified, but reference was a video file"
            )
        ref_stream = args.reference_stream
        if ref_stream is not None and not ref_stream.startswith("0:"):
            ref_stream = "0:" + ref_stream
        if getattr(args, "multi_segment_sync", False):
            return Pipeline(
                [
                    (
                        "speech_extract",
                        MultiSegmentVideoSpeechTransformer(
                            vad=vad,
                            sample_rate=SAMPLE_RATE,
                            frame_rate=args.frame_rate,
                            non_speech_label=args.non_speech_label,
                            segment_count=getattr(args, "segment_count", 8),
                            skip_intro_outro=getattr(
                                args, "skip_intro_outro", False
                            ),
                            parallel_workers=getattr(args, "parallel_workers", 4),
                            ffmpeg_path=args.ffmpeg_path,
                            ref_stream=ref_stream,
                            vlc_mode=args.vlc_mode,
                            gui_mode=args.gui_mode,
                        ),
                    ),
                ]
            )
        return Pipeline(
            [
                (
                    "speech_extract",
                    VideoSpeechTransformer(
                        vad=vad,
                        sample_rate=SAMPLE_RATE,
                        frame_rate=args.frame_rate,
                        non_speech_label=args.non_speech_label,
                        start_seconds=args.start_seconds,
                        ffmpeg_path=args.ffmpeg_path,
                        ref_stream=ref_stream,
                        vlc_mode=args.vlc_mode,
                        gui_mode=args.gui_mode,
                        max_duration_seconds=getattr(
                            args, "max_duration_seconds", None
                        ),
                        extract_audio_first=getattr(
                            args, "extract_audio_first", False
                        ),
                        progress_handler=progress_handler,
                    ),
                ),
            ]
        )


def extract_subtitles_from_reference(args: argparse.Namespace) -> int:
    stream = args.extract_subs_from_stream
    if not stream.startswith("0:s:"):
        stream = "0:s:{}".format(stream)
    elif not stream.startswith("0:") and stream.startswith("s:"):
        stream = "0:{}".format(stream)
    if not stream.startswith("0:s:"):
        logger.error(
            "invalid stream for subtitle extraction: %s", args.extract_subs_from_stream
        )
    ffmpeg_args = [
        ffmpeg_bin_path("ffmpeg", args.gui_mode, ffmpeg_resources_path=args.ffmpeg_path)
    ]
    ffmpeg_args.extend(
        [
            "-y",
            "-nostdin",
            "-loglevel",
            "fatal",
            "-i",
            args.reference,
            "-map",
            "{}".format(stream),
            "-f",
            "srt",
        ]
    )
    if args.srtout is None:
        ffmpeg_args.append("-")
    else:
        ffmpeg_args.append(args.srtout)
    logger.info(
        "attempting to extract subtitles to {} ...".format(
            "stdout" if args.srtout is None else args.srtout
        )
    )
    retcode = subprocess.call(ffmpeg_args)
    if retcode == 0:
        logger.info("...done")
    else:
        logger.error(
            "ffmpeg unable to extract subtitles from reference; return code %d", retcode
        )
    return retcode


def _detect_srtin_from_reference(reference: str) -> List[str]:
    """Find subtitle files that sit next to the reference and share its name.

    Matches `<reference-stem>.srt` and `<reference-stem>.<suffix>.srt` (e.g.
    `movie.srt`, `movie.en.srt` for a `movie.mkv` reference) in the reference's
    own directory, so detection works regardless of the process's current
    working directory. The reference itself is never returned, so this is safe
    even when the reference is already a subtitle file.
    """
    reference_dir = os.path.dirname(reference) or "."
    reference_stem = os.path.splitext(os.path.basename(reference))[0]
    reference_abspath = os.path.abspath(reference)
    matches = []
    for name in sorted(os.listdir(reference_dir)):
        stem, ext = os.path.splitext(name)
        if ext.lower() != ".srt":
            continue
        if name.endswith(".synced.srt"):
            continue  # skip our own previously-synced outputs so re-runs are idempotent
        if stem != reference_stem and not stem.startswith(reference_stem + "."):
            continue
        path = os.path.join(reference_dir, name)
        if os.path.abspath(path) == reference_abspath:
            continue
        matches.append(path)
    return matches


def validate_args(args: argparse.Namespace) -> None:
    if args.vlc_mode:
        logger.setLevel(logging.CRITICAL)
    if args.reference is None:
        if args.apply_offset_seconds == 0 or not args.srtin:
            raise ValueError(
                "`reference` required unless `--apply-offset-seconds` specified"
            )
    if args.apply_offset_seconds != 0:
        if not args.srtin:
            args.srtin = [args.reference]
        if not args.srtin:
            raise ValueError(
                "at least one of `srtin` or `reference` must be specified to apply offset seconds"
            )
    if args.srtin:
        if len(args.srtin) > 1 and not args.overwrite_input:
            raise ValueError(
                "cannot specify multiple input srt files without overwriting"
            )
        if len(args.srtin) > 1 and args.make_test_case:
            raise ValueError("cannot specify multiple input srt files for test cases")
        if len(args.srtin) > 1 and args.gui_mode:
            raise ValueError("cannot specify multiple input srt files in GUI mode")
    elif (
        args.reference is not None
        and not is_remote_url(args.reference)  # can't list a remote dir for sibling srts
        and args.extract_subs_from_stream is None
        and not args.gui_mode
        and not args.make_test_case
        and sys.stdin.isatty()  # don't hijack subtitles piped in on stdin
    ):
        # No input subtitles were given (and nothing is piped in), so look for
        # subtitle files alongside the reference that share its name.
        logger.info("no input srt specified; detecting input srt from reference")
        detected = _detect_srtin_from_reference(args.reference)
        if detected:
            for path in detected:
                logger.info("detected input srt: %s", path)
            args.srtin = detected
            if len(detected) > 1 and args.srtout is not None:
                raise ValueError(
                    "detected multiple input srt files but an output file was "
                    "specified; re-run with --overwrite-input or a single input"
                )
            if args.srtout is None and not args.overwrite_input:
                args.auto_srtout = True
                logger.info(
                    "writing synced output alongside each input as "
                    "<name>.synced.srt; pass --overwrite-input to overwrite the "
                    "input file(s) in place instead"
                )
        else:
            logger.info("no input srt detected from reference")
    if (
        args.make_test_case and not args.gui_mode
    ):  # this validation not necessary for gui mode
        if not args.srtin or args.srtout is None:
            raise ValueError(
                "need to specify input and output srt files for test cases"
            )
    if args.overwrite_input:
        if args.extract_subs_from_stream is not None:
            raise ValueError(
                "input overwriting not allowed for extracting subtitles from reference"
            )
        if not args.srtin:
            raise ValueError(
                "need to specify input srt if --overwrite-input "
                "is specified since we cannot overwrite stdin"
            )
        if args.srtout is not None:
            raise ValueError(
                "overwrite input set but output file specified; "
                "refusing to run in case this was not intended"
            )
    if args.extract_subs_from_stream is not None:
        if args.make_test_case:
            raise ValueError("test case is for sync and not subtitle extraction")
        if args.srtin:
            raise ValueError(
                "stream specified for reference subtitle extraction; "
                "-i flag for sync input not allowed"
            )


def validate_file_permissions(args: argparse.Namespace) -> None:
    error_string_template = (
        "unable to {action} {file}; "
        "try ensuring file exists and has correct permissions"
    )
    if (
        args.reference is not None
        and not is_remote_url(args.reference)  # ffmpeg streams URLs directly; no local file to check
        and not os.access(args.reference, os.R_OK)
    ):
        raise ValueError(
            error_string_template.format(action="read reference", file=args.reference)
        )
    if args.srtin:
        for srtin in args.srtin:
            if srtin is not None and not os.access(srtin, os.R_OK):
                raise ValueError(
                    error_string_template.format(
                        action="read input subtitles", file=srtin
                    )
                )
    if (
        args.srtout is not None
        and os.path.exists(args.srtout)
        and not os.access(args.srtout, os.W_OK)
    ):
        raise ValueError(
            error_string_template.format(
                action="write output subtitles", file=args.srtout
            )
        )
    if args.make_test_case or args.serialize_speech:
        npy_savename = os.path.splitext(args.reference)[0] + ".npz"
        if os.path.exists(npy_savename) and not os.access(npy_savename, os.W_OK):
            raise ValueError(
                "unable to write test case file archive %s (try checking permissions)"
                % npy_savename
            )


def _setup_logging(
    args: argparse.Namespace,
) -> Tuple[Optional[str], Optional[logging.FileHandler]]:
    log_handler = None
    log_path = None
    if args.make_test_case or args.log_dir_path is not None:
        log_path = "ffsubsync.log"
        if args.log_dir_path is not None and os.path.isdir(args.log_dir_path):
            log_path = os.path.join(args.log_dir_path, log_path)
        log_handler = logging.FileHandler(log_path)
        logger.addHandler(log_handler)
        logger.info("this log will be written to %s", os.path.abspath(log_path))
    return log_path, log_handler


def _npy_savename(args: argparse.Namespace) -> str:
    return os.path.splitext(args.reference)[0] + ".npz"


def _run_impl(
    args: argparse.Namespace,
    result: Dict[str, Any],
    progress_handler: Optional[Callable[[ProgressInfo], None]] = None,
) -> bool:
    if args.extract_subs_from_stream is not None:
        result["retval"] = extract_subtitles_from_reference(args)
        return True
    if args.srtin is not None and (
        args.reference is None
        or (len(args.srtin) == 1 and args.srtin[0] == args.reference)
    ):
        return try_sync(args, None, result)
    reference_pipe = make_reference_pipe(args, progress_handler=progress_handler)
    logger.info("extracting speech segments from reference '%s'...", args.reference)
    reference_pipe.fit(args.reference)
    logger.info("...done")
    if args.make_test_case or args.serialize_speech:
        logger.info("serializing speech...")
        np.savez_compressed(
            _npy_savename(args), speech=reference_pipe.transform(args.reference)
        )
        logger.info("...done")
        if not args.srtin:
            logger.info(
                "unsynchronized subtitle file not specified; skipping synchronization"
            )
            return False
    return try_sync(args, reference_pipe, result)


def validate_and_transform_args(
    parser_or_args: Union[argparse.ArgumentParser, argparse.Namespace],
) -> Optional[argparse.Namespace]:
    if isinstance(parser_or_args, argparse.Namespace):
        parser = None
        args = parser_or_args
    else:
        parser = parser_or_args
        args = parser.parse_args()
    try:
        validate_args(args)
    except ValueError as e:
        logger.error(e)
        if parser is not None:
            parser.print_usage()
        return None
    if args.gui_mode and args.srtout is None:
        args.srtout = "{}.synced.srt".format(os.path.splitext(args.srtin[0])[0])
    try:
        validate_file_permissions(args)
    except ValueError as e:
        logger.error(e)
        return None
    ref_format = _ref_format(args.reference)
    if args.merge_with_reference and ref_format not in SUBTITLE_EXTENSIONS:
        logger.error(
            "merging synced output with reference only valid "
            "when reference composed of subtitles"
        )
        return None
    return args


def run(
    parser_or_args: Union[argparse.ArgumentParser, argparse.Namespace],
    progress_handler: Optional[Callable[[ProgressInfo], None]] = None,
) -> Dict[str, Any]:
    """Synchronize subtitles.

    ``progress_handler``, if given, is called repeatedly during reference speech
    extraction with a :class:`~ffsubsync.speech_transformers.ProgressInfo`
    describing how much of the reference audio has been processed. This lets a
    host application (e.g. a web UI) display sync progress. The handler is only
    invoked for the video-reference path; exceptions it raises are logged and
    swallowed so a buggy handler cannot abort syncing.
    """
    sync_was_successful = False
    result = {
        "retval": 0,
        "offset_seconds": None,
        "framerate_scale_factor": None,
    }
    args = validate_and_transform_args(parser_or_args)
    if args is None:
        result["retval"] = 1
        return result
    log_path, log_handler = _setup_logging(args)
    try:
        sync_was_successful = _run_impl(
            args, result, progress_handler=progress_handler
        )
        result["sync_was_successful"] = sync_was_successful
        return result
    finally:
        if log_handler is not None and log_path is not None:
            log_handler.close()
            logger.removeHandler(log_handler)
            if args.make_test_case:
                result["retval"] += make_test_case(
                    args, _npy_savename(args), sync_was_successful
                )
            if args.log_dir_path is None or not os.path.isdir(args.log_dir_path):
                os.remove(log_path)


def add_main_args_for_cli(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "reference",
        nargs="?",
        help=(
            "Reference (video, subtitles, or a numpy array with VAD speech) "
            "to which to synchronize input subtitles."
        ),
    )
    parser.add_argument(
        "-i",
        "--srtin",
        nargs="*",
        help=(
            "Input subtitles file (default=stdin). If omitted (and nothing is "
            "piped in), subtitles sharing the reference's name in its directory "
            "are auto-detected (e.g. `movie.srt`, `movie.en.srt` for `movie.mkv`) "
            "and each is synced to a `<name>.synced.srt` next to it; pass "
            "--overwrite-input to overwrite the detected file(s) in place."
        ),
    )
    parser.add_argument(
        "-o", "--srtout", help="Output subtitles file (default=stdout)."
    )
    parser.add_argument(
        "--merge-with-reference",
        "--merge",
        action="store_true",
        help="Merge reference subtitles with synced output subtitles.",
    )
    parser.add_argument(
        "--make-test-case",
        "--create-test-case",
        action="store_true",
        help="If specified, serialize reference speech to a numpy array, "
        "and create an archive with input/output subtitles "
        "and serialized speech.",
    )
    parser.add_argument(
        "--reference-stream",
        "--refstream",
        "--reference-track",
        "--reftrack",
        default=None,
        help=(
            "Which stream/track in the video file to use as reference, "
            "formatted according to ffmpeg conventions. For example, 0:s:0 "
            "uses the first subtitle track; 0:a:3 would use the third audio track. "
            "You can also drop the leading `0:`; i.e. use s:0 or a:3, respectively. "
            "Example: `ffs ref.mkv -i in.srt -o out.srt --reference-stream s:2`"
        ),
    )
    parser.add_argument(
        "--pgs-ref-stream",
        "--pgsstream",
        nargs="?",
        const="auto",
        default=None,
        help=(
            "Use a PGS (Presentation Graphic Stream) image-based subtitle track from "
            "the reference MKV as the sync reference instead of audio VAD. "
            "Optionally specify the stream (leading `0:` is optional, e.g. `s:0` or `3`). "
            "Omit the value to auto-detect the first hdmv_pgs_subtitle track. "
            "Example: `ffs ref.mkv -i in.srt -o out.srt --pgs-ref-stream` (auto) "
            "or `ffs ref.mkv -i in.srt -o out.srt --pgs-ref-stream s:2` (explicit)."
        ),
    )


def add_cli_only_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "-v",
        "--version",
        action="version",
        version="{package} {version}".format(
            package=__package__ or "ffsubsync", version=get_version()
        ),
    )
    parser.add_argument(
        "--overwrite-input",
        action="store_true",
        help=(
            "If specified, will overwrite the input srt "
            "instead of writing the output to a new file."
        ),
    )
    parser.add_argument(
        "--encoding",
        default=DEFAULT_ENCODING,
        help="What encoding to use for reading input subtitles "
        "(default=%s)." % DEFAULT_ENCODING,
    )
    parser.add_argument(
        "--max-subtitle-seconds",
        type=float,
        default=DEFAULT_MAX_SUBTITLE_SECONDS,
        help="Maximum duration for a subtitle to appear on-screen "
        "(default=%.3f seconds)." % DEFAULT_MAX_SUBTITLE_SECONDS,
    )
    parser.add_argument(
        "--start-seconds",
        type=int,
        default=DEFAULT_START_SECONDS,
        help="Start time for processing "
        "(default=%d seconds)." % DEFAULT_START_SECONDS,
    )
    parser.add_argument(
        "--max-offset-seconds",
        type=float,
        default=DEFAULT_MAX_OFFSET_SECONDS,
        help="The max allowed offset seconds for any subtitle segment "
        "(default=%d seconds)." % DEFAULT_MAX_OFFSET_SECONDS,
    )
    parser.add_argument(
        "--max-duration-seconds",
        type=float,
        default=None,
        help="If specified, only process the first this-many seconds of the "
        "reference (measured from --start-seconds). Useful for speeding up "
        "long or remote references, since ffmpeg stops reading/downloading "
        "once this duration is reached.",
    )
    parser.add_argument(
        "--extract-audio-first",
        action="store_true",
        help="For remote URL references, first copy the audio track to a local "
        "temp file (no re-encode) and run speech detection on that, instead of "
        "streaming the full container over the network during detection. Can be "
        "more stable on flaky connections; ignored for local references.",
    )
    parser.add_argument(
        "--multi-segment-sync",
        action="store_true",
        help="Sample a few short segments spread across the reference and run "
        "speech detection only on those, instead of the whole reference. Speeds "
        "up long or remote references; the usual framerate and offset search is "
        "unchanged. Only applies to video / audio references.",
    )
    parser.add_argument(
        "--segment-count",
        type=int,
        default=8,
        help="Number of segments to sample for --multi-segment-sync (default=8).",
    )
    parser.add_argument(
        "--skip-intro-outro",
        action="store_true",
        help="With --multi-segment-sync, skip the first 30s and last 60s of the "
        "reference when placing segments (intros/credits often lack dialogue).",
    )
    parser.add_argument(
        "--parallel-workers",
        type=int,
        default=4,
        help="How many segments to extract in parallel for --multi-segment-sync "
        "(default=4); useful for overlapping downloads of remote references.",
    )
    parser.add_argument(
        "--apply-offset-seconds",
        type=float,
        default=DEFAULT_APPLY_OFFSET_SECONDS,
        help="Apply a predefined offset in seconds to all subtitle segments "
        "(default=%d seconds)." % DEFAULT_APPLY_OFFSET_SECONDS,
    )
    parser.add_argument(
        "--skip-sync-on-low-quality",
        action="store_true",
        help="If the alignment looks untrustworthy (see the thresholds below), "
        "leave the subtitles unmodified instead of applying a probably-wrong "
        "sync. Useful for batch jobs where a bad sync is worse than none.",
    )
    parser.add_argument(
        "--min-score",
        type=float,
        default=DEFAULT_MIN_SCORE,
        help="With --skip-sync-on-low-quality, reject alignments scoring below "
        "this. The score's magnitude is not normalized, but its sign is "
        "meaningful, so the default of %.1f rejects only anti-correlated "
        "(clearly wrong) alignments." % DEFAULT_MIN_SCORE,
    )
    parser.add_argument(
        "--quality-max-offset-seconds",
        type=float,
        default=DEFAULT_QUALITY_MAX_OFFSET_SECONDS,
        help="With --skip-sync-on-low-quality, reject alignments whose offset "
        "exceeds this many seconds (default=%.1f)."
        % DEFAULT_QUALITY_MAX_OFFSET_SECONDS,
    )
    parser.add_argument(
        "--max-framerate-deviation",
        type=float,
        default=DEFAULT_MAX_FRAMERATE_DEVIATION,
        help="With --skip-sync-on-low-quality, reject alignments whose framerate "
        "scale deviates from 1.0 by more than this. The default of %.2f permits "
        "every framerate correction ffsubsync can make (so it never rejects a "
        "legitimate one); tighten it only when you know the framerate should not "
        "change." % DEFAULT_MAX_FRAMERATE_DEVIATION,
    )
    parser.add_argument(
        "--frame-rate",
        type=int,
        default=DEFAULT_FRAME_RATE,
        help="Frame rate for audio extraction (default=%d)." % DEFAULT_FRAME_RATE,
    )
    parser.add_argument(
        "--skip-infer-framerate-ratio",
        action="store_true",
        help="If set, do not try to infer framerate ratio based on duration ratio.",
    )
    parser.add_argument(
        "--non-speech-label",
        type=float,
        default=DEFAULT_NON_SPEECH_LABEL,
        help="Label to use for frames detected as non-speech (default=%f)"
        % DEFAULT_NON_SPEECH_LABEL,
    )
    parser.add_argument(
        "--output-encoding",
        default="utf-8",
        help="What encoding to use for writing output subtitles "
        '(default=utf-8). Can indicate "same" to use same '
        "encoding as that of the input.",
    )
    parser.add_argument(
        "--reference-encoding",
        help="What encoding to use for reading / writing reference subtitles "
        "(if applicable, default=infer).",
    )
    parser.add_argument(
        "--vad",
        choices=[
            "subs_then_webrtc",
            "webrtc",
            "subs_then_auditok",
            "auditok",
            "subs_then_silero",
            "silero",
            "fused",
            "fused:weighted",
            "fused:intersection",
            "fused:union",
        ],
        default=None,
        help="Which voice activity detector to use for speech extraction "
        "(if using video / audio as a reference, default={}). The `fused` "
        "options combine webrtc and silero and require the optional silero "
        "dependency (torch).".format(DEFAULT_VAD),
    )
    parser.add_argument(
        "--no-fix-framerate",
        action="store_true",
        help="If specified, subsync will not attempt to correct a framerate "
        "mismatch between reference and subtitles.",
    )
    parser.add_argument(
        "--serialize-speech",
        action="store_true",
        help="If specified, serialize reference speech to a numpy array.",
    )
    parser.add_argument(
        "--extract-subs-from-stream",
        "--extract-subtitles-from-stream",
        default=None,
        help="If specified, do not attempt sync; instead, just extract subtitles"
        " from the specified stream using the reference.",
    )
    parser.add_argument(
        "--suppress-output-if-offset-less-than",
        type=float,
        default=None,
        help="If specified, do not produce output if offset below provided threshold.",
    )
    parser.add_argument(
        "--ffmpeg-path",
        "--ffmpegpath",
        default=None,
        help="Where to look for ffmpeg and ffprobe. Uses the system PATH by default.",
    )
    parser.add_argument(
        "--log-dir-path",
        default=None,
        help=(
            "If provided, will save log file ffsubsync.log to this path "
            "(must be an existing directory)."
        ),
    )
    parser.add_argument(
        "--gss",
        action="store_true",
        help="If specified, use golden-section search to try to find"
        "the optimal framerate ratio between video and subtitles.",
    )
    parser.add_argument(
        "--strict",
        action="store_true",
        help="If specified, refuse to parse srt files with formatting issues.",
    )
    parser.add_argument("--vlc-mode", action="store_true", help=argparse.SUPPRESS)
    parser.add_argument("--gui-mode", action="store_true", help=argparse.SUPPRESS)
    parser.add_argument("--skip-sync", action="store_true", help=argparse.SUPPRESS)


def make_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Synchronize subtitles with video.")
    add_main_args_for_cli(parser)
    add_cli_only_args(parser)
    return parser


def main() -> int:
    parser = make_parser()
    return run(parser)["retval"]


if __name__ == "__main__":
    sys.exit(main())