rw_ffmpeg.py

"""
Record Wizard — everything that talks to ffmpeg.

Deliberately free of Qt widgets so the command building and process
handling can be exercised on their own.

Design notes
------------
* stderr goes to a **log file**, never a pipe.  ffmpeg writes to stderr the
  whole time it runs; an unread pipe fills its ~64 KB buffer after a few
  minutes and ffmpeg then blocks forever mid-recording.
* the child is put in a Win32 job object with KILL_ON_JOB_CLOSE, so a crash
  of this app can't leave an invisible ffmpeg recording forever.
* stopping is a *blocking* operation with a timeout — the caller must wait
  for it, otherwise the MP4 never gets its moov atom and won't play.
"""

from __future__ import annotations

import os
import re
import sys
import ctypes
import shutil
import subprocess
import threading
import time
from datetime import datetime

from rw_util import app_data_dir, NO_WINDOW, even

# ═══════════════════════════════════════════════════════════════════
# Locating ffmpeg
# ═══════════════════════════════════════════════════════════════════
_ffmpeg_path: str | None = None
_ffmpeg_checked = False


def _candidates() -> list[str]:
    out = []
    env = os.environ.get("RECORD_WIZARD_FFMPEG")
    if env:
        out.append(env)
    here = getattr(sys, "_MEIPASS", os.path.dirname(os.path.abspath(__file__)))
    exe_dir = os.path.dirname(sys.executable) if getattr(sys, "frozen", False) else here
    for base in {here, exe_dir}:
        out.append(os.path.join(base, "ffmpeg.exe"))
        out.append(os.path.join(base, "bin", "ffmpeg.exe"))
    which = shutil.which("ffmpeg")
    if which:
        out.append(which)
    local = os.environ.get("LOCALAPPDATA", "")
    out += [
        r"C:\ffmpeg\bin\ffmpeg.exe",
        r"C:\Program Files\ffmpeg\bin\ffmpeg.exe",
        os.path.join(local, "Microsoft", "WinGet", "Links", "ffmpeg.exe"),
    ]
    return out


def find_ffmpeg(force=False) -> str | None:
    """Absolute path to a working ffmpeg, or None. Result is cached."""
    global _ffmpeg_path, _ffmpeg_checked
    if _ffmpeg_checked and not force:
        return _ffmpeg_path
    _ffmpeg_checked = True
    _ffmpeg_path = None
    for cand in _candidates():
        if not cand or not os.path.isfile(cand):
            continue
        try:
            r = subprocess.run([cand, "-version"], capture_output=True,
                               timeout=10, creationflags=NO_WINDOW)
            if r.returncode == 0:
                _ffmpeg_path = cand
                break
        except Exception:
            continue
    return _ffmpeg_path


FFMPEG_HELP = (
    "ffmpeg was not found.\n\n"
    "Install it with:    winget install Gyan.FFmpeg\n"
    "or download it from https://ffmpeg.org/download.html and either put "
    "ffmpeg.exe on your PATH or next to Record Wizard."
)


# ═══════════════════════════════════════════════════════════════════
# Device enumeration
# ═══════════════════════════════════════════════════════════════════
# Names that indicate a loopback / "what the speakers are playing" device.
# Only such a device can record computer audio; a plain mic cannot.
LOOPBACK_HINTS = (
    "stereo mix", "stereomix", "what u hear", "what you hear", "wave out",
    "cable output", "voicemeeter out", "vb-audio", "loopback",
    "mixage stéréo", "mixagem estéreo", "stereomischung", "missaggio stereo",
)

_DEV_LINE = re.compile(r'^\s*(?:\[[^\]]*\]\s*)?"(?P<name>[^"]+)"\s*(?:\((?P<kind>audio|video)\))?\s*$')


def parse_device_list(stderr_text: str) -> list[str]:
    """Extract dshow *audio* device names from `-list_devices true` output.

    Handles both layouts:
      old (<= ffmpeg 7):  a "DirectShow audio devices" header, then bare names
      new (ffmpeg 8.x):   every line tagged, e.g.  "Microphone (X)" (audio)
    """
    names: list[str] = []
    section: str | None = None
    for raw in stderr_text.splitlines():
        low = raw.lower()
        if "directshow" in low and "devices" in low:
            section = "audio" if "audio" in low else "video"
            continue
        if "alternative name" in low:
            continue
        m = _DEV_LINE.match(raw)
        if not m:
            continue
        kind = m.group("kind") or section
        if kind == "audio":
            name = m.group("name").strip()
            if name and name not in names:
                names.append(name)
    return names


