import media.CDDB import time import subprocess from media import common import math from PIL import Image import os import logging import sys try: from config import APP_NAME as ROOT_LOGGER_NAME except ImportError: ROOT_LOGGER_NAME = 'root' logger = logging.getLogger(ROOT_LOGGER_NAME).getChild(__name__) try: import discid except OSError: logger.exception("You might install python3-libdiscid") __KEY_CAMERA_VENDOR__ = 'camera_vendor' __KEY_CAMERA_MODEL__ = 'camera_model' def get_media_data(full_path, user_callback=None): # ft = common.get_filetype(full_path) # if ft == common.FILETYPE_AUDIO: return get_audio_data(full_path) elif ft == common.FILETYPE_IMAGE: return get_image_data(full_path) elif ft == common.FILETYPE_VIDEO: return get_video_data(full_path) elif ft == common.FILETYPE_DISC: return get_disc_data(full_path, user_callback) else: logger.warning('Filetype not known: %s', full_path) def get_audio_data(full_path): conv_key_dict = {} conv_key_dict['album'] = (str, common.KEY_ALBUM) conv_key_dict['TAG:album'] = (str, common.KEY_ALBUM) conv_key_dict['TAG:artist'] = (str, common.KEY_ARTIST) conv_key_dict['artist'] = (str, common.KEY_ARTIST) conv_key_dict['bit_rate'] = (__int_conv__, common.KEY_BITRATE) conv_key_dict['duration'] = (float, common.KEY_DURATION) conv_key_dict['TAG:genre'] = (str, common.KEY_GENRE) conv_key_dict['genre'] = (str, common.KEY_GENRE) conv_key_dict['TAG:title'] = (str, common.KEY_TITLE) conv_key_dict['title'] = (str, common.KEY_TITLE) conv_key_dict['TAG:track'] = (__int_conv__, common.KEY_TRACK) conv_key_dict['track'] = (__int_conv__, common.KEY_TRACK) conv_key_dict['TAG:date'] = (__int_conv__, common.KEY_YEAR) conv_key_dict['date'] = (__int_conv__, common.KEY_YEAR) return __adapt__data__(__get_xxprobe_data__(full_path, conv_key_dict), full_path) def get_disc_data(full_path, user_callback): # # Read Information from CDDB database # did = media.CDDB.discid() if did is None: logger.error("Could not determine disc id...") sys.exit(1) q = media.CDDB.query(did) if q is None: data = { common.KEY_ARTIST: None, common.KEY_ALBUM: None, common.KEY_YEAR: None, common.KEY_GENRE: None } for i in range(0, int(did.split('+')[1])): data["track_%02d" % (i + 1)] = None data = user_callback(common.CALLBACK_MAN_INPUT, data) return media.CDDB.my_disc_metadata(**data) if len(q) == 1: # Only one entry did = tuple(q.keys())[0] else: # multiple entries (choose) if user_callback is None: logger.warning("No usercallback to handle multiple cddb choices...") sys.exit(1) did = user_callback(common.CALLBACK_CDDB_CHOICE, q) return media.CDDB.cddb(did) """ musicbrainzngs.set_useragent("pyrip", "0.1", "your@mail") disc = discid.read() disc_id = disc.id disc_data = {} try: result = musicbrainzngs.get_releases_by_discid(disc_id, includes=["artists", "recordings"]) except musicbrainzngs.ResponseError: logger.exception("disc not found or bad response") sys.exit(1) else: disc_data[common.KEY_ARTIST] = result["disc"]["release-list"][0]["artist-credit-phrase"] disc_data[common.KEY_ALBUM] = result["disc"]["release-list"][0]["title"] disc_data[common.KEY_YEAR] = int(result["disc"]["release-list"][0]["date"][:4]) data_copy = dict(disc_data) disc_data["id"] = result["disc"]["release-list"][0]["id"] disc_data["tracks"] = [] # get tracklist for entry in result["disc"]["release-list"][0]["medium-list"][0]["track-list"]: track = dict(data_copy) track[common.KEY_TITLE] = entry["recording"]["title"] track[common.KEY_TRACK] = int(entry['number']) disc_data["tracks"].append(track) return disc_data """ def get_video_data(full_path): conv_key_dict = {} conv_key_dict['creation_time'] = (__vid_datetime_conv__, common.KEY_TIME) conv_key_dict['TAG:creation_time'] = (__vid_datetime_conv__, common.KEY_TIME) conv_key_dict['bit_rate'] = (__int_conv__, common.KEY_BITRATE) conv_key_dict['duration'] = (float, common.KEY_DURATION) conv_key_dict['height'] = (__int_conv__, common.KEY_HEIGHT) conv_key_dict['width'] = (__int_conv__, common.KEY_WIDTH) conv_key_dict['display_aspect_ratio'] = (__ratio_conv__, common.KEY_RATIO) return __adapt__data__(__get_xxprobe_data__(full_path, conv_key_dict), full_path) def get_image_data(full_path): return __adapt__data__(__get_exif_data__(full_path), full_path) def __adapt__data__(data, full_path): data[common.KEY_SIZE] = os.path.getsize(full_path) # Join Camera Vendor and Camera Model if __KEY_CAMERA_MODEL__ in data and __KEY_CAMERA_VENDOR__ in data: model = data.pop(__KEY_CAMERA_MODEL__) vendor = data.pop(__KEY_CAMERA_VENDOR__) data[common.KEY_CAMERA] = '%s: %s' % (vendor, model) # Add time if not exists if common.KEY_TIME not in data: if common.KEY_YEAR in data and common.KEY_TRACK in data: if data[common.KEY_YEAR] != 0: # ignore year 0 - must be wrong # Use a date where track 1 is the newest in the given year minute = int(data[common.KEY_TRACK] / 60) second = (data[common.KEY_TRACK] - 60 * minute) % 60 # data[common.KEY_TIME] = int(time.mktime((data[common.KEY_YEAR], 1, 1, 0, 59 - minute, 59 - second, 0, 0, 0))) data[common.KEY_TIME_IS_SUBSTITUTION] = True else: data[common.KEY_TIME] = int(os.path.getmtime(full_path)) data[common.KEY_TIME_IS_SUBSTITUTION] = True return data def __get_xxprobe_data__(full_path, conv_key_dict): def _ffprobe_command(full_path): return ['ffprobe', '-v', 'quiet', '-show_format', '-show_streams', full_path] def _avprobe_command(full_path): return ['avprobe', '-v', 'quiet', '-show_format', '-show_streams', full_path] try: xxprobe_text = subprocess.check_output(_avprobe_command(full_path)) except FileNotFoundError: try: xxprobe_text = subprocess.check_output(_ffprobe_command(full_path)) except FileNotFoundError: logger.warning('ffprobe and avprobe seem to be not installed') return {} # rv = {} for line in xxprobe_text.decode('utf-8').splitlines(): try: key, val = [snippet.strip() for snippet in line.split('=')] except ValueError: continue else: if key in conv_key_dict: tp, name = conv_key_dict[key] try: rv[name] = tp(val) except ValueError: logger.log(logging.WARNING if val else logger.INFO, 'Can\'t convert %s (%s) for %s', repr(val), name, name) return rv def __get_exif_data__(full_path): rv = {} im = Image.open(full_path) try: exif = dict(im._getexif().items()) except AttributeError: logger.debug('%s does not have any exif information', full_path) else: conv_key_dict = {} # IMAGE conv_key_dict[0x9003] = (__datetime_conv__, common.KEY_TIME) conv_key_dict[0x8822] = (__exposure_program_conv__, common.KEY_EXPOSURE_PROGRAM) conv_key_dict[0x829A] = (__num_denum_conv__, common.KEY_EXPOSURE_TIME) conv_key_dict[0x9209] = (__flash_conv__, common.KEY_FLASH) conv_key_dict[0x829D] = (__num_denum_conv__, common.KEY_APERTURE) conv_key_dict[0x920A] = (__num_denum_conv__, common.KEY_FOCAL_LENGTH) conv_key_dict[0x8825] = (__gps_conv__, common.KEY_GPS) conv_key_dict[0xA003] = (__int_conv__, common.KEY_HEIGHT) conv_key_dict[0x8827] = (__int_conv__, common.KEY_ISO) conv_key_dict[0x010F] = (str, __KEY_CAMERA_VENDOR__) conv_key_dict[0x0110] = (str, __KEY_CAMERA_MODEL__) conv_key_dict[0x0112] = (__int_conv__, common.KEY_ORIENTATION) conv_key_dict[0xA002] = (__int_conv__, common.KEY_WIDTH) for key in conv_key_dict: if key in exif: tp, name = conv_key_dict[key] raw_value = exif[key] logger.debug("Converting %s out of %s", name, repr(raw_value)) value = tp(raw_value) if value is not None: rv[name] = value return rv # TODO: Join datetime converter __datetime_conv__ and __vid_datetime_conv_ def __datetime_conv__(dt): format_string = "%Y:%m:%d %H:%M:%S" return int(time.mktime(time.strptime(dt, format_string))) def __vid_datetime_conv__(dt): try: dt = dt[:dt.index('.')] except ValueError: pass # time string seems to have no '.' dt = dt.replace('T', ' ').replace('/', '').replace('\\', '') if len(dt) == 16: dt += ':00' format_string = '%Y-%m-%d %H:%M:%S' return int(time.mktime(time.strptime(dt, format_string))) def __exposure_program_conv__(n): return { 0: 'Unidentified', 1: 'Manual', 2: 'Program Normal', 3: 'Aperture Priority', 4: 'Shutter Priority', 5: 'Program Creative', 6: 'Program Action', 7: 'Portrait Mode', 8: 'Landscape Mode' }.get(n, None) def __flash_conv__(n): return { 0: 'No', 1: 'Fired', 5: 'Fired (?)', # no return sensed 7: 'Fired (!)', # return sensed 9: 'Fill Fired', 13: 'Fill Fired (?)', 15: 'Fill Fired (!)', 16: 'Off', 24: 'Auto Off', 25: 'Auto Fired', 29: 'Auto Fired (?)', 31: 'Auto Fired (!)', 32: 'Not Available' }.get(n, None) def __int_conv__(value): try: return int(value) except ValueError: for c in ['.', '/', '-']: p = value.find(c) if p >= 0: value = value[:p] try: return int(value) except ValueError: return None def __num_denum_conv__(data): try: return float(data) except TypeError: num, denum = data return num / denum def __gps_conv__(data): def lat_lon_cal(lon_or_lat): lon_lat = 0. fac = 1. for data in lon_or_lat: try: lon_lat += float(data[0]) / float(data[1]) * fac except TypeError: lon_lat += data * fac except ZeroDivisionError: return 0. fac *= 1. / 60. if math.isnan(lon_lat): return 0. return lon_lat try: lon = lat_lon_cal(data[0x0004]) lat = lat_lon_cal(data[0x0002]) if lon != 0 or lat != 0: # do not use lon and lat equal 0, caused by motorola gps weakness return {'lon': lon, 'lat': lat} except KeyError: logger.warning('GPS data extraction failed for %s', repr(data)) def __ratio_conv__(ratio): ratio = ratio.replace('\\', '') num, denum = ratio.split(':') return float(num) / float(denum)