"""Lambda function module.""" from datetime import datetime, timedelta import json import uuid import boto3 import config from constants.genre import SME_ANALYTICS_PROVIDER_GENRE_ID, \ SME_ANALYTICS_PROVIDER_SUBGENRE_ID from ddex_ingester_common.constants.ddex_providers import \ SME_ANALYTICS_PROVIDER, SOM_LIVRE_VENDOR_ID from ddex_ingester_common.constants.format_mapping import FORMAT_MAPPING from ddex_ingester_common.helpers.lambda_warning import LambdaWarning from ddex_ingester_common.helpers.s3_ddex import load_ddex_json from ddex_ingester_common.logging import utils as logging_utils from ddex_ingester_common.models.s3.body import Body as S3Body from ddex_ingester_common.models.state_machine.body import ( Body as StateMachineBody) from ddex_ingester_common.schemas.s3_schema import ParticipantSchema, S3Schema from ddex_ingester_common.schemas.state_machine_schema import ( StateMachineSchema ) from ddex_ingester_common.utils import genre_mapping as genre_mapping_util from ddex_ingester_common.utils.prep_ddex import prep_s3_context from helpers import genre_mapping as genre_mapping_helper logger = logging_utils.get_logger(config.app_logger) def is_ccm_som_livre_grps_ddex_ingestion_enabled(): """Lookup if ccm_som_livre_grps_ddex_ingestion split is enabled.""" split = config.split_client is_enabled = split and split.get_treatment( config.APPLICATION_NAME, 'ccm_som_livre_grps_ddex_ingestion', {'service': config.APPLICATION_NAME} ) == 'on' return is_enabled def handler(event, context): """Prep SME DDEX data with extra info, for instance genre/subgenre.""" s3_context = S3Schema().load(load_ddex_json(event)) context = StateMachineSchema().load(event) correlation_id = context.correlation_id or str(uuid.uuid4()) context.correlation_id = correlation_id logging_utils.update_logger_correlation_id(logger, correlation_id) logger.info(f'Triggered prep_sme_ddex: {event}') logging_utils.update_logger_with_message_ids( logger, s3_context.message_id, s3_context.message_thread_id, s3_context.execution_name ) update_context_with_genre_ids(context, s3_context) add_product_level_producers(s3_context) add_product_level_remixers(s3_context) # Update release type for Som Livre based on their rules if (is_ccm_som_livre_grps_ddex_ingestion_enabled() and s3_context.product.vendor_id == SOM_LIVRE_VENDOR_ID): update_context_with_release_type(context, s3_context) release_type = context.product.release_type prep_s3_context(context.ddex_provider, release_type, s3_context, context.artwork_ingestion_only, is_ccm_som_livre_grps_ddex_ingestion_enabled(), logger) # update context filed with prepared values context.tracks = s3_context.tracks context.product.product_name = s3_context.product.product_name logger.info('Saving s3 context JSON') save_s3_context(context, s3_context) return StateMachineSchema().dump(context) def update_context_with_genre_ids( context: StateMachineBody, s3_context: S3Body): """Update context with genre ids.""" if s3_context.product.genres: genre_code = s3_context.product.genres[0].genre if (not is_ccm_som_livre_grps_ddex_ingestion_enabled() or genre_mapping_helper.is_itunes_genre(genre_code)): genre_map = genre_mapping_util \ .get_orchard_genre_mapping_for_itunes(genre_code) else: rows = genre_mapping_helper.get_genre_and_subgenre_for_grps_rows( logger, genre_code) if not rows or not rows[0]: # Get default genre and subgenre for label vendor_id = s3_context.product.vendor_id subaccount_id = s3_context.product.subaccount_id if vendor_id or subaccount_id: rows = genre_mapping_helper \ .get_default_genre_and_subgenre_for_major_label(logger, vendor_id, subaccount_id) # noqa genre_map = rows[0] exception = ('The GRPS genre was unable to be mapped ' + 'so a default was chosen.') if type(context.warnings) is not list: context.warnings = [] context.warnings.append( LambdaWarning( 'genre', 'GenreMappingNotFoundException', exception )._asdict() ) elif context.ddex_provider == SME_ANALYTICS_PROVIDER: # default SMEAnalytics products to Pop/Pop context.product.genre_id = SME_ANALYTICS_PROVIDER_GENRE_ID context.product.subgenre_id = SME_ANALYTICS_PROVIDER_SUBGENRE_ID # noqa exception = ('We could not map the GRPS genre ' f'"{genre_code}" so we defaulted ' 'the product to Pop/Pop') if type(context.warnings) is not list: context.warnings = [] context.warnings.append( LambdaWarning( 'genre', 'GenreMappingNotFoundException', exception )._asdict()) return else: genre_map = rows[0] if genre_map: context.product.genre_id = genre_map['genre_id'] context.product.subgenre_id = genre_map['subgenre_id'] def update_context_with_release_type( context: StateMachineBody, s3_context: S3Body): """Update context with product type based on Som Livre requirements.""" if s3_context.product.duration and s3_context.tracks: track_length_seconds = convert_duration_str_to_seconds( s3_context.product.duration ) if (len(s3_context.tracks) <= config.SINGLE_MAX_NUM_TRACKS and track_length_seconds <= config.SINGLE_MAX_LENGTH_SECONDS): context.product.release_type = FORMAT_MAPPING['Single'] elif (len(s3_context.tracks) <= config.EP_MAX_NUM_TRACKS and track_length_seconds <= config.EP_MAX_LENGTH_SECONDS): context.product.release_type = FORMAT_MAPPING['EP'] elif (len(s3_context.tracks) > config.EP_MAX_NUM_TRACKS and track_length_seconds > config.EP_MAX_LENGTH_SECONDS): context.product.release_type = FORMAT_MAPPING['Album'] def add_product_level_producers(s3_context: S3Body): """Add missing producers to product level if they are on every track.""" producer_role = 'Producer' producer_track_count = {} for track in s3_context.tracks: artist_list =\ (track.display_artists or []) + (track.resource_contributors or []) for artist in artist_list: if producer_role in (artist.roles or []): if artist.name not in producer_track_count: producer_track_count[artist.name] = 1 else: producer_track_count[artist.name] += 1 nr_tracks = len(s3_context.tracks) producers_to_add = { artist_name for artist_name, count in producer_track_count.items() if count >= nr_tracks } for artist_data in s3_context.product.display_artists: if artist_data.name in producers_to_add: if producer_role not in artist_data.roles: artist_data.roles.append(producer_role) producers_to_add.remove(artist_data.name) nr_artists = len(s3_context.product.display_artists) s3_context.product.display_artists.extend([ ParticipantSchema().load({ 'name': producer_name, 'sequence_number': nr_artists + i + 1, 'spotify_uri': None, 'apple_id': None, 'roles': [producer_role], 'localized_names': [] }) for i, producer_name in enumerate(producers_to_add) ]) def add_product_level_remixers(s3_context: S3Body): """Add missing remixers to product level if they are on every track.""" remixer_role = 'Remixer' remixer_track_count = {} for track in s3_context.tracks: artist_list =\ (track.display_artists or []) + (track.resource_contributors or []) for artist in artist_list: if remixer_role in (artist.roles or []): if artist.name not in remixer_track_count: remixer_track_count[artist.name] = 1 else: remixer_track_count[artist.name] += 1 nr_tracks = len(s3_context.tracks) remixers_to_add = { artist_name for artist_name, count in remixer_track_count.items() if count >= nr_tracks } for artist_data in s3_context.product.display_artists: if artist_data.name in remixers_to_add: if remixer_role not in artist_data.roles: artist_data.roles.append(remixer_role) remixers_to_add.remove(artist_data.name) nr_artists = len(s3_context.product.display_artists) s3_context.product.display_artists.extend([ ParticipantSchema().load({ 'name': remixer_name, 'sequence_number': nr_artists + i + 1, 'spotify_uri': None, 'apple_id': None, 'roles': [remixer_role], 'localized_names': [] }) for i, remixer_name in enumerate(remixers_to_add) ]) def convert_duration_str_to_seconds(duration_string): """Convert duration string to seconds.""" duration_datetime = datetime.strptime(duration_string, 'PT%HH%MM%SS') duration_timedelta = timedelta( minutes=duration_datetime.minute, seconds=duration_datetime.second ) return duration_timedelta.total_seconds() def save_s3_context(context: StateMachineBody, s3_context: S3Body): """Save s3 context.""" s3_client = boto3.client('s3') json_file_path = f'{context.key}parsed_ddex.json' s3_client.put_object( Bucket=context.bucket, Key=json_file_path, Body=json.dumps(S3Schema().dump(s3_context)).encode(encoding='UTF-8') )