#!/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, Callable, 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__) # iOpenPod parsers/writers log ERROR for transient I/O (USB resettling) logging.getLogger("iTunesDB_Parser").setLevel(logging.CRITICAL) logging.getLogger("iTunesDB_Writer").setLevel(logging.CRITICAL) # ── 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]) # Compilation flag (TCMP frame) — only set when explicitly present try: id3 = ID3(filepath) if "TCMP" in id3: tcmp = id3["TCMP"] if hasattr(tcmp, "text") and tcmp.text: info["compilation"] = str(tcmp.text[0]) == "1" except Exception: pass 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] # Compilation flag (cpil atom) if "cpil" in t: cpil = t["cpil"] if isinstance(cpil, bool): info["compilation"] = cpil elif isinstance(cpil, list) and len(cpil) > 0: info["compilation"] = bool(cpil[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 # Compilation flag (vorbis comment) try: if "compilation" in audio.tags: comp = audio.tags["compilation"] comp_val = str(comp[0]) if isinstance(comp, list) else str(comp) info["compilation"] = comp_val == "1" 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, compilation_flag=s.get("compilation", False), 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) sc = bs.get("skip_count", 0) dd_rt = dd.get("rating", 0) rating = dd_rt if dd_rt >= 0 else 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 "", "genre": s.get("genre", ""), "duration_ms": s.get("duration_ms", 0), "track_number": s.get("track_number", 0), "file_size": s.get("file_size", 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 # ── playlist creation ──────────────────────────────────────────── def create_playlist(self, name: str, track_paths: list[str]) -> bool: """Create a user playlist on the iPod. Two-pass matching: content hash first, then title+artist+album fallback. Only tracks already present on the iPod will be included in the playlist. Args: name: Playlist name. track_paths: Local file paths of the tracks to include. Returns: True if the playlist was created, False otherwise. """ scanned = self._scan_ipod_files() if not scanned: logger.warning("No tracks on iPod — cannot create playlist") return False playlist_created = False self._merge_play_stats(scanned) def _build_pl_cb(scanned, tracks): """Build playlist using real db_track_id from TrackInfo.""" nonlocal playlist_created try: IopPlaylistInfo = _import_iop( "iTunesDB_Writer.mhyp_writer" ).PlaylistInfo except Exception as e: logger.exception("Failed to import PlaylistInfo: %s", e) return None # Pass 1: content_hash → db_track_id hash_to_db_id: dict[str, int] = {} for i, s in enumerate(scanned): ch = s.get("content_hash", "") if ch: hash_to_db_id[ch] = tracks[i].db_track_id found_ids: list[int] = [] unmatched: list[str] = [] for path in track_paths: if not os.path.exists(path): continue ch = content_hash(path) db_id = hash_to_db_id.get(ch) if db_id is not None: found_ids.append(db_id) else: unmatched.append(path) hash_matched_count = len(found_ids) # Pass 2: metadata fallback if unmatched: scanned_meta = [] for i, s in enumerate(scanned): scanned_meta.append(( s.get("title", "").lower(), s.get("artist", "").lower(), (s.get("album") or "").lower(), tracks[i].db_track_id, )) found_set = set(found_ids) for path in unmatched: tags = _read_audio_tags(path) title = (tags.get("title") or "").lower() artist = (tags.get("artist") or "").lower() album = (tags.get("album") or "").lower() if not title or not artist: continue for s_title, s_artist, s_album, db_id in scanned_meta: if (db_id not in found_set and s_title == title and s_artist == artist and s_album == album): found_ids.append(db_id) found_set.add(db_id) break if not found_ids: logger.warning( "No tracks from playlist '%s' found on iPod", name, ) return None matched_by_hash = hash_matched_count matched_by_meta = len(found_ids) - hash_matched_count logger.info( "Playlist '%s': %d track(s) matched (%d by hash, %d by metadata)", name, len(found_ids), matched_by_hash, matched_by_meta, ) playlist_created = True return [IopPlaylistInfo(name=name, track_ids=found_ids)] db_ok = self._write_databases_from_tracks( scanned, build_extra_playlists=_build_pl_cb, ) self._cleanup_play_counts_file() return db_ok and playlist_created # ── sync: regenerate everything via iOpenPod ────────────────────── def sync_itunescdb(self) -> bool: """Regenerate ALL databases using iOpenPod's writers. Returns True if the databases were written successfully. """ try: self._sync_itunescdb_impl() return True except Exception as e: logger.exception("iTunesDB sync failed: %s", e) return False 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_cache, play_stats_store) + ipod_delta`` Includes a **validation gate** that refuses to produce all-zero play counts when we have previously-recorded non-zero data in the local PlayStatsStore — this prevents accidental data loss from silent iTunesDB parse failures. """ base_stats, delta_stats = self.read_play_stats() local_play_stats = self._load_local_play_stats() # Supplement local_play_stats with durable PlayStatsStore (SQLite) from library_cache import get_library_cache db_store = get_library_cache() 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", "") # Three-tier fallback: iTunesDB → LibraryCache (SQLite) → old JSON cache local_pc = local_play_stats.get(ch, {}) db_entry = db_store.get_by_content_hash(ch) if ch else None s["play_count"] = max( bs.get("play_count", 0), (db_entry or {}).get("play_count", 0), local_pc.get("play_count", 0), ) s["skip_count"] = max( bs.get("skip_count", 0), (db_entry or {}).get("skip_count", 0), local_pc.get("skip_count", 0), ) dd_rating = dd.get("rating", 0) store_rating = (db_entry or {}).get("rating", 0) s["rating"] = dd_rating if dd_rating >= 0 else ( bs.get("rating", 0) or store_rating or local_pc.get("rating", 0) ) s["last_played"] = max( bs.get("last_played", 0), (db_entry or {}).get("last_played", 0), local_pc.get("last_played", 0), ) # ── validation gate ──────────────────────────────────────────── # If every scanned track would get play_count=0 but the local DB # previously had non-zero values, restore them. all_zero = all(s.get("play_count", 0) == 0 for s in scanned) if all_zero: restored = 0 for s in scanned: ch = s.get("content_hash", "") if not ch: continue db_entry = db_store.get_by_content_hash(ch) if ch else None stored_pc = (db_entry or {}).get("play_count", 0) if stored_pc > 0: s["play_count"] = stored_pc restored += 1 if restored: logger.warning( "Validation gate triggered — restored play_count for %d track(s) " "from local database after merge produced all zeros", restored, ) # ─────────────────────────────────────────────────────────────── 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 _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) if not self._write_databases_from_tracks(scanned): logger.error("iTunesDB sync failed — play counts not cleaned up") return 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") or 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") or 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 def read_play_counts_delta(self) -> dict[str, dict[str, int]]: """Read Play Counts file using filesystem scan (no iTunesDB needed). Positional mapping uses ``_scan_ipod_files()`` order (correct after any sync written by this app). Returns ``{location: {play_count, skip_count, rating}}`` or empty dict on failure. """ pc_path = os.path.join( self.mountpoint, "iPod_Control", "iTunes", "Play Counts", ) if not os.path.exists(pc_path): return {} pc_mod = _import_iop("iTunesDB_Parser.playcounts") try: entries = pc_mod.parse_playcounts(pc_path) except Exception: return {} if not entries: return {} scanned = self._scan_ipod_files() delta: dict[str, dict[str, int]] = {} for i, entry in enumerate(entries): if i >= len(scanned): break if entry.play_count == 0 and entry.skip_count == 0 and entry.rating < 0: continue rel = os.path.relpath(scanned[i]["file_path"], self.music_base) loc = f":iPod_Control:Music:{rel.replace(os.sep, ':')}" delta[loc] = { "play_count": entry.play_count, "skip_count": entry.skip_count, "rating": entry.rating, } return delta @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 _path, entry in cache.iter_entries(): 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, extra_playlists: list | None = None, build_extra_playlists: Callable | None = None, ) -> bool: """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. extra_playlists: Pre-built list of PlaylistInfo (ignored when build_extra_playlists is provided). build_extra_playlists: Optional callable(scanned, tracks) → list of PlaylistInfo. Called after TrackInfo conversion and db_track_id generation, so the callback can use real db_track_id values from TrackInfo objects. Returns: True if databases were written successfully. """ 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)) # 2b. Pre-generate db_track_id for all tracks so callbacks can use them generate_db_track_id = _import_iop("iTunesDB_Writer.mhit_writer").generate_db_track_id for track in tracks: if track.db_track_id == 0: track.db_track_id = generate_db_track_id() # 2c. Build extra playlists via callback if provided if build_extra_playlists is not None: result = build_extra_playlists(scanned, tracks) if result is not None: extra_playlists = result else: extra_playlists = None # 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", playlists=extra_playlists or None, ) if not ok: logger.error("iTunesDB write failed") return False 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()