def list_audio_devices(ffmpeg: str | None = None) -> list[str]:
    ffmpeg = ffmpeg or find_ffmpeg()
    if not ffmpeg:
        return []
    try:
        r = subprocess.run(
            [ffmpeg, "-hide_banner", "-list_devices", "true",
             "-f", "dshow", "-i", "dummy"],
            capture_output=True, text=True, errors="replace",
            timeout=15, creationflags=NO_WINDOW)
    except Exception:
        return []
    return parse_device_list(r.stderr or "")


def is_loopback(name: str) -> bool:
    low = (name or "").lower()
    return any(h in low for h in LOOPBACK_HINTS)


def pick_default_device(devices: list[str]) -> str | None:
    """Prefer a loopback device (real computer audio), else the first mic."""
    for d in devices:
        if is_loopback(d):
            return d
    return devices[0] if devices else None


class AudioDevices:
    """Cached, background-refreshed dshow audio device list.

    Enumeration shells out to ffmpeg and can take a second or two, which is
    far too long to do on the UI thread every time Record is pressed.
    """

    def __init__(self):
        self._devices: list[str] = []
        self._ready = threading.Event()
        self._lock = threading.Lock()

    def refresh_async(self, on_done=None):
        def _run():
            devs = list_audio_devices()
            with self._lock:
                self._devices = devs
            self._ready.set()
            if on_done:
                try:
                    on_done(devs)
                except Exception:
                    pass
        threading.Thread(target=_run, daemon=True).start()

    def get(self, timeout=6.0) -> list[str]:
        self._ready.wait(timeout)
        with self._lock:
            return list(self._devices)

    @property
    def ready(self) -> bool:
        return self._ready.is_set()

    def cached(self) -> list[str]:
        with self._lock:
            return list(self._devices)


# ═══════════════════════════════════════════════════════════════════
# Encoders
# ═══════════════════════════════════════════════════════════════════
HW_ENCODERS = [
    ("h264_nvenc", "NVIDIA (NVENC)"),
    ("h264_qsv",   "Intel Quick Sync"),
    ("h264_amf",   "AMD (AMF)"),
]

# Plain-language names + an explanation of the trade-off. "Encoder" means
# nothing to most people; what they care about is what it costs them.
ENCODER_LABELS = {
    "libx264": (
        "Use the processor",
        "Works on every PC. The CPU does the compression, so a busy "
        "machine may drop frames while recording."),
    "h264_nvenc": (
        "Use the graphics card (NVIDIA)",
        "Your NVIDIA card does the compression. Much lighter on the "
        "processor — best choice for games and full-screen video."),
    "h264_qsv": (
        "Use the graphics chip (Intel)",
        "Your Intel graphics does the compression. Much lighter on the "
        "processor."),
    "h264_amf": (
        "Use the graphics card (AMD)",
        "Your AMD card does the compression. Much lighter on the "
        "processor — best choice for games and full-screen video."),
}


def encoder_label(enc: str) -> tuple[str, str]:
    return ENCODER_LABELS.get(enc, (enc, ""))

_encoder_cache: list[str] | None = None
_probe_cache: dict[str, bool] = {}


def probe_encoder(enc: str, ffmpeg: str | None = None, timeout=25) -> bool:
    """Does this encoder actually initialise on this machine?

    A build can advertise h264_qsv with no Intel GPU present, so the only
    honest test is a throwaway encode.  Costs well under a second.
    """
    if enc in _probe_cache:
        return _probe_cache[enc]
    ffmpeg = ffmpeg or find_ffmpeg()
    ok = False
    if ffmpeg:
        try:
            r = subprocess.run(
                [ffmpeg, "-hide_banner", "-loglevel", "error",
                 "-f", "lavfi", "-i", "nullsrc=s=256x144:d=0.2",
                 "-c:v", enc, "-f", "null", "-"],
                capture_output=True, timeout=timeout, creationflags=NO_WINDOW)
            ok = r.returncode == 0
        except Exception:
            ok = False
    _probe_cache[enc] = ok
    return ok


