import asyncio import time from typing import Optional import discord from discord import app_commands from discord.ext import commands from app.internal.c2_client import c2 from app.internal.logger import logger # How long /join keeps editing its reply while the node reports back. JOIN_WATCH_SECONDS = 30 JOIN_POLL_SECONDS = 2.5 OVERVIEW_TTL = 30.0 STATE_TEXT = { "pending": "Waiting for the node to pick it up…", "connecting": "Connecting…", "connected": "Connected.", "node_offline": "The node carrying this channel is offline.", "paused_locally": "Paused from the node's local dashboard.", } def _stream_name(s: dict) -> str: """How one radio stream is named to a Discord user. Several streams can share a voice channel, so the system and the bot are what tell them apart.""" bot = f" ({s['token_name']})" if s.get("token_name") else "" return f"{s.get('system_name') or s.get('system_id') or 'radio'}{bot}" def _refusal_text(body: dict) -> str: text = body.get("detail") or "DRB could not start radio in that channel." if isinstance(text, dict): # FastAPI HTTPException detail text = text.get("detail") or str(text) invites = [i for i in (body.get("invite") or []) if i.get("invite_url")] if invites: text += "\n" + "\n".join(f"- [Invite {i['name']}]({i['invite_url']})" for i in invites[:5]) return text class RadioCommands(commands.Cog): """Radio channels for ONE Discord server at a time (server-26#203). Every C2 call carries this interaction's guild_id; C2 resolves the org that server is linked to and never returns or touches anything outside it. /join and /leave default to members with Manage Channels — a server admin can hand them to a DJ role under Server Settings > Integrations. A voice channel can carry SEVERAL radio streams, each its own bot playing its own system (server-26#213 follow-up). So: * /join adds a stream for the chosen system; it never replaces another system's stream in the channel. The same system again is a no-op. * /leave ``stream:`` stops exactly one (autocomplete lists this server's). * /leave ``channel:`` stops that channel's stream; if it has several it lists them and asks which (``stream:``) — or ``all:True`` for every one. * /leave with nothing stops every stream in this server (unchanged). """ def __init__(self, bot: commands.Bot): self.bot = bot self._overview: dict[int, tuple[float, dict]] = {} async def _guild(self, guild_id: int, fresh: bool = False) -> dict: hit = self._overview.get(guild_id) if hit and not fresh and time.monotonic() - hit[0] < OVERVIEW_TTL: return hit[1] data = await c2.guild_overview(guild_id) if not data.get("error"): self._overview[guild_id] = (time.monotonic(), data) return data # ------------------------------------------------------------------ # Autocomplete — this server's org only # ------------------------------------------------------------------ async def system_autocomplete( self, interaction: discord.Interaction, current: str ) -> list[app_commands.Choice[str]]: if not interaction.guild_id: return [] data = await self._guild(interaction.guild_id) return [ app_commands.Choice(name=s["name"], value=s["system_id"]) for s in data.get("systems", []) if s.get("name") and current.lower() in s["name"].lower() ][:25] async def bot_autocomplete( self, interaction: discord.Interaction, current: str ) -> list[app_commands.Choice[str]]: if not interaction.guild_id: return [] data = await self._guild(interaction.guild_id) out = [] for b in data.get("bots", []): name = b.get("name") or b.get("token_id") if current.lower() not in name.lower(): continue if b.get("busy_here"): label = "busy here" elif b.get("member") is False: label = "not in this server" else: label = "free" out.append(app_commands.Choice(name=f"{name} ({label})", value=b["token_id"])) return out[:25] async def stream_autocomplete( self, interaction: discord.Interaction, current: str ) -> list[app_commands.Choice[str]]: if not interaction.guild_id: return [] data = await self._guild(interaction.guild_id) out = [] for s in data.get("streams", []): ch = interaction.guild.get_channel(int(s["channel_id"])) if interaction.guild else None where = f"#{ch.name}" if ch else (s.get("channel_name") or f"channel {s['channel_id']}") name = f"{where} — {_stream_name(s)}" if current.lower() in name.lower(): out.append(app_commands.Choice(name=name[:100], value=s["stream_id"])) return out[:25] # ------------------------------------------------------------------ # /join # ------------------------------------------------------------------ @app_commands.command(name="join", description="Stream a radio system to your voice channel.") @app_commands.guild_only() @app_commands.default_permissions(manage_channels=True) @app_commands.describe( system="The radio system to listen to.", bot="Optionally pick which radio bot joins.", persistent="Keep this channel on radio until someone runs /leave (reconnects by itself).", ) @app_commands.autocomplete(system=system_autocomplete, bot=bot_autocomplete) async def join(self, interaction: discord.Interaction, system: str, bot: Optional[str] = None, persistent: bool = False): await interaction.response.defer(ephemeral=True) voice = getattr(interaction.user, "voice", None) if not voice or not voice.channel: await interaction.followup.send("You need to be in a voice channel first.", ephemeral=True) return channel = voice.channel ok, body = await c2.start_stream({ "guild_id": str(interaction.guild_id), "channel_id": str(channel.id), "system_id": system, "mode": "persistent" if persistent else "adhoc", "preferred_token_id": bot, "channel_name": channel.name, "guild_name": interaction.guild.name if interaction.guild else None, "created_by": str(interaction.user.id), }) self._overview.pop(interaction.guild_id, None) if not ok: await interaction.followup.send(_refusal_text(body), ephemeral=True) return system_name = body.get("system_name") or system head = f"**{system_name}** → {channel.mention}" if body.get("already_on"): head += "\nThat system is already playing in this channel — nothing was replaced." if body.get("delivered") is False: head += "\nDRB's message broker is down; the node will get this as soon as it is back." msg = await interaction.followup.send(f"{head}\n{STATE_TEXT['pending']}", ephemeral=True, wait=True) # The node reports what actually happened (node-26#25 item 5) — show # that, not an optimistic "Streaming…" (server-26#203 item 4). await self._watch(msg, body["stream_id"], head) async def _watch(self, msg: discord.WebhookMessage, stream_id: str, head: str) -> None: last = None deadline = time.monotonic() + JOIN_WATCH_SECONDS while time.monotonic() < deadline: await asyncio.sleep(JOIN_POLL_SECONDS) s = await c2.get_stream(stream_id) if not s: continue actual = s.get("actual") or {} state = actual.get("state") or "pending" text = actual.get("error") if state == "error" else STATE_TEXT.get(state, state) if text != last: last = text try: await msg.edit(content=f"{head}\n{text}") except discord.HTTPException as e: logger.warning(f"/join status edit failed: {e}") return if state in ("connected", "error"): return if last == STATE_TEXT["pending"]: try: await msg.edit(content=f"{head}\nThe node has not answered yet — it may be offline. " "Check `/status`.") except discord.HTTPException: pass # ------------------------------------------------------------------ # /leave — this server only # ------------------------------------------------------------------ @app_commands.command(name="leave", description="Stop radio in this server, one channel, or one stream.") @app_commands.guild_only() @app_commands.default_permissions(manage_channels=True) @app_commands.describe( channel="Only stop radio in this voice channel.", stream="Stop just this radio stream (a channel can carry several).", every="With channel: stop every stream in it without asking.", ) @app_commands.rename(every="all") @app_commands.autocomplete(stream=stream_autocomplete) async def leave(self, interaction: discord.Interaction, channel: Optional[discord.VoiceChannel] = None, stream: Optional[str] = None, every: bool = False): await interaction.response.defer(ephemeral=True) if channel and not stream and not every: # Several streams in one channel: say which, never guess. data = await self._guild(interaction.guild_id, fresh=True) here = [s for s in data.get("streams", []) if str(s.get("channel_id")) == str(channel.id)] if len(here) > 1: names = "\n".join(f"- {_stream_name(s)}" for s in here) await interaction.followup.send( f"{channel.mention} has {len(here)} radio streams:\n{names}\n" "Run `/leave stream:` and pick one, or `/leave channel:` with `all:True` to stop them all.", ephemeral=True) return stopped = await c2.stop_guild(interaction.guild_id, str(channel.id) if channel and not stream else None, stream_id=stream) self._overview.pop(interaction.guild_id, None) if stopped is None: await interaction.followup.send("Could not reach DRB. Try again in a minute.", ephemeral=True) elif not stopped: where = "that stream" if stream else (channel.mention if channel else "this server") await interaction.followup.send(f"No radio is playing in {where}.", ephemeral=True) else: await interaction.followup.send(f"Stopped {len(stopped)} radio stream(s).", ephemeral=True) # ------------------------------------------------------------------ # /status — this server's org only # ------------------------------------------------------------------ @app_commands.command(name="status", description="Show this server's radio channels and your nodes.") @app_commands.guild_only() async def status(self, interaction: discord.Interaction): await interaction.response.defer(ephemeral=True) data = await self._guild(interaction.guild_id, fresh=True) embed = discord.Embed(title="DRB status", color=0x2b2d31) if data.get("error"): embed.description = "Could not reach DRB." elif not data.get("org_id"): embed.description = "This server is not linked to a DRB organization yet." else: systems = {s["system_id"]: s.get("name") for s in data.get("systems", [])} lines = [] for s in data.get("streams", []): actual = (s.get("actual") or {}) state = actual.get("state") or "pending" lines.append(f"<#{s['channel_id']}> — {_stream_name(s)} · `{state}`" + (" · persistent" if s.get("mode") == "persistent" else "")) embed.add_field(name="Radio channels", value="\n".join(lines) or "None playing.", inline=False) dot = {"online": "🟢", "recording": "🔴", "offline": "⚫", "unconfigured": "🟡"} for n in sorted(data.get("nodes", []), key=lambda n: n.get("name") or n["node_id"])[:20]: st = n.get("status") or "offline" embed.add_field( name=f"{dot.get(st, '⚪')} {n.get('name') or n['node_id']}", value=f"`{st}` — {systems.get(n.get('system_id'), 'Unassigned')}", inline=True, ) await interaction.followup.send(embed=embed, ephemeral=True) # ------------------------------------------------------------------ # /help # ------------------------------------------------------------------ @app_commands.command(name="help", description="How to use this bot.") async def help(self, interaction: discord.Interaction): embed = discord.Embed(title="DRB Radio Bot — Help", color=0x5865f2) embed.add_field( name="/drb link `code`", value=( "Link this server to your DRB organization — needed before anything below works. " "An organization owner gets the code in DRB under Settings → Organization → Discord servers; " "someone with Manage Server runs this.\n" "Or skip this bot entirely: **Connect a Discord server** on that same page adds a radio bot " "and links the server in one step, and radio channels can be started from a node's page in DRB." ), inline=False, ) embed.add_field( name="/join `system` [`bot`] [`persistent`]", value=( "Put a radio system on your current voice channel. A radio bot joins and plays it.\n" "Run it again with another system to add a second stream (its own bot) to the same channel — " "it never replaces one that is playing. A bot can be in one voice channel per server.\n" "`persistent` keeps the channel on radio until `/leave`, reconnecting by itself.\n" "Needs Manage Channels unless a server admin grants it to another role." ), inline=False, ) embed.add_field( name="/leave [`channel`] [`stream`] [`all`]", value=("Stop radio in this server, one voice channel, or one `stream`. If a channel has several " "streams, `/leave channel:` asks which (or add `all:True`). Never affects other servers."), inline=False, ) embed.add_field( name="/status", value="This server's radio channels and your organization's nodes.", inline=False, ) embed.add_field( name="Radio bots", value=( "A radio bot has to be a member of this server before it can join a channel — " "if none is, `/join` replies with an invite link. It needs View Channel, Connect and Speak " "there; if it can't, `/join` says which permission is missing and where to allow it.\n" "Radio bots work without this bot: it only adds these commands.\n" "The bot's green speaking ring lights up only while radio is transmitting." ), inline=False, ) embed.set_footer(text="Audio comes live from DRB edge nodes' receivers, straight from their audio " "path — not a re-broadcast stream.") await interaction.response.send_message(embed=embed, ephemeral=True) async def setup(bot: commands.Bot): await bot.add_cog(RadioCommands(bot))