489d99c068
- New src/artwork/ module with iPod Nano 7G cover art support - codecs: RGB565/RGB555/UYVY/JPEG encode/decode via numpy - presets: 50+ iPod artwork format definitions - chunks: ArtworkDB binary parser/writer (MHFD→MHSD→MHLI→MHII→MHNI) - writer: ithmb file generation + ArtworkDB assembly - extractor: cover extraction from audio files (APIC/covr/FLAC pictures) - cache: local JPEG cache for player UI - iTunes-style playback bar with 60x60 cover art display - Scan library now caches cover art by artist+album - iPod Nano7 transfer writes artwork to ArtworkDB + ithmb - FFmpeg conversion preserves embedded cover art (-map 0:t?) - One-time reembed_covers.py script in /tmp - MetadataHandler process_file() fallback to cover_path - Bug fixes: missing read_existing_artwork import, signed PID in Q format
578 lines
20 KiB
Python
578 lines
20 KiB
Python
#!/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 tempfile
|
|
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
|
|
|
|
# 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 tempfile.gettempdir()
|
|
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
|
|
|
|
|
|
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
|
|
# )
|
|
# )
|