from apollo_utils.core.constants.dsp import DSP from apollo_utils.core.constants.market import Market from apollo_utils.service.schemas.fields.datetime import DateStringField, DateTimeStringField from apollo_utils.service.schemas.fields.list import SplitUniqueListField from apollo_utils.service.schemas.fields.market import MarketField from apollo_utils.service.schemas.fields.playlists import PlaylistIdField from apollo_utils.service.schemas.mixins.isrc import SourceTrackISRCLowerMixin from apollo_utils.service.schemas.mixins.search import SearchNoneLowerMixin from marshmallow import fields, post_load, validate from server.constants.playlists.v0.misc import PlaylistTracklistOrder from server.schemas.base.pagination import PaginationMixin from server.schemas.base.schemas import RequestSchema, Schema from server.schemas.images.core_image_service import CoreImageServiceImageSizeMixin from server.schemas.images.image_url import ImageUrlResolutionChangeField from server.schemas.playlists.common.blue_dot import IsEntryBlueDotMixin from server.utils.common import get_inner_value class TrackItemBase(Schema): id = fields.Integer(attribute="track_id") name = fields.String(attribute="track.name") storefront = fields.Function(lambda item, context: context["country_code"]) artworkUrl = ImageUrlResolutionChangeField(attribute="track.image_url") isrc = fields.String() albumId = fields.Integer(attribute="track.album.album_id") albumName = fields.String(attribute="track.album.name") albumReleaseDate = DateStringField(attribute="track.album.release_date") artistId = fields.Function(lambda item: get_inner_value(item, "track", "artists", 0, "artist_id", convert_func=int)) artistIds = fields.Function( lambda item: [ int(i["artist_id"]) for i in get_inner_value(item, "track", "artists", default_value=[]) if i.get("artist_id") ] ) artistName = fields.Function(lambda item: get_inner_value(item, "track", "artists", 0, "name")) position = fields.Integer(attribute="current") releaseDate = DateStringField(attribute="track.release_date") class V1PlaylistsAppleTracks: class Request(RequestSchema): playlist_id = PlaylistIdField(required=True, dsp=DSP.APPLE, validate=validate.Length(max=50)) country_code = MarketField( data_key="country", missing=None, validate=validate.Length(min=2, max=10), gl=Market.WORLDWIDE ) class Response(Schema): class TracklistItem(TrackItemBase): addedDate = DateTimeStringField(attribute="last_added_date_time") earliestAdded = DateTimeStringField(attribute="earliest_position_date_time") earliestPosition = fields.Integer(attribute="earliest_position") positionChange = fields.Integer(attribute="trend") positionChangeDate = DateTimeStringField(attribute="change_position_date_time") isEntry = fields.Boolean(attribute="is_entry") marketStreams = fields.Function( lambda i, c: get_inner_value(i, "streaming_info", c["country_code"], "7", "isrc_in_playlist") ) tracklist = fields.List(fields.Nested(TracklistItem)) class V1PlaylistsAppleTracksHistory: class Request(RequestSchema): playlist_id = PlaylistIdField(required=True, dsp=DSP.APPLE, validate=validate.Length(max=50)) date = fields.Date(required=True) country_code = MarketField( data_key="country", missing=None, validate=validate.Length(min=2, max=10), gl=Market.WORLDWIDE ) class Response(Schema): class TracklistItem(TrackItemBase): added = DateTimeStringField(attribute="last_added_date_time") marketStreams = fields.Function( lambda i, c: get_inner_value(i, "streaming_info", c["country_code"], "1", "isrc_in_playlist") ) tracklist = fields.List(fields.Nested(TracklistItem)) class V1PlaylistsAppleTracklist: class Request( RequestSchema, SearchNoneLowerMixin, SourceTrackISRCLowerMixin, CoreImageServiceImageSizeMixin, PaginationMixin, ): playlist_id = PlaylistIdField(required=True, dsp=DSP.APPLE, validate=validate.Length(max=50)) country_code = MarketField( data_key="market", missing=Market.US, validate=validate.Length(min=2, max=10), gl=Market.US ) order_by = SplitUniqueListField( fields.String(validate=validate.OneOf(PlaylistTracklistOrder.ALL)), default=[PlaylistTracklistOrder.ADDED_DATE, PlaylistTracklistOrder.CURR_POSITION], missing=[PlaylistTracklistOrder.ADDED_DATE, PlaylistTracklistOrder.CURR_POSITION], split=False, ) @post_load def reformat_search(self, data, *args, **kwargs): search = data.get("search") if search: data["search"] = search.replace(",", " ").split() if search.find(",") > 0 else [search] return data class Response(Schema): class TracklistItem(Schema, IsEntryBlueDotMixin): is_new = fields.Boolean(attribute="is_entry") current_position = fields.Integer(attribute="current", default=None) isrc = fields.String(default=None) track_id = fields.Integer(attribute="track_id", default=None) image_url = ImageUrlResolutionChangeField(attribute="track.image_url", default=None) name = fields.String(attribute="track.name", default=None) previous_position = fields.Integer(attribute="previous", default=None) trend = fields.Integer(attribute="trend", default=None) position_change_date = DateTimeStringField(attribute="change_position_date_time", default=None) artists = fields.Function( lambda item: [ i["name"] for i in get_inner_value(item, "track", "artists", default_value=[]) if i.get("name") ] ) added_date = DateStringField(attribute="earliest_position_date_time", default=None, truncate_str=True) items = fields.List(fields.Nested(TracklistItem))