"""Validation rules.""" import re from typing import Dict, NamedTuple from ddex_ingester_common.constants import \ ddex_providers, validation_rule_error_codes as rule_error_codes from ddex_ingester_common.constants import validation_queries as queries from ddex_ingester_common.constants import validation_rules as rule_constants from ddex_ingester_common.constants.catalog_ingestion import ( REJECT, SUCCESS, WARNING, ) from ddex_ingester_common.constants.format_mapping import FORMAT_MAPPING from ddex_ingester_common.constants.roles import ( DDEX_ARTIST_ROLE_TO_ORCHARD_ARTIST_TYPE, DDEX_ARTIST_ROLE_TO_ORCHARD_PERFORMER_ROLE, DDEX_ARTIST_ROLE_TO_ORCHARD_WRITER, DDEX_RESOURCE_CONTRIBUTOR_ROLE_TO_ORCHARD_PARTICIPANT_ROLE, DDEX_RESOURCE_CONTRIBUTOR_ROLE_TO_ORCHARD_PERFORMER_ROLE, DDEX_VIDEO_ROLE_TO_ORCHARD_ARTIST_TYPE, ) from ddex_ingester_common.constants.validation_accepted_values import ( AWAL_NOT_FOR_DISTRIBUTION, VALID_UPDATE_INDICATORS ) from ddex_ingester_common.helpers import metadata as metadata_helper from ddex_ingester_common.lambda_exceptions import ( InvalidGenreException, InvalidSubgenreException ) from ddex_ingester_common.utils import genre_mapping as genre_mapping_util class RuleResult(NamedTuple): """A result from a rule.""" validation_rule_id: int response: str = SUCCESS message: str = None def is_ccm_som_livre_grps_ddex_ingestion_enabled(split, application_name): """Lookup if ccm_som_livre_grps_ddex_ingestion split is enabled.""" is_enabled = split and split.get_treatment( application_name, 'ccm_som_livre_grps_ddex_ingestion', {'service': application_name} ) == 'on' return is_enabled def get_require_valid_sme_genre(splitio_client, application_name): """Validate SME genre.""" def require_valid_sme_genre(data: Dict) -> RuleResult: """Validate SME genre.""" genres = data['product'].get('genres') if genres: first_genre = genres[0].get('genre') if not genres or not first_genre: return RuleResult( rule_error_codes.MISSING_GENRE, REJECT, 'Genre is missing' ) # Only validate genre if iTunes genre, since rest are Som Livre if ((not is_ccm_som_livre_grps_ddex_ingestion_enabled(splitio_client, application_name)) # noqa or (first_genre.endswith('-00') or first_genre.endswith( '01'))): genre_mapping = genre_mapping_util. \ get_orchard_genre_mapping_for_itunes(first_genre) if not genre_mapping: return RuleResult( rule_error_codes.INVALID_GENRE, REJECT, 'Invalid genre' ) return RuleResult(rule_error_codes.MISSING_GENRE) return require_valid_sme_genre def require_party_id(data: Dict) -> RuleResult: """Rule for validating a valid party id.""" party_id = data.get('party_id') if not party_id: return RuleResult( rule_error_codes.INVALID_PARTY_ID, REJECT, 'Party Id is missing' ) if party_id not in rule_constants.VALID_PARTY_IDS: return RuleResult( rule_error_codes.INVALID_PARTY_ID, REJECT, f'Invalid Party Id: {party_id}' ) return RuleResult(rule_error_codes.INVALID_PARTY_ID) def require_message_id(data: Dict) -> RuleResult: """Rule for requiring a message id.""" if not data.get('message_id'): return RuleResult( rule_error_codes.MISSING_MESSAGE_ID, REJECT, 'Message Id is missing' ) return RuleResult(rule_error_codes.MISSING_MESSAGE_ID) def require_product(data: Dict) -> RuleResult: """Rule for requiring a product.""" if not data.get('product'): return RuleResult( rule_error_codes.MISSING_PRODUCT, REJECT, 'Product is missing' ) return RuleResult(rule_error_codes.MISSING_PRODUCT) def require_product_name(data: Dict) -> RuleResult: """Rule for requiring a product name.""" if not data['product'].get('product_name'): return RuleResult( rule_error_codes.MISSING_PRODUCT_NAME, REJECT, 'Product Name is missing' ) return RuleResult(rule_error_codes.MISSING_PRODUCT_NAME) def require_product_release_reference(data: Dict) -> RuleResult: """Rule for requiring a product release reference.""" if not data['product'].get('release_reference'): return RuleResult( rule_error_codes.MISSING_PRODUCT_RELEASE_REFERENCE, REJECT, 'Product Release Reference is missing' ) return RuleResult(rule_error_codes.MISSING_PRODUCT_RELEASE_REFERENCE) def require_product_display_artist_name(data: Dict) -> RuleResult: """Rule for requiring a product display artist name.""" if not data['product'].get('display_artist_name'): return RuleResult( rule_error_codes.MISSING_PRODUCT_DISPLAY_ARTIST_NAME, REJECT, 'Product Display Artist Name is missing' ) return RuleResult(rule_error_codes.MISSING_PRODUCT_DISPLAY_ARTIST_NAME) def require_product_release_type(data: Dict) -> RuleResult: """Rule for requiring a product release type.""" if not data['product'].get('release_type'): return RuleResult( rule_error_codes.MISSING_PRODUCT_RELEASE_TYPE, REJECT, 'Product Release Type is missing' ) return RuleResult(rule_error_codes.MISSING_PRODUCT_RELEASE_TYPE) def require_product_is_main_release(data: Dict) -> RuleResult: """Rule for requiring a product is main release.""" if type(data['product'].get('is_main_release')) is not bool: return RuleResult( rule_error_codes.MISSING_PRODUCT_IS_MAIN_RELEASE, REJECT, 'Product Main Release is missing' ) return RuleResult(rule_error_codes.MISSING_PRODUCT_IS_MAIN_RELEASE) def require_product_upc(data: Dict) -> RuleResult: """Rule for requiring a product UPC.""" if not data['product'].get('upc'): return RuleResult( rule_error_codes.MISSING_PRODUCT_UPC, REJECT, 'Product UPC is missing' ) return RuleResult(rule_error_codes.MISSING_PRODUCT_UPC) def get_require_valid_awal_genre(graphql_gateway): """Validate genre/subgenre.""" def require_valid_awal_genre(data: Dict) -> RuleResult: """Validate genre/subgenre.""" genres = data['product'].get('genres') if not genres: return RuleResult( rule_error_codes.MISSING_GENRE, REJECT, 'Genre is missing' ) genre = genres[0].get('genre') subgenre = genres[0].get('subgenre') if not subgenre: return RuleResult( rule_error_codes.MISSING_SUBGENRE, REJECT, 'Subgenre is missing' ) try: genre_mapping_util.get_orchard_genre_mapping( graphql_gateway, genre, subgenre) except InvalidGenreException as err: return RuleResult( rule_error_codes.INVALID_GENRE, REJECT, str(err) ) except InvalidSubgenreException as err: return RuleResult( rule_error_codes.INVALID_SUBGENRE, REJECT, str(err) ) return RuleResult(rule_error_codes.MISSING_GENRE) return require_valid_awal_genre def require_valid_pline(data: Dict) -> RuleResult: """Validate track p-line.""" tracks = data['tracks'] for track in tracks: p_line = track.get('p_line') if p_line and not re.match('[0-9]{4}[ ]+\\S+', p_line, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_PLINE, REJECT, f'Invalid P-Line on track ' f'{track["volume"]}-{track["sequence_number"]}') return RuleResult(rule_error_codes.INVALID_PLINE) def require_valid_cline(data: Dict) -> RuleResult: """Validate product c-line.""" c_line = data['product'].get('c_line') if c_line and not re.match('[0-9]{4}[ ]+\\S+', c_line, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_CLINE, REJECT, 'Invalid product C-Line' ) return RuleResult(rule_error_codes.INVALID_CLINE) def validate_product_localization_language_codes(data: Dict) -> RuleResult: """Validate product localized title language codes.""" localized_titles = data['product'].get('localized_titles') return validate_language_codes( localized_titles, rule_error_codes.DUPLICATE_LOCALIZED_PRODUCT_LANGUAGE_CODE, 'localized product title' ) def validate_product_localized_titles(data: Dict) -> RuleResult: """Check product localized titles for missing data.""" localized_titles = data['product'].get('localized_titles') for localized_title in localized_titles or []: result = validate_localized_title(localized_title, 'Product') if result: return result return RuleResult(rule_error_codes.MISSING_LOCALIZED_TITLE_DATA) def validate_track_localized_titles(data: Dict) -> RuleResult: """Check track localized titles for missing data.""" for track in data.get('tracks') or []: localized_titles = track.get('localized_titles') for localized_title in localized_titles or []: result = validate_localized_title(localized_title, 'Track') if result: return result return RuleResult(rule_error_codes.MISSING_LOCALIZED_TITLE_DATA) def validate_localized_title(localized_title: Dict, field: str) -> RuleResult: """Check localized title for missing data.""" if not localized_title.get('title'): return RuleResult( rule_error_codes.MISSING_LOCALIZED_TITLE_DATA, REJECT, f'{field} Localization Title is missing' ) if not localized_title.get('language_code'): return RuleResult( rule_error_codes.MISSING_LOCALIZED_TITLE_DATA, REJECT, f'{field} Localization Language Code is missing' ) return None def validate_product_localized_artists(data: Dict) -> RuleResult: """Check product localized artists for missing data.""" for display_artist in data['product'].get('display_artists') or []: for localized_name in display_artist.get('localized_names') or []: result = validate_localized_artist(localized_name, 'Product') if result: return result return RuleResult(rule_error_codes.MISSING_LOCALIZED_ARTIST_DATA) def validate_track_localized_artists(data: Dict) -> RuleResult: """Check track localized artists for missing data.""" for track in data.get('tracks') or []: for display_artist in track.get('display_artists') or []: for localized_name in display_artist.get('localized_names') or []: result = validate_localized_artist(localized_name, 'Track') if result: return result return RuleResult(rule_error_codes.MISSING_LOCALIZED_ARTIST_DATA) def validate_localized_artist( localized_artist: Dict, field: str) -> RuleResult: """Check localized artist for missing data.""" if not localized_artist.get('name'): return RuleResult( rule_error_codes.MISSING_LOCALIZED_ARTIST_DATA, REJECT, f'{field} Artist Localization Name is missing' ) if not localized_artist.get('language_code'): return RuleResult( rule_error_codes.MISSING_LOCALIZED_ARTIST_DATA, REJECT, f'{field} Artist Localization Language Code is missing' ) return None def validate_language_codes( localized_titles: Dict, validation_rule_id: int, field_name: str) -> RuleResult: """Validate language codes.""" if localized_titles: language_ids = [] for localized_title in localized_titles: language_id = metadata_helper.get_language_id( localized_title['language_code']) if language_id in language_ids: return RuleResult( validation_rule_id, REJECT, f'Duplicate language code' f' {localized_title["language_code"]}' f' not allowed for localization' ) language_ids.append(language_id) if (rule_constants.CHINESE_SIMPLIFIED in language_ids and rule_constants.CHINESE_TRADITIONAL not in language_ids) or ( rule_constants.CHINESE_SIMPLIFIED not in language_ids and rule_constants.CHINESE_TRADITIONAL in language_ids): return RuleResult( rule_error_codes.MISSING_CHINESE_SIMPLIFIED_OR_TRADITIONAL, WARNING, f'Both Simplified and Traditional Chinese must be present for' f' {field_name}' ) return RuleResult(validation_rule_id) def validate_product_display_artist_language_codes(data: Dict) -> RuleResult: """Validate product localized title language codes.""" display_artists = data['product'].get('display_artists') result = RuleResult( rule_error_codes.DUPLICATE_LOCALIZED_PRODUCT_ARTIST_LANGUAGE_CODE) for display_artist in display_artists: result = validate_language_codes( display_artist.get('localized_names'), rule_error_codes.DUPLICATE_LOCALIZED_PRODUCT_ARTIST_LANGUAGE_CODE, f'localized product display artist {display_artist["name"]}' ) if result.response == REJECT: return result return result def require_track_name(data: Dict) -> RuleResult: """Rule for requiring a track name.""" for track in data.get('tracks') or []: if not track.get('track_name'): return RuleResult( rule_error_codes.MISSING_TRACK_NAME, REJECT, 'Track Name is missing' ) return RuleResult(rule_error_codes.MISSING_TRACK_NAME) def require_track_isrc(data: Dict) -> RuleResult: """Rule for requiring a track ISRC.""" for track in data.get('tracks') or []: if not track.get('isrc'): return RuleResult( rule_error_codes.MISSING_TRACK_ISRC, REJECT, 'Track ISRC is missing' ) return RuleResult(rule_error_codes.MISSING_TRACK_ISRC) def require_track_unique_isrc(data: Dict) -> RuleResult: """Rule for requiring a track ISRC to be unique.""" existing_isrc = set() for track in data.get('tracks') or []: track_isrc = track.get('isrc') if track_isrc in existing_isrc: return RuleResult( rule_error_codes.DUPLICATE_TRACK_ISRC, REJECT, f'Duplicate track ISRC {track_isrc}' ) else: existing_isrc.add(track_isrc) return RuleResult(rule_error_codes.DUPLICATE_TRACK_ISRC) def require_track_sequence_number(data: Dict) -> RuleResult: """Rule for requiring a track sequence number.""" for track in data.get('tracks') or []: if not track.get('sequence_number'): return RuleResult( rule_error_codes.MISSING_TRACK_SEQUENCE_NUMBER, REJECT, 'Track Sequence Number is missing' ) return RuleResult(rule_error_codes.MISSING_TRACK_SEQUENCE_NUMBER) def require_track_volume(data: Dict) -> RuleResult: """Rule for requiring a track volume.""" for track in data.get('tracks') or []: if not track.get('volume'): return RuleResult( rule_error_codes.MISSING_TRACK_VOLUME, REJECT, 'Track Volume is missing' ) return RuleResult(rule_error_codes.MISSING_TRACK_VOLUME) def require_track_release_type(data: Dict) -> RuleResult: """Rule for requiring a track release type.""" for track in data.get('tracks') or []: if not track.get('release_type'): return RuleResult( rule_error_codes.MISSING_TRACK_RELEASE_TYPE, REJECT, 'Track Release Type is missing' ) return RuleResult(rule_error_codes.MISSING_TRACK_RELEASE_TYPE) def require_track_release_reference(data: Dict) -> RuleResult: """Rule for requiring a track release reference.""" for track in data.get('tracks') or []: if not track.get('release_reference'): return RuleResult( rule_error_codes.MISSING_TRACK_RELEASE_REFERENCE, REJECT, 'Track Release Reference is missing' ) return RuleResult(rule_error_codes.MISSING_TRACK_RELEASE_REFERENCE) def require_track_resource_reference(data: Dict) -> RuleResult: """Rule for requiring a track resource reference.""" for track in data.get('tracks') or []: if not track.get('resource_reference'): return RuleResult( rule_error_codes.MISSING_TRACK_RESOURCE_REFERENCE, REJECT, 'Track Resource Reference is missing' ) return RuleResult(rule_error_codes.MISSING_TRACK_RESOURCE_REFERENCE) def validate_track_display_artist_language_codes(data: Dict) -> RuleResult: """Validate track localized title language codes.""" result = RuleResult( rule_error_codes.DUPLICATE_LOCALIZED_TRACK_ARTIST_LANGUAGE_CODE) if data['tracks']: for track in data['tracks']: display_artists = track.get('display_artists') for display_artist in display_artists: result = validate_language_codes( display_artist.get('localized_names'), rule_error_codes.DUPLICATE_LOCALIZED_TRACK_ARTIST_LANGUAGE_CODE, # noqa f'localized product display artist ' f'{display_artist["name"]}' ) if result.response == REJECT: return result return result def validate_track_localization_language_codes(data: Dict) -> RuleResult: """Validate track localized title language codes.""" result = RuleResult( rule_error_codes.DUPLICATE_LOCALIZED_TRACK_LANGUAGE_CODE) if data['tracks']: for track in data['tracks']: localized_titles = track.get('localized_titles') result = validate_language_codes( localized_titles, rule_error_codes.DUPLICATE_LOCALIZED_TRACK_LANGUAGE_CODE, track['track_name'] ) if result.response == REJECT: return result return result def require_orchard_label(data: Dict) -> RuleResult: """Require orchard label ID.""" if not data['product'].get('vendor_id'): return RuleResult( rule_error_codes.MISSING_ORCHARD_LABEL, REJECT, 'Orchard Label ID is missing' ) return RuleResult(rule_error_codes.MISSING_ORCHARD_LABEL) def require_project_from_ext_meta(data: Dict) -> RuleResult: """Require project data from extended metadata.""" if not data.get('project'): return RuleResult( rule_error_codes.MISSING_PROJECT_DATA, REJECT, 'Project section is missing' ) if not data['project'].get('project_code'): return RuleResult( rule_error_codes.MISSING_PROJECT_CODE, REJECT, 'Project code is missing' ) if not data['project'].get('artist'): return RuleResult( rule_error_codes.MISSING_PROJECT_ARTIST, REJECT, 'Project artist is missing' ) return RuleResult(rule_error_codes.MISSING_PROJECT_DATA) def get_require_target_vendor_subaccount_matches_existing_product( graphql_gateway): """Require target vendor/subaccount to match existing Product vendor.""" def require_target_vendor_subaccount_matches_existing_product( data: Dict) -> RuleResult: upc = data['product']['upc'] switchboard_vendor_id = data['product']['vendor_id'] switchboard_subaccount_id = data['product']['subaccount_id'] orchard_product = graphql_gateway.execute( queries.GET_PRODUCT_BY_UPC, {'upc': upc} )['data']['productByUpc'] if orchard_product: orchard_vendor_id = orchard_product.get('vendorId') orchard_subaccount_id = orchard_product.get('subaccountId') invalid_target = \ (orchard_vendor_id != switchboard_vendor_id or orchard_subaccount_id != switchboard_subaccount_id ) if invalid_target: msg = (f'Product {upc} exists in the Orchard but under ' 'a different vendor or subaccount than what the ' 'Switchboard deal is targeting. Switchboard Vendor ID: ' f'{switchboard_vendor_id}, Orchard Vendor ID: ' f' {orchard_vendor_id}, Switchboard Subaccount ID: ' f' {switchboard_subaccount_id}, Orchard Subaccount ID: ' f' {orchard_subaccount_id}') return RuleResult( rule_error_codes.MISMATCHING_VENDOR_OR_SUBACCOUNT, REJECT, msg ) return RuleResult(rule_error_codes.MISMATCHING_VENDOR_OR_SUBACCOUNT) return require_target_vendor_subaccount_matches_existing_product def require_video_name(data: Dict) -> RuleResult: """Rule for requiring a video name.""" if data.get('video'): if not data['video'].get('video_name'): return RuleResult( rule_error_codes.MISSING_VIDEO_NAME, REJECT, 'Video Name is missing' ) return RuleResult(rule_error_codes.MISSING_VIDEO_NAME) def require_video_isrc(data: Dict) -> RuleResult: """Rule for requiring a video ISRC.""" if data.get('video'): if not data['video'].get('isrc'): return RuleResult( rule_error_codes.MISSING_VIDEO_ISRC, REJECT, 'Video ISRC is missing' ) return RuleResult(rule_error_codes.MISSING_VIDEO_ISRC) def require_video_release_type(data: Dict) -> RuleResult: """Rule for requiring a video release type.""" if data.get('video'): if not data['video'].get('release_type'): return RuleResult( rule_error_codes.MISSING_VIDEO_RELEASE_TYPE, REJECT, 'Video Release Type is missing' ) return RuleResult(rule_error_codes.MISSING_VIDEO_RELEASE_TYPE) def require_video_release_reference(data: Dict) -> RuleResult: """Rule for requiring a video release reference.""" if data.get('video'): if not data['video'].get('release_reference'): return RuleResult( rule_error_codes.MISSING_VIDEO_RELEASE_REFERENCE, REJECT, 'Video Release Reference is missing' ) return RuleResult(rule_error_codes.MISSING_VIDEO_RELEASE_REFERENCE) def validate_video_p_line(data: Dict) -> RuleResult: """Rule for requiring a video pline to be complete or null.""" if data.get('video'): if not data['video'].get('p_line'): return RuleResult( rule_error_codes.MISSING_VIDEO_PLINE_DATA, REJECT, 'Video PLine is missing from XML' ) if not data['video']['p_line'].get('line'): return RuleResult( rule_error_codes.MISSING_VIDEO_PLINE_DATA, REJECT, 'Video PLine Text is missing' ) if not data['video']['p_line'].get('year'): return RuleResult( rule_error_codes.MISSING_VIDEO_PLINE_DATA, REJECT, 'Video PLine Year is missing' ) return RuleResult(rule_error_codes.MISSING_VIDEO_PLINE_DATA) def validate_video_c_line(data: Dict) -> RuleResult: """Rule for requiring a video cline to be complete or null.""" if data.get('video'): if not data['video'].get('c_line'): return RuleResult( rule_error_codes.MISSING_VIDEO_CLINE_DATA, REJECT, 'Video CLine is missing from XML' ) if not data['video']['c_line'].get('line'): return RuleResult( rule_error_codes.MISSING_VIDEO_CLINE_DATA, REJECT, 'Video CLine Text is missing' ) if not data['video']['c_line'].get('year'): return RuleResult( rule_error_codes.MISSING_VIDEO_CLINE_DATA, REJECT, 'Video CLine Year is missing' ) return RuleResult(rule_error_codes.MISSING_VIDEO_CLINE_DATA) def validate_product_participants(data: Dict) -> RuleResult: """Check product participants for missing data.""" for participant in data['product'].get('display_artists') or []: result = validate_participant(participant, 'Product') if result: return result return RuleResult(rule_error_codes.MISSING_PARTICIPANT_DATA) def validate_track_participants(data: Dict) -> RuleResult: """Check track participants for missing data.""" for track in data.get('tracks') or []: if not track.get('display_artists', []): return RuleResult( rule_error_codes.MISSING_PARTICIPANT_DATA, REJECT, 'Display Artists is missing' ) for participant in track.get('display_artists') or []: result = validate_participant(participant, 'Track') if result: return result for participant in track.get('resource_contributors') or []: result = validate_participant(participant, 'Track') if result: return result return RuleResult(rule_error_codes.MISSING_PARTICIPANT_DATA) def require_video_composer(data: Dict) -> RuleResult: """Rule for requiring a composer contributor on a video.""" if data.get('video'): has_composer = any( 'Composer' in (contributor.get('roles') or []) for contributor in (data['video'].get('display_artists') or []) + (data['video'].get('resource_contributors') or []) ) if not has_composer: return RuleResult( rule_error_codes.MISSING_VIDEO_COMPOSER, REJECT, 'Video Composer is required' ) return RuleResult(rule_error_codes.MISSING_VIDEO_COMPOSER) def validate_video_participants(data: Dict) -> RuleResult: """Check video participants for missing data.""" if data.get('video'): for participant in data['video'].get('display_artists') or []: result = validate_participant(participant, 'Video') if result: return result for participant in data['video'].get('resource_contributors') or []: result = validate_participant(participant, 'Video') if result: return result return RuleResult(rule_error_codes.MISSING_PARTICIPANT_DATA) def validate_participant(participant: Dict, field: str) -> RuleResult: """Check participant for missing data.""" if not participant.get('name'): return RuleResult( rule_error_codes.MISSING_PARTICIPANT_DATA, REJECT, f'{field} Participant Name is missing' ) if not participant.get('sequence_number'): return RuleResult( rule_error_codes.MISSING_PARTICIPANT_DATA, REJECT, f'{field} Participant Sequence Number is missing' ) return None def validate_deal_release_references(data: Dict) -> RuleResult: """Check deals for missing data.""" for deal in data.get('deals') or []: if not deal.get('release_references'): return RuleResult( rule_error_codes.MISSING_DEAL_RELEASE_REFERENCES, REJECT, 'Deal Release References are missing' ) return RuleResult(rule_error_codes.MISSING_DEAL_RELEASE_REFERENCES) def validate_deal_terms(data: Dict) -> RuleResult: """Check deal terms for missing data.""" for deal in data.get('deals') or []: references = deal.get('release_references') if not deal.get('deal_terms'): return RuleResult( rule_error_codes.MISSING_DEAL_TERMS, REJECT, f'Deal Terms are missing for references: {references}' ) return RuleResult(rule_error_codes.MISSING_DEAL_TERMS) def validate_deal_term_commercial_model_types(data: Dict) -> RuleResult: """Check deal terms for missing data.""" for deal in data.get('deals') or []: for deal_term in deal.get('deal_terms') or []: if not deal_term.get('commercial_model_types'): return RuleResult( rule_error_codes.MISSING_DEAL_TERM_COMMERCIAL_MODEL_TYPES, REJECT, 'Deal Term Commercial Model Types are missing' ) return RuleResult( rule_error_codes.MISSING_DEAL_TERM_COMMERCIAL_MODEL_TYPES) # noqa def validate_deal_term_usage_type(data: Dict) -> RuleResult: """Check deal terms for missing data.""" for deal in data.get('deals') or []: for deal_term in deal.get('deal_terms') or []: if not deal_term.get('usage_type'): return RuleResult( rule_error_codes.MISSING_DEAL_TERM_USAGE_TYPE, REJECT, 'Deal Term Usage Type is missing' ) return RuleResult(rule_error_codes.MISSING_DEAL_TERM_USAGE_TYPE) def validate_deal_term_territories(data: Dict) -> RuleResult: """Check deal terms for missing data.""" for deal in data.get('deals') or []: for deal_term in deal.get('deal_terms') or []: if not deal_term.get('territories') and \ not deal_term.get('excluded_territories'): return RuleResult( rule_error_codes.MISSING_DEAL_TERM_TERRITORIES, REJECT, 'Deal Term Territories are missing' ) return RuleResult(rule_error_codes.MISSING_DEAL_TERM_TERRITORIES) def validate_instant_gratification_references(data: Dict) -> RuleResult: """Check instant gratification for missing data.""" for deal in data.get('deals') or []: for deal_term in deal.get('deal_terms') or []: if deal_term.get('instant_gratifications'): if not deal_term['instant_gratifications'].get('references'): return RuleResult( rule_error_codes.MISSING_INSTANT_GRATIFICATION_REFERENCES, # noqa REJECT, 'Instant Gratification References are missing' ) return RuleResult( rule_error_codes.MISSING_INSTANT_GRATIFICATION_REFERENCES) def validate_update_indicator(data: Dict) -> RuleResult: """Check update indicator for invalid data.""" update_indicator = data.get('update_indicator') if update_indicator and update_indicator not in VALID_UPDATE_INDICATORS: return RuleResult( rule_error_codes.INVALID_UPDATE_INDICATOR, REJECT, 'Invalid value for update indicator' ) return RuleResult(rule_error_codes.INVALID_UPDATE_INDICATOR) def get_validate_product_price_type(graphql_gateway): """Check product album price type for invalid data.""" def validate_product_price_type(data: Dict) -> RuleResult: """Check product album price type for invalid data.""" deal_terms = next(( deal['deal_terms'] for deal in data['deals'] if 'R0' in deal['release_references'] or 'R1' in deal['release_references']), [] ) price_type = next(( deal_term['price_type'] for deal_term in deal_terms if deal_term['price_type']), None ) # Allow null values if not price_type: return RuleResult(rule_error_codes.INVALID_PRODUCT_PRICE_TYPE) if not str.isdigit(price_type): return RuleResult( rule_error_codes.INVALID_PRODUCT_PRICE_TYPE, REJECT, f'Product price type "{price_type}" is not an integer' ) pricing_tiers = graphql_gateway.execute( queries.ORCHARD_PRICING_FAMILY, {'pricingFamilies': [rule_constants.MUSIC_ALBUM]} )['data']['orchardPricingFamily'][0]['orchardPricingTiers'] for pricing_tier in pricing_tiers: if int(pricing_tier['id']) == int(price_type): return RuleResult(rule_error_codes.INVALID_PRODUCT_PRICE_TYPE) return RuleResult( rule_error_codes.INVALID_PRODUCT_PRICE_TYPE, REJECT, f'Product price type "{price_type}" is not valid' ) return validate_product_price_type def get_validate_track_price_type(graphql_gateway): """Check product track price type for invalid data.""" def validate_track_price_type(data: Dict) -> RuleResult: """Check product track price type for invalid data.""" pricing_tiers = graphql_gateway.execute( queries.ORCHARD_PRICING_FAMILY, {'pricingFamilies': [rule_constants.MUSIC_TRACK]} )['data']['orchardPricingFamily'][0]['orchardPricingTiers'] for track in data['tracks']: isrc = track['isrc'] deal_terms = next(( deal['deal_terms'] for deal in data['deals'] if track['release_reference'] in deal['release_references']), [] ) price_type = next(( deal_term['price_type'] for deal_term in deal_terms if deal_term['price_type']), None ) # Allow null values if not price_type: continue if not str.isdigit(price_type): return RuleResult( rule_error_codes.INVALID_TRACK_PRICE_TYPE, REJECT, f'On track {isrc} price type "{price_type}" is not an integer' # noqa ) valid_price_type = False for pricing_tier in pricing_tiers: if int(pricing_tier['id']) == int(price_type): valid_price_type = True if not valid_price_type: return RuleResult( rule_error_codes.INVALID_TRACK_PRICE_TYPE, REJECT, f'On track {isrc} price type "{price_type}" is not valid' ) return RuleResult(rule_error_codes.INVALID_TRACK_PRICE_TYPE) return validate_track_price_type def validate_awal_not_for_distribution(data: Dict) -> RuleResult: """Validate AWAL delivery Not For Distribution value.""" nfd = data['product'].get('not_for_distribution') if nfd and nfd not in AWAL_NOT_FOR_DISTRIBUTION: return RuleResult( rule_error_codes.INVALID_NOT_FOR_DISTRIBUTION, REJECT, f"Invalid Not For Distribution '{nfd}' for AWAL" ) return RuleResult(rule_error_codes.INVALID_NOT_FOR_DISTRIBUTION) def get_warn_on_format_mismatch(graphql_gateway): """Check if format mismatch between existing product and incoming ddex.""" def warn_on_format_mismatch( data: Dict) -> RuleResult: upc = data['product']['upc'] ddex_format = FORMAT_MAPPING.get(data['product']['release_type']) orchard_product = graphql_gateway.execute( queries.GET_PRODUCT_BY_UPC, {'upc': upc} )['data']['productByUpc'] if orchard_product: orchard_format = orchard_product.get('format') if orchard_format != ddex_format: msg = ('Formats mismatch between incoming DDEX and existing ' f'Orchard Product. Orchard format: {orchard_format} ' f'/ DDEX format: {ddex_format}') return RuleResult( rule_error_codes.MISMATCHING_FORMAT, WARNING, msg ) return RuleResult(rule_error_codes.MISMATCHING_FORMAT) return warn_on_format_mismatch def validate_roles(data: Dict) -> RuleResult: """Apply Orchard validation to track publishing.""" VALID_ROLES = set( list(DDEX_ARTIST_ROLE_TO_ORCHARD_ARTIST_TYPE.keys()) + list(DDEX_ARTIST_ROLE_TO_ORCHARD_WRITER.keys()) + list(DDEX_ARTIST_ROLE_TO_ORCHARD_PERFORMER_ROLE.keys()) + list(DDEX_RESOURCE_CONTRIBUTOR_ROLE_TO_ORCHARD_PERFORMER_ROLE.keys()) + list( DDEX_RESOURCE_CONTRIBUTOR_ROLE_TO_ORCHARD_PARTICIPANT_ROLE.keys()) # noqa + list(DDEX_VIDEO_ROLE_TO_ORCHARD_ARTIST_TYPE.keys()) ) if not data or not data.get('project'): return RuleResult(rule_error_codes.INVALID_ROLE) artist = data['project'].get('artist', {}) or {} for role in artist.get('roles') or []: if role not in VALID_ROLES: return RuleResult( rule_error_codes.INVALID_ROLE, WARNING, f'Invalid role {role} on project artist {artist.get("name")}') for artist in data['product'].get('display_artists') or []: for role in artist.get('roles') or []: if role not in VALID_ROLES: return RuleResult( rule_error_codes.INVALID_ROLE, WARNING, f'Invalid role {role} on product artist ' f'{artist.get("name")}') video = data.get('video') or {} video_artists = video.get('display_artists') or [] video_artists.extend(video.get('resource_contributors') or []) for artist in video_artists: for role in artist.get('roles') or []: if role not in VALID_ROLES: return RuleResult( rule_error_codes.INVALID_ROLE, WARNING, f'Invalid role {role} on video artist ' f'{artist.get("name")}') for track in data['tracks']: track_artists = track.get('display_artists') or [] track_artists.extend(track.get('resource_contributors') or []) for artist in track_artists: for role in artist.get('roles') or []: if role not in VALID_ROLES: return RuleResult( rule_error_codes.INVALID_ROLE, WARNING, f'Invalid role {role} on artist {artist.get("name")} ' f'on track {track.get("isrc")}') return RuleResult(rule_error_codes.INVALID_ROLE) def validate_sme_analytics_not_confidential(data: Dict) -> RuleResult: """Validate sme_analytics isConfidential is false.""" if data['product'].get('is_confidential'): return RuleResult( rule_error_codes.DATA_IS_CONFIDENTIAL, REJECT, 'SME Analytics Data is confidential' ) return RuleResult(rule_error_codes.DATA_IS_CONFIDENTIAL) def validate_sme_analytics_tracks_original_release_date_not_null( data: Dict) -> RuleResult: """Validate sme_analytics original release date is not null. See prep_ddex.prep_preorder_tracks for more details. """ for track in data['tracks']: if 'original_release_date' not in track: return RuleResult( rule_error_codes.MISSING_TRACK_ORIGINAL_RELEASE_DATE, REJECT, 'SME Analytics track is missing original_release_date' ) return RuleResult(rule_error_codes.MISSING_TRACK_ORIGINAL_RELEASE_DATE) def get_common_rules(graphql_gateway): """List of common validation rules.""" return [ require_party_id, require_message_id, require_product, require_product_name, require_product_release_reference, require_product_display_artist_name, require_product_release_type, require_product_is_main_release, require_product_upc, validate_product_participants, validate_product_localized_titles, validate_product_localized_artists, validate_product_localization_language_codes, validate_product_display_artist_language_codes, require_track_name, require_track_isrc, require_track_unique_isrc, require_track_sequence_number, require_track_volume, require_track_release_type, require_track_release_reference, require_track_resource_reference, validate_track_participants, validate_track_localized_titles, validate_track_localized_artists, require_video_composer, require_video_name, require_video_isrc, require_video_release_type, require_video_release_reference, validate_video_participants, validate_video_p_line, validate_video_c_line, require_valid_pline, require_valid_cline, validate_deal_release_references, validate_deal_term_commercial_model_types, validate_deal_term_usage_type, validate_deal_term_territories, validate_instant_gratification_references, validate_update_indicator, require_project_from_ext_meta, validate_roles, get_warn_on_format_mismatch(graphql_gateway), ] def get_validation_rules(ddex_provider: str, application_name, splitio_client, graphql_gateway): """List of validation rules with required clients.""" match ddex_provider: case ddex_providers.SME: return [ *get_common_rules(graphql_gateway), get_require_valid_sme_genre(splitio_client, application_name), get_require_target_vendor_subaccount_matches_existing_product( graphql_gateway), validate_deal_terms, ] case ddex_providers.AWAL: return [ *get_common_rules(graphql_gateway), get_require_valid_awal_genre(graphql_gateway), require_orchard_label, get_validate_product_price_type(graphql_gateway), get_validate_track_price_type(graphql_gateway), validate_awal_not_for_distribution, validate_deal_terms, ] case ddex_providers.RISING_88: return [] case ddex_providers.ALTAFONTE: return [ *get_common_rules(graphql_gateway), get_require_valid_awal_genre(graphql_gateway), require_orchard_label, get_validate_product_price_type(graphql_gateway), get_validate_track_price_type(graphql_gateway), validate_awal_not_for_distribution, validate_deal_terms, ] case ddex_providers.SME_ANALYTICS_PROVIDER: return [ *get_common_rules(graphql_gateway), validate_sme_analytics_tracks_original_release_date_not_null ]