"""Logic for Track.""" from collections import ChainMap from collections import defaultdict import json from connector_neo4j import Neo4jSession from backend.models import ows_account from flask import g from oto import response from oto import status as response_code from backend import features from backend.config import EXCLUDE_PRODUCT_ID_OSR_VALIDATION, NEO4J_DATABASE_NAME from backend.constants import artist_role from backend.constants import api as api_constants from backend.constants import error from backend.constants import field as field_const from backend.constants import header from backend.constants import product as product_consts from backend.constants import services as services_constants from backend.constants import track_field from backend.constants import track_field as tf from backend.constants import validation as validation_constants from backend.exceptions import RequestError from backend.logic import audio_attribute as audio_attribute_logic from backend.logic import rights_attribute as rights_attribute_logic from backend.logic import isrc as isrc_logic from backend.logic import performer as performer_logic from backend.logic import track_sample as sample_logic from backend.models import ows_assets from backend.models import ows_blacklist_manager from backend.models import track_spatial as track_spatial_model from backend.models import ows_lyrics from backend.models import ows_permissions from backend.models import ows_product from backend.models import ows_product_digital from backend.models import ows_sound_recordings from backend.models import ows_video from backend.models import track as track_model from backend.models.track_persister import TrackPersister from backend.utils import api as api_utils from backend.utils import localization as loc_util from backend.utils import logic as logic_util from backend.utils import lyrics as lyrics_util from backend.utils import performers as performers_util from backend.utils import product_utils from backend.utils import track_utils from backend.utils import validation as validation_util from backend.utils.product_utils import get_product_artist_correction_names from backend.utils.product_utils import ( get_product_artist_corrections_for_key, get_product_featuring_artist_corrections, get_product_primary_artist_corrections, get_product_name_correction, get_product_version_correction, get_product_imprint_correction ) from backend.utils.product_utils import get_product_metalanguage_correction from backend.utils.product_utils import get_track_corrections from backend.validation import track_validators def create(track_data, account_type, account_id): """Create a new track. Args: track_data (dict): track data to add Returns: Response: A Response obj with track JSON data """ product_id = track_data.pop('product_id', None) product_response = logic_util.get_product_data_and_validate_ownership( product_id, account_type, account_id) if not product_response: return product_response upc = product_response.message['upc'] for_volume = track_data.pop(tf.VOLUME_NUMBER, None) res = TrackPersister.create_track( product_id, upc, track_data, for_volume=for_volume) return _make_track_response(res) def bulk_create( product_id, track_data_list, for_volume, account_type, account_id): """Bulk create new tracks. Args: track_data_list (list): list of track data dicts to add Returns: Response: A Response obj with tracks JSON data """ product_response = logic_util.get_product_data_and_validate_ownership( product_id, account_type, account_id) if not product_response: return product_response upc = product_response.message['upc'] res = TrackPersister.bulk_create_tracks( product_id, upc, track_data_list, for_volume=for_volume) return _make_track_response(res) def bulk_create_track_with_metadata( product_id, track_data, account_type, account_id): """Create multiple tracks and populate them with data. Args: product_id (int): Unique product id. track_data (list(dict)): List of track data. account_type (str): the user account type in the header. account_id (int): the user account id in the header. Returns: Response: Created tracks data. """ product_response = logic_util.get_product_data_and_validate_ownership( product_id, account_type, account_id) if not product_response: return product_response upc = product_response.message['upc'] create_and_update_tracks_response = ( TrackPersister.create_and_update_multiple_tracks( product_id, upc, track_data['tracks'])) return create_and_update_tracks_response def set_product_tracks( product_id, payload, account_type, account_id, generate_isrc=False): """Set the tracks on a product so they match the payload exactly. This function will create, update and delete tracks on a product. It maps tracks from the payload to the DB based on volume and track number. Args: product_id (int): Unique product id. payload (list(dict)): List of track data. account_type (str): the user account type in the header. account_id (int): the user account id in the header. Returns: Response: Created tracks data. """ product_response = logic_util.get_product_data_and_validate_ownership( product_id, account_type, account_id) if not product_response: return product_response upc = product_response.message['upc'] existing_track_data = TrackPersister.get_all_by_product_id( product_id ).message['items'] existing_tracks = { (track[tf.VOLUME_NUMBER], track[tf.TRACK_NUMBER]): track[tf.TUID] for track in existing_track_data } payload_tracks = { (track[tf.VOLUME_NUMBER], track[tf.TRACK_NUMBER]): track for track in payload } tuids_to_delete = [ tuid for key, tuid in existing_tracks.items() if key not in payload_tracks ] if tuids_to_delete: bulk_delete( tuids_to_delete, product_id, account_type=account_type, account_id=account_id, delete_assets=True) tracks_to_create = {} for track in payload: if (track[tf.VOLUME_NUMBER], track[tf.TRACK_NUMBER]) in existing_tracks: continue volume_number = track[tf.VOLUME_NUMBER] if not tracks_to_create.get(volume_number): tracks_to_create[volume_number] = [] tracks_to_create[volume_number].append(track) created_tracks = {} for volume_number, track_data in tracks_to_create.items(): create_data = [{ tf.TRACK_NAME: track[tf.TRACK_NAME], tf.VOLUME_NUMBER: track[tf.VOLUME_NUMBER], } for track in track_data] created_tracks[volume_number] = TrackPersister.bulk_create_tracks( product_id, upc, create_data, for_volume=volume_number ).message['items'] existing_tracks = { (track[tf.VOLUME_NUMBER], track[tf.TRACK_NUMBER]): track for track in existing_track_data } for volume, tracks in created_tracks.items(): for track in tracks: existing_tracks[(volume, track[tf.TRACK_NUMBER])] = track tracks_needing_isrc = [] for payload_track in payload: existing_track = existing_tracks[ (payload_track[tf.VOLUME_NUMBER], payload_track[tf.TRACK_NUMBER])] track_isrc = payload_track.get(tf.ISRC) or existing_track.get(tf.ISRC) if not track_isrc and generate_isrc: tracks_needing_isrc.append( (payload_track[tf.VOLUME_NUMBER], payload_track[tf.TRACK_NUMBER])) claimed_isrcs = {} if tracks_needing_isrc: isrc_response = TrackPersister.claim_new_isrcs(len(tracks_needing_isrc)) if not isrc_response: return isrc_response isrcs = isrc_response.message['isrcs'] for i, track_key in enumerate(tracks_needing_isrc): claimed_isrcs[track_key] = isrcs[i] update_data = [] for payload_track in payload: track_key = (payload_track[tf.VOLUME_NUMBER], payload_track[tf.TRACK_NUMBER]) existing_track = existing_tracks[track_key] track_isrc = payload_track.get(tf.ISRC) or existing_track.get(tf.ISRC) if not track_isrc and track_key in claimed_isrcs: track_isrc = claimed_isrcs[track_key] update_data.append({ **payload_track, tf.TUID: existing_track[tf.TUID], tf.ISRC: track_isrc, }) if not update_data: return api_utils.create_get_list_response([]) return TrackPersister.update_multiple_tracks( update_data, update_track_volume_numbers=True ) @logic_util.verify_product_ownership def get_by_tuid(tuid, account_type, account_id, exclude=None, include=None): """Get track by tuid. Args: tuid (int): unique id of track account_type (str): the user account type in the header. account_id (int): the user account id in the header. exclude (list): list of fields to exclude from response include (list): list of fields to include to response Returns: Response: Track JSON data """ res = TrackPersister.get_by_tuid(tuid) exclude = exclude or [] include = include or [] if tf.LOCALIZATIONS not in exclude: res = loc_util.add_track_localizations_to_response(res) if tf.LYRICS in include: res = lyrics_util.add_track_lyrics_to_response(res) if tf.TRACK_SAMPLES in include: res = track_utils.add_track_samples_to_response(res) return _make_track_response(res) def get_multiple_by_tuids(tuids, exclude=None): """Get mutliple tracks. Args: tuids (list): unique id of track exclude (list): list of fields to exclude from response Returns: Response: Track JSON data """ res = TrackPersister.get_multiple_by_tuids(tuids) exclude = exclude or [] if tf.LOCALIZATIONS not in exclude: res = loc_util.add_track_localizations_to_response(res) return _make_track_response(res) def get_multiple_by_tuids_with_nones(tuids): """Get mutliple tracks. Args: tuids (list): unique id of track Returns: Response: Track JSON data """ res = TrackPersister.get_multiple_by_tuids_with_nones(tuids) for track in res.message[api_constants.ITEMS]: if track: _make_track_response_dict(track) return res @Neo4jSession(transaction=True, use_v2=True, database=NEO4J_DATABASE_NAME) def get_tracks_by_osrid(osr_id, limit, offset): """Get a paginated list of tracks by OrchardSoundRecording ID. Args: osr_id (str): The OrchardSoundRecording to search by. limit (int): The number of tracks to return per page. offset (int): The offset to return a page of results from. Returns: dict: A standard pagination object with total_records and items[]. """ return track_model.get_tracks_by_osrid(osr_id, limit, offset) @logic_util.verify_product_ownership def update( tuid, data, account_type, account_id, user_type=None, user_id=None, patch_response=False, generate_isrc=False, get_artist_info_ids=False): """Update track. Args: tuid (int): id of track data (dict): track data to update account_type (str): the user account type in the header. account_id (int): the user account id in the header. user_type (str): the user type in the header. user_id (int): the user id in the header. patch_response (bool): Response is a patch delta instead of all fields generate_isrc (bool): Populate ISRC for track get_artist_info_ids (bool): Allow artist_info_ids without track_artist Returns: Response: A Response obj with track JSON data """ has_lyrics = tf.LYRICS in data lyrics = data.pop(tf.LYRICS, None) # This must be discarded, raising an exception will break existing code if user_type != header.OA_USER_TYPE: data.pop(tf.OFFER_TYPE, None) track_response = TrackPersister.get_by_tuid(tuid) if not track_response: return track_response product_id = track_response.message[tf.PRODUCT_ID] digital_product_response = \ ows_product_digital.get_product_by_product_id(product_id) if not digital_product_response: return digital_product_response digital_product = digital_product_response.message meta_language_code = \ digital_product.get(product_consts.META_LANGUAGE) if meta_language_code and not _is_valid_metalanguage( tuid, meta_language_code): return api_utils.create_validation_error_response( error.INVALID_META_LANGUAGE_MSG) focus_track = data.get('focus_track', None) if focus_track == 'N': data['focus_track_start_date'] = None data['focus_track_end_date'] = None elif focus_track == 'Y': # if the request values are different from the current add user data if logic_util.focus_track_values_updated(data, track_response.message): data['user_id'] = user_id data['user_type'] = user_type if not features.is_incfeatures_single_focus_track_enabled(): data = logic_util.verify_and_update_focus_track_dates( tuid, product_id, data) if not data: return api_utils.create_validation_error_response( error.INVALID_FOCUS_TRACK_DATE_RANGE_ALREADY_EXISTS_MSG) track_response = TrackPersister.update_track( tuid=tuid, data=data, generate_isrc=generate_isrc, get_artist_info_ids_from_existing_rows=get_artist_info_ids) if not track_response: return track_response if has_lyrics: if not lyrics: lyrics_response = ows_lyrics.delete_track_lyrics(tuid) else: lyrics_response = ows_lyrics.put_track_lyrics(tuid, lyrics) if not lyrics_response: return lyrics_response if patch_response: # Filter out response data to only tuids and updated fields in data if track_response: keys = set(data.keys()) keys.add(tf.TUID) if generate_isrc: keys.add(tf.ISRC) if {'user_type', 'user_id'} <= keys: keys.remove('user_type') keys.remove('user_id') track_response.message = { key: track_response.message[key] for key in keys} return track_response else: res = loc_util.add_track_localizations_to_response(track_response) return _make_track_response(res) def update_track_duration(tuid, data, account_type, account_id): """Update track duration. Args: tuid (int): id of track data (dict): track data to update account_type (str): the user account type in the header. account_id (int): the user account id in the header. """ response = TrackPersister.update_track_duration( tuid=tuid, data=data) return response @logic_util.verify_product_ownership def delete(tuid, account_type, account_id): """Delete track. Args: tuid (int): id of track. account_type (str): the user account type in the header. account_id (int): the user account id in the header. Returns: Response: A Response with status 200 if deleted """ track_response = TrackPersister.get_by_tuid(tuid) if not track_response: return track_response product_id = track_response.message[tf.PRODUCT_ID] audio_attribute_logic.bulk_delete_track_audio_attributes([tuid]) rights_attribute_logic.bulk_delete_track_rights_attributes([tuid]) # OWS Assets retrieves the track data from OWS Track before deleting, # so the asset must be deleted before deleting the track metadata. ows_assets.bulk_delete_track_assets([tuid]) delete_response = TrackPersister.delete_by_tuid( tuid=tuid, product_id=product_id) if not delete_response: return delete_response performer_logic.delete_performers_by_tuid(tuid) ows_product.delete_track_localizations([tuid]) ows_lyrics.delete_track_lyrics(tuid) ows_video.disassociate_tracks([tuid]) return delete_response @logic_util.verify_product_ownership def bulk_delete(tuids, product_id, account_type, account_id, delete_assets=False): """Bulk delete a list of tracks. Args: tuids (list): list of tuids to delete. product_id (int): product id. account_type (str): the user account type in the header. account_id (int): the user account id in the header. delete_assets (bool): if True, also delete track assets via ows_assets. Default False for backward compatibility. Returns: Response: A Response with status 200 if deleted """ # FIXME: Verify list of tuids is valid before deleting assets! audio_attribute_logic.bulk_delete_track_audio_attributes(tuids) rights_attribute_logic.bulk_delete_track_rights_attributes(tuids) if delete_assets: ows_assets.bulk_delete_track_assets(tuids) delete_response = TrackPersister.delete_many_by_tuids( tuids=tuids, product_id=product_id) if not delete_response: return delete_response performer_logic.bulk_delete_performers_by_tuids(tuids) localization_response = ows_product.delete_track_localizations(tuids) if not localization_response: error_msg = '{message} {error}'.format( message=error.OWS_PRODUCT_FAILED_TO_DELETE_LOCALIZATIONS_MSG, error=localization_response.errors.get('message')) g.log.error(error_msg) lyrics_response = ows_lyrics.delete_tracks_lyrics(tuids) if not lyrics_response: error_msg = '{message} {error}'.format( message=error.OWS_PRODUCT_FAILED_TO_DELETE_LYRICS_MSG, error=lyrics_response.errors.get('message')) g.log.error(error_msg) samples_deleted = sample_logic.bulk_delete_samples_by_tuids(tuids) if not samples_deleted: error_msg = '{message} {error}'.format( message=error.OWS_PRODUCT_FAILED_TO_DELETE_SAMPLES_MSG, error=samples_deleted.errors.get('message')) g.log.error(error_msg) video_response = ows_video.disassociate_tracks(tuids) if not video_response: error_msg = '{message} {error}'.format( message=error.OWS_PRODUCT_FAILED_TO_DISASSOCIATE_VIDEO_TRACKS_MSG, error=video_response.errors.get('message')) g.log.error(error_msg) return delete_response @logic_util.verify_product_ownership def get_all_tracks_by_product_id( product_id, account_type, account_id, exclude=None, is_overview=False): """Get a track based on a given product_id. Args: product_id (int): product_id of tracks to retrieve. account_type (str): the user account type in the header. account_id (int): the user account id in the header. exclude (array): list of track fields to exlude. Returns: Response: A Response obj with track JSON data """ res = TrackPersister.get_all_by_product_id(product_id, is_overview=is_overview) exclude = exclude or [] if tf.LOCALIZATIONS not in exclude: res = loc_util.add_track_localizations_to_response(res) return _make_track_response(res) def get_tracks_ids_by_product_ids_with_order( product_ids, account_type, account_id, order_by_fields, ): """Get a track ID based on a given product_id. Args: product_ids (list): product_id of tracks to retrieve. account_type (str): the user account type in the header. account_id (int): the user account id in the header. order_by_fields (iterable): iterable of fields names for ordering. Returns: Response: A Response obj with track JSON data. """ products_tracks = [] tracks = TrackPersister.get_tuids_by_product_ids_with_order( product_ids=product_ids, order_by_fields=order_by_fields, ) tracks_for_product_dict = defaultdict(list) for track in tracks: tracks_for_product_dict[track[track_field.PRODUCT_ID]].append(track) products_tracks = [ tracks_for_product_dict[product_id] for product_id in product_ids ] return api_utils.create_get_nested_lists_response( products_tracks, key_name='tracks', ) @logic_util.verify_product_ownership def get_all_tracks_by_product_id_light( product_id, account_type, account_id): """Get a track based on a given product_id. Args: product_id (int): product_id of tracks to retrieve. account_type (str): the user account type in the header. account_id (int): the user account id in the header. Returns: Response: A Response obj with track JSON data """ res = TrackPersister.get_all_by_product_id_light(product_id) return _make_track_response(res) def get_all_tracks_by_product_ids_medium(product_ids): """Get a tracks based on a given product_ids. Args: product_ids (list): ids of product. Returns: Response: A Response obj with track JSON data """ res = TrackPersister.get_all_by_product_ids_medium(product_ids) return _make_track_response(res) @logic_util.verify_product_ownership def update_field_for_all_tracks_in_product( product_id, data, account_type, account_id, generate_isrc=False): """Update all tracks for product for a field. Yes, the logic here is a bit convoluted. This is unavoidable due to how fields with localizations are updated. The entire localization data for the track needs to be retrieved before it can be updated. TODO: Update function to be able to update multiple fields at once Args: product_id (int): product_id of tracks to arrange. data (dict): contains single field key/value pair to update. account_type (str): the user account type in the header. account_id (int): the user account id in the header. generate_isrc (bool): populate ISRC for track. Returns: Response: A Response obj with track JSON data """ if generate_isrc: return TrackPersister.assign_track_isrcs_for_product_id(product_id) # Remove localization updates localizations_delta = data.pop(tf.LOCALIZATIONS, []) if len(data) != 1: return response.create_error_response( code=error.INVALID_VALUE_ERROR_CODE, message='There can be only one') field_name = next(iter(data)) field_value = data[field_name] response_include_fields = [] if localizations_delta: # Right now only the version field supports localizations # Rest of validation of localization data is done via RAML if field_name != tf.VERSION: return api_utils.create_validation_error_response( error.VALIDATION_ERROR_BAD_LOCALIZATION_DATA_MSG) # Artist data is needed to build localization response response_include_fields = [tf.META_LANGUAGE_CODE, tf.ARTISTS] # FIXME: Validate language_id does not repeat and exist in database # Validate meta language code can be updated if field_name == tf.META_LANGUAGE_CODE and field_value: if not _is_valid_metalanguage_for_product(product_id, field_value): return api_utils.create_validation_error_response( error.META_LANGUAGE_APPLY_TO_ALL_CONFLICT_MSG) track_response = TrackPersister.update_field_for_all_tracks_in_product( product_id, field_name, field_value, response_include_fields=response_include_fields) result = loc_util.patch_tracks_localizations( track_response, localizations_delta) # Clean up any extra data attached by response_include_fields if result and response_include_fields: for item in result.message[api_constants.ITEMS]: for field in response_include_fields: if field in item: del item[field] return result @logic_util.verify_product_ownership def reorder(product_id, arrangement, account_type, account_id): """Reorder tracks to specified arrangement. Args: product_id (int): product_id of tracks to arrange. arrangement (list): list of dicts that has key/value pairs specifying where to arrange each track. Returns: Response: A Response obj with track JSON data """ reorder_response = TrackPersister.reorder(product_id, arrangement) if not reorder_response: return reorder_response product_response = TrackPersister.get_all_by_product_id(product_id) product_body = product_response.message item_list = [] for track in product_body['items']: item = {k: v for k, v in track.items() if k in ( tf.TUID, tf.TRACK_NUMBER, tf.VOLUME_NUMBER)} item_list.append(item) reorder_list = { 'pagination': product_body['pagination'], 'items': item_list } return response.Response(status=200, message=reorder_list) @logic_util.verify_product_ownership def validate_tracks_for_product( product_id, account_type, account_id, user_type, user_id, validation_context='pre_submission', profile_type='', identity_id='', ): """Validate all tracks of a product. Returns the total number of tracks and the number of valid ones. Args: product_id (int): product id. account_type (str): the user account type in the header. account_id (int): the user account id from the header. Returns: response.Response: a Response object with JSON data """ tracks_response = get_all_tracks_by_product_id(product_id) if not tracks_response: return tracks_response tracks = tracks_response.message[api_constants.ITEMS] if not tracks: return api_utils.create_validation_response(tracks, errors=[]) # Get product data orchard_user_id = '{}:{}'.format(user_type, user_id)\ if user_type and user_id else None digital_product_response = ows_product_digital.get_product_by_product_id( product_id, orchard_user_id) if not digital_product_response: return digital_product_response digital_product = digital_product_response.message product_meta_language = \ digital_product.get(product_consts.META_LANGUAGE) genres = logic_util.get_digital_product_genre_type(digital_product) product_genre_type = genres['genre'] product_subgenre_type = genres['subgenre'] product_release_date = digital_product.get('release_date') is_correction_mode = product_utils.is_in_correction_mode(digital_product) track_corrections = {} product_artists = digital_product.get('product_artists', []) product_featuring_artists = product_utils.get_featuring_artists(product_artists, 'role') product_primary_artists = product_utils.get_primary_artists(product_artists, 'role') product_name = digital_product.get('product_name') product_version = digital_product.get('delivered_version') or '' product_imprint = digital_product.get('imprint') or '' if is_correction_mode: track_corrections = get_track_corrections(digital_product) product_metalanguage_corrections = \ get_product_metalanguage_correction(digital_product) product_metalanguage_data = product_metalanguage_corrections.get(product_id, {}) for meta_language_key, meta_language_value in product_metalanguage_data.items(): if meta_language_value: product_meta_language = meta_language_value product_featuring_artist_corrections = \ get_product_featuring_artist_corrections(digital_product) if product_featuring_artist_corrections: product_featuring_artists = get_product_artist_correction_names( product_featuring_artist_corrections, product_id) product_primary_artist_corrections = \ get_product_primary_artist_corrections(digital_product) if product_primary_artist_corrections: product_primary_artists = get_product_artist_correction_names( product_primary_artist_corrections, product_id) product_name_correction = \ get_product_name_correction(digital_product) if product_name_correction: product_name = product_name_correction product_version_correction = \ get_product_version_correction(digital_product) if product_version_correction is not None: product_version = product_version_correction product_imprint_correction = \ get_product_imprint_correction(digital_product) if product_imprint_correction is not None: product_imprint = product_imprint_correction # Get Assets product_assets = None product_response = ows_product.get_product_by_product_id(product_id) if not product_response: return product_response vendor_id = product_response.message.get('vendor_id') assets_response = ows_assets.get_product_assets_v2(product_id) if assets_response: product_assets = assets_response.message[api_constants.ITEMS] spatial_isrc_by_track_id = {} if features.is_cdam_spatial_isrc_validation_enabled(): track_ids = [t['tuid'] for t in tracks] spatial_response = track_spatial_model.get_spatial_isrc_map_by_track_ids(track_ids) if not spatial_response: return spatial_response spatial_isrc_by_track_id = spatial_response.message or {} used_track_names = set() errors = [] warnings = [] for idx, track in enumerate(tracks): track_id = track['tuid'] # we also modify the original object tracks[idx] here so corrections are always considered # after this operation track = tracks[idx] = ChainMap(track_corrections.get(track_id, {}), track) error_obj = track_validators.validate_track_complete( product_featuring_artists, track, product_genre_type, product_subgenre_type, product_release_date, product_assets, used_track_names, product_meta_language, is_correction_mode, tracks, spatial_isrc_by_track_id, track_corrections, profile_type, identity_id) # Updated set of used track names when no errors in track name if tf.TRACK_NAME not in error_obj: used_track_names.add( track_utils.get_track_artists_name_ver_artists_tpl(track)) if error_obj: errors.append(error_obj) if len(tracks) == 1 and \ features.is_single_track_and_product_mismatch_enabled(vendor_id): track = tracks[0] error_objs = [] if product_genre_type != product_consts.CLASSICAL_GENRE: error_objs.append(( track_validators.validate_track_name_matches_product_name( track, product_name ), tf.TRACK_NAME )) error_objs.append(( track_validators.validate_track_version_matches_product_version( track, product_version ), tf.VERSION )) error_objs.append(( track_validators.validate_track_primary_artists_matches_product_primary_artists( track, product_primary_artists), tf.PERFORMER )) error_objs.append(( track_validators.validate_track_featuring_artists_matches_product_featuring_artists( track, product_featuring_artists), tf.FEATURING)) is_classical = product_genre_type == product_consts.CLASSICAL_GENRE is_composer_required_soundtrack = product_genre_type == product_consts.SOUNDTRACK_GENRE and \ product_subgenre_type in product_consts.SOUNDTRACK_SUBGENRE_IDS.values() is_composer_required_world_music = product_genre_type == product_consts.WORLD_MUSIC_GENRE and \ product_subgenre_type in product_consts.WORLD_MUSIC_SUBGENRE_IDS.values() product_details = [product_id, product_artists, is_correction_mode, digital_product, track] if features.is_single_track_composer_mismatch_enabled(vendor_id) and \ (is_classical or is_composer_required_soundtrack or is_composer_required_world_music): _validate_track_artists_by_key(artist_role.COMPOSER, error_objs, *product_details) if is_classical and features.is_single_track_classical_artists_mismatch_enabled(vendor_id): _validate_track_artists_by_key(artist_role.ORCHESTRA, error_objs, *product_details) _validate_track_artists_by_key(artist_role.CONDUCTOR, error_objs, *product_details) _validate_track_artists_by_key(artist_role.ENSEMBLE, error_objs, *product_details) for err in error_objs: if err[0]: if len(errors): errors[0][err[1]] = err[0][err[1]] else: errors.append(err[0]) try: post_submission_validation( validation_context, product_id, vendor_id, errors, warnings, tracks, product_imprint ) except RequestError as e: return api_utils.create_error_response( status=e.http_status, code=e.error_code, message=e.message ) return api_utils.create_validation_response( tracks, errors, warnings=warnings) def _post_submission_validation_isrc_usage(product_id): isrc_usage_stats = isrc_logic.get_isrc_usage_stats(product_id) return [ { tf.ISRC: { 'validator': validation_constants.MAX_ISRC_REUSE, 'validator_value': validation_constants.MAX_ISRC_REUSE_VALUE, 'message': json.dumps(track), 'code': error.EXCESSIVE_ISRC_REUSE, }, 'tuid': track['id'] } for track in isrc_usage_stats['product']['tracks'] if track['isrc_usage']['count'] > validation_constants.MAX_ISRC_REUSE_VALUE ] def _post_submission_validation_potential_audio_infringement(product_id, tracks, imprint): def __check_if_self_match(result, match, tracks_dict): track = tracks_dict[result['track_id']] track_metadata = { 'track_name': track.get('track_name'), 'artist_names': track_utils.get_track_primary_artist_names(track), 'imprint': imprint } match_metadata = { 'track_name': match.get('title'), 'artist_names': match.get('artists'), 'imprint': match.get('label') } return track_metadata == match_metadata match_audio_results = ows_assets.get_match_audio_results(product_id) tracks_dict = {} for track in tracks: tracks_dict[track['tuid']] = track for result in match_audio_results: filtered_matches = [] for match in result.get('matches', []): if not __check_if_self_match(result, match, tracks_dict): filtered_matches.append(match) result['matches'] = filtered_matches return [ { services_constants.AUDIO_FILE: { 'validator': validation_constants.POTENTIAL_AUDIO_INFRINGEMENT, 'validator_value': validation_constants.POTENTIAL_AUDIO_INFRINGEMENT_VALUE, 'message': json.dumps( { key: match_audio_result[key] for key in ['asset_final_id', 'matches'] } ), 'code': error.POTENTIAL_AUDIO_INFRINGEMENT, }, 'tuid': match_audio_result['track_id'] } for match_audio_result in match_audio_results if match_audio_result['code'] == 'WARNING_MATCHES_FOUND' and match_audio_result['matches'] ] def _post_submission_validation_ai_generated_audio_suspected(product_id, as_error=False): """Build AI-generated audio notices for tracks flagged with suspected AI audio. When ``as_error`` is True the notices are emitted as hard-blocker track errors so the review UI categorizes them as a Blocker and disables approval; the inner payload uses ``error_code`` to match the track error shape consumed downstream. Otherwise they are emitted as warnings (keyed by ``code``). """ ai_audio_results = ows_assets.get_ai_generated_audio_results(product_id) code_key = 'error_code' if as_error else 'code' return [ { validation_constants.AI_GENERATED_AUDIO: { 'validator': validation_constants.AI_GENERATED_AUDIO, 'validator_value': validation_constants.AI_GENERATED_AUDIO_VALUE, 'message': json.dumps( { key: ai_audio_result[key] for key in ['asset_final_id'] } ), code_key: error.AI_GENERATED_AUDIO, }, 'tuid': ai_audio_result['track_id'], } for ai_audio_result in ai_audio_results if ai_audio_result['code'] == 'AI_GENERATED_AUDIO_SUSPECTED' ] def _post_submission_validation_spotify_watchlist_artists(tracks): """Check track artists against the Spotify watchlist via ows-blacklist-manager. Args: tracks (list): list of track dicts from the product. Returns: list: warning dicts keyed by tuid for any matched watchlist artists. """ payload = { 'track_artists': [ { tf.TUID: track[tf.TUID], tf.ARTISTS: track_utils.get_artists_with_corrections(track) } for track in tracks ] } validation_response = ows_blacklist_manager.validate_artists(payload) track_artist_results = validation_response.get('validation_errors', {}).get('track_artists', []) return [ { validation_constants.SPOTIFY_WATCHLIST_ARTIST: { 'validator': validation_constants.SPOTIFY_WATCHLIST_ARTIST, 'validator_value': validation_constants.SPOTIFY_WATCHLIST_ARTIST_VALUE, 'message': json.dumps(track_artist['matches']), 'code': error.SPOTIFY_WATCHLIST_ARTIST, }, 'tuid': track_artist['tuid'], } for track_artist in track_artist_results if track_artist.get('matches') ] def post_submission_validation(validation_context, product_id, vendor_id, errors, warnings, tracks, imprint): """If applicable, run post_submission validation.""" if validation_context != 'post_submission': return isrc_usage_warnings = _post_submission_validation_isrc_usage(product_id) potential_audio_infringement_warnings = _post_submission_validation_potential_audio_infringement( product_id, tracks, imprint ) suspected_ai_generated_audio_errors = [] suspected_ai_generated_audio_warnings = [] if features.is_ai_generated_audio_error_enabled(vendor_id): suspected_ai_generated_audio_errors = _post_submission_validation_ai_generated_audio_suspected( product_id, as_error=True) elif features.is_ai_generated_audio_validation_enabled(vendor_id): suspected_ai_generated_audio_warnings = _post_submission_validation_ai_generated_audio_suspected(product_id) spotify_watchlist_artist_warnings = [] if features.is_spotify_watchlist_artist_validation_enabled(vendor_id): spotify_watchlist_artist_warnings = _post_submission_validation_spotify_watchlist_artists(tracks) cross_account_osr_conflict_warnings = [] cross_track_isrc_mismatch_warnings = [] cross_isrc_osr_mismatch_warnings = [] if product_id not in EXCLUDE_PRODUCT_ID_OSR_VALIDATION: sound_recording_matches = ows_sound_recordings.get_formatted_sound_recording_matches(product_id) cross_account_osr_conflict_warnings = get_cross_account_osr_conflict_warnings( vendor_id, sound_recording_matches ) cross_track_isrc_mismatch_warnings = get_cross_track_isrc_mismatch_warnings(sound_recording_matches) cross_isrc_osr_mismatch_warnings = get_cross_isrc_osr_mismatch_warnings(product_id, tracks) track_id_to_warnings = defaultdict(dict) for warning in ( isrc_usage_warnings + potential_audio_infringement_warnings + cross_account_osr_conflict_warnings + cross_track_isrc_mismatch_warnings + cross_isrc_osr_mismatch_warnings + suspected_ai_generated_audio_warnings + spotify_watchlist_artist_warnings ): track_id_to_warnings[warning['tuid']].update(warning) warnings.extend(track_id_to_warnings.values()) # AI-generated audio is a hard blocker, so merge it into the per-track errors. _merge_post_submission_track_errors(errors, suspected_ai_generated_audio_errors) def _merge_post_submission_track_errors(errors, post_submission_errors): """Merge post-submission track errors into the per-track errors list, keyed by tuid. Each entry is keyed by validator/field name with a ``tuid`` key. The downstream formatter only reads the first error entry per track, so post-submission errors are merged into any existing entry for the same track rather than appended separately. """ errors_by_tuid = { track_error[tf.TUID]: track_error for track_error in errors if tf.TUID in track_error } for post_submission_error in post_submission_errors: tuid = post_submission_error[tf.TUID] existing_error = errors_by_tuid.get(tuid) if existing_error is not None: existing_error.update(post_submission_error) else: errors.append(post_submission_error) errors_by_tuid[tuid] = post_submission_error def get_cross_track_isrc_mismatch_warnings(sound_recording_matches): """Get cross track isrc mismatch warnings.""" return validate_cross_track_isrc_mismatch(sound_recording_matches) def validate_cross_track_isrc_mismatch(sound_recording_matches): """Validate cross track isrc mismatch.""" matched_tracks = [] for tuid, data in sound_recording_matches.items(): isrc = data['isrc'] filtered_matches = [] for matched_track in data['matched_tracks']: if isrc != matched_track.get('isrc'): filtered_matches.append(matched_track) if filtered_matches: matched_tracks.append({ validation_constants.CROSS_TRACK_ISRC_MISMATCH: { 'validator': validation_constants.CROSS_TRACK_ISRC_MISMATCH, 'validator_value': True, 'message': json.dumps({'matches': filtered_matches}), 'code': validation_constants.CROSS_TRACK_ISRC_MISMATCH, }, 'tuid': tuid }) return matched_tracks def get_cross_account_osr_conflict_warnings(vendor_id, sound_recording_matches): """Get cross account osr conflict warnings.""" return validate_cross_account_osr_conflict(sound_recording_matches, vendor_id) def validate_cross_account_osr_conflict(sound_recording_matches, vendor_id): """Validate cross account OSR conflict.""" matched_tracks = [] company_brand = ows_account.get_vendor_company_brand(vendor_id) for tuid, data in sound_recording_matches.items(): filtered_matches = [] for matched_track in data['matched_tracks']: if vendor_id == matched_track.get('vendor_id'): continue matched_track['company_brand'] = company_brand.message filtered_matches.append(matched_track) if filtered_matches: matched_tracks.append({ validation_constants.CROSS_ACCOUNT_OSR_CONFLICT: { 'validator': validation_constants.CROSS_ACCOUNT_OSR_CONFLICT, 'validator_value': True, 'message': json.dumps({'matches': filtered_matches}), 'code': validation_constants.CROSS_ACCOUNT_OSR_CONFLICT, }, 'tuid': tuid }) return matched_tracks def get_cross_isrc_osr_mismatch_warnings(product_id, tracks): """Get potential isrc misuse warnings.""" tuids_to_isrcs = {track['tuid']: track['isrc'] for track in tracks} isrc_matches = ows_sound_recordings.get_formatted_isrc_matches( product_id, list(tuids_to_isrcs.values()), tuids_to_isrcs.keys()) return validate_cross_isrc_osr_mismatch(isrc_matches, tuids_to_isrcs) def validate_cross_isrc_osr_mismatch(isrc_matches, tuids_to_isrcs): """Validate potential isrc misuse.""" results = [] for current_tuid, current_osr_id in isrc_matches['tuids_to_osr_ids'].items(): if current_osr_id is None: # If the current product isn't attached to a sound recording, # the current_osr_id will be `None`. # To show this validation in that case, remove this if/continue block continue matched_tracks = get_same_isrc_different_osr( current_osr_id, tuids_to_isrcs.get(current_tuid), isrc_matches) if matched_tracks: results.append({ validation_constants.CROSS_ISRC_OSR_MISMATCH: { 'validator': validation_constants.CROSS_ISRC_OSR_MISMATCH, 'validator_value': True, 'message': json.dumps({'matches': matched_tracks}), 'code': validation_constants.CROSS_ISRC_OSR_MISMATCH, }, 'tuid': current_tuid }) return results def get_same_isrc_different_osr(current_osr_id, product_track_isrc, isrc_matches): """Get list of tracks with same isrc and different osr for a given track.""" results = [] for osr_id, track_matches in isrc_matches['osrs_to_tracks'].items(): if current_osr_id and osr_id == current_osr_id: continue for track in track_matches: if track['isrc'] == product_track_isrc: results.append(track) return results def import_tracks( request_data, product_id, orchard_user_id, account_type, account_id, correlation_id): """Import existing tracks. Args: request_data (dict): Request data with a list of TUIDs. product_id (int): product id. orchard_user_id (str): ALW user id. account_type (str): the user account type in the header. account_id (int): the user account id from the header. correlation_id (str): UUID used for logging. Returns: response.Response: a Response object with JSON data """ # Validate the product ownership and find out UPC by product id product_response = logic_util.get_product_data_and_validate_ownership( product_id, account_type, account_id) if not product_response: return product_response upc = product_response.message[tf.UPC] source_tuids = request_data[field_const.TUIDS] tracks_response = TrackPersister.get_multiple_by_tuids(source_tuids) if not tracks_response: return tracks_response source_product_ids = set() for track in tracks_response.message[api_constants.ITEMS]: source_product_ids.add(track[tf.PRODUCT_ID]) # Verify imported tracks ownership products_ownership_response = logic_util.verify_products_ownership( source_product_ids, account_type, account_id) if not products_ownership_response: return products_ownership_response import_response = TrackPersister.import_tracks( source_tuids, product_id, upc) return _process_import_response( import_response, orchard_user_id, correlation_id, product_id) def copy_tracks_for_product( product_id, new_product_id, orchard_user_id, correlation_id, track_list=None): """Logic for copy tracks for digital product. Args: product_id (int): the product_id of the product to copy from. new_product_id (int): the product_id of the product to copy to. orchard_user_id (str): ALW user id. correlation_id (str): UUID used for logging. track_list (list): List of track ids that should be copied. Returns: response.Response """ # Get UPC by new_product_id product_response = ows_product.get_product_by_product_id(new_product_id) if not product_response: return product_response upc = product_response.message[tf.UPC] import_response = TrackPersister.copy_product_tracks( src_product_id=product_id, dest_product_id=new_product_id, dest_upc=upc, exclude_fields=[], track_list=track_list ) return _process_import_response( import_response, orchard_user_id, correlation_id, new_product_id) def get_track_by_upc_and_isrc(upc, isrc): """Get track by upc and isrc. Args: upc (str): UPC of the track's product isrc (str): ISRC of the track Returns: Response: A Response obj with the full track data """ return TrackPersister.get_by_upc_and_isrc(upc=upc, isrc=isrc) def get_all_tracks_by_type_and_isrc(track_type, isrc): """Retrieve all tracks which have the specified type and ISRC. Args: track_type (str): type of the tracks to retrieve isrc (str): ISRC of the tracks to retrieve Returns: response.Response: list of tracks """ items = TrackPersister.get_all_by_isrc_and_type(isrc, track_type) return api_utils.create_get_list_response(items) def get_account_tracks_by_type_and_isrc(account_type, account_id, track_type, isrc): """Retrieve all tracks for an account which have the specified type and ISRC. Args: account_type (str): the type of the account, either vendor or subaccount account_id (int): the id of the account track_type (str): type of the tracks to retrieve isrc (str): ISRC of the tracks to retrieve Returns: response.Response: list of tracks """ tracks = TrackPersister.get_all_for_account_by_isrc_and_type( account_type, account_id, isrc, track_type, ) return _make_track_response(tracks) def _process_import_response(import_response, orchard_user_id, correlation_id, dest_product_id): """Copy assets and localizations for imported tracks. Args: import_response (Response): Response from TrackPersister after import. orchard_user_id (str): ALW user id. correlation_id (str): UUID used for logging. dest_product_id (int): Product Id. Returns: response.Response """ if not import_response: return import_response import_source_dest_list = import_response.message[api_constants.ITEMS] logic_util.add_tracks_log_entries( import_source_dest_list, orchard_user_id, action='import') # Copy assets for successfully imported tracks source_dest_tuid_list = [ (import_dict['source'][tf.TUID], import_dict['destination'][tf.TUID],) for import_dict in import_source_dest_list] logic_util.copy_tracks_assets_in_parallel( source_dest_tuid_list, orchard_user_id, correlation_id) # Copy performers. performers_util.copy_tracks_performers(source_dest_tuid_list) sample_logic.copy_tracks(source_dest_tuid_list) ows_lyrics.copy_tracks_lyrics(import_source_dest_list) # create release_spatial record. create_release_spatial_response = _create_release_spatial_if_spatial_isrc_present( import_source_dest_list, dest_product_id, orchard_user_id ) if not create_release_spatial_response: return create_release_spatial_response copy_localization_response = loc_util.copy_tracks_localizations( import_source_dest_list) if copy_localization_response: return _make_track_response(copy_localization_response) # Localizations weren't copied imported_tracks = [ import_dict['destination'] for import_dict in import_source_dest_list] return api_utils.create_get_list_response( imported_tracks, status=response_code.OK) def _create_release_spatial_if_spatial_isrc_present(import_source_dest_list, dest_product_id, orchard_user_id): # create release_spatial record if any single spatial track found. dest_tuids = [ import_dict['destination'][tf.TUID] for import_dict in import_source_dest_list ] spatial_isrc_map = track_spatial_model.get_spatial_isrc_map_by_track_ids(dest_tuids) if not spatial_isrc_map: return spatial_isrc_map if not any(spatial_isrc_map.message.get(dest_tuid) for dest_tuid in dest_tuids): return response.Response() existing_release_spatial = ows_product_digital.get_product_spatial(dest_product_id) if existing_release_spatial.status == response_code.OK: return response.Response() if existing_release_spatial.status == response_code.NOT_FOUND: create_product_spatial_response = ows_product_digital.create_product_spatial( dest_product_id, orchard_user_id ) if not create_product_spatial_response: return create_product_spatial_response return response.Response() g.log.error(f'Failed to fetch spatial data for product {dest_product_id}') return existing_release_spatial def get_track_by_track_artist_id(track_artist_id): """Get track by track_artist_id logic. Args: track_artist_id (int): Track artist id. Returns: response.Response """ return TrackPersister.get_track_by_track_artist_id(track_artist_id) def claim_new_isrc(): """Claim a new isrc. Returns: response.Response: Contains status code and payload with ISRC (str). """ return TrackPersister.claim_new_isrc() def _is_valid_metalanguage_for_product( product_id, meta_language_code): """Validate meta language can be used for product tracks. There should be no localizations for any of the product tracks that conflict with the meta language. Args: product_id (int): id of product. meta_language_code (str): language_code for track meta language. Returns: bool """ if not validation_util.is_meta_language_code_format(meta_language_code): return False meta_language_id = loc_util.code_to_language_id(meta_language_code) if not meta_language_id: # Not every valid meta_metalanguage_code can be converted to an id return True tracks_response = get_all_tracks_by_product_id( product_id) for track in tracks_response.message[api_constants.ITEMS]: for localization in track[tf.LOCALIZATIONS]: if meta_language_id == localization[tf.LANGUAGE_ID]: return False return True def _is_valid_metalanguage(tuid, meta_language_code): """Validate meta language does not conflict with existing localization. Args: tuid (int): unique id of track. meta_language_code (str): language_code for track meta language. Returns: bool """ if not validation_util.is_meta_language_code_format(meta_language_code): return False meta_language_id = loc_util.code_to_language_id(meta_language_code) if not meta_language_id: # Not every valid meta_metalanguage_code can be converted to an id return True localizations_res = ows_product.get_track_localizations(tuid) if not localizations_res: return False for localization in localizations_res.message[api_constants.ITEMS]: if meta_language_id == localization[tf.LANGUAGE_ID]: return False return True def _make_track_response(persister_response): """Update response object so proper set of track fields are returned. This function copies the original input. Args: persister_response (Response): Response object Returns: response """ res = logic_util.copy_response(persister_response) if not res: return res if api_constants.ITEMS in res.message: for track in res.message[api_constants.ITEMS]: _make_track_response_dict(track) else: _make_track_response_dict(res.message) return res def _make_track_response_dict(track): """Make the proper response dictionary for a track. This function modifies the original input. Args: track (dict): Track dictionary """ if tf.LOCALIZATIONS not in track: track[tf.LOCALIZATIONS] = [] for field in ( tf.PUBLISHERS, tf.US_PUBLISHING_OBLIGATION, tf.THIRD_PARTY_PUBLISHER): if field in track: del track[field] def check_profile_track_access(profile_uuid, tuid): """Check profile access to a track.""" # get product id for track track_response = TrackPersister.get_by_tuid(tuid) if not track_response: return track_response product_id = track_response.message[tf.PRODUCT_ID] # get label id for product product_response = ows_product.get_product_by_product_id(product_id) if not product_response: return product_response label_id = product_response.message.get('vendor_id') # return ows permissions check return ows_permissions.check_profile_access_to_label(profile_uuid, label_id) def _validate_track_artists_by_key( key, error_objs, product_id, product_artists, is_correction_mode, digital_product, track): """Validate artists for a given artist type key.""" product_artists = product_utils.get_artists_by_key(key, product_artists, 'role') if is_correction_mode: product_artist_correction = \ get_product_artist_corrections_for_key(key, digital_product) if product_artist_correction: product_artists = get_product_artist_correction_names( product_artist_correction, product_id) error_objs.append(( track_validators.validate_track_artists_match_product_artists_for_key( key, track, product_artists), key))