Logan CusanoandClaude Opus 5.5
8af410aede
Owner-reported on live: adding a radio channel for a server + voice channel that already had one (different label, system, even bot) overwrote the existing stream and made its bot leave, because stream_id was guild:channel. - New streams get guild:channel:<12 hex> ids (leases.new_stream_id); old guild:channel docs keep working for read/stop/remove/node reports. Route validators accept both (leases.STREAM_ID_PATTERN). - start_stream: no stream_id = ADD. Same system already on in that channel is idempotent (already_on, label/mode untouched); a stopped one of that system is restarted; anything else is a new doc. stream_id = explicit restart/edit (admin Start button, reconciler), incl. moving to another bot. - preferred_token_id is strict: bot already in this server -> 409 preferred_busy naming the stream; on another node -> preferred_other_node; unusable -> preferred_unavailable. No silent swap. pool_exhausted now says one-bot-per-server and to add + invite another bot. - One-token-per-guild was already enforced inside the lease transaction; now covered by race tests. Per-guild asyncio lock serialises the already-on / server-cap check with the create (guild half of #223). - Stop by stream_id (POST /discord/guilds/{id}/stop {stream_id}); admin UI Stop/Start/Remove act on the row's stream; node picker lists every stream in a channel and adds beside other systems. - Bot: /leave stream: (autocomplete) stops one; /leave channel: with several streams asks which unless all:True; bare /leave still stops the server. /status and /join name each stream's bot. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
329 lines
15 KiB
Python
329 lines
15 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."
|
|
),
|
|
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.\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))
|