#!/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 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 _IOP_ROOT = "/tmp/iOpenPod" _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: # Evict local shadows ONCE so iOpenPod modules resolve correctly. sys.modules.pop("ipod_device", None) _IOP_READY = True logger.debug("iOpenPod boot: evicted local ipod_device") had_iop = _IOP_ROOT in sys.path if not had_iop: sys.path.insert(0, _IOP_ROOT) try: mod = importlib.import_module(module_name) finally: if not had_iop: sys.path.remove(_IOP_ROOT) _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 # ── Helpers ────────────────────────────────────────────────────────── def _to_mac_epoch(ts: int) -> int: return ts + MAC_EPOCH_OFFSET if ts > 0 else 0 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, ) 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) 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.""" scanned = self._scan_ipod_files() tracks = [] for s in scanned: 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"], }) 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 _sync_itunescdb_impl(self) -> None: # 1. Scan all audio files on iPod scanned = self._scan_ipod_files() if not scanned: logger.warning("No audio files found on iPod — writing empty databases") # Write minimal empty databases self._write_empty_databases() return # 2. 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"] try: art_result = prepare_artwork_for_track(fpath) except Exception: art_result = None 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) # 3. Convert to iOpenPod TrackInfo tracks = _scanned_tracks_to_iop(self.mountpoint, scanned, artwork_map) logger.info("Built %d iOpenPod TrackInfo objects", len(tracks)) # 4. 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) # 5. 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)) # 6. 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, )