class EncoderSupport:
    """Background probe of which encoders really work here."""

    def __init__(self):
        self._working: list[str] = ["libx264"]
        self._broken: list[str] = []
        self._ready = threading.Event()
        self._lock = threading.Lock()

    def refresh_async(self):
        def _run():
            ffmpeg = find_ffmpeg()
            working, broken = [], []
            for enc in available_encoders(ffmpeg):
                (working if probe_encoder(enc, ffmpeg) else broken).append(enc)
            if "libx264" not in working:
                working.insert(0, "libx264")   # always offer a way to record
            with self._lock:
                self._working, self._broken = working, broken
            self._ready.set()
        threading.Thread(target=_run, daemon=True).start()

    @property
    def ready(self) -> bool:
        return self._ready.is_set()

    def working(self) -> list[str]:
        with self._lock:
            return list(self._working)

    def broken(self) -> list[str]:
        with self._lock:
            return list(self._broken)


def available_encoders(ffmpeg: str | None = None) -> list[str]:
    """H.264 encoders this ffmpeg build offers (software first)."""
    global _encoder_cache
    if _encoder_cache is not None:
        return _encoder_cache
    ffmpeg = ffmpeg or find_ffmpeg()
    found = ["libx264"]
    if ffmpeg:
        try:
            r = subprocess.run([ffmpeg, "-hide_banner", "-encoders"],
                               capture_output=True, text=True, errors="replace",
                               timeout=15, creationflags=NO_WINDOW)
            text = (r.stdout or "") + (r.stderr or "")
            for enc, _label in HW_ENCODERS:
                if re.search(rf"\b{enc}\b", text):
                    found.append(enc)
        except Exception:
            pass
    _encoder_cache = found
    return found


# CRF-ish value per quality level
_QUALITY_Q = {"high": 18, "balanced": 23, "small": 28}


def _encoder_args(encoder: str, quality: str, fps: int) -> list[str]:
    q = _QUALITY_Q.get(quality, 23)
    gop = str(max(2, fps * 2))
    if encoder == "h264_nvenc":
        return ["-c:v", "h264_nvenc", "-preset", "p5", "-rc", "vbr",
                "-cq", str(q), "-b:v", "0", "-g", gop]
    if encoder == "h264_qsv":
        return ["-c:v", "h264_qsv", "-global_quality", str(q), "-g", gop]
    if encoder == "h264_amf":
        return ["-c:v", "h264_amf", "-rc", "cqp",
                "-qp_i", str(q), "-qp_p", str(q), "-g", gop]
    # libx264. 'veryfast' instead of 'ultrafast': still real-time on any
    # modern CPU and roughly half the file size.
    return ["-c:v", "libx264", "-preset", "veryfast", "-crf", str(q),
            "-g", gop]


def _dshow_escape(name: str) -> str:
    """ffmpeg's dshow parser treats ':' as a separator; escape it."""
    return name.replace("\\", "\\\\").replace(":", "\\:")


def build_command(ffmpeg: str, *, out_path: str, video: bool = True,
                  region_phys: tuple[int, int, int, int] | None = None,
                  fps: int = 30, cursor: bool = True,
                  audio_device: str | None = None,
                  encoder: str = "libx264",
                  quality: str = "balanced") -> list[str]:
    """Build the full ffmpeg argv for one capture part."""
    cmd = [ffmpeg, "-hide_banner", "-nostats", "-loglevel", "warning", "-y"]

    if not video:
        if not audio_device:
            raise ValueError("audio-only capture needs a device")
        cmd += ["-f", "dshow", "-thread_queue_size", "1024",
                "-use_wallclock_as_timestamps", "1",
                "-i", f"audio={_dshow_escape(audio_device)}",
                "-c:a", "libmp3lame", "-q:a", "2", out_path]
        return cmd

    cmd += ["-f", "gdigrab",
            "-framerate", str(fps),
            "-draw_mouse", "1" if cursor else "0",
            "-thread_queue_size", "1024"]
    if region_phys:
        x, y, w, h = region_phys
        # H.264 + yuv420p requires even dimensions, otherwise ffmpeg exits
        # immediately with "width not divisible by 2".
        cmd += ["-offset_x", str(int(x)), "-offset_y", str(int(y)),
                "-video_size", f"{even(w)}x{even(h)}"]
    cmd += ["-i", "desktop"]

    if audio_device:
        cmd += ["-f", "dshow", "-thread_queue_size", "1024",
                "-use_wallclock_as_timestamps", "1",
                "-i", f"audio={_dshow_escape(audio_device)}"]

    if not region_phys:
        # Full-screen capture can also land on an odd size; cropping a pixel
        # is cheaper and lossless compared with rescaling.
        cmd += ["-vf", "crop=trunc(iw/2)*2:trunc(ih/2)*2"]

    cmd += _encoder_args(encoder, quality, fps)
    cmd += ["-pix_fmt", "yuv420p"]

    if audio_device:
        # aresample keeps the dshow clock from drifting away from gdigrab's.
        cmd += ["-c:a", "aac", "-b:a", "192k",
                "-af", "aresample=async=1:first_pts=0"]
    else:
        cmd += ["-an"]

    cmd.append(out_path)
    return cmd


