import json import os import re import urllib.parse import boto3 from datadog_lambda.metric import lambda_metric # Initialize clients mediaconvert_endpoint = os.environ["MEDIACONVERT_ENDPOINT"] mediaconvert_client = None source_bucket = os.environ["SOURCE_BUCKET"] destination_bucket = os.environ["DESTINATION_BUCKET"] def get_mediaconvert_client(): """Get or create the MediaConvert client with the account-specific endpoint""" global mediaconvert_endpoint global mediaconvert_client if mediaconvert_client is None: # First, we need to get the custom endpoint for this account default_client = boto3.client("mediaconvert", region_name="us-east-1") if not mediaconvert_endpoint: endpoints = default_client.describe_endpoints() mediaconvert_endpoint = endpoints["Endpoints"][0]["Url"] print(f"MediaConvert endpoint: {mediaconvert_endpoint}") # Now create the real client with the custom endpoint mediaconvert_client = boto3.client( "mediaconvert", endpoint_url=mediaconvert_endpoint ) return mediaconvert_client def get_s3_client(): return boto3.client("s3") def is_video_file(key): """Check if the file is a supported video format""" video_extensions = [".mp4", ".mov", ".avi", ".mkv", ".wmv", ".flv", ".webm"] _, ext = os.path.splitext(key.lower()) return ext in video_extensions def get_clip_id_from_audio_key(audio_key): b = os.path.basename(audio_key) return os.path.splitext(b)[0] # key w/o extension def create_user_metadata( audio_key, environment, callback_url: str | None, skip_callback: bool = False ) -> dict[str, str]: metadata = { "environment": environment, "callback-url": callback_url or "", "skip-callback": "true" if skip_callback else "false", } if audio_key: metadata["clip-id"] = get_clip_id_from_audio_key(audio_key) return metadata def get_video_description( is_landscape: bool, size: int, max_bitrate: int, qvbr_quality_level: int ) -> dict: video_description = { "Width": size, "Height": size, "ScalingBehavior": "FIT_NO_UPSCALE", "CodecSettings": { "Codec": "H_264", "H264Settings": { "MaxBitrate": max_bitrate, "RateControlMode": "QVBR", "QvbrSettings": {"QvbrQualityLevel": qvbr_quality_level}, "GopSize": 6.0, # Match GOP_SECONDS from transcode pipeline "GopSizeUnits": "SECONDS", "NumberBFramesBetweenReferenceFrames": 0, # Disable B-frames to prevent timestamp offset "SceneChangeDetect": "DISABLED", # Prevent early keyframes on fades "MinIInterval": 0, # Force keyframe at start (t=0) }, }, } if is_landscape: video_description.pop("Width") else: video_description.pop("Height") return video_description def create_hls_job_params( source_bucket, source_key, destination_path, audio_bucket=None, audio_key=None, environment="staging", callback_url=None, skip_callback=False, is_landscape=False, ): """Create MediaConvert job parameters for HLS output""" params = { "Role": os.environ["MEDIACONVERT_ROLE"], "Settings": { # "ImageBasedTrickPlay": "ADVANCED", # "ImageBasedTrickPlaySettings": { # "IntervalCadence": "FOLLOW_CUSTOM", # "ThumbnailInterval": 1, # "TileWidth": 30, # "TileHeight": 16, # "ThumbnailWidth": 320, # "ThumbnailHeight": 180, # }, "OutputGroups": [ { "Name": "HLS Group", "OutputGroupSettings": { "Type": "HLS_GROUP_SETTINGS", "HlsGroupSettings": { "SegmentLength": 6, "MinSegmentLength": 0, "Destination": destination_path, "SegmentControl": "SEGMENTED_FILES", "ManifestCompression": "NONE", "CodecSpecification": "RFC_4281", "OutputSelection": "MANIFESTS_AND_SEGMENTS", "TimedMetadataId3Frame": "PRIV", # Improve Safari HLS compatibility }, }, "Outputs": [ # 1080p Output { "ContainerSettings": { # Add this section "Container": "M3U8", "M3u8Settings": { "AudioFramesPerPes": 4, "PcrControl": "PCR_EVERY_PES_PACKET", "PmtPid": 480, "PrivateMetadataPid": 503, "ProgramNumber": 1, "PatInterval": 0, "PmtInterval": 0, "TimedMetadata": "NONE", "TimedMetadataPid": 502, "VideoPid": 481, "AudioPids": [482, 483, 484], }, }, "VideoDescription": get_video_description( is_landscape, 1080, 8500000, 9 ), "AudioDescriptions": [ { "AudioSourceName": "Audio Selector 1", "CodecSettings": { "Codec": "AAC", "AacSettings": { "Bitrate": 192000, "CodingMode": "CODING_MODE_2_0", "SampleRate": 48000, }, }, "LanguageCodeControl": "FOLLOW_INPUT", }, ], "OutputSettings": { "HlsSettings": {"SegmentModifier": "_1080p"} }, "NameModifier": "_1080p", }, # 720p Output { "ContainerSettings": { "Container": "M3U8", "M3u8Settings": { "AudioFramesPerPes": 4, "PcrControl": "PCR_EVERY_PES_PACKET", "PmtPid": 480, "PrivateMetadataPid": 503, "ProgramNumber": 1, "PatInterval": 0, "PmtInterval": 0, "TimedMetadata": "NONE", "TimedMetadataPid": 502, "VideoPid": 481, "AudioPids": [482, 483, 484], }, }, "VideoDescription": get_video_description( is_landscape, 720, 4500000, 8 ), "AudioDescriptions": [ { "AudioSourceName": "Audio Selector 1", "CodecSettings": { "Codec": "AAC", "AacSettings": { "Bitrate": 192000, "CodingMode": "CODING_MODE_2_0", "SampleRate": 48000, }, }, "LanguageCodeControl": "FOLLOW_INPUT", }, ], "OutputSettings": { "HlsSettings": {"SegmentModifier": "_720p"} }, "NameModifier": "_720p", }, # 360p Output { "ContainerSettings": { "Container": "M3U8", "M3u8Settings": { "AudioFramesPerPes": 4, "PcrControl": "PCR_EVERY_PES_PACKET", "PmtPid": 480, "PrivateMetadataPid": 503, "ProgramNumber": 1, "PatInterval": 0, "PmtInterval": 0, "TimedMetadata": "NONE", "TimedMetadataPid": 502, "VideoPid": 481, "AudioPids": [482, 483, 484], }, }, "VideoDescription": get_video_description( is_landscape, 360, 1200000, 7 ), "AudioDescriptions": [ { "CodecSettings": { "Codec": "AAC", "AacSettings": { "Bitrate": 128000, "CodingMode": "CODING_MODE_2_0", "SampleRate": 48000, # "RawFormat": "NONE", # "Specification": "MPEG4", # "RateControlMode": "VBR", # "VbrQuality": 4, }, }, } ], "OutputSettings": { "HlsSettings": {"SegmentModifier": "_360p"} }, "NameModifier": "_360p", }, ], } ], "Inputs": [ { "FileInput": f"s3://{source_bucket}/{source_key}", "AudioSelectors": { "Audio Selector 1": { "ExternalAudioFileInput": f"s3://{audio_bucket}/{audio_key}", "SelectorType": "TRACK", "Tracks": [1], } } if (audio_bucket and audio_key) else {"Audio Selector 1": {"DefaultSelection": "DEFAULT"}}, "VideoSelector": {}, "TimecodeSource": "EMBEDDED", } ], }, # S3 Metadata "UserMetadata": create_user_metadata( audio_key, environment, callback_url, skip_callback ), } return params # For a record with the UUID as 'id' def get_nested_path(record): # Decode the URL-encoded UUID base_name = os.path.splitext(record)[0] # Ensure we have at least 4 characters if len(base_name) < 4: # Pad with zeros if needed base_name = base_name.ljust(4, "0") # Ensure it's a valid UUID format (optional validation) first_two = base_name[:2] second_two = base_name[2:4] # Create the nested path structure with the full filename # Sanitize characters that might be problematic in paths sanitized_name = re.sub(r"[^a-zA-Z0-9_\-\.]", "_", base_name) # Create the nested path nested_path = f"{first_two}/{second_two}/{sanitized_name}" return nested_path def get_output_path(full_key): return os.path.splitext(full_key)[0] class S3Metadata: audio_bucket: str | None = None audio_key: str | None = None environment: str callback_url: str | None = None skip_callback: bool = False is_landscape: bool = False def __init__( self, audio_bucket=None, audio_key=None, environment="staging", callback_url=None, skip_callback=False, is_landscape=False, ): self.audio_bucket = audio_bucket self.audio_key = audio_key self.environment = environment self.callback_url = callback_url self.skip_callback = skip_callback self.is_landscape = is_landscape def get_metadata_params(bucket, key) -> S3Metadata: try: response = get_s3_client().head_object(Bucket=bucket, Key=key) print(f"response is {response}") audio_file_bucket = response.get("Metadata", {}).get("audio-file-bucket", None) audio_file_key = response.get("Metadata", {}).get("audio-file-key", None) environment = response.get("Metadata", {}).get("environment", "staging") callback_url = response.get("Metadata", {}).get("callback-url", None) skip_callback = ( response.get("Metadata", {}).get("skip-callback", "false").lower() == "true" ) is_landscape = ( response.get("Metadata", {}).get("is-landscape", "false").lower() == "true" ) print( f"got audio location {audio_file_bucket, audio_file_key} and environment {environment} and callback url {callback_url} and is_landscape {is_landscape} and skip_callback {skip_callback}" ) return S3Metadata( audio_file_bucket, audio_file_key, environment, callback_url, skip_callback, is_landscape, ) except Exception as e: print(f"Error fetching metadata: {str(e)}") return S3Metadata() def handler(event, context): print(f"Event: {json.dumps(event)}") """Lambda handler for S3 event notifications""" results = [] if "Records" not in event: print("No records found in event") return { "statusCode": 200, "body": json.dumps({"message": "No records found in event"}), } for record in event["Records"]: try: # Get the bucket and key from the S3 event bucket = record["s3"]["bucket"]["name"] full_key = urllib.parse.unquote_plus(record["s3"]["object"]["key"]) key = os.path.basename(full_key) print(f"Processing new upload: {bucket}/{key}") # Skip non-video files if not is_video_file(key): print(f"Not a supported video format, skipping: {key}") results.append( { "key": key, "status": "skipped", "reason": "Not a supported video format", } ) continue s3_metadata = get_metadata_params(bucket, full_key) audio_bucket = s3_metadata.audio_bucket audio_key = s3_metadata.audio_key environment = s3_metadata.environment callback_url = s3_metadata.callback_url is_landscape = s3_metadata.is_landscape skip_callback = s3_metadata.skip_callback # Get the MediaConvert client mediaconvert = get_mediaconvert_client() # Generate UUID-based output path {first_two}/{second_two}/{key_without_extension} nested_path = get_output_path(full_key) output_path = f"s3://{destination_bucket}/{nested_path}" print(f"Output path: {output_path}") # Create and submit the job job_params = create_hls_job_params( source_bucket=bucket, source_key=full_key, destination_path=output_path, audio_bucket=audio_bucket, audio_key=audio_key, environment=environment, callback_url=callback_url, skip_callback=skip_callback, is_landscape=is_landscape, ) print(f"Job params: {job_params}") response = mediaconvert.create_job(**job_params) print(f"Response: {response}") job_id = response["Job"]["Id"] print(f"MediaConvert job created: {job_id}") results.append( { "key": key, "status": "processed", "jobId": job_id, "outputPath": output_path, } ) except Exception as e: print( f"Error processing media file {key if 'key' in locals() else 'unknown'}: {str(e)}" ) # send error to datadog instead of failing the lambda - there might be several jobs and we shouldn't fail them all lambda_metric( "media_convert_handler.error", 1, tags=[f"environment:{environment}"] ) results.append( { "key": key if "key" in locals() else "unknown", "status": "error", "error": str(e), } ) return { "statusCode": 200, "body": json.dumps( {"message": f"Processed {len(results)} records", "results": results} ), }