"""Validation rules.""" import datetime import re from typing import Dict, NamedTuple from constants import rule_error_codes from constants.genre_ids import ( CLASSICAL_GENRE_ID, COMPOSER_REQUIRED_SUBGENRE_IDS, SOUNDTRACK_SUBGENRE_IDS, ) from constants.ownership_types import ( IS_OWNER_MAP, OWNERSHIP_TYPES, ) from constants.rules import TRACK_NAME_MAX_LENGTH from constants.us_publishing_obligation import ( OBLIGATION_ENUM, US_PUBLISHING_OBLIGATION_ENUM_MAPPING, ) from ddex_ingester_common.constants.catalog_ingestion import ( REJECT, SUCCESS, ) from ddex_ingester_common.constants.ownership import OWNERSHIP_TYPE_MAPPING from ddex_ingester_common.constants.roles import ( DDEX_ARTIST_ROLE_TO_ORCHARD_WRITER, DDEX_DISPLAY_ARTIST_ROLE_TO_ORCHARD_PRODUCT_ARTIST, DDEX_ROLE_TO_ORCHARD_ARTIST_TYPE, ) from ddex_ingester_common.helpers.metadata import get_language_id from ddex_ingester_common.lambda_exceptions import ( InvalidGenreException, InvalidSubgenreException ) from ddex_ingester_common.utils import genre_mapping as genre_mapping_util from utils.utils import ( has_performer_role, is_artist_name_empty, is_artist_required, is_date_in_range, is_isrc, is_pinfo, parse_date, retrieve_release_dates, ) class RuleResult(NamedTuple): """A result from a rule.""" validation_rule_id: int response: str = SUCCESS message: str = None def orchard_validation_cline_pattern(data: Dict) -> RuleResult: """Apply Orchard pattern validation to cline.""" c_line = data['product'].get('c_line') pattern = r'^[0-9]{4}(,\s[0-9]{4})*(\s\S+)+$' if not c_line: return RuleResult( rule_error_codes.INVALID_CLINE_WORKSTATION, REJECT, 'Missing product C-Line' ) if not re.match(pattern, c_line, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_CLINE_WORKSTATION, REJECT, 'Invalid product C-Line' ) return RuleResult(rule_error_codes.INVALID_CLINE_WORKSTATION) def orchard_validation_date_pattern(data: Dict) -> RuleResult: """Apply Orchard pattern validation to dates.""" pattern = r'^[0-9]{4}-[0-9]{2}-[0-9]{2}' date = data['product'].get('sale_start_date') if date and not re.match(pattern, date, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_DATE_FORMAT, REJECT, f'Invalid product sale start date format: {date}') date = data['product'].get('original_release_date') if date and not re.match(pattern, date, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_DATE_FORMAT, REJECT, f'Invalid product original release date format: {date}') stores = data['product'].get('stores') for store in stores or []: date = store.get('delivery_date') if date and not re.match(pattern, date, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_DATE_FORMAT, REJECT, f'Invalid store delivery date format: {date}') for deal in data['deals'] or []: for deal_term in deal.get('deal_terms') or []: for instant_grat in deal_term.get('instant_gratifications') or []: date = deal_term.get('start_date') if date and not re.match(pattern, date, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_DATE_FORMAT, REJECT, f'Invalid instant grat start date format: {date}') date = deal_term.get('start_date_time') if date and not re.match(pattern, date, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_DATE_FORMAT, REJECT, f'Invalid instant grat start date time: {date}') date = deal_term.get('start_date') if date and not re.match(pattern, date, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_DATE_FORMAT, REJECT, f'Invalid deal term start date format: {date}') date = deal_term.get('start_date_time') if date and not re.match(pattern, date, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_DATE_FORMAT, REJECT, f'Invalid deal term start date time: {date}') date = deal_term.get('release_display_date') if date and not re.match(pattern, date, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_DATE_FORMAT, REJECT, f'Invalid deal term release display date format: {date}') date = deal_term.get('track_listing_preview_date') if date and not re.match(pattern, date, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_DATE_FORMAT, REJECT, f'Invalid deal term track listing preview date format: {date}') # noqa date = deal_term.get('cover_art_preview_date') if date and not re.match(pattern, date, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_DATE_FORMAT, REJECT, f'Invalid deal term cover art preview date format: {date}') date = deal_term.get('clip_preview_date') if date and not re.match(pattern, date, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_DATE_FORMAT, REJECT, f'Invalid deal term clip preview date format: {date}') return RuleResult(rule_error_codes.INVALID_DATE_FORMAT) def orchard_validation_date_range(data: Dict) -> RuleResult: """Apply Orchard range validation to dates.""" MIN_VALID_DATE = datetime.date(1900, 1, 1) MAX_VALID_DATE = datetime.date(2100, 12, 31) date = data['product'].get('sale_start_date') parsed_date = parse_date(date) if date and not is_date_in_range(parsed_date, MIN_VALID_DATE, MAX_VALID_DATE): # noqa return RuleResult( rule_error_codes.INVALID_DATE_RANGE, REJECT, f'Product sale start date {date} must be between ' f'{MIN_VALID_DATE} and {MAX_VALID_DATE}') date = data['product'].get('original_release_date') parsed_date = parse_date(date) if date and not is_date_in_range(parsed_date, MIN_VALID_DATE, MAX_VALID_DATE): # noqa return RuleResult( rule_error_codes.INVALID_DATE_RANGE, REJECT, f'Product original release date {date} must be between ' f'{MIN_VALID_DATE} and {MAX_VALID_DATE}') stores = data['product'].get('stores') for store in stores or []: date = store.get('delivery_date') parsed_date = parse_date(date) if date and not is_date_in_range(parsed_date, MIN_VALID_DATE, MAX_VALID_DATE): # noqa return RuleResult( rule_error_codes.INVALID_DATE_RANGE, REJECT, f'Store delivery date {date} must be between ' f'{MIN_VALID_DATE} and {MAX_VALID_DATE}') for deal in data['deals'] or []: for deal_term in deal.get('deal_terms') or []: for instant_grat in deal_term.get('instant_gratifications') or []: date = deal_term.get('start_date') parsed_date = parse_date(date) if date and not is_date_in_range(parsed_date, MIN_VALID_DATE, MAX_VALID_DATE): # noqa return RuleResult( rule_error_codes.INVALID_DATE_RANGE, REJECT, f'Instant grat start date {date} must be between ' f'{MIN_VALID_DATE} and {MAX_VALID_DATE}') date = deal_term.get('start_date_time') parsed_date = parse_date(date) if date and not is_date_in_range(parsed_date, MIN_VALID_DATE, MAX_VALID_DATE): # noqa return RuleResult( rule_error_codes.INVALID_DATE_RANGE, REJECT, f'Instant grat start date time: {date} must be ' f'between {MIN_VALID_DATE} and {MAX_VALID_DATE}') date = deal_term.get('start_date') parsed_date = parse_date(date) if date and not is_date_in_range(parsed_date, MIN_VALID_DATE, MAX_VALID_DATE): # noqa return RuleResult( rule_error_codes.INVALID_DATE_RANGE, REJECT, f'Deal term start date {date} must be between ' f'{MIN_VALID_DATE} and {MAX_VALID_DATE}') date = deal_term.get('start_date_time') parsed_date = parse_date(date) if date and not is_date_in_range(parsed_date, MIN_VALID_DATE, MAX_VALID_DATE): # noqa return RuleResult( rule_error_codes.INVALID_DATE_RANGE, REJECT, f'Deal term start date time: {date} must be between ' f'{MIN_VALID_DATE} and {MAX_VALID_DATE}') date = deal_term.get('release_display_date') parsed_date = parse_date(date) if date and not is_date_in_range(parsed_date, MIN_VALID_DATE, MAX_VALID_DATE): # noqa return RuleResult( rule_error_codes.INVALID_DATE_RANGE, REJECT, f'Deal term release display date {date} must be between ' f'{MIN_VALID_DATE} and {MAX_VALID_DATE}') date = deal_term.get('track_listing_preview_date') parsed_date = parse_date(date) if date and not is_date_in_range(parsed_date, MIN_VALID_DATE, MAX_VALID_DATE): # noqa return RuleResult( rule_error_codes.INVALID_DATE_RANGE, REJECT, f'Deal term track listing preview date {date} must be ' f'between {MIN_VALID_DATE} and {MAX_VALID_DATE}') date = deal_term.get('cover_art_preview_date') parsed_date = parse_date(date) if date and not is_date_in_range(parsed_date, MIN_VALID_DATE, MAX_VALID_DATE): # noqa return RuleResult( rule_error_codes.INVALID_DATE_RANGE, REJECT, f'Deal term cover art preview date {date} must be between ' f'{MIN_VALID_DATE} and {MAX_VALID_DATE}') date = deal_term.get('clip_preview_date') parsed_date = parse_date(date) if date and not is_date_in_range(parsed_date, MIN_VALID_DATE, MAX_VALID_DATE): # noqa return RuleResult( rule_error_codes.INVALID_DATE_RANGE, REJECT, f'Deal term clip preview date {date} must be between ' f'{MIN_VALID_DATE} and {MAX_VALID_DATE}') return RuleResult(rule_error_codes.INVALID_DATE_RANGE) def orchard_validation_product_required_roles(data: Dict) -> RuleResult: """Apply Orchard validation of required product roles.""" required_roles = {'PRIMARY_ARTIST'} display_artists = data['product'].get('display_artists') for display_artist in display_artists or []: for role in display_artist['roles'] or []: mapped_role = DDEX_DISPLAY_ARTIST_ROLE_TO_ORCHARD_PRODUCT_ARTIST.get(role) # noqa required_roles.discard(mapped_role) if required_roles: return RuleResult( rule_error_codes.MISSING_PRODUCT_ARTIST_ROLE, REJECT, f'Missing required product artist roles: {list(required_roles)}') return RuleResult(rule_error_codes.MISSING_PRODUCT_ARTIST_ROLE) def orchard_validation_composer_required(data: Dict) -> RuleResult: """Apply Orchard validation required composer role for certain genres.""" genres = data['product'].get('genres') if not genres: return RuleResult(rule_error_codes.MISSING_COMPOSER) genre = genres[0].get('genre') subgenre = genres[0].get('subgenre') if not subgenre: return RuleResult(rule_error_codes.MISSING_COMPOSER) try: genre_ids = genre_mapping_util.get_orchard_genre_mapping(genre, subgenre) # noqa except InvalidGenreException: return RuleResult(rule_error_codes.MISSING_COMPOSER) except InvalidSubgenreException: return RuleResult(rule_error_codes.MISSING_COMPOSER) genre_id = genre_ids['genre_id'] subgenre_id = genre_ids['subgenre_id'] require_composer = ( genre_id == CLASSICAL_GENRE_ID or subgenre_id in COMPOSER_REQUIRED_SUBGENRE_IDS or subgenre_id in SOUNDTRACK_SUBGENRE_IDS) if require_composer: display_artists = data['product'].get('display_artists') has_composer = False for display_artist in display_artists or []: for role in display_artist['roles'] or []: mapped_role = DDEX_DISPLAY_ARTIST_ROLE_TO_ORCHARD_PRODUCT_ARTIST.get(role) # noqa if mapped_role == 'COMPOSER': has_composer = True if not has_composer: return RuleResult( rule_error_codes.MISSING_COMPOSER, REJECT, f'Composer is required for genre {genre} and subgenre {subgenre}') # noqa for track in data['tracks'] or []: track_artists = track.get('display_artists') or [] track_artists.extend(track.get('resource_contributors') or []) has_composer = False for artist in track_artists: for role in artist.get('roles') or []: mapped_role = DDEX_DISPLAY_ARTIST_ROLE_TO_ORCHARD_PRODUCT_ARTIST.get(role) # noqa if mapped_role == 'COMPOSER': has_composer = True if not has_composer: return RuleResult( rule_error_codes.MISSING_COMPOSER, REJECT, f'Missing composer on track {track.get("isrc")}') return RuleResult(rule_error_codes.MISSING_COMPOSER) def orchard_validation_product_required_dates(data: Dict) -> RuleResult: """Apply Orchard validation, require certain product dates.""" if not data.get('deals'): return RuleResult( rule_error_codes.MISSING_PRODUCT_DATES, REJECT, 'Product deals are missing') deals = data['deals'] release_dates = retrieve_release_dates(deals) sales_start_date = release_dates.get('sales_start_date') release_date = data['product'].get('original_release_date') \ if data['product'].get('original_release_date') else sales_start_date if not release_date: return RuleResult( rule_error_codes.MISSING_PRODUCT_DATES, REJECT, 'Product release date and sales start date are missing') if not sales_start_date: return RuleResult( rule_error_codes.MISSING_PRODUCT_DATES, REJECT, 'Product sales start date is missing') return RuleResult(rule_error_codes.MISSING_PRODUCT_DATES) def orchard_validation_product_metadata_language(data: Dict) -> RuleResult: """Apply Orchard validation, metadata language is required.""" if not data['product'].get('metadata_language'): return RuleResult( rule_error_codes.MISSING_PRODUCT_METADATA_LANGUAGE, REJECT, 'Product metadata language is missing') return RuleResult(rule_error_codes.MISSING_PRODUCT_METADATA_LANGUAGE) def orchard_validation_product_imprint(data: Dict) -> RuleResult: """Apply Orchard validation, imprint is required.""" if not data['product'].get('imprint'): return RuleResult( rule_error_codes.MISSING_PRODUCT_IMPRINT, REJECT, 'Product imprint is missing') return RuleResult(rule_error_codes.MISSING_PRODUCT_IMPRINT) def orchard_validation_product_code(data: Dict) -> RuleResult: """Apply Orchard validation to product code.""" product_code = data['product'].get('catalog_number') pattern = r'^[a-zA-Z0-9-]*$' if product_code and not re.match(pattern, product_code, flags=re.ASCII): return RuleResult( rule_error_codes.INVALID_PRODUCT_CODE, REJECT, 'Invalid product code') return RuleResult(rule_error_codes.INVALID_PRODUCT_CODE) def orchard_validation_track_name(data: Dict) -> RuleResult: """Apply Orchard validation to track names.""" track_names = set() for track in data.get('tracks') or []: track_full_name = ( track.get('track_name'), track.get('track_version') or '') if track_full_name not in track_names: track_names.add(track_full_name) else: return RuleResult( rule_error_codes.DUPLICATE_TRACK_NAME, REJECT, f'Duplicate name and version {track_full_name} ' f'on track {track.get("isrc")}' ) return RuleResult(rule_error_codes.DUPLICATE_TRACK_NAME) def orchard_validation_track_meta_language_code(data: Dict) -> RuleResult: """Apply Orchard validation to track meta language code.""" for track in data.get('tracks') or []: if not track.get('lyrics_language'): return RuleResult( rule_error_codes.MISSING_TRACK_LANGUAGE, REJECT, f'Missing language on track {track.get("isrc")}' ) return RuleResult(rule_error_codes.MISSING_TRACK_LANGUAGE) def orchard_validation_track_isrc(data: Dict) -> RuleResult: """Apply Orchard validation to track ISRC.""" for track in data.get('tracks') or []: if not is_isrc(track.get('isrc')): return RuleResult( rule_error_codes.INVALID_TRACK_ISRC, REJECT, f'Invalid ISRC format on track {track.get("isrc")}' ) return RuleResult(rule_error_codes.INVALID_TRACK_ISRC) def orchard_validation_track_explicit(data: Dict) -> RuleResult: """Apply Orchard validation to track explicit.""" for track in data.get('tracks') or []: if not track.get('explicit'): return RuleResult( rule_error_codes.MISSING_TRACK_EXPLICIT, REJECT, f'Missing explicit on track {track.get("isrc")}' ) return RuleResult(rule_error_codes.MISSING_TRACK_EXPLICIT) def orchard_validation_track_ownership_rights(data: Dict) -> RuleResult: """Apply Orchard validation to track ownership rights.""" for track in data.get('tracks') or []: ownership_rights = OWNERSHIP_TYPE_MAPPING.get( track.get('ownership_rights')) if ownership_rights not in OWNERSHIP_TYPES: return RuleResult( rule_error_codes.INVALID_TRACK_OWNERSHIP_RIGHTS, REJECT, f'Invalid ownership rights {track.get("ownership_rights")} ' f'on track {track.get("isrc")}' ) return RuleResult(rule_error_codes.INVALID_TRACK_OWNERSHIP_RIGHTS) def orchard_validation_track_recording_country(data: Dict) -> RuleResult: """Apply Orchard validation to track recording country.""" for track in data.get('tracks') or []: ownership_rights = OWNERSHIP_TYPE_MAPPING.get( track.get('ownership_rights')) is_owner = IS_OWNER_MAP.get(ownership_rights) if is_owner: if not track.get('recording_country_code'): return RuleResult( rule_error_codes.INVALID_TRACK_RECORDING_COUNTRY, REJECT, f'Missing recording country on track {track.get("isrc")}' ) return RuleResult(rule_error_codes.INVALID_TRACK_RECORDING_COUNTRY) def orchard_validation_track_copyright_owner_country(data: Dict) -> RuleResult: """Apply Orchard validation to track copyright owner country.""" for track in data.get('tracks') or []: ownership_rights = OWNERSHIP_TYPE_MAPPING.get( track.get('ownership_rights')) is_owner = IS_OWNER_MAP.get(ownership_rights) if is_owner: if not track.get('copyright_owner_country'): return RuleResult( rule_error_codes.INVALID_TRACK_COPYRIGHT_OWNER_COUNTRY, REJECT, f'Missing original rights holder country on track {track.get("isrc")}' # noqa ) return RuleResult(rule_error_codes.INVALID_TRACK_COPYRIGHT_OWNER_COUNTRY) def orchard_validation_track_pline(data: Dict) -> RuleResult: """Apply Orchard validation to pline.""" tracks = data['tracks'] for track in tracks: p_line = track.get('p_line') if not p_line: return RuleResult( rule_error_codes.INVALID_PLINE_WORKSTATION, REJECT, f'Missing P-Line on track {track.get("isrc")}') if not is_pinfo(p_line): return RuleResult( rule_error_codes.INVALID_PLINE_WORKSTATION, REJECT, f'Invalid P-Line on track {track.get("isrc")}') return RuleResult(rule_error_codes.INVALID_PLINE_WORKSTATION) def orchard_validation_track_artists(data: Dict) -> RuleResult: """Apply Orchard validation to track artists.""" tracks = data['tracks'] for track in tracks: track_artists = track.get('display_artists') or [] track_artists.extend(track.get('resource_contributors') or []) if not track_artists: return RuleResult( rule_error_codes.INVALID_TRACK_ARTISTS, REJECT, f'Missing artists on track {track.get("isrc")}') for artist in track_artists: for role in artist['roles']: artist_type = DDEX_ROLE_TO_ORCHARD_ARTIST_TYPE.get(role) if artist_type and is_artist_required(artist_type) and \ is_artist_name_empty(artist['name']): return RuleResult( rule_error_codes.INVALID_TRACK_ARTISTS, REJECT, f'Missing artist name on track {track.get("isrc")}') if not has_performer_role(track): return RuleResult( rule_error_codes.INVALID_TRACK_ARTISTS, REJECT, f'Missing primary artist on track {track.get("isrc")}') return RuleResult(rule_error_codes.INVALID_TRACK_ARTISTS) def orchard_validation_track_featuring(data: Dict) -> RuleResult: """Apply Orchard validation to track featured artists.""" FEATURED_ARTIST_MUST_BE_FEATURED_ON_ALL_TRACKS = True if not FEATURED_ARTIST_MUST_BE_FEATURED_ON_ALL_TRACKS: return RuleResult(rule_error_codes.INVALID_TRACK_FEATURING) product_artists = data['product'].get('display_artists') or [] product_featured_names = set() for artist in product_artists: if 'FeaturedArtist' in (artist.get('roles') or []) \ and artist.get('name'): product_featured_names.add(artist['name']) tracks = data['tracks'] for track in tracks: track_artists = track.get('display_artists') or [] track_artists.extend(track.get('resource_contributors') or []) track_featured_names = set() for artist in track_artists: if 'FeaturedArtist' in (artist.get('roles') or []) \ and artist.get('name'): track_featured_names.add(artist['name']) for name in product_featured_names: if name not in track_featured_names: return RuleResult( rule_error_codes.INVALID_TRACK_FEATURING, REJECT, f'Missing product featured artist on ' f'track {track.get("isrc")}') return RuleResult(rule_error_codes.INVALID_TRACK_FEATURING) def orchard_validation_track_writers(data: Dict) -> RuleResult: """Apply Orchard validation to track writers.""" tracks = data['tracks'] for track in tracks: track_artists = track.get('display_artists') or [] track_artists.extend(track.get('resource_contributors') or []) has_writers = False for artist in track_artists or []: for role in artist.get('roles') or []: if role in DDEX_ARTIST_ROLE_TO_ORCHARD_WRITER: has_writers = True if not has_writers: return RuleResult( rule_error_codes.INVALID_TRACK_ARTISTS, REJECT, f'Missing writers on track {track.get("isrc")}') return RuleResult(rule_error_codes.INVALID_TRACK_WRITERS) def orchard_validation_track_localization(data: Dict) -> RuleResult: """Apply Orchard validation to track localizations.""" tracks = data['tracks'] for track in tracks: localizations = track.get('localized_titles') or [] if len(localizations) > 5: return RuleResult( rule_error_codes.INVALID_TRACK_LOCALIZATION, REJECT, f'Too many localizations on track {track.get("isrc")}') code = data['product'].get('metadata_language') language_id = get_language_id(code, exception=False) CHINESE_TRADITIONAL = 'Chinese Traditional' CHINESE_TRADITIONAL_LANG_ID = 7 CHINESE_SIMPLIFIED = 'Chinese Simplified' CHINESE_SIMPLIFIED_LANG_ID = 6 required_lang_ids = { CHINESE_TRADITIONAL_LANG_ID: CHINESE_SIMPLIFIED_LANG_ID, CHINESE_SIMPLIFIED_LANG_ID: CHINESE_TRADITIONAL_LANG_ID } id_to_language = { CHINESE_TRADITIONAL_LANG_ID: CHINESE_TRADITIONAL, CHINESE_SIMPLIFIED_LANG_ID: CHINESE_SIMPLIFIED } localized_lang_ids = [ get_language_id(localization.get('language_code'), exception=False) # noqa for localization in localizations] if language_id in required_lang_ids: required_lang_id = required_lang_ids[language_id] if required_lang_id not in localized_lang_ids: return RuleResult( rule_error_codes.INVALID_TRACK_LOCALIZATION, REJECT, f'Missing {id_to_language[required_lang_id]} ' f'localization on track {track.get("isrc")}') chinese_lang_localizations_exist = [ required_lang_id in localized_lang_ids for required_lang_id in required_lang_ids] if any(chinese_lang_localizations_exist): if not all(chinese_lang_localizations_exist): return RuleResult( rule_error_codes.INVALID_TRACK_LOCALIZATION, REJECT, f'Track {track.get("isrc")} does not have both ' 'chinese localizations') return RuleResult(rule_error_codes.INVALID_TRACK_LOCALIZATION) def orchard_validation_track_classical(data: Dict) -> RuleResult: """Apply Orchard validation to a classical track.""" genres = data['product'].get('genres') if not genres: return RuleResult(rule_error_codes.INVALID_TRACK_CLASSICAL) genre = genres[0].get('genre') subgenre = genres[0].get('subgenre') if not subgenre: return RuleResult(rule_error_codes.INVALID_TRACK_CLASSICAL) try: genre_ids = genre_mapping_util.get_orchard_genre_mapping(genre, subgenre) # noqa except InvalidGenreException: return RuleResult(rule_error_codes.INVALID_TRACK_CLASSICAL) except InvalidSubgenreException: return RuleResult(rule_error_codes.INVALID_TRACK_CLASSICAL) genre_id = genre_ids['genre_id'] if genre_id != CLASSICAL_GENRE_ID: return RuleResult(rule_error_codes.INVALID_TRACK_CLASSICAL) for track in data['tracks'] or []: track_name = track.get('track_name') or '' work, _, movement = track_name.partition(': ') work = work.strip() if work else '' movement = movement.strip() if movement else '' if not work: return RuleResult( rule_error_codes.INVALID_TRACK_CLASSICAL, REJECT, f'Work of track {track.get("isrc")} is missing') if len(work) > TRACK_NAME_MAX_LENGTH: return RuleResult( rule_error_codes.INVALID_TRACK_CLASSICAL, REJECT, f'Work of track {track.get("isrc")} exceeds the maximum length') # noqa if len(movement) > TRACK_NAME_MAX_LENGTH: return RuleResult( rule_error_codes.INVALID_TRACK_CLASSICAL, REJECT, f'Movement of track {track.get("isrc")} exceeds the maximum length') # noqa return RuleResult(rule_error_codes.INVALID_TRACK_CLASSICAL) def orchard_validation_track_publishing(data: Dict) -> RuleResult: """Apply Orchard validation to track publishing.""" for track in data['tracks'] or []: us_publishing_obligation = US_PUBLISHING_OBLIGATION_ENUM_MAPPING.get( track.get('us_publishing_obligation')) if us_publishing_obligation not in OBLIGATION_ENUM: return RuleResult( rule_error_codes.INVALID_TRACK_PUBLISHING, REJECT, f'Invalid US publishing obligation ' f'{track.get("us_publishing_obligation")} ' f'on track {track.get("isrc")}') return RuleResult(rule_error_codes.INVALID_TRACK_PUBLISHING) ORCHARD_VALIDATION_RULES = [ orchard_validation_cline_pattern, orchard_validation_date_pattern, orchard_validation_date_range, orchard_validation_product_required_dates, orchard_validation_product_required_roles, orchard_validation_composer_required, orchard_validation_track_name, orchard_validation_track_meta_language_code, orchard_validation_track_isrc, orchard_validation_track_explicit, orchard_validation_track_pline, orchard_validation_product_metadata_language, orchard_validation_product_imprint, orchard_validation_product_code, orchard_validation_track_artists, orchard_validation_track_featuring, orchard_validation_track_writers, orchard_validation_track_localization, orchard_validation_track_classical, orchard_validation_track_publishing, orchard_validation_track_ownership_rights, orchard_validation_track_recording_country, orchard_validation_track_copyright_owner_country, ]