# ═══════════════════════════════════════════════════════════════════
# Kill-on-close job object  (no orphaned ffmpeg if we die unexpectedly)
# ═══════════════════════════════════════════════════════════════════
_jobs: list = []  # keep handles alive for the life of the process


def _assign_kill_on_close_job(proc: subprocess.Popen) -> bool:
    if sys.platform != "win32":
        return False
    try:
        class IO_COUNTERS(ctypes.Structure):
            _fields_ = [(n, ctypes.c_ulonglong) for n in (
                "ReadOperationCount", "WriteOperationCount",
                "OtherOperationCount", "ReadTransferCount",
                "WriteTransferCount", "OtherTransferCount")]

        class BASIC_LIMIT(ctypes.Structure):
            _fields_ = [
                ("PerProcessUserTimeLimit", ctypes.c_longlong),
                ("PerJobUserTimeLimit", ctypes.c_longlong),
                ("LimitFlags", ctypes.c_ulong),
                ("MinimumWorkingSetSize", ctypes.c_size_t),
                ("MaximumWorkingSetSize", ctypes.c_size_t),
                ("ActiveProcessLimit", ctypes.c_ulong),
                ("Affinity", ctypes.c_size_t),
                ("PriorityClass", ctypes.c_ulong),
                ("SchedulingClass", ctypes.c_ulong)]

        class EXTENDED_LIMIT(ctypes.Structure):
            _fields_ = [
                ("BasicLimitInformation", BASIC_LIMIT),
                ("IoInfo", IO_COUNTERS),
                ("ProcessMemoryLimit", ctypes.c_size_t),
                ("JobMemoryLimit", ctypes.c_size_t),
                ("PeakProcessMemoryUsed", ctypes.c_size_t),
                ("PeakJobMemoryUsed", ctypes.c_size_t)]

        JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x2000
        JobObjectExtendedLimitInformation = 9

        k32 = ctypes.windll.kernel32
        k32.CreateJobObjectW.restype = ctypes.c_void_p
        job = k32.CreateJobObjectW(None, None)
        if not job:
            return False
        info = EXTENDED_LIMIT()
        info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
        if not k32.SetInformationJobObject(
                ctypes.c_void_p(job), JobObjectExtendedLimitInformation,
                ctypes.byref(info), ctypes.sizeof(info)):
            k32.CloseHandle(ctypes.c_void_p(job))
            return False
        handle = int(proc._handle)  # Popen keeps a Windows HANDLE on win32
        if not k32.AssignProcessToJobObject(ctypes.c_void_p(job),
                                            ctypes.c_void_p(handle)):
            k32.CloseHandle(ctypes.c_void_p(job))
            return False
        _jobs.append(job)
        return True
    except Exception:
        return False


# ═══════════════════════════════════════════════════════════════════
# One ffmpeg invocation
# ═══════════════════════════════════════════════════════════════════
def log_dir() -> str:
    d = os.path.join(app_data_dir(), "logs")
    try:
        os.makedirs(d, exist_ok=True)
    except OSError:
        pass
    return d


_log_dir = log_dir   # backwards-compatible alias


def prune_logs(keep=20):
    try:
        d = _log_dir()
        files = sorted((os.path.join(d, f) for f in os.listdir(d)
                        if f.endswith(".log")), key=os.path.getmtime)
        for f in files[:-keep]:
            try:
                os.remove(f)
            except OSError:
                pass
    except Exception:
        pass


