|
| 1 | +from flask import Blueprint, make_response, request |
| 2 | + |
| 3 | +from src.database.crud.album import ( |
| 4 | + add_genres_to_album, |
| 5 | + get_album_artists, |
| 6 | + get_album_genres, |
| 7 | + get_user_albums, |
| 8 | + update_album, |
| 9 | +) |
| 10 | +from src.database.crud.genre import create_genre |
| 11 | +from src.database.crud.playlist import ( |
| 12 | + create_playlist, |
| 13 | + get_playlist_albums, |
| 14 | + get_playlist_by_id_or_none, |
| 15 | + get_user_playlists, |
| 16 | + update_playlist, |
| 17 | +) |
| 18 | +from src.database.crud.user import get_or_create_user |
| 19 | +from src.musicbrainz import MusicbrainzClient |
| 20 | +from src.spotify import SpotifyClient |
| 21 | +from time import sleep |
| 22 | + |
| 23 | + |
| 24 | +def database_controller(spotify: SpotifyClient, musicbrainz: MusicbrainzClient): |
| 25 | + database_controller = Blueprint( |
| 26 | + name="database_controller", import_name=__name__, url_prefix="/database" |
| 27 | + ) |
| 28 | + |
| 29 | + @database_controller.route("populate_user", methods=["GET"]) |
| 30 | + def populate_user(): |
| 31 | + access_token = request.cookies.get("spotify_access_token") |
| 32 | + user = spotify.get_current_user(access_token) |
| 33 | + simplified_playlists = spotify.get_all_playlists( |
| 34 | + user_id=user.id, access_token=access_token |
| 35 | + ) |
| 36 | + (db_user,) = get_or_create_user(user) |
| 37 | + |
| 38 | + for simplified_playlist in simplified_playlists: |
| 39 | + if "Albums" in simplified_playlist.name: |
| 40 | + db_playlist = get_playlist_by_id_or_none(simplified_playlist.id) |
| 41 | + |
| 42 | + if db_playlist is None: |
| 43 | + [playlist, albums] = [ |
| 44 | + spotify.get_playlist( |
| 45 | + access_token=access_token, id=simplified_playlist.id |
| 46 | + ), |
| 47 | + spotify.get_playlist_album_info( |
| 48 | + access_token=access_token, id=simplified_playlist.id |
| 49 | + ), |
| 50 | + ] |
| 51 | + create_playlist(playlist, albums, db_user) |
| 52 | + else: |
| 53 | + if db_playlist.snapshot_id != simplified_playlist.snapshot_id: |
| 54 | + [playlist, albums] = [ |
| 55 | + spotify.get_playlist( |
| 56 | + access_token=access_token, id=simplified_playlist.id |
| 57 | + ), |
| 58 | + spotify.get_playlist_album_info( |
| 59 | + access_token=access_token, id=simplified_playlist.id |
| 60 | + ), |
| 61 | + ] |
| 62 | + update_playlist(playlist, albums) |
| 63 | + |
| 64 | + return make_response("Playlist data populated", 201) |
| 65 | + |
| 66 | + @database_controller.route( |
| 67 | + "populate_additional_album_details_from_playlist", methods=["GET"] |
| 68 | + ) |
| 69 | + def populate_additional_album_details_from_playlist(): |
| 70 | + access_token = request.cookies.get("spotify_access_token") |
| 71 | + user = spotify.get_current_user(access_token) |
| 72 | + playlists = get_user_playlists(user.id) |
| 73 | + |
| 74 | + for playlist in playlists: |
| 75 | + albums = get_playlist_albums(playlist.id) |
| 76 | + if albums == []: |
| 77 | + continue |
| 78 | + batch_albums = split_list(albums, 20) |
| 79 | + for album_chunk in batch_albums: |
| 80 | + sleep(0.5) |
| 81 | + albums = spotify.get_multiple_albums( |
| 82 | + access_token=access_token, ids=[album.id for album in album_chunk] |
| 83 | + ) |
| 84 | + for db_album in albums: |
| 85 | + album = spotify.get_album(access_token=access_token, id=db_album.id) |
| 86 | + update_album(album) |
| 87 | + |
| 88 | + return make_response("Playlist data populated", 201) |
| 89 | + |
| 90 | + @database_controller.route("populate_additional_album_details", methods=["GET"]) |
| 91 | + def populate_additional_album_details(): |
| 92 | + access_token = request.cookies.get("spotify_access_token") |
| 93 | + user = spotify.get_current_user(access_token) |
| 94 | + albums = get_user_albums(user.id) |
| 95 | + albums_without_label = [album for album in albums] # if album.label is None] |
| 96 | + if albums_without_label == []: |
| 97 | + return make_response("No Albums to process", 204) |
| 98 | + batch_albums = split_list(albums_without_label, 20) |
| 99 | + for album_chunk in batch_albums: |
| 100 | + sleep(0.5) |
| 101 | + albums = spotify.get_multiple_albums( |
| 102 | + access_token=access_token, ids=[album.id for album in album_chunk] |
| 103 | + ) |
| 104 | + for db_album in albums: |
| 105 | + db_album.genres = musicbrainz.get_album_genres( |
| 106 | + db_album.artists[0].name, db_album.name |
| 107 | + ) |
| 108 | + update_album(db_album) |
| 109 | + |
| 110 | + return make_response("Playlist details populated", 201) |
| 111 | + |
| 112 | + @database_controller.route("populate_universal_genre_list", methods=["GET"]) |
| 113 | + def populate_universal_genre_list(): |
| 114 | + genre_list = musicbrainz.get_genre_list() |
| 115 | + [create_genre(genre) for genre in genre_list] |
| 116 | + return make_response("Genre data populated", 201) |
| 117 | + |
| 118 | + @database_controller.route("populate_user_album_genres", methods=["GET"]) |
| 119 | + def populate_user_album_genres(): |
| 120 | + access_token = request.cookies.get("spotify_access_token") |
| 121 | + user = spotify.get_current_user(access_token) |
| 122 | + populate_album_genres_by_user_id(user.id, musicbrainz) |
| 123 | + return make_response("User album genres populated", 201) |
| 124 | + |
| 125 | + return database_controller |
| 126 | + |
| 127 | + |
| 128 | +def split_list(input_list, max_length=20): |
| 129 | + return [ |
| 130 | + input_list[i : i + max_length] for i in range(0, len(input_list), max_length) |
| 131 | + ] |
| 132 | + |
| 133 | + |
| 134 | +def populate_album_genres_by_user_id( |
| 135 | + user_id: str, musicbrainz: MusicbrainzClient = MusicbrainzClient() |
| 136 | +): |
| 137 | + albums = get_user_albums(user_id=user_id) |
| 138 | + print(f"processing album {0} of {len(albums)}") |
| 139 | + skip_count = 0 |
| 140 | + for idx, db_album in enumerate(albums): |
| 141 | + print("\033[A \033[A") |
| 142 | + print(f"processing album {idx} of {len(albums)}, skipped {skip_count}") |
| 143 | + if get_album_genres(db_album.id) != []: |
| 144 | + skip_count += 1 |
| 145 | + continue |
| 146 | + album_artists = get_album_artists(db_album) |
| 147 | + genres = musicbrainz.get_album_genres( |
| 148 | + artist_name=album_artists[0].name, album_title=db_album.name |
| 149 | + ) |
| 150 | + add_genres_to_album(db_album, genres) |
| 151 | + print("\033[A \033[A") |
| 152 | + print(f"completed. Processed {len(albums)} albums. Skipped ") |
0 commit comments