Add album details info generation scripts

backend/src/app.py

@@ -4,13 +4,16 @@
 from src.controllers.spotify import spotify_controller
 from src.exceptions.Unauthorized import UnauthorizedException
 from src.flask_config import Config
+from src.musicbrainz import MusicbrainzClient
 from src.spotify import SpotifyClient
 from src.controllers.auth import auth_controller
 
 
 def create_app():
     app = Flask(__name__)
     spotify = SpotifyClient()
+    musicbrainz = MusicbrainzClient()
+
     app.config.from_object(Config())
     app.config["CORS_HEADERS"] = "Content-Type"
 
@@ -45,6 +48,8 @@ def handle_unauthorized_exception(_):
 
     app.register_blueprint(auth_controller(spotify=spotify))
     app.register_blueprint(spotify_controller(spotify=spotify))
-    app.register_blueprint(database_controller(spotify=spotify))
+    app.register_blueprint(
+        database_controller(spotify=spotify, musicbrainz=musicbrainz)
+    )
 
     return app

backend/src/controllers/database.py

@@ -1,16 +1,27 @@
-from logging import Logger
 from flask import Blueprint, make_response, request
 
+from src.database.crud.album import (
+    add_genres_to_album,
+    get_album_artists,
+    get_album_genres,
+    get_user_albums,
+    update_album,
+)
+from src.database.crud.genre import create_genre
 from src.database.crud.playlist import (
     create_playlist,
+    get_playlist_albums,
     get_playlist_by_id_or_none,
+    get_user_playlists,
     update_playlist,
 )
 from src.database.crud.user import get_or_create_user
+from src.musicbrainz import MusicbrainzClient
 from src.spotify import SpotifyClient
+from time import sleep
 
 
-def database_controller(spotify: SpotifyClient):
+def database_controller(spotify: SpotifyClient, musicbrainz: MusicbrainzClient):
     database_controller = Blueprint(
         name="database_controller", import_name=__name__, url_prefix="/database"
     )
@@ -22,7 +33,7 @@ def populate_user():
         simplified_playlists = spotify.get_all_playlists(
             user_id=user.id, access_token=access_token
         )
-        get_or_create_user(user)
+        (db_user,) = get_or_create_user(user)
 
         for simplified_playlist in simplified_playlists:
             if "Albums" in simplified_playlist.name:
@@ -37,7 +48,7 @@ def populate_user():
                             access_token=access_token, id=simplified_playlist.id
                         ),
                     ]
-                    create_playlist(playlist, albums)
+                    create_playlist(playlist, albums, db_user)
                 else:
                     if db_playlist.snapshot_id != simplified_playlist.snapshot_id:
                         [playlist, albums] = [
@@ -52,4 +63,81 @@ def populate_user():
 
         return make_response("Playlist data populated", 201)
 
+    @database_controller.route(
+        "populate_additional_album_details_from_playlist", methods=["GET"]
+    )
+    def populate_additional_album_details_from_playlist():
+        access_token = request.cookies.get("spotify_access_token")
+        user = spotify.get_current_user(access_token)
+        playlists = get_user_playlists(user.id)
+
+        for playlist in playlists:
+            albums = get_playlist_albums(playlist.id)
+            if albums == []:
+                continue
+            batch_albums = split_list(albums, 20)
+            for album_chunk in batch_albums:
+                sleep(0.5)
+                albums = spotify.get_multiple_albums(
+                    access_token=access_token, ids=[album.id for album in album_chunk]
+                )
+                for db_album in albums:
+                    album = spotify.get_album(access_token=access_token, id=db_album.id)
+                    update_album(album)
+
+        return make_response("Playlist data populated", 201)
+
+    @database_controller.route("populate_additional_album_details", methods=["GET"])
+    def populate_additional_album_details():
+        access_token = request.cookies.get("spotify_access_token")
+        user = spotify.get_current_user(access_token)
+        albums = get_user_albums(user.id)
+        albums_without_label = [album for album in albums]  # if album.label is None]
+        if albums_without_label == []:
+            return make_response("No Albums to process", 204)
+        batch_albums = split_list(albums_without_label, 20)
+        for album_chunk in batch_albums:
+            sleep(0.5)
+            albums = spotify.get_multiple_albums(
+                access_token=access_token, ids=[album.id for album in album_chunk]
+            )
+            for db_album in albums:
+                db_album.genres = musicbrainz.get_album_genres(
+                    db_album.artists[0].name, db_album.name
+                )
+                update_album(db_album)
+
+        return make_response("Playlist data populated", 201)
+
+    @database_controller.route("populate_universal_genre_list", methods=["GET"])
+    def populate_universal_genre_list():
+        genre_list = musicbrainz.get_genre_list()
+        [create_genre(genre) for genre in genre_list]
+        return make_response("Playlist data populated", 201)
+
     return database_controller
+
+
+def split_list(input_list, max_length=20):
+    return [
+        input_list[i : i + max_length] for i in range(0, len(input_list), max_length)
+    ]
+
+
+def populate_user_album_genres(user_id: str):
+    albums = get_user_albums(user_id=user_id)
+    print(f"processing album {0} of {len(albums)}")
+    skip_count = 0
+    for idx, db_album in enumerate(albums):
+        print("\033[A                             \033[A")
+        print(f"processing album {idx} of {len(albums)}, skipped {skip_count}")
+        if get_album_genres(db_album) != []:
+            skip_count += 1
+            continue
+        album_artists = get_album_artists(db_album)
+        genres = MusicbrainzClient().get_album_genres(
+            artist_name=album_artists[0].name, album_title=db_album.name
+        )
+        add_genres_to_album(db_album, genres)
+    print("\033[A                             \033[A")
+    print(f"completed. Processed {len(albums)} albums. Skipped ")

backend/src/database/crud/album.py

@@ -1,10 +1,14 @@
+from typing import List
+from src.database.crud.artist import create_or_update_artist
 from src.dataclasses.album import Album
 from src.database.models import (
     AlbumArtistRelationship,
     AlbumGenreRelationship,
+    DbArtist,
     DbGenre,
     DbAlbum,
-    DbArtist,
+    DbPlaylist,
+    PlaylistAlbumRelationship,
 )
 
 
@@ -23,15 +27,72 @@ def create_album_or_none(album: Album):
         uri=album.uri,
     )
     for artist in album.artists:
