Dispy-P3BOT/radio_bot.py

#!/usr/bin/env python3
"""One-shot Discord netradio relay.

Connects to a Discord voice channel, streams a netradio source into it for a
configured number of seconds, then disconnects and exits. Designed to be fired
by cron (e.g. "transmit for 5 minutes every day at 6 pm").

Configuration lives in ./radioconfig.json (see radioconfig.json.template).
The bot token may live in the config as "token" or, preferably, in the
DISCORD_BOT_TOKEN environment variable (which takes precedence).
"""

import asyncio
import json
import logging
import os
import sys
from pathlib import Path

import discord

CONFIG_PATH = Path(__file__).resolve().parent / "radioconfig.json"

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger("radio_bot")

# Reconnect/keep-alive flags for ffmpeg so a brief network hiccup on the
# netradio stream doesn't kill the whole broadcast.
FFMPEG_BEFORE_OPTIONS = (
    "-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5"
)
FFMPEG_OPTIONS = "-vn"


def load_config() -> dict:
    if not CONFIG_PATH.exists():
        log.error(
            "Config not found at %s. Copy radioconfig.json.template to "
            "radioconfig.json and fill it in.",
            CONFIG_PATH,
        )
        sys.exit(1)

    try:
        with CONFIG_PATH.open(encoding="utf-8") as fh:
            config = json.load(fh)
    except (json.JSONDecodeError, OSError) as exc:
        log.error("Failed to read %s: %s", CONFIG_PATH, exc)
        sys.exit(1)

    # Token: env var wins, config is the fallback.
    token = os.environ.get("DISCORD_BOT_TOKEN") or config.get("token")
    if not token:
        log.error(
            "No bot token. Set DISCORD_BOT_TOKEN or add \"token\" to %s.",
            CONFIG_PATH,
        )
        sys.exit(1)
    config["token"] = token

    missing = [k for k in ("netradio_source", "channel_id") if not config.get(k)]
    if missing:
        log.error("Missing required config keys: %s", ", ".join(missing))
        sys.exit(1)

    try:
        config["channel_id"] = int(config["channel_id"])
    except (TypeError, ValueError):
        log.error("channel_id must be a numeric Discord channel ID.")
        sys.exit(1)

    runtime = config.get("runtime_seconds", 300)
    try:
        config["runtime_seconds"] = int(runtime)
    except (TypeError, ValueError):
        log.error("runtime_seconds must be an integer number of seconds.")
        sys.exit(1)
    if config["runtime_seconds"] <= 0:
        log.error("runtime_seconds must be greater than 0.")
        sys.exit(1)

    return config


class RadioBot(discord.Client):
    def __init__(self, config: dict) -> None:
        super().__init__(intents=discord.Intents.default())
        self.config = config
        self._exit_code = 0

    async def on_ready(self) -> None:
        log.info("Logged in as %s (id=%s)", self.user, self.user.id)
        try:
            await self.broadcast()
        except Exception:  # noqa: BLE001 - log any failure, still exit cleanly
            log.exception("Broadcast failed")
            self._exit_code = 1
        finally:
            await self.close()

    async def broadcast(self) -> None:
        channel = self.get_channel(self.config["channel_id"])
        if channel is None:
            channel = await self.fetch_channel(self.config["channel_id"])

        if not isinstance(channel, discord.VoiceChannel) and not isinstance(
            channel, discord.StageChannel
        ):
            raise RuntimeError(
                f"Channel {self.config['channel_id']} is not a voice/stage "
                f"channel (got {type(channel).__name__})."
            )

        source_url = self.config["netradio_source"]
        runtime = self.config["runtime_seconds"]

        log.info("Connecting to voice channel #%s", channel.name)
        voice = await channel.connect()

        # Stage channels put bots in the audience by default; try to move to
        # speaker so audio is actually heard.
        if isinstance(channel, discord.StageChannel):
            try:
                await self.user_to_speaker(channel)
            except Exception:  # noqa: BLE001
                log.warning("Could not become stage speaker; audio may be muted.")

        try:
            audio = discord.FFmpegPCMAudio(
                source_url,
                before_options=FFMPEG_BEFORE_OPTIONS,
                options=FFMPEG_OPTIONS,
            )
            log.info("Streaming %s for %s seconds", source_url, runtime)
            voice.play(audio)

            # Wait for the runtime, but stop early if the stream ends on its own.
            elapsed = 0
            while voice.is_playing() and elapsed < runtime:
                await asyncio.sleep(1)
                elapsed += 1

            log.info("Broadcast finished after %s seconds", elapsed)
        finally:
            if voice.is_playing():
                voice.stop()
            await voice.disconnect()

    async def user_to_speaker(self, channel: discord.StageChannel) -> None:
        me = channel.guild.me
        await me.edit(suppress=False)


def main() -> None:
    config = load_config()
    bot = RadioBot(config)
    try:
        bot.run(config["token"], log_handler=None)
    except discord.LoginFailure:
        log.error("Login failed: invalid bot token.")
        sys.exit(1)
    sys.exit(bot._exit_code)


if __name__ == "__main__":
    main()