339 lines
15 KiB
Python
339 lines
15 KiB
Python
 
|
#!/usr/bin/env python3
|
|||
|
|
"""Read a plain text file aloud with the lektor_ai voice — no video, no SRT.
|
|||
|
|
|
|||
|
|
The dubbing pipeline is built around subtitles: every clip is placed at a
|
|||
|
|
timestamp against a film. Prose has no timings, so this script does the one
|
|||
|
|
thing that is genuinely missing — it splits the text into sentence-sized
|
|||
|
|
chunks itself — and then reuses lektor_ai_comfy for everything else: the same
|
|||
|
|
dictionary, the same Polish number normalizer, the same ComfyUI TTS call with
|
|||
|
|
its truncation retries and crash recovery, the same per-clip trim and speed-up.
|
|||
|
|
The clips are laid end to end with a short pause instead of at an SRT time.
|
|||
|
|
|
|||
|
|
python3 tts_txt.py input/oko_proroka.txt
|
|||
|
|
python3 tts_txt.py input/oko_proroka.txt -o output/oko.mp3 --speed 1.1
|
|||
|
|
python3 tts_txt.py input/oko_proroka.txt --dry-run # just show the chunks
|
|||
|
|
|
|||
|
|
Clips live in output/clips/<stem>/ exactly as for a film, so an interrupted
|
|||
|
|
run resumes with --resume-from N and a re-assembly costs nothing.
|
|||
|
|
"""
|
|||
|
|
import argparse
|
|||
|
|
import random
|
|||
|
|
import re
|
|||
|
|
import sys
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
import lektor_ai_comfy as lek
|
|||
|
|
|
|||
|
|
# Chunk size is a compromise, not a limit of the model: longer chunks read with
|
|||
|
|
# better prosody but are likelier to come back cut off mid-word, and a retry
|
|||
|
|
# then costs the whole chunk. Subtitle blocks — what the voice was tuned on —
|
|||
|
|
# sit well under this.
|
|||
|
|
MAX_CHARS = 240
|
|||
|
|
|
|||
|
|
SENTENCE_GAP = 0.35 # s of silence between chunks
|
|||
|
|
PARAGRAPH_GAP = 0.90 # s at a blank line
|
|||
|
|
|
|||
|
|
# A dot after one of these is not the end of a sentence. Single letters are
|
|||
|
|
# handled separately ("p. kasztelana", initials).
|
|||
|
|
ABBREV = {
|
|||
|
|
"np.", "itd.", "itp.", "tzn.", "tj.", "ok.", "dr.", "prof.", "św.",
|
|||
|
|
"ul.", "nr.", "str.", "m.in.", "tzw.", "ds.", "godz.", "mln.", "mld.",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
_SENT_SPLIT = re.compile(r'(?<=[.!?…])["»”\'’)\]]*\s+')
|
|||
|
|
_CLOSERS = '"»”\'’)]'
|
|||
|
|
_CLAUSE_SPLIT = re.compile(r'(?<=[,;:])\s+|\s+(?=—\s)')
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _terminal_punct(text: str) -> str:
|
|||
|
|
"""Give the chunk an ending the model can land on.
|
|||
|
|
|
|||
|
|
lek._ensure_terminal_punct is written for subtitles and does not know the
|
|||
|
|
single-character ellipsis or a closing quote, so it would turn `uciekło…`
|
|||
|
|
into `uciekło……` and `jeść!”` into `jeść!”...`.
|
|||
|
|
"""
|
|||
|
|
if not text:
|
|||
|
|
return text
|
|||
|
|
bare = text.rstrip(_CLOSERS)
|
|||
|
|
if bare and bare[-1] in '.,!?;:…':
|
|||
|
|
return text
|
|||
|
|
return text + '…'
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _prepare(text: str, dictionary) -> str:
|
|||
|
|
text = lek._strip_sub_tags(text)
|
|||
|
|
# A chunk cut out of the middle of a long sentence starts lowercase, which
|
|||
|
|
# the TTS reads with the wrong intonation — the pipeline capitalises these
|
|||
|
|
# for the same reason.
|
|||
|
|
if text and text[0].islower():
|
|||
|
|
text = text[0].upper() + text[1:]
|
|||
|
|
if dictionary is not None:
|
|||
|
|
text = dictionary.apply(text)
|
|||
|
|
return _terminal_punct(text)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _is_real_boundary(before: str) -> bool:
|
|||
|
|
"""False when the dot that ended `before` belongs to an abbreviation."""
|
|||
|
|
tail = before.rstrip('"»”\'’)]').split()[-1] if before.split() else ""
|
|||
|
|
if not tail.endswith("."):
|
|||
|
|
return True # ended on ! ? … — always a boundary
|
|||
|
|
if tail.lower() in ABBREV:
|
|||
|
|
return False
|
|||
|
|
stem = tail[:-1]
|
|||
|
|
return not (len(stem) == 1 and stem.isalpha()) # "p.", initials
|
|||
|
|
|
|||
|
|
|
|||
|
|
def split_sentences(paragraph: str) -> list:
|
|||
|
|
parts, pos = [], 0
|
|||
|
|
for m in _SENT_SPLIT.finditer(paragraph):
|
|||
|
|
candidate = paragraph[pos:m.start()]
|
|||
|
|
if not _is_real_boundary(candidate):
|
|||
|
|
continue
|
|||
|
|
parts.append(paragraph[pos:m.end()].strip())
|
|||
|
|
pos = m.end()
|
|||
|
|
tail = paragraph[pos:].strip()
|
|||
|
|
if tail:
|
|||
|
|
parts.append(tail)
|
|||
|
|
return parts
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _split_long(sentence: str, max_chars: int) -> list:
|
|||
|
|
"""A sentence longer than one chunk, broken at commas and dashes.
|
|||
|
|
|
|||
|
|
Nineteenth-century prose runs to sentences of several hundred characters;
|
|||
|
|
handing one to the model whole is exactly the case that comes back
|
|||
|
|
truncated.
|
|||
|
|
"""
|
|||
|
|
pieces, current = [], ""
|
|||
|
|
for part in _CLAUSE_SPLIT.split(sentence):
|
|||
|
|
if current and len(current) + 1 + len(part) > max_chars:
|
|||
|
|
pieces.append(current)
|
|||
|
|
current = part
|
|||
|
|
else:
|
|||
|
|
current = f"{current} {part}".strip()
|
|||
|
|
if current:
|
|||
|
|
pieces.append(current)
|
|||
|
|
return pieces
|
|||
|
|
|
|||
|
|
|
|||
|
|
def chunk_text(raw: str, max_chars: int = MAX_CHARS) -> list:
|
|||
|
|
"""Text -> [(chunk_text, ends_paragraph)].
|
|||
|
|
|
|||
|
|
Hard line wraps inside a paragraph are joined: they are an artefact of how
|
|||
|
|
the file was typeset, not pauses the reader should hear. A blank line is a
|
|||
|
|
real break and becomes the longer pause.
|
|||
|
|
"""
|
|||
|
|
chunks = []
|
|||
|
|
paragraphs = [p for p in re.split(r'\n\s*\n', raw) if p.strip()]
|
|||
|
|
for para in paragraphs:
|
|||
|
|
joined = re.sub(r'\s*\n\s*', ' ', para.strip())
|
|||
|
|
joined = re.sub(r'([?!.…])(—)', r'\1 \2', joined) # "jak?— Ja" -> "jak? — Ja"
|
|||
|
|
joined = re.sub(r'\s{2,}', ' ', joined)
|
|||
|
|
buf = []
|
|||
|
|
for sentence in split_sentences(joined):
|
|||
|
|
for piece in ([sentence] if len(sentence) <= max_chars
|
|||
|
|
else _split_long(sentence, max_chars)):
|
|||
|
|
if buf and len(" ".join(buf)) + 1 + len(piece) > max_chars:
|
|||
|
|
chunks.append([" ".join(buf), False])
|
|||
|
|
buf = []
|
|||
|
|
buf.append(piece)
|
|||
|
|
if buf:
|
|||
|
|
chunks.append([" ".join(buf), False])
|
|||
|
|
if chunks:
|
|||
|
|
chunks[-1][1] = True # last chunk of the paragraph
|
|||
|
|
return [(t, end) for t, end in chunks]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main():
|
|||
|
|
pre = argparse.ArgumentParser(add_help=False)
|
|||
|
|
pre.add_argument("--settings", default=str(lek.SETTINGS_FILE), metavar="FILE")
|
|||
|
|
settings_path = Path(pre.parse_known_args()[0].settings).expanduser()
|
|||
|
|
cfg = lek.load_settings(settings_path)
|
|||
|
|
|
|||
|
|
p = argparse.ArgumentParser(
|
|||
|
|
description="Turn a plain text file into a lektor-voiced audio file.",
|
|||
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
|
|||
|
|
p.add_argument("text", help="UTF-8 .txt file to read aloud")
|
|||
|
|
p.add_argument("-o", "--output", default=None, metavar="FILE",
|
|||
|
|
help="output audio (extension picks the format); "
|
|||
|
|
"default <output_dir>/<stem>.mp3")
|
|||
|
|
p.add_argument("--settings", default=str(lek.SETTINGS_FILE), metavar="FILE")
|
|||
|
|
p.add_argument("--voice", default=cfg["voice"],
|
|||
|
|
help="voice name in voices/ or a path to a WAV")
|
|||
|
|
# 1.0 rather than the settings default: the pipeline's speed-up exists to
|
|||
|
|
# fit a clip into its subtitle slot, and prose has no slot to fit.
|
|||
|
|
p.add_argument("--speed", type=float, default=1.0, metavar="FACTOR")
|
|||
|
|
p.add_argument("--temperature", type=float, default=cfg["temperature"], metavar="T")
|
|||
|
|
p.add_argument("--max-tokens", type=int, default=cfg["max_tokens"], metavar="N")
|
|||
|
|
p.add_argument("--seed", type=int, default=0, help="0 = random")
|
|||
|
|
p.add_argument("--truncation-retries", type=int,
|
|||
|
|
default=cfg["truncation_retries"], metavar="N")
|
|||
|
|
p.add_argument("--strip-leading-ms", type=float, default=cfg["strip_leading_ms"])
|
|||
|
|
p.add_argument("--trailing-fade-ms", type=float, default=cfg["trailing_fade_ms"])
|
|||
|
|
p.add_argument("--dictionary", default=cfg["dictionary"], metavar="FILE",
|
|||
|
|
help='replacement list; "" disables')
|
|||
|
|
p.add_argument("--dictionary-extra", default=None, metavar="FILE",
|
|||
|
|
help="second list merged on top, for rules specific to this text")
|
|||
|
|
numbers = p.add_mutually_exclusive_group()
|
|||
|
|
numbers.add_argument("--normalize-numbers", dest="normalize_numbers",
|
|||
|
|
action="store_true", default=cfg["normalize_numbers"])
|
|||
|
|
numbers.add_argument("--no-normalize-numbers", dest="normalize_numbers",
|
|||
|
|
action="store_false")
|
|||
|
|
p.add_argument("--max-chars", type=int, default=MAX_CHARS, metavar="N",
|
|||
|
|
help="longest chunk handed to the model in one go")
|
|||
|
|
p.add_argument("--sentence-gap", type=float, default=SENTENCE_GAP, metavar="S")
|
|||
|
|
p.add_argument("--paragraph-gap", type=float, default=PARAGRAPH_GAP, metavar="S")
|
|||
|
|
p.add_argument("--limit", type=int, default=0, metavar="N",
|
|||
|
|
help="generate only the first N chunks (for a quick listen)")
|
|||
|
|
p.add_argument("--resume-from", type=int, default=0, metavar="N",
|
|||
|
|
help="reuse clips 1..N-1 already in the job folder")
|
|||
|
|
p.add_argument("--rebuild", action="store_true",
|
|||
|
|
help="re-assemble from the clips on disk, no TTS")
|
|||
|
|
p.add_argument("--dry-run", action="store_true",
|
|||
|
|
help="print the chunks and exit without touching ComfyUI")
|
|||
|
|
p.add_argument("--output-dir", default=cfg["output_dir"])
|
|||
|
|
p.add_argument("--comfy-url", default=cfg["comfy_url"])
|
|||
|
|
p.add_argument("--no-comfy-autostart", action="store_true")
|
|||
|
|
args = p.parse_args()
|
|||
|
|
|
|||
|
|
lek.COMFY_URL = args.comfy_url
|
|||
|
|
lek.COMFY_AUTOSTART = cfg["comfy_autostart"] and not args.no_comfy_autostart
|
|||
|
|
start_cmd = Path(cfg["comfy_start_command"]).expanduser()
|
|||
|
|
lek.COMFY_START_SCRIPT = (start_cmd if start_cmd.is_absolute()
|
|||
|
|
else Path(lek.__file__).resolve().parent / start_cmd)
|
|||
|
|
|
|||
|
|
src = Path(args.text)
|
|||
|
|
if not src.is_file():
|
|||
|
|
sys.exit(f"No such text file: {src}")
|
|||
|
|
out_dir = Path(args.output_dir)
|
|||
|
|
clips_dir = lek.job_dir(out_dir, str(src))
|
|||
|
|
clips_dir.mkdir(parents=True, exist_ok=True)
|
|||
|
|
final = Path(args.output) if args.output else out_dir / f"{src.stem}.mp3"
|
|||
|
|
|
|||
|
|
print(f" Settings: {settings_path if settings_path.is_file() else 'built-in defaults'}")
|
|||
|
|
print(f" Job folder: {clips_dir}")
|
|||
|
|
|
|||
|
|
# ── Step 1: Text -> blocks ────────────────────────────────────────────────
|
|||
|
|
print(f"\n[1/3] Splitting {src.name}…")
|
|||
|
|
chunks = chunk_text(src.read_text(encoding="utf-8-sig"), args.max_chars)
|
|||
|
|
if not chunks:
|
|||
|
|
sys.exit(f"{src} has no text.")
|
|||
|
|
if args.limit:
|
|||
|
|
chunks = chunks[:args.limit]
|
|||
|
|
print(f" --limit {args.limit}: only the first {len(chunks)} chunk(s).")
|
|||
|
|
|
|||
|
|
wanted = []
|
|||
|
|
if args.dictionary.strip():
|
|||
|
|
wanted.append(lek.resolve_dictionary_path(args.dictionary))
|
|||
|
|
if args.dictionary_extra and args.dictionary_extra.strip():
|
|||
|
|
wanted.append(Path(args.dictionary_extra).expanduser())
|
|||
|
|
dictionary = lek.Dictionary.load_all(wanted) if wanted else None
|
|||
|
|
if dictionary is not None:
|
|||
|
|
for path, count in dictionary.sources:
|
|||
|
|
print(f" Dictionary: {count} rule(s) from {path}")
|
|||
|
|
|
|||
|
|
# The synthetic timings are placeholders — every clip asks to start at 0 and
|
|||
|
|
# generate_clips pushes it after the previous one. The real spacing is
|
|||
|
|
# applied after generation, when the durations are known.
|
|||
|
|
blocks, gaps = [], []
|
|||
|
|
for text, ends_para in chunks:
|
|||
|
|
clean = _prepare(text, dictionary)
|
|||
|
|
blocks.append((0.0, 0.0, clean))
|
|||
|
|
gaps.append(args.paragraph_gap if ends_para else args.sentence_gap)
|
|||
|
|
if dictionary is not None and dictionary.hits:
|
|||
|
|
print(f" {dictionary.hits} dictionary replacement(s) applied.")
|
|||
|
|
|
|||
|
|
|
|||
|
|
if args.normalize_numbers:
|
|||
|
|
py = Path(cfg["normalizer_python"]).expanduser()
|
|||
|
|
if not py.is_absolute():
|
|||
|
|
py = Path(lek.__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(lek.__file__).resolve().parent / llm
|
|||
|
|
blocks = lek.normalize_block_numbers(blocks, py, cfg["normalizer_timeout"],
|
|||
|
|
free_vram=cfg["free_comfy_vram"],
|
|||
|
|
llm_dir=llm)
|
|||
|
|
|
|||
|
|
chars = sum(len(t) for _s, _e, t in blocks)
|
|||
|
|
print(f" {len(blocks)} chunk(s), {chars} characters "
|
|||
|
|
f"(longest {max(len(t) for _s, _e, t in blocks)}).")
|
|||
|
|
|
|||
|
|
preview = clips_dir / "tts_input.txt"
|
|||
|
|
preview.write_text("\n\n".join(f"[{i + 1}] {t}" for i, (_s, _e, t) in enumerate(blocks)),
|
|||
|
|
encoding="utf-8")
|
|||
|
|
print(f" Saved to {preview} for inspection.")
|
|||
|
|
|
|||
|
|
if args.dry_run:
|
|||
|
|
for i, (_s, _e, text) in enumerate(blocks, 1):
|
|||
|
|
print(f" [{i:4d}] ({len(text):3d}) {text}")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# ── Step 2: TTS ───────────────────────────────────────────────────────────
|
|||
|
|
if args.rebuild:
|
|||
|
|
clips = [(0.0, path) for path in
|
|||
|
|
sorted(clips_dir.glob("clip_[0-9][0-9][0-9][0-9].flac"))]
|
|||
|
|
if not clips:
|
|||
|
|
sys.exit(f"--rebuild: no clips in {clips_dir}.")
|
|||
|
|
print(f"\n[2/3] Rebuilding from {len(clips)} existing clip(s) — no TTS.")
|
|||
|
|
else:
|
|||
|
|
print(f"\n[2/3] Generating TTS ({len(blocks)} clips, speed {args.speed}×)…")
|
|||
|
|
wav_path = Path(args.voice)
|
|||
|
|
if not wav_path.is_file():
|
|||
|
|
wav_path = lek.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}")
|
|||
|
|
|
|||
|
|
# A cold backend answers the first submission with a connection error,
|
|||
|
|
# which run_workflow would count against the clip's crash budget.
|
|||
|
|
if not lek.wait_for_backend(2.0):
|
|||
|
|
print(" ComfyUI is not up — starting it…")
|
|||
|
|
if not (lek.COMFY_AUTOSTART and lek.start_comfy_backend(lek.COMFY_START_SCRIPT)
|
|||
|
|
and lek.wait_for_backend(lek.BACKEND_HEALTH_TIMEOUT)):
|
|||
|
|
sys.exit(f"ComfyUI did not come up at {lek.COMFY_URL}.")
|
|||
|
|
|
|||
|
|
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 = lek.upload_voice_wav(wav_path)
|
|||
|
|
|
|||
|
|
clips = lek.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)
|
|||
|
|
if cfg["free_comfy_vram"]:
|
|||
|
|
lek.free_comfy_vram()
|
|||
|
|
|
|||
|
|
# ── Step 3: Lay the clips end to end ─────────────────────────────────────
|
|||
|
|
print(f"\n[3/3] Assembling…")
|
|||
|
|
timed, cursor = [], 0.0
|
|||
|
|
for _start, path in clips:
|
|||
|
|
# generate_clips skips a clip that repeatedly killed the backend, so the
|
|||
|
|
# gap belongs to the block the clip came from, not to its position here.
|
|||
|
|
idx = int(re.search(r'(\d+)', path.stem).group(1)) - 1
|
|||
|
|
timed.append((cursor, path))
|
|||
|
|
cursor += lek.get_audio_duration(path)
|
|||
|
|
cursor += gaps[idx] if idx < len(gaps) else args.sentence_gap
|
|||
|
|
|
|||
|
|
assembled = clips_dir / "tts_assembled.flac"
|
|||
|
|
lek.assemble_timed_audio(timed, str(assembled))
|
|||
|
|
|
|||
|
|
final.parent.mkdir(parents=True, exist_ok=True)
|
|||
|
|
if final.suffix.lower() == ".flac":
|
|||
|
|
assembled.replace(final)
|
|||
|
|
else:
|
|||
|
|
lek._run(["ffmpeg", "-y", "-i", str(assembled), "-b:a", "192k", str(final)],
|
|||
|
|
f"Encoding {final.name}")
|
|||
|
|
|
|||
|
|
total = lek.get_audio_duration(final)
|
|||
|
|
print(f"\nDone! Output: {final} ({int(total // 60)}:{int(total % 60):02d}, "
|
|||
|
|
f"{final.stat().st_size // 1024} KB)")
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
main()
|