Files
server-26/drb-server-discord-bot/app/commands/radio.py
T
Logan CusanoandClaude Opus 5.5 9c124d4c8b S8: main DRB bot optional end to end + owner-flow integration tests (server-26#225)
- Copy: picker, admin Servers/Streams/How-it-works, org Discord card, bot
  /help and the server-limit refusal no longer imply the DRB server bot is
  required; a radio bot alone is enough.
- Admin overview: per-server main_bot_present, radio_bots, linked_via,
  directory_source; Servers tab gets a Bots column.
- discord_web docstring: states the reversed "never calls Discord" rule
  (plan finding 5) and the optional server bot.
- Server bot directory cog: re-push (which drops C2's cached permission
  answers) on overwrite changes, role permission changes and role deletes.
- tests/test_discord_owner_flows.py: one-bot org (BYO bot, owner-set
  preferred, OAuth link, picker, start, reconciler) and full setup (server
  bot + 4 radio bots: placement, shared channel, every refusal explained),
  against a modelled Discord over httpx.MockTransport.
- storage.download_audio removed: no caller on main since #233.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 03:24:44 -04:00

333 lines
16 KiB
Python

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))