radio/scripts/validate_config.py

#!/usr/bin/env python3
"""Validate all configuration files for the local radio station."""

import os
import sys
from pathlib import Path

import yaml

BASE_DIR = Path(os.environ.get("LOCALRADIO_BASE", "/opt/localradio"))
errors = []
warnings = []


def err(msg):
    errors.append(msg)
    print(f"  ERROR: {msg}")


def warn(msg):
    warnings.append(msg)
    print(f"  WARN:  {msg}")


def ok(msg):
    print(f"  OK:    {msg}")


def validate_feeds():
    print("=== Validating feeds.yaml ===")
    path = BASE_DIR / "config" / "feeds.yaml"
    if not path.exists():
        err(f"Missing: {path}")
        return

    with open(path) as f:
        data = yaml.safe_load(f)

    if not data or "feeds" not in data:
        err("feeds.yaml must have a 'feeds' key")
        return

    feeds = data["feeds"]
    enabled_count = 0
    for i, feed in enumerate(feeds):
        if not feed.get("name"):
            err(f"Feed [{i}]: missing 'name'")
        if not feed.get("url"):
            err(f"Feed [{i}]: missing 'url'")
        elif not feed["url"].startswith(("http://", "https://")):
            err(f"Feed '{feed.get('name', i)}': URL must start with http:// or https://")
        if feed.get("enabled", True):
            enabled_count += 1

    if enabled_count == 0:
        warn("No feeds are enabled")
    else:
        ok(f"{enabled_count} feed(s) enabled")


def validate_station():
    print("\n=== Validating station.yaml ===")
    path = BASE_DIR / "config" / "station.yaml"
    if not path.exists():
        err(f"Missing: {path}")
        return

    with open(path) as f:
        data = yaml.safe_load(f)

    if not data.get("schedule"):
        err("station.yaml must have a 'schedule' section")
        return

    for block in data["schedule"]:
        name = block.get("name", "unnamed")
        if not block.get("folder"):
            err(f"Schedule '{name}': missing 'folder'")
        if not block.get("days"):
            err(f"Schedule '{name}': missing 'days'")
        if not block.get("start") or not block.get("end"):
            err(f"Schedule '{name}': missing 'start' or 'end'")

    ok(f"{len(data['schedule'])} schedule block(s) defined")

    fmt = data.get("audio", {}).get("format", "mp3")
    bitrate = data.get("audio", {}).get("bitrate", 320)
    ok(f"Audio: {fmt} @ {bitrate} kbps")


def validate_directories():
    print("\n=== Validating directories ===")
    required = [
        "config",
        "scripts",
        "state",
        "state/queue",
        "media/music/morning",
        "media/music/day",
        "media/music/night",
        "media/music/weekend",
        "media/podcasts",
        "media/fallback",
        "logs",
    ]
    for d in required:
        full = BASE_DIR / d
        if full.exists():
            ok(f"{d}/")
        else:
            err(f"Missing directory: {d}/")

    # Check fallback has content
    fallback = BASE_DIR / "media" / "fallback"
    if fallback.exists():
        audio_exts = {".mp3", ".flac", ".ogg", ".opus", ".m4a", ".wav"}
        files = [f for f in fallback.iterdir() if f.suffix.lower() in audio_exts]
        if not files:
            warn("media/fallback/ has no audio files — stream may go silent!")
        else:
            ok(f"media/fallback/ has {len(files)} audio file(s)")


def validate_icecast():
    print("\n=== Validating icecast.xml ===")
    path = BASE_DIR / "config" / "icecast.xml"
    if not path.exists():
        err(f"Missing: {path}")
        return

    content = path.read_text()
    if "localradio_source" in content:
        warn("icecast.xml still uses default source password — change it for security")
    if "localradio_admin" in content:
        warn("icecast.xml still uses default admin password — change it for security")
    ok("icecast.xml exists and is parseable")


def validate_liquidsoap():
    print("\n=== Validating station.liq ===")
    path = BASE_DIR / "config" / "station.liq"
    if not path.exists():
        err(f"Missing: {path}")
        return
    ok("station.liq exists")


def main():
    print(f"Validating configuration at: {BASE_DIR}\n")
    validate_feeds()
    validate_station()
    validate_directories()
    validate_icecast()
    validate_liquidsoap()

    print("\n" + "=" * 50)
    if errors:
        print(f"RESULT: {len(errors)} error(s), {len(warnings)} warning(s)")
        sys.exit(1)
    elif warnings:
        print(f"RESULT: OK with {len(warnings)} warning(s)")
    else:
        print("RESULT: All checks passed")


if __name__ == "__main__":
    main()