Add metadata extraction from local audio files for proper tag preservation
- Add extract_tags() to metadata_handler.py: reads tags from FLAC, MP3, OGG, M4A - Support ID3v2 (MP3), Vorbis comments (FLAC/OGG), MP4 tags (M4A) - Extract cover art from METADATA_BLOCK_PICTURE (FLAC), APIC (MP3), covr (M4A) - Update _run_convert to use extract_tags instead of guessing from filename - Fix tag reading in ipod_nano7_db.get_audio_info() using EasyID3 for MP3
This commit is contained in:
Maksim Totmin
parent
761f896ea8
commit
1b4c092a04
@ -15,9 +15,9 @@ import shutil
|
||||
from typing import Optional, Dict, Any
|
||||
|
||||
from mutagen import File as MutagenFile
|
||||
from mutagen.easyid3 import EasyID3
|
||||
from mutagen.mp4 import MP4
|
||||
from mutagen.mp3 import MP3
|
||||
from mutagen.id3 import ID3
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -117,51 +117,118 @@ class Nano7Database:
|
||||
else:
|
||||
info["audio_format"] = 1
|
||||
|
||||
# Extract tags
|
||||
for tag_name, file_key in [
|
||||
("title", "title"),
|
||||
("artist", "artist"),
|
||||
("album", "album"),
|
||||
("albumartist", "album_artist"),
|
||||
("composer", "composer"),
|
||||
("genre", "genre"),
|
||||
("comment", "comment"),
|
||||
]:
|
||||
if tag_name in audio:
|
||||
val = audio[tag_name]
|
||||
if isinstance(val, list):
|
||||
val = val[0]
|
||||
if isinstance(val, str):
|
||||
info[file_key] = val
|
||||
# Extract tags - handle different formats
|
||||
ext = os.path.splitext(filepath)[1].lower()
|
||||
|
||||
# Track/disc numbers
|
||||
for tag_name, file_key in [
|
||||
("tracknumber", "track_number"),
|
||||
("discnumber", "disc_number"),
|
||||
]:
|
||||
if tag_name in audio:
|
||||
val = audio[tag_name]
|
||||
if isinstance(val, list):
|
||||
val = val[0]
|
||||
if isinstance(val, str):
|
||||
parts = val.split("/")
|
||||
info[file_key] = int(parts[0]) if parts[0].isdigit() else 0
|
||||
if len(parts) > 1:
|
||||
info[file_key + "_count"] = (
|
||||
int(parts[1]) if parts[1].isdigit() else 0
|
||||
)
|
||||
elif isinstance(val, int):
|
||||
info[file_key] = val
|
||||
# Use EasyID3 for MP3 files (supports easy key access)
|
||||
if ext == ".mp3":
|
||||
try:
|
||||
easy = EasyID3(filepath)
|
||||
for tag_name, file_key in [
|
||||
("title", "title"),
|
||||
("artist", "artist"),
|
||||
("album", "album"),
|
||||
("albumartist", "albumartist"),
|
||||
("composer", "composer"),
|
||||
("genre", "genre"),
|
||||
("comment", "comment"),
|
||||
]:
|
||||
if tag_name in easy:
|
||||
val = easy[tag_name]
|
||||
if isinstance(val, list) and val:
|
||||
info[file_key] = str(val[0])
|
||||
|
||||
# Year
|
||||
if "date" in audio:
|
||||
val = audio["date"]
|
||||
if isinstance(val, list):
|
||||
val = val[0]
|
||||
if isinstance(val, str) and len(val) >= 4:
|
||||
info["year"] = int(val[:4])
|
||||
elif isinstance(val, int):
|
||||
info["year"] = val
|
||||
# Track/disc numbers
|
||||
for tag_name, file_key in [
|
||||
("tracknumber", "track_number"),
|
||||
("discnumber", "disc_number"),
|
||||
]:
|
||||
if tag_name in easy:
|
||||
val = easy[tag_name]
|
||||
if isinstance(val, list) and val:
|
||||
val = str(val[0])
|
||||
if '/' in val:
|
||||
parts = val.split('/')
|
||||
info[file_key] = int(parts[0]) if parts[0].isdigit() else 0
|
||||
elif val.isdigit():
|
||||
info[file_key] = int(val)
|
||||
|
||||
# Date/year
|
||||
if "date" in easy:
|
||||
val = easy["date"]
|
||||
if isinstance(val, list) and val:
|
||||
date_str = str(val[0])
|
||||
if len(date_str) >= 4:
|
||||
info["year"] = int(date_str[:4])
|
||||
except Exception as e:
|
||||
logger.debug(f"EasyID3 failed for {filepath}: {e}")
|
||||
|
||||
# MP4/M4A uses MutagenFile directly
|
||||
elif ext in (".m4a", ".aac", ".mp4"):
|
||||
if audio and audio.tags:
|
||||
tags = audio.tags
|
||||
for tag_name, mp4_key, file_key in [
|
||||
("title", "\xa9nam", "title"),
|
||||
("artist", "\xa9ART", "artist"),
|
||||
("album", "\xa9alb", "album"),
|
||||
]:
|
||||
if mp4_key in tags:
|
||||
val = tags[mp4_key]
|
||||
if isinstance(val, list) and val:
|
||||
info[file_key] = str(val[0])
|
||||
if "trkn" in tags:
|
||||
trkn = tags["trkn"]
|
||||
if isinstance(trkn, list) and trkn:
|
||||
t = trkn[0]
|
||||
if isinstance(t, tuple) and len(t) >= 1:
|
||||
info["track_number"] = t[0]
|
||||
else:
|
||||
# FLAC/OGG/Vorbis - use MutagenFile with tag access
|
||||
if audio and audio.tags:
|
||||
tags = audio.tags
|
||||
for tag_name, file_key in [
|
||||
("title", "title"),
|
||||
("artist", "artist"),
|
||||
("album", "album"),
|
||||
("albumartist", "album_artist"),
|
||||
("composer", "composer"),
|
||||
("genre", "genre"),
|
||||
("comment", "comment"),
|
||||
]:
|
||||
try:
|
||||
if tag_name in tags:
|
||||
val = tags[tag_name]
|
||||
if isinstance(val, list) and val:
|
||||
info[file_key] = str(val[0])
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
for tag_name, file_key in [
|
||||
("tracknumber", "track_number"),
|
||||
("discnumber", "disc_number"),
|
||||
]:
|
||||
try:
|
||||
if tag_name in tags:
|
||||
val = tags[tag_name]
|
||||
if isinstance(val, list) and val:
|
||||
val = str(val[0])
|
||||
if '/' in val:
|
||||
parts = val.split('/')
|
||||
info[file_key] = int(parts[0]) if parts[0].isdigit() else 0
|
||||
elif val.isdigit():
|
||||
info[file_key] = int(val)
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
try:
|
||||
if "date" in tags:
|
||||
val = tags["date"]
|
||||
if isinstance(val, list) and val:
|
||||
date_str = str(val[0])
|
||||
if len(date_str) >= 4:
|
||||
info["year"] = int(date_str[:4])
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
return info
|
||||
|
||||
|
||||
27
src/main.py
27
src/main.py
@ -108,27 +108,16 @@ class WorkerThread(QThread):
|
||||
# Filter out tracks without download paths
|
||||
valid_tracks = [track for track in tracks if hasattr(track, 'download_path') and track.download_path and os.path.exists(track.download_path)]
|
||||
|
||||
# Process local files into TrackInfo objects
|
||||
# Process local files: extract real metadata from tags
|
||||
metadata_handler = MetadataHandler()
|
||||
for f in local_files:
|
||||
if os.path.exists(f):
|
||||
filename = os.path.basename(f)
|
||||
title = os.path.splitext(filename)[0]
|
||||
# Try to parse "Artist - Title" format
|
||||
if " - " in title:
|
||||
artist, track_title = title.split(" - ", 1)
|
||||
else:
|
||||
artist = "Unknown"
|
||||
track_title = title
|
||||
track = TrackInfo(
|
||||
video_id=f"local_{hash(f)}",
|
||||
title=track_title,
|
||||
artist=artist,
|
||||
album="",
|
||||
thumbnail_url="",
|
||||
duration=0,
|
||||
download_path=f
|
||||
)
|
||||
valid_tracks.append(track)
|
||||
# Try to extract metadata from file tags first
|
||||
track = metadata_handler.extract_tags(f)
|
||||
if track:
|
||||
# Ensure download_path is set
|
||||
track.download_path = f
|
||||
valid_tracks.append(track)
|
||||
|
||||
if not valid_tracks:
|
||||
self.finished_signal.emit(False, "No valid tracks to convert", None)
|
||||
|
||||
@ -5,14 +5,18 @@ 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
|
||||
@ -227,6 +231,308 @@ class MetadataHandler:
|
||||
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
|
||||
|
||||
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,
|
||||
download_path=file_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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user