from __future__ import annotations import sys import tempfile import time from pathlib import Path from typing import Optional, List try: from PIL import Image HAS_PIL = True except ImportError: HAS_PIL = False try: import requests HAS_REQUESTS = True except ImportError: HAS_REQUESTS = False try: import musicbrainzngs as mb HAS_MB = True except ImportError: HAS_MB = False try: from mutagen.id3 import ID3, APIC, error as ID3Error from mutagen.flac import FLAC, Picture from mutagen.mp4 import MP4, MP4Cover HAS_MUTAGEN = True except ImportError: HAS_MUTAGEN = False _MIN_COVER_SIZE = 200 # pixels def _image_ok(path: Path) -> bool: if not HAS_PIL: return path.stat().st_size > 5000 try: with Image.open(path) as img: w, h = img.size return w >= _MIN_COVER_SIZE and h >= _MIN_COVER_SIZE except Exception: return False def find_local_cover(image_files: List[Path]) -> Optional[Path]: priority = ("folder", "front", "cover", "album") # Sort by priority keyword, then size descending def key(p: Path): name = p.name.lower() score = next((i for i, kw in enumerate(priority) if kw in name), len(priority)) size = p.stat().st_size if p.exists() else 0 return (score, -size) for p in sorted(image_files, key=key): if _image_ok(p): return p return None def _mb_cover_url(release_mbid: str) -> Optional[str]: url = f"https://coverartarchive.org/release/{release_mbid}/front" if not HAS_REQUESTS: return None try: r = requests.head(url, timeout=5, allow_redirects=True) if r.status_code == 200: return url except Exception: pass return None def download_cover(release_mbid: Optional[str], dest_dir: Path) -> Optional[Path]: if not release_mbid or not HAS_REQUESTS: return None url = _mb_cover_url(release_mbid) if not url: return None dest = dest_dir / "folder.jpg" try: r = requests.get(url, timeout=15) if r.status_code != 200: return None ct = r.headers.get("content-type", "") if "png" in ct and HAS_PIL: # PNG → JPEG konvertieren import io with Image.open(io.BytesIO(r.content)) as img: buf = io.BytesIO() img.convert("RGB").save(buf, format="JPEG", quality=92) dest.write_bytes(buf.getvalue()) else: dest.write_bytes(r.content) if _image_ok(dest): return dest dest.unlink(missing_ok=True) except Exception as e: print(f" ⚠️ Cover-Download-Fehler: {e}", file=sys.stderr) dest.unlink(missing_ok=True) return None def _load_cover_data(cover_path: Path) -> tuple[bytes, str]: """ Liest Cover-Bilddaten und gibt (bytes, mime_type) zurück. WebP wird zu JPEG konvertiert wenn PIL verfügbar (bessere Player-Kompatibilität). """ suffix = cover_path.suffix.lower() if suffix in (".jpg", ".jpeg"): return cover_path.read_bytes(), "image/jpeg" if suffix == ".webp" and HAS_PIL: try: with Image.open(cover_path) as img: img = img.convert("RGB") buf = tempfile.SpooledTemporaryFile(max_size=10 * 1024 * 1024) img.save(buf, format="JPEG", quality=90) buf.seek(0) return buf.read(), "image/jpeg" except Exception as e: print(f" ⚠️ WebP→JPEG-Konvertierung fehlgeschlagen ({cover_path.name}): {e}", file=sys.stderr) if suffix == ".webp": return cover_path.read_bytes(), "image/webp" if suffix == ".png": return cover_path.read_bytes(), "image/png" # Fallback: raw bytes, JPEG assumed return cover_path.read_bytes(), "image/jpeg" def embed_cover(audio_path: Path, cover_path: Path) -> bool: if not HAS_MUTAGEN: return False try: img_data, mime = _load_cover_data(cover_path) ext = audio_path.suffix.lower() if ext == ".mp3": try: tags = ID3(str(audio_path)) except ID3Error: tags = ID3() tags.delall("APIC") tags.add(APIC(encoding=3, mime=mime, type=3, desc="Cover", data=img_data)) tags.save(str(audio_path), v2_version=4) return True elif ext == ".flac": audio = FLAC(str(audio_path)) audio.clear_pictures() pic = Picture() pic.type = 3 pic.mime = mime pic.desc = "Cover" pic.data = img_data audio.add_picture(pic) audio.save() return True elif ext == ".m4a": audio = MP4(str(audio_path)) fmt = MP4Cover.FORMAT_JPEG if mime == "image/jpeg" else MP4Cover.FORMAT_PNG # WebP wurde bereits zu JPEG konvertiert, mime ist dann "image/jpeg" audio.tags["covr"] = [MP4Cover(img_data, imageformat=fmt)] audio.save() return True else: # Generic mutagen fallback from mutagen import File as MutagenFile audio = MutagenFile(str(audio_path), easy=False) if audio is not None: if audio.tags is None: audio.add_tags() if hasattr(audio.tags, "add"): audio.tags.add( APIC(encoding=3, mime=mime, type=3, desc="Cover", data=img_data) ) audio.save() return True except Exception as e: print(f" ⚠️ Cover-Einbettungsfehler {audio_path.name}: {e}", file=sys.stderr) return False def resolve_cover( image_files: List[Path], release_mbid: Optional[str], album_dir: Path, ) -> tuple[Optional[Path], Optional[str]]: """Returns (cover_path, source_label).""" local = find_local_cover(image_files) if local: return local, "local" if release_mbid: downloaded = download_cover(release_mbid, album_dir) if downloaded: return downloaded, "musicbrainz" return None, None