feat(events): add LavaLyrics events

mafic/events.py

@@ -19,6 +19,7 @@
         TrackStuckEvent as TrackStuckEventPayload,
         WebSocketClosedEvent as WebSocketClosedEventPayload,
     )
+    from .typings.common import LyricsLine, LyricsObject
 
 
 # This needs HKTs in python - as Player is generic on ClientT.
@@ -31,6 +32,9 @@
     "TrackStartEvent",
     "TrackStuckEvent",
     "WebSocketClosedEvent",
+    "LyricsLineEvent",
+    "LyricsFoundEvent",
+    "LyricsNotFoundEvent",
 )
 
 
@@ -89,6 +93,69 @@ def __repr__(self) -> str:
         )
 
 
+class LyricsLineEvent(Generic[PlayerT]):
+    """Represents a lyrics line event.
+
+    Attributes
+    ----------
+    guildId: :class:`str`
+        The guild ID that received the lyrics line.
+    line: :class:`LyricsLine`
+        Information about the lyrics line.
+    """
+
+    __slots__ = ("guildId", "line")
+
+    def __init__(self, *, guildId: str, line: LyricsLine) -> None:
+        self.guildId: str = guildId
+        self.line: LyricsLine = line
+
+    def __repr__(self) -> str:
+        """Get a string representation of the event."""
+        return f"<LyricsLineEvent guildId={self.guildId} line={self.line!r}>"
+
+
+class LyricsFoundEvent(Generic[PlayerT]):
+    """Represents a lyrics found event.
+
+    Attributes
+    ----------
+    guildId: :class:`str`
+        The guild ID that received the lyrics line.
+    lyrics: :class:`LyricsObject`
+        Information about all lyrics, including provider and platform.
+    """
+
+    __slots__ = ("guildId", "lyrics")
+
+    def __init__(self, *, guildId: str, lyrics: LyricsObject) -> None:
+        self.guildId: str = guildId
+        self.lyrics: LyricsObject = lyrics
+
+    def __repr__(self) -> str:
+        """Get a string representation of the event."""
+        return f"<LyricsFoundEvent guildId={self.guildId} lyrics={self.lyrics!r}>"
+
+
+class LyricsNotFoundEvent(Generic[PlayerT]):
+    """Represents a lyrics not found event.
+
+    Attributes
+    ----------
+    guildId: :class:`str`
+        The guild ID that received event.
+    """
+
+    __slots__ = ("guildId",)
+
+    def __init__(self, *, guildId: str) -> None:
+        self.guildId: str = guildId
+
+    def __repr__(self) -> str:
+        """Get a string representation of the event."""
+        return f"<LyricsNotFoundEvent guildId={self.guildId}>"
+
+
 class TrackStartEvent(Generic[PlayerT]):
     """Represents an event when a track starts playing.
 

mafic/node.py

@@ -31,6 +31,7 @@
 from .type_variables import ClientT
 from .typings import (
     BalancingIPRouteDetails,
+    LyricsObject,
     NanoIPRouteDetails,
     RotatingIPRouteDetails,
     RotatingNanoIPRouteDetails,
@@ -231,9 +232,9 @@ def __init__(
         self.regions: list[VoiceRegion] | None = _wrap_regions(regions)
 
         self._rest_uri = yarl.URL.build(
-            scheme=f"http{'s'*secure}", host=host, port=port
+            scheme=f"http{'s' * secure}", host=host, port=port
         )
-        self._ws_uri = yarl.URL.build(scheme=f"ws{'s'*secure}", host=host, port=port)
+        self._ws_uri = yarl.URL.build(scheme=f"ws{'s' * secure}", host=host, port=port)
         self._resume_key = resume_key or f"{host}:{port}:{label}"
         self._resuming_session_id: str = resuming_session_id or ""
 
@@ -1397,3 +1398,93 @@ async def sync_players(
                 for player_id in expected_player_ids - actual_player_ids
             ),
         )
+
+    async def subscribe_to_lyrics(
+        self, guild_id: int, *, skip_track_source: bool = False
+    ) -> None:
+        """Subscribe to lyrics-related events.
+
+        This requires the `LavaLyrics`_ plugin to be installed.
+
+        .. _LavaLyrics: https://github.com/topi314/LavaLyrics/
+
+        Parameters
+        ----------
+        guild_id:
+            The guild that will receive lyrics events
+        skip_track_source:
+            Skip the current track source and fetch from highest priority source
+        """
+        await self.__request(
+            "POST",
+            f"sessions/{self._session_id}/players/{guild_id}/lyrics/subscribe",
+            params={"skipTrackSource": str(skip_track_source)},
+        )
+
+    async def unsubscribe_from_lyrics(self, guild_id: int) -> None:
+        """Unsubscribe from lyrics-related events.
+
+        This requires the `LavaLyrics`_ plugin to be installed.
+
+        .. _LavaLyrics: https://github.com/topi314/LavaLyrics/
+
+        Parameters
+        ----------
+        guild_id:
+            The guild that will no longer receive lyrics events
+        """
+        await self.__request(
+            "DELETE", f"sessions/{self._session_id}/players/{guild_id}/lyrics/subscribe"
+        )
+
+    async def get_playing_lyrics(
+        self, guild_id: int, *, skip_track_source: bool = False
+    ) -> LyricsObject:
+        """Get the lyrics of the current playing track.
+
+        By default, it will try to fetch the lyrics from where the track is sourced from
+
+        This requires the `LavaLyrics`_ plugin to be installed.
+
+        .. _LavaLyrics: https://github.com/topi314/LavaLyrics/
+
+        Parameters
+        ----------
+        guild_id:
+            The guild that with playing track
+        skip_track_source:
+            Skip the current track source and fetch from highest priority source
+        """
+        data = await self.__request(
+            "GET",
+            f"sessions/{self._session_id}/players/{guild_id}/track/lyrics",
+            params={"skipTrackSource": str(skip_track_source)},
+        )
+        # logger.debug
+        return LyricsObject(data)
+
+    async def get_lyrics(
+        self, track: str, *, skip_track_source: bool = False
+    ) -> LyricsObject:
+        """Get the lyrics for a given encoded track.
+
+        By default, it will try to fetch the lyrics from where the track is sourced from
+
+        This requires the `LavaLyrics`_ plugin to be installed.
+
+        .. _LavaLyrics: https://github.com/topi314/LavaLyrics/
+
+        Parameters
+        ----------
+        track:
+            The encoded track to fetch lyrics for
+        skip_track_source:
+            Skip the current track source and fetch from highest priority source
+        """
+        data = await self.__request(
+            "GET",
+            "lyrics",
+            params={"track": track, "skipTrackSource": str(skip_track_source)},
+        )
+        # logger.debug
+        return LyricsObject(data)

mafic/player.py

@@ -320,6 +320,18 @@ def dispatch_event(self, data: EventPayload) -> None:
             event = TrackStuckEvent(player=self, track=track, payload=data)
             self.client.dispatch("track_stuck", event)
             _log.debug("Received track stuck event: %s", event)
+        elif data["type"] == "LyricsLineEvent":
+            event = LyricsLineEvent(guildId=data["guildId"], line=data["line"])
+            self.client.dispatch("lyrics_line", event)
+            _log.debug("Received lyrics line event: %s", event)
+        elif data["type"] == "LyricsFoundEvent":
+            event = LyricsFoundEvent(guildId=data["guildId"], lyrics=data["lyrics"])
+            self.client.dispatch("lyrics_found", event)
+            _log.debug("Received lyrics found event: %s", event)
+        elif data["type"] == "LyricsNotFoundEvent":
+            event = LyricsNotFoundEvent(guildId=data["guildId"])
+            self.client.dispatch("lyrics_not_found", event)
+            _log.debug("Received lyrics not found event: %s", event)
         else:
             # Pyright expects this to never happen, so do I, I really hope.
             # Nobody expects the Spanish Inquisition, neither does pyright.

mafic/typings/common.py

@@ -9,6 +9,8 @@
 if TYPE_CHECKING:
     from typing_extensions import NotRequired
 
+    from . import PluginData
+
 __all__ = (
     "Filters",
     "ChannelMix",
@@ -31,6 +33,8 @@
     "CPU",
     "FrameStats",
     "Stats",
+    "LyricsObject",
+    "LyricsLine",
 )
 
 
@@ -173,3 +177,18 @@ class Stats(TypedDict):
     cpu: CPU
     # V3 is NotRequired, V4 is None
     frameStats: NotRequired[FrameStats | None]
+
+
+class LyricsObject(TypedDict):
+    sourceName: str
+    provider: str
+    text: str
+    lines: list[LyricsLine]
+    plugin: PluginData
+
+
+class LyricsLine(TypedDict):
+    timestamp: int
+    duration: int
+    line: str
+    plugin: PluginData

mafic/typings/incoming.py

@@ -4,7 +4,7 @@
 
 from typing import TYPE_CHECKING, TypedDict, Union
 
-from .common import Stats, TrackWithInfo
+from .common import LyricsLine, LyricsObject, Stats, TrackWithInfo
 from .misc import PayloadWithGuild
 
 if TYPE_CHECKING:
@@ -24,6 +24,9 @@
     "TrackStartEvent",
     "TrackStuckEvent",
     "WebSocketClosedEvent",
+    "LyricsFoundEvent",
+    "LyricsNotFoundEvent",
+    "LyricsLineEvent",
 )
 
 
@@ -99,6 +102,26 @@ class TrackStuckEvent(PayloadWithGuild):
     thresholdMs: int
 
 
+class LyricsFoundEvent(PayloadWithGuild):
+    lyrics: LyricsObject
+    op: Literal["event"]
+    type: Literal["LyricsFoundEvent"]
+
+
+class LyricsNotFoundEvent(PayloadWithGuild):
+    # "This event does not contain any additional fields."
+    op: Literal["event"]
+    type: Literal["LyricsNotFoundEvent"]
+
+
+class LyricsLineEvent(PayloadWithGuild):
+    lineIndex: int
+    line: LyricsLine
+    skipped: bool
+    op: Literal["event"]
+    type: Literal["LyricsLineEvent"]
+
+
 class ReadyPayload(TypedDict):
     op: Literal["ready"]
     resumed: bool

test_bot/bot/__main__.py

@@ -75,7 +75,7 @@ async def before_identify_hook(
         # gateway-proxy
         return
 
-    async def add_nodes(self) -> None:  # noqa: PLR0912
+    async def add_nodes(self) -> None:
         with open(environ["LAVALINK_FILE"], "rb") as f:
             data: list[LavalinkInfo] = orjson.loads(f.read())