Restructure to make transformer and other functions redundant.

musicbot/commands.py

@@ -10,10 +10,8 @@
  MusicPlayer,
  MusicQueueView,
  ShortTime,
- WavelinkSearchTransformer,
  create_track_embed,
  ensure_voice_hook,
- generate_tracks_add_notification,
  is_in_bot_vc,
 )
 
@@ -59,38 +57,48 @@ async def muse_connect(itx: discord.Interaction[MusicBot]) -> None:
 @app_commands.command(name="play")
 @app_commands.guild_only()
 @ensure_voice_hook
-async def muse_play(
- itx: discord.Interaction[MusicBot],
- search: app_commands.Transform[wavelink.Playable | wavelink.Playlist, WavelinkSearchTransformer],
-) -> None:
+async def muse_play(itx: discord.Interaction[MusicBot], query: str) -> None:
  """Play audio from a YouTube url or search term.
 
  Parameters
  ----------
  itx : :class:`discord.Interaction`
  The invocation context.
- search : AnyTrack | AnyTrackIterable
- A search term/url that is converted into a track or list of tracks.
+ query : :class:`str`
+ A search term/url that is converted into a track or playlist.
  """
 
  # Known at runtime.
  assert itx.guild
  vc = itx.guild.voice_client
  assert isinstance(vc, MusicPlayer) # Known due to ensure_voice_hook.
 
- if isinstance(search, wavelink.Playable):
- search.requester = itx.user.mention # type: ignore # Runtime attribute assignment.
- else:
- search.track_extras(requester=itx.user.mention)
+ await itx.response.defer()
 
- await vc.queue.put_wait(search)
- notif_text = generate_tracks_add_notification(search)
- await itx.followup.send(notif_text)
+ tracks: wavelink.Search = await wavelink.Playable.search(query)
+ if not tracks:
+ await itx.followup.send(f"Could not find any tracks based on the given query: `{query}`.")
+
+ if isinstance(tracks, wavelink.Playlist):
+ tracks.track_extras(requester=itx.user.mention)
+ added = await vc.queue.put_wait(tracks)
+ await itx.followup.send(f"Added {added} tracks from the `{tracks.name}` playlist to the queue.")
+ else:
+ track = tracks[0]
+ track.requester = itx.user.mention # type: ignore # Runtime attribute assignment.
+ await vc.queue.put_wait(track)
+ await itx.followup.send(f"Added `{track.title}` to the queue.")
 
  if not vc.playing:
  await vc.play(vc.queue.get())
 
 
+@muse_play.autocomplete("query")
+async def muse_play_autocomplete(_: discord.Interaction[MusicBot], current: str) -> list[app_commands.Choice[str]]:
+ tracks: wavelink.Search = await wavelink.Playable.search(current)
+ return [app_commands.Choice(name=track.title, value=track.uri or track.title) for track in tracks][:25]
+
+
 @app_commands.command(name="pause")
 @app_commands.guild_only()
 @is_in_bot_vc()

musicbot/utils.py

@@ -27,16 +27,13 @@
  "NotInVoiceChannel",
  "NotInBotVoiceChannel",
  "InvalidShortTimeFormat",
- "WavelinkSearchError",
  "LavalinkCreds",
  "ShortTime",
- "WavelinkSearchTransformer",
  "MusicQueue",
  "MusicPlayer",
  "MusicQueueView",
  "resolve_path_with_links",
  "create_track_embed",
- "generate_tracks_add_notification",
  "ensure_voice_hook",
  "is_in_bot_vc",
 )
@@ -92,17 +89,6 @@ def __init__(self, value: str, *args: object) -> None:
  super().__init__(message, *args)
 
 
-class WavelinkSearchError(MusicBotError):
- """Exception raised when a wavelink search fails to find any tracks.
-
- This inherits from :exc:`app_commands.AppCommandError`.
- """
-
- def __init__(self, value: str, *args: object) -> None:
- message = f"Failed to find any tracks matching that query: {value}."
- super().__init__(message, *args)
-
-
 class LavalinkCreds(NamedTuple):
  """Credentials for the Lavalink node this bot is connecting to."""
 
@@ -127,23 +113,6 @@ async def transform(cls: type[Self], _: discord.Interaction, position_str: str,
  return cls(position_str, position_seconds)
 
 
-class WavelinkSearchTransformer(app_commands.Transformer):
- """Transforms command argument to a wavelink track or collection of tracks."""
-
- async def transform(self, itx: discord.Interaction, value: str, /) -> wavelink.Playable | wavelink.Playlist:
- # Searching can take a while sometimes.
- await itx.response.defer()
-
- tracks: wavelink.Search = await wavelink.Playable.search(value)
- if not tracks:
- raise WavelinkSearchError(value, discord.AppCommandOptionType.string, self)
- return tracks if isinstance(tracks, wavelink.Playlist) else tracks[0]
-
- async def autocomplete(self, _: discord.Interaction, value: str) -> list[app_commands.Choice[str]]: # type: ignore # Narrowing.
- tracks: wavelink.Search = await wavelink.Playable.search(value)
- return [app_commands.Choice(name=track.title, value=track.uri or track.title) for track in tracks][:25]
-
-
 class MusicQueue(wavelink.Queue):
  """A version of :class:`wavelink.Queue` with extra operations."""
 
@@ -480,22 +449,6 @@ def create_track_embed(title: str, track: wavelink.Playable) -> discord.Embed:
  return embed
 
 
-def generate_tracks_add_notification(tracks: wavelink.Playable | wavelink.Playlist | list[wavelink.Playable]) -> str:
- """Return the appropriate notification string for tracks being added to a queue.
-
- This accounts for the tracks being indvidual, in a list, or in async iterator format — no others.
- """
-
- if isinstance(tracks, wavelink.Playlist):
- return f"Added {len(tracks.tracks)} tracks from the `{tracks.name}` playlist to the queue."
- if isinstance(tracks, list) and (len(tracks)) > 1:
- return f"Added `{len(tracks)}` tracks to the queue."
- if isinstance(tracks, list):
- return f"Added `{tracks[0].title}` to the queue."
-
- return f"Added `{tracks.title}` to the queue."
-
-
 def ensure_voice_hook(func: UnboundCommandCallback[P, T]) -> UnboundCommandCallback[P, T]:
  """A makeshift pre-command hook, ensuring that a voice client automatically connects the right channel.