class FfmpegProcess:
    """A single running ffmpeg, with its stderr safely parked in a file."""

    def __init__(self, cmd: list[str]):
        self.cmd = cmd
        self.proc: subprocess.Popen | None = None
        self.log_path = os.path.join(
            _log_dir(),
            f"ffmpeg_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}.log")
        self._log_file = None
        self.started_at = 0.0

    def start(self):
        """Launch ffmpeg. Raises RuntimeError with a readable message."""
        try:
            self._log_file = open(self.log_path, "wb")
            self._log_file.write((" ".join(self.cmd) + "\n\n").encode(
                "utf-8", "replace"))
            self._log_file.flush()
        except OSError:
            self._log_file = None

        try:
            self.proc = subprocess.Popen(
                self.cmd,
                stdin=subprocess.PIPE,
                stdout=subprocess.DEVNULL,
                # stderr -> file, never a pipe: an unread pipe deadlocks
                # ffmpeg once its buffer fills.
                stderr=self._log_file or subprocess.DEVNULL,
                creationflags=NO_WINDOW)
        except FileNotFoundError as exc:
            self._close_log()
            raise RuntimeError(FFMPEG_HELP) from exc
        except OSError as exc:
            self._close_log()
            raise RuntimeError(f"Could not start ffmpeg: {exc}") from exc

        self.started_at = time.monotonic()
        _assign_kill_on_close_job(self.proc)

    def _close_log(self):
        if self._log_file:
            try:
                self._log_file.close()
            except OSError:
                pass
            self._log_file = None

    # -- state ------------------------------------------------------
    def poll(self):
        return self.proc.poll() if self.proc else None

    @property
    def running(self) -> bool:
        return self.proc is not None and self.proc.poll() is None

    @property
    def uptime(self) -> float:
        return time.monotonic() - self.started_at if self.started_at else 0.0

    # -- stopping ---------------------------------------------------
    def stop(self, timeout=20.0) -> bool:
        """Ask ffmpeg to finish and wait for it. BLOCKS — call off the UI thread.

        Returns True when ffmpeg exited on its own (file properly finalised).
        """
        if not self.proc:
            return True
        p = self.proc
        clean = False
        try:
            if p.poll() is None:
                try:
                    if p.stdin:
                        p.stdin.write(b"q\n")
                        p.stdin.flush()
                except (OSError, ValueError):
                    pass
                try:
                    p.wait(timeout=timeout)
                    clean = True
                except subprocess.TimeoutExpired:
                    p.kill()
                    try:
                        p.wait(timeout=5)
                    except subprocess.TimeoutExpired:
                        pass
            else:
                clean = p.returncode == 0
        finally:
            for stream in (p.stdin,):
                try:
                    if stream:
                        stream.close()
                except OSError:
                    pass
            self._close_log()
        return clean

    def kill(self):
        if self.proc and self.proc.poll() is None:
            try:
                self.proc.kill()
                self.proc.wait(timeout=5)
            except Exception:
                pass
        self._close_log()

    # -- diagnostics ------------------------------------------------
    def log_tail(self, max_chars=4000) -> str:
        try:
            with open(self.log_path, "rb") as f:
                try:
                    f.seek(-max_chars, os.SEEK_END)
                except OSError:
                    f.seek(0)
                return f.read().decode("utf-8", "replace")
        except OSError:
            return ""

    def error_summary(self, max_lines=3) -> str:
        """A short, human-usable reason for a failed start."""
        lines = [ln.strip() for ln in self.log_tail().splitlines() if ln.strip()]
        lines = [ln for ln in lines if not ln.lower().startswith(("ffmpeg version",
                                                                 "configuration:",
                                                                 "built with"))]
        # skip the echoed command line
        lines = [ln for ln in lines if not ln.startswith(self.cmd[0])]
        interesting = [ln for ln in lines
                       if any(k in ln.lower() for k in
                              ("error", "invalid", "could not", "cannot",
                               "no such", "denied", "unknown", "failed",
                               "not supported", "unable"))]
        chosen = (interesting or lines)[-max_lines:]
        return "\n".join(chosen)


def short_reason(text: str) -> str:
    """Squeeze an ffmpeg error blob into something that fits a tiny label."""
    low = (text or "").lower()
    table = [
        ("could not find video device", "no screen"),
        ("could not run filter", "filter error"),
        ("i/o error", "disk error"),
        ("permission denied", "no access"),
        ("no such file or directory", "bad folder"),
        ("could not open file", "bad folder"),
        ("unknown encoder", "no encoder"),
        ("cannot load", "gpu error"),
        ("no device available", "gpu error"),
        ("device creation failed", "gpu error"),
        ("openencodesessionex failed", "gpu busy"),
        ("not divisible by 2", "odd size"),
        ("invalid argument", "bad option"),
        ("could not find audio only device", "no audio dev"),
        ("i/o error occurred", "disk error"),
    ]
    for needle, label in table:
        if needle in low:
            return label
    return "ffmpeg error"


