diff --git a/src/ipod_mhbd_writer.py b/src/ipod_mhbd_writer.py new file mode 100644 index 0000000..725b40c --- /dev/null +++ b/src/ipod_mhbd_writer.py @@ -0,0 +1,982 @@ +#!/usr/bin/env python3 +""" +Self-contained iTunesDB (mhbd) binary writer for iPod Nano 7G. + +Generates a complete, properly-structured mhbd database from SQLite track +data, with HASHAB signing at scheme=4. Replaces the broken sync_itunescdb() +approach that used "mhfd" magic and raw SQLite compression. + +The binary layout documented below is cross-referenced against: + - libgpod (itdb_itunesdb.c) + - iPodLinux wiki (https://ipodlinux.org/wiki/ITunesDB) + - iOpenPod reference implementation +""" + +import hashlib +import logging +import os +import random +import sqlite3 +import struct +import time +import zlib +from typing import Optional + +from hashab import compute_hashab + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Global header-size constants +# --------------------------------------------------------------------------- +MHBD_HEADER_SIZE = 244 +MHSD_HEADER_SIZE = 96 +MHLT_HEADER_SIZE = 92 +MHIT_HEADER_SIZE = 0x270 # 624 — iTunes 10+/modern +MHLP_HEADER_SIZE = 92 +MHYP_HEADER_SIZE = 184 +MHIP_HEADER_SIZE = 76 +MHLA_HEADER_SIZE = 92 +MHIA_HEADER_SIZE = 88 +MHLI_HEADER_SIZE = 92 +MHII_HEADER_SIZE = 80 +MHOD_HEADER_SIZE = 24 +MHOD_STRING_SUBHEADER_SIZE = 16 +MHOD100_POSITION_BODY_SIZE = 20 + +# MHOD type constants +MHOD_TYPE_TITLE = 1 +MHOD_TYPE_LOCATION = 2 +MHOD_TYPE_ALBUM = 3 +MHOD_TYPE_ARTIST = 4 +MHOD_TYPE_ALBUM_ARTIST = 22 +MHOD_TYPE_FILETYPE = 6 +MHOD_TYPE_ARTIST_NAME = 300 +MHOD_TYPE_ALBUM_ALBUM = 200 +MHOD_TYPE_ALBUM_ARTIST_ITEM = 201 + +# File type codes (big-endian ASCII stored as LE u32) +FILETYPE_CODES = { + "mp3": 0x4D503320, # "MP3 " + "m4a": 0x4D344120, # "M4A " + "m4p": 0x4D345020, # "M4P " + "mp4": 0x4D503420, # "MP4 " + "aac": 0x41414320, # "AAC " + "wav": 0x57415620, # "WAV " +} + +# Mac epoch offset: seconds between 1904-01-01 and 1970-01-01 +MAC_EPOCH_OFFSET = 2082844800 + +# Audio format flags +AUDIO_FORMAT_FLAG_DEFAULT = 0xFFFF +AUDIO_FORMAT_FLAG_MAP = {"wav": 0x0000, "aiff": 0x0000, "aif": 0x0000, "m4b": 0x0001} + + +# --------------------------------------------------------------------------- +# Low-level helpers +# --------------------------------------------------------------------------- + +def _to_mac_epoch(ts: int) -> int: + return ts + MAC_EPOCH_OFFSET if ts > 0 else 0 + + +def _sample_rate_to_fixed(hz: int) -> int: + return (hz << 16) & 0xFFFFFFFF + + +def _write_generic_header(buf: bytearray, offset: int, tag: bytes, + header_len: int, total_or_count: int) -> None: + buf[offset:offset + 4] = tag + struct.pack_into(" int: + return random.getrandbits(64) + + +# --------------------------------------------------------------------------- +# MHOD string writer +# --------------------------------------------------------------------------- + +def _write_mhod_string(mhod_type: int, value: str) -> bytes: + if not value: + return b"" + string_data = value.encode("utf-16-le") + string_len = len(string_data) + total_len = MHOD_HEADER_SIZE + MHOD_STRING_SUBHEADER_SIZE + string_len + header = struct.pack( + "<4sIIIII", b"mhod", MHOD_HEADER_SIZE, total_len, + mhod_type, 0, 0, + ) + sub = struct.pack(" bytes: + total_len = MHOD_HEADER_SIZE + MHOD100_POSITION_BODY_SIZE + header = struct.pack( + "<4sIIIII", b"mhod", MHOD_HEADER_SIZE, total_len, 100, 0, 0, + ) + body = struct.pack(" bytes: + total_len = 0x288 + data = bytearray(total_len) + hdr = struct.pack("<4sIIIII", b"mhod", MHOD_HEADER_SIZE, total_len, 100, 0, 0) + data[:24] = hdr + struct.pack_into(" bytes: + """Build 8-byte sort_mhod_indicators.""" + ind = bytearray(8) + ind[0] = 0x81 if item.get("sort_title") else 0x80 + ind[1] = 0x81 if item.get("sort_album") else 0x80 + ind[2] = 0x81 if item.get("sort_artist") else 0x80 + ind[3] = 0x81 if item.get("sort_album_artist") else 0x80 + ind[4] = 0x80 + ind[5] = 0x80 + return bytes(ind) + + +# --------------------------------------------------------------------------- +# Track info normalized dict (parsed from SQLite + Filesystem) +# --------------------------------------------------------------------------- + +class TrackInfo: + """All fields needed to write a single MHIT chunk.""" + + def __init__(self, title: str = "", artist: str = "", album: str = "", + location: str = "", size: int = 0, length_ms: int = 0, + filetype: str = "mp3", bitrate: int = 0, sample_rate: int = 44100, + year: int = 0, track_number: int = 0, track_count: int = 0, + disc_number: int = 0, disc_count: int = 0, + album_artist: str = "", genre: str = "", composer: str = "", + art_count: int = 0, art_size: int = 0, mhii_link: int = 0, + album_id: int = 0, artist_id: int = 0, + db_track_id: int = 0): + self.title = title + self.artist = artist + self.album = album + self.album_artist = album_artist + self.location = location + self.size = size + self.length_ms = length_ms + self.filetype = filetype.lower() or "mp3" + self.bitrate = bitrate + self.sample_rate = sample_rate + self.year = year + self.track_number = track_number + self.track_count = track_count + self.disc_number = disc_number + self.disc_count = disc_count + self.genre = genre + self.composer = composer + self.art_count = art_count + self.art_size = art_size + self.mhii_link = mhii_link + self.album_id = album_id + self.artist_id = artist_id + # db_track_id: SQLite uses signed int but MHIT wants unsigned u64. + # Convert negative (signed) to unsigned 64-bit representation. + raw_id = db_track_id or _gen_id() + if raw_id < 0: + raw_id = raw_id + (1 << 64) + self.db_track_id = raw_id + self.date_added = int(time.time()) + self.sort_title = "" + self.sort_artist = "" + self.sort_album = "" + self.sort_album_artist = "" + self.compilation = False + self.rating = 0 + self.play_count = 0 + self.volume = 0 + self.vbr = False + self.purchased_aac = False + self.explicit = 0 + self.mpeg_audio_type = 0 + + @property + def filetype_code(self) -> int: + return FILETYPE_CODES.get(self.filetype, FILETYPE_CODES["mp3"]) + + @property + def audio_format_flag(self) -> int: + return AUDIO_FORMAT_FLAG_MAP.get(self.filetype, AUDIO_FORMAT_FLAG_DEFAULT) + + +# --------------------------------------------------------------------------- +# MHIT writer +# --------------------------------------------------------------------------- + +def _write_mhit(track: TrackInfo, track_id: int, db_id_2: int) -> bytes: + """Write a single MHIT chunk + child MHODs.""" + ft = track.filetype + + # --- Build child MHODs --- + mhod_parts: list[bytes] = [] + mhod_parts.append(_write_mhod_string(MHOD_TYPE_TITLE, track.title)) + mhod_parts.append(_write_mhod_string(MHOD_TYPE_LOCATION, track.location)) + if track.artist: + mhod_parts.append(_write_mhod_string(MHOD_TYPE_ARTIST, track.artist)) + if track.album: + mhod_parts.append(_write_mhod_string(MHOD_TYPE_ALBUM, track.album)) + if track.album_artist and track.album_artist != track.artist: + mhod_parts.append(_write_mhod_string(MHOD_TYPE_ALBUM_ARTIST, track.album_artist)) + if track.genre: + mhod_parts.append(_write_mhod_string(5, track.genre)) + if track.composer: + mhod_parts.append(_write_mhod_string(12, track.composer)) + # Sort MHODs + if track.sort_title: + mhod_parts.append(_write_mhod_string(27, track.sort_title)) + if track.sort_album: + mhod_parts.append(_write_mhod_string(28, track.sort_album)) + if track.sort_artist: + mhod_parts.append(_write_mhod_string(23, track.sort_artist)) + if track.sort_album_artist: + mhod_parts.append(_write_mhod_string(29, track.sort_album_artist)) + # Filetype desc + ft_desc = {"mp3": "MPEG audio file", "m4a": "AAC audio file", + "aac": "AAC audio file", "mp4": "AAC audio file", + "m4p": "Purchased AAC audio file"} + desc = ft_desc.get(ft, "MPEG audio file") + mhod_parts.append(_write_mhod_string(MHOD_TYPE_FILETYPE, desc)) + + mhod_data = b"".join(mhod_parts) + mhod_count = len(mhod_parts) + + header_size = MHIT_HEADER_SIZE + total_length = header_size + len(mhod_data) + + header = bytearray(header_size) + _write_generic_header(header, 0, b"mhit", header_size, total_length) + + # Pack all MHIT fields densely at correct offsets + mac_ts = _to_mac_epoch(int(time.time())) + mac_added = _to_mac_epoch(track.date_added) + struct.pack_into(" 0 else 2) # has_artwork + struct.pack_into("B", header, 0xA5, 0) # skip_when_shuffling + struct.pack_into("B", header, 0xA6, 0) # remember_position + struct.pack_into("B", header, 0xA7, 0) # podcast_flag + struct.pack_into(" 0 else 2) # played/not + struct.pack_into("B", header, 0xB3, 0) # unk0xB3 + struct.pack_into(" tuple[bytes, int]: + track_chunks = [] + tid = start_track_id + for trk in tracks: + track_chunks.append(_write_mhit(trk, tid, db_id_2)) + tid += 1 + header = bytearray(MHLT_HEADER_SIZE) + _write_generic_header(header, 0, b"mhlt", MHLT_HEADER_SIZE, len(tracks)) + return bytes(header) + b"".join(track_chunks), tid + + +# --------------------------------------------------------------------------- +# MHIP writer +# --------------------------------------------------------------------------- + +def _write_mhip(track_id: int, position: int, mhip_id: int = 0, pod_group_flag: int = 0, + pod_group_ref: int = 0, track_db_id: int = 0, mhip_persist_id: int = 0) -> bytes: + mhod_pos = _write_mhod100_position(position) + total = MHIP_HEADER_SIZE + len(mhod_pos) + header = bytearray(MHIP_HEADER_SIZE) + _write_generic_header(header, 0, b"mhip", MHIP_HEADER_SIZE, total) + struct.pack_into(" bytes: + pid = playlist_id or _gen_id() + ts = int(time.time()) + + # MHOD title + mhod_title = _write_mhod_string(MHOD_TYPE_TITLE, "iPod") + # MHOD 100 (playlist prefs) + mhod_prefs = _write_playlist_prefs_mhod100() + + # MHIP entries + mhips = b"".join(_write_mhip(tid, i) + for i, tid in enumerate(track_ids)) + mhip_count = len(track_ids) + + mhod_count = 2 # title + prefs + + total = MHYP_HEADER_SIZE + len(mhod_title) + len(mhod_prefs) + len(mhips) + header = bytearray(MHYP_HEADER_SIZE) + _write_generic_header(header, 0, b"mhyp", MHYP_HEADER_SIZE, total) + + struct.pack_into(" bytes: + """Write empty smart playlist list (type 5 stub).""" + header = bytearray(MHLP_HEADER_SIZE) + _write_generic_header(header, 0, b"mhlp", MHLP_HEADER_SIZE, 0) + return bytes(header) + + +def _write_splpref() -> bytes: + """Write SPLPref MHOD type 50 (132 bytes).""" + total = MHOD_HEADER_SIZE + 132 + data = bytearray(total) + hdr = struct.pack("<4sIIIII", b"mhod", MHOD_HEADER_SIZE, total, 50, 0, 0) + data[:24] = hdr + # liveUpdate=0, checkRules=0 (no rules → match all), checkLimits=0 + struct.pack_into("B", data, 24, 0) # liveUpdate + struct.pack_into("B", data, 25, 0) # checkRules = 0 = match all + struct.pack_into("B", data, 26, 0) # checkLimits + struct.pack_into("B", data, 27, 0) # limitType + struct.pack_into("B", data, 28, 0) # limitSort + struct.pack_into(" bytes: + """Write empty SLst MHOD type 51 (0 rules = match all).""" + SLST_SIZE = 136 + total = MHOD_HEADER_SIZE + SLST_SIZE + data = bytearray(total) + hdr = struct.pack("<4sIIIII", b"mhod", MHOD_HEADER_SIZE, total, 51, 0, 0) + data[:24] = hdr + data[24:28] = b"SLst" + # rule_count=0, conjunction=0, no rules → match all tracks + return bytes(data) + + +def _write_music_smart_mhyp(db_id_2: int) -> bytes: + """Write a 'Music' smart playlist (type 5) matching all tracks.""" + ts = int(time.time()) + mhod_title = _write_mhod_string(MHOD_TYPE_TITLE, "Music") + mhod_prefs = _write_playlist_prefs_mhod100() + mhod_splpref = _write_splpref() + mhod_slst = _write_slst_empty() + + # MHOD count: title + playlist prefs + splpref + slst = 4 + mhod_count = 4 + mhip_count = 0 # Smart playlists have no MHIPs — firmware evaluates rules at runtime + + total = (MHYP_HEADER_SIZE + len(mhod_title) + len(mhod_prefs) + + len(mhod_splpref) + len(mhod_slst)) + + header = bytearray(MHYP_HEADER_SIZE) + _write_generic_header(header, 0, b"mhyp", MHYP_HEADER_SIZE, total) + + struct.pack_into(" bytes: + total = MHSD_HEADER_SIZE + len(child_data) + header = bytearray(MHSD_HEADER_SIZE) + _write_generic_header(header, 0, b"mhsd", MHSD_HEADER_SIZE, total) + struct.pack_into(" bytes: + children = bytearray() + child_count = 0 + if album_name: + children.extend(_write_mhod_string(MHOD_TYPE_ALBUM_ALBUM, album_name)) + child_count += 1 + if album_artist: + children.extend(_write_mhod_string(MHOD_TYPE_ALBUM_ARTIST_ITEM, album_artist)) + child_count += 1 + total = MHIA_HEADER_SIZE + len(children) + header = bytearray(MHIA_HEADER_SIZE) + _write_generic_header(header, 0, b"mhia", MHIA_HEADER_SIZE, total) + struct.pack_into(" tuple[bytes, dict, int]: + """Returns (MHLA bytes, album_map, next_album_id).""" + groups: dict[tuple, list] = {} + for t in tracks: + key = (t.album.lower(), t.album_artist.lower()) if t.album else ("", "") + groups.setdefault(key, []).append(t) + + items = bytearray() + album_map: dict = {} + aid = 1 + for (alb_lower, aa_lower), group in sorted(groups.items()): + rep = group[0] + album_map[(rep.album, rep.album_artist or rep.artist)] = aid + items.extend(_write_mhia(aid, rep.album or "", rep.album_artist or rep.artist or "")) + aid += 1 + + header = bytearray(MHLA_HEADER_SIZE) + _write_generic_header(header, 0, b"mhla", MHLA_HEADER_SIZE, len(album_map)) + return bytes(header) + bytes(items), album_map, aid + + +# --------------------------------------------------------------------------- +# MHII / MHLI writer (artists) +# --------------------------------------------------------------------------- + +def _write_mhii_artist(artist_id: int, name: str) -> bytes: + children = bytearray() + child_count = 0 + if name: + children.extend(_write_mhod_string(MHOD_TYPE_ARTIST_NAME, name)) + child_count = 1 + total = MHII_HEADER_SIZE + len(children) + header = bytearray(MHII_HEADER_SIZE) + _write_generic_header(header, 0, b"mhii", MHII_HEADER_SIZE, total) + struct.pack_into(" tuple[bytes, dict, int]: + """Returns (MHLI bytes, artist_map, next_artist_id).""" + seen: dict[str, str] = {} + for t in tracks: + a = t.artist.lower() if t.artist else "" + if a and a not in seen: + seen[a] = t.artist + + items = bytearray() + amap: dict[str, int] = {} + art_id = 1 + for key in sorted(seen): + amap[key] = art_id + items.extend(_write_mhii_artist(art_id, seen[key])) + art_id += 1 + + header = bytearray(MHLI_HEADER_SIZE) + _write_generic_header(header, 0, b"mhli", MHLI_HEADER_SIZE, len(amap)) + return bytes(header) + bytes(items), amap, art_id + + +# =========================================================================== +# Top-level: Build the complete iTunesDB +# =========================================================================== + +def build_itunesdb(tracks: list[TrackInfo], + db_id: int = 0, + language: str = "ru", + timezone_offset: int = 0) -> bytes: + """Build complete mhbd iTunesDB bytes from TrackInfo list. + + Args: + tracks: list of TrackInfo objects. + db_id: database ID (generated if 0). + language: 2-char language code. + timezone_offset: signed int offset in seconds (0 = auto). + + Returns: + Full iTunesDB file content as bytes. + """ + db_id = db_id or _gen_id() + db_id_2 = _gen_id() + lib_pid = _gen_id() + + if timezone_offset == 0: + timezone_offset = -time.altzone if time.daylight else -time.timezone + + # ---- Build sub-datasets in order: 4 (albums), 8 (artists) ---------- + mhla_bytes, album_map, next_album_id = _write_mhla(tracks) + mhli_bytes, artist_map, next_artist_id = _write_mhli(tracks) + + # Assign album_id and artist_id to each track + for t in tracks: + t.album_id = album_map.get((t.album, t.album_artist or t.artist), 0) + t.artist_id = artist_map.get(t.artist.lower() if t.artist else "", 0) + + # ---- Type 1: Tracks ------------------------------------------------- + mhlt_bytes, next_track_id = _write_mhlt(tracks, next_artist_id + 1, db_id_2) + track_ids = list(range(next_artist_id + 1, next_track_id)) + + # ---- Type 3: Podcast list (master playlist copy) ------------------- + master_pl = _write_master_playlist(track_ids, db_id_2) + mhlp_hdr = bytearray(MHLP_HEADER_SIZE) + _write_generic_header(mhlp_hdr, 0, b"mhlp", MHLP_HEADER_SIZE, 1) + mhlp_type3 = bytes(mhlp_hdr) + master_pl + + # ---- Type 2: Playlist list (master playlist — primary) ------------- + mhlp_type2 = bytes(bytearray(mhlp_hdr)) + master_pl + + # ---- Type 5: Music smart playlist ----------------------------------- + music_mhyp = _write_music_smart_mhyp(db_id_2) + mhlp_type5 = bytearray(MHLP_HEADER_SIZE) + _write_generic_header(mhlp_type5, 0, b"mhlp", MHLP_HEADER_SIZE, 1) + mhlp_type5 = bytes(mhlp_type5) + music_mhyp + + # ---- Type 6, 10: Empty stubs ---------------------------------------- + mhlt_empty = bytearray(MHLT_HEADER_SIZE) + _write_generic_header(mhlt_empty, 0, b"mhlt", MHLT_HEADER_SIZE, 0) + mhsd_type6 = _write_mhsd(6, bytes(mhlt_empty)) + mhsd_type10 = _write_mhsd(10, bytes(mhlt_empty)) + + # ---- Assemble datasets: 4, 8, 1, 3, 2, 5, 6, 10 ----------------- + datasets = b"" + datasets += _write_mhsd(4, mhla_bytes) # albums + datasets += _write_mhsd(8, mhli_bytes) # artists + datasets += _write_mhsd(1, mhlt_bytes) # tracks + datasets += _write_mhsd(3, mhlp_type3) # podcast list copy + datasets += _write_mhsd(2, mhlp_type2) # playlist list (primary) + datasets += _write_mhsd(5, mhlp_type5) # smart playlists (Music) + datasets += mhsd_type6 # empty stub + datasets += mhsd_type10 # empty stub + child_count = 8 + + total_length = MHBD_HEADER_SIZE + len(datasets) + + # ---- Build MHBD header ---------------------------------------------- + header = bytearray(MHBD_HEADER_SIZE) + _write_generic_header(header, 0, b"mhbd", MHBD_HEADER_SIZE, total_length) + + struct.pack_into(" bool: + """Generate iTunesDB from tracks, sign with HASHAB scheme=4, write to iPod. + + Args: + ipod_mount: iPod mountpoint (e.g. /run/media/mat/IPOD). + tracks: list of TrackInfo objects. + firewire_id: 8+ byte FireWire GUID from SysInfo. + backup: create .backup of existing file. + + Returns: + True on success. + """ + itunes_dir = os.path.join(ipod_mount, "iPod_Control", "iTunes") + itdb_path = os.path.join(itunes_dir, "iTunesDB") + os.makedirs(itunes_dir, exist_ok=True) + + # Build the database + itdb_data = bytearray(build_itunesdb(tracks)) + + # ---- HASHAB signing ------------------------------------------------- + # MUST happen with scheme=4 set in header (already done in build_itunesdb) + if firewire_id and len(firewire_id) >= 8: + # Backup fields that will be zeroed for SHA1 + backup_db_id = bytes(itdb_data[0x18:0x20]) + backup_unk32 = bytes(itdb_data[0x32:0x46]) + + # Ensure scheme=4 + struct.pack_into("= 8: + data_for_hash = bytearray(cdb_data) + data_for_hash[0x18:0x20] = b"\x00" * 8 + data_for_hash[0x32:0x46] = b"\x00" * 20 + data_for_hash[0x58:0x6C] = b"\x00" * 20 + data_for_hash[0x72:0xA0] = b"\x00" * 46 + data_for_hash[0xAB:0xE4] = b"\x00" * 57 + struct.pack_into(" list[TrackInfo]: + """Read all tracks from Library.itdb SQLite and build TrackInfo list. + + Uses Locations.itdb to resolve file paths. + """ + itlp = os.path.join(mountpoint, "iPod_Control", "iTunes", "iTunes Library.itlp") + lib_path = os.path.join(itlp, "Library.itdb") + loc_path = os.path.join(itlp, "Locations.itdb") + + conn = sqlite3.connect(lib_path) + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + + # Attach Locations.itdb so we can JOIN across databases + if os.path.exists(loc_path): + conn.execute("ATTACH DATABASE ? AS locdb", (loc_path,)) + location_join = "LEFT JOIN locdb.location l ON i.pid = l.item_pid" + else: + location_join = "" + + query = f""" + SELECT i.pid, i.title, i.artist, i.album, i.album_artist, + i.total_time_ms, i.track_number, i.track_count, + i.disc_number, i.disc_count, i.year, i.artwork_cache_id, + i.artwork_status, i.physical_order, + a.audio_format, a.bit_rate, a.sample_rate, a.duration, + l.location, l.extension, l.file_size + FROM item i + LEFT JOIN avformat_info a ON i.pid = a.item_pid + {location_join} + ORDER BY i.physical_order + """ + cursor.execute(query) + rows = cursor.fetchall() + conn.close() + + # Filetype from extension code + ext_map: dict[int, str] = { + 0x4D503320: "mp3", 0x4D344120: "m4a", 0x4D345020: "m4p", + 0x41414320: "aac", 0x4D503420: "mp4", + } + + tracks = [] + for r in rows: + title = r["title"] or "Unknown" + artist = r["artist"] or "Unknown Artist" + album = r["album"] or "" + + location = r["location"] or "" + if location: + # Convert relative path to iPod colon path + loc = f":iPod_Control:Music:{location.replace('/', ':')}" + else: + loc = "" + + ext_code = r["extension"] + ft = ext_map.get(ext_code, "mp3") + duration_ms = int(r["total_time_ms"] or 0) + bitrate = int(r["bit_rate"] or 256) # already stored as kbps + sr = int(r["sample_rate"] or 44100) + file_size = int(r["file_size"] or 0) + + art_count = 1 if r["artwork_status"] == 1 else 0 + mhii_link = r["artwork_cache_id"] or 0 + + t = TrackInfo( + title=title, artist=artist, album=album, + album_artist=r["album_artist"] or artist, + location=loc, size=file_size, length_ms=duration_ms, + filetype=ft, bitrate=bitrate, sample_rate=sr, + year=r["year"] or 0, + track_number=r["track_number"] or 0, + track_count=r["track_count"] or 0, + disc_number=r["disc_number"] or 0, + disc_count=r["disc_count"] or 0, + art_count=art_count, art_size=0, mhii_link=mhii_link, + # db_track_id = item pid (negative values are fine for 64-bit) + db_track_id=r["pid"], + ) + tracks.append(t) + + logger.info("Loaded %d tracks from SQLite", len(tracks)) + return tracks diff --git a/src/ipod_nano7_db.py b/src/ipod_nano7_db.py index c5d060a..3737a49 100644 --- a/src/ipod_nano7_db.py +++ b/src/ipod_nano7_db.py @@ -1,1444 +1,555 @@ #!/usr/bin/env python3 """ -iPod Nano 7 Database Manager -Handles SQLite database and iTunesCDB for iPod Nano 7 (firmware 2.x+) +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 os +import sys +import hashlib +import logging +import random +import shutil +import sqlite3 import struct import time -import random -import zlib -import hashlib -import sqlite3 -import logging -import shutil -import re -from typing import Optional, Dict, Any +from typing import Any, Optional + +os.environ.setdefault("IOPENOD_HOME", "/tmp/iOpenPod") +_iop = "/tmp/iOpenPod" +if _iop not in sys.path: + sys.path.insert(0, _iop) from mutagen import File as MutagenFile from mutagen.easyid3 import EasyID3 -from mutagen.mp4 import MP4 -from mutagen.mp3 import MP3 -from artwork.extractor import extract_cover_with_fallback -from artwork.cache import cover_cache -from artwork.writer import prepare_artwork_for_track, write_artworkdb -from artwork.chunks import read_existing_artwork -from artwork.types import ArtworkEntry +from iTunesDB_Writer.mhit_writer import TrackInfo as IopTrackInfo +from iTunesDB_Writer.mhyp_writer import PlaylistInfo +from iTunesDB_Writer import write_itunesdb as iop_write_itunesdb +from SQLiteDB_Writer import write_sqlite_databases +from ipod_device import ( + DeviceCapabilities, + ChecksumType, + get_firewire_id, + capabilities_for_family_gen, + get_current_device, +) logger = logging.getLogger(__name__) -MASTER_CONTAINER_PID = 203092939621887772 # "iPod" master playlist -MUSIC_CONTAINER_PID = 4872745547565090077 # Music smart playlist +# ── constants ──────────────────────────────────────────────────────── +FILETYPE_CODES = { + "mp3": 0x4D503320, + "m4a": 0x4D344120, + "m4p": 0x4D345020, + "mp4": 0x4D503420, + "aac": 0x41414320, +} -LIBRARY_SCHEMA_SQL = """ -CREATE TABLE IF NOT EXISTS version_info ( - id INTEGER PRIMARY KEY, - major INTEGER, minor INTEGER, compatibility INTEGER DEFAULT 0, - update_level INTEGER DEFAULT 0, - device_update_level INTEGER DEFAULT 0, platform INTEGER DEFAULT 0 -); - -CREATE TABLE IF NOT EXISTS db_info ( - pid INTEGER NOT NULL PRIMARY KEY, - primary_container_pid INTEGER, - media_folder_url TEXT, audio_language INTEGER DEFAULT -1, - subtitle_language INTEGER DEFAULT -1, genius_cuid TEXT, - bib BLOB, rib BLOB -); - -CREATE TABLE IF NOT EXISTS container ( - pid INTEGER NOT NULL PRIMARY KEY, - distinguished_kind INTEGER, date_created INTEGER, - date_modified INTEGER, name TEXT, name_order INTEGER, - parent_pid INTEGER DEFAULT 0, media_kinds INTEGER, - workout_template_id INTEGER DEFAULT 0, is_hidden INTEGER, - smart_is_folder INTEGER DEFAULT 0, smart_is_dynamic INTEGER, - smart_is_filtered INTEGER, smart_is_genius INTEGER DEFAULT 0, - smart_enabled_only INTEGER DEFAULT 0, smart_is_limited INTEGER, - smart_limit_kind INTEGER, smart_limit_order INTEGER, - smart_evaluation_order INTEGER DEFAULT 1, - smart_limit_value INTEGER, smart_reverse_limit_order INTEGER, - smart_criteria BLOB, description TEXT -); - -CREATE TABLE IF NOT EXISTS genre_map ( - id INTEGER NOT NULL PRIMARY KEY, - genre TEXT NOT NULL UNIQUE, genre_order INTEGER DEFAULT 0, - is_unknown INTEGER DEFAULT 0, has_music INTEGER DEFAULT 0, - artist_count_calc INTEGER DEFAULT 0 NOT NULL, - album_count_calc INTEGER DEFAULT 0 NOT NULL, - compilation_count_calc INTEGER DEFAULT 0 NOT NULL -); - -CREATE TABLE IF NOT EXISTS location_kind_map ( - pid INTEGER PRIMARY KEY, name TEXT -); - -CREATE TABLE IF NOT EXISTS category_map ( - pid INTEGER PRIMARY KEY, name TEXT, - is_protected INTEGER, is_subscribed INTEGER, - feed_url TEXT, user_pid INTEGER -); - -CREATE TABLE IF NOT EXISTS item ( - pid INTEGER PRIMARY KEY, media_kind INTEGER, is_song INTEGER, - date_modified INTEGER, year INTEGER, is_compilation INTEGER, - is_user_disabled INTEGER, remember_bookmark INTEGER, - exclude_from_shuffle INTEGER, part_of_gapless_album INTEGER, - chosen_by_auto_fill INTEGER, artwork_status INTEGER, - artwork_cache_id INTEGER, total_time_ms INTEGER, - track_number INTEGER, track_count INTEGER, disc_number INTEGER, - disc_count INTEGER, album_pid INTEGER, artist_pid INTEGER, - genre_id INTEGER, - title TEXT, artist TEXT, album TEXT, album_artist TEXT, - sort_title TEXT, sort_artist TEXT, sort_album TEXT, - sort_album_artist TEXT, title_order INTEGER, artist_order INTEGER, - album_order INTEGER, physical_order INTEGER -); - -CREATE TABLE IF NOT EXISTS artist ( - pid INTEGER PRIMARY KEY, kind INTEGER, artwork_status INTEGER, - artwork_album_pid INTEGER, name TEXT, name_order INTEGER, - sort_name TEXT, is_unknown INTEGER, has_songs INTEGER, - has_music_videos INTEGER, has_non_compilation_tracks INTEGER, - album_count INTEGER -); - -CREATE TABLE IF NOT EXISTS album ( - pid INTEGER PRIMARY KEY, kind INTEGER, artwork_status INTEGER, - artwork_item_pid INTEGER, artist_pid INTEGER, user_rating INTEGER, - name TEXT, name_order INTEGER, all_compilations INTEGER, - feed_url TEXT, season_number INTEGER, is_unknown INTEGER, - has_songs INTEGER, has_music_videos INTEGER, sort_order INTEGER, - artist_order INTEGER, has_any_compilations INTEGER, sort_name TEXT, - artist_count_calc INTEGER, has_movies INTEGER, item_count INTEGER, - min_volume_normalization_energy INTEGER -); - -CREATE TABLE IF NOT EXISTS avformat_info ( - item_pid INTEGER, sub_id INTEGER, audio_format INTEGER, - bit_rate INTEGER, channels INTEGER, sample_rate INTEGER, - duration INTEGER, gapless_heuristic_info INTEGER, - gapless_encoding_delay INTEGER, gapless_encoding_drain INTEGER, - gapless_last_frame_resynch INTEGER, analysis_inhibit_flags INTEGER, - audio_fingerprint INTEGER, volume_normalization_energy INTEGER -); - -CREATE TABLE IF NOT EXISTS store_info ( - item_pid INTEGER PRIMARY KEY -); - -CREATE TABLE IF NOT EXISTS item_to_container ( - item_pid INTEGER, container_pid INTEGER, physical_order INTEGER, - shuffle_order INTEGER -); - -CREATE TABLE IF NOT EXISTS track_size_calc ( - kind TEXT PRIMARY KEY, size INTEGER DEFAULT 0 -); - -INSERT OR IGNORE INTO track_size_calc (kind, size) VALUES ('audio', 0); - -INSERT OR IGNORE INTO version_info (id, major, minor, compatibility, update_level, device_update_level, platform) - VALUES (1, 1, 111, 0, 0, 1104, 2); - -INSERT OR IGNORE INTO db_info (pid, primary_container_pid, media_folder_url, audio_language, subtitle_language, genius_cuid, bib, rib) - VALUES (203092939621887772, 203092939621887772, NULL, -1, -1, NULL, NULL, NULL); - -INSERT OR IGNORE INTO location_kind_map (pid, name) VALUES (1, 'MPEG audio file'); -INSERT OR IGNORE INTO location_kind_map (pid, name) VALUES (2, 'Purchased AAC audio file'); -INSERT OR IGNORE INTO location_kind_map (pid, name) VALUES (3, 'AAC audio file'); - -INSERT OR IGNORE INTO container (pid, distinguished_kind, date_created, date_modified, - name, name_order, parent_pid, media_kinds, workout_template_id, is_hidden, - smart_is_folder, smart_is_dynamic, smart_is_filtered, - smart_is_genius, smart_enabled_only, smart_is_limited, - smart_limit_kind, smart_limit_order, smart_evaluation_order, - smart_limit_value, smart_reverse_limit_order, smart_criteria, description) - VALUES (203092939621887772, 0, 0, 0, 'iPod', 100, 0, 1, 0, 1, - 0, NULL, NULL, 0, 0, NULL, NULL, NULL, 0, NULL, NULL, NULL, NULL); - -INSERT OR IGNORE INTO container (pid, distinguished_kind, date_created, date_modified, - name, name_order, parent_pid, media_kinds, workout_template_id, is_hidden, - smart_is_folder, smart_is_dynamic, smart_is_filtered, - smart_is_genius, smart_enabled_only, smart_is_limited, - smart_limit_kind, smart_limit_order, smart_evaluation_order, - smart_limit_value, smart_reverse_limit_order, smart_criteria, description) - VALUES (4872745547565090077, 4, 0, 0, 'Music', 200, 0, 1, 0, 1, - 0, 1, 1, 0, 0, NULL, NULL, NULL, 1, NULL, NULL, NULL, NULL); -""" - -DYNAMIC_SCHEMA_SQL = """ -CREATE TABLE IF NOT EXISTS item ( - item_pid INTEGER PRIMARY KEY, play_count INTEGER, - last_played_date INTEGER, skip_count INTEGER, - rating INTEGER, bookmark INTEGER -); -""" - -EXTRAS_SCHEMA_SQL = """ -CREATE TABLE IF NOT EXISTS item ( - item_pid INTEGER PRIMARY KEY, lyrics TEXT, - chapter_data BLOB, description TEXT -); -""" - -GENIUS_SCHEMA_SQL = """ -CREATE TABLE IF NOT EXISTS item ( - item_pid INTEGER PRIMARY KEY -); -""" - -LOCATIONS_SCHEMA_SQL = """ -CREATE TABLE IF NOT EXISTS location ( - item_pid INTEGER, sub_id INTEGER, base_location_id INTEGER, - location_type INTEGER, location TEXT, extension INTEGER, - kind_id INTEGER, date_created INTEGER, file_size INTEGER, - file_creator INTEGER, file_type INTEGER, - num_dir_levels_file INTEGER, num_dir_levels_lib INTEGER -); -""" +MAC_EPOCH_OFFSET = 2082844800 +MASTER_CONTAINER_PID = 203092939621887772 +MUSIC_CONTAINER_PID = 4872745547565090077 -class Nano7Database: - """Manages iPod Nano 7 SQLite database and iTunesCDB""" +# ── Helpers ────────────────────────────────────────────────────────── - def __init__(self, mountpoint: str): - self.mountpoint = mountpoint - self.itunes_dir = os.path.join(mountpoint, "iPod_Control", "iTunes") - self.sqlite_path = os.path.join( - self.itunes_dir, "iTunes Library.itlp", "Library.itdb" - ) - self.locations_path = os.path.join( - self.itunes_dir, "iTunes Library.itlp", "Locations.itdb" - ) - self.locations_cbk_path = os.path.join( - self.itunes_dir, "iTunes Library.itlp", "Locations.itdb.cbk" - ) - self.cdb_path = os.path.join(self.itunes_dir, "iTunesCDB") - self.cdb_ext_path = os.path.join(self.itunes_dir, "iTunesCDB.ext") - self.music_base = os.path.join(mountpoint, "iPod_Control", "Music") +def _to_mac_epoch(ts: int) -> int: + return ts + MAC_EPOCH_OFFSET if ts > 0 else 0 - # FireWire GUID for HASHAB signing (Nano 6G/7G) - self.firewire_id = self._detect_firewire_id() - # Extension codes as 4-byte big-endian ASCII - self.EXT_CODES = { - ".mp3": struct.unpack(">I", b"MP3 ")[0], - ".m4a": struct.unpack(">I", b"M4A ")[0], - ".aac": struct.unpack(">I", b"AAC ")[0], - ".mp4": struct.unpack(">I", b"MP4 ")[0], - } - - self._ensure_schema() - - def _ensure_schema(self) -> None: - """Create required tables in all SQLite databases on the iPod.""" - itlp_dir = os.path.join(self.itunes_dir, "iTunes Library.itlp") - os.makedirs(itlp_dir, exist_ok=True) - os.makedirs(os.path.dirname(self.locations_path), exist_ok=True) - - dbs = [ - (self.sqlite_path, LIBRARY_SCHEMA_SQL), - (self.locations_path, LOCATIONS_SCHEMA_SQL), - (os.path.join(itlp_dir, "Dynamic.itdb"), DYNAMIC_SCHEMA_SQL), - (os.path.join(itlp_dir, "Extras.itdb"), EXTRAS_SCHEMA_SQL), - (os.path.join(itlp_dir, "Genius.itdb"), GENIUS_SCHEMA_SQL), - ] - for db_path, schema in dbs: - try: - conn = sqlite3.connect(db_path) - conn.executescript(schema) - conn.commit() - conn.close() - except Exception as e: - logger.warning(f"Failed to ensure schema in {os.path.basename(db_path)}: {e}") - - def _detect_firewire_id(self) -> bytes: - """Parse FirewireGuid from SysInfo file on the iPod.""" - sysinfo_path = os.path.join( - self.mountpoint, "iPod_Control", "Device", "SysInfo" - ) - if not os.path.exists(sysinfo_path): - logger.warning("SysInfo not found, HASHAB signing disabled") - return b"" - try: - with open(sysinfo_path, 'r') as f: - for line in f: - if line.startswith('FirewireGuid:'): - hex_str = line.split(':')[1].strip() - if hex_str.startswith('0x'): - hex_str = hex_str[2:] - fwid = bytes.fromhex(hex_str) - logger.info(f"FireWire GUID: {hex_str}") - return fwid - except Exception as e: - logger.warning(f"Failed to parse FirewireGuid: {e}") - return b"" - - def get_next_pid(self) -> int: - """Generate a unique negative pid (like existing tracks)""" - return random.randint(-(2**63), -1) - - def find_free_music_dir(self) -> str: - """Find a FXX directory with the least files""" - dirs = sorted(os.listdir(self.music_base)) - min_files = float("inf") - target = None - for d in dirs: - full = os.path.join(self.music_base, d) - if os.path.isdir(full): - count = len(os.listdir(full)) - if count < min_files: - min_files = count - target = d - return target or "F00" - - def copy_track_to_ipod(self, filepath: str) -> tuple: - """Copy track to iPod_Control/Music/FXX/ and return (relative_path, full_ipod_path)""" - filename = os.path.basename(filepath) - base, ext = os.path.splitext(filename) - - # Generate unique name like pygpod does - prefix = "pygpod" if "pygpod" in filename else "" - if not prefix: - prefix = "".join(random.choices("ABCDEFGHIJKLMNOPQRSTUVWXYZ", k=4)) - - rand_num = random.randint(0, 999999) - new_name = f"{prefix}{rand_num:06d}{ext}" - - target_dir = self.find_free_music_dir() - target_path = os.path.join(self.music_base, target_dir, new_name) - os.makedirs(os.path.dirname(target_path), exist_ok=True) - shutil.copy2(filepath, target_path) - - relative_path = f"{target_dir}/{new_name}" - ipod_path = f":iPod_Control:Music:{target_dir}:{new_name}" - logger.info(f"Copied track to {ipod_path}") - return relative_path, ipod_path - - def get_audio_info(self, filepath: str) -> Dict[str, Any]: - """Extract audio information from file""" - info = {} +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: - raise ValueError(f"Cannot read audio file: {filepath}") - - # Duration in ms - if hasattr(audio, "info") and audio.info.length: + 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 - - # Bitrate if hasattr(audio.info, "bitrate"): - info["bitrate"] = audio.info.bitrate // 1000 + info["bitrate"] = audio.info.bitrate else: - info["bitrate"] = 256 - - # Channels + 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 - # Sample rate - if hasattr(audio.info, "sample_rate"): - info["sample_rate"] = audio.info.sample_rate - else: - info["sample_rate"] = 44100 - - # Audio format code (502=AAC, 1=MP3) - ext = os.path.splitext(filepath)[1].lower() - if ext in (".m4a", ".aac", ".mp4"): - info["audio_format"] = 502 # AAC - elif ext == ".mp3": - info["audio_format"] = 1 # MP3 - else: - info["audio_format"] = 1 - - # Extract tags - handle different formats - ext = os.path.splitext(filepath)[1].lower() - - # Use EasyID3 for MP3 files (supports easy key access) if ext == ".mp3": try: easy = EasyID3(filepath) - for tag_name, file_key in [ - ("title", "title"), - ("artist", "artist"), - ("album", "album"), - ("albumartist", "albumartist"), - ("composer", "composer"), - ("genre", "genre"), - ("comment", "comment"), - ]: - if tag_name in easy: - val = easy[tag_name] + 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[file_key] = str(val[0]) - - # Track/disc numbers - for tag_name, file_key in [ - ("tracknumber", "track_number"), - ("discnumber", "disc_number"), - ]: - if tag_name in easy: - val = easy[tag_name] + 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: - val = str(val[0]) - if '/' in val: - parts = val.split('/') - info[file_key] = int(parts[0]) if parts[0].isdigit() else 0 - elif val.isdigit(): - info[file_key] = int(val) - - # Date/year + 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: - date_str = str(val[0]) - if len(date_str) >= 4: - info["year"] = int(date_str[:4]) - except Exception as e: - logger.debug(f"EasyID3 failed for {filepath}: {e}") - - # MP4/M4A uses MutagenFile directly - elif ext in (".m4a", ".aac", ".mp4"): + 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: - tags = audio.tags - for tag_name, mp4_key, file_key in [ - ("title", "\xa9nam", "title"), - ("artist", "\xa9ART", "artist"), - ("album", "\xa9alb", "album"), - ]: - if mp4_key in tags: - val = tags[mp4_key] - if isinstance(val, list) and val: - info[file_key] = str(val[0]) - if "trkn" in tags: - trkn = tags["trkn"] - if isinstance(trkn, list) and trkn: - t = trkn[0] - if isinstance(t, tuple) and len(t) >= 1: - info["track_number"] = t[0] + 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: - # FLAC/OGG/Vorbis - use MutagenFile with tag access if audio and audio.tags: - tags = audio.tags - for tag_name, file_key in [ - ("title", "title"), - ("artist", "artist"), - ("album", "album"), - ("albumartist", "album_artist"), - ("composer", "composer"), - ("genre", "genre"), - ("comment", "comment"), - ]: + for tag_k, dest in [("title", "title"), ("artist", "artist"), + ("album", "album"), ("albumartist", "album_artist"), + ("genre", "genre"), ("composer", "composer")]: try: - if tag_name in tags: - val = tags[tag_name] - if isinstance(val, list) and val: - info[file_key] = str(val[0]) + 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 - - for tag_name, file_key in [ - ("tracknumber", "track_number"), - ("discnumber", "disc_number"), - ]: - try: - if tag_name in tags: - val = tags[tag_name] - if isinstance(val, list) and val: - val = str(val[0]) - if '/' in val: - parts = val.split('/') - info[file_key] = int(parts[0]) if parts[0].isdigit() else 0 - elif val.isdigit(): - info[file_key] = int(val) - except (ValueError, TypeError): - pass - - try: - if "date" in tags: - val = tags["date"] - if isinstance(val, list) and val: - date_str = str(val[0]) - if len(date_str) >= 4: - info["year"] = int(date_str[:4]) - except (ValueError, TypeError): - pass + except Exception: + pass + return info - return info - def add_track(self, filepath: str, **kwargs) -> Optional[Dict[str, Any]]: - """Add a track to the iPod Nano 7 database. - - Returns None if track already exists (duplicate detection). +def _scanned_tracks_to_iop(mountpoint: str, + scanned: list[dict]) -> list[IopTrackInfo]: + """Convert our scanned track dicts to iOpenPod TrackInfo objects.""" + tracks: list[IopTrackInfo] = [] + 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) + + t = IopTrackInfo( + title=s.get("title") or base, + location=location, + size=size, + length=s.get("duration_ms", 0), + filetype=ft, + bitrate=s.get("bitrate", 0) // 1000, + sample_rate=s.get("sample_rate", 44100), + 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=s.get("art_count", 0), + artwork_size=s.get("art_size", 0), + mhii_link=s.get("mhii_link", 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}") - # Get audio info for duplicate detection - audio_info = self.get_audio_info(filepath) - for key, value in kwargs.items(): - audio_info[key] = value + 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", "") - title = audio_info.get("title", os.path.splitext(os.path.basename(filepath))[0]) - artist = audio_info.get("artist", "Unknown Artist") - album = audio_info.get("album", "Unknown 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 - # Check for duplicate: same title + artist + album (case-insensitive) - conn = sqlite3.connect(self.sqlite_path) - cursor = conn.cursor() - cursor.execute( - "SELECT pid, title FROM item WHERE LOWER(title) = LOWER(?) AND LOWER(artist) = LOWER(?) AND LOWER(album) = LOWER(?)", - (title, artist, album) - ) - existing = cursor.fetchone() - if existing: - logger.info(f"Track already exists: '{existing[1]}' by {artist} ({album}), skipping") - conn.close() - 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) - # Copy file to iPod - relative_path, ipod_path = self.copy_track_to_ipod(filepath) + logger.info("Copied: %s → :iPod_Control:Music:%s:%s", + os.path.basename(filepath), target_dir, fname) - # Generate pid - pid = self.get_next_pid() - - # Get file size - file_size = os.path.getsize(filepath) - - # Get extension code - ext = os.path.splitext(filepath)[1].lower() - ext_code = self.EXT_CODES.get(ext, self.EXT_CODES[".m4a"]) - - # Get current max physical_order - cursor.execute("SELECT MAX(physical_order) FROM item") - max_order = cursor.fetchone()[0] or 0 - physical_order = max_order + 1 - - # Get current timestamp (mac epoch: seconds since 1904) - mac_epoch = int(time.time()) + 2082844800 - - # Check if artist exists - cursor.execute("SELECT pid FROM artist WHERE name = ?", (artist,)) - row = cursor.fetchone() - if row: - artist_pid = row[0] - else: - artist_pid = self.get_next_pid() - cursor.execute( - """ - INSERT INTO artist ( - pid, kind, artwork_status, artwork_album_pid, name, - name_order, sort_name, is_unknown, has_songs, - has_music_videos, has_non_compilation_tracks, album_count - ) VALUES (?, 2, 0, 0, ?, 100, ?, 0, 1, 0, 1, 1) - """, - (artist_pid, artist, artist), - ) - - # Check if album exists - cursor.execute("SELECT pid FROM album WHERE name = ?", (album,)) - row = cursor.fetchone() - if row: - album_pid = row[0] - else: - album_pid = self.get_next_pid() - cursor.execute( - """ - INSERT INTO album ( - pid, kind, artwork_status, artwork_item_pid, artist_pid, - user_rating, name, name_order, all_compilations, feed_url, - season_number, is_unknown, has_songs, has_music_videos, - sort_order, artist_order, has_any_compilations, sort_name, - artist_count_calc, has_movies, item_count, min_volume_normalization_energy - ) VALUES (?, 2, 0, 0, ?, 0, ?, 100, 0, NULL, 0, 0, 1, 0, 100, 100, 0, ?, 1, 0, 1, 0) - """, - (album_pid, artist_pid, album, album), - ) - - # Insert into item table - cursor.execute( - """ - INSERT INTO item ( - pid, media_kind, is_song, date_modified, year, - is_compilation, is_user_disabled, remember_bookmark, - exclude_from_shuffle, part_of_gapless_album, chosen_by_auto_fill, - artwork_status, artwork_cache_id, - total_time_ms, track_number, track_count, - disc_number, disc_count, - album_pid, artist_pid, - title, artist, album, album_artist, - sort_title, sort_artist, sort_album, sort_album_artist, - title_order, artist_order, album_order, - physical_order - ) VALUES ( - ?, 1, 1, ?, ?, 0, 0, 0, 0, 0, 0, 1, 0, - ?, ?, ?, ?, ?, - ?, ?, - ?, ?, ?, ?, - ?, ?, ?, ?, - 1000, 1000, 1000, - ? - ) - """, - ( - pid, - mac_epoch, - audio_info.get("year", 0), - audio_info["duration_ms"], - audio_info.get("track_number", 0), - audio_info.get("track_count", 0), - audio_info.get("disc_number", 0), - audio_info.get("disc_count", 0), - album_pid, - artist_pid, - title, - artist, - album, - audio_info.get("album_artist", artist), - title, - artist, - album, - audio_info.get("album_artist", artist), - physical_order, - ), - ) - - # Insert into avformat_info - # CRITICAL: duration in avformat_info is in SAMPLES, not milliseconds! - # Firmware uses this to determine when to stop playback. - # samples = duration_ms * sample_rate / 1000 - duration_samples = int(audio_info["duration_ms"] * audio_info["sample_rate"] / 1000) - cursor.execute( - """ - INSERT INTO avformat_info ( - item_pid, sub_id, audio_format, bit_rate, channels, - sample_rate, duration, gapless_heuristic_info, - gapless_encoding_delay, gapless_encoding_drain, - gapless_last_frame_resynch, analysis_inhibit_flags, - audio_fingerprint, volume_normalization_energy - ) VALUES (?, 0, ?, ?, 0, ?, ?, 1, 2112, 316, 0, 0, 0, 0) - """, - ( - pid, - audio_info["audio_format"], - audio_info["bitrate"], - audio_info["sample_rate"], - duration_samples, - ), - ) - - # Insert into store_info - cursor.execute( - """ - INSERT INTO store_info (item_pid) VALUES (?) - """, - (pid,), - ) - - # Add to master playlist (iPod container) - cursor.execute( - """ - INSERT INTO item_to_container (item_pid, container_pid, physical_order, shuffle_order) - VALUES (?, ?, ?, NULL) - """, - (pid, MASTER_CONTAINER_PID, physical_order), - ) - - # Add to Music container (pid 4872745547565090077) - music_container_pid = 4872745547565090077 - cursor.execute( - """ - INSERT INTO item_to_container (item_pid, container_pid, physical_order, shuffle_order) - VALUES (?, ?, ?, NULL) - """, - (pid, music_container_pid, physical_order), - ) - - # Add to artist album container (Music container) - - conn.commit() - conn.close() - - # Insert location record in Locations.itdb - self._insert_location(pid, relative_path, ext_code, file_size, mac_epoch) - - result = { - "pid": pid, - "title": title, - "artist": artist, - "album": album, - "ipod_path": ipod_path, - "physical_order": physical_order, - } - logger.info(f"Added track: {title} by {artist}") - - # Extract and write album art to iPod - try: - self._write_artwork_for_track(filepath, artist, album, pid) - except Exception as exc: - logger.warning("Failed to write artwork for track '%s': %s", title, exc) - - # Sync iTunesCDB so firmware accepts the modified database + # Regenerate all databases via iOpenPod self.sync_itunescdb() - # Regenerate HASHAB-signed checksums for Locations.itdb (Nano 6G/7G) - self.sync_locations_cbk() + 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} - return result + 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 _write_artwork_for_track(self, source_path: str, artist: str, album: str, item_pid: int) -> None: - """Extract cover art, encode for iPod, write to ithmb/ArtworkDB, update SQLite.""" - result = prepare_artwork_for_track(source_path) - if result is None: - return - - art_hash, encoded_formats = result - - artwork_dir = os.path.join(self.mountpoint, "iPod_Control", "Artwork") - os.makedirs(artwork_dir, exist_ok=True) - artdb_path = os.path.join(artwork_dir, "ArtworkDB") - - existing = read_existing_artwork(artdb_path, artwork_dir) - next_img_id = max(existing.keys(), default=99) + 1 - - src_img_size = sum(p.size for p in encoded_formats.values()) - - new_entry = ArtworkEntry( - img_id=next_img_id, - db_track_id=item_pid, - art_hash=art_hash, - src_img_size=src_img_size, - formats=encoded_formats, - db_track_ids=[item_pid], - ) - - all_entries = [] - for eid in sorted(existing.keys()): - e = existing[eid] - all_entries.append(ArtworkEntry( - img_id=e["img_id"], - db_track_id=e["song_id"], - art_hash=None, - src_img_size=e["src_img_size"], - formats={fid: ref for fid, ref in e["formats"].items()}, - db_track_ids=[e["song_id"]], - )) - all_entries.append(new_entry) - - write_artworkdb(self.mountpoint, all_entries, start_img_id=min(existing.keys(), default=100)) - - conn = sqlite3.connect(self.sqlite_path) - try: - conn.execute( - "UPDATE item SET artwork_cache_id = ?, artwork_status = 1 WHERE pid = ?", - (next_img_id, item_pid), - ) - conn.commit() - finally: - conn.close() + 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: - """Get number of tracks in database""" - conn = sqlite3.connect(self.sqlite_path) - cursor = conn.cursor() - cursor.execute("SELECT COUNT(*) FROM item") - count = cursor.fetchone()[0] - conn.close() - return count - - def get_tracks(self, limit: int = 10) -> list: - """Get recent tracks from database""" - conn = sqlite3.connect(self.sqlite_path) - cursor = conn.cursor() - cursor.execute( - "SELECT pid, title, artist, album, physical_order FROM item ORDER BY physical_order DESC LIMIT ?", - (limit,), - ) - tracks = cursor.fetchall() - conn.close() - return tracks - - def get_all_tracks(self) -> list: - """Get all tracks with full metadata and file paths.""" - conn = sqlite3.connect(self.sqlite_path) - conn.row_factory = sqlite3.Row - cursor = conn.cursor() - cursor.execute( - """ - SELECT pid, title, artist, album, album_artist, - total_time_ms, track_number, disc_number, - physical_order, date_modified, year - FROM item - ORDER BY physical_order - """ - ) - rows = cursor.fetchall() - conn.close() + 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 row in rows: - track = { - "pid": row["pid"], - "title": row["title"] or "Unknown", - "artist": row["artist"] or "Unknown Artist", - "album": row["album"] or "", - "album_artist": row["album_artist"] or "", - "duration_ms": row["total_time_ms"] or 0, - "track_number": row["track_number"] or 0, - "disc_number": row["disc_number"] or 0, - "physical_order": row["physical_order"] or 0, - "date_modified": row["date_modified"] or 0, - "year": row["year"] or 0, + 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": None, - } - tracks.append(track) - - # Resolve physical file paths by scanning Music/FXX/ directories - # and matching tags to tracks - track_lookup = {} - for t in tracks: - key = (t["title"].lower(), t["artist"].lower(), t["duration_ms"]) - track_lookup[key] = t - - music_base = os.path.join(self.mountpoint, "iPod_Control", "Music") - if os.path.exists(music_base): - for folder in sorted(os.listdir(music_base)): - folder_path = os.path.join(music_base, folder) - if not os.path.isdir(folder_path): - continue - for fname in os.listdir(folder_path): - fpath = os.path.join(folder_path, fname) - if not os.path.isfile(fpath): - continue - ext = os.path.splitext(fname)[1].lower() - if ext not in (".mp3", ".m4a", ".aac", ".mp4"): - continue - - # Try to match by tags - try: - audio = MutagenFile(fpath) - if audio is None or audio.tags is None: - continue - - # Get title - file_title = "" - file_artist = "" - if ext == ".mp3": - try: - easy = EasyID3(fpath) - if "title" in easy: - file_title = str(easy["title"][0]) - if "artist" in easy: - file_artist = str(easy["artist"][0]) - except Exception: - pass - elif ext in (".m4a", ".aac", ".mp4"): - from mutagen.mp4 import MP4 - tags = audio.tags - if "\xa9nam" in tags: - file_title = str(tags["\xa9nam"][0]) - if "\xa9ART" in tags: - file_artist = str(tags["\xa9ART"][0]) - else: - tags = audio.tags - try: - if "title" in tags: - val = tags["title"] - file_title = str(val[0] if isinstance(val, list) else val) - if "artist" in tags: - val = tags["artist"] - file_artist = str(val[0] if isinstance(val, list) else val) - except (ValueError, TypeError): - pass - - duration_ms = int(audio.info.length * 1000) if hasattr(audio, "info") and audio.info.length else 0 - - key = (file_title.lower(), file_artist.lower(), duration_ms) - if key in track_lookup: - track = track_lookup[key] - track["ipod_path"] = f":iPod_Control:Music:{folder}:{fname}" - track["file_path"] = fpath - del track_lookup[key] - - except Exception: - pass - - # For unmatched tracks, try to find file by ipod_path pattern - for t in tracks: - if t["file_path"]: - continue - # Try to find by scanning for any audio file in FXX dirs - for folder in sorted(os.listdir(music_base)): - folder_path = os.path.join(music_base, folder) - if not os.path.isdir(folder_path): - continue - for fname in os.listdir(folder_path): - fpath = os.path.join(folder_path, fname) - if not os.path.isfile(fpath): - continue - ext = os.path.splitext(fname)[1].lower() - if ext in (".mp3", ".m4a", ".aac", ".mp4"): - t["ipod_path"] = f":iPod_Control:Music:{folder}:{fname}" - t["file_path"] = fpath - break - if t["file_path"]: - break - + "file_path": s["file_path"], + }) return tracks def delete_track(self, pid: int) -> bool: - """Delete a track from the iPod database and remove its physical file.""" - try: - conn = sqlite3.connect(self.sqlite_path) - cursor = conn.cursor() - - # Get the track info first (for file path) - cursor.execute( - "SELECT title, artist, album FROM item WHERE pid = ?", - (pid,), - ) - row = cursor.fetchone() - if row is None: - conn.close() - logger.warning(f"Track pid={pid} not found in database") - return False - - title, artist, album = row - - # Find the physical file by matching title+artist in tags - file_path = self._find_track_file(pid, title, artist) - - # Remove from item_to_container - cursor.execute("DELETE FROM item_to_container WHERE item_pid = ?", (pid,)) - - # Remove from avformat_info - cursor.execute("DELETE FROM avformat_info WHERE item_pid = ?", (pid,)) - - # Remove from store_info - cursor.execute("DELETE FROM store_info WHERE item_pid = ?", (pid,)) - - # Remove from item - cursor.execute("DELETE FROM item WHERE pid = ?", (pid,)) - - # Clean up orphaned artist/album entries - self._cleanup_orphaned_entries(cursor) - - conn.commit() - conn.close() - - # Remove location record - self._delete_location(pid) - - # Remove physical file - if file_path and os.path.exists(file_path): - os.remove(file_path) - logger.info(f"Deleted physical file: {file_path}") - else: - logger.warning( - f"Physical file not found for pid={pid} ({artist} - {title}). " - "Use 'Clean Orphaned Files' to find and remove orphaned files." - ) - - logger.info(f"Deleted track pid={pid}: {artist} - {title}") - - # Sync iTunesCDB so firmware accepts the modified database + """Delete track by pid. Scans for matching file, deletes, re-syncs.""" + deleted = False + for s in self._scan_ipod_files(): + if s["pid"] == pid: + if os.path.exists(s["file_path"]): + os.remove(s["file_path"]) + logger.info("Deleted: %s", s["file_path"]) + deleted = True + break + if deleted: self.sync_itunescdb() - - # Regenerate HASHAB-signed checksums for Locations.itdb (Nano 6G/7G) - self.sync_locations_cbk() - return True + # Try deleting by file path if pid lookup fails + # pid might be a negative hash; try all files + logger.warning("No file found for pid=%d", pid) + return False - except Exception as e: - logger.error(f"Failed to delete track pid={pid}: {e}") - return False + def remove_duplicates(self) -> 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) - def remove_duplicates(self) -> List[Dict[str, Any]]: - """Find and remove duplicate tracks from the iPod database. - - Duplicates are identified by LOWER(title) + LOWER(artist) + LOWER(album). - Keeps the entry with the lowest physical_order (original). - - Returns list of removed track info dicts. - """ - conn = sqlite3.connect(self.sqlite_path) - cursor = conn.cursor() - - # Find duplicate groups: same title + artist + album, more than one entry - cursor.execute(""" - SELECT LOWER(title) AS lt, LOWER(artist) AS la, LOWER(album) AS lb, - COUNT(*) AS cnt, MIN(physical_order) AS keep_order - FROM item - GROUP BY LOWER(title), LOWER(artist), LOWER(album) - HAVING cnt > 1 - """) - dup_groups = cursor.fetchall() - - if not dup_groups: - conn.close() - logger.info("No duplicate tracks found") - return [] - - # Collect PIDs to remove - pids_to_remove = [] - files_to_remove = [] - removed_info = [] - - for lt, la, lb, cnt, keep_order in dup_groups: - # Get the track to keep (lowest physical_order) - cursor.execute( - "SELECT pid, title, artist, album FROM item WHERE LOWER(title)=? AND LOWER(artist)=? AND LOWER(album)=? AND physical_order=?", - (lt, la, lb, keep_order) - ) - keep_row = cursor.fetchone() - keep_pid = keep_row[0] if keep_row else None - - # Get all duplicates for this group - cursor.execute( - "SELECT pid, title, artist, album, physical_order FROM item WHERE LOWER(title)=? AND LOWER(artist)=? AND LOWER(album)=? AND pid!=? ORDER BY physical_order", - (lt, la, lb, keep_pid) - ) - dup_rows = cursor.fetchall() - - for dup_pid, dup_title, dup_artist, dup_album, dup_order in dup_rows: - pids_to_remove.append(dup_pid) - removed_info.append({ - "pid": dup_pid, - "title": dup_title, - "artist": dup_artist, - "album": dup_album, - "physical_order": dup_order, - }) - - # Find the physical file for this duplicate - file_path = self._find_track_file(dup_pid, dup_title, dup_artist) - if file_path: - files_to_remove.append(file_path) - - logger.info(f"Found {len(pids_to_remove)} duplicate entries to remove across {len(dup_groups)} groups") - - # Remove from all tables - for pid in pids_to_remove: - cursor.execute("DELETE FROM item_to_container WHERE item_pid = ?", (pid,)) - cursor.execute("DELETE FROM avformat_info WHERE item_pid = ?", (pid,)) - cursor.execute("DELETE FROM store_info WHERE item_pid = ?", (pid,)) - cursor.execute("DELETE FROM item WHERE pid = ?", (pid,)) - - # Clean up orphaned artist/album entries - self._cleanup_orphaned_entries(cursor) - - conn.commit() - - # Delete physical files - for fpath in files_to_remove: - if os.path.exists(fpath): - os.remove(fpath) - logger.info(f"Deleted duplicate file: {fpath}") - - # Update Locations.itdb - for pid in pids_to_remove: - self._delete_location(pid) - - conn.close() - - # Sync databases - self.sync_itunescdb() - self.sync_locations_cbk() - - logger.info(f"Removed {len(removed_info)} duplicate track(s)") - return removed_info - - def _find_track_file(self, pid: int, title: str, artist: str) -> Optional[str]: - """Find the physical file for a track by scanning Music/FXX/ directories.""" - music_base = os.path.join(self.mountpoint, "iPod_Control", "Music") - if not os.path.exists(music_base): - return None - - title_lower = (title or "").lower() - artist_lower = (artist or "").lower() - - for folder in sorted(os.listdir(music_base)): - folder_path = os.path.join(music_base, folder) - if not os.path.isdir(folder_path): + removed = [] + for key, items in seen.items(): + if len(items) <= 1: continue - for fname in os.listdir(folder_path): - fpath = os.path.join(folder_path, fname) - if not os.path.isfile(fpath): - continue - ext = os.path.splitext(fname)[1].lower() - if ext not in (".mp3", ".m4a", ".aac", ".mp4"): - continue + # Keep first, delete rest + 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) - try: - audio = MutagenFile(fpath) - if audio is None or audio.tags is None: - continue - - file_title = "" - file_artist = "" - if ext == ".mp3": - try: - easy = EasyID3(fpath) - if "title" in easy: - file_title = str(easy["title"][0]) - if "artist" in easy: - file_artist = str(easy["artist"][0]) - except Exception: - pass - elif ext in (".m4a", ".aac", ".mp4"): - tags = audio.tags - if "\xa9nam" in tags: - file_title = str(tags["\xa9nam"][0]) - if "\xa9ART" in tags: - file_artist = str(tags["\xa9ART"][0]) - else: - tags = audio.tags - try: - if "title" in tags: - val = tags["title"] - file_title = str(val[0] if isinstance(val, list) else val) - if "artist" in tags: - val = tags["artist"] - file_artist = str(val[0] if isinstance(val, list) else val) - except (ValueError, TypeError): - pass - - if file_title.lower() == title_lower and file_artist.lower() == artist_lower: - return fpath - except Exception: - continue - - return None - - def _cleanup_orphaned_entries(self, cursor): - """Remove artist/album records with no remaining tracks.""" - # Remove artists with no tracks referencing them - cursor.execute( - """ - DELETE FROM artist - WHERE pid NOT IN (SELECT DISTINCT artist_pid FROM item WHERE artist_pid IS NOT NULL) - """ - ) - - # Remove albums with no tracks referencing them - cursor.execute( - """ - DELETE FROM album - WHERE pid NOT IN (SELECT DISTINCT album_pid FROM item WHERE album_pid IS NOT NULL) - """ - ) - - def _insert_location(self, pid: int, relative_path: str, ext_code: int, file_size: int, mac_epoch: int): - """Insert a file location record into Locations.itdb and update track_size_calc.""" - if not os.path.exists(self.locations_path): - logger.warning(f"Locations database not found at {self.locations_path}") - return - - conn = sqlite3.connect(self.locations_path) - cursor = conn.cursor() - cursor.execute( - """ - INSERT INTO location ( - item_pid, sub_id, base_location_id, location_type, - location, extension, kind_id, date_created, file_size, - file_creator, file_type, num_dir_levels_file, num_dir_levels_lib - ) VALUES (?, 0, 1, 1179208773, ?, ?, 1, ?, ?, 0, 0, 0, 0) - """, - (pid, relative_path, ext_code, mac_epoch, file_size), - ) - conn.commit() - conn.close() - logger.info(f"Added location record: {relative_path} for pid={pid}") - - # Update total audio size in Library.itdb - conn = sqlite3.connect(self.sqlite_path) - cursor = conn.cursor() - cursor.execute("UPDATE track_size_calc SET size = size + ? WHERE kind = 'audio'", (file_size,)) - conn.commit() - conn.close() - - def _delete_location(self, pid: int): - """Remove a file location record from Locations.itdb and update track_size_calc.""" - if not os.path.exists(self.locations_path): - return - - # Get file size before deleting - conn = sqlite3.connect(self.locations_path) - cursor = conn.cursor() - cursor.execute("SELECT file_size FROM location WHERE item_pid = ?", (pid,)) - row = cursor.fetchone() - file_size = row[0] if row else 0 - - cursor.execute("DELETE FROM location WHERE item_pid = ?", (pid,)) - conn.commit() - conn.close() - - # Update total audio size in Library.itdb - if file_size > 0: - conn = sqlite3.connect(self.sqlite_path) - cursor = conn.cursor() - cursor.execute("UPDATE track_size_calc SET size = MAX(0, size - ?) WHERE kind = 'audio'", (file_size,)) - conn.commit() - conn.close() - - # ---- Helper methods for tag extraction (used by multiple callers) ---- - - def sync_itunescdb(self): - """Regenerate iTunesCDB compressed database and update hash in iTunesCDB.ext. - - Must be called after any modification to Library.itdb so the iPod firmware - accepts the database as valid. - """ - import zlib - import re - from hashab import compute_hashab - - # Read current Library.itdb - with open(self.sqlite_path, "rb") as f: - lib_data = f.read() - - # Compute new SHA1 hash of Library.itdb for iTunesCDB.ext - new_hash = hashlib.sha1(lib_data).hexdigest() - - compressed = zlib.compress(lib_data) - total_size = 244 + len(compressed) - - # Build mhfd header - header = bytearray(244) - header[0:4] = b"mhfd" - struct.pack_into("= 8: - try: - signature = compute_hashab(hashlib.sha1(itdb_data).digest(), self.firewire_id) - header[0xAB:0xAB + 57] = signature - itdb_data = bytes(header) + compressed - logger.info("HASHAB signature written to iTunesCDB header") - except Exception as e: - logger.warning("HASHAB signing failed for iTunesCDB: %s", e) - - # Write new iTunesCDB - with open(self.cdb_path, "wb") as f: - f.write(itdb_data) - - logger.info(f"Synced iTunesCDB: hash={new_hash}, compressed={len(compressed)} bytes") - - # Update hash in iTunesCDB.ext - if not os.path.exists(self.cdb_ext_path): - ext_content = f"[iTunesCDB]\nitunesdb_hash={new_hash}\n" - else: - with open(self.cdb_ext_path, "r") as f: - ext_content = f.read() - new_ext_content = re.sub(r"itunesdb_hash=\w+", f"itunesdb_hash={new_hash}", ext_content) - with open(self.cdb_ext_path, "w") as f: - f.write(new_ext_content) - - logger.info(f"Updated iTunesCDB.ext: hash={new_hash}") - - def sync_locations_cbk(self): - """Regenerate Locations.itdb.cbk with HASHAB-signed block checksums. - - Nano 6G/7G firmware requires this file to accept Locations.itdb. - Without valid cbk, the device will ignore location records. - - File format: - [57 bytes] HASHAB signature of final_sha1 (Nano 6G/7G) - [20 bytes] final_sha1 = SHA1(all block SHA1s concatenated) - [Nx20 bytes] SHA1 of each 1024-byte block of Locations.itdb - """ - if not os.path.exists(self.locations_path): - logger.warning("Locations.itdb not found, skipping cbk generation") - return - - if not self.firewire_id or len(self.firewire_id) < 8: - logger.warning("No FireWire GUID available, cbk generation disabled") - return - - BLOCK_SIZE = 1024 - - # Read Locations.itdb - with open(self.locations_path, 'rb') as f: - locations_data = f.read() - - # Compute SHA1 of each 1024-byte block - block_sha1s = [] - offset = 0 - while offset < len(locations_data): - block = locations_data[offset:offset + BLOCK_SIZE] - block_sha1s.append(hashlib.sha1(block).digest()) - offset += BLOCK_SIZE - - # Compute final SHA1 = SHA1(concatenation of all block SHA1s) - all_sha1s = b''.join(block_sha1s) - final_sha1 = hashlib.sha1(all_sha1s).digest() - - logger.debug( - "Locations.itdb: %d bytes, %d blocks, final_sha1: %s", - len(locations_data), len(block_sha1s), final_sha1.hex() - ) - - # Compute 57-byte HASHAB signature - try: - from hashab import compute_hashab - header = compute_hashab(final_sha1, self.firewire_id[:8]) - if len(header) != 57: - logger.error(f"HASHAB returned {len(header)} bytes, expected 57") - return - except Exception as e: - logger.error(f"Failed to compute HASHAB signature: {e}") - return - - # Write cbk file: header + final_sha1 + block_sha1s - with open(self.locations_cbk_path, 'wb') as f: - f.write(header) - f.write(final_sha1) - for bsha1 in block_sha1s: - f.write(bsha1) - - total_size = len(header) + 20 + len(block_sha1s) * 20 - logger.info( - "Wrote Locations.itdb.cbk: %d bytes " - "(57-byte HASHAB + 20-byte final SHA1 + %d block SHA1s)", - total_size, len(block_sha1s) - ) - - # ---- Helper methods for tag extraction (used by multiple callers) ---- - - def _get_file_tags(self, fpath: str) -> tuple: - """Extract title and artist from an audio file's tags. - - Returns: - (title, artist) tuple, both strings (may be empty) - """ - title = "" - artist = "" - ext = os.path.splitext(fpath)[1].lower() - - try: - audio = MutagenFile(fpath) - if audio is None: - return title, artist - - if ext == ".mp3": - try: - easy = EasyID3(fpath) - if "title" in easy: - title = str(easy["title"][0]) - if "artist" in easy: - artist = str(easy["artist"][0]) - except Exception: - pass - elif ext in (".m4a", ".aac", ".mp4"): - from mutagen.mp4 import MP4 - tags = audio.tags - if "\xa9nam" in tags: - title = str(tags["\xa9nam"][0]) - if "\xa9ART" in tags: - artist = str(tags["\xa9ART"][0]) - else: - # FLAC, OGG, etc. - tags = audio.tags - if tags: - try: - if "title" in tags: - val = tags["title"] - title = str(val[0] if isinstance(val, list) else val) - if "artist" in tags: - val = tags["artist"] - artist = str(val[0] if isinstance(val, list) else val) - except (ValueError, TypeError): - pass - except Exception: - pass - - return title, artist - - # ---- Orphaned file detection and cleanup ---- + if removed: + self.sync_itunescdb() + logger.info("Removed %d duplicate(s)", len(removed)) + return removed def get_orphaned_files(self) -> list[dict]: - """Find audio files on disk that have no corresponding DB entry. - - Returns: - List of dicts with 'file_path', 'size', 'title', 'artist' for each orphaned file. - """ - music_base = os.path.join(self.mountpoint, "iPod_Control", "Music") - if not os.path.exists(music_base): - return [] - - # Get all (title, artist) pairs from the database (case-insensitive) - conn = sqlite3.connect(self.sqlite_path) - cursor = conn.cursor() - cursor.execute("SELECT LOWER(title), LOWER(artist) FROM item") - db_pairs = {(r[0] or "", r[1] or "") for r in cursor.fetchall()} - conn.close() - + """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 = [] - for folder in sorted(os.listdir(music_base)): - folder_path = os.path.join(music_base, folder) - if not os.path.isdir(folder_path): + 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(folder_path): - fpath = os.path.join(folder_path, fname) + 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"): - continue - - file_title, file_artist = self._get_file_tags(fpath) - key = (file_title.lower(), file_artist.lower()) - - # If the file's title+artist pair is NOT in the DB, it's orphaned - if key not in db_pairs: - size = os.path.getsize(fpath) + if ext not in (".mp3", ".m4a", ".aac", ".mp4", ".m4p"): orphans.append({ "file_path": fpath, - "size": size, - "title": file_title or fname, - "artist": file_artist or "Unknown", + "size": os.path.getsize(fpath), + "title": fname, + "artist": "Unknown", }) - return orphans def delete_orphaned_files(self, file_paths: list[str] = None) -> int: - """Delete orphaned files from disk. - - Args: - file_paths: List of file paths to delete. If None, delete all orphaned files. - - Returns: - Number of files successfully deleted. - """ + deleted = 0 if file_paths is None: orphans = self.get_orphaned_files() file_paths = [o["file_path"] for o in orphans] - - deleted = 0 - for fpath in file_paths: - try: - if os.path.exists(fpath): - os.remove(fpath) - logger.info(f"Deleted orphaned file: {fpath}") - deleted += 1 - except Exception as e: - logger.error(f"Failed to delete orphaned file {fpath}: {e}") - + 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. + + Scans all audio files on iPod, builds correct TrackInfo list, + writes SQLite databases (Library/Locations/Dynamic/Extras/Genius), + writes binary iTunesDB, writes iTunesCDB.ext and Locations.cbk. + """ + # 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. Convert to iOpenPod TrackInfo + tracks = _scanned_tracks_to_iop(self.mountpoint, scanned) + logger.info("Built %d iOpenPod TrackInfo objects", len(tracks)) + + # 3. Determine capabilities for this device + capabilities = None + try: + dev = get_current_device() + if dev and dev.model_family: + capabilities = 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 = get_firewire_id(self.mountpoint) + except Exception: + firewire_id = self.firewire_id + + 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: + from ipod_device import resolve_itdb_path + cdb_path = 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.""" + empty_tracks: list[IopTrackInfo] = [] + try: + fwid = get_firewire_id(self.mountpoint) + except Exception: + fwid = self.firewire_id + capabilities = None + try: + dev = get_current_device() + if dev and dev.model_family: + capabilities = capabilities_for_family_gen( + dev.model_family, dev.generation or "") + except Exception: + pass + iop_write_itunesdb(self.mountpoint, empty_tracks, firewire_id=fwid, + backup=True, capabilities=capabilities) + write_sqlite_databases( + ipod_path=self.mountpoint, tracks=empty_tracks, + master_playlist_name="iPod", + db_pid=random.getrandbits(64), + capabilities=capabilities, firewire_id=fwid, backup=True, + )