"""Base packager module.""" import asyncio import json import os import pathlib import subprocess import tempfile import uuid import xml.etree.ElementTree as ET from abc import ABC from abc import abstractmethod import config import requests import src.exceptions from src.connectors.graphql_router import GraphQLConnector from src.connectors.ows_assets import OWSAssetsConnector from src.connectors.ows_metadata import OWSMetadataConnector from src.connectors.s3 import S3Connector from src.utils import images class AbstractPackager(ABC): """Parent packager class.""" dms_id = None s3_connector = None ows_assets_connector = None ows_metadata_connector = None graphql_connector = None def __init__(self): """Create new instance of packager.""" self.s3_connector = S3Connector() self.ows_metadata_connector = OWSMetadataConnector() self.ows_assets_connector = OWSAssetsConnector() self.graphql_connector = GraphQLConnector() async def package(self, stereo_upc, spatial_upc, job_id): """Validate and assemble spatial audio package. Args: stereo_upc (str): delivered stereo product spatial_upc (str): new spatial product job_id (str): unique identifier of this job run Returns: None """ await self.verify_output_dir_empty(job_id) # get manually configured data and validate it stereo_metadata = self.get_product_metadata(stereo_upc) spatial_metadata = self.get_product_metadata(spatial_upc) spatial_audio_objects = \ await self.get_spatial_audio_objects(spatial_upc) self.validate_setup( stereo_metadata, spatial_metadata, spatial_audio_objects ) # get spatial files metadata and validate it self.validate_spatial_audio_metadata( await self.get_media_metadata(spatial_audio_objects) ) # fetch data to be transformated stereo_xml = self.get_xml(stereo_upc, spatial_upc) stereo_artwork_filename = await self.download_artwork( stereo_metadata['productId'] ) # calculate MD5 and filesize of objects on S3 (slow operation) spatial_audio_details = \ await self.get_objects_details(spatial_audio_objects) # perform data transformations spatial_artwork_filename = self.transform_image( stereo_artwork_filename ) spatial_artwork_details = \ self.get_image_details(spatial_artwork_filename) spatial_xml = self.transform_xml( stereo_xml, stereo_metadata, spatial_metadata, spatial_audio_details, spatial_artwork_details ) # copy/upload final package data to S3 await self.commit_package( spatial_upc, stereo_upc, spatial_audio_objects, spatial_xml, spatial_artwork_filename, job_id ) async def verify_output_dir_empty(self, job_id): """Verify if output dir exists already. Raises: OutputDirectoryAlreadyExists Returns: None """ if await self.s3_connector.list_objects( config.OUTPUT_BUCKET, job_id): raise src.exceptions.OutputDirectoryAlreadyExists() def get_product_metadata(self, upc): """Retrieve product details from graphql. Args: upc (str): product to fetch data for Returns: dict: metadata about product """ return self.graphql_connector.get_metadata_by_upc(upc)['metadata'] def get_xml(self, stereo_upc, spatial_upc): """Retrieve xml from ows-metadata. Args: stereo_upc (str): stereo product upc spatial_upc (str): spatial product upc Returns: str: xml for this dms Raises: XMLFetchFailedReleaseCarveout XMLFetchFailedLabelCarveout XMLFetchFailedSubaccountCarveout XMLFetchFailedLabelDeleted XMLFetchFailedLabelContractNotActive """ try: (response, body) = self.ows_metadata_connector.get_metadata( self.get_xml_url(stereo_upc, spatial_upc) ) response.raise_for_status() except requests.exceptions.HTTPError as e: if e.response.status_code == 500: error_message = e.response.content.decode('utf-8') if 'Release has master DMS' in error_message: raise src.exceptions.XMLFetchFailedReleaseCarveout() elif 'Label contract has master DMS' in error_message: raise src.exceptions.XMLFetchFailedLabelCarveout() elif 'Subaccount has master DMS' in error_message: raise src.exceptions.XMLFetchFailedSubaccountCarveout() elif 'Label is deleted' in error_message: raise src.exceptions.XMLFetchFailedLabelDeleted() elif 'Label does not have an active contract' in error_message: raise src.exceptions.XMLFetchFailedLabelContractNotActive() raise e return body['metadata'] @abstractmethod def get_xml_url(self, stereo_upc, spatial_upc): """Format URL to use in get_xml.""" pass async def get_spatial_audio_objects(self, spatial_upc): """Retrieve list of spatial objects from S3. Args: spatial_upc (upc): product to fetch data for Returns: list[tuple]: location of objects on S3 0 - object bucket 1 - object path Raises: NoObjectsFound NoWavAssetsFound """ keys = await self.s3_connector.list_objects( config.INPUT_BUCKET, config.INPUT_BUCKET_PREFIX + spatial_upc ) if not keys: raise src.exceptions.NoObjectsFound() keys = [x for x in keys if x.lower().endswith('.wav')] if not keys: raise src.exceptions.NoWavAssetsFound() return [ (config.INPUT_BUCKET, key) for key in keys ] async def get_media_metadata(self, media_objects): """Analyze objects on S3 using mediainfo. Args: media_objects (list[tuple]): s3 object locations Returns: list[tuple]: 0 (str): object bucket location 1 (str): object key location 2 (dict): object media derived metadata Raises: MediaInfoStderr MediaInfoNonZeroExit """ results = list() signed_objects = await asyncio.gather( *[ self.s3_connector.get_signed_url(bucket, key) for bucket, key in media_objects ] ) for bucket, key, url in signed_objects: try: cmd_result = subprocess.run( ['mediainfo', '-f', '--Output=JSON', url], stdout=subprocess.PIPE, stderr=subprocess.PIPE ) # mediainfo returns exit code 0 on HTTP errors, check stderr if cmd_result.stderr: raise src.exceptions.MediaInfoStderr() media_info_binary_data = cmd_result.stdout except OSError: raise src.exceptions.MediaInfoNonZeroExit() media_info = media_info_binary_data.decode('utf-8').replace('\n', '') # noqa:E501 results.append(( bucket, key, json.loads(media_info) )) return results async def get_objects_details(self, objects): """Analyze objects on S3 for additional details. Args: objects (list[tuple]): s3 object locations Returns: list[tuple]: 0 (str): bucket location of object 1 (str): key location of object 2 (str): md5 of object body 3 (int): size of object body """ return await asyncio.gather( *[ self.s3_connector.analyze_object(bucket, key) for bucket, key in objects ] ) def get_image_details(self, filename): """Analyze image file on disk. Args: filename (str): image file location on disk Returns: tuple: 0 (str): MD5 of file body 1 (int): width of image 2 (int): height of image """ return images.analyze(filename) async def download_artwork(self, product_id): """Save product artwork to disk. Args: product_id (int): unique id Returns: str: filename of image on disk """ (bucket, key) = \ self.ows_assets_connector.get_mezzanine_tif_s3_location(product_id) output_filename = os.path.join( tempfile.gettempdir(), str(uuid.uuid4()) + pathlib.Path(key).suffix ) await self.s3_connector.download_fileobj(bucket, key, output_filename) return output_filename def validate_setup( self, stereo_metadata, spatial_metadata, spatial_audio_objects ): """Check setup is correct. Args: stereo_metadata (dict): details of stereo product spatial_metadata (dict): details of spatial product spatial_audio_objects (list[tuple]): spatial audio files on s3 Raises: InvalidNotForDistributionFlag InvalidNumberOfSpatialTracks StereoProductNotDeliveredToDMSError NoStereoMatchForSpatialTrackVolumeIndex DuplicateSpatialTrackIsrcError StereoIsrcMatchesSpatialIsrcError InvalidSpatialFileUpc Returns: None """ stereo_upc = stereo_metadata['upc'] spatial_upc = spatial_metadata['upc'] # Collect stereo tracks data for validation later on stereo_isrcs = list() stereo_tracks = stereo_metadata['tracks'] stereo_volume_index_map = list() for stereo_track in stereo_tracks: volume = stereo_track['volumeNumber'] index = stereo_track['trackNumber'] isrc = stereo_track['isrc'] stereo_volume_index_map.append((volume, index)) stereo_isrcs.append(isrc) # Validate S3 spatial filename format and contents spatial_isrcs = list() for bucket, path in spatial_audio_objects: (upc, volume, track_index, track_isrc) = \ self._extract_spatial_filename_parts(path) # Validate spatial track volume & index maps to a stereo track if (volume, track_index) not in stereo_volume_index_map: raise src.exceptions.NoStereoMatchForSpatialTrackVolumeIndex( volume, track_index ) # Validate spatial UPC is correct # The UPC from S3 filename (upc) can include a leading zero, so to # ensure correct comparison, we convert both UPCs into integers if int(upc) != int(spatial_upc): raise src.exceptions.InvalidSpatialFileUpc(upc) # Build spatial ISRCs list spatial_isrcs.append(track_isrc) # Validate no duplicate spatial ISRCs in S3 files if len(set(spatial_isrcs)) != len(spatial_isrcs): raise src.exceptions.DuplicateSpatialTrackIsrcError( 'Duplicate track ISRC in S3 spatial product.' ) # Validate spatial tracks <= stereo tracks if len(spatial_isrcs) > len(stereo_isrcs): raise src.exceptions.InvalidNumberOfSpatialTracks( f'Number of spatial tracks ({len(spatial_isrcs)}) exceeds ' f'number of stereo tracks ({len(stereo_isrcs)}).' ) # Validate exactly 1 audio file per spatial track if len(spatial_metadata['tracks']) != len(spatial_audio_objects): raise src.exceptions.InvalidNumberOfSpatialFiles( f"Number of spatial files ({len(spatial_metadata['tracks'])})" f' != number of spatial tracks ({len(spatial_audio_objects)}).' ) # Validate no intersection between stereo and spatial ISRCs intersecting_isrcs = \ [isrc for isrc in spatial_isrcs if isrc in stereo_isrcs] if intersecting_isrcs: raise src.exceptions.StereoIsrcMatchesSpatialIsrcError( intersecting_isrcs ) # Delivery history validations stereo_delivery_history = stereo_metadata['deliveryHistory'] stereo_is_delivered = False for record in stereo_delivery_history: # Validate stereo product delivered if record['store']['storeId'] == self.dms_id: stereo_is_delivered = True break if not stereo_is_delivered: raise src.exceptions.StereoProductNotDeliveredToDMSError( stereo_upc ) # Validate notForDistribution flag spatial_nfd_flag = spatial_metadata['notForDistribution'] if spatial_nfd_flag != 'NotforFurtherDistribution': raise src.exceptions.InvalidNotForDistributionFlag( spatial_nfd_flag ) def validate_spatial_audio_metadata(self, audio_metadata): """Check spatial files are valid. Args: audio_metadata (list[dict]): mediainfo output from spatial files Raises: MissingAudioTrack MissingExtraMetadata InvalidAudioChannelCount InvalidAudioMetadataFormat InvalidAdmProfile InvalidAudioFormat InvalidAudioBitDepth InvalidAudioSamplingRate Returns: None """ # examine each file's data for _, key, metadata in audio_metadata: # extract audio track try: audio_track = [ x for x in metadata['media']['track'] if x['@type'] == 'Audio' ][0] except (IndexError, KeyError): raise src.exceptions.MissingAudioTrack(key) # perform general validations channels = int(audio_track['Channels']) if channels <= 2: raise src.exceptions.InvalidAudioChannelCount(key, channels) audio_format = audio_track['Format'] if not audio_format == 'PCM': raise src.exceptions.InvalidAudioFormat(key, audio_format) bit_depth = int(audio_track['BitDepth']) if not bit_depth == 24: raise src.exceptions.InvalidAudioBitDepth(key, bit_depth) sampling_rate = int(audio_track['SamplingRate']) if not sampling_rate == 48000: raise src.exceptions.InvalidAudioSamplingRate( key, sampling_rate ) # extract extra data for spatial try: extra_metadata = audio_track['extra'] except KeyError: raise src.exceptions.MissingExtraMetadata(key) # perform spatial validations adm_format = extra_metadata.get('Metadata_Format', '') if not adm_format.startswith('ADM'): raise src.exceptions.InvalidAudioMetadataFormat( key, adm_format ) adm_profile = extra_metadata.get('AdmProfile_Format', '') if not adm_profile == 'Dolby Atmos Master': raise src.exceptions.InvalidAdmProfile(key, adm_profile) def format_xml(self, xml): """Format xml into string. Args: xml (xml.etree.ElementTree.Element): input parsed xml Returns: str: formatted decoded xml """ ET.indent(xml, space=' ', level=0) return ET.tostring( xml, encoding='UTF-8', xml_declaration=True ).decode('utf-8') @abstractmethod def transform_xml( self, input_xml, stereo_metadata, spatial_metadata, spatial_audio_details, spatial_artwork_details): """Splice stereo xml into spatial xml to store's spec. Args: input_xml (str): stock xml to transform stereo_metadata (dict): details of stereo product spatial_metadata (dict): details of spatial product spatial_audio_details (list[dict]): md5 and filesize spatial_artwork_details (tuple): details of artwork file Returns: str: spatial xml """ pass @abstractmethod def transform_image(self, image_location): """Modify artwork to store's spec. Args: image_location (str): filename of image on disk to modify Returns: str: filename of modified image """ pass async def commit_package( self, spatial_upc, stereo_upc, audio_objects, xml_data, image_filename, job_id): """Prepare final package on s3 to store's spec. Args: spatial_upc (str): unique upc for spatial product stereo_upc (str): unique upc for stereo product audio_objects (list[dict]): s3 locations of spatial files xml_data (str): spatial formatted xml data image_filename (str): spatial formatted artwork filename job_id (str): unique id for this job Returns: None """ base = self.output_base_path(job_id, spatial_upc, stereo_upc) await self.s3_connector.upload_object( xml_data.encode('utf-8'), config.OUTPUT_BUCKET, self.output_metadata_filepath(base, spatial_upc) ) trigger_filename = self.output_trigger_filename(base) if trigger_filename: await self.s3_connector.upload_object( b'', config.OUTPUT_BUCKET, job_id + '/' + trigger_filename ) if image_filename: await self.s3_connector.upload_object( open(image_filename, 'rb'), config.OUTPUT_BUCKET, self.output_image_filepath(base, spatial_upc) ) copy_objects = [ ( bucket, path, config.OUTPUT_BUCKET, self.output_audio_path(base, spatial_upc) + '/' + self.output_audio_filename(path) # noqa:E501 ) for (bucket, path) in audio_objects ] await asyncio.gather( *[ self.s3_connector.copy_object(*x) for x in copy_objects ] ) def output_trigger_filename(self, base_path): """Generate filename of zero byte trigger file. Args: base_path (str): path to package of files Returns str """ pass @abstractmethod def output_base_path(self, job_id, spatial_upc, stereo_upc): """Root directory for output files. Args: job_id (str): unique id for this job run spatial_upc (str): spatial upc of product being packaged stereo_upc (str): stereo sibling of product being packaged Returns: str: s3 directory path """ pass @abstractmethod def output_image_filepath(self, base, upc): """Image file output location. Args: base (str): dir path base from output_base_path() upc (str): upc of product being packaged Returns: str: s3 file path """ pass @abstractmethod def output_metadata_filepath(self, base, upc): """Metadata XML file output location. Args: base (str): dir path base from output_base_path() upc (str): upc of product being packaged Returns: str: s3 file path """ pass @abstractmethod def output_audio_path(self, base, upc): """Audio file output dir location. Args: base (str): dir path base from output_base_path() upc (str): upc of product being packaged Returns: str: s3 directory path """ pass def output_audio_filename(self, s3_path): """Translate input filename to output filename. Args: s3_path (str): input filename Returns: str: output filename """ (upc, volume, track, _) = self._extract_spatial_filename_parts(s3_path) return f'{upc}_{volume}_{track}.wav' def _extract_spatial_filename_parts(self, s3_path): """Help function to split spatial s3 filepath. Args: s3_path (str): location of file on s3 Returns: tuple: 0 (str): upc 1 (str): volume index 2 (str): track index 3 (str): track isrc """ try: assert s3_path.lower().endswith('.wav') filename = s3_path.split('/')[-1].split('.')[0] file_parts = filename.split('_') formatted_parts = ( file_parts[0], int(file_parts[1]), int(file_parts[2]), file_parts[3] ) except (IndexError, ValueError, AssertionError): raise src.exceptions.InvalidSpatialFilename(s3_path) return formatted_parts def _format_spatial_file_list(self, spatial_audio_details): """Help function to format file data by name. Args: spatial_audio_details (list[tuple]): output from get_object_details 0 (str): s3 bucket 1 (str): s3 path 2 (str): file md5 3 (int): file size in bytes Returns: list[dict]: int: volume index extracted from filename int: track index extracted from filename isrc (str): isrc extracted from fioename md5 (str): md5 of file body size (int): byte size of file body """ spatial_files = dict() for (bucket, key, md5, filesize) in spatial_audio_details: (upc, volume_index, track_index, isrc) = \ self._extract_spatial_filename_parts(key) if volume_index not in spatial_files: spatial_files[volume_index] = dict() if track_index in spatial_files[volume_index]: raise src.exceptions.DuplicateSpatialFileIndexes(f'{volume_index}_{track_index}') # noqa:E501 spatial_files[volume_index][track_index] = { 'upc': upc, 'isrc': isrc, 'md5': md5, 'size': filesize, 'bucket': bucket, 'path': key } return spatial_files