"""validation handler.""" import datetime import os import shlex import subprocess import ffmpeg from botocore.errorfactory import ClientError from ddex_ingester_common.models.state_machine.body import \ Body as StateMachineContext from ddex_ingester_common.schemas.state_machine_schema import \ StateMachineSchema import config from constants.constants import (ALLOWED_RESOLUTIONS, ALLOWED_VERTICAL_RESOLUTIONS, BLACK_VIDEO_FILE_NAME, MERGE_ORDER_FILE, MIN_BIT_RATE, NEAR_BLACK, NO_AUDIO_BLACK_VIDEO_FILE_NAME, NR_FRAMES) logger = config.app_logger s3_client = config.s3_client def handler(event, context): """Lambda entry point.""" logger.info(f'Triggered fix_video_resolution: {event}') if 'context' in event: context = StateMachineSchema().load(event.get('context')) else: context = StateMachineSchema().load(event) local_path = config.PATH_WITH_WRITE_ACCESS if not context.video or not context.video.assets: logger.info('Mising video asset.') return # Create a new name for the original file new_file_name = generate_file_name(context) # Upload original video file with a different name to avoid losing it rename_video_file(context, new_file_name) # Get URL of the video file from S3 to use in video processing input_file_url = get_video_file_url(context, new_file_name) # Add black frame to start of video processed_video_file = fix_video_resolution(local_path, input_file_url) # Replace original video file with the fixed video upload_video_file(context, processed_video_file) # Start another State Machine Execution to ingest the new video trigger_new_state_machine_execution(context) return StateMachineSchema().dump(context) def generate_file_name(context: StateMachineContext) -> str: """Generate a new name for the original video file.""" original_file_name = context.video.assets[0].filename current_datetime = datetime.datetime.now() time = current_datetime.strftime('%Y-%m-%d_%H-%M-%S') split_name = original_file_name.split('.') new_file_name = split_name[0] + '_Original_' + time + '.' + split_name[1] logger.info(f'Generated new video file name: {new_file_name}') return new_file_name def rename_video_file( context: StateMachineContext, new_name: str): """Change the name of the original video file to make sure we keep it.""" video_asset = context.video.assets[0] bucket = video_asset.bucket original_name_key = video_asset.key key_no_file_name = get_key_without_file_name(context) new_name_key = key_no_file_name + new_name logger.info( f'Renaming original video file to {new_name}. ' f'Bucket: {bucket} Key: {new_name_key}') # You can't rename object in S3, you need to create a new one # No need to delete the old one as it will be replaced by the fixed video copy_source = { 'Bucket': bucket, 'Key': original_name_key } s3_client.copy(copy_source, bucket, new_name_key) def get_video_file_url( context: StateMachineContext, new_file_name: str): """Get a URL for the video file with the given name from S3.""" video_asset = context.video.assets[0] bucket = video_asset.bucket key_no_file_name = get_key_without_file_name(context) key = key_no_file_name + new_file_name logger.info( 'Generating presigned URL for S3 video file. ' f'Bucket: {bucket} Key: {key} Timeout: {config.S3_SIGNED_URL_TIMEOUT}') return s3_client.generate_presigned_url( 'get_object', Params={'Bucket': bucket, 'Key': key}, ExpiresIn=config.S3_SIGNED_URL_TIMEOUT) def upload_video_file( context: StateMachineContext, video_data: bytes): """Upload the fixed video file to S3.""" video_asset = context.video.assets[0] bucket = video_asset.bucket key = video_asset.key logger.info( 'Uploading processed video file to S3. ' f' Bucket: {bucket} Key: {key} Video Size: {len(video_data)} bytes') s3_client.put_object( Body=video_data, Bucket=bucket, Key=key) def get_key_without_file_name(context: StateMachineContext) -> str: """Get video asset key without the file name. Always has "/" at the end.""" # Example of video asset context data # bucket: 'qa-ddex-ingester' # key: 'sme_ddex/Gui_Video_Scratch_Folder/resources/EXPV494614_RECV2091549.mov' # noqa # filepath: 'resources/' # filename: 'EXPV494614_RECV2091549.mov' video_asset = context.video.assets[0] key = video_asset.key filepath = video_asset.filepath split_s3_key = key.split(filepath) s3_key_without_file = split_s3_key[0] + '/' + filepath + '/' s3_key_without_file = s3_key_without_file.replace('//', '/') logger.info( f'Retrieved the key without file name: {s3_key_without_file} ' f'From the original key: {key}') return s3_key_without_file def trigger_new_state_machine_execution(context: StateMachineContext): """Trigger a new State Machine Execution to ingest the new video.""" # We can't start an execution in the same way as trigger_state_machine # That method requires the State Machine ARN # A lambda inside the State Machine can't have its ARN as an environment # variable since terraform creates the lambdas before the State Machine # To trigger another execution we place the XML on the S3 bucket again product = context.product file_identifier = product.grid if product.grid else product.upc xml_file_name = file_identifier + '.xml' bucket = context.bucket key = context.key ddex_key = key + xml_file_name new_file_key = key + 'reupload_' + xml_file_name # Check if file already exists to avoid infinite loops try: logger.info(f'Checking if {bucket} file exists: {new_file_key}') s3_client.head_object(Bucket=bucket, Key=new_file_key) logger.info(f'Will not start a new execution, file already exists: {new_file_key}') # noqa: E501 return except ClientError: logger.info(f'File does not exist: {new_file_key}') logger.info(f'Starting DDEX file upload to Bucket: {bucket} Key: {ddex_key}') # noqa: E501 xml_file_data = s3_client.get_object( Bucket=bucket, Key=ddex_key, )['Body'].read() s3_client.put_object( Bucket=bucket, Key=new_file_key, Body=xml_file_data, ) def fix_video_resolution( local_path: str, input_file_url: str, output_file_name: str = '') -> bytes: """Script entry point.""" # output_file name is only set when runninng locally file_ext =\ '.' + output_file_name.split('.')[1] if output_file_name else '.mp4' black_video_file = BLACK_VIDEO_FILE_NAME + file_ext no_audio_black_video_file = NO_AUDIO_BLACK_VIDEO_FILE_NAME + file_ext black_video_path = local_path + black_video_file no_audio_black_video_path = local_path + no_audio_black_video_file merge_order_path = local_path + MERGE_ORDER_FILE logger.info( f'Starting video processing. Input: {input_file_url}') probe = ffmpeg.probe(input_file_url) video_stream = next( (stream for stream in probe['streams'] if stream['codec_type'] == 'video'), None ) audio_stream = next( (stream for stream in probe['streams'] if stream['codec_type'] == 'audio'), None ) logger.info(f'Video stream data: {video_stream}') logger.info(f'Audio stream data: {audio_stream}') width = int(video_stream['width']) height = int(video_stream['height']) video_codec = video_stream['codec_name'] video_bit_rate = int(video_stream['bit_rate']) video_bit_rate = video_bit_rate if video_bit_rate > MIN_BIT_RATE else MIN_BIT_RATE # noqa size_in_gb = float(probe['format']['size']) / (1024 * 1024 * 1024) re_encode, width, height = is_reencode_needed(width, height) preset = get_encoder_preset(size_in_gb) logger.info(f'ReEncode: {re_encode} Preset: {preset}') metadata = get_file_metadata(video_stream, audio_stream, probe) # From https://video.stackexchange.com/questions/20717/ffmpeg-add-3-seconds-of-black-to-video-head-and-tail # noqa # and https://video.stackexchange.com/questions/29791/ffmpeg-adding-blankspace-to-end-of-video # noqa # and https://superuser.com/questions/1096921/concatenating-videos-with-ffmpeg-produces-silent-video-when-the-first-video-has # noqa # Also check: https://ffmpeg.org/documentation.html create_black_video( width, height, video_stream, audio_stream, black_video_path, no_audio_black_video_path) create_merge_order_file(black_video_file, input_file_url, merge_order_path) video_data = merge_videos( width, height, re_encode, video_codec, preset, video_bit_rate, metadata, merge_order_path, output_file_name, ) clear_local_files([ no_audio_black_video_path, black_video_path, merge_order_path, ]) return video_data def is_reencode_needed(width: int, height: int) -> tuple: """Check if we need to reencode the output video and to what resolution. It is only needed if the source video does not have an allowed resolution. Black bars are added to pad the video to a standard resolution. """ def find_allowed_res(width: int, height: int, res_list: list): for allowed_width, allowed_height in res_list: if width == allowed_width and allowed_height > height: return width, allowed_height if height == allowed_height and allowed_width > width: return allowed_width, height return None, None if (width, height) in ALLOWED_RESOLUTIONS or \ (width, height) in ALLOWED_VERTICAL_RESOLUTIONS: return False, width, height if width > height: res_list = ALLOWED_RESOLUTIONS else: res_list = ALLOWED_VERTICAL_RESOLUTIONS for i in range(500): new_width, new_height = find_allowed_res(width + i, height, res_list) if new_width and new_height: return True, new_width, new_height new_width, new_height = find_allowed_res(width, height + i, res_list) if new_width and new_height: return True, new_width, new_height raise Exception(f'Video Resolution {height}x{width} cannot be fixed.') def get_encoder_preset(size: float) -> str: """Determine what encoder preset to use based on file size. This is needed to use the best quality encoder possible without the Lambda exceeding the maximum processing time of 15 minutes. """ if size > 10: raise Exception(f'File does not fit in Lambda memory. Size: {size} GB') # Lower limits if we get timeouts if size < 0.5: preset = 'medium' elif size < 1.0: preset = 'fast' elif size < 1.5: preset = 'faster' elif size < 2.0: preset = 'veryfast' elif size < 2.5: preset = 'superfast' else: preset = 'ultrafast' return preset def get_file_metadata( video_stream: dict, audio_stream: dict, probe: dict) -> str: """Return the file metadata as a string of ffmpeg tags.""" global_tags = probe.get('format', {}).get('tags') metadata = '' for tag in global_tags or {}: value = global_tags[tag] metadata = metadata + f'-metadata {tag}="{value}" ' video_tags = video_stream.get('tags') if video_stream else [] for tag in video_tags: value = video_tags[tag] metadata = metadata + f'-metadata:s:v {tag}="{value}" ' audio_tags = audio_stream.get('tags') if audio_stream else [] for tag in audio_tags: value = audio_tags[tag] metadata = metadata + f'-metadata:s:a {tag}="{value}" ' return metadata def create_black_video( width: int, height: int, video_stream: dict, audio_stream: dict, black_video_path: str, no_audio_black_video_path: str): """Create a video file with a few frames of near black video.""" frame_rate = eval(video_stream['avg_frame_rate']) duration = (1 / frame_rate) * NR_FRAMES time_scale = 1 / eval(video_stream['time_base']) video_codec = video_stream['codec_name'] audio_codec = audio_stream['codec_name'] sample_rate = audio_stream['sample_rate'] if audio_stream else '44100' channels = int(audio_stream['channels']) if audio_stream else 1 silence_expr = '0' if channels == 1 else '0|0' # Create a video with 3 frames of a color that is very close to black result = os.system( f'ffmpeg -y -f lavfi -i ' f'color=c={NEAR_BLACK}:s={width}x{height}:r={frame_rate}:d={duration} ' f'-c:v {video_codec} -c:a {audio_codec} ' f'-video_track_timescale {time_scale} ' f'{no_audio_black_video_path} {config.LOG_LEVEL}') if result: logger.info(f'Failed to create black video. Exit code: {result}') # Add audio to the black video # If this video has no audio the merged result will lack audio as well result = os.system( f'ffmpeg -y -i {no_audio_black_video_path} -f lavfi -i ' f'aevalsrc="{silence_expr}:s={sample_rate}" -shortest -y ' f'-c:v {video_codec} -c:a {audio_codec} ' f'-video_track_timescale {time_scale} ' f'{black_video_path} {config.LOG_LEVEL}') if result: logger.info(f'Failed to add audio to black video. Exit code: {result}') def create_merge_order_file( black_video_file: str, input_file_url: str, merge_order_path: str): """Create a file with the merge order for the video files. The expected format is a text file with the file paths of the files: file 'PATH_TO_FILE.mp4' file 'PATH_TO_OTHER_FILE.mp4' The path to the file can also be a URL. """ result = os.system( 'echo ' f'"file \'{black_video_file}\'\nfile \'{input_file_url}\'"' f' > {merge_order_path}') if result: logger.info(f'Failed to create merge order file. Exit code: {result}') def merge_videos( width: int, height: int, re_encode: bool, video_codec: str, preset: str, video_bit_rate: int, metadata: str, merge_order_path: str, output_file_name: str): """Merge the black video with the source video. If there is no need to reencode it just concatenates both videos. It never reencodes audio and always copies the original as is. """ # From https://stackoverflow.com/questions/49343174/can-ffmpeg-concatenate-files-from-a-different-domain # noqa allow_network_files = '-safe 0 -protocol_whitelist file,http,https,tcp,tls' # If the video is not the correct resolution, add black bars # From https://stackoverflow.com/questions/46671252/how-to-add-black-borders-to-video # noqa padding = ( f'"scale={width}:{height}:force_original_aspect_ratio=decrease,' f'pad={width}:{height}:(ow-iw)/2:(oh-ih)/2,setsar=1"') bit_rate = ( f'-vf {padding} -c:v {video_codec} -preset {preset} -b:v {video_bit_rate} -c:a copy' # noqa # TODO Should use this line instead but the output frame rate is wrong # if re_encode else '-c copy') if re_encode else f'-c:v {video_codec} -preset {preset} -b:v {video_bit_rate} -c:a copy') # noqa out_format = ( output_file_name if output_file_name else # Only works for h264, from https://stackoverflow.com/questions/55698581/create-mp4-file-from-raw-h264-using-a-pipe-instead-of-files # noqa # Pipe the output video bytes to stdout, from https://aws.amazon.com/blogs/media/processing-user-generated-content-using-aws-lambda-and-ffmpeg/ # noqa '-f mp4 -movflags frag_keyframe+empty_moov pipe:1' ) # Merge the video files ffmpeg_command = ( f'ffmpeg -y -f concat {allow_network_files} -i {merge_order_path} ' f'{bit_rate} {metadata} {config.LOG_LEVEL} {out_format}') logger.info(f'FFMPEG command: {ffmpeg_command}') # Run the FFMPEG command in a subprocess to capture the stdout result # By returning the command output we return the bytes of the output video split_command = shlex.split(ffmpeg_command) return subprocess.check_output(split_command) def clear_local_files(file_list: list): """Remove local files created during the execution. We want to remove them in case the Lambda is reused. """ for path in file_list: os.system(f'rm -rf {path}') def run_locally(input_file_name: str, output_file_name: str): """Run video processing locally to help with debugging.""" # Setting output_file_name writes the file to disk # The file is exported by other operations in the makefile fix_video_resolution( '', input_file_name, output_file_name=output_file_name )