Build the episode
Log project parent
Tech Stack
Log notes
we have the pod manager configured to generate the .ODT source file and the commands for the Ubuntu server. Here is the code for the script that can grab the source file, chunk the text, send it off to eleven labs one chunk at a time, pull them together into a full audio file with the introduction and out roll tunes. the Script places everything in the proper folders and is archived ready for upload to the server
Log Code
#!/usr/bin/env python3
"""
build_episode.py — Wiki Waltz build pipeline (ElevenLabs)
Derived from build_episode_v2.py (OpenAI / ad-system version). Differences:
- No ad system (no [[ad-begin]]/[[ad-end]] handling, no interludes/promos)
- TTS via ElevenLabs eleven_v3 instead of OpenAI tts-1-hd
- Chunker is tag-aware ([voice tag] never split across a chunk boundary)
and emits a second, tag-stripped transcript.txt for the Drupal node
- Final composite is intro + core + outro only (single "podcast" output,
no separate youtube variant)
- Gentle/SRT alignment and YouTube video generation are discontinued
Pipeline:
1. ODT -> source_txt + meta dict (odt_to_source_txt)
2. source_txt -> numbered chunks + transcript.txt (chunk_text)
3. chunks -> per-chunk mp3 via ElevenLabs (generate_tts_audio)
4. chunk mp3s -> merged core mp3 (merge_audio_chunks)
5. intro + core + outro -> final wav (+dup to published) -> mp3
(build_final_composite, convert_to_mp3)
6. metadata.json for the uploader (write_build_metadata)
"""
import os
import re
import json
import time
import glob
import argparse
import unicodedata
import subprocess
from pathlib import Path
from pydub import AudioSegment
from pydub.exceptions import CouldntDecodeError
from nltk.tokenize import sent_tokenize
from elevenlabs.client import ElevenLabs
# ====== CONFIGURATION ======
BASE_PATH = "/path/to/podcast_files"
SENTINEL_BEGIN = "### BEGIN EPISODE"
SENTINEL_END = "### END EPISODE"
META_BEGIN = "[[EPISODE_META_JSON"
META_END = "EPISODE_META_JSON]]"
## VOICE_ID = "uYXf8XasLslADfZ2MB4u" # Hope
VOICE_ID = "<redacted>" # Tiffany
MODEL_ID = "eleven_v3"
OUTPUT_FORMAT = "mp3_44100_128"
TTS_SILENCE_RATIO_MIN = 0.94
TTS_MAX_ATTEMPTS = 3
TTS_RETRY_SLEEP = 5 # seconds between retry attempts
TAG_RE = re.compile(r"\[[^\[\]]*\]")
_client = None
# ====== SMALL HELPERS ======
def safe_mkdir(path):
if not os.path.exists(path):
os.makedirs(path)
def _normalize_eol(text: str) -> str:
return text.replace("\r\n", "\n").replace("\r", "\n")
def clean_feed_text(text: str) -> str:
"""Normalize text for UTF-8 podcast feeds."""
if not text:
return ""
text = unicodedata.normalize("NFKC", text)
replacements = {
"\u2018": "'", "\u2019": "'", "\u201C": '"', "\u201D": '"',
"\u2013": "-", "\u2014": "-", "\u00A0": " ",
}
for bad, good in replacements.items():
text = text.replace(bad, good)
text = re.sub(r"[^\x09\x0A\x0D\x20-\x7E\u00A0-\uFFFF]", "", text)
text = re.sub(r"\s+", " ", text).strip()
return text
def load_podcast_defaults(podcast: str) -> dict:
"""Load per-podcast defaults.json."""
config_path = os.path.join(BASE_PATH, "static-reuse", podcast, "defaults.json")
defaults = {
"feed_id": f"{podcast.lower()}_feed",
"season_number": 1,
"author": "Harmony",
"episode_type": "full",
"explicit": "no",
"link_text": "Wiki Page",
"published": False,
}
if os.path.isfile(config_path):
try:
with open(config_path, "r", encoding="utf-8") as f:
defaults.update(json.load(f))
except Exception as e:
print(f"[WARN] defaults.json unreadable, using built-ins: {e}")
return defaults
def ffprobe_duration_hms(media_path: str) -> str:
"""Return duration as HH:MM:SS using ffprobe."""
if not (media_path and os.path.isfile(media_path)):
return "00:00:00"
try:
result = subprocess.run(
[
"ffprobe", "-v", "error",
"-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1",
media_path,
],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=True,
)
duration = float(result.stdout.strip())
except Exception as e:
print(f"[WARN] ffprobe duration failed: {e}")
return "00:00:00"
total_seconds = int(round(duration))
h, rem = divmod(total_seconds, 3600)
m, s = divmod(rem, 60)
return f"{h:02}:{m:02}:{s:02}"
def _derive_episode_num(odt_name: str) -> str:
m = re.match(r"^(\d{3})-.*\.odt$", odt_name)
if not m:
raise ValueError(f"ODT filename must look like NNN-subject.odt, got: {odt_name}")
return m.group(1)
# ====== 1. INTAKE (ODT -> source text + meta) ======
def extract_episode_meta(full_text: str) -> dict:
s = _normalize_eol(full_text)
start = s.find(META_BEGIN)
if start == -1:
return {}
end = s.find(META_END, start)
if end == -1:
raise RuntimeError(f"Found {META_BEGIN} but not {META_END}")
block = s[start + len(META_BEGIN):end].strip()
m = re.search(r"\{.*\}", block, flags=re.DOTALL)
if not m:
raise RuntimeError("Meta block found but JSON object not detected.")
obj = json.loads(m.group(0))
if not isinstance(obj, dict):
raise RuntimeError("Meta JSON must be an object.")
return obj
def odt_to_source_txt(base_path: str, podcast: str, odt_name: str) -> tuple[str, str, dict]:
"""
Convert unpublished/NNN-subject.odt -> build/NNN/work/NNN.source.txt
Returns (ep, source_txt_path, meta_dict).
"""
ep = _derive_episode_num(odt_name)
odt_path = os.path.join(base_path, podcast, "unpublished", odt_name)
if not os.path.exists(odt_path):
raise FileNotFoundError(f"ODT not found: {odt_path}")
work_dir = os.path.join(base_path, podcast, "build", ep, "work")
chunks_dir = os.path.join(base_path, podcast, "build", ep, "chunks")
os.makedirs(work_dir, exist_ok=True)
os.makedirs(chunks_dir, exist_ok=True)
full_txt = os.path.join(work_dir, f"{ep}.full.txt")
source_txt = os.path.join(work_dir, f"{ep}.source.txt")
subprocess.run(
["pandoc", odt_path, "-t", "plain", "--wrap=none", "-o", full_txt],
check=True,
)
txt = Path(full_txt).read_text(encoding="utf-8")
txt = _normalize_eol(txt)
meta = extract_episode_meta(txt)
on, out_lines = False, []
found_begin, found_end = False, False
for line in txt.split("\n"):
s = line.strip()
if s == SENTINEL_BEGIN:
on = True; found_begin = True; continue
if s == SENTINEL_END:
on = False; found_end = True; continue
if on:
out_lines.append(line)
if not found_begin or not found_end or not out_lines:
raise RuntimeError(
"Episode slice failed. Ensure your ODT contains:\n"
f" {SENTINEL_BEGIN}\n ...episode body...\n {SENTINEL_END}"
)
Path(source_txt).write_text("\n".join(out_lines) + "\n", encoding="utf-8")
return ep, source_txt, meta
# ====== 2. CHUNKING (tag-aware, ElevenLabs v3) ======
def strip_tags(text: str) -> str:
"""Remove ElevenLabs audio tags and collapse resulting whitespace."""
cleaned = TAG_RE.sub("", text)
cleaned = re.sub(r"[ \t]+", " ", cleaned)
cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
cleaned = re.sub(r"\n[ \t]+", "\n", cleaned) # stray leading spaces after a stripped tag
return cleaned.strip()
def _ends_with_orphan_tag(text: str) -> bool:
"""
True if the buffer's trailing content is just a bare tag with nothing
following it yet -- avoid flushing a chunk right there.
"""
trimmed = text.rstrip()
if not trimmed:
return False
last_unit = trimmed.split("\n\n")[-1].strip()
return bool(last_unit) and TAG_RE.sub("", last_unit).strip() == ""
def chunk_text(input_path, output_dir, chunk_size=2000):
"""
Marker-free, tag-aware chunker for ElevenLabs v3.
Emits:
output_dir/000.txt, 001.txt, ... (tags intact -> TTS input)
output_dir/chunks.plan.json (chunk index only)
output_dir/transcript.txt (tags stripped -> Drupal node transcript)
"""
safe_mkdir(output_dir)
with open(input_path, "r", encoding="utf-8") as f:
raw = f.read().replace("\r\n", "\n")
transcript_path = os.path.join(output_dir, "transcript.txt")
with open(transcript_path, "w", encoding="utf-8") as t:
t.write(strip_tags(raw) + "\n")
chunk_idx = 0
plan = []
buf = ""
def flush():
nonlocal buf, chunk_idx
text = buf.strip()
if not text:
return
out_path = os.path.join(output_dir, f"{chunk_idx:03}.txt")
with open(out_path, "w", encoding="utf-8") as out:
out.write(text)
plan.append({"i": chunk_idx})
chunk_idx += 1
buf = ""
paragraphs = raw.split("\n\n")
for para in paragraphs:
para = para.strip()
if not para:
continue
if TAG_RE.fullmatch(para.strip()):
sents = [para]
else:
sents = sent_tokenize(para)
for idx, s in enumerate(sents):
s = s.strip()
if not s:
continue
new_paragraph = (idx == 0)
if not buf:
sep = ""
elif new_paragraph:
sep = "\n\n"
else:
sep = " "
to_add = sep + s
fits = len(buf) + len(to_add) <= chunk_size
if not fits and _ends_with_orphan_tag(buf):
buf += to_add
continue
if fits:
buf += to_add
else:
flush()
buf = s
flush()
plan_path = os.path.join(output_dir, "chunks.plan.json")
with open(plan_path, "w", encoding="utf-8") as fp:
json.dump({"version": 1, "chunks": plan}, fp, ensure_ascii=False, indent=2)
print(f"Chunked into {chunk_idx} pieces (tags intact). Transcript written to {transcript_path}.")
# ====== 3. TTS GENERATION (ElevenLabs) ======
def _get_client():
global _client
if _client is None:
_client = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))
return _client
def _chunk_is_dead_air(audio_path):
"""Same silence/corruption check as before, decoding MP3 instead of WAV."""
try:
audio = AudioSegment.from_file(audio_path, format="mp3")
except CouldntDecodeError as e:
return True, f"[WARN] Corrupt MP3 (ffmpeg could not decode): {e}"
total_seconds = len(audio) // 1000
if total_seconds == 0:
return True, "[WARN] Sub-second file — certainly wrong"
active = sum(
1 for i in range(total_seconds)
if audio[i * 1000:(i + 1) * 1000].rms > 50
)
ratio = active / total_seconds
if ratio < TTS_SILENCE_RATIO_MIN:
return True, f"[WARN] Dead-air detected: {active}/{total_seconds}s active ({ratio:.0%})"
else:
return False, f"[INFO] Clip has full audio: {active}/{total_seconds}s active ({ratio:.0%})"
def generate_tts_audio(chunk_dir, audio_dir, voice_id=VOICE_ID, model_id=MODEL_ID):
"""Generate per-chunk mp3s via ElevenLabs, with retry + dead-air QC."""
client = _get_client()
if not os.path.exists(audio_dir):
os.makedirs(audio_dir)
for filename in sorted(os.listdir(chunk_dir)):
if not filename.endswith(".txt"):
continue
if filename in ("chunks.txt", "transcript.txt"):
continue
base_name = os.path.splitext(filename)[0]
text_path = os.path.join(chunk_dir, filename)
audio_path = os.path.join(audio_dir, base_name + ".mp3")
with open(text_path, "r", encoding="utf-8") as f:
text = f.read().strip()
if not text:
print(f"Skipping empty chunk: {filename}")
continue
print(f"Generating audio for {filename}...")
success = False
for attempt in range(1, TTS_MAX_ATTEMPTS + 1):
if attempt > 1:
print(f" [RETRY] Attempt {attempt} of {TTS_MAX_ATTEMPTS} for {filename}...")
time.sleep(TTS_RETRY_SLEEP)
try:
audio_stream = client.text_to_speech.convert(
text=text,
voice_id=voice_id,
model_id=model_id,
output_format=OUTPUT_FORMAT,
voice_settings={
"stability": 0.2, # roughly maps to "Natural" — try 0.3 for more "Creative" range
"similarity_boost": 0.5,
"style": 0
}
)
with open(audio_path, "wb") as out_file:
for chunk in audio_stream:
if isinstance(chunk, bytes):
out_file.write(chunk)
except Exception as e:
print(f" [ERROR] TTS API call failed for {filename} (attempt {attempt}): {e}")
continue
is_bad, message = _chunk_is_dead_air(audio_path)
print(f" {message}")
if is_bad:
if os.path.exists(audio_path):
os.remove(audio_path)
continue
success = True
break
if not success:
raise RuntimeError(
f"\n[FATAL] Chunk '{filename}' could not be recovered after "
f"{TTS_MAX_ATTEMPTS} attempts.\n"
f"Inspect the source text at: {text_path}\n"
f"Re-run with --skip-tts once the issue is resolved."
)
time.sleep(0.3) # rate limit cushion
print("TTS audio generation complete.")
# ====== 4. MERGE CHUNK AUDIO ======
def merge_audio_chunks(audio_dir, output_path):
"""Concatenate chunk mp3s in numeric order into one core-episode mp3."""
combined = AudioSegment.empty()
mp3s = sorted(glob.glob(os.path.join(audio_dir, "*.mp3")))
for mp3_file in mp3s:
combined += AudioSegment.from_file(mp3_file, format="mp3")
combined.export(output_path, format="mp3")
apply_loudnorm_ffmpeg(output_path)
print(f"Merged audio to {output_path}")
def apply_loudnorm_ffmpeg(input_path):
ext = os.path.splitext(input_path)[1]
temp_path = input_path.replace(ext, f"-norm{ext}")
os.system(
f'ffmpeg -y -i "{input_path}" '
f'-af "loudnorm=I=-22:TP=-1.8:LRA=11,alimiter=limit=0.75" '
f'"{temp_path}" && mv "{temp_path}" "{input_path}"'
)
print("Applied loudness normalization")
# ====== 5. FINAL COMPOSITE (intro + core + outro) ======
def build_final_composite(podcast, episode, core_episode_path, out_dir, published_dir):
"""
Stitch intro + core + outro into the final episode WAV.
Exports WAV to build_dir and duplicates it into published_dir (required by
Pod Manager's outroll-replacement step, which expects a .wav there whose
basename matches the eventual Drupal mp3 filename).
"""
static_dir = os.path.join(BASE_PATH, "static-reuse", podcast)
safe_mkdir(out_dir)
final = AudioSegment.empty()
final += AudioSegment.from_wav(os.path.join(static_dir, "intro.wav"))
final += AudioSegment.from_file(core_episode_path, format="mp3")
final += AudioSegment.from_wav(os.path.join(static_dir, "outro.wav"))
out_wav = os.path.join(out_dir, f"{podcast}-episode-{episode}-podcast.wav")
final.export(out_wav, format="wav")
print(f"Exported final WAV: {out_wav}")
pub_wav = os.path.join(published_dir, f"{podcast}-episode-{episode}-podcast.wav")
final.export(pub_wav, format="wav")
print(f"Exported duplicate WAV to published dir: {pub_wav}")
return out_wav
def convert_to_mp3(input_wav_path):
mp3_path = input_wav_path.replace(".wav", ".mp3")
audio = AudioSegment.from_wav(input_wav_path)
audio.export(mp3_path, format="mp3")
print(f"MP3 saved: {mp3_path}")
return mp3_path
# ====== 6. METADATA ======
def write_build_metadata(podcast: str, episode: str, build_dir: str, meta_from_doc: dict, audio_mp3_path: str) -> str:
"""Create build/<ep>/metadata.json for the uploader."""
defaults = load_podcast_defaults(podcast)
title = clean_feed_text(meta_from_doc.get("title", ""))
field_episode_name = clean_feed_text(meta_from_doc.get("field_episode_name", ""))
subtitle = clean_feed_text(meta_from_doc.get("subtitle", ""))
description = clean_feed_text(meta_from_doc.get("description", ""))
about = clean_feed_text(meta_from_doc.get("about", ""))
keywords_csv = clean_feed_text(meta_from_doc.get("keywords_csv", ""))
subject_type = clean_feed_text(meta_from_doc.get("subject_type", ""))
episode_date = meta_from_doc.get("episode_date", "")
last_page_url = (meta_from_doc.get("last_page_url") or "").strip()
wiki_url = (meta_from_doc.get("wiki_url") or "").strip()
next_page_url = (meta_from_doc.get("next_page_url") or "").strip()
duration_hms = ffprobe_duration_hms(audio_mp3_path)
metadata = {
"title": title,
"field_episode_name": field_episode_name,
"subtitle": subtitle,
"description": description,
"about": about,
"episode_number": int(episode),
"season_number": defaults.get("season_number", 1),
"itunes_keywords": keywords_csv,
"subject_type": subject_type,
"last_page_url": last_page_url,
"wiki_url": wiki_url,
"next_page_url": next_page_url,
"episode_date": episode_date,
"defaults": defaults,
"auto": {
"audio_file": audio_mp3_path,
"duration": duration_hms,
},
"meta_source": "EPISODE_META_JSON",
}
target = os.path.join(build_dir, "metadata.json")
with open(target, "w", encoding="utf-8") as f:
json.dump(metadata, f, indent=2, ensure_ascii=False)
print(f"[OK] Wrote metadata.json: {target}")
return target
# ====== MAIN ======
def main(podcast, odt_filename, skip_tts=False, chunk_size=2000, meta_only=False):
ep, source_txt, meta_from_doc = odt_to_source_txt(BASE_PATH, podcast, odt_filename)
build_dir = os.path.join(BASE_PATH, podcast, "build", ep)
published_dir = os.path.join(BASE_PATH, podcast, "published-files")
chunks_dir = os.path.join(BASE_PATH, podcast, "build", ep, "chunks")
chunk_audio_dir = os.path.join(build_dir, "audio")
core_episode_path = os.path.join(build_dir, f"{podcast}-episode-{ep}-core.mp3")
os.makedirs(chunks_dir, exist_ok=True)
os.makedirs(published_dir, exist_ok=True)
if meta_only:
existing_mp3 = os.path.join(build_dir, f"{podcast}-episode-{ep}-podcast.mp3")
write_build_metadata(podcast, ep, build_dir, meta_from_doc, existing_mp3)
return
if not skip_tts:
chunk_text(source_txt, chunks_dir, chunk_size=chunk_size)
generate_tts_audio(chunks_dir, chunk_audio_dir)
else:
print("Skipping TTS audio generation...")
merge_audio_chunks(chunk_audio_dir, core_episode_path)
final_wav = build_final_composite(podcast, ep, core_episode_path, build_dir, published_dir)
final_mp3 = convert_to_mp3(final_wav)
print("Workflow complete.")
write_build_metadata(
podcast=podcast,
episode=ep,
build_dir=build_dir,
meta_from_doc=meta_from_doc,
audio_mp3_path=final_mp3,
)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("podcast", help="Podcast key / slug (folder under BASE_PATH, matches Pod Manager podcasts.slug)")
parser.add_argument("odt_filename", help="ODT file name (no path), e.g. 010-mosaic.odt")
parser.add_argument("--skip-tts", action="store_true", help="Skip TTS audio generation, reuse existing chunk audio")
parser.add_argument("--chunk-size", type=int, default=2000, help="Max characters per TTS chunk (v3 hard cap is 5000)")
parser.add_argument("--meta-only", action="store_true", help="Only write metadata.json (requires existing podcast MP3)")
args = parser.parse_args()
main(args.podcast, args.odt_filename, skip_tts=args.skip_tts, chunk_size=args.chunk_size, meta_only=args.meta_only)
Log entry date