"""Product utils.""" from typing import Dict, List from bulk_metadata_ingester_common.constants.file import CSV_VALUE_SEPARATOR from bulk_metadata_ingester_common.models.bulk_release import BulkRelease from bulk_metadata_ingester_common.utils.catalog_ingestion import \ save_catalog_ingestion_action from bulk_metadata_ingester_common.utils.error import graphql_execute from config import graphql_gateway from constants import queries from constants.exceptions import ( DuplicateParticipantException, ParticipantRoleGenreException) from constants.languages import LANGUAGE_CODE_MAP from constants.product import ( BULK_NOT_FOR_DISTRIBUTION, BULK_PRODUCT_HIGHLIGHTS) from constants.roles import ( BULK_FIELD_REQUIRED_GENRES_SUBGENRES, BULKRELEASE_FIELD_TO_RELEASE_ARTIST_TYPE ) from ddex_ingester_common.constants.catalog_ingestion import ( INSERT_ACTION, UPDATE_ACTION) from ddex_ingester_common.constants.format_mapping \ import FORMAT_MAPPING as DDEX_FORMAT_MAPPING def check_for_product(model: BulkRelease, logger: object) -> Dict: """Check if a product exists using its UPC. Args: model (BulkRelease): The product model to check. logger (object): The logger object to use for logging. Returns: Dict: A dictionary containing the product information if it exists, otherwise None. """ # If no UPC passed, create product and get UPC. if not model.upc: return payload = { 'upc': str(model.upc) } result = graphql_execute( graphql_gateway, queries.GET_PRODUCT_BY_UPC, payload, logger )['data']['productByUpc'] return result def map_release_type(release: BulkRelease) -> str: """Map the release type of a BulkRelease object to a DDEX format. Args: release: A BulkRelease object representing the release to be mapped. Returns: str: A string representing the mapped release type. Raises: ValueError: If the release type is not recognized. """ # Handle Orchard-style (i.e. Bulk) release type using DDEX mapping targets if release.release_type not in DDEX_FORMAT_MAPPING.values(): # Maybe they passed a DDEX format release_type = DDEX_FORMAT_MAPPING.get(release.release_type) if not release_type: raise ValueError(f'Unknown format: {release.release_type}') else: # Type recognized release_type = release.release_type return release_type def create_product(event: Dict, release: BulkRelease, logger: object) -> Dict: """Create a product using the provided release data. Args: event (Dict): The event that triggered the creation of the product. release (BulkRelease): The bulk release data to use for creating the product. logger (object): The logger object to use for logging information. Returns: Dict: A dictionary containing the result of the createProduct GraphQL mutation. """ # Convert the release type for the payload. release_type = map_release_type(release) # Get the participations for the release participations = get_participations(release, logger) # Shape the payload payload = { 'data': { 'productName': release.release_name, 'productCode': str(release.product_code), 'productHighlights': BULK_PRODUCT_HIGHLIGHTS, 'projectId': release.project_id, 'accountId': release.vendor_id, 'subaccountId': release.subaccount_id or None, 'upc': str(release.upc), 'metaLanguage': LANGUAGE_CODE_MAP[release.metadata_language], # TWO VERSION FIELDS # 'deliveredVersion': release.product_version, # The below is really product_version_notes. # This is where the bulk template product version data is stored in OA. # noqa 'version': release.product_version, # TODO This should be renamed and extended in the template # noqa 'pLine': release.p_line, # 'cLine': release.c_line, 'format': release_type, 'imprint': release.imprint, 'notForDistribution': BULK_NOT_FOR_DISTRIBUTION, 'participations': participations['artists'], 'genreId': release.genre_id, 'subgenreId': release.subgenre_id, 'manufacturerUpc': ( str(release.manufacturer_upc) if release.manufacturer_upc else str(release.upc)), # Saved for future use: # 'specialInstructions': sanitize_special_instructions( # model.special_instructions), # 'productLocalizations': get_title_localizations(s3_data.product), # 'vendorReleaseIdentifier': model.manufacturer_upc, } } c_line = release.c_line if c_line: payload['data']['cLine'] = c_line result = graphql_execute( graphql_gateway, queries.CREATE_PRODUCT, payload, logger )['data']['createProduct'] # Update the DB if result: save_catalog_ingestion_action( event, release, {}, INSERT_ACTION ) return result def update_product(event: Dict, release: BulkRelease, logger: object) -> Dict: """Update an existing product in the catalog. Args: event (Dict): A dictionary containing information about the event that triggered the function. release (BulkRelease): An instance of the BulkRelease class representing the product to be updated. logger (object): An object used for logging messages. Returns: Dict: A dictionary containing information about the updated product. """ # Convert the release type for the payload. release_type = map_release_type(release) # Get the participations for the release participations = get_participations(release, logger) payload = { 'data': { 'productName': release.release_name, # 'productCode': str(release.product_code), # Can't update 'productId': release.product_id, # 'projectId': str(release.project_id), # Can't Update 'accountId': release.vendor_id, 'productHighlights': BULK_PRODUCT_HIGHLIGHTS, 'metaLanguage': LANGUAGE_CODE_MAP[release.metadata_language], # TWO VERSION FIELDS # 'deliveredVersion': release.product_version, # The below is really product_version_notes. # This is where the bulk template product version data is stored in OA. # noqa # 'version': release.product_version, # TODO This should be renamed and extended in the template # noqa 'participations': participations['artists'], 'pLine': release.p_line, 'genreId': release.genre_id, 'subgenreId': release.subgenre_id, 'imprint': release.imprint, 'notForDistribution': BULK_NOT_FOR_DISTRIBUTION, 'format': release_type, 'manufacturerUpc': ( str(release.manufacturer_upc) if release.manufacturer_upc else str(release.upc)), # Saved for future use: # 'specialInstructions': sanitize_special_instructions( # context.product.special_instructions), # 'productLocalizations': get_title_localizations(s3_data.product), # 'vendorReleaseIdentifier': model.project_code, } } c_line = release.c_line if c_line: payload['data']['cLine'] = c_line # TODO This should be renamed and extended in the template # noqa product_version_notes = release.product_version if product_version_notes: payload['data']['version'] = product_version_notes result = graphql_execute( graphql_gateway, queries.UPDATE_PRODUCT, payload, logger )['data']['updateProduct'] save_catalog_ingestion_action( event, release, {}, UPDATE_ACTION ) return result def get_all_roles(release: BulkRelease, logger: object) -> Dict: """Return a dictionary with every role along with the participant's name. Args: release (BulkRelease): An instance of BulkRelease class. logger (object): An instance of logger class. Returns: Dict: A dictionary containing all roles and their respective participants' names. """ # Log message logger.info( f'Getting all roles for: {release.upc} - "{release.release_name}"') genre = release.genre subgenre = release.subgenre release_performer_list = [] # Handle release-level artists for field in BULKRELEASE_FIELD_TO_RELEASE_ARTIST_TYPE: try: # If field value is empty, move along. if not getattr(release, field): continue except AttributeError: # If field is something else, move along. continue # Check if role is special case / classical if field in BULK_FIELD_REQUIRED_GENRES_SUBGENRES: # Get genre list required_genres = \ BULK_FIELD_REQUIRED_GENRES_SUBGENRES[field].keys() # Normalize case for comparison required_genres = [g.lower() for g in required_genres] # Check if genre is acceptable for role if genre.lower() not in required_genres: # TODO: explode, or drop silently? msg = f'Genre "{genre}" is not acceptable for role "{field}"' raise ParticipantRoleGenreException(msg) # continue # Get subgenre list required_subgenres = \ BULK_FIELD_REQUIRED_GENRES_SUBGENRES[field][genre] # Normalize case for comparison required_subgenres = [sg.lower() for sg in required_subgenres] # Check if subgenre is acceptable for role # (Empty list means all subgenres are acceptable) if len(required_subgenres) > 0 and \ subgenre.lower() not in required_subgenres: # TODO: explode, or drop silently? msg = f'Subgenre "{subgenre}" is not acceptable for role "{field}"' # noqa raise ParticipantRoleGenreException(msg) # continue try: # Try to split the field, if multiple artists # Does this work for XLSX? artists = getattr(release, field).split(CSV_VALUE_SEPARATOR) # make list, and strip artists = [a.strip() for a in artists if a.strip()] except (AttributeError, ValueError, TypeError): # Garbage - move along continue # If empty set, move along if not artists: continue # Format for payload for artist in artists: release_performer = { 'role': BULKRELEASE_FIELD_TO_RELEASE_ARTIST_TYPE[field], 'name': artist } # Add to payload release_performer_list.append(release_performer) # Return payloads all_roles = { 'artists': release_performer_list } return all_roles def add_release_artist( label_participant_uuid: str, performer: Dict, release_performers: List[Dict]) -> List[Dict]: """Add artist to list of release performers for create/update product. Args: label_participant_uuid (str): The UUID of the label participant. performer (Dict): A dictionary containing details of the performer. release_performers (List[Dict]): A list of dictionaries containing details of all the release performers. This list is updated. Returns: None: But the list of release performers is updated with the the new artist if it's not already present. """ # Get role from from full role list role = performer['role'] # Format for payload new_artist = { 'labelParticipantUuid': label_participant_uuid, 'role': role, } # Omit duplicates if new_artist not in release_performers: release_performers.append(new_artist) def get_participations( release: BulkRelease, logger: object) -> Dict: """Get the payload of participants for createProduct GraphQL mutation. Args: release (BulkRelease): An instance of BulkRelease class containing a `participants` section. logger (object): logger Returns: Dict: A dictionary containing details of all the artists who participated in the release. """ # Log message logger.info( f'Getting all participants for: {release.upc} - ' f'"{release.release_name}"') release_artists = [] # Get the participation list from set_project via context. all_participants = release.participants # Get all roles all_roles = get_all_roles(release, logger) # Loop through artists for artist in all_roles['artists']: context_participant = \ get_context_participant(all_participants, artist['name']) label_participant_uuid = context_participant['label_participant_uuid'] add_release_artist( label_participant_uuid, artist, release_artists, ) return { 'artists': release_artists, } def get_context_participant( all_participants: List[Dict], name_to_find: str) -> Dict: """Return the participant object from context by name. Args: all_participants (List[Dict]): Participants section of the BulkRelease. name_to_find (str): The name of the participant to be found. Returns: Dict: A dictionary containing the label participant uuid, etc. of the participant with the passed name. Raises: DuplicateParticipantException: If multiple participants are found with the same name. """ matching_participants = [] for participant in all_participants: if participant['name'] == name_to_find: matching_participants.append(participant) if not matching_participants: return None elif len(matching_participants) > 1: raise DuplicateParticipantException( f'Found multiple participants with the same name: ' f'{matching_participants}') else: return matching_participants[0]