feat(skills): add modular text-to-audio podcast producer skill
This commit is contained in:
Moritz Graf
+427
@@ -0,0 +1,427 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
text_to_audio.py - OpenRouter Text-to-Speech & Podcast Synthesizer
|
||||
|
||||
Synthesizes single-speaker dossiers or multi-speaker dialog podcasts using OpenRouter TTS
|
||||
(default: deepgram/aura-2) and stitches segments with ffmpeg into an MP3 file.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
DEFAULT_MODEL = "deepgram/aura-2"
|
||||
OPENROUTER_SPEECH_URL = "https://openrouter.ai/api/v1/audio/speech"
|
||||
|
||||
# Speaker voice mapping presets
|
||||
DEFAULT_VOICES = {
|
||||
"de": {
|
||||
"host": "aura-2-julius-de",
|
||||
"cohost": "aura-2-lara-de",
|
||||
"expert": "aura-2-lara-de",
|
||||
"male": "aura-2-julius-de",
|
||||
"female": "aura-2-lara-de",
|
||||
"julius": "aura-2-julius-de",
|
||||
"lara": "aura-2-lara-de",
|
||||
"fabian": "aura-2-fabian-de",
|
||||
"viktoria": "aura-2-viktoria-de",
|
||||
},
|
||||
"en": {
|
||||
"host": "aura-2-zeus-en",
|
||||
"cohost": "aura-2-athena-en",
|
||||
"expert": "aura-2-athena-en",
|
||||
"male": "aura-2-zeus-en",
|
||||
"female": "aura-2-athena-en",
|
||||
"zeus": "aura-2-zeus-en",
|
||||
"athena": "aura-2-athena-en",
|
||||
"apollo": "aura-2-apollo-en",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def find_ffmpeg() -> str:
|
||||
"""Locate the ffmpeg binary across common container and host locations."""
|
||||
env_path = os.environ.get("FFMPEG_PATH")
|
||||
if env_path and os.path.exists(env_path):
|
||||
return env_path
|
||||
|
||||
which_path = shutil.which("ffmpeg")
|
||||
if which_path:
|
||||
return which_path
|
||||
|
||||
candidates = [
|
||||
"/home/node/.openclaw/bin/ffmpeg",
|
||||
"/usr/local/bin/ffmpeg",
|
||||
"/usr/bin/ffmpeg",
|
||||
]
|
||||
for c in candidates:
|
||||
if os.path.exists(c) and os.access(c, os.X_OK):
|
||||
return c
|
||||
|
||||
raise RuntimeError("ffmpeg binary not found. Please ensure ffmpeg is installed or set FFMPEG_PATH.")
|
||||
|
||||
|
||||
def split_text_into_chunks(text: str, max_chars: int = 1500) -> list[str]:
|
||||
"""Split long text into natural sentence-level chunks to satisfy API limits."""
|
||||
text = text.strip()
|
||||
if not text:
|
||||
return []
|
||||
if len(text) <= max_chars:
|
||||
return [text]
|
||||
|
||||
# Split on sentence boundaries
|
||||
sentences = re.split(r"(?<=[.?!;:\n])\s+", text)
|
||||
chunks = []
|
||||
current = []
|
||||
current_len = 0
|
||||
|
||||
for s in sentences:
|
||||
s = s.strip()
|
||||
if not s:
|
||||
continue
|
||||
if current_len + len(s) + 1 > max_chars and current:
|
||||
chunks.append(" ".join(current))
|
||||
current = [s]
|
||||
current_len = len(s)
|
||||
else:
|
||||
current.append(s)
|
||||
current_len += len(s) + 1
|
||||
|
||||
if current:
|
||||
chunks.append(" ".join(current))
|
||||
|
||||
return chunks
|
||||
|
||||
|
||||
def synthesize_speech(
|
||||
text: str,
|
||||
voice: str,
|
||||
output_path: str,
|
||||
model: str = DEFAULT_MODEL,
|
||||
api_key: str | None = None,
|
||||
) -> None:
|
||||
"""Call OpenRouter audio speech API and save MP3 response."""
|
||||
key = api_key or os.environ.get("OPENROUTER_API_KEY")
|
||||
if not key:
|
||||
raise ValueError("OPENROUTER_API_KEY environment variable is not set.")
|
||||
|
||||
payload = {
|
||||
"model": model,
|
||||
"input": text,
|
||||
"voice": voice,
|
||||
"response_format": "mp3",
|
||||
}
|
||||
data = json.dumps(payload).encode("utf-8")
|
||||
req = urllib.request.Request(
|
||||
OPENROUTER_SPEECH_URL,
|
||||
data=data,
|
||||
headers={
|
||||
"Authorization": f"Bearer {key}",
|
||||
"Content-Type": "application/json",
|
||||
"User-Agent": "OpenClaw-AudioDossier/1.0",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=60) as resp:
|
||||
content = resp.read()
|
||||
with open(output_path, "wb") as f:
|
||||
f.write(content)
|
||||
except urllib.error.HTTPError as e:
|
||||
body = e.read().decode("utf-8", errors="replace")
|
||||
raise RuntimeError(f"OpenRouter TTS API error {e.code}: {body}") from e
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"TTS request failed: {e}") from e
|
||||
|
||||
|
||||
def generate_silence(ffmpeg_bin: str, duration_sec: float, output_path: str) -> None:
|
||||
"""Generate a short silence MP3 segment for natural conversational pauses."""
|
||||
cmd = [
|
||||
ffmpeg_bin,
|
||||
"-y",
|
||||
"-f",
|
||||
"lavfi",
|
||||
"-i",
|
||||
f"anullsrc=r=24000:cl=mono",
|
||||
"-t",
|
||||
str(duration_sec),
|
||||
"-c:a",
|
||||
"libmp3lame",
|
||||
"-b:a",
|
||||
"48k",
|
||||
output_path,
|
||||
]
|
||||
res = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
|
||||
if res.returncode != 0:
|
||||
raise RuntimeError(f"Failed to generate silence with ffmpeg: {res.stderr}")
|
||||
|
||||
|
||||
def stitch_audio_segments(
|
||||
ffmpeg_bin: str,
|
||||
audio_files: list[str],
|
||||
output_file: str,
|
||||
title: str = "Audio Dossier",
|
||||
artist: str = "OpenClaw",
|
||||
album: str = "OpenClaw Audio Briefings",
|
||||
) -> None:
|
||||
"""Concatenate audio segments and apply ID3 tags."""
|
||||
if not audio_files:
|
||||
raise ValueError("No audio segments to stitch.")
|
||||
|
||||
if len(audio_files) == 1:
|
||||
# Just copy and add metadata
|
||||
cmd = [
|
||||
ffmpeg_bin,
|
||||
"-y",
|
||||
"-i",
|
||||
audio_files[0],
|
||||
"-metadata",
|
||||
f"title={title}",
|
||||
"-metadata",
|
||||
f"artist={artist}",
|
||||
"-metadata",
|
||||
f"album={album}",
|
||||
"-c:a",
|
||||
"copy",
|
||||
output_file,
|
||||
]
|
||||
res = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
|
||||
if res.returncode != 0:
|
||||
# Fallback to re-encoding if copy fails
|
||||
cmd_reencode = [
|
||||
ffmpeg_bin,
|
||||
"-y",
|
||||
"-i",
|
||||
audio_files[0],
|
||||
"-metadata",
|
||||
f"title={title}",
|
||||
"-metadata",
|
||||
f"artist={artist}",
|
||||
"-metadata",
|
||||
f"album={album}",
|
||||
output_file,
|
||||
]
|
||||
res_re = subprocess.run(cmd_reencode, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
|
||||
if res_re.returncode != 0:
|
||||
raise RuntimeError(f"FFmpeg error: {res_re.stderr}")
|
||||
return
|
||||
|
||||
# Multi-file concat using concat demuxer
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as list_file:
|
||||
for f in audio_files:
|
||||
safe_path = f.replace("'", "'\\''")
|
||||
list_file.write(f"file '{safe_path}'\n")
|
||||
list_file_path = list_file.name
|
||||
|
||||
try:
|
||||
cmd = [
|
||||
ffmpeg_bin,
|
||||
"-y",
|
||||
"-f",
|
||||
"concat",
|
||||
"-safe",
|
||||
"0",
|
||||
"-i",
|
||||
list_file_path,
|
||||
"-metadata",
|
||||
f"title={title}",
|
||||
"-metadata",
|
||||
f"artist={artist}",
|
||||
"-metadata",
|
||||
f"album={album}",
|
||||
"-c:a",
|
||||
"libmp3lame",
|
||||
"-b:a",
|
||||
"64k",
|
||||
output_file,
|
||||
]
|
||||
res = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True)
|
||||
if res.returncode != 0:
|
||||
raise RuntimeError(f"FFmpeg concatenation failed: {res.stderr}")
|
||||
finally:
|
||||
if os.path.exists(list_file_path):
|
||||
os.remove(list_file_path)
|
||||
|
||||
|
||||
def resolve_voice(speaker_name: str, lang: str = "de", explicit_voice: str | None = None) -> str:
|
||||
"""Resolve a speaker name or role into a concrete OpenRouter voice ID."""
|
||||
if explicit_voice:
|
||||
return explicit_voice
|
||||
|
||||
lang_map = DEFAULT_VOICES.get(lang, DEFAULT_VOICES["de"])
|
||||
clean_name = speaker_name.lower().strip()
|
||||
|
||||
if clean_name in lang_map:
|
||||
return lang_map[clean_name]
|
||||
|
||||
if clean_name.startswith("aura-"):
|
||||
return clean_name
|
||||
|
||||
if any(k in clean_name for k in ["host2", "cohost", "female", "lara", "expert", "guest"]):
|
||||
return lang_map.get("cohost", "aura-2-lara-de")
|
||||
|
||||
return lang_map.get("host", "aura-2-julius-de")
|
||||
|
||||
|
||||
def process_dialog(
|
||||
dialog_items: list[dict],
|
||||
ffmpeg_bin: str,
|
||||
lang: str,
|
||||
model: str,
|
||||
output_file: str,
|
||||
title: str,
|
||||
pause_duration: float = 0.35,
|
||||
) -> None:
|
||||
"""Generate audio for structured multi-speaker dialog."""
|
||||
with tempfile.TemporaryDirectory(prefix="audio_dossier_") as tmpdir:
|
||||
silence_file = os.path.join(tmpdir, "pause.mp3")
|
||||
generate_silence(ffmpeg_bin, pause_duration, silence_file)
|
||||
|
||||
segment_files = []
|
||||
seg_idx = 0
|
||||
|
||||
for item in dialog_items:
|
||||
speaker = item.get("speaker", "host")
|
||||
text = item.get("text", "").strip()
|
||||
if not text:
|
||||
continue
|
||||
|
||||
voice = resolve_voice(speaker, lang=lang, explicit_voice=item.get("voice"))
|
||||
chunks = split_text_into_chunks(text)
|
||||
|
||||
for chunk in chunks:
|
||||
chunk_file = os.path.join(tmpdir, f"seg_{seg_idx:04d}.mp3")
|
||||
synthesize_speech(chunk, voice=voice, output_path=chunk_file, model=model)
|
||||
segment_files.append(chunk_file)
|
||||
seg_idx += 1
|
||||
|
||||
# Append silence between speaker turns
|
||||
segment_files.append(silence_file)
|
||||
|
||||
if segment_files and segment_files[-1] == silence_file:
|
||||
segment_files.pop()
|
||||
|
||||
stitch_audio_segments(
|
||||
ffmpeg_bin=ffmpeg_bin,
|
||||
audio_files=segment_files,
|
||||
output_file=output_file,
|
||||
title=title,
|
||||
)
|
||||
|
||||
|
||||
def process_solo_text(
|
||||
text: str,
|
||||
ffmpeg_bin: str,
|
||||
voice: str,
|
||||
model: str,
|
||||
output_file: str,
|
||||
title: str,
|
||||
) -> None:
|
||||
"""Generate audio for a single speaker text."""
|
||||
chunks = split_text_into_chunks(text)
|
||||
if not chunks:
|
||||
raise ValueError("No text provided to synthesize.")
|
||||
|
||||
with tempfile.TemporaryDirectory(prefix="audio_solo_") as tmpdir:
|
||||
segment_files = []
|
||||
for i, chunk in enumerate(chunks):
|
||||
chunk_file = os.path.join(tmpdir, f"chunk_{i:04d}.mp3")
|
||||
synthesize_speech(chunk, voice=voice, output_path=chunk_file, model=model)
|
||||
segment_files.append(chunk_file)
|
||||
|
||||
stitch_audio_segments(
|
||||
ffmpeg_bin=ffmpeg_bin,
|
||||
audio_files=segment_files,
|
||||
output_file=output_file,
|
||||
title=title,
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="OpenClaw Text-to-Audio / Podcast Generator")
|
||||
parser.add_argument("--text", help="Plaintext to synthesize (solo mode)")
|
||||
parser.add_argument("--file", help="Path to plaintext file (solo mode)")
|
||||
parser.add_argument("--dialog-json", help="Raw JSON string containing dialog array: [{'speaker':'...','text':'...'}]")
|
||||
parser.add_argument("--dialog-file", help="Path to JSON file containing dialog array")
|
||||
parser.add_argument("--output", "-o", help="Target output MP3 path")
|
||||
parser.add_argument("--title", default="Audio Dossier", help="Audio title for ID3 metadata")
|
||||
parser.add_argument("--voice", help="Primary speaker voice (e.g. aura-2-julius-de)")
|
||||
parser.add_argument("--co-voice", help="Secondary speaker voice for dialog (e.g. aura-2-lara-de)")
|
||||
parser.add_argument("--lang", default="de", choices=["de", "en"], help="Language preset (de/en)")
|
||||
parser.add_argument("--model", default=DEFAULT_MODEL, help="OpenRouter speech model")
|
||||
parser.add_argument("--pause", type=float, default=0.35, help="Pause in seconds between dialog turns")
|
||||
parser.add_argument("--media-tag", action="store_true", default=True, help="Print MEDIA:<path> tag to stdout")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
ffmpeg_bin = find_ffmpeg()
|
||||
|
||||
# Determine output path
|
||||
if args.output:
|
||||
out_path = Path(args.output).expanduser().resolve()
|
||||
else:
|
||||
import time
|
||||
ts = int(time.time())
|
||||
out_path = Path(f"/tmp/audio-dossier-{ts}.mp3")
|
||||
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 1. Dialog mode via file or json string
|
||||
dialog_data = None
|
||||
if args.dialog_file:
|
||||
with open(args.dialog_file, "r", encoding="utf-8") as f:
|
||||
dialog_data = json.load(f)
|
||||
elif args.dialog_json:
|
||||
dialog_data = json.loads(args.dialog_json)
|
||||
|
||||
if dialog_data:
|
||||
if not isinstance(dialog_data, list):
|
||||
raise ValueError("Dialog input must be a JSON array of objects with 'speaker' and 'text'.")
|
||||
process_dialog(
|
||||
dialog_items=dialog_data,
|
||||
ffmpeg_bin=ffmpeg_bin,
|
||||
lang=args.lang,
|
||||
model=args.model,
|
||||
output_file=str(out_path),
|
||||
title=args.title,
|
||||
pause_duration=args.pause,
|
||||
)
|
||||
else:
|
||||
# 2. Solo text mode
|
||||
raw_text = ""
|
||||
if args.file:
|
||||
with open(args.file, "r", encoding="utf-8") as f:
|
||||
raw_text = f.read()
|
||||
elif args.text:
|
||||
raw_text = args.text
|
||||
elif not sys.stdin.isatty():
|
||||
raw_text = sys.stdin.read()
|
||||
else:
|
||||
parser.error("Must provide either --text, --file, --dialog-json, --dialog-file, or pipe text to stdin.")
|
||||
|
||||
primary_voice = args.voice or resolve_voice("host", lang=args.lang)
|
||||
process_solo_text(
|
||||
text=raw_text,
|
||||
ffmpeg_bin=ffmpeg_bin,
|
||||
voice=primary_voice,
|
||||
model=args.model,
|
||||
output_file=str(out_path),
|
||||
title=args.title,
|
||||
)
|
||||
|
||||
print(f"Generated audio: {out_path} ({out_path.stat().st_size} bytes)")
|
||||
if args.media_tag:
|
||||
print(f"MEDIA:{out_path}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user