-        DbArtist.get_or_create(
-            id=artist.id,
-            image_url=album.images[0].url if album.images else None,
-            name=artist.name,
-            uri=artist.uri,
-        )
+        create_or_update_artist(artist)
         AlbumArtistRelationship.create(album=album.id, artist=artist.id)
     for genre in album.genres or []:
         db_genre = DbGenre.get_or_create(name=genre)
         AlbumGenreRelationship.create(album=album.id, genre=db_genre.id)
 
     return album
+
+
+def update_album(album: Album):
+    DbAlbum.update(
+        album_type=album.album_type,
+        total_tracks=album.total_tracks,
+        image_url=album.images[0].url if album.images else None,
+        name=album.name,
+        release_date=album.release_date,
+        release_date_precision=album.release_date_precision,
+        label=album.label,
+        uri=album.uri,
+    ).where(DbAlbum.id == album.id).execute()
+
+    for artist in album.artists:
+        create_or_update_artist(artist)
+        AlbumArtistRelationship.get_or_create(album=album.id, artist=artist.id)
+    for genre in album.genres or []:
+        db_genre = DbGenre.get_or_none(name=genre)
+        if db_genre:
+            AlbumGenreRelationship.get_or_create(album=album.id, genre=db_genre.id)
+
+    return album
+
+
+def get_album_genres(album: DbAlbum) -> List[str]:
+    query = (
+        DbGenre.select()
+        .join(AlbumGenreRelationship)
+        .join(DbAlbum)
+        .where(DbAlbum.id == album.id)
+    )
+    return list(query)
+
+
+def add_genres_to_album(album: DbAlbum, genres: List[str]) -> DbAlbum:
+    for genre in genres or []:
+        db_genre = DbGenre.get_or_none(name=genre)
+        if db_genre:
+            AlbumGenreRelationship.get_or_create(album=album.id, genre=db_genre.id)
+
+    return album
+
+
+def get_album_artists(album: DbAlbum) -> List[DbArtist]:
+    query = (
+        DbArtist.select()
+        .join(AlbumArtistRelationship)
+        .join(DbAlbum)
+        .where(DbAlbum.id == album.id)
+    )
+    return list(query)
+
+
+def get_user_albums(user_id: str) -> List[DbAlbum]:
+    query = (
+        DbAlbum.select()
+        .join(PlaylistAlbumRelationship)
+        .join(DbPlaylist)
+        .where(DbPlaylist.user == user_id)
+    )
+    return list(query)

backend/src/database/crud/artist.py

@@ -0,0 +1,20 @@
+from src.database.models import DbArtist
+from src.dataclasses.artist import Artist
+
+
+def create_or_update_artist(artist: Artist):
+    db_artist, created = DbArtist.get_or_create(
+        id=artist.id,
+        defaults={
+            "image_url": artist.images[0].url if artist.images else None,
+            "name": artist.name,
+            "uri": artist.uri,
+        },
+    )
+    if not created:
+        if artist.images:
+            db_artist.image_url = artist.images[0].url
+        db_artist.name = artist.name
+        db_artist.uri = artist.uri
+        db_artist.save()
+    return db_artist

