"""Common DDEX metadata helper functions.""" import re from typing import Dict, List, Optional, Union from ddex_ingester_common.constants.cline import ( INVALID_CHAR_SEQUENCES as INVALID_CLINE_SEQUENCES) from ddex_ingester_common.constants.country_codes import ( COUNTRY_CODE_TO_COUNTRY_ID) from ddex_ingester_common.constants.language_codes import ( ISO_639_2_LETTER_TO_3_LETTER, ISO_639_3_LETTER_TO_2_LETTER, LANGUAGE_CODE_TO_LANGUAGE_ID, ORCHARD_ALLOWED_LANGUAGE_CODES, OTHER_CODES_TO_ORCHARD_LANGUAGE_CODES, PRODUCT_LANGUAGE_CODE, PRODUCT_ONLY_CODES_TO_ORCHARD_LANGUAGE_CODES, TRACK_LANGUAGE_CODE, TRACK_ONLY_CODES_TO_ORCHARD_LANGUAGE_CODES, VIDEO_LANGUAGE_CODE, VIDEO_ONLY_CODES_TO_ORCHARD_LANGUAGE_CODES, ) from ddex_ingester_common.constants.pline import ( INVALID_CHAR_SEQUENCES as INVALID_PLINE_SEQUENCES) from ddex_ingester_common.constants.publishing import ( US_PUBLISHING_OBLIGATION_MAPPING ) from ddex_ingester_common.constants.video import ( DDEX_VIDEO_TYPE_TO_ORCHARD_VIDEO_TYPE, DEFAULT_ORCHARD_VIDEO_TYPE ) def map_explicit_string(ddex_explicit: str) -> str: """Get explicit string to be used in track metadata. Args: ddex_explicit: ParentalWarningType field from DDEX """ explicit_string_map = { 'EXPLICIT': 'Y', 'EXPLICITCONTENTEDITED': 'C', 'NOTEXPLICIT': 'N', } if not ddex_explicit: return None return explicit_string_map.get(ddex_explicit.upper()) def format_lyrics(lyrics: Union[str, None]) -> Union[str, None]: """Format lyrics by removing CDATA sections.""" if lyrics: return lyrics.replace('', '') def format_pline(pline: str, pline_year: str = None) -> str: """Format pLine by removing invalid characters.""" if not pline: return None formatted_pline = '' try: formatted_pline = re.sub( r'|'.join( map( re.escape, [*INVALID_CLINE_SEQUENCES, *INVALID_PLINE_SEQUENCES] ) ), '', pline, 1).strip() # Combine whitespace throughout the pline formatted_pline = re.sub(r'\s\s+', ' ', formatted_pline) if pline_year and not re.match( '[0-9]{4}[ ]+', formatted_pline, flags=re.ASCII): formatted_pline = f'{pline_year} {formatted_pline}' except (AttributeError, TypeError): pass return formatted_pline def format_publishers(publishers: List[Dict]) -> Optional[List[str]]: """Format publishers.""" key = 'PublisherName' return [p.get(key) for p in publishers if p.get(key)] or None def process_language_code(language_code: str, code_type: str) -> str: """Map DDEX language code to Orchard language code. Args: language_code: language code to process code_type: if the code applies to track or product """ if not language_code: return None code = language_code.upper() if code_type == TRACK_LANGUAGE_CODE: unique_mappings = TRACK_ONLY_CODES_TO_ORCHARD_LANGUAGE_CODES elif code_type == PRODUCT_LANGUAGE_CODE: unique_mappings = PRODUCT_ONLY_CODES_TO_ORCHARD_LANGUAGE_CODES elif code_type == VIDEO_LANGUAGE_CODE: unique_mappings = VIDEO_ONLY_CODES_TO_ORCHARD_LANGUAGE_CODES else: unique_mappings = {} if code in unique_mappings: return unique_mappings[code] elif code in ISO_639_2_LETTER_TO_3_LETTER: return ISO_639_2_LETTER_TO_3_LETTER[code] elif code in OTHER_CODES_TO_ORCHARD_LANGUAGE_CODES: return OTHER_CODES_TO_ORCHARD_LANGUAGE_CODES[code] elif code in ORCHARD_ALLOWED_LANGUAGE_CODES: return code else: raise Exception(f'Invalid language code {code}') def remove_non_alphanumeric_characters(text: str) -> str: """Remove non alphanumeric characters from a string.""" return re.sub('[^0-9a-zA-Z]+', '', text) def get_country_id(country_code: str) -> int: """Get country ID for given country code.""" if country_code: return COUNTRY_CODE_TO_COUNTRY_ID.get(country_code.upper()) return None def get_language_id(language_code: str, exception=True) -> int: """Get language ID for given language code.""" if language_code: language_id = LANGUAGE_CODE_TO_LANGUAGE_ID.get(language_code.upper()) if not language_id and exception: raise Exception(f'Invalid language code {language_code}') if not language_id: processed_code = ISO_639_3_LETTER_TO_2_LETTER.get( language_code.upper()) language_id = LANGUAGE_CODE_TO_LANGUAGE_ID.get( processed_code.upper()) return language_id return None def get_us_publishing_obligation(text: str) -> str: """Map US Publishing Obligation.""" publishing_obligation = None if text: processed_text = remove_non_alphanumeric_characters(text).upper() publishing_obligation =\ US_PUBLISHING_OBLIGATION_MAPPING.get(processed_text) return publishing_obligation def map_video_type(video_type: str) -> str: """Map DDEX video type to Orchard video type.""" if not video_type: return DEFAULT_ORCHARD_VIDEO_TYPE video_type_key = remove_non_alphanumeric_characters(video_type) return DDEX_VIDEO_TYPE_TO_ORCHARD_VIDEO_TYPE.get( video_type_key, DEFAULT_ORCHARD_VIDEO_TYPE) def map_video_parental_warning(parental_warning: str) -> str: """Map DDEX video parental warning to Orchard video parental warning.""" explicit_string_map = { 'EXPLICIT': 'Yes', 'EXPLICITCONTENTEDITED': 'Clean Version', 'NOTEXPLICIT': 'No', } if not parental_warning: return None return explicit_string_map.get(parental_warning.upper())