#!/usr/bin/env python3 """ iPod Nano 7 Database Manager — iOpenPod-powered edition. Uses iOpenPod's battle-tested SQLiteDB_Writer and iTunesDB_Writer to generate correct databases for Nano 6G/7G firmware. The firmware on Nano 6G/7G reads SQLite databases exclusively, with cross-reference validation against the binary iTunesDB. Both must be written with the same db_pid. iOpenPod handles all of this correctly. Public API preserved for main.py compatibility: Nano7Database(mount_point) — constructor add_track(filepath) — copy + sync get_all_tracks() — scan iPod filesystem get_track_count() — count audio files on iPod delete_track(pid) — delete by pid (db_track_id) remove_duplicates() — find + remove duplicate tracks get_orphaned_files() — files not in DB delete_orphaned_files(paths) — remove orphan files """ import importlib import importlib.util import os import sys import hashlib import logging import random import shutil import sqlite3 import struct import subprocess import time from typing import Any, Optional from mutagen import File as MutagenFile from mutagen.easyid3 import EasyID3 from artwork import prepare_artwork_for_track, ArtworkEntry, write_artworkdb def _find_iop_root() -> str: """Find iOpenPod's installed location. Tries in order: 1. pip-installed package (via importlib.util.find_spec) 2. /tmp/iOpenPod (backward compat / development) """ for pkg in ("iTunesDB_Writer", "SQLiteDB_Writer", "ipod_device"): spec = importlib.util.find_spec(pkg) if spec and spec.origin: parent = os.path.dirname(spec.origin) if os.path.basename(parent) == pkg: return os.path.dirname(parent) fallbacks = [ os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "iOpenPod"), "/tmp/iOpenPod", os.path.join(os.path.expanduser("~"), "iOpenPod"), ] for path in fallbacks: if os.path.isdir(path): return path raise ImportError( "iOpenPod not found. Install with: pip install iopenpod" ) _IOP_ROOT = _find_iop_root() _IOP_CACHE: dict[str, Any] = {} _IOP_READY = False def _import_iop(module_name: str): """Import a module from iOpenPod, evicting our local ipod_device once. main.py imports our local ``ipod_device`` early, caching it in ``sys.modules``. On the FIRST call here we permanently evict the local module so iOpenPod's ``ipod_device`` takes its place in ``sys.modules``. Our local classes (IPodDevice) were already bound by ``from ipod_device import IPodDevice`` in main.py before the eviction, so they stay alive through their own references. Subsequent calls are served from a simple module cache. """ global _IOP_READY if module_name in _IOP_CACHE: return _IOP_CACHE[module_name] if not _IOP_READY: sys.modules.pop("ipod_device", None) _IOP_READY = True logger.debug("iOpenPod boot: evicted local ipod_device") src_paths = [p for p in sys.path if p.endswith('/src') or p == 'src'] for p in src_paths: sys.path.remove(p) had_iop = _IOP_ROOT in sys.path if not had_iop: sys.path.insert(0, _IOP_ROOT) importlib.invalidate_caches() try: mod = importlib.import_module(module_name) finally: for p in src_paths: if p not in sys.path: sys.path.insert(0, p) if not had_iop: try: sys.path.remove(_IOP_ROOT) except ValueError: pass _IOP_CACHE[module_name] = mod return mod logger = logging.getLogger(__name__) # ── constants ──────────────────────────────────────────────────────── FILETYPE_CODES = { "mp3": 0x4D503320, "m4a": 0x4D344120, "m4p": 0x4D345020, "mp4": 0x4D503420, "aac": 0x41414320, } MAC_EPOCH_OFFSET = 2082844800 MASTER_CONTAINER_PID = 203092939621887772 MUSIC_CONTAINER_PID = 4872745547565090077 _REMOVE_ARTWORK = object() """Sentinel: used in artwork_overrides to remove artwork from a track.""" # ── Helpers ────────────────────────────────────────────────────────── def _to_mac_epoch(ts: int) -> int: return ts + MAC_EPOCH_OFFSET if ts > 0 else 0 def content_hash(filepath: str, max_bytes: int = 65536) -> str: """SHA1 of first max_bytes of audio data (fast content fingerprint).""" h = hashlib.sha1() try: with open(filepath, "rb") as f: h.update(f.read(max_bytes)) except OSError: return "" return h.hexdigest() def _read_audio_tags(filepath: str) -> dict[str, Any]: """Extract title, artist, album, duration, etc. from an audio file.""" info: dict[str, Any] = {} ext = os.path.splitext(filepath)[1].lower() try: audio = MutagenFile(filepath) if audio is None: return info if hasattr(audio, "info") and audio.info and audio.info.length: info["duration_ms"] = int(audio.info.length * 1000) else: info["duration_ms"] = 0 if hasattr(audio.info, "bitrate"): info["bitrate"] = audio.info.bitrate else: info["bitrate"] = 0 if hasattr(audio.info, "sample_rate"): info["sample_rate"] = audio.info.sample_rate or 44100 else: info["sample_rate"] = 44100 if hasattr(audio.info, "channels"): info["channels"] = audio.info.channels or 2 else: info["channels"] = 2 if ext == ".mp3": try: easy = EasyID3(filepath) for tag_k, dest in [("title", "title"), ("artist", "artist"), ("album", "album"), ("albumartist", "album_artist"), ("genre", "genre"), ("composer", "composer"), ("comment", "comment")]: if tag_k in easy: val = easy[tag_k] if isinstance(val, list) and val: info[dest] = str(val[0]) for tag_k, dest in [("tracknumber", "track_number"), ("discnumber", "disc_number")]: if tag_k in easy: val = easy[tag_k] if isinstance(val, list) and val: s = str(val[0]) if "/" in s: parts = s.split("/") info[dest] = int(parts[0]) if parts[0].isdigit() else 0 elif s.isdigit(): info[dest] = int(s) if "date" in easy: val = easy["date"] if isinstance(val, list) and val: ds = str(val[0]) if len(ds) >= 4: info["year"] = int(ds[:4]) except Exception: pass elif ext in (".m4a", ".aac", ".mp4", ".m4p"): if audio and audio.tags: t = audio.tags for mp4_k, dest in [("\xa9nam", "title"), ("\xa9ART", "artist"), ("\xa9alb", "album"), ("aART", "album_artist"), ("\xa9gen", "genre"), ("\xa9wrt", "composer"), ("\xa9day", "date")]: if mp4_k in t: v = t[mp4_k] if isinstance(v, list) and v: if dest == "date": ds = str(v[0]) if len(ds) >= 4: info["year"] = int(ds[:4]) else: info[dest] = str(v[0]) if "trkn" in t: tr = t["trkn"] if isinstance(tr, list) and tr and isinstance(tr[0], tuple): info["track_number"] = tr[0][0] else: if audio and audio.tags: for tag_k, dest in [("title", "title"), ("artist", "artist"), ("album", "album"), ("albumartist", "album_artist"), ("genre", "genre"), ("composer", "composer")]: try: if tag_k in audio.tags: v = audio.tags[tag_k] info[dest] = str(v[0]) if isinstance(v, list) and v else str(v) except (ValueError, TypeError): pass except Exception: pass return info def _probe_gapless(filepath: str, sample_rate: int, duration_ms: int) -> tuple[int, int, int]: """Probe AAC/MP3 file for encoder delay and padding samples. Returns (pregap, postgap, sample_count). """ pregap = 0 postgap = 0 try: result = subprocess.run( ["ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", "stream_tags=encoder_delay,padding", "-of", "csv=p=0", filepath], capture_output=True, text=True, timeout=10, ) if result.returncode == 0 and result.stdout.strip(): parts = result.stdout.strip().split(",") pregap = int(parts[0]) if parts[0] else 0 postgap = int(parts[1]) if len(parts) > 1 and parts[1] else 0 except Exception: pass # Fallback: typical AAC-LC encoding delay (1 frame = 1024 samples at 44100 Hz) if pregap == 0 and sample_rate == 44100: ext = os.path.splitext(filepath)[1].lower() if ext in (".m4a", ".aac", ".mp4"): pregap = 1024 total_samples = int(duration_ms * sample_rate / 1000) if duration_ms else 0 sample_count = max(0, total_samples - pregap - postgap) return pregap, postgap, sample_count def _scanned_tracks_to_iop(mountpoint: str, scanned: list[dict], artwork_map: dict[str, tuple[int, int, int]] | None = None) -> list: """Convert our scanned track dicts to iOpenPod TrackInfo objects.""" IopTrackInfo = _import_iop("iTunesDB_Writer.mhit_writer").TrackInfo tracks: list = [] for s in scanned: fpath = s["file_path"] fname = os.path.basename(fpath) base, ext = os.path.splitext(fname) ft = ext.lstrip(".").lower() rel = os.path.relpath(fpath, os.path.join(mountpoint, "iPod_Control", "Music")) location = f":iPod_Control:Music:{rel.replace('/', ':')}" size = os.path.getsize(fpath) art_count = 0 art_size = 0 mhii_link = 0 if artwork_map and fpath in artwork_map: mhii_link, art_count, art_size = artwork_map[fpath] sr = s.get("sample_rate", 44100) dur_ms = s.get("duration_ms", 0) pregap, postgap, sample_count = _probe_gapless(fpath, sr, dur_ms) has_album = bool(s.get("album")) t = IopTrackInfo( title=s.get("title") or base, location=location, size=size, length=dur_ms, filetype=ft, bitrate=s.get("bitrate", 0) // 1000, sample_rate=sr, artist=s.get("artist") or "", album=s.get("album") or "", album_artist=s.get("album_artist") or s.get("artist") or "", genre=s.get("genre") or "", composer=s.get("composer") or "", year=s.get("year", 0), track_number=s.get("track_number", 0), total_tracks=s.get("track_count", 0), disc_number=s.get("disc_number", 0), total_discs=s.get("disc_count", 0), artwork_count=art_count, artwork_size=art_size, mhii_link=mhii_link, pregap=pregap, postgap=postgap, sample_count=sample_count, gapless_track_flag=1, gapless_album_flag=1 if has_album else 0, play_count=s.get("play_count", 0), rating=s.get("rating", 0), last_played=s.get("last_played", 0), skip_count=s.get("skip_count", 0), played_mark=-1, ) tracks.append(t) return tracks # ═══════════════════════════════════════════════════════════════════════ # Nano7Database (public API kept compatible with main.py) # ═══════════════════════════════════════════════════════════════════════ class Nano7Database: """Manages iPod Nano 7 database via iOpenPod writers.""" def __init__(self, mountpoint: str): self.mountpoint = mountpoint self.music_base = os.path.join(mountpoint, "iPod_Control", "Music") self.firewire_id = self._detect_firewire_id() # ── device info ──────────────────────────────────────────────────── def _detect_firewire_id(self) -> bytes: sysinfo = os.path.join(self.mountpoint, "iPod_Control", "Device", "SysInfo") if not os.path.exists(sysinfo): return b"" try: with open(sysinfo, "r") as f: for line in f: if line.startswith("FirewireGuid:"): hs = line.split(":")[1].strip() if hs.startswith("0x"): hs = hs[2:] fw = bytes.fromhex(hs) logger.info("FireWire GUID: %s", hs) return fw except Exception as e: logger.warning("Failed to parse FirewireGuid: %s", e) return b"" # ── scan tracks from iPod filesystem ─────────────────────────────── def _scan_ipod_files(self) -> list[dict]: """Walk iPod_Control/Music/FXX/ and extract metadata from every audio file.""" results: list[dict] = [] if not os.path.isdir(self.music_base): return results for folder in sorted(os.listdir(self.music_base)): d = os.path.join(self.music_base, folder) if not os.path.isdir(d): continue for fname in sorted(os.listdir(d)): fpath = os.path.join(d, fname) if not os.path.isfile(fpath): continue ext = os.path.splitext(fname)[1].lower() if ext not in (".mp3", ".m4a", ".aac", ".mp4", ".m4p"): continue tags = _read_audio_tags(fpath) tags["file_path"] = fpath tags["file_name"] = fname tags["file_size"] = os.path.getsize(fpath) tags["pid"] = self._hash_to_pid(fpath) tags["content_hash"] = content_hash(fpath) results.append(tags) return results def _hash_to_pid(self, filepath: str) -> int: """Generate a stable negative pid from file path (for API compat).""" h = hashlib.sha1(filepath.encode()).digest() # Map to signed 64-bit negative range raw = int.from_bytes(h[:8], "big") if raw >= (1 << 63): raw = -(raw & ((1 << 63) - 1)) if raw >= 0: raw = -raw - 1 return raw # ── add / list / delete / dedup ─────────────────────────────────── def add_track(self, filepath: str, **kwargs) -> Optional[dict[str, Any]]: """Copy track to iPod and regenerate all databases. Returns None if track is a duplicate (title+artist+album match). """ if not os.path.exists(filepath): raise FileNotFoundError(f"File not found: {filepath}") tags = _read_audio_tags(filepath) title = tags.get("title", os.path.splitext(os.path.basename(filepath))[0]) artist = tags.get("artist", "Unknown Artist") album = tags.get("album", "") # Duplicate check: scan existing files on iPod existing = self.get_all_tracks() for e in existing: if (e["title"].lower() == title.lower() and e["artist"].lower() == artist.lower() and (e.get("album") or "").lower() == album.lower()): logger.info("Duplicate: '%s' by %s, skipping", title, artist) return None # Copy file target_dir = self._find_free_music_dir() os.makedirs(os.path.join(self.music_base, target_dir), exist_ok=True) fname = self._gen_ipod_name(filepath) dest = os.path.join(self.music_base, target_dir, fname) shutil.copy2(filepath, dest) logger.info("Copied: %s → :iPod_Control:Music:%s:%s", os.path.basename(filepath), target_dir, fname) # Regenerate all databases via iOpenPod self.sync_itunescdb() return {"title": title, "artist": artist, "album": album, "pid": self._hash_to_pid(dest), "ipod_path": f":iPod_Control:Music:{target_dir}:{fname}", "physical_order": 0} def _find_free_music_dir(self) -> str: dirs = sorted(os.listdir(self.music_base)) if os.path.isdir(self.music_base) else [] min_files = float("inf") target = "F00" for d in dirs: full = os.path.join(self.music_base, d) if os.path.isdir(full): cnt = len(os.listdir(full)) if cnt < min_files: min_files = cnt target = d return target def _gen_ipod_name(self, filepath: str) -> str: base, ext = os.path.splitext(os.path.basename(filepath)) prefix = "".join(random.choices("ABCDEFGHIJKLMNOPQRSTUVWXYZ", k=4)) num = random.randint(0, 999999) return f"{prefix}{num:06d}{ext}" def get_track_count(self) -> int: return len(self._scan_ipod_files()) def get_all_tracks(self) -> list[dict]: """Return all tracks on iPod with full metadata including play stats.""" scanned = self._scan_ipod_files() base_stats, delta_stats = self.read_play_stats() tracks = [] for s in scanned: rel = os.path.relpath(s["file_path"], self.music_base) location = f":iPod_Control:Music:{rel.replace(os.sep, ':')}" bs = base_stats.get(location, {}) dd = delta_stats.get(location, {}) pc = bs.get("play_count", 0) + dd.get("play_count", 0) sc = bs.get("skip_count", 0) + dd.get("skip_count", 0) rating = dd.get("rating", 0) or bs.get("rating", 0) tracks.append({ "pid": s["pid"], "title": s.get("title") or os.path.splitext(s["file_name"])[0], "artist": s.get("artist") or "Unknown Artist", "album": s.get("album") or "", "album_artist": s.get("album_artist") or "", "duration_ms": s.get("duration_ms", 0), "track_number": s.get("track_number", 0), "disc_number": s.get("disc_number", 0), "physical_order": 0, "date_modified": 0, "year": s.get("year", 0), "ipod_path": None, "file_path": s["file_path"], "content_hash": s.get("content_hash", ""), "play_count": pc, "rating": rating, "last_played": bs.get("last_played", 0), "skip_count": sc, }) return tracks def delete_track(self, pid: int) -> bool: """Delete a single track by pid (scans, deletes, syncs).""" return self.delete_tracks([pid]) > 0 def delete_tracks(self, pids: list[int], progress_callback=None) -> int: """Delete tracks by pid. Scans once, deletes all matching files, then regenerates the database exactly once. Args: pids: List of track pids to delete. progress_callback: Optional callable(percent, status) for progress. Returns: Number of files actually deleted. """ pid_set = set(pids) scanned = self._scan_ipod_files() deleted = 0 for i, s in enumerate(scanned): if s["pid"] in pid_set and os.path.exists(s["file_path"]): os.remove(s["file_path"]) logger.info("Deleted: %s", s["file_path"]) deleted += 1 pid_set.discard(s["pid"]) if progress_callback and i % 5 == 0: progress_callback( min(50, int((i + 1) / max(len(scanned), 1) * 50)), f"Deleting files... {deleted} removed", ) if deleted: if progress_callback: progress_callback(55, "Regenerating iPod database...") self.sync_itunescdb() if progress_callback: progress_callback(100, f"Removed {deleted} track(s)") return deleted def remove_duplicates(self, progress_callback=None) -> list[dict]: """Find and remove duplicate tracks (same title+artist+album).""" all_tracks = self._scan_ipod_files() seen: dict[tuple, list[dict]] = {} for t in all_tracks: key = (t.get("title", "").lower(), t.get("artist", "").lower(), t.get("album", "").lower()) seen.setdefault(key, []).append(t) removed = [] for key, items in seen.items(): if len(items) <= 1: continue for item in items[1:]: fpath = item["file_path"] if os.path.exists(fpath): os.remove(fpath) removed.append({ "pid": item["pid"], "title": item.get("title", ""), "artist": item.get("artist", ""), "album": item.get("album", ""), "file_path": fpath, }) logger.info("Removed duplicate: %s", fpath) if progress_callback: progress_callback( min(50, int(len(removed) / max(len(all_tracks), 1) * 50)), f"Removing duplicates... {len(removed)} found", ) if removed: if progress_callback: progress_callback(55, "Regenerating iPod database...") self.sync_itunescdb() if progress_callback: progress_callback(100, f"Removed {len(removed)} duplicate(s)") logger.info("Removed %d duplicate(s)", len(removed)) return removed def get_orphaned_files(self) -> list[dict]: """Return files on iPod with no matching DB entry.""" # Since we regenerate the DB from scratch every sync, there # shouldn't be orphans. Still check for non-audio files. orphans = [] if not os.path.isdir(self.music_base): return orphans for folder in sorted(os.listdir(self.music_base)): d = os.path.join(self.music_base, folder) if not os.path.isdir(d): continue for fname in os.listdir(d): fpath = os.path.join(d, fname) if not os.path.isfile(fpath): continue ext = os.path.splitext(fname)[1].lower() if ext not in (".mp3", ".m4a", ".aac", ".mp4", ".m4p"): orphans.append({ "file_path": fpath, "size": os.path.getsize(fpath), "title": fname, "artist": "Unknown", }) return orphans def delete_orphaned_files(self, file_paths: list[str] = None) -> int: deleted = 0 if file_paths is None: orphans = self.get_orphaned_files() file_paths = [o["file_path"] for o in orphans] for p in file_paths: if os.path.exists(p): os.remove(p) deleted += 1 return deleted # ── sync: regenerate everything via iOpenPod ────────────────────── def sync_itunescdb(self) -> None: """Regenerate ALL databases using iOpenPod's writers.""" try: self._sync_itunescdb_impl() except Exception as e: logger.exception("iTunesDB sync failed: %s", e) def _merge_play_stats(self, scanned: list[dict]) -> None: """Merge iPod + local play stats into *scanned* track dicts in-place. Uses additive model (like iTunes): ``new_value = max(ipod_base, local_count) + ipod_delta`` """ base_stats, delta_stats = self.read_play_stats() local_play_stats = self._load_local_play_stats() for s in scanned: rel = os.path.relpath(s["file_path"], self.music_base) location = f":iPod_Control:Music:{rel.replace(os.sep, ':')}" bs = base_stats.get(location, {}) dd = delta_stats.get(location, {}) ch = s.get("content_hash", "") local_pc = local_play_stats.get(ch, {}) s["play_count"] = max( bs.get("play_count", 0), local_pc.get("play_count", 0), ) + dd.get("play_count", 0) s["skip_count"] = max( bs.get("skip_count", 0), local_pc.get("skip_count", 0), ) + dd.get("skip_count", 0) s["rating"] = dd.get("rating", 0) or bs.get("rating", 0) or local_pc.get("rating", 0) s["last_played"] = max( bs.get("last_played", 0), local_pc.get("last_played", 0), ) def _cleanup_play_counts_file(self) -> None: """Delete the iPod 'Play Counts' delta file after a full sync.""" pc_file = os.path.join( self.mountpoint, "iPod_Control", "iTunes", "Play Counts", ) if os.path.exists(pc_file): try: os.remove(pc_file) logger.debug("Deleted Play Counts file after sync") except OSError as e: logger.debug("Could not delete Play Counts file: %s", e) def consume_play_deltas(self) -> None: """Public entry point: delete the Play Counts file after consuming its data during mount. Prevents double-counting on subsequent ``read_play_stats()`` calls.""" self._cleanup_play_counts_file() def _sync_itunescdb_impl(self) -> None: scanned = self._scan_ipod_files() if not scanned: logger.warning("No audio files found on iPod — writing empty databases") self._write_empty_databases() return self._merge_play_stats(scanned) self._write_databases_from_tracks(scanned) self._cleanup_play_counts_file() def find_track_by_content_hash(self, content_hash: str) -> dict | None: """Find an iPod track by its content hash.""" if not content_hash: return None for t in self.get_all_tracks(): if t.get("content_hash") == content_hash: return t return None def find_track_by_metadata( self, title: str, artist: str, album: str = "", ) -> dict | None: """Find an iPod track by title+artist+album (case-insensitive, fallback matching).""" for t in self.get_all_tracks(): if (t["title"].lower() == title.lower() and t["artist"].lower() == artist.lower() and (t.get("album") or "").lower() == album.lower()): return t return None def read_play_stats(self) -> tuple[dict[str, dict[str, int]], dict[str, dict[str, int]]]: """Read iPod play stats split into base (written) and delta (since last write). Returns ``(base_stats, delta_stats)`` keyed by iPod ``location``. **base_stats** — values from the binary iTunesDB / Dynamic.itdb (the last written totals). Fields: ``play_count``, ``rating``, ``last_played``, ``skip_count``. **delta_stats** — deltas from the ``Play Counts`` file accumulated since the last database write. Fields: ``play_count``, ``skip_count``, ``rating`` (0 if no delta activity). Both return empty dicts when the databases cannot be read (new or corrupted iPod state). """ ipod_device = _import_iop("ipod_device") try: itdb_path = ipod_device.resolve_itdb_path(self.mountpoint) except Exception: itdb_path = os.path.join( self.mountpoint, "iPod_Control", "iTunes", "iTunesDB", ) base_stats: dict[str, dict[str, int]] = {} delta_stats: dict[str, dict[str, int]] = {} try: ipl = _import_iop("iTunesDB_Parser.ipod_library") data = ipl.load_ipod_library(itdb_path, merge_playcounts=False) except Exception as e: logger.debug("Cannot read iPod play base stats: %s", e) return base_stats, delta_stats if not data: return base_stats, delta_stats tracks = data.get("mhlt", []) for track in tracks: loc = track.get("location", "") if not loc: continue base_stats[loc] = { "play_count": track.get("play_count_1", 0) or 0, "rating": track.get("rating", 0) or 0, "last_played": track.get("last_played", 0) or 0, "skip_count": track.get("skip_count", 0) or 0, } try: pc_mod = _import_iop("iTunesDB_Parser.playcounts") pc_dir = os.path.dirname(itdb_path) pc_path = os.path.join(pc_dir, "Play Counts") entries = pc_mod.parse_playcounts(pc_path) if entries: for i, entry in enumerate(entries): if i >= len(tracks): break loc = tracks[i].get("location", "") if not loc: continue delta_stats[loc] = { "play_count": entry.play_count, "skip_count": entry.skip_count, "rating": entry.rating, } except Exception as e: logger.debug("Cannot read iPod Play Counts deltas: %s", e) logger.debug( "Read play stats — base: %d tracks, delta: %d tracks", len(base_stats), len(delta_stats), ) return base_stats, delta_stats @staticmethod def _load_local_play_stats() -> dict[str, dict[str, int]]: """Read accumulated play stats from the local library cache. Returns ``{content_hash: {play_count, rating, last_played, skip_count}}`` for every cached entry that has play data. """ from library_cache import get_library_cache cache = get_library_cache() result: dict[str, dict[str, int]] = {} for entry in cache._data.values(): ch = entry.get("_content_hash", "") pc = entry.get("play_count", 0) rating = entry.get("rating", 0) lp = entry.get("last_played", 0) sc = entry.get("skip_count", 0) if pc or rating or lp or sc: if ch: result[ch] = { "play_count": pc, "rating": rating, "last_played": lp, "skip_count": sc, } return result def update_track_metadata( self, pid: int, new_metadata: dict, local_source_path: str | None = None, ) -> bool: """Update metadata for a single iPod track without touching audio files. Only updates the databases (iTunesDB + SQLite + ArtworkDB). The audio file on the iPod is never modified — new metadata is written directly into the database records. Args: pid: The iPod track pid (from get_all_tracks). new_metadata: Dict with optional keys: title, artist, album, album_artist, genre, composer, year, track_number, disc_number, comment. local_source_path: If provided, artwork is extracted from this local file and encoded into iPod-native formats. Returns: True if the track was found and updated, False otherwise. """ scanned = self._scan_ipod_files() if not scanned: logger.warning("No tracks on iPod to update") return False target_idx = None for idx, s in enumerate(scanned): if s["pid"] == pid: target_idx = idx break if target_idx is None: logger.warning("Track with pid=%d not found on iPod", pid) return False target = scanned[target_idx] text_fields = { "title": "title", "artist": "artist", "album": "album", "album_artist": "album_artist", "genre": "genre", "composer": "composer", "comment": "comment", } for meta_key, scan_key in text_fields.items(): if meta_key in new_metadata: target[scan_key] = str(new_metadata[meta_key]) if new_metadata[meta_key] else "" if "year" in new_metadata: target["year"] = int(new_metadata["year"] or 0) if "track_number" in new_metadata: target["track_number"] = int(new_metadata["track_number"] or 0) if "disc_number" in new_metadata: target["disc_number"] = int(new_metadata["disc_number"] or 0) artwork_overrides: dict[str, object] = {} if local_source_path and os.path.exists(local_source_path): try: art_result = prepare_artwork_for_track(local_source_path) if art_result is not None: artwork_overrides[target["file_path"]] = art_result else: artwork_overrides[target["file_path"]] = _REMOVE_ARTWORK except Exception as e: logger.warning("Failed to extract artwork from %s: %s", local_source_path, e) self._merge_play_stats(scanned) self._write_databases_from_tracks(scanned, artwork_overrides=artwork_overrides) self._cleanup_play_counts_file() logger.info("Updated metadata for track: %s", target.get("title", "?")) return True def _write_databases_from_tracks( self, scanned: list[dict], artwork_overrides: dict[str, object] | None = None, ) -> None: """Build artwork, convert to TrackInfo, write iTunesDB + SQLite. Args: scanned: List of track dicts from _scan_ipod_files(). artwork_overrides: Optional dict mapping iPod file_path → (art_hash, {fmt_id: EncodedFormatPayload}) for tracks whose artwork should come from a local file instead of the iPod file. Use the ``_REMOVE_ARTWORK`` sentinel to explicitly remove artwork for a track. Omit a key to keep existing iPod artwork. """ artwork_overrides = artwork_overrides or {} # 1. Prepare artwork for all tracks artwork_map: dict[str, tuple[int, int, int]] = {} art_entries: list[ArtworkEntry] = [] art_hash_to_img_id: dict[str, int] = {} next_img_id = 100 for idx, s in enumerate(scanned): fpath = s["file_path"] override = artwork_overrides.get(fpath) if override is None: try: art_result = prepare_artwork_for_track(fpath) except Exception: art_result = None elif override is _REMOVE_ARTWORK: art_result = None else: art_result = override if art_result is not None: art_hash, formats = art_result art_count = len(formats) total_size = sum(p.size for p in formats.values()) if art_hash not in art_hash_to_img_id: img_id = next_img_id next_img_id += 1 art_hash_to_img_id[art_hash] = img_id art_entries.append(ArtworkEntry( img_id=img_id, db_track_id=idx, art_hash=art_hash, src_img_size=total_size, formats=formats, db_track_ids=[idx], )) else: img_id = art_hash_to_img_id[art_hash] for entry in art_entries: if entry.img_id == img_id: if idx not in entry.db_track_ids: entry.db_track_ids.append(idx) break artwork_map[fpath] = (img_id, art_count, total_size) if art_entries: try: write_artworkdb(self.mountpoint, art_entries) logger.info("ArtworkDB written: %d unique covers, %d total tracks", len(art_entries), len(artwork_map)) except Exception as e: logger.warning("ArtworkDB write failed: %s", e) # 2. Convert to iOpenPod TrackInfo tracks = _scanned_tracks_to_iop(self.mountpoint, scanned, artwork_map) logger.info("Built %d iOpenPod TrackInfo objects", len(tracks)) # 3. Determine capabilities for this device ipod_device = _import_iop("ipod_device") capabilities = None try: dev = ipod_device.get_current_device() if dev and dev.model_family: capabilities = ipod_device.capabilities_for_family_gen( dev.model_family, dev.generation or "", ) if capabilities: logger.debug("Device capabilities: %s", capabilities) except Exception as e: logger.debug("Capabilities detection failed: %s", e) # 4. Generate and write binary iTunesDB first (to get db_pid) try: firewire_id = ipod_device.get_firewire_id(self.mountpoint) except Exception: firewire_id = self.firewire_id iop_write_itunesdb = _import_iop("iTunesDB_Writer").write_itunesdb ok = iop_write_itunesdb( self.mountpoint, tracks, firewire_id=firewire_id, backup=True, capabilities=capabilities, master_playlist_name="iPod", ) if not ok: logger.error("iTunesDB write failed") return logger.info("iTunesDB written (%d tracks)", len(tracks)) # 5. Extract db_pid from the binary DB for SQLite cross-reference db_pid = 0 try: ipod_dev = _import_iop("ipod_device") cdb_path = ipod_dev.resolve_itdb_path(self.mountpoint) if cdb_path: with open(cdb_path, "rb") as f: hdr = f.read(0x20) if len(hdr) >= 0x20 and hdr[:4] == b"mhbd": db_pid = struct.unpack_from(" None: """Write minimal empty databases when no tracks exist.""" ipod_device = _import_iop("ipod_device") iop_write_itunesdb = _import_iop("iTunesDB_Writer").write_itunesdb write_sqlite_databases = _import_iop("SQLiteDB_Writer").write_sqlite_databases try: fwid = ipod_device.get_firewire_id(self.mountpoint) except Exception: fwid = self.firewire_id capabilities = None try: dev = ipod_device.get_current_device() if dev and dev.model_family: capabilities = ipod_device.capabilities_for_family_gen( dev.model_family, dev.generation or "") except Exception: pass iop_write_itunesdb(self.mountpoint, [], firewire_id=fwid, backup=True, capabilities=capabilities) write_sqlite_databases( ipod_path=self.mountpoint, tracks=[], master_playlist_name="iPod", db_pid=random.getrandbits(64), capabilities=capabilities, firewire_id=fwid, backup=True, ) self._cleanup_play_counts_file()