radio-susan/announce_tracks.py

#!/usr/bin/env python3
"""
Generate a track announcement TTS clip for Radio Susan.
Called by liquidsoap via request.dynamic between tracks.
Reads track_state.json (written by liquidsoap on_track) and
track_history.json (maintained here) to announce what just played.

Outputs an MP3 file path to stdout, or nothing if no announcement needed.
"""

import hashlib
import json
import os
import random
import subprocess
import tempfile
from pathlib import Path

VENV_BIN = Path("/var/lib/radio/venv/bin")
EDGE_TTS = VENV_BIN / "edge-tts"
ANNOUNCE_DIR = Path("/var/lib/radio/announcements")
ANNOUNCE_DIR.mkdir(parents=True, exist_ok=True)
STATE_FILE = Path("/var/lib/radio/track_state.json")
HISTORY_FILE = Path("/var/lib/radio/track_history.json")

VOICES = [
    "en-US-GuyNeural",
    "en-US-ChristopherNeural",
    "en-GB-RyanNeural",
    "en-US-AriaNeural",
    "en-US-JennyNeural",
]

# Station ID variations
STATION_IDS = [
    "Radio Susan, ninety nine point oh.",
    "You're listening to Radio Susan.",
    "Radio Susan.",
    "Susan Radio, always on.",
    "This is Radio Susan.",
]


def pick_voice(text):
    h = int(hashlib.md5(text.encode()).hexdigest(), 16)
    return VOICES[h % len(VOICES)]


def generate_tts(text, outfile, voice=None):
    if voice is None:
        voice = pick_voice(text)
    cmd = [
        str(EDGE_TTS),
        "--voice", voice,
        "--rate", "+10%",
        "--text", text,
        "--write-media", outfile,
    ]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
        return result.returncode == 0
    except subprocess.TimeoutExpired:
        return False


def normalize_audio(infile, outfile):
    cmd = [
        "ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
        "-i", infile,
        "-af", "loudnorm=I=-14:TP=-1:LRA=11,aformat=sample_rates=44100:channel_layouts=stereo",
        "-b:a", "192k",
        outfile,
    ]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
        return result.returncode == 0
    except subprocess.TimeoutExpired:
        return False


def clean_title(s):
    if not s or s == "Unknown":
        return None
    for ext in ('.flac', '.mp3', '.ogg', '.m4a', '.opus', '.wav'):
        s = s.replace(ext, '')
    s = s.strip()
    # Strip leading track numbers like "01 -", "03.", "001 "
    if len(s) > 3 and s[:2].isdigit() and s[2] in '. -':
        s = s[3:].strip()
    if len(s) > 4 and s[:3].isdigit() and s[3] in '. -':
        s = s[4:].strip()
    # Strip "Artist - " prefix if title already contains it
    return s.strip() if s.strip() else None


def fmt_track(entry):
    """Format a history entry as 'Artist - Title'."""
    artist = entry.get("artist", "")
    title = entry.get("title", "")
    if artist and title:
        return f"{artist} - {title}"
    return clean_title(title or artist) or None


def load_history():
    if HISTORY_FILE.exists():
        try:
            return json.loads(HISTORY_FILE.read_text())
        except Exception:
            pass
    return []


def save_history(history):
    HISTORY_FILE.write_text(json.dumps(history[-20:]))


def main():
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--state", action="store_true")
    args = parser.parse_args()

    if not args.state:
        return

    # Read current track from liquidsoap's state file
    current = None
    if STATE_FILE.exists():
        try:
            current = json.loads(STATE_FILE.read_text())
        except Exception:
            return

    if not current:
        return

    history = load_history()

    # Check if this is the same track we already announced (avoid repeats)
    current_key = f"{current.get('artist', '')}|{current.get('title', '')}"
    if history and f"{history[-1].get('artist', '')}|{history[-1].get('title', '')}" == current_key:
        return  # Already announced this one

    # Add current to history
    history.append(current)
    save_history(history)

    # Need at least 2 tracks in history to announce (we announce what JUST played)
    if len(history) < 2:
        return

    # Announce the two most recent tracks before this one.
    # With 2:1 rotation these roughly match what the listener just heard.
    last = fmt_track(history[-2]) if len(history) >= 2 else None
    prev = fmt_track(history[-3]) if len(history) >= 3 else None

    if not last:
        return

    # Build announcement — keep it vague ("recently" not "just played")
    station_id = random.choice(STATION_IDS)

    if prev:
        text = f"You've been listening to {last}, and {prev}. {station_id}"
    else:
        text = f"You've been listening to {last}. {station_id}"

    # Cache based on content hash
    content_hash = hashlib.md5(text.encode()).hexdigest()[:12]
    cached = ANNOUNCE_DIR / f"announce_{content_hash}.mp3"

    if cached.exists():
        print(str(cached))
        return

    # Generate TTS
    with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as tmp:
        tmp_path = tmp.name

    try:
        if not generate_tts(text, tmp_path):
            return
        if normalize_audio(tmp_path, str(cached)):
            print(str(cached))
        else:
            os.rename(tmp_path, str(cached))
            print(str(cached))
    finally:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass

    # Cleanup old announcements (keep last 200)
    announcements = sorted(ANNOUNCE_DIR.glob("announce_*.mp3"), key=lambda f: f.stat().st_mtime)
    for old in announcements[:-200]:
        old.unlink()


if __name__ == "__main__":
    main()