ListenBrainz/plugin.py

###
# Copyright (c) 2025, Pratyush Desai
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
#   * Redistributions of source code must retain the above copyright notice,
#     this list of conditions, and the following disclaimer.
#   * Redistributions in binary form must reproduce the above copyright notice,
#     this list of conditions, and the following disclaimer in the
#     documentation and/or other materials provided with the distribution.
#   * Neither the name of the author of this software nor the name of
#     contributors to this software may be used to endorse or promote products
#     derived from this software without specific prior written consent.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

###

import liblistenbrainz
from liblistenbrainz.errors import ListenBrainzAPIException


from supybot import utils, plugins, ircutils, callbacks
from supybot.commands import *
from supybot.i18n import PluginInternationalization


_ = PluginInternationalization("ListenBrainz")


class ListenBrainz(callbacks.Plugin):
    """Fetches scrobbled metadata for users."""
    threaded = True

    def __init__(self, irc):
        super().__init__(irc)
        self.client = liblistenbrainz.ListenBrainz()
        self.DISPLAY_LIMIT = 3

    @wrap(["text"])
    def np(self, irc, msg, args, user):
        """<user>

        Announces the track currently being played by <user>.
        """
        try:
            listen = self.client.get_playing_now(user)
        except ListenBrainzAPIException as e:
            irc.error(f"ListenBrainz API Error fetching listens for {user}: {e}", Raise=True)
            return
        except Exception as e:
                    irc.error(f"An unexpected error occurred: {e}", Raise=True)
                    return

        if not listen:
            response = f"{user} doen't seem to be listening to anything"
        else:
            response = (
                f"{user} is currently playing: \x02{listen.track_name}\x0f "
                f"from \x02{listen.release_name}\x0f by \x02{listen.artist_name}\x0f "
                f"at \x02{listen.listened_at}\x0f"
            )
        irc.reply(response)


    @wrap(["text"])
    def listencount(self, irc, msg, args, user):
        """<user>

        Announces total number of tracks scrobbled by <user>
        """
        try:
            count = self.client.get_user_listen_count(user)
        except ListenBrainzAPIException as e:
            irc.error(f"ListenBrainz API Error fetching listens for {user}: {e}", Raise=True)
            return
        except Exception as e:
            irc.error(f"An unexpected error occurred: {e}", Raise=True)
            return

        if not count:
            response = "Nothing recorded"
        else:
            response = (f"{user} has recorded listening to {count} tracks")
        irc.reply(response)


    @wrap(["text"])
    def tracks(self, irc, msg, args, user):
        """<user>

        <user>'s top listened to album releases.
        """
        try:
            data = self.client.get_user_recordings(user)
            payload = data.get("payload", {})
        except ListenBrainzAPIException as e:
            irc.error(f"ListenBrainz API Error fetching listens for {user}: {e}", Raise=True)
            return
        except Exception as e:
            irc.error(f"An unexpected error occurred: {e}", Raise=True)
            return
            
        total_tracks = payload.get("total_recording_count", "N/A")
        recordings = payload.get("recordings", [])
        if not recordings:
            irc.reply(f"{user} has no recorded tracks.")
            return

        top_tracks = recordings[:self.DISPLAY_LIMIT]
        header = f"{user}'s Top {len(top_tracks)} Tracks (Total unique tracks: {total_tracks:,}):"
        irc.reply(header)
        for i, track in enumerate(top_tracks):
            rank = i + 1
            artist = track.get("artist_name", "Unknown Artist")
            track_name = track.get("track_name", "Unknown Track")
            release_name = track.get("release_name")
            count = track.get("listen_count", 0)

            release_info = f" from \x02{release_name}\x0f" if release_name else ""

            response = (
                f"({count:,} listens): "
                f"\x02{track_name}\x0f{release_info} by "
                f"\x02{artist}\x0f"
            )
            irc.reply(response)


    @wrap(["text"])
    def albums(self, irc, msg, args, user):
        """<user>

        Announces the top albums/releases listened to by <user>.
        """
        try:
            data = self.client.get_user_releases(user)
            payload = data.get("payload", {})
        except ListenBrainzAPIException as e:
            irc.error(f"ListenBrainz API Error fetching listens for {user}: {e}", Raise=True)
            return
        except Exception as e:
            irc.error(f"An unexpected error occurred: {e}", Raise=True)
            return

        total_albums = payload.get("total_release_count", "N/A")
        releases = payload.get("releases", [])

        if not releases:
            irc.reply(f"{user} has no recorded releases.")
            return

        top_releases = releases[:self.DISPLAY_LIMIT]

        header = f"{user}'s Top {len(top_releases)} Releases (Total unique releases: {total_albums:,}):"
        irc.reply(header)

        for i, release in enumerate(top_releases):
            rank = i + 1
            artist = release.get("artist_name", "Unknown Artist")
            release_name = release.get("release_name", "Unknown Album")
            count = release.get("listen_count", 0)

            response = (
                f"({count:,} listens): "
                f"\x02{release_name}\x0f by "
                f"\x02{artist}\x0f"
            )
            irc.reply(response)


Class = ListenBrainz


# vim:set shiftwidth=4 softtabstop=4 expandtab textwidth=79: