api/endpoints/lastfm.py

#!/usr/bin/env python3.12
# pylint: disable=bare-except

import importlib
import traceback
from fastapi import FastAPI
from pydantic import BaseModel

class ValidArtistSearchRequest(BaseModel):
    """
    - **a**: artist name
    """

    a: str
    
    class Config: # pylint: disable=missing-class-docstring
        schema_extra = {
             "example": {
                  "a": "eminem"
                  }
                }    

class ValidAlbumDetailRequest(BaseModel):
    """
    - **a**: artist name
    - **a2**: album/release name (as sourced from here/LastFM)
    """

    a: str
    a2: str

    class Config: # pylint: disable=missing-class-docstring
        schema_extra = {
            "example": {
                "a": "eminem",
                "a2": "houdini"
                }
            }
                
class ValidTrackInfoRequest(BaseModel):
    """
    - **a**: artist name
    - **t**: track
    """

    a: str
    t: str

    class Config: # pylint: disable=missing-class-docstring
        schema_extra = {
            "example": {
                "a": "eminem",
                "t": "rap god"
                }
            }

class LastFM(FastAPI):
    """Last.FM Endpoints"""   
    def __init__(self, app: FastAPI, util, constants, glob_state):  # pylint: disable=super-init-not-called
        self.app = app
        self.util = util
        self.constants = constants
        self.glob_state = glob_state
        self.lastfm = importlib.import_module("lastfm_wrapper").LastFM()

        self.endpoints = {
            "lastfm/get_artist_by_name": self.artist_by_name_handler,
            "lastfm/get_artist_albums": self.artist_album_handler,
            "lastfm/get_release": self.release_detail_handler,
            "lastfm/get_release_tracklist": self.release_tracklist_handler,
            "lastfm/get_track_info": self.track_info_handler,
            #tbd
            }

        for endpoint, handler in self.endpoints.items():
            app.add_api_route(f"/{endpoint}", handler, methods=["POST"],
                              include_in_schema=True)
    
    async def artist_by_name_handler(self, data: ValidArtistSearchRequest):
        """
        Get artist info
        - **a**: Artist to search
        """
        artist = data.a.strip()
        if not artist:
            return {
                'err': True,
                'errorText': 'No artist specified'
            }
        
        artist_result = await self.lastfm.search_artist(artist=artist)
        if not artist_result or "err" in artist_result.keys():
            return {
                'err': True,
                'errorText': 'Search failed (no results?)'
            }
        
        return {
            'success': True,
            'result': artist_result
        }
    
    async def artist_album_handler(self, data: ValidArtistSearchRequest):
        """
        Get artist's albums/releases
        - **a**: Artist to search
        """
        artist = data.a.strip()
        if not artist:
            return {
                'err': True,
                'errorText': 'No artist specified'
            }
        
        album_result = await self.lastfm.get_artist_albums(artist=artist)
        album_result_out = []
        seen_release_titles = []
        
        for release in album_result:
            release_title = release.get('title')
            if release_title.lower() in seen_release_titles:
                continue
            seen_release_titles.append(release_title.lower())
            album_result_out.append(release)

        return {
            'success': True,
            'result': album_result_out
            }

    async def release_detail_handler(self, data: ValidAlbumDetailRequest):
        """
        Get details of a particular release by an artist
        - **a**: Artist to search
        - **a2**: Release title to search (subject to change)
        """
        artist = data.a.strip()
        release = data.a2.strip()
        
        if not artist or not release:
            return {
                'err': True,
                'errorText': 'Invalid request'
            }
        
        release_result = await self.lastfm.get_release(artist=artist, album=release)
        ret_obj = {
            'id': release_result.get('id'),
            'artists': release_result.get('artists'),
            'title': release_result.get('title'),
            'summary': release_result.get('summary'),
            'tracks': release_result.get('tracks')
            } 

        return {
            'success': True,
            'result': ret_obj
            }
    
    async def release_tracklist_handler(self, data: ValidAlbumDetailRequest):
        """
        Get track list for a particular release by an artist
        - **a**: Artist to search
        - **a2**: Release title to search (subject to change)
        """
        artist = data.a.strip()
        release = data.a2.strip()
        
        if not artist or not release:
            return {
                'err': True,
                'errorText': 'Invalid request'
            }
        
        tracklist_result = await self.lastfm.get_album_tracklist(artist=artist, album=release)
        return {
            'success': True,
            'id': tracklist_result.get('id'),
            'artists': tracklist_result.get('artists'),
            'title': tracklist_result.get('title'),
            'summary': tracklist_result.get('summary'),
            'tracks': tracklist_result.get('tracks')
            }

    async def track_info_handler(self, data: ValidTrackInfoRequest):
        """
         Get track info from Last.FM given an artist/track
         - **a**: Artist to search
         - **t**: Track title to search
         """    
        try:
            artist = data.a
            track = data.t
            
            if not artist or not track:
                return {
                    'err': True,
                    'errorText': 'Invalid request'
                    }
            
            track_info_result = await self.lastfm.get_track_info(artist=artist, track=track)
            assert not "err" in track_info_result.keys()
            return {
                'success': True,
                'result': track_info_result
            }
        except:
            traceback.print_exc()
            return {
                'err': True,
                'errorText': 'General error',
            }