def is_encoder_failure(text: str) -> bool:
    """Does this error look like the hardware encoder is to blame?"""
    low = (text or "").lower()
    return any(k in low for k in (
        "unknown encoder", "cannot load", "no device available",
        "device creation failed", "openencodesession", "nvenc", "qsv",
        "amf", "d3d11", "hwaccel", "impossible to convert",
        "encoder not found", "no capable devices"))


# ═══════════════════════════════════════════════════════════════════
# A recording session — one or more parts (pause/resume) + final merge
# ═══════════════════════════════════════════════════════════════════
class RecordingSession:
    """Owns the output naming, the part files and the final concat."""

    def __init__(self, ffmpeg: str, out_path: str, options: dict):
        self.ffmpeg = ffmpeg
        self.final_path = out_path
        self.options = options
        self.parts: list[str] = []
        self.current: FfmpegProcess | None = None
        self.encoder = options.get("encoder", "libx264")

    # -- naming -----------------------------------------------------
    def _next_part_path(self) -> str:
        base, ext = os.path.splitext(self.final_path)
        return f"{base}.part{len(self.parts) + 1}{ext}"

    # -- running ----------------------------------------------------
    def start_part(self) -> FfmpegProcess:
        path = self._next_part_path()
        cmd = build_command(self.ffmpeg, out_path=path,
                            encoder=self.encoder, **self.options_for_build())
        proc = FfmpegProcess(cmd)
        proc.start()
        self.current = proc
        self.parts.append(path)
        return proc

    def options_for_build(self) -> dict:
        o = dict(self.options)
        o.pop("encoder", None)
        return o

    def retry_with_software(self) -> FfmpegProcess:
        """Re-launch the failed part using libx264."""
        if self.parts:
            failed = self.parts.pop()
            _quiet_remove(failed)
        self.encoder = "libx264"
        return self.start_part()

    def stop_part(self, timeout=20.0) -> bool:
        if not self.current:
            return True
        ok = self.current.stop(timeout=timeout)
        self.current = None
        return ok

    # -- finishing --------------------------------------------------
    def finalize(self) -> tuple[str | None, str | None]:
        """Merge parts into final_path. BLOCKS — run it off the UI thread.

        Returns (path, error). On a merge failure the parts are kept and
        their location is reported rather than silently thrown away.
        """
        real = [p for p in self.parts
                if os.path.exists(p) and os.path.getsize(p) > 0]
        for p in self.parts:
            if p not in real:
                _quiet_remove(p)

        if not real:
            return None, "ffmpeg produced no output"

        if len(real) == 1:
            try:
                if os.path.exists(self.final_path):
                    _quiet_remove(self.final_path)
                os.replace(real[0], self.final_path)
                return self.final_path, None
            except OSError:
                return real[0], None  # keep the part; it is a valid file

        list_path = os.path.join(app_data_dir(), "concat_list.txt")
        try:
            with open(list_path, "w", encoding="utf-8") as f:
                for p in real:
                    f.write("file '%s'\n" % p.replace("\\", "/").replace("'", "'\\''"))
            cmd = [self.ffmpeg, "-hide_banner", "-nostats", "-loglevel", "error",
                   "-y", "-f", "concat", "-safe", "0", "-i", list_path,
                   "-c", "copy", self.final_path]
            r = subprocess.run(cmd, capture_output=True, text=True,
                               errors="replace", timeout=1800,
                               creationflags=NO_WINDOW)
            if r.returncode == 0 and os.path.exists(self.final_path):
                for p in real:
                    _quiet_remove(p)
                return self.final_path, None
            return (os.path.dirname(real[0]),
                    "Could not merge the paused segments — the individual "
                    "parts were kept:\n" + "\n".join(os.path.basename(p) for p in real))
        except Exception as exc:
            return (os.path.dirname(real[0]),
                    f"Could not merge segments ({exc}). Parts kept:\n"
                    + "\n".join(os.path.basename(p) for p in real))
        finally:
            _quiet_remove(list_path)

    def abort(self):
        if self.current:
            self.current.kill()
            self.current = None


def _quiet_remove(path: str):
    try:
        os.remove(path)
    except OSError:
        pass