"""Media metadata extraction module.""" import json import math import subprocess from src import config from src.config import logger from src.constants import mediainfo_const def get_audio_asset_media_info(file_path): """Collect all information about audio media file using mediainfo. Args: file_path (str): path to file or url. Returns: dict: dictionary with formatted mediainfo output. """ return format_media_info(get_media_info(file_path)) def get_media_info(file_path): """Get mediainfo output for file by filter pattern. Args: file_path (str): File path or url. Returns: dict: Mediainfo output. Raises: RuntimeError: Unable to read asset file technical and tag data. OSError: Any error occurred during meadiainfo work. """ try: cmd_result = subprocess.run( config.MEDIAINFO_CMD + [file_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE ) # mediainfo returns exit code 0 on HTTP errors, check stderr cmd_err = cmd_result.stderr if cmd_err: raise Exception(cmd_err.decode('utf-8').replace('\n', ' ')) media_info_binary_data = cmd_result.stdout except OSError as e: logger.exception(str(e)) raise if not media_info_binary_data: return {} media_info = ( media_info_binary_data.decode('utf-8').replace('\n', '')) if not media_info: return {} result_dict = json.loads(media_info) return result_dict def format_media_info(media_info): """Transform raw mediainfo output. Args: media_info (dict): json loaded direct from mediainfo Returns: dict: flat formatted metadata """ # format data to match mediainfo_const.OUTPUT_MAPPING media_info = { track['@type']: track for track in media_info['media']['track'] if track.get('@typeorder', '1') == '1' } # populate data from file into expected mapping result_dict = {} for track_type, data_map in mediainfo_const.OUTPUT_MAPPING.items(): for raw_name, formatted_name in data_map.items(): data = media_info.get(track_type, {}).get(raw_name) result_dict[formatted_name] = type_cast(formatted_name, data) # JSON output returns seconds, convert for backwards compatibility result_dict['duration_ms'] = ( math.ceil(result_dict['duration'] * 1000) if result_dict['duration'] else None ) del result_dict['duration'] return result_dict def type_cast(key, value): """Cast type of output value from mediainfo. Args: key (str): result key for mediainfo cli output value (str): result value Returns: str or int: return cast value """ if key in mediainfo_const.NUMERIC_FIELDS: if not value: return None if '/' in value: value = value.split('/')[0] float_value = float(value) int_value = int(float_value) if int_value == float_value: return int_value return float_value elif value: return value.lower()