#!/usr/bin/env python3 """ Metadata Handler Module for iPod Nano Transfer Tool Handles metadata injection and album art embedding """ import os import sys import logging import struct import base64 from typing import Optional, Dict, Any from io import BytesIO from pathlib import Path import requests from PIL import Image from mutagen import File as MutagenFile from mutagen.mp4 import MP4, MP4Cover from mutagen.id3 import ID3, APIC, TIT2, TPE1, TALB, TRCK from mutagen.easyid3 import EasyID3 from youtube_downloader import TrackInfo from artwork.cache import cover_cache, xdg_cache_path # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class MetadataHandler: """Class to handle metadata injection and album art embedding""" def __init__(self, temp_dir: Optional[str] = None): """ Initialize the metadata handler Args: temp_dir: Directory for temporary files (optional) """ self.temp_dir = temp_dir or xdg_cache_path("thumbnails") os.makedirs(self.temp_dir, exist_ok=True) def download_thumbnail(self, thumbnail_url: str) -> Optional[str]: """ Download thumbnail image from URL Args: thumbnail_url: URL of the thumbnail image Returns: Path to downloaded thumbnail file, or None if download failed """ if not thumbnail_url: logger.warning("No thumbnail URL provided") return None try: logger.info(f"Downloading thumbnail: {thumbnail_url}") response = requests.get(thumbnail_url, timeout=10) response.raise_for_status() # Save thumbnail to temporary file thumbnail_path = os.path.join(self.temp_dir, f"thumbnail_{hash(thumbnail_url)}.jpg") # Process image with PIL to ensure it's a valid format img = Image.open(BytesIO(response.content)) img.save(thumbnail_path, "JPEG") logger.info(f"Thumbnail saved to: {thumbnail_path}") return thumbnail_path except (requests.RequestException, IOError) as e: logger.error(f"Failed to download thumbnail: {e}") return None def embed_metadata_m4a(self, file_path: str, track_info: TrackInfo, thumbnail_path: Optional[str] = None) -> bool: """ Embed metadata and album art in M4A/AAC file Args: file_path: Path to M4A/AAC file track_info: TrackInfo object with track metadata thumbnail_path: Path to thumbnail image (optional) Returns: True if successful, False otherwise """ logger.info(f"Embedding metadata in M4A file: {file_path}") try: audio = MP4(file_path) # Clear existing tags audio.clear() # Add metadata tags audio['\xa9nam'] = [track_info.title] # Title audio['\xa9ART'] = [track_info.artist] # Artist audio['\xa9alb'] = [track_info.album or track_info.artist] # Album # Add track number if available if track_info.track_number is not None: audio['trkn'] = [(track_info.track_number, 0)] # Add album art if available if thumbnail_path: with open(thumbnail_path, 'rb') as f: album_art_data = f.read() # Determine image format if album_art_data.startswith(b'\xff\xd8'): # JPEG cover = MP4Cover(album_art_data, imageformat=MP4Cover.FORMAT_JPEG) else: # PNG cover = MP4Cover(album_art_data, imageformat=MP4Cover.FORMAT_PNG) audio['covr'] = [cover] # Save changes audio.save() logger.info(f"Metadata embedded successfully: {file_path}") return True except Exception as e: logger.error(f"Failed to embed metadata in M4A file: {e}") return False def embed_metadata_mp3(self, file_path: str, track_info: TrackInfo, thumbnail_path: Optional[str] = None) -> bool: """ Embed metadata and album art in MP3 file Args: file_path: Path to MP3 file track_info: TrackInfo object with track metadata thumbnail_path: Path to thumbnail image (optional) Returns: True if successful, False otherwise """ logger.info(f"Embedding metadata in MP3 file: {file_path}") try: # First try with EasyID3 for basic tags try: audio = EasyID3(file_path) except: # If the file doesn't have an ID3 tag, add one audio = ID3() audio.save(file_path) audio = EasyID3(file_path) # Clear existing tags audio.clear() # Add metadata tags audio['title'] = track_info.title audio['artist'] = track_info.artist audio['album'] = track_info.album or track_info.artist # Add track number if available if track_info.track_number is not None: audio['tracknumber'] = str(track_info.track_number) # Save basic tags audio.save() # Now add album art using ID3 if thumbnail_path: audio = ID3(file_path) with open(thumbnail_path, 'rb') as f: album_art_data = f.read() # Add album art audio.add( APIC( encoding=3, # UTF-8 mime='image/jpeg', type=3, # Cover (front) desc='Cover', data=album_art_data ) ) # Save with album art audio.save() logger.info(f"Metadata embedded successfully: {file_path}") return True except Exception as e: logger.error(f"Failed to embed metadata in MP3 file: {e}") return False def process_file(self, file_path: str, track_info: TrackInfo) -> bool: """ Process audio file to embed metadata and album art Args: file_path: Path to audio file track_info: TrackInfo object with track metadata Returns: True if successful, False otherwise """ logger.info(f"Processing metadata for file: {file_path}") # Download thumbnail if available, or use cached cover from source thumbnail_path = None if track_info.thumbnail_url: thumbnail_path = self.download_thumbnail(track_info.thumbnail_url) elif track_info.cover_path and os.path.exists(track_info.cover_path): thumbnail_path = track_info.cover_path # Determine file type and process accordingly file_ext = Path(file_path).suffix.lower() if file_ext in ['.m4a', '.aac', '.mp4']: return self.embed_metadata_m4a(file_path, track_info, thumbnail_path) elif file_ext == '.mp3': return self.embed_metadata_mp3(file_path, track_info, thumbnail_path) else: logger.warning(f"Unsupported file format: {file_ext}") return False def extract_tags(self, file_path: str) -> Optional[TrackInfo]: """ Extract metadata from any supported audio file. Uses mutagen to read tags from FLAC, MP3, OGG, OPUS, M4A, WAV, APE, etc. Also extracts embedded cover art if present. Args: file_path: Path to audio file Returns: TrackInfo with extracted metadata, or None if file can't be read """ import struct import base64 logger.info(f"Extracting metadata from: {file_path}") try: audio = MutagenFile(file_path) if audio is None: logger.warning(f"Cannot read tags from: {file_path}") return self._parse_filename_trackinfo( os.path.splitext(os.path.basename(file_path))[0], file_path ) filename = os.path.basename(file_path) base_name = os.path.splitext(filename)[0] tags = audio.tags # Extract common fields - handle both vorbis (FLAC/OGG) and MP4 tags title = self._safe_get_tag(tags, 'title', '\xa9nam') artist = self._safe_get_tag(tags, 'artist', '\xa9ART') album = self._safe_get_tag(tags, 'album', '\xa9alb') album_artist = self._safe_get_tag(tags, 'albumartist', 'aART') genre = self._safe_get_tag(tags, 'genre', '\xa9gen') date_str = self._safe_get_tag(tags, 'date', '\xa9day') # Track number track_number = 0 track_count = 0 for key in ('tracknumber', 'TRK', 'trkn'): val = self._safe_get_tag_value(tags, key) if val is not None: if '/' in val: parts = val.split('/') track_number = int(parts[0]) if parts[0].isdigit() else 0 track_count = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 0 else: try: track_number = int(val) except ValueError: pass if track_number: break # M4A trkn is stored as list of tuples - check directly if not track_number and tags is not None: try: if 'trkn' in tags: trkn = tags['trkn'] if isinstance(trkn, list) and trkn: t = trkn[0] if isinstance(t, tuple) and len(t) >= 1: track_number = t[0] if len(t) >= 2: track_count = t[1] elif isinstance(trkn, tuple) and len(trkn) >= 1: track_number = trkn[0] if len(trkn) >= 2: track_count = trkn[1] except (ValueError, TypeError): pass # Disc number disc_number = 0 for key in ('discnumber', 'DISC', 'disk'): val = self._safe_get_tag_value(tags, key) if val is not None: if '/' in val: try: disc_number = int(val.split('/')[0]) except ValueError: pass else: try: disc_number = int(val) except ValueError: pass if disc_number: break # Duration duration = int(audio.info.length) if hasattr(audio, 'info') and audio.info.length else 0 # Extract cover art cover_data = self._extract_cover(audio, tags) thumbnail_path = None if cover_data: thumbnail_path = os.path.join(self.temp_dir, f"cover_{hash(file_path)}.jpg") self._save_cover_as_jpeg(cover_data, thumbnail_path) # Parse year from date year = 0 if date_str: try: year = int(str(date_str)[:4]) except (ValueError, TypeError): pass # If essential tags are missing, fall back to filename if not title: fallback = self._parse_filename_trackinfo(base_name) title = title or fallback.title artist = artist or fallback.artist # Cache cover art for the player UI cover_path = None if cover_data: cover_path = cover_cache.put(artist or "Unknown Artist", album or "", cover_data) return TrackInfo( video_id=f"local_{hash(file_path)}", title=title or base_name, artist=artist or "Unknown Artist", album=album or "", thumbnail_url="", duration=duration, track_number=track_number if track_number else None, genre=genre, download_path=file_path, cover_path=cover_path, ) except Exception as e: logger.error(f"Failed to extract tags from {file_path}: {e}") base_name = os.path.splitext(os.path.basename(file_path))[0] return self._parse_filename_trackinfo(base_name, file_path) def _safe_get_tag(self, tags, easy_key: str, mp4_key: str) -> Optional[str]: """Get a tag value safely, handling vorbis, ID3, and MP4 tag formats.""" if tags is None: return None # Try MP4 key first (for MP4/M4A files) try: if mp4_key in tags: val = tags[mp4_key] if isinstance(val, list) and val: return str(val[0]) return str(val) except ValueError: pass # vorbis dict raises ValueError for invalid keys # Try vorbis comment key (case-insensitive) try: easy_lower = easy_key.lower() for k in tags.keys(): if k.lower() == easy_lower: val = tags[k] if isinstance(val, list) and val: return str(val[0]) return str(val) except (ValueError, AttributeError): pass # Try ID3 frame mapping for common tags id3_map = { 'title': 'TIT2', 'artist': 'TPE1', 'album': 'TALB', 'albumartist': 'TPE2', 'genre': 'TCON', 'date': 'TDRC', 'comment': 'COMM', } frame_id = id3_map.get(easy_key.lower()) if frame_id and hasattr(tags, 'getall'): try: frames = tags.getall(frame_id) if frames: return str(frames[0]) except Exception: pass return None def _safe_get_tag_value(self, tags, key: str) -> Optional[str]: """Get raw tag value safely.""" if tags is None: return None # Direct check try: if key in tags: val = tags[key] if isinstance(val, list) and val: return str(val[0]) return str(val) except (ValueError, KeyError, TypeError): pass # Case-insensitive for vorbis try: key_lower = key.lower() for k in tags.keys(): if k.lower() == key_lower: val = tags[k] if isinstance(val, list) and val: return str(val[0]) return str(val) except (ValueError, AttributeError): pass # ID3 frame mapping id3_map = { 'tracknumber': 'TRCK', 'discnumber': 'TPOS', } frame_id = id3_map.get(key.lower()) if frame_id and hasattr(tags, 'getall'): try: frames = tags.getall(frame_id) if frames: return str(frames[0]) except Exception: pass return None def _extract_cover(self, audio, tags) -> Optional[bytes]: """Extract cover art data from audio file tags.""" # MP3 (ID3v2) - APIC frames if hasattr(tags, 'getall'): try: for frame in tags.getall('APIC'): return frame.data except Exception: pass # FLAC / Vorbis comments - METADATA_BLOCK_PICTURE try: if 'metadata_block_picture' in tags: pic_data = tags['metadata_block_picture'] if isinstance(pic_data, list): pic_data = pic_data[0] pic_data = str(pic_data) decoded = base64.b64decode(pic_data) # Parse the picture block: type(4), mime_len(4), mime, desc_len(4), desc, w(4), h(4), d(4), colors(4), data_len(4), data pos = 0 pos += 4 # picture type mime_len = struct.unpack('>I', decoded[pos:pos+4])[0] pos += 4 + mime_len # mime string desc_len = struct.unpack('>I', decoded[pos:pos+4])[0] pos += 4 + desc_len # description pos += 20 # width(4) + height(4) + depth(4) + colors(4) + data_len(4) return decoded[pos:] except Exception: pass # MP4/M4A - covr try: if 'covr' in tags: cover_list = tags['covr'] if isinstance(cover_list, list) and cover_list: return bytes(cover_list[0]) except Exception: pass return None def _save_cover_as_jpeg(self, cover_data: bytes, output_path: str) -> None: """Save cover art data as JPEG.""" try: from PIL import Image from io import BytesIO logger.debug(f"Saving cover art: {len(cover_data)} bytes") stream = BytesIO(cover_data) img = Image.open(stream) img = img.convert('RGB') max_size = 600 if img.width > max_size or img.height > max_size: img.thumbnail((max_size, max_size), Image.Resampling.LANCZOS) img.save(output_path, 'JPEG', quality=85) logger.debug(f"Cover art saved to: {output_path}") except Exception as e: logger.warning(f"Could not save cover art: {e}") logger.debug(f"Cover data length: {len(cover_data)}, first bytes: {cover_data[:4].hex()}") def _parse_filename_trackinfo(self, filename: str, file_path: Optional[str] = None) -> TrackInfo: """Create TrackInfo by parsing filename as fallback.""" title = filename artist = "Unknown" if " - " in filename: artist, title = filename.split(" - ", 1) return TrackInfo( video_id=f"local_{hash(file_path or filename)}", title=title, artist=artist, album="", thumbnail_url="", duration=0, download_path=file_path, ) def enhance_metadata_with_musicbrainz(self, track_info: TrackInfo) -> TrackInfo: """ Enhance track metadata using MusicBrainz API Args: track_info: TrackInfo object with track metadata Returns: Enhanced TrackInfo object """ # This is a placeholder for future implementation # MusicBrainz integration would go here return track_info def update_tags(self, file_path: str, tags: Dict[str, Any], new_cover_path: Optional[str] = None) -> bool: """ Update metadata tags in an audio file. Args: file_path: Path to audio file tags: Dict of tag_name -> value Supported keys: title, artist, album, album_artist, genre, date, track_number, track_total, disc_number, disc_total, comment new_cover_path: Path to new cover image, None to keep existing, empty string '' to remove cover art Returns: True if successful, False otherwise """ from mutagen.flac import FLAC, Picture as FLACPicture from mutagen.oggvorbis import OggVorbis from mutagen.oggopus import OggOpus file_ext = Path(file_path).suffix.lower() try: if file_ext in ('.m4a', '.aac', '.mp4'): return self._update_mp4_tags(file_path, tags, new_cover_path) elif file_ext == '.mp3': return self._update_mp3_tags(file_path, tags, new_cover_path) elif file_ext == '.flac': return self._update_vorbis_tags(file_path, tags, new_cover_path, use_picture_block=True) elif file_ext in ('.ogg', '.opus'): return self._update_vorbis_tags(file_path, tags, new_cover_path, use_picture_block=False) else: logger.warning(f"Unsupported format for tag update: {file_ext}") return False except Exception as e: logger.error(f"Failed to update tags in {file_path}: {e}") return False def _load_cover_data(self, cover_path: str) -> Optional[bytes]: """Load cover image from path, convert to JPEG if needed.""" try: img = Image.open(cover_path) if img.mode != 'RGB': img = img.convert('RGB') buf = BytesIO() img.save(buf, 'JPEG', quality=90) return buf.getvalue() except Exception as e: logger.error(f"Failed to load cover image {cover_path}: {e}") return None def _update_mp4_tags(self, file_path: str, tags: Dict[str, Any], new_cover_path: Optional[str]) -> bool: audio = MP4(file_path) text_fields = { 'title': '\xa9nam', 'artist': '\xa9ART', 'album': '\xa9alb', 'album_artist': 'aART', 'genre': '\xa9gen', 'date': '\xa9day', 'comment': '\xa9cmt', } for key, mp4_key in text_fields.items(): if key in tags: val = tags[key] if val: audio[mp4_key] = [str(val)] elif mp4_key in audio: del audio[mp4_key] if 'track_number' in tags or 'track_total' in tags: trkn_val = list(audio.get('trkn', [(0, 0)])[0]) if 'trkn' in audio else [0, 0] if 'track_number' in tags: trkn_val[0] = tags['track_number'] or 0 if 'track_total' in tags: trkn_val[1] = tags['track_total'] or 0 if trkn_val[0] or trkn_val[1]: audio['trkn'] = [tuple(trkn_val)] elif 'trkn' in audio: del audio['trkn'] if 'disc_number' in tags or 'disc_total' in tags: disk_val = list(audio.get('disk', [(0, 0)])[0]) if 'disk' in audio else [0, 0] if 'disc_number' in tags: disk_val[0] = tags['disc_number'] or 0 if 'disc_total' in tags: disk_val[1] = tags['disc_total'] or 0 if disk_val[0] or disk_val[1]: audio['disk'] = [tuple(disk_val)] elif 'disk' in audio: del audio['disk'] if new_cover_path == '': if 'covr' in audio: del audio['covr'] elif new_cover_path is not None: data = self._load_cover_data(new_cover_path) if data: cover = MP4Cover(data, imageformat=MP4Cover.FORMAT_JPEG) audio['covr'] = [cover] audio.save() return True def _update_mp3_tags(self, file_path: str, tags: Dict[str, Any], new_cover_path: Optional[str]) -> bool: try: audio = EasyID3(file_path) except Exception: audio = ID3() audio.save(file_path) audio = EasyID3(file_path) easy_fields = { 'title': 'title', 'artist': 'artist', 'album': 'album', 'album_artist': 'albumartist', 'genre': 'genre', 'date': 'date', } for key, easy_key in easy_fields.items(): if key in tags: val = tags[key] if val: audio[easy_key] = str(val) elif easy_key in audio: del audio[easy_key] if 'comment' in tags: try: audio['comment'] = str(tags['comment']) if tags['comment'] else '' except Exception: pass if 'track_number' in tags or 'track_total' in tags: current = audio.get('tracknumber', [''])[0] if 'tracknumber' in audio else '' tn, tt = self._parse_slash_value(current) if 'track_number' in tags: tn = tags['track_number'] or 0 if 'track_total' in tags: tt = tags['track_total'] or 0 if tn or tt: audio['tracknumber'] = [f"{tn}/{tt}" if tt else str(tn)] elif 'tracknumber' in audio: del audio['tracknumber'] if 'disc_number' in tags or 'disc_total' in tags: current = audio.get('discnumber', [''])[0] if 'discnumber' in audio else '' dn, dt = self._parse_slash_value(current) if 'disc_number' in tags: dn = tags['disc_number'] or 0 if 'disc_total' in tags: dt = tags['disc_total'] or 0 if dn or dt: audio['discnumber'] = [f"{dn}/{dt}" if dt else str(dn)] elif 'discnumber' in audio: del audio['discnumber'] audio.save() if new_cover_path is not None: audio = ID3(file_path) if new_cover_path == '': audio.delall('APIC') else: data = self._load_cover_data(new_cover_path) if data: audio.delall('APIC') audio.add(APIC(encoding=3, mime='image/jpeg', type=3, desc='Cover', data=data)) audio.save() return True def _parse_slash_value(self, current: str) -> Tuple[int, int]: """Parse '5/12' format into (5, 12).""" try: parts = str(current).split('/') first = int(parts[0]) if parts[0].strip().isdigit() else 0 second = int(parts[1]) if len(parts) > 1 and parts[1].strip().isdigit() else 0 return first, second except Exception: return 0, 0 def _update_vorbis_tags(self, file_path: str, tags: Dict[str, Any], new_cover_path: Optional[str], use_picture_block: bool = False) -> bool: from mutagen.flac import FLAC, Picture as FLACPicture from mutagen.oggvorbis import OggVorbis from mutagen.oggopus import OggOpus ext = Path(file_path).suffix.lower() if ext == '.flac': audio = FLAC(file_path) elif ext == '.opus': audio = OggOpus(file_path) else: audio = OggVorbis(file_path) if audio.tags is None: audio.add_tags() vorbis_fields = { 'title': 'TITLE', 'artist': 'ARTIST', 'album': 'ALBUM', 'album_artist': 'ALBUMARTIST', 'genre': 'GENRE', 'date': 'DATE', 'comment': 'COMMENT', } for key, vorbis_key in vorbis_fields.items(): if key in tags: val = tags[key] if val: audio.tags[vorbis_key] = str(val) elif vorbis_key in audio.tags: try: del audio.tags[vorbis_key] except (KeyError, TypeError): pass if 'track_number' in tags or 'track_total' in tags: current = audio.tags.get('TRACKNUMBER', [''])[0] if 'TRACKNUMBER' in audio.tags else '' tn, tt = self._parse_slash_value(current) if 'track_number' in tags: tn = tags['track_number'] or 0 if 'track_total' in tags: tt = tags['track_total'] or 0 if tn or tt: audio.tags['TRACKNUMBER'] = f"{tn}/{tt}" if tt else str(tn) elif 'TRACKNUMBER' in audio.tags: try: del audio.tags['TRACKNUMBER'] except (KeyError, TypeError): pass if 'disc_number' in tags or 'disc_total' in tags: current = audio.tags.get('DISCNUMBER', [''])[0] if 'DISCNUMBER' in audio.tags else '' dn, dt = self._parse_slash_value(current) if 'disc_number' in tags: dn = tags['disc_number'] or 0 if 'disc_total' in tags: dt = tags['disc_total'] or 0 if dn or dt: audio.tags['DISCNUMBER'] = f"{dn}/{dt}" if dt else str(dn) elif 'DISCNUMBER' in audio.tags: try: del audio.tags['DISCNUMBER'] except (KeyError, TypeError): pass if new_cover_path is not None: self._clear_cover_vorbis(audio) if new_cover_path != '': data = self._load_cover_data(new_cover_path) if data: self._embed_cover_vorbis(audio, data) elif use_picture_block: pass else: pass audio.save() return True def _clear_cover_vorbis(self, audio) -> None: """Remove all cover art from Vorbis/FLAC file.""" if audio.tags is None: return if 'METADATA_BLOCK_PICTURE' in audio.tags: del audio.tags['METADATA_BLOCK_PICTURE'] if 'COVERART' in audio.tags: del audio.tags['COVERART'] if hasattr(audio, 'clear_pictures'): audio.clear_pictures() def _embed_cover_vorbis(self, audio, data: bytes) -> None: """Embed JPEG cover art into Vorbis/FLAC file.""" from mutagen.flac import Picture as FLACPicture pic = FLACPicture() pic.type = 3 # Cover (front) pic.mime = 'image/jpeg' pic.desc = 'Cover' pic.width = 0 pic.height = 0 pic.depth = 24 pic.data = data pic_data = pic.write() if hasattr(audio, 'add_picture'): audio.clear_pictures() audio.add_picture(pic) else: encoded = base64.b64encode(pic_data).decode('ascii') audio.tags['METADATA_BLOCK_PICTURE'] = encoded if __name__ == "__main__": # Example usage handler = MetadataHandler() # Test metadata embedding # handler.process_file( # "test.m4a", # TrackInfo( # video_id="dQw4w9WgXcQ", # title="Never Gonna Give You Up", # artist="Rick Astley", # album="Whenever You Need Somebody", # thumbnail_url="https://i.ytimg.com/vi/dQw4w9WgXcQ/maxresdefault.jpg", # duration=213, # track_number=1 # ) # )