lukas/lektor-ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e541ef7ee9fd59bde08800444fbdff2a5b4e8b74
lektor-ai/lektor_ai_comfy.py
T
claudeandClaude Opus 5 e541ef7ee9 Give a wedged ComfyUI ten restarts per clip instead of three 
Three was not enough in practice: a backend that dies during model load can
burn all three before it ever comes up healthy, and the clip is then skipped —
or, when the backend is simply gone, the run stalls waiting for it.

Both caps move together. COMFY_AUTOSTART_MAX is what actually gates a relaunch,
so raising CLIP_CRASH_LIMIT alone would still have stopped at three restarts.
Counting is unchanged: both are per clip, not per run.

The cost of a truly dead backend goes up accordingly — up to ten rounds of a
90s cooldown plus the 180s health timeout before a clip is given up on.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NkWDx9dVB7xBpqntvjvm6A
2026-08-31 21:15:36 +02:00

2674 lines
115 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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"(?<!\w)" + source
if pattern[-1].isalnum() or pattern[-1] == "_":
source += r"(?!\w)"
parts.append(f"(?P<{name}>{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'(?<!\d)(?<!\d[.,:/\-])(?<!\d[\u00a0\u202f ])'
r'\d'
r'(?!\d)(?![.,:/\-]\d)(?![\u00a0\u202f ]\d{3}(?!\d))')
_ANY_DIGIT_RE = re.compile(r'\d')
_TRAILING_PUNCT_RE = re.compile(r'[.…!?,:;]+$')
def needs_number_normalization(text: str) -> 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: <i>, </i>, <b>, <font …>
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'</?font[^>]*>', 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 <font face="Serif"><font size="18">. 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: <output_dir>/clips/<video filename without extension>/
Everything that belongs to one video lives here — the clips, the processed
tts_input.srt, the assembled audio, any extracted subtitle or video
fragment. Only the finished <stem>_lektor.mkv is written to <output_dir>.
Without this separation every job overwrote the previous one's clip_NNNN.flac
and tts_input.srt, so --rebuild and --regen were only ever safe on the most
recent run. The clips and tts_input.srt are a matched pair — block N in the
SRT is clip_NNNN.flac — which is why the SRT lives here too rather than at
the top level.
"""
return out_dir / "clips" / Path(video_path).stem
def _legacy_hint(out_dir: Path, job: Path) -> str:
"""Point at the pre-per-job layout if that is what the user actually has."""
if (out_dir / "tts_input.srt").is_file():
return (f"\n\nA tts_input.srt exists at {out_dir} in the old shared layout. "
f"Clips are now kept per job in {job}. Move that run's files there, "
f"or regenerate it.")
return ""
def _rebuild_video(args, out_dir: Path):
"""Re-assemble audio from existing clips and rebuild the final video.
No TTS generation — use this to change volumes or fix assembly without
waiting for the model to run again.
"""
job = job_dir(out_dir, args.video)
tts_srt = job / "tts_input.srt"
if not tts_srt.exists():
sys.exit(f"No tts_input.srt found in {job} — run a full generation first."
+ _legacy_hint(out_dir, job))
blocks = load_tts_input_srt(tts_srt)
total = len(blocks)
clips_dir = job
missing = [
clips_dir / f"clip_{i+1:04d}.flac"
for i in range(total)
if not (clips_dir / f"clip_{i+1:04d}.flac").exists()
]
if missing:
sys.exit("Missing clip files — regenerate them first:\n "
+ "\n ".join(str(p) for p in missing[:10]))
clips = []
cursor = 0.0
for idx, (t_start, t_end, text) in enumerate(blocks):
clip_path = clips_dir / f"clip_{idx+1:04d}.flac"
duration = get_audio_duration(clip_path)
actual_start = max(t_start, cursor)
clips.append((actual_start, clip_path))
cursor = actual_start + duration
assembled_tts = job / "tts_assembled.flac"
print(f"\n Assembling {total} clips…")
assemble_timed_audio(clips, str(assembled_tts))
print(f" Assembled: {assembled_tts} ({assembled_tts.stat().st_size // 1024} KB)")
# TTS timestamps are relative to --start, so we need the same video source
# the original run used: a fragment when --end was given, full video otherwise.
if args.end:
start_sec = _parse_mmss(args.start)
end_sec = _parse_mmss(args.end)
video_fragment = job / "video_fragment.mkv"
if not video_fragment.exists():
extract_video_fragment(args.video, start_sec, end_sec, str(video_fragment))
video_src = str(video_fragment)
else:
video_src = args.video
stem = Path(args.video).stem
final_output = out_dir / f"{stem}_lektor.mkv"
print(f"\n Building 2-track video "
f"(TTS {int(args.tts_volume*100)}% / orig {int(args.orig_volume*100)}%)…")
merge_voiceover(video_src, str(assembled_tts), str(final_output),
orig_volume=args.orig_volume, tts_volume=args.tts_volume,
extra_subs=voiceover_subs_file(job) if args.include_subs else None)
print(f"\nDone! Output: {final_output}")
def _check_job_clips(args, out_dir: Path):
"""List clips in this video's job folder that end mid-word.
Prints them as a --regen argument list, so fixing them is a copy and paste.
"""
job = job_dir(out_dir, args.video)
clips = sorted(job.glob("clip_*.flac"))
if not clips:
sys.exit(f"No clips found in {job}." + _legacy_hint(out_dir, job))
print(f"\n Checking {len(clips)} clips in {job}…")
bad = []
for clip in clips:
if ends_abruptly(clip):
try:
number = int(clip.stem.split("_")[1])
except (IndexError, ValueError):
continue
bad.append(number)
print(f" {clip.name} ends at full level "
f"({get_audio_duration(clip):.2f}s)")
if not bad:
print(f"\n No truncated clips. All {len(clips)} end cleanly.")
return
print(f"\n {len(bad)} of {len(clips)} clips ({100 * len(bad) / len(clips):.0f}%) "
f"end mid-word. To regenerate them:")
print(f" python3 {Path(__file__).name} \"{args.video}\" --regen "
f"{' '.join(str(n) for n in bad)}")
def _trim_job_clips(args, out_dir: Path):
"""Trim trailing silence from every clip already on disk for this video.
Repairs a job generated before trimming existed, or one where the model ran
to the token ceiling. Only silence is removed, so the speech is untouched
and nothing needs regenerating — follow with --rebuild.
"""
job = job_dir(out_dir, args.video)
clips = sorted(job.glob("clip_*.flac"))
if not clips:
sys.exit(f"No clips found in {job}." + _legacy_hint(out_dir, job))
print(f"\n Scanning {len(clips)} clips in {job}…")
total_removed, changed, runaways = 0.0, 0, []
for i, clip in enumerate(clips, 1):
before = get_audio_duration(clip)
tmp = clip.with_suffix(".trim.flac")
removed = trim_trailing_silence(clip, tmp)
if removed > 0.02:
tmp.replace(clip)
total_removed += removed
changed += 1
if removed > 20:
runaways.append((clip.name, before, removed))
print(f" {clip.name} {before:7.2f}s -> {before - removed:6.2f}s "
f"(-{removed:.2f}s)")
else:
tmp.unlink(missing_ok=True)
print(f"\n Trimmed {changed}/{len(clips)} clips, {total_removed:.0f}s of silence removed.")
if runaways:
print(f" {len(runaways)} clip(s) had run to the token ceiling:")
for name, before, removed in runaways:
print(f" {name}: {before:.0f}s generated for {before - removed:.1f}s of speech")
print(" Their speech is intact, but listen to them — if any line is cut "
"short, regenerate it with --regen.")
print("\n Now run --rebuild to reassemble with the corrected timing.")
def _regen_clips(args, out_dir: Path):
"""Regenerate specific clips by tts_input block number.
Only replaces FLAC files — run --rebuild afterwards to reassemble audio and video.
"""
job = job_dir(out_dir, args.video)
tts_srt = job / "tts_input.srt"
if not tts_srt.exists():
sys.exit(f"No tts_input.srt found in {job} — run a full generation first."
+ _legacy_hint(out_dir, job))
blocks = load_tts_input_srt(tts_srt)
total = len(blocks)
regen_set = set(args.regen)
clips_dir = job
clips_dir.mkdir(parents=True, exist_ok=True)
bad = [n for n in regen_set if n < 1 or n > total]
if bad:
sys.exit(f"Block number(s) out of range (tts_input.srt has {total} blocks): {bad}")
# Set up voice
wav_path = Path(args.voice)
if not wav_path.is_file():
wav_path = VOICE_PROMPTS_DIR / f"{args.voice}.wav"
if not wav_path.is_file():
sys.exit(f"Voice WAV not found: {wav_path}")
voice_filename, voice_ref_text = upload_voice_wav(wav_path)
base_seed = args.seed if args.seed != 0 else random.randint(1, 2**31 - 1)
print(f"\n Regenerating blocks {sorted(regen_set)} with seed {base_seed}…")
failed = []
for idx, (t_start, t_end, text) in enumerate(blocks):
clip_n = idx + 1
if clip_n not in regen_set:
continue
label = f"clip_{clip_n:04d}"
print(f"\n REGEN [{clip_n}/{total}] {text[:70]!r}")
seed = base_seed + idx
for attempt in range(CLIP_CRASH_LIMIT):
try:
clip_path, duration, truncated = _generate_one(
text, voice_filename, voice_ref_text,
seed + attempt * 100_003, args.speed, clips_dir, label,
args.strip_leading_ms, args.temperature,
trailing_fade_ms=args.trailing_fade_ms,
max_new_tokens=args.max_tokens,
retries=args.truncation_retries,
)
print(f" → saved as {clip_path.name} ({duration:.2f}s)")
break
except BackendCrashed as exc:
restart_comfy_backend()
wait_for_backend(BACKEND_HEALTH_TIMEOUT)
print(f" ComfyUI died on this clip ({attempt + 1}/"
f"{CLIP_CRASH_LIMIT}): {exc}", file=sys.stderr)
else:
failed.append(clip_n)
print(f" ! gave up on block {clip_n} — it keeps killing ComfyUI. "
f"Try editing its text in tts_input.srt.", file=sys.stderr)
if failed:
print(f"\n {len(failed)} block(s) could not be generated: "
f"{' '.join(str(n) for n in failed)}")
print(f"\nDone. Run --rebuild to reassemble audio and rebuild the video.")
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
global COMFY_URL, COMFY_AUTOSTART, COMFY_START_SCRIPT
# --settings is resolved before the main parser is built, because the file's
# contents become that parser's defaults.
pre = argparse.ArgumentParser(add_help=False)
pre.add_argument("--settings", default=str(SETTINGS_FILE), metavar="FILE")
pre_args, _ = pre.parse_known_args()
settings_path = Path(pre_args.settings)
cfg = load_settings(settings_path)
parser = argparse.ArgumentParser(
description="Generate AI voiceover via ComfyUI + Higgs Audio V3, per-subtitle.",
epilog=f"Defaults are read from {SETTINGS_FILE.name} next to this script; "
f"any flag given on the command line overrides it.",
)
parser.add_argument("--settings", default=str(SETTINGS_FILE), metavar="FILE",
help=f"Settings file to read defaults from "
f"(default: {SETTINGS_FILE.name})")
parser.add_argument("srt", nargs="?", default=None,
help="Input SRT subtitle file (omit with --embedded-subs)")
parser.add_argument("video", nargs="?", default=None,
help="Input video file (MKV, MP4, …)")
parser.add_argument("--start", default="00:00", metavar="MM:SS",
help="Fragment start time (default: 00:00)")
parser.add_argument("--end", default=None, metavar="MM:SS",
help="Fragment end time")
parser.add_argument("--voice", default=cfg["voice"],
help=f"Voice prompt name or WAV path (default: {cfg['voice']})")
parser.add_argument("--speed", type=float, default=cfg["speed"], metavar="FACTOR",
help=f"Per-clip playback speed via atempo (default: {cfg['speed']})")
parser.add_argument("--seed", type=int, default=0,
help="Base generation seed (default: 0 = random each run)")
parser.add_argument("--dictionary", default=cfg["dictionary"], metavar="FILE",
help="Word/phrase replacements applied to the subtitles "
f"before TTS (default: {cfg['dictionary']}). "
"Pass an empty string to disable.")
parser.add_argument("--dictionary-extra", default=None, metavar="FILE",
help="A second replacement list, merged with the first. "
"The GUI writes one per queue so a batch can carry "
"its own names and fixes without touching the "
"shared dictionary.")
numbers = parser.add_mutually_exclusive_group()
numbers.add_argument("--normalize-numbers", dest="normalize_numbers",
action="store_true", default=cfg["normalize_numbers"],
help="Spell digits out as Polish words before TTS "
f"(default: {'on' if cfg['normalize_numbers'] else 'off'})")
numbers.add_argument("--no-normalize-numbers", dest="normalize_numbers",
action="store_false",
help="Leave digits as digits; the TTS often misreads them.")
parser.add_argument("--orig-volume", type=float, default=cfg["orig_volume"], metavar="0-1",
help=f"Original audio volume in mix (default: {cfg['orig_volume']})")
parser.add_argument("--tts-volume", type=float, default=cfg["tts_volume"], metavar="0-2",
help=f"Voiceover volume in mix (default: {cfg['tts_volume']})")
parser.add_argument("--output-dir", default=cfg["output_dir"],
help=f"Output directory (default: {cfg['output_dir']}/)")
parser.add_argument("--comfy-url", default=cfg["comfy_url"],
help=f"ComfyUI URL (default: {cfg['comfy_url']})")
parser.add_argument("--skip-tts", metavar="AUDIO_FILE",
help="Skip TTS and use this pre-assembled audio file")
parser.add_argument("--temperature", type=float, default=cfg["temperature"], metavar="T",
help=f"Sampling temperature (default: {cfg['temperature']}; "
f"lower = more deterministic)")
parser.add_argument("--strip-leading-ms", type=float, default=cfg["strip_leading_ms"],
metavar="MS",
help=f"Strip N ms from the start of each TTS clip to remove "
f"leading breath/noise artifact "
f"(default: {cfg['strip_leading_ms']}, try 80–150)")
parser.add_argument("--trailing-fade-ms", type=float, default=cfg["trailing_fade_ms"],
metavar="MS",
help=f"Fade out the last N ms of each TTS clip to suppress "
f"tail pops/noise (default: {cfg['trailing_fade_ms']}, "
f"set 0 to disable)")
parser.add_argument("--max-tokens", type=int, default=cfg["max_tokens"], metavar="N",
help=f"Max audio tokens the TTS model may generate per clip "
f"(default: {cfg['max_tokens']}; raise if long clips are "
f"cut off mid-word)")
parser.add_argument("--resume-from", type=int, default=0, metavar="N",
help="Resume a crashed run: skip blocks 1..(N-1) by loading "
"their existing clip files, generate from block N onwards.")
parser.add_argument("--rebuild", action="store_true",
help="Re-assemble audio from existing clips and rebuild the "
"final video without any TTS generation. "
"Use this to change --orig-volume / --tts-volume.")
parser.add_argument("--regen", nargs="+", type=int, metavar="N",
help="Regenerate specific clip(s) by tts_input block number "
"(same number as in clips/clip_NNNN.flac and tts_input.srt). "
"Only replaces the selected FLAC files; run --rebuild "
"afterwards to reassemble audio and rebuild the video.")
parser.add_argument("--no-comfy-autostart", action="store_true",
help="Do not launch ComfyUI automatically when it is not "
"answering. By default it is started via the script named "
"by [comfy] start_command in settings.ini.")
parser.add_argument("--truncation-retries", type=int,
default=cfg["truncation_retries"], metavar="N",
help=f"How many extra attempts to make when the model returns a "
f"clip cut off mid-word (default: {cfg['truncation_retries']}; "
f"0 disables retrying).")
parser.add_argument("--check-clips", action="store_true",
help="Report which clips in this video's job folder end mid-word, "
"as a ready-made --regen list, then stop.")
parser.add_argument("--trim-silence", action="store_true",
help="Trim trailing silence from the clips already generated for "
"this video, then stop. Repairs runs where the model failed "
"to stop and padded a clip out to --max-tokens, which pushes "
"every later clip out of sync. Follow with --rebuild.")
parser.add_argument("--include-subs", action="store_true",
help="Also mux the subtitles the voiceover was made from into "
"the finished video, as an extra track alongside whatever "
"subtitle tracks the source already had. Works with "
"--rebuild too: the run keeps its own copy of them.")
parser.add_argument("--embedded-sub-index", type=int, default=None, metavar="N",
help="With --embedded-subs, pick this ffmpeg stream index instead "
"of prompting. Used by ui.py and for scripted runs.")
parser.add_argument("--embedded-subs", action="store_true",
help="Pick a subtitle track embedded in the video file instead of "
"providing an external SRT. Lists available tracks and prompts "
"for a choice. The video path may be given as the first "
"positional argument (no SRT needed).")
args = parser.parse_args()
COMFY_URL = args.comfy_url
COMFY_AUTOSTART = cfg["comfy_autostart"] and not args.no_comfy_autostart
start_cmd = Path(cfg["comfy_start_command"]).expanduser()
COMFY_START_SCRIPT = start_cmd if start_cmd.is_absolute() \
else Path(__file__).resolve().parent / start_cmd
if settings_path.is_file():
print(f" Settings: {settings_path}")
else:
print(f" Settings: {settings_path.name} not found — using built-in defaults")
# ── Resolve positional ambiguity for modes that don't need an SRT file ────
# When the user writes `script.py video.mkv --embedded-subs` (or --rebuild /
# --regen without an SRT), argparse puts the lone positional into args.srt.
# Detect this and move it to args.video.
needs_srt = not (args.embedded_subs or args.rebuild or args.trim_silence
or args.check_clips or bool(args.regen))
if args.video is None and args.srt is not None and not needs_srt:
args.video = args.srt
args.srt = None
if args.video is None:
parser.error("A video file is required.")
if needs_srt and args.srt is None:
parser.error("Provide an SRT subtitle file, or use --embedded-subs to pick one "
"from the video.")
out_dir = Path(args.output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
# ── Maintenance: report clips the model cut short, then stop ─────────────
if args.check_clips:
_check_job_clips(args, out_dir)
return
# ── Maintenance: trim silence from existing clips, then stop ─────────────
if args.trim_silence:
_trim_job_clips(args, out_dir)
return
# ── Rebuild mode: reassemble + remux from existing clips, no TTS ─────────
if args.rebuild:
_rebuild_video(args, out_dir)
return
# ── Regen mode: replace specific clips, then rebuild audio + video ────────
if args.regen:
_regen_clips(args, out_dir)
return
# Everything belonging to this video lives together, so successive jobs
# cannot overwrite each other's clips or tts_input.srt.
clips_dir = job_dir(out_dir, args.video)
clips_dir.mkdir(parents=True, exist_ok=True)
print(f" Job folder: {clips_dir}")
start_sec = _parse_mmss(args.start)
end_sec = _parse_mmss(args.end) if args.end else None
# ── Step 1: Parse subtitles ───────────────────────────────────────────────
# Resuming reuses the text the interrupted run already produced. Re-deriving
# it would be worse than wasteful: the dictionary or the normalizer may have
# changed since, and --resume-from relies on block N still being
# clip_NNNN.flac. A different block count, or the same count with different
# text, silently pairs old clips with new lines.
resume_srt = clips_dir / "tts_input.srt"
if args.resume_from and resume_srt.is_file():
print(f"\n[1/4] Resuming from block {args.resume_from} — reusing "
f"{resume_srt.name}…")
blocks = load_tts_input_srt(resume_srt)
if not blocks:
sys.exit(f"{resume_srt} is empty — cannot resume from it.")
if args.resume_from > len(blocks):
sys.exit(f"Cannot resume from block {args.resume_from}: "
f"{resume_srt.name} has {len(blocks)}.")
print(f" {len(blocks)} blocks, subtitles not reprocessed — the "
f"numbering still matches the clips on disk.")
return _run_generation(args, cfg, blocks, clips_dir, start_sec, end_sec)
if args.resume_from:
print(f" WARNING: no {resume_srt.name} in {clips_dir} — the subtitles "
f"will be reprocessed, which may not match the existing clips.",
file=sys.stderr)
print(f"\n[1/4] Parsing subtitles {args.start}–{args.end or 'end'}…")
if args.embedded_subs:
sub_list = list_subtitle_streams(args.video)
if not sub_list:
sys.exit(f"No subtitle tracks found in {Path(args.video).name}.")
if args.embedded_sub_index is not None:
# Non-interactive selection by ffmpeg stream index, for the GUI and
# for scripted batch runs. Interactive behaviour is unchanged when
# this is omitted.
chosen = next((s for s in sub_list
if s["index"] == args.embedded_sub_index), None)
if chosen is None:
avail = ", ".join(str(s["index"]) for s in sub_list)
sys.exit(f"No subtitle stream #{args.embedded_sub_index} in "
f"{Path(args.video).name}. Available: {avail}")
print(f" Using subtitle stream #{chosen['index']} "
f"({chosen['codec_name']}{' ' + chosen['language'] if chosen['language'] else ''})")
else:
print(f" Available subtitle tracks in {Path(args.video).name}:")
for i, s in enumerate(sub_list):
lang = f" [{s['language']}]" if s['language'] else ""
title = f" — {s['title']}" if s['title'] else ""
compat = ("" if s['codec_name'].lower() in _TEXT_SUB_CODECS
else " (image-based, cannot be used for TTS)")
print(f" [{i + 1}] Stream #{s['index']}: {s['codec_name']}{lang}{title}{compat}")
while True:
try:
choice = int(input(" Select track number: ").strip()) - 1
if 0 <= choice < len(sub_list):
break
print(f" Enter a number from 1 to {len(sub_list)}.")
except (ValueError, EOFError, KeyboardInterrupt):
sys.exit("\nAborted.")
chosen = sub_list[choice]
if chosen['codec_name'].lower() not in _TEXT_SUB_CODECS:
sys.exit(f"Stream #{chosen['index']} ({chosen['codec_name']}) is image-based "
"and cannot be converted to text for TTS.")
embedded_srt = clips_dir / "embedded_sub.srt"
extract_subtitle_track(args.video, chosen['index'], str(embedded_srt))
srt_path = str(embedded_srt)
else:
srt_path = args.srt
if end_sec is None:
end_sec = 99 * 3600
# Kept before any processing, and for every run: --rebuild --include-subs
# is given only the video, so this is the only place the original wording
# can be preserved from.
save_voiceover_subs(srt_path, clips_dir, start_sec, end_sec)
wanted = []
if args.dictionary.strip():
wanted.append(resolve_dictionary_path(args.dictionary))
if args.dictionary_extra and args.dictionary_extra.strip():
wanted.append(Path(args.dictionary_extra).expanduser())
if wanted:
dictionary = Dictionary.load_all(wanted)
if not dictionary.sources:
print(f" Dictionary: none of {', '.join(str(w) for w in wanted)} "
f"found — no replacements")
else:
for src, count in dictionary.sources:
print(f" Dictionary: {count} rule(s) from {src}")
else:
dictionary = None
blocks = parse_subtitle_blocks(srt_path, start_sec, end_sec, dictionary)
if not blocks:
sys.exit("No subtitle blocks found in the specified range.")
if dictionary is not None and dictionary.hits:
print(f" {dictionary.hits} dictionary replacement(s) applied.")
blocks = split_dialogue_blocks(blocks)
print(f" {len(blocks)} entries after dialogue splitting.")
# Numbers go last, on the final per-clip text: the dictionary may have turned
# "16%" into "16 procent" first, and dialogue splitting decides which clip a
# number ends up in.
if args.normalize_numbers:
py = Path(cfg["normalizer_python"]).expanduser()
if not py.is_absolute():
py = Path(__file__).resolve().parent / py
llm = Path(cfg["normalizer_llm"]).expanduser() if cfg["normalizer_llm"] else None
if llm and not llm.is_absolute():
llm = Path(__file__).resolve().parent / llm
blocks = normalize_block_numbers(blocks, py, cfg["normalizer_timeout"],
free_vram=cfg["free_comfy_vram"],
llm_dir=llm)
srt_preview = clips_dir / "tts_input.srt"
srt_preview.write_text(blocks_to_srt(blocks), encoding="utf-8")
print(f" Saved to {srt_preview} for inspection.")
return _run_generation(args, cfg, blocks, clips_dir, start_sec, end_sec)
def _run_generation(args, cfg, blocks, clips_dir, start_sec, end_sec):
"""Steps 2-4: TTS, assembly, mux. Shared by a fresh run and a resumed one,
which differ only in where the blocks came from."""
out_dir = Path(args.output_dir)
# ── Step 2: Generate TTS per block ────────────────────────────────────────
if args.skip_tts:
assembled_tts = Path(args.skip_tts)
print(f"\n[2/4] Using existing TTS audio: {assembled_tts}")
else:
print(f"\n[2/4] Generating TTS ({len(blocks)} clips, speed {args.speed}×)…")
wav_path = Path(args.voice)
if not wav_path.is_file():
wav_path = VOICE_PROMPTS_DIR / f"{args.voice}.wav"
if not wav_path.is_file():
sys.exit(f"Voice WAV not found: {wav_path}")
print(f" Voice: {wav_path.name}")
base_seed = args.seed if args.seed != 0 else random.randint(1, 2**31 - 1)
print(f" Base seed: {base_seed}")
voice_filename, voice_ref_text = upload_voice_wav(wav_path)
clips = generate_clips(blocks, voice_filename, voice_ref_text,
base_seed, args.speed, clips_dir,
strip_leading_ms=args.strip_leading_ms,
resume_from=args.resume_from,
temperature=args.temperature,
trailing_fade_ms=args.trailing_fade_ms,
max_new_tokens=args.max_tokens,
retries=args.truncation_retries)
# The queue is done, so ComfyUI is holding ~8 GB it no longer needs
# while ffmpeg does the rest. Unload rather than stop it: the backend
# stays warm and reloads by itself on the next TTS request.
if cfg["free_comfy_vram"]:
free_comfy_vram()
assembled_tts = clips_dir / "tts_assembled.flac"
assemble_timed_audio(clips, str(assembled_tts))
print(f" Assembled: {assembled_tts} ({assembled_tts.stat().st_size // 1024} KB)")
print(f" Clips saved to {clips_dir}")
# ── Step 3: Extract video fragment ───────────────────────────────────────
print(f"\n[3/4] Extracting video fragment…")
real_end = _parse_mmss(args.end) if args.end else None
if real_end is not None:
video_fragment = clips_dir / "video_fragment.mkv"
extract_video_fragment(args.video, start_sec, real_end, str(video_fragment))
else:
video_fragment = Path(args.video)
# ── Step 4: Mix voiceover with video ─────────────────────────────────────
stem = Path(args.video).stem
final_output = out_dir / f"{stem}_lektor.mkv"
print(f"\n[4/4] Building 2-track video (Original + Polish lektor "
f"TTS {int(args.tts_volume*100)}% / orig {int(args.orig_volume*100)}%)…")
merge_voiceover(str(video_fragment), str(assembled_tts), str(final_output),
orig_volume=args.orig_volume, tts_volume=args.tts_volume,
extra_subs=(voiceover_subs_file(clips_dir)
if args.include_subs else None))
print(f"\nDone! Output: {final_output}")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink