def generate_master_playlist()

in source/controlplaneapi/infrastructure/lambda/Replay/replay_lambda.py [0:0]

• 90 lines of code
• 14 McCabe index (conditional complexity)

def generate_master_playlist(event, context):
    '''
        Generates a master Playlist with various Quality levels representing different resolutions
        #EXTM3U
        #EXT-X-STREAM-INF:BANDWIDTH=25000000,RESOLUTION=3840x2160
        4K/TestProgram_AK555_1_00002Part-1.m3u8
        #EXT-X-STREAM-INF:BANDWIDTH=25000000,RESOLUTION=2560x1440
        2K/TestProgram_AK555_1_00002Part-1.m3u8
        #EXT-X-STREAM-INF:BANDWIDTH=6000000,RESOLUTION=1920x1080
        1080p/TestProgram_AK555_1_00002Part-1.m3u8
        #EXT-X-STREAM-INF:BANDWIDTH=3000000,RESOLUTION=1280x720
        720p/TestProgram_AK555_1_00002Part-1.m3u8
    '''

    from MediaReplayEngineWorkflowHelper import ControlPlane
    controlplane = ControlPlane()

    playlist_content = []
    playlist_content.append('#EXTM3U')
    bucket = ''
    thumbnail_location = '-'
    does_play_list_exist = False
    for jobResult in event['CreateHlsJobsResult']["Payload"]:

        resolution = jobResult['Resolution']

        res_desc = get_resolution_hls_desc(resolution)
        playlist_content.append(res_desc)
        if 'JobMetadata' in jobResult:
            if len(jobResult['JobMetadata']) > 0:
                s3_path = jobResult['JobMetadata'][0]['OutputDestination'].split('/')

                #['s3', '', 'aws-mre-clip-gen-output', 'HLS', 'ak555-testprogram', '4K', '']
                bucket = s3_path[2]
                keyprefix = f"{s3_path[3]}/{s3_path[4]}/{s3_path[5]}/"

                manifests = get_all_manifests(bucket, keyprefix)
                for manifest in manifests:
                    manifest_path = manifest.split('/')
                    playlist_content.append(f"{resolution.replace(':', '')}/{manifest_path[3]}")
                    
                does_play_list_exist = True #We know at least 1 HLS manifest exists, mark this so we can trigger the creation of the Master Manifest

        # Get the location of the Thumbnail generated for the Hls stream
        # We pick up the last thumbnail location in the CreateHlsJobsResult list
        if 'ThumbnailLocations' in jobResult:
            if len(jobResult['ThumbnailLocations']) > 0:
                # S3 location of the Thumbnail (without the file name generated by MediaConvert).
                key_name = list(jobResult['ThumbnailLocations'][0].keys())[0]
                thumbnail_loc = jobResult['ThumbnailLocations'][0][key_name]

                # Get the thumbnail file s3 Path
                tmp_loc = thumbnail_loc.split('/')
                thumbnail_location_tmp_loc = f"{tmp_loc[3]}/{tmp_loc[4]}/{tmp_loc[5]}/{tmp_loc[6]}/"
                thumbnail_location = get_s3_thumbnail_path(thumbnail_location_tmp_loc)
                thumbnail_location = f"s3://{OUTPUT_BUCKET}/{thumbnail_location}"

    playlist_location = ''
    # If the Playlist content has no manifest file locations of HLS, theres no point in creating a new Master Manifest
    if does_play_list_exist:
        create_manifest_file(playlist_content)
        
        event_name = event['ReplayRequest']['Event']
        program_name = event['ReplayRequest']['Program']
        replay_request_id = event['ReplayRequest']['ReplayId']
        batch_id = event['CreateHlsJobsResult']["Payload"][0]['JobMetadata'][0]['BatchId']

        #key_prefix = f"HLS/{event_name.lower()}-{program_name.lower()}-{replay_request_id}/master-playlist.m3u8"
        key_prefix = f"HLS/{batch_id}/master-playlist.m3u8"

        # Upload final Manifest File to S3
        # s3_client.upload_file("/tmp/main.m3u8", bucket, f"{keyPrefix}/main.m3u8", ExtraArgs={'ACL':'public-read'})
        s3_client.upload_file("/tmp/main.m3u8", bucket, key_prefix)

        playlist_location = f"s3://{bucket}/{key_prefix}"

        payload = {
            "Event": event_name,
            "Program": program_name,
            "ReplayRequestId": replay_request_id,
            "HlsLocation": playlist_location,
            "Thumbnail": thumbnail_location
        }

        controlplane.update_replay_request_with_hls_location(payload)

        # Update the Event that a replay has been created for the Event
        controlplane.update_event_has_replays(event_name, program_name)

        # Notify EventBridge
        detail = {
            "State": "REPLAY_PROCESSED",
            "Event": {
                "Event": event_name,
                "Program": program_name,
                "ReplayId": replay_request_id,
                "EventType": "REPLAY_GEN_DONE"
            }
        }
        eb_client.put_events(
            Entries=[
                {
                    "Source": "awsmre",
                    "DetailType": "Base Replay Data Exported",
                    "Detail": json.dumps(detail),
                    "EventBusName": EB_EVENT_BUS_NAME
                }
            ]
        )

        runId = 'NA'
        if 'CurrentReplayResult' in event:
            if 'RunId' in event['CurrentReplayResult']:
                runId = event['CurrentReplayResult']['RunId']
        
        if runId != 'NA':
            end_time = datetime.datetime.now()
            print(f"------- Updated HLS Location - RunID - {runId} ----------------- {end_time}")

    return {
        "MasterPlaylist": 'No playlist was found. This may not always be an error.' if playlist_location == '' else playlist_location
    }