"""Schema for Tracks.""" import marshmallow from marshmallow import fields from marshmallow import ValidationError from marshmallow.validate import Length from marshmallow.validate import OneOf from marshmallow.validate import Range from backend import features from backend.constants import track_field from backend.constants.ownership_type import OWNERSHIP_TYPES from backend.constants.track_role import ARTISTS from backend.constants.track_role import WRITERS from backend.models import ows_lyrics from backend.utils import validation as validation_utils from backend.utils.schema_validator import IsNotBlankOrWhitespace EXPLICIT_CHOICES = ('Y', 'N', 'C', '') OFFER_TYPE_CHOICES = ( track_field.OFFER_TYPE_ALL, track_field.OFFER_TYPE_ALBUM_DOWNLOAD_ONLY, track_field.OFFER_TYPE_TRACK_DOWNLOAD_ONLY, track_field.OFFER_TYPE_TRACK_DOWNLOAD_STREAM, track_field.OFFER_TYPE_ALBUM_TRACK_DOWNLOAD, track_field.OFFER_TYPE_ALBUM_DOWNLOAD_STREAM, track_field.OFFER_TYPE_STREAM_ONLY, track_field.OFFER_TYPE_NONE, ) FOCUS_TRACK_CHOICES = ('Y', 'N') class StringOrNumeric(fields.String): """Class that allows input field value to be string or numeric.""" def _deserialize(self, value, attr, obj, **kwargs): if value is None: return None if isinstance(value, int): return value if isinstance(value, str): return value raise ValidationError('Not a valid number or string.') class TrackArtistSchema(marshmallow.Schema): """Track Artist Schema.""" type = fields.String( # noqa: A003 required=True, validate=[OneOf(ARTISTS)]) name = fields.String( required=True, validate=[IsNotBlankOrWhitespace(), Length(max=255)]) artist_info_id = fields.Int(required=False, allow_none=True) track_artist_id = StringOrNumeric(required=False) class TrackWriterSchema(marshmallow.Schema): """Track Writer Schema.""" type = fields.String( # noqa: A003 required=False, validate=[OneOf(WRITERS)]) name = fields.String( required=True, validate=[IsNotBlankOrWhitespace(), Length(max=255)]) artist_info_id = fields.Int(required=False, allow_none=True) track_writer_id = StringOrNumeric(required=False) class TrackPostSchema(marshmallow.Schema): """Create New Track Schema.""" product_id = fields.Int(required=True, validate=Range(min=1)) track_name = fields.String( required=True, validate=[Length(max=200)]) volume_number = fields.Int(required=False, validate=Range(min=1, max=255)) track_type = fields.String(required=False) @marshmallow.post_load def sanitize_input(self, data): """Clean up and sanitize raw input data (post validation). Args: data (dict): Unserialized data Returns: data (dict) """ data['track_name'] = data['track_name'].strip() return data class TrackUpdateSchema(marshmallow.Schema): """Update Track Schema.""" track_name = fields.String( required=True, validate=[Length(max=200)]) meta_language_code = fields.String( required=True, allow_none=True) version = fields.String( required=True, validate=[Length(max=255)], allow_none=True) isrc = fields.String( required=False) track_type = fields.String( required=False) upc = fields.Int( required=False) p_info = fields.String( required=True, validate=[Length(max=255)], allow_none=True) explicit = fields.String( required=True, validate=[OneOf(EXPLICIT_CHOICES)]) ownership_rights = fields.String( required=True, allow_none=True, validate=[OneOf(OWNERSHIP_TYPES)]) recording_country_id = fields.Int( required=True, allow_none=True, validate=[Range(min=1)]) original_rights_holder_country_id = fields.Int( required=True, allow_none=True, validate=[Range(min=1)]) focus_track = fields.String( required=False, allow_none=False, validate=[OneOf(FOCUS_TRACK_CHOICES)]) focus_track_start_date = fields.Date( required=False, allow_none=True) focus_track_end_date = fields.Date( required=False, allow_none=True) # Optional Fields preview_start_time = fields.Int( required=False, allow_none=True, validate=[Range(min=0)]) artists = fields.Nested( required=False, nested=TrackArtistSchema, many=True) writers = fields.Nested( required=False, nested=TrackWriterSchema, many=True) lyrics = fields.String(required=False, allow_none=True) offer_type = fields.String( required=False, validate=[OneOf(OFFER_TYPE_CHOICES)]) @marshmallow.validates('meta_language_code') def validate_meta_language_code(self, value): """Validate Meta Language Code.""" if value and not validation_utils.is_meta_language_code_format(value): raise marshmallow.ValidationError( 'Invalid meta_language_code format') @marshmallow.validates('isrc') def validate_isrc(self, value): """Validate ISRC.""" if value and not validation_utils.is_isrc(value): raise marshmallow.ValidationError('Invalid isrc') @marshmallow.validates('p_info') def validate_p_info(self, value): """Validate p_info.""" if value and not validation_utils.is_pinfo(value): raise marshmallow.ValidationError('Invalid p_info') def _validate_explicit_matches_lyrics(self, lyrics, explicit, track_type): if track_type and track_type == 'video': return if explicit not in ['N', 'C']: return has_explicit_lyrics = ows_lyrics.has_explicit_lyrics(lyrics) if has_explicit_lyrics: raise marshmallow.ValidationError( 'Swear word found', 'explicit') def _validate_focus_track(self, focus_track, focus_track_start_date, focus_track_end_date): if not focus_track and (focus_track_start_date or focus_track_end_date): raise marshmallow.ValidationError( 'Specify whether track is a focus track when updating dates') if focus_track == 'Y' and not features.is_incfeatures_single_focus_track_enabled(): if not focus_track_start_date: raise marshmallow.ValidationError( 'Focus track start date missing') if focus_track_end_date and (focus_track_start_date > focus_track_end_date): raise marshmallow.ValidationError( 'Focus track start date comes after focus track end date') @marshmallow.validates_schema(skip_on_field_errors=True) def validate_schema(self, data): """Do schema validations only when passing all field validators. Does additional checks to make sure duplicate data isn't passed in. Args: data (dict): Unserialized data Raises: marshmallow.ValidationError """ artists = data.get('artists', []) unique_artists = { (artist['type'], artist['name']) for artist in artists} if len(artists) != len(unique_artists): raise marshmallow.ValidationError( 'Duplicate artist entries found', 'artists') writers = data.get('writers', []) unique_writers = {writer['name'] for writer in writers} if len(writers) != len(unique_writers): raise marshmallow.ValidationError( 'Duplicate writer found', 'writers') if data.get('lyrics'): self._validate_explicit_matches_lyrics( data.get('lyrics').lower(), data.get('explicit'), data.get('track_type')) self._validate_focus_track( data.get('focus_track'), data.get('focus_track_start_date'), data.get('focus_track_end_date') ) @marshmallow.validates('lyrics') def validate_lyrics(self, value): """Validate lyrics.""" @marshmallow.post_load def sanitize_input(self, data): """Clean up and sanitize raw input data when validation passes. Args: data (dict): Unserialized data Returns: data (dict) """ for field_name in ['track_name', 'version']: if data.get(field_name): data[field_name] = data[field_name].strip() return data class TrackDurationSchema(marshmallow.Schema): """Track duration schema.""" duration_minutes = fields.Int( required=False, allow_none=True, validate=[Range(min=0)]) duration_seconds = fields.Int( required=False, allow_none=True, validate=[Range(min=0)]) class TrackCreateWithMetadata(TrackUpdateSchema): """Create track with metadata Schema.""" track_number = fields.Int( required=False, allow_none=True, validate=[Range(min=0)]) volume_number = fields.Int( required=False, allow_none=True, validate=[Range(min=1)]) class TracksCreateWithMetadata(marshmallow.Schema): """Create tracks with metadata batch Schema.""" tracks = fields.List( fields.Nested(TrackCreateWithMetadata), required=True) class SetProductTrackSchema(marshmallow.Schema): """Set a product track Schema.""" track_name = fields.String( required=True, validate=[Length(max=200)]) track_number = fields.Int( required=True, allow_none=True, validate=[Range(min=1)]) volume_number = fields.Int( required=True, allow_none=True, validate=[Range(min=1)]) isrc = fields.String( required=False, allow_none=True) @marshmallow.validates('isrc') def validate_isrc(self, value): """Validate ISRC.""" if value and not validation_utils.is_isrc(value): raise marshmallow.ValidationError('Invalid isrc') class SetProductTracksSchema(marshmallow.Schema): """Set all product tracks Schema.""" tracks = fields.List( fields.Nested(SetProductTrackSchema), required=True) generate_isrc = fields.Boolean( required=False)