#!/usr/bin/env python3 """ lektor_ai_comfy — Convert SRT subtitles to AI voiceover via ComfyUI + Higgs Audio V3. Usage: python3 lektor_ai_comfy.py input/film.srt input/film.mkv --start 00:00 --end 01:40 Requires ComfyUI with TTS-Audio-Suite running at http://127.0.0.1:8189 """ import argparse import configparser import json import os import queue import random import re import shlex import signal import shutil import subprocess import sys import threading import time import uuid from pathlib import Path from urllib.error import URLError from urllib.request import urlopen, Request # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- COMFY_OUTPUT_DIR = Path("/home/lukas/comfy-tts-app/ComfyUI/output") COMFY_INPUT_DIR = Path("/home/lukas/comfy-tts-app/ComfyUI/input") VOICE_PROMPTS_DIR = Path(__file__).resolve().parent / "voices" SETTINGS_FILE = Path(__file__).resolve().parent / "settings.ini" def _as_bool(raw) -> bool: """yes/no setting. configparser's own getboolean is bypassed by the per-key cast in load_settings, so accept the same spellings here.""" value = str(raw).strip().lower() if value in ("1", "true", "yes", "on"): return True if value in ("0", "false", "no", "off"): return False raise ValueError(f"{raw!r} is not a yes/no value") # Fallback values, used when settings.ini is absent or omits a key. Keeping them # here means the script still runs correctly with no settings file at all. # Each entry is (ini_section, ini_key, type, value). _DEFAULTS = { "voice": ("tts", "voice", str, "gudowski_short"), "speed": ("tts", "speed", float, 1.2), "temperature": ("tts", "temperature", float, 0.9), "max_tokens": ("tts", "max_tokens", int, 4096), "truncation_retries": ("tts", "truncation_retries", int, 2), "strip_leading_ms": ("tts", "strip_leading_ms", float, 0.0), "trailing_fade_ms": ("tts", "trailing_fade_ms", float, 80.0), "orig_volume": ("mix", "orig_volume", float, 0.60), "tts_volume": ("mix", "tts_volume", float, 1.8), "output_dir": ("paths", "output_dir", str, "output"), "dictionary": ("paths", "dictionary", str, "dictionary.txt"), "normalize_numbers": ("normalize", "numbers", _as_bool, True), "normalizer_python": ("normalize", "python", str, "normalizer/venv/bin/python"), "normalizer_timeout": ("normalize", "timeout", float, 60.0), "normalizer_llm": ("normalize", "llm", str, "normalizer/models/PLLuM-4B-instruct-2512"), "free_comfy_vram": ("normalize", "free_comfy_vram", _as_bool, True), "comfy_url": ("comfy", "comfy_url", str, "http://127.0.0.1:8189"), "comfy_autostart": ("comfy", "autostart", _as_bool, True), "comfy_start_command": ("comfy", "start_command", str, "start_comfyui.sh"), } NORMALIZER_DIR = Path(__file__).resolve().parent / "normalizer" COMFY_URL = _DEFAULTS["comfy_url"][3] # rebound from settings/CLI in main() # Rebound in main() too; used when the backend stops answering. COMFY_AUTOSTART = True COMFY_START_SCRIPT = Path(__file__).resolve().parent / "start_comfyui.sh" COMFY_LOG = Path(__file__).resolve().parent / "comfyui.log" # The backend log is appended to for ever otherwise — it reached 10 MB here in # a couple of days. Trimmed to the newest lines rather than rotated, so there is # one file to look at after a crash and no .1 .2 .3 to hunt through. COMFY_LOG_MAX_LINES = 120_000 _LOG_TRIM_MIN_BYTES = 2 << 20 # below this it cannot plausibly be over the cap def load_settings(path: Path) -> dict: """Read settings.ini, falling back to _DEFAULTS for anything missing. Values here become argparse's defaults, so precedence ends up as CLI flag > settings.ini > _DEFAULTS without any extra bookkeeping. A malformed or unreadable file is reported and then ignored rather than being fatal — a typo in a config file should not cost you a long run. """ values = {key: spec[3] for key, spec in _DEFAULTS.items()} if not path.is_file(): return values parser = configparser.ConfigParser(inline_comment_prefixes=("#", ";")) try: parser.read(path, encoding="utf-8") except (configparser.Error, OSError) as exc: print(f" WARNING: could not read {path.name} ({exc}) — using built-in defaults", file=sys.stderr) return values for key, (section, option, cast, fallback) in _DEFAULTS.items(): if not parser.has_option(section, option): continue raw = parser.get(section, option).strip() if not raw: continue try: values[key] = cast(raw) except ValueError: print(f" WARNING: {path.name} [{section}] {option}={raw!r} is not a valid " f"{cast.__name__} — using {fallback!r}", file=sys.stderr) return values # Subtitle codecs that carry real text and can therefore be extracted and fed # to the TTS engine. Used only by the --embedded-subs picker; image-based # formats (hdmv_pgs_subtitle, dvd_subtitle) hold bitmaps and would need OCR. # The final mux copies every subtitle track regardless of codec. _TEXT_SUB_CODECS = {"subrip", "srt", "ass", "ssa", "webvtt", "mov_text"} # --------------------------------------------------------------------------- # Replacement dictionary # --------------------------------------------------------------------------- class Dictionary: r"""Word and phrase replacements applied to subtitle text before TTS. The file holds one rule per line, `pattern = replacement`: km/h = kilometrów na godzinę /(\d+)\s*%/ = \1 procent A pattern wrapped in slashes is a regular expression (`\1` refers to its groups); anything else is matched literally. Matching is **case-sensitive** and only on whole words, so `np.` leaves `znp.` alone and `Joshua` leaves `joshua` alone. The replacement is inserted exactly as written. Every rule is compiled into a single alternation and applied in one pass, so replacement text is never rescanned and rules cannot chain into each other. Literal patterns are tried longest first, which is why `km/h` wins over a plain `km` rule no matter which order the two are written in; regular expressions are tried before literals, in file order. """ def __init__(self, rules: list, path: Path = None, missing: bool = False): # rules: list of (pattern, is_regex, replacement) self.rules = rules self.path = path self.missing = missing self.sources = [(path, len(rules))] if path else [] self.hits = 0 self._groups = [] # [(group_name, own_regex_or_None, replacement)] self._pattern = None self._compile() def __len__(self): return len(self.rules) # -- loading ------------------------------------------------------------ @classmethod def load(cls, path) -> "Dictionary": """Read a dictionary file. A missing file is not an error — it just yields an empty dictionary. Unparseable lines are reported and skipped, so one typo cannot cost you a long run.""" p = Path(path) if not p.is_file(): return cls([], path=p, missing=True) try: text = p.read_text(encoding="utf-8-sig") except OSError as exc: print(f" WARNING: could not read {p} ({exc}) — no replacements", file=sys.stderr) return cls([], path=p, missing=True) rules = [] for lineno, raw in enumerate(text.splitlines(), 1): line = raw.strip() if not line or line[0] in "#;": continue cut, width = cls._find_separator(line) if cut is None: print(f" WARNING: {p.name}:{lineno} has no '=' — skipped: {line!r}", file=sys.stderr) continue pattern = line[:cut].strip() replacement = line[cut + width:].strip() if not pattern: print(f" WARNING: {p.name}:{lineno} has an empty pattern — skipped", file=sys.stderr) continue is_regex = len(pattern) > 1 and pattern[0] == "/" and pattern[-1] == "/" if is_regex: pattern = pattern[1:-1] try: re.compile(pattern) except re.error as exc: print(f" WARNING: {p.name}:{lineno} is not a valid regex " f"({exc}) — skipped", file=sys.stderr) continue rules.append((pattern, is_regex, replacement)) return cls(rules, path=p) @classmethod def load_all(cls, paths) -> "Dictionary": """Merge several dictionary files into one rule set. Used for the queue dictionary: the shared `dictionary.txt` holds the rules that are true of Polish, and a per-queue file holds the ones true only of this batch — a character's name, a recurring mistranslation. Merging rather than chaining keeps the single-pass guarantee, so a rule from one file still cannot rewrite the output of a rule from the other. """ rules, sources = [], [] for path in paths: if not path: continue one = cls.load(path) if one.missing: print(f" Dictionary: {path} not found — skipped", file=sys.stderr) continue rules.extend(one.rules) sources.append((Path(path), len(one))) merged = cls(rules) merged.sources = sources return merged @staticmethod def _find_separator(line: str): """Locate the pattern/replacement separator: '=', '=>' or '->', whichever comes first. Returns (index, length) or (None, 0).""" found = [i for i in (line.find("="), line.find("->")) if i != -1] if not found: return None, 0 cut = min(found) return cut, 2 if line[cut:cut + 2] in ("=>", "->") else 1 # -- applying ----------------------------------------------------------- def _compile(self): regexes = [r for r in self.rules if r[1]] literals = sorted((r for r in self.rules if not r[1]), key=lambda r: len(r[0]), reverse=True) parts = [] for i, (pattern, is_regex, replacement) in enumerate(regexes + literals): name = f"_d{i}" if is_regex: source, own = pattern, re.compile(pattern) else: source, own = re.escape(pattern), None # Whole-word matching, but only on the edges that are word # characters: "km/h" needs the guards, "np." only the leading one. if pattern[0].isalnum() or pattern[0] == "_": source = r"(?{source})") self._groups.append((name, own, replacement)) if parts: # Case-sensitive: a rule matches the case you wrote it in, and the # replacement goes in verbatim. Matching loosely and then guessing # how to re-case the replacement is the kind of cleverness that # turns "NASA" into "Nasa" behind your back. self._pattern = re.compile("|".join(parts)) def apply(self, text: str) -> str: if not self._pattern or not text: return text before = self.hits result = self._pattern.sub(self._substitute, text) if self.hits != before: # A rule replacing a word with nothing would otherwise leave a # double space behind. result = re.sub(r"[ \t]{2,}", " ", result).strip() return result def _substitute(self, m) -> str: for name, own, replacement in self._groups: if m.group(name) is None: continue matched = m.group(0) if own is not None: inner = own.fullmatch(matched) if inner is not None: try: replacement = inner.expand(replacement) except (re.error, IndexError): pass # bad group reference — use the text as written self.hits += 1 return replacement return m.group(0) # unreachable, but never eat text def resolve_dictionary_path(value: str) -> Path: """Absolute paths are taken as given; a relative one is looked for in the current directory first and then next to this script, so a per-film dictionary can sit beside the film without hiding the shared one.""" p = Path(value).expanduser() if p.is_absolute() or p.is_file(): return p beside = Path(__file__).resolve().parent / p return beside if beside.is_file() else p # --------------------------------------------------------------------------- # Number normalizer (client side) # --------------------------------------------------------------------------- # # Higgs Audio mispronounces raw digits and garbles long ones, so digits are # spelled out as Polish words — with the right grammatical case — before the # text reaches the TTS. That is done by a small neural model that needs # transformers>=5, which must not be installed next to the pinned ComfyUI/Higgs # dependencies. So it lives in normalizer/ with its own venv and runs as a # separate process; this is the client half. See normalizer/README.md. # Mirrors normalizer.KEEP_RE. A standalone one-digit number is read correctly by # the TTS and is left alone; anything else with a digit is worth sending. Numeric # punctuation only continues a number when a digit follows it, so "85," ends at # the comma while "3,5" does not. The worker applies the same rule # authoritatively — this copy only keeps blocks that need nothing from crossing # the process boundary at all. _KEEP_DIGIT_RE = re.compile(r'(? bool: """True when the block holds a number the TTS would get wrong.""" return bool(_ANY_DIGIT_RE.search(_KEEP_DIGIT_RE.sub("", text))) def _restore_trailing_punct(original: str, result: str) -> str: """_clean_text() adds terminal punctuation deliberately, to give the model a falling tail. Generation occasionally drops it; put it back.""" m = _TRAILING_PUNCT_RE.search(original) if m and not _TRAILING_PUNCT_RE.search(result): return result + m.group(0) return result class NumberNormalizer: """Talks to normalizer/worker.py over JSON lines, one block at a time. The model costs far more to load than to run, so the worker is started once per run and kept alive. Nothing here is allowed to be fatal: if the venv is missing, the worker dies, or a block times out, the affected text is passed through unchanged and the run carries on — a number read badly is a worse outcome than the alternative only if the alternative isn't losing the film. """ STARTUP_TIMEOUT = 300.0 # first load pulls 1.1 GB off disk onto the GPU def __init__(self, python_path: Path, timeout: float = 60.0, llm_dir: Path = None): self.python = Path(python_path) self.timeout = timeout self.llm_dir = llm_dir self.script = NORMALIZER_DIR / "worker.py" self.proc = None self.mode = None # "model" | "fallback" self.device = "" self.llm = False # inflection leg available self.calls = 0 self.changed = 0 self.legs = {} # leg -> count self.guards = {} # guard reason -> count self._out = queue.Queue() self._err = [] self._next_id = 0 # -- lifecycle ---------------------------------------------------------- def start(self) -> bool: if not self.python.is_file(): print(f" WARNING: normalizer interpreter not found at {self.python} — " f"numbers stay as digits. Run normalizer/setup.sh to install it.", file=sys.stderr) return False if not self.script.is_file(): print(f" WARNING: {self.script} is missing — numbers stay as digits.", file=sys.stderr) return False env = dict(os.environ, HF_HUB_OFFLINE="1", TRANSFORMERS_OFFLINE="1", PYTHONUNBUFFERED="1") if self.llm_dir: env["PL_NORMALIZER_LLM"] = str(self.llm_dir) try: self.proc = subprocess.Popen( [str(self.python), str(self.script)], cwd=str(NORMALIZER_DIR), env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", bufsize=1) except OSError as exc: print(f" WARNING: could not start the normalizer ({exc}) — " f"numbers stay as digits.", file=sys.stderr) return False threading.Thread(target=self._pump_stdout, daemon=True).start() threading.Thread(target=self._pump_stderr, daemon=True).start() ready = self._read(self.STARTUP_TIMEOUT) if not ready or not ready.get("ready"): print(f" WARNING: the normalizer did not start — numbers stay as " f"digits.{self._stderr_tail()}", file=sys.stderr) self.close() return False self.mode = ready.get("mode", "fallback") self.device = ready.get("device") or "" self.llm = bool(ready.get("llm")) return True def close(self, grace: float = 10.0): """Closing stdin is the worker's exit signal; `grace` is how long a wedged one is given before it is killed.""" if not self.proc: return try: if self.proc.stdin and not self.proc.stdin.closed: self.proc.stdin.close() self.proc.wait(timeout=grace) except (OSError, subprocess.TimeoutExpired): self.proc.kill() finally: self.proc = None def __enter__(self): return self def __exit__(self, *exc): self.close() return False # -- plumbing ----------------------------------------------------------- def _pump_stdout(self): for line in self.proc.stdout: self._out.put(line) self._out.put(None) # EOF def _pump_stderr(self): for line in self.proc.stderr: self._err.append(line.rstrip()) del self._err[:-40] # keep the tail, not the flood def _stderr_tail(self, n: int = 6) -> str: tail = [l for l in self._err[-n:] if l.strip()] return "\n " + "\n ".join(tail) if tail else "" def _read(self, timeout: float): try: line = self._out.get(timeout=timeout) except queue.Empty: return None if line is None: return None try: return json.loads(line) except ValueError: return None # -- use ---------------------------------------------------------------- def normalize(self, text: str) -> str: """Return `text` with its numbers spelled out. Returns the input unchanged — and disables itself — on any failure.""" if not self.proc: return text self._next_id += 1 try: self.proc.stdin.write( json.dumps({"id": self._next_id, "text": text}, ensure_ascii=False) + "\n") self.proc.stdin.flush() except (OSError, ValueError): self._give_up("the normalizer stopped accepting input") return text reply = self._read(self.timeout) if reply is None: self._give_up(f"no answer within {self.timeout:g}s") return text if reply.get("id") != self._next_id: self._give_up("the normalizer replied out of order") return text if reply.get("error"): print(f"\n WARNING: normalizer: {reply['error']} — block left as-is", file=sys.stderr) return text result = (reply.get("text") or "").strip() if not result: return text result = _restore_trailing_punct(text, result) self.calls += 1 if result != text: self.changed += 1 leg = reply.get("leg") or "?" self.legs[leg] = self.legs.get(leg, 0) + 1 if reply.get("guard"): key = reply["guard"].split(" (")[0] self.guards[key] = self.guards.get(key, 0) + 1 # Name the first few, because a downgraded block is the one worth # reading in tts_input.srt. The tally at the end covers the rest. if sum(self.guards.values()) <= 3: print(f"\n {reply['guard']}: {text[:70]}", flush=True) return result def _give_up(self, why: str): print(f"\n WARNING: {why} — the rest of the numbers stay as digits." f"{self._stderr_tail()}", file=sys.stderr) self.close(grace=2.0) # it is already wedged; don't wait on it def free_comfy_vram(timeout: float = 30.0) -> bool: """Ask ComfyUI to unload its models, so the normalizer can have the GPU. Higgs stays resident between runs — around 8.4 GB of a 16 GB card — and a normalizer that cannot fit does not fail loudly: accelerate answers a shortage by spilling layers to RAM, which measured 5x slower here with no error printed. So the models are unloaded first and ComfyUI reloads them on the first clip of step 2, about 20-30 s, once per run. Only full runs reach this. --regen and --rebuild work from tts_input.srt, which is already normalized, so they never pay the reload. The unload is asynchronous — /free sets a queue flag that ComfyUI's worker picks up on its next loop — so this waits for the memory to actually come back instead of assuming it did. Returns False if it could not be confirmed; the caller carries on regardless, because a slow normalizer beats no run. """ QUIET = 1 << 30 # torch holding under 1 GiB means the models are out def torch_vram() -> float: """Bytes ComfyUI's torch has reserved, or None if it cannot be read.""" try: devices = _comfy_get("/system_stats").get("devices") or [] except (URLError, OSError, ValueError): return None cuda = [d for d in devices if d.get("type") == "cuda"] return max((d.get("torch_vram_total") or 0) for d in cuda) if cuda else 0 before = torch_vram() if before is None: return True # not running: nothing to free, and autostart handles it if before < QUIET: return True # nothing loaded yet — a freshly started backend print(f" Asking ComfyUI to unload its models ({before / 2**30:.1f} GB)…", flush=True) try: req = Request(f"{COMFY_URL}/free", method="POST", data=json.dumps({"unload_models": True, "free_memory": True}).encode(), headers={"Content-Type": "application/json"}) urlopen(req, timeout=30).read() # 200 with an empty body except (URLError, OSError) as exc: print(f" WARNING: could not ask ComfyUI to unload ({exc}).", file=sys.stderr) return False deadline = time.time() + timeout while time.time() < deadline: time.sleep(0.5) now = torch_vram() if now is not None and now < QUIET: print(f" Freed {(before - now) / 2**30:.1f} GB.") return True print(f" WARNING: ComfyUI still holds {torch_vram() / 2**30:.1f} GB after " f"{timeout:g}s — the normalizer may run partly on the CPU.", file=sys.stderr) return False def normalize_block_numbers(blocks: list, python_path: Path, timeout: float, free_vram: bool = True, llm_dir: Path = None) -> list: """Final stage of subtitle processing: rewrite digits as Polish words. Only blocks that actually contain a number worth expanding are sent, so a film with no numbers costs nothing but a process start — and not even that if nothing qualifies. """ targets = [i for i, (_s, _e, text) in enumerate(blocks) if needs_number_normalization(text)] if not targets: print(" No numbers to normalize.") return blocks print(f" {len(targets)} block(s) contain numbers — normalizing…") if free_vram: free_comfy_vram() normalizer = NumberNormalizer(python_path, timeout, llm_dir) if not normalizer.start(): return blocks where = (f"{normalizer.mode} on {normalizer.device}" if normalizer.device else normalizer.mode) legs = "two models" if normalizer.llm else "fine-tune only" print(f" Normalizer ready ({where}, {legs}).", flush=True) if not normalizer.llm: print(" NOTE: no inflection model — large numbers in oblique cases " "will keep the fine-tune's grammar. See normalizer/README.md.", file=sys.stderr) if normalizer.mode != "model": print(" WARNING: running on the num2words fallback — numbers will be " "spelled out in the nominative case only, and times, dates and " "decimals are left as digits. See normalizer/README.md.", file=sys.stderr) out = list(blocks) with normalizer: for n, i in enumerate(targets, 1): t_start, t_end, text = out[i] print(f"\r [{n}/{len(targets)}] normalizing…", end="", flush=True) out[i] = (t_start, t_end, normalizer.normalize(text)) detail = ", ".join(f"{n} {leg}" for leg, n in sorted(normalizer.legs.items()) if leg != "skipped") print(f"\r {normalizer.changed} of {len(targets)} block(s) rewritten" f"{' (' + detail + ')' if detail else ''}.{' ' * 20}") if normalizer.guards: worst = ", ".join(f"{reason} x{n}" for reason, n in sorted(normalizer.guards.items(), key=lambda kv: -kv[1])) print(f" {sum(normalizer.guards.values())} block(s) failed a safety " f"check and fell back ({worst}).") return out # --------------------------------------------------------------------------- # SRT helpers # --------------------------------------------------------------------------- _SRT_TIME_RE = re.compile(r'(\d+):(\d+):(\d+)[,.](\d+)') def _parse_srt_time(s: str) -> float: m = _SRT_TIME_RE.match(s.strip()) if not m: raise ValueError(f"Bad SRT timestamp: {s!r}") h, mn, sec, ms = int(m[1]), int(m[2]), int(m[3]), int(m[4]) return h * 3600 + mn * 60 + sec + ms / 1000 def _fmt_srt_time(seconds: float) -> str: h = int(seconds // 3600) m = int((seconds % 3600) // 60) s = int(seconds % 60) ms = int(round((seconds % 1) * 1000)) if ms == 1000: ms = 999 return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" def _parse_mmss(value: str) -> float: parts = value.strip().split(":") if len(parts) == 2: return int(parts[0]) * 60 + float(parts[1]) if len(parts) == 3: return int(parts[0]) * 3600 + int(parts[1]) * 60 + float(parts[2]) raise ValueError(f"Cannot parse time: {value!r}") def _ensure_terminal_punct(text: str, single_dot: bool = False) -> str: if not text or text[-1] in ('.', ',', '!', '?', ';', ':'): return text return text + ('.' if single_dot else '...') def _strip_sub_tags(line: str) -> str: """Remove SRT/ASS formatting codes from a subtitle line.""" line = re.sub(r'\{[^}]*\}', '', line) # ASS overrides: {\an8}, {\pos(…)}, {\c&H…&} line = re.sub(r'<[^>]+>', '', line) # HTML tags: , , , return line _TOKEN_RE = re.compile(r'\S+') def _title_case(text: str) -> str: """Every word capitalised, the rest lowercased. ALL-CAPS captions are shouted by the TTS, so they have to be recased. The obvious choice — lowercase everything and capitalise the first letter — also lowercases the proper nouns inside them: "NASA – MARGO MADISON" became "Nasa – margo madison", and a lowercase name is both wrong to read and an invitation for the number normalizer to "correct" it (it produced "marga"). Capitalising every word keeps the names intact, and A/B clips showed no difference in how the TTS reads it. Hyphenated parts are capitalised too (Jean-Luc), but text after an apostrophe is not — "DON'T" gives "Don't", not "Don'T". """ def cap(word: str) -> str: return "-".join(_cap_first(part) for part in word.split("-")) def _cap_first(part: str) -> str: for i, ch in enumerate(part): if ch.isalpha(): return part[:i] + ch.upper() + part[i + 1:].lower() return part.lower() return _TOKEN_RE.sub(lambda m: cap(m.group(0)), text) def _clean_text(lines: list, dictionary: Dictionary = None) -> str: # Strip formatting tags first so they don't interfere with allcaps detection # or the leading-dash check. lines = [_strip_sub_tags(l) for l in lines] # Only split into multi-part (for dialogue splitting later) when lines # have explicit speaker dashes. Wrapped single-speaker lines join with space. has_dialogue = any(re.match(r"^-\s*\S", l.strip()) for l in lines if l.strip()) out = [] for line in lines: line = line.strip() line = re.sub(r"^-\s*", "", line) if line: out.append(line) result = "\n".join(out) if has_dialogue else " ".join(out) # If the whole block is ALL-CAPS, recase it and end with a single dot. letters = [c for c in result if c.isalpha()] was_allcaps = bool(letters) and all(c == c.upper() for c in letters) if was_allcaps: result = _title_case(result) # Sentence fragments that continue from a previous subtitle start with a # lowercase letter and confuse TTS models — capitalise the first character. if result and result[0].islower(): result = result[0].upper() + result[1:] # Dictionary replacements run after case normalisation, so an ALL-CAPS block # is still recognised as one, and before the terminal punctuation is added, # so a rule may end a phrase with punctuation of its own. if dictionary is not None: result = "\n".join(dictionary.apply(p) for p in result.split("\n")) # Ensure each sentence ends with punctuation so TTS produces a natural tail. if has_dialogue: result = "\n".join(_ensure_terminal_punct(p, was_allcaps) for p in result.split("\n")) else: result = _ensure_terminal_punct(result, was_allcaps) return result def load_tts_input_srt(path: Path) -> list: """Read a pre-processed tts_input.srt into (t_start, t_end, text) tuples. Unlike parse_subtitle_blocks this does NOT re-run _clean_text or dialogue splitting — the file is already in final form. """ text = path.read_text(encoding="utf-8") blocks = re.split(r"\n{2,}", text.strip()) result = [] for block in blocks: lines = block.strip().splitlines() ti = next((i for i, l in enumerate(lines) if "-->" in l), None) if ti is None: continue parts = lines[ti].split("-->") if len(parts) != 2: continue t_start = _parse_srt_time(parts[0]) t_end = _parse_srt_time(parts[1]) payload = " ".join(l.strip() for l in lines[ti + 1:] if l.strip()) if payload: result.append((t_start, t_end, payload)) return result def parse_subtitle_blocks(srt_path: str, start_sec: float, end_sec: float, dictionary: Dictionary = None) -> list: """Return list of (start_relative, end_relative, plain_text) blocks.""" text = Path(srt_path).read_text(encoding="utf-8-sig") blocks = re.split(r"\n{2,}", text.strip()) result = [] for block in blocks: lines = block.strip().splitlines() ti = next((i for i, l in enumerate(lines) if "-->" in l), None) if ti is None: continue parts = lines[ti].split("-->") if len(parts) != 2: continue t_start = _parse_srt_time(parts[0]) t_end = _parse_srt_time(parts[1]) if t_start < start_sec or t_start > end_sec: continue plain = _clean_text(lines[ti + 1:], dictionary) if plain: result.append((t_start - start_sec, t_end - start_sec, plain)) return result def split_dialogue_blocks(blocks: list) -> list: """Split multi-line blocks into one entry per line with proportional timing.""" result = [] for t_start, t_end, text in blocks: lines = [l for l in text.split("\n") if l.strip()] if len(lines) <= 1: result.append((t_start, t_end, text)) continue duration = t_end - t_start total_chars = sum(len(l) for l in lines) current = t_start for i, line in enumerate(lines): share = len(line) / total_chars if total_chars else 1 / len(lines) end = t_end if i == len(lines) - 1 else current + duration * share result.append((current, end, line)) current = end return result def save_voiceover_subs(srt_path: str, clips_dir: Path, start_sec: float, end_sec: float) -> Path: """Keep a copy of the subtitles this run speaks, timed to the output video. Written for every run, so a later --rebuild --include-subs still has them when the original .srt is no longer on the command line. Cue text is copied verbatim: this is the track a viewer reads, not the text handed to the model, so dictionary replacements and spelled-out numbers stay out of it. Timestamps are shifted by --start, because the video they are muxed into begins there, and cues outside --start..--end are dropped. """ out = clips_dir / "voiceover_subs.srt" text = Path(srt_path).read_text(encoding="utf-8-sig") kept = [] for block in re.split(r"\n{2,}", text.strip()): lines = block.strip().splitlines() ti = next((i for i, l in enumerate(lines) if "-->" in l), None) if ti is None: continue parts = lines[ti].split("-->") if len(parts) != 2: continue t_start = _parse_srt_time(parts[0]) t_end = _parse_srt_time(parts[1]) if t_start < start_sec or t_start > end_sec: continue body = [l for l in lines[ti + 1:] if l.strip()] if not body: continue kept.append(f"{len(kept) + 1}\n" f"{_fmt_srt_time(t_start - start_sec)} --> " f"{_fmt_srt_time(max(0.0, t_end - start_sec))}\n" + "\n".join(body) + "\n") out.write_text("\n".join(kept), encoding="utf-8") return out def voiceover_subs_file(job: Path): """The subtitle file --include-subs should mux, or None if there is none. Prefers the verbatim copy kept by the run itself. Runs made before that copy existed still have tts_input.srt, which is the same timing but the model's wording — better than no track at all, so it is used with a note. """ saved = job / "voiceover_subs.srt" if saved.is_file() and saved.stat().st_size: return saved fallback = job / "tts_input.srt" if fallback.is_file() and fallback.stat().st_size: print(f" --include-subs: no {saved.name} from the original run — using " f"{fallback.name} instead (the model's text, not the original wording).") return fallback print(" --include-subs: no subtitles found in the job folder — skipping " "the extra track.", file=sys.stderr) return None def blocks_to_srt(blocks: list) -> str: """Convert (start, end, text) tuples back to SRT string for inspection.""" lines = [] for i, (t_start, t_end, text) in enumerate(blocks, 1): lines.append(str(i)) lines.append(f"{_fmt_srt_time(t_start)} --> {_fmt_srt_time(t_end)}") lines.append(text) lines.append("") return "\n".join(lines) # --------------------------------------------------------------------------- # FFmpeg helpers # --------------------------------------------------------------------------- def _ffmpeg_ts(seconds: float) -> str: h = int(seconds // 3600) m = int((seconds % 3600) // 60) s = seconds % 60 return f"{h:02d}:{m:02d}:{s:06.3f}" def _run(cmd: list, label: str): print(f" {label}...") result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: print(f"ffmpeg error ({label}):\n{result.stderr[-2000:]}", file=sys.stderr) sys.exit(1) def _run_progress(cmd: list, label: str, total_seconds: float): """Run ffmpeg with a live progress bar that rewrites the current line.""" # Inject -progress pipe:1 before the output path (last arg) prog_cmd = cmd[:-1] + ["-progress", "pipe:1", "-nostats", cmd[-1]] proc = subprocess.Popen( prog_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, ) stderr_chunks: list = [] def _drain(): for line in proc.stderr: stderr_chunks.append(line) threading.Thread(target=_drain, daemon=True).start() BAR = 28 print(f" {label}...", end="", flush=True) for line in proc.stdout: if not line.startswith("out_time_us="): continue try: us = int(line.split("=", 1)[1]) if us < 0: continue sec = us / 1_000_000 except (ValueError, IndexError): continue pct = min(sec / total_seconds, 1.0) if total_seconds > 0 else 0.0 filled = int(BAR * pct) bar = "█" * filled + "░" * (BAR - filled) t_cur = f"{int(sec // 60):02d}:{int(sec % 60):02d}" t_tot = f"{int(total_seconds // 60):02d}:{int(total_seconds % 60):02d}" print(f"\r {label}... [{bar}] {t_cur}/{t_tot}", end="", flush=True) proc.wait() if proc.returncode != 0: err = "".join(stderr_chunks) print(f"\nffmpeg error ({label}):\n{err[-2000:]}", file=sys.stderr) sys.exit(1) t_tot = f"{int(total_seconds // 60):02d}:{int(total_seconds % 60):02d}" print(f"\r {label}... [{'█' * BAR}] {t_tot}/{t_tot}") def get_audio_duration(path: Path) -> float: result = subprocess.run( ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", str(path)], capture_output=True, text=True, ) info = json.loads(result.stdout) for stream in info.get("streams", []): if "duration" in stream: return float(stream["duration"]) raise RuntimeError(f"Cannot determine duration of {path}") def speed_up_audio(input_path: Path, output_path: Path, speed: float, fade_out_ms: float = 0): """Apply atempo (and optional fade-out) to a single clip.""" filters = [] if speed != 1.0: remaining = speed while remaining > 2.0: filters.append("atempo=2.0") remaining /= 2.0 filters.append(f"atempo={remaining:.4f}") if fade_out_ms > 0: fade_sec = fade_out_ms / 1000 # Compute fade start from the post-speed duration so we can use a plain # afade=t=out (areverse caused MP3 tail truncation on VBR files). raw_dur = get_audio_duration(input_path) post_dur = raw_dur / speed if speed != 1.0 else raw_dur fade_start = max(0.0, post_dur - fade_sec) filters.append(f"afade=t=out:st={fade_start:.3f}:d={fade_sec:.3f}") if not filters: shutil.copy2(input_path, output_path) return label = f"Speed {speed:.2f}×" if fade_out_ms: label += f" + fade-out {fade_out_ms:.0f}ms" _run([ "ffmpeg", "-y", "-i", str(input_path), "-filter:a", ",".join(filters), str(output_path), ], label) def trim_audio(input_path: Path, output_path: Path, max_duration: float): """Hard-trim a clip to max_duration seconds.""" _run([ "ffmpeg", "-y", "-i", str(input_path), "-t", f"{max_duration:.3f}", "-c", "copy", str(output_path), ], f"Trim to {max_duration:.2f}s") def strip_leading_audio(input_path: Path, output_path: Path, strip_ms: float): """Remove the first strip_ms milliseconds from a clip.""" _run([ "ffmpeg", "-y", "-i", str(input_path), "-ss", f"{strip_ms / 1000:.3f}", "-c", "copy", str(output_path), ], f"Strip {strip_ms:.0f}ms leading audio") # A clip's trailing silence must not count towards its duration: placement uses # a push-forward cursor, so padding at the end of one clip delays every clip # after it. Usually that padding is a fraction of a second, but when the model # fails to emit a stop token it generates until --max-tokens runs out — a short # line then yields a clip of speech followed by two minutes of digital silence, # which shunts the rest of the film out of sync. SILENCE_DB = -45.0 # anything quieter counts as silence SILENCE_MIN = 0.35 # ignore shorter gaps; commas and beats are not padding SILENCE_KEEP_MS = 120 # leave this much tail so speech does not end abruptly def trailing_silence_start(path: Path) -> float: """Time at which the clip's final run of silence begins, or None.""" result = subprocess.run( ["ffmpeg", "-hide_banner", "-i", str(path), "-af", f"silencedetect=noise={SILENCE_DB}dB:d={SILENCE_MIN}", "-f", "null", "-"], capture_output=True, text=True, ) try: duration = get_audio_duration(path) except RuntimeError: return None open_at = None for kind, value in re.findall(r"silence_(start|end):\s*([0-9.]+)", result.stderr): t = float(value) if kind == "start": open_at = t elif t < duration - 0.05: open_at = None # that silence ended mid-clip, so it is a pause return open_at # Some runaway generations pad with a steady tone instead of digital silence. # It sits above any sane silence threshold, so silencedetect cannot see it — # but real speech swings tens of dB between windows while a tone does not. # Measured here: tone tails spread 3-5 dB, speech tails 43-61 dB. FLAT_TAIL_DB = 15.0 # spread below this is not speech FLAT_TAIL_MIN = 5.0 # the flat run must last at least this long RUNAWAY_SECONDS = 20.0 # only look for tone padding on clips this long # A clip the model finished decays to near silence; one it cut off mid-word # stops at full level. Measured across 80 raw outputs the two groups sit about # 50 dB apart (-65 dB vs -16 dB), so the test is unambiguous. TRUNCATION_PEAK_DB = -25.0 TRUNCATION_WINDOW = 0.08 def ends_abruptly(path: Path, window: float = TRUNCATION_WINDOW) -> bool: """True when a clip stops at full level, i.e. the model cut it mid-word.""" try: duration = get_audio_duration(path) except RuntimeError: return False if duration <= window: return False result = subprocess.run( ["ffmpeg", "-hide_banner", "-ss", f"{duration - window:.3f}", "-t", f"{window}", "-i", str(path), "-af", "volumedetect", "-f", "null", "-"], capture_output=True, text=True, ) peak = re.search(r"max_volume:\s*(-?[0-9.]+)", result.stderr) return bool(peak) and float(peak.group(1)) > TRUNCATION_PEAK_DB def flat_tail_start(path: Path, window: float = 0.5) -> float: """Start of a long, unnaturally constant tail, or None.""" result = subprocess.run( ["ffmpeg", "-hide_banner", "-i", str(path), "-af", f"astats=metadata=1:reset={window}," "ametadata=print:key=lavfi.astats.Overall.RMS_level", "-f", "null", "-"], capture_output=True, text=True, ) blob = result.stderr + result.stdout values = [(-120.0 if "inf" in m else float(m)) for m in re.findall(r"RMS_level=(-?[0-9.]+|-?inf)", blob)] need = int(FLAT_TAIL_MIN / window) if len(values) < need + 1: return None for i in range(len(values) - need): tail = values[i:] if max(tail) - min(tail) <= FLAT_TAIL_DB: return i * window return None def trim_trailing_silence(input_path: Path, output_path: Path) -> float: """Cut the final silence off a clip. Returns the seconds removed.""" duration = get_audio_duration(input_path) start = trailing_silence_start(input_path) if start is None and duration > RUNAWAY_SECONDS: start = flat_tail_start(input_path) keep = None if start is None else start + SILENCE_KEEP_MS / 1000 if keep is None or keep >= duration - 0.02 or keep <= 0.05: if input_path != output_path: shutil.copy2(input_path, output_path) return 0.0 _run([ "ffmpeg", "-y", "-i", str(input_path), "-t", f"{keep:.3f}", str(output_path), ], f"Trim {duration - keep:.1f}s trailing silence") return duration - keep def _assemble_batch(clips: list, output_path: str, label: str, total_seconds: float = 0.0, start_offset: float = 0.0): """Mix a batch of (start_sec, path) clips with adelay into output_path. start_offset is subtracted from every clip's start time so sub-batch files contain no leading silence. The caller re-applies the offset at merge time. """ run = (lambda cmd, lbl: _run_progress(cmd, lbl, total_seconds)) if total_seconds > 0 else _run if len(clips) == 1: delay_ms = max(0, int((clips[0][0] - start_offset) * 1000)) run([ "ffmpeg", "-y", "-i", str(clips[0][1]), "-af", f"adelay={delay_ms}:all=1", output_path, ], label) return inputs, filter_parts = [], [] for i, (start_sec, path) in enumerate(clips): inputs += ["-i", str(path)] delay_ms = max(0, int((start_sec - start_offset) * 1000)) filter_parts.append(f"[{i}:a]adelay={delay_ms}:all=1[a{i}]") mix_inputs = "".join(f"[a{i}]" for i in range(len(clips))) filter_parts.append( f"{mix_inputs}amix=inputs={len(clips)}:duration=longest:normalize=0[out]" ) run( ["ffmpeg", "-y"] + inputs + ["-filter_complex", ";".join(filter_parts), "-map", "[out]", output_path], label, ) def assemble_timed_audio(clips: list, output_path: str, batch_size: int = 500): """Place each clip at its start time via adelay then mix. clips: list of (start_sec, Path). Processes in batches to stay under OS file-descriptor limits. """ n_batches = (len(clips) - 1) // batch_size + 1 if len(clips) <= batch_size: total_sec = clips[-1][0] if clips else 0.0 _assemble_batch(clips, output_path, f"Assembling {len(clips)} timed clips", total_seconds=total_sec) return tmp_dir = Path(output_path).parent / "_batch_tmp" tmp_dir.mkdir(exist_ok=True) batch_files = [] batch_offsets = [] try: for b, start in enumerate(range(0, len(clips), batch_size)): batch = clips[start:start + batch_size] batch_offset = batch[0][0] # first clip's global start time total_sec = batch[-1][0] - batch_offset # content duration only batch_path = str(tmp_dir / f"batch_{b:04d}.flac") _assemble_batch( batch, batch_path, f"Assembling batch {b + 1}/{n_batches} ({len(batch)} clips)", total_seconds=total_sec, start_offset=batch_offset, ) batch_files.append(batch_path) batch_offsets.append(batch_offset) # Merge: re-apply each batch's offset so clips land at their global times batch_clips = [(offset, Path(p)) for offset, p in zip(batch_offsets, batch_files)] _assemble_batch(batch_clips, output_path, f"Merging {len(batch_files)} batches") finally: for p in batch_files: Path(p).unlink(missing_ok=True) tmp_dir.rmdir() if tmp_dir.exists() and not any(tmp_dir.iterdir()) else None def extract_video_fragment(video_path: str, start_sec: float, end_sec: float, output_path: str): _run([ "ffmpeg", "-y", "-ss", _ffmpeg_ts(start_sec), "-i", video_path, "-t", str(end_sec - start_sec), "-c", "copy", output_path, ], f"Extracting video {_ffmpeg_ts(start_sec)}–{_ffmpeg_ts(end_sec)}") def list_subtitle_streams(video_path: str) -> list: """Return subtitle stream info dicts for all subtitle tracks in a video file.""" result = subprocess.run( ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", video_path], capture_output=True, text=True, ) subs = [] for s in json.loads(result.stdout).get("streams", []): if s.get("codec_type") != "subtitle": continue tags = s.get("tags", {}) subs.append({ "index": s["index"], "codec_name": s.get("codec_name", "unknown"), "language": tags.get("language", ""), "title": tags.get("title", ""), }) return subs def extract_subtitle_track(video_path: str, stream_index: int, output_path: str): """Extract one subtitle track from a video to an SRT file.""" _run([ "ffmpeg", "-y", "-i", video_path, "-map", f"0:{stream_index}", "-c:s", "subrip", output_path, ], f"Extracting subtitle stream #{stream_index}") def get_media_duration(path: str) -> float: """Container duration in seconds (0.0 if unknown).""" result = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", path], capture_output=True, text=True, ) try: return float(result.stdout.strip()) except ValueError: return 0.0 _FONT_TAG_RE = re.compile(r']*>', re.IGNORECASE) def extract_mov_text_srt(video_path: str, stream_index: int, out_path: Path) -> bool: """Decode one mov_text track to a SubRip file, without its styling. ffmpeg's tx3g decoder turns the format's style records into markup, so every cue comes out wrapped in . Players that honour those tags then render the subtitles at whatever pixel size the MP4 happened to specify — usually far too small on a 1080p screen — and the ones that don't honour them show the tags as literal text. Neither is what the source looked like, and the tags carry nothing worth keeping, so they are dropped and the plain cue text is muxed instead. Returns False if the track holds no cues, which is common for the empty placeholder track MP4 rips often carry. """ result = subprocess.run( ["ffmpeg", "-y", "-v", "error", "-i", video_path, "-map", f"0:{stream_index}", "-c:s", "srt", "-f", "srt", str(out_path)], capture_output=True, text=True, ) if result.returncode != 0 or not out_path.exists(): print(f" WARNING: could not extract subtitle stream #{stream_index}: " f"{result.stderr.strip()[-300:]}", file=sys.stderr) return False text = out_path.read_text(encoding="utf-8", errors="replace") cleaned = _FONT_TAG_RE.sub("", text) if not cleaned.strip(): return False out_path.write_text(cleaned, encoding="utf-8") return True def merge_voiceover(video_path: str, tts_path: str, output_path: str, orig_volume: float = 0.60, tts_volume: float = 1.8, extra_subs: Path = None): """Produce a video with two audio tracks: track 0 — Original (first audio stream from source, stream-copied) track 1 — Polish (lektor mix: TTS at tts_volume + original at orig_volume) plus every subtitle track from the source, stream-copied, and — when extra_subs is given — the subtitles that were spoken, as one more track. Done in two passes: 1. Render the lektor mix on its own to a temporary audio file. 2. Stream-copy everything (video + both audio tracks + all subtitles) into the final MKV. Combining both in a single ffmpeg call fails on sources with many subtitle tracks: the sparse subtitle streams stall the muxer while the audio filter graph keeps producing, and ffmpeg eventually aborts with "Failed to inject frame into filter network: Invalid argument". Splitting the work keeps the filter graph free of subtitle streams, and makes the final mux a pure copy (roughly 10x faster too). MKV rips often have audio PTS starting at a non-zero offset (e.g. 4.043s). Without correction amix can't output from PTS 0 until BOTH inputs have frames, which delays the TTS. asetpts resets the PTS to 0 and adelay re-inserts the offset as explicit silence, so the background audio stays at its correct position while amix gets frames from both inputs from PTS 0. """ probe = subprocess.run( ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", video_path], capture_output=True, text=True, ) streams = json.loads(probe.stdout).get("streams", []) audio_start = next( (float(s.get("start_time", 0)) for s in streams if s.get("codec_type") == "audio"), 0.0, ) if audio_start > 0.01: print(f" Audio PTS offset: {audio_start:.3f}s — normalising orig_mix for lektor track") delay_ms = int(audio_start * 1000) norm = f"asetpts=PTS-STARTPTS,adelay={delay_ms}:all=1,aresample=async=1" else: norm = "aresample=async=1" # The lektor track is stereo on purpose. ffmpeg's AAC encoder does not # signal a 5.1 channel layout — it comes back as "unknown" in every # container — so players cannot route the channels and the track plays # silent. Stereo is declared correctly and plays everywhere; 192k is also # a good rate for two channels where it is a poor one spread over six. # The 5.1 source is downmixed with ffmpeg's standard coefficients, and the # mono TTS becomes equal-level L/R, i.e. centred. stereo = "aformat=channel_layouts=stereo" filter_complex = ( f"[0:a:0]{norm},{stereo},volume={orig_volume}[orig_mix];" f"[1:a]{stereo},volume={tts_volume}[tts];" "[orig_mix][tts]amix=inputs=2:duration=first:normalize=0[lektor]" ) out = Path(output_path) mix_path = out.parent / f"{out.stem}_mix.m4a" sub_tmp: list = [] # extracted .srt files, removed after the mux try: # ── Pass 1: render the lektor mix on its own ───────────────────────── _run_progress([ "ffmpeg", "-y", "-i", video_path, "-i", tts_path, "-filter_complex", filter_complex, "-map", "[lektor]", "-c:a", "aac", "-b:a", "192k", str(mix_path), ], "Mixing lektor audio track", get_media_duration(video_path)) # ── Pass 2: stream-copy mux — video + both audio tracks + subtitles ── sub_streams = [s for s in streams if s.get("codec_type") == "subtitle"] if sub_streams: print(f" Including {len(sub_streams)} subtitle track(s) in output") # Subtitles cannot all be stream-copied: an MP4 source carries them as # mov_text (tx3g), which the matroska muxer rejects outright — # "Subtitle codec 94213 is not supported", and no output file is # written at all. Those tracks are decoded to SubRip in a temporary # file and muxed from there; everything else is still copied, which is # what the image-based formats (PGS, VOBSUB) require since they cannot # become text at all. cmd = [ "ffmpeg", "-y", "-i", video_path, "-i", str(mix_path), ] maps = ["-map", "0:v", "-map", "0:a:0", "-map", "1:a:0"] sub_codecs, sub_meta = [], [] next_input = 2 converted = 0 out_idx = 0 # position among the OUTPUT subtitle streams, # which a skipped track must not advance for s in sub_streams: if s.get("codec_name") == "mov_text": srt_path = out.parent / f"{out.stem}_sub{s['index']}.srt" if not extract_mov_text_srt(video_path, s["index"], srt_path): srt_path.unlink(missing_ok=True) print(f" Subtitle stream #{s['index']} has no cues — skipping it") continue sub_tmp.append(srt_path) cmd += ["-i", str(srt_path)] maps += ["-map", f"{next_input}:0"] # A standalone .srt carries no language, so the tag the source # track had has to be re-applied by hand or the output track # shows up as "undefined". lang = (s.get("tags") or {}).get("language") if lang: sub_meta += [f"-metadata:s:s:{out_idx}", f"language={lang}"] title = (s.get("tags") or {}).get("title") if title: sub_meta += [f"-metadata:s:s:{out_idx}", f"title={title}"] next_input += 1 converted += 1 else: maps += ["-map", f"0:{s['index']}"] sub_codecs += [f"-c:s:{out_idx}", "copy"] out_idx += 1 # The subtitles the voiceover was made from, added as one more track # so the file carries them even when the source had none of its own. if extra_subs is not None: print(f" Adding the voiceover subtitles as an extra track: " f"{Path(extra_subs).name}") cmd += ["-i", str(extra_subs)] maps += ["-map", f"{next_input}:0"] sub_codecs += [f"-c:s:{out_idx}", "copy"] sub_meta += [f"-metadata:s:s:{out_idx}", "language=pol", f"-metadata:s:s:{out_idx}", "title=Lektor"] next_input += 1 out_idx += 1 if converted: print(f" Converted {converted} mov_text subtitle track(s) to SubRip — " f"Matroska cannot store the MP4 subtitle format") cmd += maps + ["-c:v", "copy", "-c:a", "copy"] + sub_codecs + sub_meta cmd += [ # Must be an OUTPUT option (before the output path) — placed before # -i it silently configures the demuxer instead and does nothing. # Sparse subtitle tracks make the muxer give up on interleaving # after its default 10s window, which strands the lektor audio # hundreds of MB away from the video it belongs with. Players read # MKV sequentially, so they never reach it and the track is silent. # 0 means "buffer until every stream has a packet" — correct # interleaving, and only a couple of GB of transient buffering. "-max_interleave_delta", "0", "-metadata:s:a:0", "title=Original", "-metadata:s:a:1", "title=Polish", "-metadata:s:a:1", "language=pol", output_path, ] _run(cmd, "Muxing final video (2 audio tracks + subtitles)") finally: mix_path.unlink(missing_ok=True) for p in sub_tmp: p.unlink(missing_ok=True) # --------------------------------------------------------------------------- # ComfyUI API helpers # --------------------------------------------------------------------------- def _comfy_post(path: str, data: dict) -> dict: payload = json.dumps(data).encode() req = Request( f"{COMFY_URL}{path}", data=payload, headers={"Content-Type": "application/json"}, ) with urlopen(req, timeout=30) as r: return json.loads(r.read()) def _comfy_get(path: str) -> dict: with urlopen(f"{COMFY_URL}{path}", timeout=30) as r: return json.loads(r.read()) def upload_voice_wav(wav_path: Path) -> tuple: """Copy voice WAV into ComfyUI's input dir; return (filename, reference_text).""" dst = COMFY_INPUT_DIR / wav_path.name if not dst.exists() or dst.stat().st_mtime < wav_path.stat().st_mtime: shutil.copy2(wav_path, dst) print(f" Copied {wav_path.name} → {dst}") txt_path = wav_path.with_suffix(".txt") ref_text = txt_path.read_text(encoding="utf-8").strip() if txt_path.exists() else "" if ref_text: print(f" Reference text loaded ({len(ref_text)} chars)") else: print(" WARNING: no reference .txt found — voice cloning quality will be lower") return wav_path.name, ref_text def build_text_workflow(text: str, voice_filename: str, voice_ref_text: str, seed: int, temperature: float = 0.9, max_new_tokens: int = 4096) -> dict: """ComfyUI workflow for a single text utterance with proper voice reference.""" return { "1": { "class_type": "HiggsAudioV3EngineNode", "inputs": { "model": "higgs-audio-v3-tts-4b", "device": "auto", "dtype": "auto", "attention": "auto", "temperature": temperature, "top_p": 0.95, "top_k": 50, "max_new_tokens": max_new_tokens, }, }, "2": { "class_type": "LoadAudio", "inputs": {"audio": voice_filename}, }, "3": { # Pairs audio with its transcript so the model can clone the voice correctly "class_type": "CharacterVoicesNode", "inputs": { "voice_name": "none", "reference_text": voice_ref_text, "opt_audio_input": ["2", 0], }, }, "4": { "class_type": "UnifiedTTSTextNode", "inputs": { "TTS_engine": ["1", 0], "text": text, "narrator_voice": "none", "seed": seed, "opt_narrator": ["3", 0], "enable_audio_cache": False, }, }, "5": { "class_type": "SaveAudio", "inputs": { "audio": ["4", 0], "filename_prefix": "audio/lektor_clip", }, }, } def submit_workflow(workflow: dict) -> str: client_id = str(uuid.uuid4()) resp = _comfy_post("/prompt", {"prompt": workflow, "client_id": client_id}) return resp["prompt_id"] BACKEND_RETRY_SECONDS = 3.0 # how often to look for ComfyUI while it is down # Launching the backend is guarded: a ComfyUI that dies on startup would # otherwise spawn a new terminal window every few seconds. COMFY_AUTOSTART_MAX = 10 # restarts allowed while one clip is in flight COMFY_AUTOSTART_COOLDOWN = 90.0 # seconds between attempts # A clip that kills the backend this many times is skipped rather than retried # for ever. Some inputs abort ComfyUI outright — a one-word line like "Joshua." # tripped a CUDA device-side assert inside the audio tokenizer, which cannot be # caught in Python — and without a cap an overnight queue stops there and does # nothing until morning. CLIP_CRASH_LIMIT = 10 # How long a restarted backend gets to answer before we conclude it is not the # clip's fault but the backend's absence. BACKEND_HEALTH_TIMEOUT = 180.0 class BackendCrashed(Exception): """The backend died repeatedly on one clip. Raised so the caller can skip it and keep the queue moving.""" def _comfy_port() -> int: from urllib.parse import urlparse return urlparse(COMFY_URL).port or 8189 def _proc_state(pid: int) -> str: """One-letter process state from /proc, or "" if it is gone.""" try: stat = Path(f"/proc/{pid}/stat").read_text() except OSError: return "" return stat.rpartition(")")[2].split()[0] if ")" in stat else "" def comfy_pid() -> int: """PID of whatever is listening on ComfyUI's port, or None. Found through /proc rather than `ss` or `lsof`, so the script keeps its one real virtue of needing nothing installed. Discovering it by port rather than remembering what we spawned matters: the backend is usually started by hand in a terminal, and it has to be manageable either way. """ port, inodes = _comfy_port(), set() for table in ("/proc/net/tcp", "/proc/net/tcp6"): try: rows = Path(table).read_text().splitlines()[1:] except OSError: continue for row in rows: cols = row.split() if len(cols) > 9 and cols[3] == "0A": # 0A = TCP_LISTEN try: if int(cols[1].split(":")[1], 16) == port: inodes.add(cols[9]) except (ValueError, IndexError): pass if not inodes: return None for entry in Path("/proc").iterdir(): if not entry.name.isdigit(): continue try: for fd in (entry / "fd").iterdir(): try: link = os.readlink(fd) except OSError: continue if link.startswith("socket:[") and link[8:-1] in inodes: return int(entry.name) except OSError: continue # not ours, or gone return None def stop_comfy_backend(grace: float = 10.0) -> bool: """Stop the backend and wait for its port to be free. True when nothing is listening any more. SIGTERM first, then SIGKILL — except for a stopped process, which is killed outright: a process in state T never runs its signal handlers, so SIGTERM is accepted and ignored while it keeps holding the GPU. That is not a corner case; it is how ComfyUI ends up when its terminal suspends it, and `pkill` reports success while nothing happens. """ pid = comfy_pid() if pid is None: return True stopped = _proc_state(pid) in ("T", "t") print(f" Stopping ComfyUI (pid {pid}{', suspended' if stopped else ''})…", file=sys.stderr) try: os.kill(pid, signal.SIGKILL if stopped else signal.SIGTERM) except OSError as exc: print(f" Could not signal pid {pid}: {exc}", file=sys.stderr) return False deadline = time.monotonic() + grace while time.monotonic() < deadline: time.sleep(0.5) if comfy_pid() is None: break else: try: os.kill(pid, signal.SIGKILL) except OSError: pass time.sleep(1.0) # Reap it if it was ours, so it does not linger as a zombie holding the pid. proc = _autostart.get("proc") if proc is not None: try: proc.poll() except Exception: # noqa: BLE001 pass return comfy_pid() is None _autostart = {"attempts": 0, "last": 0.0, "proc": None} _TERMINALS = ( ("gnome-terminal", lambda c: ["gnome-terminal", "--title", "ComfyUI", "--", "bash", "-lc", c]), ("konsole", lambda c: ["konsole", "-p", "tabtitle=ComfyUI", "-e", "bash", "-lc", c]), ("xfce4-terminal", lambda c: ["xfce4-terminal", "--title", "ComfyUI", "-e", f"bash -lc {shlex.quote(c)}"]), ("xterm", lambda c: ["xterm", "-T", "ComfyUI", "-e", "bash", "-lc", c]), ) def start_comfy_backend(script: Path) -> bool: """Launch ComfyUI in its own terminal window. True if one was started.""" now = time.monotonic() if _autostart["attempts"] >= COMFY_AUTOSTART_MAX: return False if _autostart["last"] and now - _autostart["last"] < COMFY_AUTOSTART_COOLDOWN: return False if not script.is_file(): print(f" Cannot start ComfyUI automatically: {script} not found.", file=sys.stderr) _autostart["attempts"] = COMFY_AUTOSTART_MAX # do not keep complaining return False _autostart["attempts"] += 1 _autostart["last"] = now trim_comfy_log() # before the backend opens it again # Pause on exit so a backend that fails to start leaves its error on screen # instead of the window vanishing. log = COMFY_LOG inner = (f"{shlex.quote(str(script))} 2>&1 | tee -a {shlex.quote(str(log))}; " f"echo; echo '[ComfyUI exited — press any key to close]'; read -n 1") if os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY"): for name, build in _TERMINALS: if not shutil.which(name): continue try: _autostart["proc"] = subprocess.Popen(build(inner), start_new_session=True) except OSError: continue print(f" Starting ComfyUI in a new {name} window " f"(attempt {_autostart['attempts']} of {COMFY_AUTOSTART_MAX})…", file=sys.stderr) return True log = COMFY_LOG try: with open(log, "ab") as handle: _autostart["proc"] = subprocess.Popen( [str(script)], stdout=handle, stderr=handle, start_new_session=True) except OSError as exc: print(f" Could not start ComfyUI: {exc}", file=sys.stderr) return False print(f" No terminal available — started ComfyUI in the background, " f"logging to {log}", file=sys.stderr) return True def trim_comfy_log(path: Path = None, max_lines: int = COMFY_LOG_MAX_LINES) -> int: """Drop the oldest lines of the backend log, keeping the newest `max_lines`. Returns how many were dropped. Rewritten in place rather than replaced. `tee` holds the file open in append mode, so the inode must survive: writing a new file and renaming it over the old one would leave tee appending to an unlinked inode, and every later line would vanish silently. Rewriting in place works because O_APPEND directs each write to the current end of file rather than to a remembered offset. """ path = path or COMFY_LOG try: if not path.is_file() or path.stat().st_size < _LOG_TRIM_MIN_BYTES: return 0 data = path.read_bytes() if data.count(b"\n") <= max_lines: return 0 lines = data.splitlines(keepends=True) dropped = len(lines) - max_lines with path.open("r+b") as handle: handle.write(b"".join(lines[-max_lines:])) handle.truncate() return dropped except OSError as exc: print(f" Could not trim {path.name}: {exc}", file=sys.stderr) return 0 def wait_for_backend(timeout: float = 300.0) -> bool: """Block until the backend answers again. After a restart the port is not listening for a while, and submitting into it would look like another crash and burn one of the clip's attempts.""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: _comfy_get("/system_stats") return True except Exception: # noqa: BLE001 time.sleep(BACKEND_RETRY_SECONDS) return False def restart_comfy_backend() -> bool: """Stop whatever is on the port, then start a fresh backend. Restarting without stopping first is what fails in practice: the old process may be wedged, suspended or holding the GPU, and the new one then cannot bind the port or fit in VRAM. """ stop_comfy_backend() return start_comfy_backend(COMFY_START_SCRIPT) def _prompt_known(prompt_id: str) -> bool: """True while ComfyUI still has this prompt queued, running or in history. A restart wipes both, so this is how a job lost to a crash is spotted. If the answer cannot be determined, say yes: resubmitting a job that is actually still running would waste a generation. """ try: if prompt_id in _comfy_get(f"/history/{prompt_id}"): return True queue = _comfy_get("/queue") except Exception: # noqa: BLE001 return True for key in ("queue_running", "queue_pending"): for item in queue.get(key, []): if isinstance(item, (list, tuple)) and len(item) > 1 and item[1] == prompt_id: return True return False def run_workflow(workflow: dict, poll_interval: float = 0.25, max_interval: float = 2.0, backoff_after: float = 10.0) -> dict: """Submit a workflow, wait for it, and return its history entry. Survives ComfyUI dying mid-job. Its queue and history do not persist across a restart, so a prompt submitted beforehand exists nowhere afterwards and polling it would wait for ever — the job has to be sent again. While the backend is down this reports once rather than once per poll, and slows to BACKEND_RETRY_SECONDS instead of hammering a closed port. The poll interval, not the GPU, governs perceived speed: a clip that finishes 100 ms after a poll still waits for the next one. It was a flat 3 s, which quantised every clip to a 3 s boundary — across 2075 real clips 91% of the gaps landed exactly on a multiple of 3 s. So polling starts tight and only backs off once a job looks slow. """ prompt_id, offline = None, False start, interval, next_dot = time.monotonic(), poll_interval, 1.0 # Restarts are counted per clip, not per run. A crash in block 4000 of an # overnight queue deserves the same number of attempts the first one got. # One death is enough to hand control back: the caller retries with a # different seed, which is the only thing that changes the outcome. crashes = 0 _autostart["attempts"] = 0 while True: if prompt_id is None: # first submission, or a resubmit try: prompt_id = submit_workflow(workflow) except Exception as exc: # noqa: BLE001 if not offline: crashes += 1 print(f"\n ComfyUI unreachable ({exc}) " f"[{crashes}/{CLIP_CRASH_LIMIT} for this clip].", file=sys.stderr) if crashes >= 1: raise BackendCrashed(str(exc)) offline = True if not (COMFY_AUTOSTART and restart_comfy_backend()): print(" Waiting for it to come back — start it and this " "will carry on.", file=sys.stderr) time.sleep(BACKEND_RETRY_SECONDS) continue start, interval, next_dot = time.monotonic(), poll_interval, 1.0 print(f" Waiting for {prompt_id[:8]}…", end="", flush=True) try: history = _comfy_get(f"/history/{prompt_id}") except Exception as exc: # noqa: BLE001 if not offline: crashes += 1 print(f"\n ComfyUI went away ({exc}) " f"[{crashes}/{CLIP_CRASH_LIMIT} for this clip].", file=sys.stderr) # Dying *while this clip is in flight* is the signal that the # clip is what killed it — some inputs abort the process from # inside CUDA, where no exception can be raised. if crashes >= 1: raise BackendCrashed(str(exc)) offline = True if not (COMFY_AUTOSTART and restart_comfy_backend()): print(" Waiting for it to come back — restart it and this " "will carry on.", file=sys.stderr) time.sleep(BACKEND_RETRY_SECONDS) continue if offline: # it answered again offline = False print(" ComfyUI is back.", file=sys.stderr) if not _prompt_known(prompt_id): print(" The queue did not survive the restart — resubmitting " "this clip.", file=sys.stderr) prompt_id = None continue if prompt_id in history: entry = history[prompt_id] status = entry.get("status", {}) for msg_type, msg_data in status.get("messages", []): if msg_type == "execution_error": print(f"\nComfyUI error: {msg_data}", file=sys.stderr) sys.exit(1) if status.get("completed"): print(f" done ({time.monotonic() - start:.1f}s).") return entry elapsed = time.monotonic() - start # One dot per second regardless of poll rate, or the line fills with dots. if elapsed >= next_dot: print(".", end="", flush=True) next_dot += 1.0 time.sleep(interval) if elapsed > backoff_after: interval = min(interval * 1.5, max_interval) def extract_output_audio(history_entry: dict) -> Path: outputs = history_entry.get("outputs", {}) for node_output in outputs.values(): for audio_info in node_output.get("audio", []): subfolder = audio_info.get("subfolder", "") path = COMFY_OUTPUT_DIR / subfolder / audio_info["filename"] if path.exists(): return path raise RuntimeError(f"Output audio not found in history: {outputs}") def _generate_one(text: str, voice_filename: str, voice_ref_text: str, seed: int, speed: float, out_dir: Path, label: str, strip_leading_ms: float, temperature: float = 0.9, trailing_fade_ms: float = 0, max_new_tokens: int = 4096, retries: int = 2) -> tuple: """Submit one TTS job and return (path, duration_after_speedup, truncated). About one clip in twelve comes back cut off mid-word — the model stops generating before the line is finished. It is not predictable from the text, but it is obvious in the audio, so a clip that ends at full level is simply asked for again with a different seed. """ attempt, truncated = 0, False while True: workflow = build_text_workflow(text, voice_filename, voice_ref_text, seed, temperature, max_new_tokens) raw_clip = extract_output_audio(run_workflow(workflow)) truncated = ends_abruptly(raw_clip) if not truncated or attempt >= retries: break attempt += 1 seed += 7919 # a distant seed, not the neighbouring sample print(f" ! ends mid-word — retrying with seed {seed} " f"(attempt {attempt + 1} of {retries + 1})") if truncated: print(" ! still ends mid-word after " f"{attempt + 1} attempt{'s' if attempt else ''} — " "regenerate this one with --regen") if strip_leading_ms > 0: stripped = out_dir / f"{label}_s.flac" strip_leading_audio(raw_clip, stripped, strip_leading_ms) raw_clip = stripped raw_duration = get_audio_duration(raw_clip) trimmed = out_dir / f"{label}_t.flac" removed = trim_trailing_silence(raw_clip, trimmed) if removed > 0: raw_clip = trimmed # Losing most of the clip to silence means the model never stopped and # ran to the token ceiling. The speech is usually fine, but say so — # it can also mean the line was cut short, which needs a --regen. if removed > 20 and removed > raw_duration * 0.5: print(f" ! {removed:.0f}s of trailing silence removed " f"({raw_duration:.0f}s generated for {raw_duration - removed:.1f}s of " f"speech) — the model did not stop on its own; check this clip") fast_clip = out_dir / f"{label}.flac" speed_up_audio(raw_clip, fast_clip, speed, fade_out_ms=trailing_fade_ms) trimmed.unlink(missing_ok=True) return fast_clip, get_audio_duration(fast_clip), truncated def generate_clips(blocks: list, voice_filename: str, voice_ref_text: str, base_seed: int, speed: float, clips_dir: Path, strip_leading_ms: float = 0.0, resume_from: int = 0, temperature: float = 0.9, trailing_fade_ms: float = 0, max_new_tokens: int = 4096, retries: int = 2) -> list: """ Generate one audio clip per block via ComfyUI, apply per-clip speed-up, then calculate start positions with "push forward if overlap" logic. If resume_from > 0, blocks 1..(resume_from-1) are loaded from existing clip files instead of being regenerated (crash-recovery mode). Returns list of (actual_start_sec, clip_path). """ clips = [] cursor = 0.0 total = len(blocks) still_truncated = [] crashed = [] for idx, (t_start, t_end, text) in enumerate(blocks): clip_n = idx + 1 label = f"clip_{clip_n:04d}" canonical = clips_dir / f"{label}.flac" if resume_from > 0 and clip_n < resume_from: if not canonical.exists(): sys.exit(f"Resume from block {resume_from}: {canonical} not found — " f"lower --resume-from or regenerate from an earlier block.") duration = get_audio_duration(canonical) print(f" [{clip_n}/{total}] skip (resuming) — {canonical.name} {duration:.2f}s") actual_start = max(t_start, cursor) clips.append((actual_start, canonical)) cursor = actual_start + duration continue display = text.replace("\n", " ") print(f" [{clip_n}/{total}] {_fmt_srt_time(t_start)} {display[:60]!r}") skip = False attempt = 0 while True: # A crash is seed-dependent, not purely text-dependent: the block # that killed ComfyUI three times in a 2073-clip run generated # perfectly on its own with a different seed. So each attempt gets a # fresh seed — repeated tries at the identical computation would just # reproduce the same abort. seed = base_seed + idx + attempt * 100_003 try: fast_clip, duration, truncated = _generate_one( text, voice_filename, voice_ref_text, seed, speed, clips_dir, label, strip_leading_ms, temperature, trailing_fade_ms=trailing_fade_ms, max_new_tokens=max_new_tokens, retries=retries, ) break except BackendCrashed as exc: attempt += 1 # The backend died on this clip. That means one of # two very different things, and the difference decides whether # skipping is right: either this clip kills ComfyUI — some # inputs abort it from inside CUDA — or ComfyUI is simply not # running. Skipping the clip is right for the first and # disastrous for the second, where it would skip every # remaining clip and produce a silent film. So: restart, and # ask whether the backend is healthy. restart_comfy_backend() if wait_for_backend(BACKEND_HEALTH_TIMEOUT): if attempt < CLIP_CRASH_LIMIT: print(f" ComfyUI died on this clip " f"({attempt}/{CLIP_CRASH_LIMIT}) — retrying with " f"a different seed.", file=sys.stderr) continue crashed.append(clip_n) print(f" ! killed ComfyUI {CLIP_CRASH_LIMIT} times " f"({exc}) — skipping this clip and carrying on.", file=sys.stderr) skip = True break print(f" ComfyUI is not answering — waiting for it rather " f"than skipping clips. Start it and this will carry on.", file=sys.stderr) while not wait_for_backend(3600.0): pass # an overnight queue can afford to wait if skip: continue if truncated: still_truncated.append(clip_n) # Place at SRT time or after previous clip, whichever is later. # No trimming — clips play to their natural end; re-align at the next # natural SRT gap that is wider than the accumulated overrun. actual_start = max(t_start, cursor) clips.append((actual_start, fast_clip)) cursor = actual_start + duration delay = actual_start - t_start delay_str = f" [+{delay:.2f}s behind SRT]" if delay > 0.05 else "" print(f" → placed at {_fmt_srt_time(actual_start)}, duration {duration:.2f}s{delay_str}") if still_truncated: print(f"\n {len(still_truncated)} clip(s) still end mid-word after retries. " f"To try again:") print(f" --regen {' '.join(str(n) for n in still_truncated)}") if crashed: print(f"\n {len(crashed)} clip(s) crashed ComfyUI and were skipped — the " f"voiceover is silent there. Edit their text in tts_input.srt " f"(a one-word line is the usual culprit), then:") print(f" --regen {' '.join(str(n) for n in crashed)}") return clips # --------------------------------------------------------------------------- # Rebuild / Regen helpers # --------------------------------------------------------------------------- def job_dir(out_dir: Path, video_path: str) -> Path: """Per-job working directory: /clips/