backend/src/database/crud/genre.py

@@ -0,0 +1,5 @@
+from src.database.models import DbGenre
+
+
+def create_genre(name: str):
+    return DbGenre.get_or_create(name=name)

backend/src/database/crud/playlist.py

@@ -1,6 +1,6 @@
 from typing import List
 from src.database.crud.album import create_album_or_none
-from src.database.models import DbPlaylist, PlaylistAlbumRelationship
+from src.database.models import DbAlbum, DbPlaylist, DbUser, PlaylistAlbumRelationship
 from src.dataclasses.album import Album
 from src.dataclasses.playlist import Playlist
 
@@ -9,13 +9,13 @@ def get_playlist_by_id_or_none(id: str):
     return DbPlaylist.get_or_none(DbPlaylist.id == id)
 
 
-def create_playlist(playlist: Playlist, albums: List[Album]):
+def create_playlist(playlist: Playlist, albums: List[Album], user: DbUser):
     playlist = DbPlaylist.create(
         id=playlist.id,
         description=playlist.description,
         image_url=playlist.images[0].url if playlist.images else None,
         name=playlist.name,
-        owner=playlist.owner.id,
+        user=user.id,
         snapshot_id=playlist.snapshot_id,
         uri=playlist.uri,
     )
@@ -33,7 +33,7 @@ def update_playlist(playlist: Playlist, albums: List[Album]):
         description=playlist.description,
         image_url=playlist.images[0].url if playlist.images else None,
         name=playlist.name,
-        owner=playlist.owner.id,
+        user_id=playlist.owner.id,
         snapshot_id=playlist.snapshot_id,
         uri=playlist.uri,
     )
@@ -44,3 +44,17 @@ def update_playlist(playlist: Playlist, albums: List[Album]):
         PlaylistAlbumRelationship.create(playlist=playlist.id, album=album.id)
 
     return playlist
+
+
+def get_user_playlists(user_id: str) -> List[DbPlaylist]:
+    return DbPlaylist.select().where(DbPlaylist.user == user_id).execute()
+
+
+def get_playlist_albums(playlist_id: str) -> List[DbAlbum]:
+    query = (
+        DbAlbum.select()
+        .join(PlaylistAlbumRelationship)
+        .join(DbPlaylist)
+        .where(DbPlaylist.id == playlist_id)
+    )
+    return list(query)

backend/src/database/models.py

@@ -31,7 +31,7 @@ class DbPlaylist(BaseModel):
     description = CharField()
     image_url = CharField(null=True)
     name = CharField()
-    owner = DbUser()
+    user = ForeignKeyField(DbUser, backref="owner", to_field="id")
     snapshot_id = CharField()
     uri = CharField()
 
@@ -89,7 +89,7 @@ class Meta:
 
 class AlbumGenreRelationship(BaseModel):
     album = ForeignKeyField(DbAlbum, backref="genres")
-    genre = ForeignKeyField(DbArtist, backref="albums")
+    genre = ForeignKeyField(DbGenre, backref="albums")
 
     class Meta:
         indexes = ((("album", "genre"), True),)

backend/src/dataclasses/linked_from.py

@@ -1,10 +1,11 @@
+from typing import Optional
 from pydantic import BaseModel
 from src.dataclasses.external_urls import ExternalUrls
 
 
-class LinkedFrom(BaseModel):     
-    external_urls: ExternalUrls
-    href: str
-    id: str
-    type: str
-    uri: str
+class LinkedFrom(BaseModel):
+    external_urls: Optional[ExternalUrls] = None
+    href: Optional[str] = None
+    id: Optional[str] = None
+    type: Optional[str] = None
+    uri: Optional[str] = None

backend/src/dataclasses/musicbrainz/release_group_response.py

@@ -0,0 +1,16 @@
+from typing import List, Optional
+from pydantic import BaseModel, Field
+
+
+class Tag(BaseModel):
+    count: int
+    name: str
+
+
+class ReleaseGroup(BaseModel):
+    tags: Optional[List[Tag]] = None
+
+
+class ReleaseGroupResponse(BaseModel):
+    count: int
+    release_groups: List[ReleaseGroup] = Field(alias="release-groups")

backend/src/dataclasses/simplified_track.py

@@ -19,7 +19,7 @@ class SimplifiedTrack(BaseModel):
     linked_from: Optional[LinkedFrom] = None
     restrictions: Optional[Restrictions] = None
     name: str
-    preview_url: str
+    preview_url: Optional[str] = None
     track_number: int
     type: str
     uri: str