"""Validations for audio product fields.""" from collections import defaultdict import datetime import json import math import regex from jsonschema import Draft3Validator from oto import response from product_digital import config from product_digital.constants import error from product_digital.constants import validations from product_digital.constants import warning from product_digital.constants.metadata import ( DICTIONARY, METADATA_FIELD, METADATA_VALUE, MIN_LENGTH, NAME, PRODUCT_ARTISTS, PRODUCT_NAME, SEARCH_RATIO, ROLE, TRACK_NAME, TUID, ) from product_digital.constants.valid_dates import MAX_VALID_DATE, MIN_VALID_DATE from product_digital.constants.error_correction import FIELD_NAME_MAP from product_digital.constants.product import FEATURING, PRODUCER, PERFORMER, REMIXER, ValidationType from product_digital.models import audio_product as audio_product_model from product_digital.models import content_review_account_blocklist from product_digital.models import language as language_model from product_digital.models import ows_assets from product_digital.models import ows_blocklist_manager from product_digital.models import ows_product from product_digital.models import ows_product_workflow from product_digital.models import ows_pricing from product_digital.models import ows_track from product_digital.models import project as project_model from product_digital.models import release from product_digital.models import release_spatial as release_spatial_model from product_digital.validation import json_schema def validate_product_on_create(product_data): """Validate the contents of a product payload on create. Args: product_data (dict): payload of audio product data Returns: response.Response: response with 200 status if valid """ errors = {} if product_data.get('product_code'): product_code_validation = _validate_product_code_unique(product_data) if not product_code_validation: return product_code_validation errors.update(product_code_validation.message) if product_data.get('upc'): upc_validation = _validate_upc_available(product_data.get('upc')) if not upc_validation: return upc_validation errors.update(upc_validation.message) if errors: return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=errors) return response.Response() def validate_product_on_update(product_data): """Validate the contents of a product payload. Only used for partial updates now, so all fields are treated as optional. Args: product_data (dict): payload of audio product data to validate Returns: response.Response: response with 200 status if valid or 400 if not """ errors = {} errors.update(_validate_dates(product_data)) if 'product_code' in product_data: product_code_validation = _validate_product_code_unique(product_data) if not product_code_validation: return product_code_validation errors.update(product_code_validation.message) if errors: return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=errors) return response.Response() def validate_product_basics( product_basics, tracks_data=None, single_track_product=None, ff_context=None): """Check the completeness of product basics data. Used for insuring that all required product basics are stored and valid. Args: product_basics (dict): the product basics data to validate featuring_artist_mismatch (bool): featuring artist mismatch feature flag Returns: response.Response: 200 response with validation payload """ basics_errors = {} validator = Draft3Validator(config.PRODUCT_BASICS_DEFINITION) json_validation = json_schema.validate( _remove_blank_values(product_basics), validator) if not json_validation: basics_errors.update(json_validation.errors['message']) artists = product_basics.get('product_artists', []) artist_validation = _validate_required_artist_role( artists, 'primary_artist') if not artist_validation: basics_errors.update(artist_validation.errors['message']) product_name = product_basics.get('product_name') if ff_context.product_name_length_check and len(product_name) > 255: basics_errors.update({ 'product_name': _format_error( validator='productNameLengthError', validator_value=True, message=error.ERROR_MESSAGE_PRODUCT_NAME_LENGTH, error_code=error.ERROR_CODE_PRODUCT_NAME_LENGTH ) }) tracks_corrections = product_basics.get('corrections_by_track_id', {}) product_primary_artists = _get_product_artist_names_by_type( artists, FIELD_NAME_MAP[PERFORMER]) product_featuring_artists = _get_product_artist_names_by_type( artists, FIELD_NAME_MAP.get(FEATURING, FEATURING)) product_remixers = _get_product_artist_names_by_type( artists, FIELD_NAME_MAP.get(REMIXER, REMIXER)) product_producers = _get_product_artist_names_by_type( artists, FIELD_NAME_MAP.get(PRODUCER, PRODUCER)) if (ff_context.featuring_artist_mismatch and not _validate_various_artists(product_featuring_artists) and not _validate_featuring_artists( product_featuring_artists, tracks_data, tracks_corrections)): basics_errors.update({ FEATURING: _format_error( validator='featuringArtistMismatch', validator_value=True, message=error.ERROR_MESSAGE_FEATURING_ARTIST_MISMATCH, error_code=error.ERROR_CODE_FEATURING_ARTIST_MISMATCH ) }) if ((ff_context.primary_artist_mismatch_error and not _is_subgenre_theatre_scores(product_basics)) and not _validate_primary_artists( product_primary_artists, tracks_data, tracks_corrections)): basics_errors.update(primary_artist=_format_error( validator='primaryArtistMismatch', validator_value=True, message=error.ERROR_MESSAGE_PRIMARY_ARTIST_MISMATCH_ERROR, error_code=error.ERROR_CODE_PRIMARY_ARTIST_MISMATCH_ERROR ) ) if (ff_context.various_artist_on_featuring and _validate_various_artists(list(product_featuring_artists))): basics_errors.update({ FEATURING: _format_error( validator='variousArtistsOnFeaturingArtist', validator_value=True, message=error.ERROR_MESSAGE_VARIOUS_ARTISTS, error_code=error.ERROR_CODE_VARIOUS_ARTISTS_ON_FEATURING ) }) if ff_context.single_track_product_various_artists and single_track_product: if _validate_various_artists(list(product_primary_artists)): basics_errors.update(primary_artist=_format_error( validator='stVariousArtists', validator_value=True, message='Various Artists is not allowed on single track products', error_code='st_various_artists_res' )) if (ff_context.remixer_mismatch and not _validate_remixers( product_remixers, tracks_data, tracks_corrections) ): basics_errors.update({ REMIXER: _format_error( validator='productAndTrackRemixersMismatch', validator_value=True, message=error.ERROR_MESSAGE_REMIXER_MISMATCH, error_code=error.ERROR_CODE_REMIXER_MISMATCH ) }) if ff_context.producer_mismatch and not _validate_producers(product_producers, tracks_data, tracks_corrections): basics_errors.update({ PRODUCER: _format_error( validator_value=True, validator='productAndTrackProducersMismatch', error_code=error.ERROR_CODE_PRODUCER_MISMATCH, message=error.ERROR_MESSAGE_PRODUCER_MISMATCH ) }) genre_id = product_basics.get('genre_id') if 'subgenre_id' in product_basics and \ _composer_required(product_basics) or \ genre_id == config.CLASSICAL_GENRE_ID: composer_validation = _validate_required_artist_role( artists, 'composer') if not composer_validation: basics_errors.update(composer_validation.errors['message']) if ff_context.release_level_lyricist and \ 'genre_id' in product_basics and \ 'subgenre_id' in product_basics and \ _lyricist_required(product_basics): lyricist_validation = _validate_required_artist_role( artists, 'lyricist') if not lyricist_validation: basics_errors.update(lyricist_validation.errors['message']) if 'meta_language' not in basics_errors: meta_language = get_latest_meta_language(product_basics) if _meta_language_is_invalid(meta_language): basics_errors.update(meta_language=_format_error( validator='invalid', validator_value=True, message='meta_language must be a valid value', error_code='meta_language_invalid' )) return response.Response( _create_validation_message( validation_type=ValidationType.PRODUCT_BASICS, errors=basics_errors, ) ) def validate_product_dates(scheduling_data): """Validate product dates for product submission. Args: scheduling_data (dict): dictionary of product scheduling meta-data Returns: response.Response: 200 response with validation payload """ scheduling_and_pricing_errors = {} validator = Draft3Validator(config.PRODUCT_DATES_DEFINITION) json_validation = json_schema.validate( _remove_blank_values(scheduling_data), validator) if not json_validation: scheduling_and_pricing_errors.update(json_validation.errors['message']) return response.Response( _create_validation_message( validation_type=ValidationType.SCHEDULING_AND_PRICING, errors=scheduling_and_pricing_errors, ) ) def validate_artwork(product_id, require_v1_artwork=False): """Validate artwork for a product. If artwork has not been uploaded, the result of the validation will be a not found response with status = 404. Args: product_id (int): the unique id for the product to validate require_v1_artwork (bool): is v1 artwork required (in addition to v2). Returns: response.Response: the outcome of the validation call. """ artwork_response = ows_assets.validate_artwork_v2(product_id, require_v1_artwork) errors = [] if artwork_response.status == 404: errors.append('Artwork has not been uploaded successfully.') elif not artwork_response: return artwork_response return response.Response(_create_validation_message( validation_type=ValidationType.ARTWORK, errors=errors, )) def validate_tracks( product_id, orchard_user_id='', ): """Validate tracks for a product. Args: product_id (int): the unique id for the product to validate. orchard_user_id (string): user id of an orchard user. Returns: response.Response: the outcome of the validation call. """ tracks_response = ows_track.validate_tracks( product_id, orchard_user_id ) if not tracks_response: return tracks_response return response.Response(_create_tracks_validation_message( validation_type=ValidationType.TRACKS, ows_track_validation_results=tracks_response.message, )) def validate_release_correction(product_id, release_correction_id): """Validate release_correction has the correct product id. Args: product_id (int): id of the product to validate against. release_correction_id (int): id of the release correction for the given product. Returns: boolean: indicating if the release correction product_id is valid. """ correction_response = ows_product_workflow.get_release_correction( release_correction_id=release_correction_id, ) if correction_response.status != 200: return False, {} release_id = int(correction_response.message.get('release_id')) return release_id == product_id, correction_response.message def validate_publishing_obligation( product_id, account_type, account_id, orchard_user_id=''): """Validate publishing obligation information for a product. Args: product_id (int): the product relating the obligations to be confirmed account_type (str): the user account type in the header. account_id (int): the user account id in the header. orchard_user_id (string): user id of an orchard user. Returns: response.Response: the outcome of the validation call. """ publishing_response = ows_track.validate_publishing_obligation( product_id, account_type, account_id, orchard_user_id) if not publishing_response: return publishing_response return response.Response(_create_tracks_validation_message( validation_type=ValidationType.PUBLISHING_OBLIGATION, ows_track_validation_results=publishing_response.message, )) def validate_product_blocklist(product_id): """Validate blocklisted words for a product. Args: product_id (int): the product to validate Returns: response.Response: the outcome of the validation call. """ validation_response = ows_blocklist_manager.validate_product(product_id=product_id) if not validation_response: return validation_response validation_error = validation_response.message.get('validation_error', {}) return response.Response( _create_validation_message( validation_type=ValidationType.PRODUCT_BLOCKLIST, errors=validation_error.get('items', []), ) ) def validate_account(product_id): """Validate account that product belongs to. Args: product_id (int): id of product to validate Returns: response.Response: the outcome of the validation call """ product_response = audio_product_model.get_product(product_id) if not product_response: return product_response product = product_response.message project_id = product['project_id'] project_response = project_model.get_project_by_id(project_id) if not project_response: return project_response project = project_response.message vendor_id = project['vendor_id'] subaccount_id = project['subaccount_id'] validation_response = content_review_account_blocklist.is_vendor_or_subaccount_on_blocklist( vendor_id=vendor_id, subaccount_id=subaccount_id, ) if not validation_response: return validation_response is_vendor_or_subaccount_on_blocklist = validation_response.message errors = [] if is_vendor_or_subaccount_on_blocklist: errors.append( _format_validation_result( code='account_blocklist', reason=( 'This content is from a blocked account and cannot be approved. ' 'Please reach out to QC, Operations or Contract Admin for more information.' ) ) ) return response.Response( _create_validation_message( validation_type=ValidationType.ACCOUNT, errors=errors, ) ) def validate_spatial_upc(product_id, orchard_user_id, account_type='', account_id=''): """Validate that a spatial UPC record exists when the product has spatial audio assets. Args: product_id (int): the unique id for the product to validate. orchard_user_id (str): user id of an orchard user. account_type (str): the user account type (vendor/subaccount). account_id (str): the user account id. Returns: response.Response: the outcome of the validation. """ assets_response = ows_assets.get_assets( product_id, orchard_user_id, account_type=account_type, account_id=account_id) if not assets_response: return assets_response assets = assets_response.message.get('assets', []) has_atmos = any(a.get('asset_upload_type') == 'atmos' for a in assets) errors = [] if has_atmos: spatial_response = release_spatial_model.get_release_spatial(product_id) if spatial_response.status == 404: errors.append( _format_validation_result( code=error.ERROR_CODE_SPATIAL_UPC_REQUIRED, reason=error.ERROR_MESSAGE_SPATIAL_UPC_REQUIRED, ) ) elif not spatial_response: return spatial_response return response.Response( _create_validation_message( validation_type=ValidationType.SPATIAL_UPC, errors=errors, ) ) def get_latest_meta_language(product_basics): """Get current meta_language, considering corrections.""" if product_basics.get('corrections', {}).get('items'): for correction in product_basics['corrections']['items']: if ( correction.get('table_name') == 'releases' and correction.get('field_name') == 'meta_language' ): return correction.get('key_value') return product_basics.get('meta_language') def get_tracks_by_product_id(product_id): """Get tracks by product id. Args: product_id (int): product id Returns: list: List of track data """ tracks_response = ows_track.get_tracks_by_product_id(product_id) return tracks_response.message.get('items') def validate_product_artists(product_data, tracks_data, awal_artist, ff_context=None): """Validate product artists. Args: product_data (dict): the product basics data to validate. tracks_data (dict): the track data to validate. awal_artist (str): the awal artist to validate. ff_context (dict): feature flags context. Returns: response.Response: response with 200 status if valid, 400 if not. """ warnings = [] tracks_corrections = product_data.get('corrections_by_track_id', {}) product_primary_artists = _get_product_artist_names_by_type( product_data['product_artists'], FIELD_NAME_MAP[PERFORMER]) if (_validate_various_artists(product_primary_artists) and not _validate_primary_artist_various_artists( tracks_data, tracks_corrections)): warnings.append( _format_validation_result( code=warning.WARNING_CODE_VARIOUS_ARTIST_MISMATCH, reason=warning.WARNING_MESSAGE_VARIOUS_ARTIST_MISMATCH ) ) if (awal_artist and not _validate_awal_artist(product_primary_artists, awal_artist)): warnings.append( _format_validation_result( code=warning.WARNING_CODE_AWAL_ARTIST_APPLICATION_MISMATCH, reason=json.dumps( { 'awal_application_artist': awal_artist } ) ) ) if ff_context and ff_context.spotify_watchlist_artist: product_id = product_data.get('product_id') product_artists = product_data.get('product_artists', []) match_result = _validate_blocklisted_artists(product_id, product_artists) if match_result: warnings.append( _format_validation_result( code=warning.WARNING_CODE_SPOTIFY_WATCHLIST_ARTIST, reason=json.dumps(match_result) ) ) return response.Response( _create_validation_message( validation_type=ValidationType.PRODUCT_ARTISTS, errors=[], warnings=warnings ) ) def validate_artwork_compliance(product_data, tracks_data, ff_context=None): """Validate that artwork is compliant with DSP and/ or stakeholder requirements. Runs feature-flagged compliance checks (e.g. metadata mismatch) and returns a validation response containing any resulting warnings. Args: product_data (dict): Product metadata including artists, name, and product_id. tracks_data (list[dict]): Track metadata for all tracks on the product. ff_context: Feature flag context used to gate individual checks. Returns: response.Response: Validation result with an empty error list and any compliance warnings generated by the enabled checks. """ warnings = [] if ff_context and ff_context.artwork_metadata_mismatch: matches = _validate_artwork_metadata_mismatch(product_data, tracks_data) if matches: warnings.append( _format_validation_result( code=warning.WARNING_CODE_ARTWORK_METADATA_MISMATCH, reason=json.dumps({'matches': matches}) ) ) return response.Response(_create_validation_message( validation_type=ValidationType.ARTWORK_COMPLIANCE, errors=[], warnings=warnings, )) def get_validate_pricing_warnings(product_id): """Validate the pricing for a product. Args: product_id (int): the product to validate Returns: response.Response: response object containing validation result """ warnings = [] pricing_response = ows_pricing.get_product_pricing_validations(product_id) if not pricing_response: return pricing_response for pricing_warning in (pricing_response.message or {}).get('warnings') or []: warnings.append( _format_validation_result( code=pricing_warning.get('warning_code'), reason=pricing_warning.get('warning_message') ) ) return warnings def _build_metadata_search_terms(product_data, tracks_data): """Build the list of metadata terms to conduct a fuzzy search.""" product_artists = [ { METADATA_VALUE: artist.get(NAME), METADATA_FIELD: artist.get(ROLE), } for artist in product_data.get(PRODUCT_ARTISTS, []) ] product_name = { METADATA_VALUE: product_data.get(PRODUCT_NAME), METADATA_FIELD: PRODUCT_NAME, } track_corrections = product_data.get('corrections_by_track_id', {}) track_names = [] for track in tracks_data: track_name = ( track_corrections.get(track[TUID], {}).get(TRACK_NAME) or track.get(TRACK_NAME) ) if not track_name: continue track_names.append({ METADATA_VALUE: track_name, METADATA_FIELD: TRACK_NAME, }) terms = [product_name] + track_names + product_artists for t in terms: val = t[METADATA_VALUE].lower() t[DICTIONARY] = [val] + val.split() return terms def _get_normalized_text(text: str): return json.loads('"{}"'.format(text.replace('"', '\\"'))) def _validate_artwork_metadata_mismatch(product_data, tracks_data): """Detect and validate mismatches between artwork text and product/track metadata. Fetches the OCR text extracted from artwork via ows-assets. Performs fuzzy searches against the product/ track metadata. A mismatch is recorded when subtext is found that differs slightly from the exact metadata value. Args: product_data (dict): Product metadata including product name, product-level artists, and per-track corrections. tracks_data (list[dict]): Track metadata for all tracks on the product. Returns: list[dict]: Each entry contains the metadata field, expected value, and a list of fuzzy matches found in the extracted OCR text. If no results, returns an empty list. """ product_id = product_data.get('product_id') resp = ows_assets.get_image_text_extract(product_id).message block_text = resp['result']['block_text'] if not block_text: return [] normalized_block_text = _get_normalized_text(block_text) full_text_lower = normalized_block_text.lower() metadata_search_terms = _build_metadata_search_terms(product_data, tracks_data) matches = [] dictionary = set() query_completed = set() for entry in metadata_search_terms: for t in entry[DICTIONARY]: dictionary.add(t) for entry in metadata_search_terms: for term in entry[DICTIONARY]: if len(term) < MIN_LENGTH: # SKIP: Too short (reduce noise) continue if term in query_completed: # SKIP: Query completed continue calculated_dist = math.ceil(len(term) * SEARCH_RATIO) pattern = _format_artwork_metadata_mismatch_search_pattern( term, calculated_dist) search_result = regex.finditer(pattern, full_text_lower) matches_found = [] for m in search_result: matched = m.group() if matched == term: # PASS: Exact match on term (is valid) continue if len(matched) < MIN_LENGTH or matched in dictionary: # PASS: Too short/ contained in dictionary (reduce noise) continue # FAIL: Non-exact match on term (invalid) subsequence = normalized_block_text[m.start(): m.end()] matches_found.append(subsequence) query_completed.add(term) if matches_found: matches.append({ METADATA_FIELD: entry[METADATA_FIELD], METADATA_VALUE: entry[METADATA_VALUE], 'fuzzy_matches_found': matches_found }) break return matches def _format_artwork_metadata_mismatch_search_pattern(query, max_dist): """Get validation search pattern.""" query_text = query.strip('"') return rf"\b(?=\w)(?i)(?e)({regex.escape(query_text)}){{e<={max_dist}}}(?<=\S)\b" def _validate_blocklisted_artists(product_id, product_artists): """Validate artists against the blocklist service. Returns the list of matches if any artists are flagged, otherwise None. """ payload = { 'product_id': product_id, 'product_artists': product_artists } artist_response = ows_blocklist_manager.validate_artists(payload) if not artist_response: return None matches = artist_response.message.get( 'validation_errors', {}).get( 'product_artists', {}).get( 'matches', [] ) result = [ { 'name': match.get('name'), 'role': match.get('role') } for match in matches ] return result def _remove_blank_values(dictionary): """All truthy values from a dictionary. Args: dictionary (dict): dictionary to filter falsy values from Returns: dictionary: dictionary with falsy values removed """ return {key: value for key, value in dictionary.items() if value} def _meta_language_is_invalid(meta_language): """Validate if meta language is in list of known good languages.""" if meta_language is None: return False valid_language_response = language_model.get_languages() valid_language_list = [lang.get('code') for lang in valid_language_response.message.get('items')] if meta_language not in valid_language_list: return True return False def _format_validation_result(*, code, reason): """Format validation result. Use this format for errors that are not from existing validation types: product_basics, scheduling_and_pricing, artwork, tracks, and publishing_obligation, and all warnings. For errors that are from these existing validation types, use _format_error. See https://github.com/theorchard/graphql-content-review/blob/master/src/connectors/ows-product-digital/formatters/product-validation.ts for how we reformat validation results in GraphQL. Args: code (str): An error or warning code reason (str): A message explaining the error or warning, or JSON data associated with this validation. Returns: dict: A dictionary containing the formatted validation result with 'code' and 'reason' keys. """ # noqa: E501 return { 'code': code.upper(), 'reason': reason, } def _format_error(*, validator, validator_value, message, error_code): """Format legacy validation error. Use this format for errors that are from existing validation types: product_basics, scheduling_and_pricing, artwork, tracks, and publishing_obligation. For errors that are not from these existing validation types, and all warnings, use _format_validation_result. See https://github.com/theorchard/graphql-content-review/blob/master/src/connectors/ows-product-digital/formatters/product-validation.ts for how we reformat validation results in GraphQL. Args: validator (str): The validator responsible for the error validator_value (any): The value that the validator expected message (str): A description of the error error_code (str): An error code Returns: dict: A dictionary containing the formatted validation error with keys: 'validator', 'validator_value', 'message', and 'error_code' """ # noqa: E501 return { 'validator': validator, 'validator_value': validator_value, 'message': message, 'error_code': error_code, } def _validate_required_artist_role(artists, role): """Validate that the artist list includes an artist with the given role. Args: artists (list): list of artist dictionaries. Returns: response.Response: response with 200 status if valid, 400 if not. """ for artist in artists: if artist.get('role') == role: return response.Response() error_message = { role: _format_error( validator='required', validator_value=True, message="'{}' is a required property".format(role), error_code=f'{role}_required' ) } return response.create_error_response( code=error.ERROR_CODE_BAD_REQUEST, message=error_message) def _validate_awal_artist(product_primary_artists, awal_artist): """Validate if AWAL Artists match primary product artists. Args: product_primary_artists (set): set of product level primary artists awal_artist (str): AWAL artist associated with initial signup form Returns: bool: False if none of the AWAL artists are listed as a primary artist on the product. """ return awal_artist in product_primary_artists def _get_product_artist_names_by_type(artists, artist_type): """Get primary artist names from a list of artists. Args: artists (list): list of artist objects artist_type (str): artist role or type Returns: set: Set of track level primary artists """ artist_names = set(artist['name'] for artist in artists if artist['role'] == artist_type) return artist_names def _get_track_artist_names_by_type(artists, artist_type): """Get artist names by type from a list of artists. Args: artists (list): list of artist objects artist_type (str): artist role or type Returns: set: Set of track level artists by type """ return { artist['name'].strip() for artist in artists if artist.get('type') == artist_type and artist.get('name') and artist['name'].strip() } def _validate_various_artists(product_primary_artists): """Validate Various Artists if they exist in product primary artists. Args: product_primary_artists (list): list of artist names Returns: bool: False if Various Artist names does not exist in primary artist list """ for artist in product_primary_artists: if artist in validations.VARIOUS_ARTISTS: return True return False def _validate_primary_artists(product_primary_artists, tracks_data, tracks_corrections): """Validate that Primary Artists on 100% of tracks are on product primary artists. Args: product_primary_artists (list): list of artist names tracks_data (list): list of track data dicts tracks_corrections (dict): corrections by track id Returns: bool: True if primary artist on track level match on product level """ primary_artists_by_track = [] for track in tracks_data: track_artists = tracks_corrections.get(track['tuid'], {}).get('track_artist') or track['artists'] primary_artists_by_track.append(_get_track_artist_names_by_type(track_artists, PERFORMER)) primary_artist_intersection = _get_artist_intersection_across_tracks(primary_artists_by_track) return len(primary_artist_intersection - set(product_primary_artists)) == 0 def _validate_featuring_artists(product_featuring_artists, tracks_data, tracks_corrections): """Validate that a Featuring Artist on 100% of tracks is a product featuring artist. Args: product_featuring_artists (list): list of artist names tracks_data (list): list of track data dicts tracks_corrections (dict): corrections by track id Returns: bool: True if featuring artists on 100% of tracks are on the product-level """ featuring_artists_by_track = [] for track in tracks_data: track_artists = _get_artist_corrections(track, tracks_corrections, FEATURING) featuring_artists_by_track.append(_get_track_artist_names_by_type(track_artists, FEATURING)) featuring_artist_intersection = _get_artist_intersection_across_tracks(featuring_artists_by_track) return len(featuring_artist_intersection - set(product_featuring_artists)) == 0 def _validate_remixers(product_remixer_artists, tracks_data, tracks_corrections): """Validate that remixers on 100% of tracks are listed as remixers on the product-level. Args: product_remixer_artists (list): list of remixer names tracks_data (list): list of track data dicts tracks_corrections (dict): corrections by track id Returns: bool: True if it passes the validation, otherwise False """ remix_artists_by_track = [] for track in tracks_data: track_artists = _get_artist_corrections(track, tracks_corrections, REMIXER) remix_artists_by_track.append(_get_track_artist_names_by_type(track_artists, REMIXER)) remix_artist_intersection = _get_artist_intersection_across_tracks(remix_artists_by_track) return len(remix_artist_intersection - set(product_remixer_artists)) == 0 def _validate_producers(product_producer_artists, tracks_data, tracks_corrections): """Validate that Producer is on 100% of product tracks. Args: product_producer_artists (set): the product basics data to validate tracks_data (list): list of track data dicts tracks_corrections (dict): corrections by track id Returns: bool: True if each track includes producer as on product level. """ producer_artists_by_track = [] for track in tracks_data: track_artists = _get_artist_corrections(track, tracks_corrections, PRODUCER) producer_artists_by_track.append(_get_track_artist_names_by_type(track_artists, PRODUCER)) producer_artist_intersection = _get_artist_intersection_across_tracks(producer_artists_by_track) return len(producer_artist_intersection - set(product_producer_artists)) == 0 def _get_artist_intersection_across_tracks(artists_by_track): """Get the intersection of the given artists across tracks. Args: artists_by_track (list): list of the artists in order of the tracks Returns: set: the intersection of the artists from the tracks """ if len(artists_by_track) == 0: return set() artist_intersection = set(artists_by_track[0]) for i in range(1, len(artists_by_track)): artist_intersection = artist_intersection.intersection(set(artists_by_track[i])) return artist_intersection def _get_artist_corrections(track, tracks_corrections, key): artist_corrections = tracks_corrections.get(track['tuid'], {}).get(key) return artist_corrections if artist_corrections is not None else track['artists'] def _validate_primary_artist_various_artists( tracks_data, tracks_corrections ): """Validate that there are >3 unique track artists if 'Various Artists' exists on product primary artists. Args: tracks_data (list): list of track data dicts tracks_corrections (dict): corrections by track id product_genre_id (int): product genre id Returns:test_validate_product_basics_remixer_mismatch_error bool: True if > 3 unique primary artists across all tracks """ tracks_artist_count = defaultdict(int) for track in tracks_data: track_artists = tracks_corrections.get(track['tuid'], {}).get('track_artist') or track['artists'] for artist in _get_track_artist_names_by_type(track_artists, PERFORMER): tracks_artist_count[artist] += 1 return len(tracks_artist_count) > 3 def _validate_dates(product_data): """Validate dates. Args: product_data (dict): payload of audio product data to validate Returns: errors (dict): dictionary of any errors encountered """ errors = {} for field in ['preorder_date', 'release_date', 'sale_start_date']: if field not in product_data or not product_data[field]: continue field_value = product_data.get(field) date = _parse_date(field_value) if not date: errors[field] = _format_error( validator='dates', validator_value=True, message='invalid date', error_code=f'invalid_{field}' ) elif not _is_date_in_range(date, MIN_VALID_DATE, MAX_VALID_DATE): errors[field] = _format_error( validator='dates', validator_value=True, message='must be between {} and {}'.format( MIN_VALID_DATE, MAX_VALID_DATE), error_code=f'invalid_{field}' ) return errors def _parse_date(date_text): if isinstance(date_text, datetime.date): # don't try to parse datetime objects. return date_text try: date_time = datetime.datetime.strptime(date_text, '%Y-%m-%d') except ValueError: return None return date_time.date() def _is_date_in_range(date, min_date, max_date): return min_date <= date <= max_date def _validate_product_code_unique(product_data): """Call to ows-product to check if product code is unique. Args: product_data (dict): payload of audio product data to validate Returns: response.Response: response with 200 status if valid or 400 if not """ errors = {} product_code = product_data.get('product_code') account_type = product_data.get('account_type') account_id = product_data.get('account_id') product_code_response = ows_product.is_product_code_available( product_code, account_type, account_id) if product_code_response.status == 400: product_code_message = product_code_response.errors.get('message') if 'product_id' in product_data: release_response = release.get_release(product_data['product_id']) current_product_code = release_response.message.get('product_code') if current_product_code != product_code: errors['product_code'] = _format_error( validator='used', validator_value=True, message=product_code_message, error_code='product_code_used' ) else: errors['product_code'] = _format_error( validator='used', validator_value=True, message=product_code_message, error_code='product_code_used' ) if product_code_response.status == 500: return response.create_fatal_response() return response.Response(message=errors) def _validate_upc_available(upc): """Call to ows-product to check if upc is available. Args: product_data (dict): payload of audio product data to validate Returns: response.Response: response with 200 status if valid or 400 if not """ errors = {} upc_response = ows_product.check_upc_available(upc) if upc_response.status in [400, 403]: upc_message = upc_response.errors.get('message') errors['upc'] = _format_error( validator='used', validator_value=True, message=upc_message, error_code='upc_used' ) if upc_response.status == 500: return response.create_fatal_response() return response.Response(message=errors) def _composer_required(product_basics): subgenre_ids = config.COMPOSER_REQUIRED_SUBGENRE_IDS return product_basics['subgenre_id'] in subgenre_ids def _lyricist_required(product_basics): genre_id = config.WORLD_MUSIC_GENRE_ID subgenre_ids = config.LYRICIST_REQUIRED_SUBGENRE_IDS # Only applies to new products validation_start_date = validations.LYRICIST_REQUIRED_START_DATE new_release = ( product_basics.get('release_date') is not None and datetime.datetime.strptime( product_basics['release_date'], '%Y-%m-%d' ) >= datetime.datetime.strptime(validation_start_date, "%Y-%m-%d") ) return ( product_basics['genre_id'] == genre_id and product_basics['subgenre_id'] in subgenre_ids and new_release ) def _is_subgenre_theatre_scores(product_basics): if 'subgenre_id' in product_basics: genre_id = product_basics.get('genre_id') subgenre_id = product_basics.get('subgenre_id') return genre_id == config.SOUNDTRACKS_GENRE_ID and subgenre_id == config.THEATRE_SCORES_SUBGENRE_ID return False def _create_validation_message(*, validation_type, errors=None, warnings=None): """Create validation message with errors and warnings. See https://github.com/theorchard/graphql-content-review/blob/master/src/connectors/ows-product-digital/formatters/product-validation.ts for how we reformat validation results in GraphQL. Args: validation_type (str): Type of validation performed for product errors (list or dict, optional): A list of validation errors. dict is used only by product_basics and scheduling_and_pricing validation_types. warnings (list, optional): A list of validation warnings Returns: dict: A dictionary representing product validation """ # noqa: E501 result = { 'valid': not errors, # only errors affect validity, not warnings 'errors': errors, } if warnings: result['warnings'] = warnings return { validation_type: result } def _create_tracks_validation_message( *, validation_type, ows_track_validation_results): """Create validation message with errors and warnings from tracks. Args: validation_type (str): Type of track validation performed for product. ows_track_validation_results (dict): response message from ows-track validation ows_assets_bit_depths_validation_results (response.Response, optional): Indicates bit depths are valid by status == 200. If None, bit_depths_valid is not included in the response. Returns: dict: A dictionary representing product validation """ validation_message = _create_validation_message( validation_type=validation_type, errors=ows_track_validation_results['errors'], warnings=ows_track_validation_results.get('warnings'), ) total_tracks = ows_track_validation_results['total_tracks'] valid_tracks = ows_track_validation_results['valid_tracks'] has_tracks = bool(total_tracks) validation_message[validation_type].update({ 'total_tracks': total_tracks, 'valid_tracks': valid_tracks, 'valid': validation_message[validation_type]['valid'] and has_tracks, }) return validation_message