From 11c40c8a99c46af243eae188ae01f85060e2e71d Mon Sep 17 00:00:00 2001 From: Maksim Totmin Date: Sun, 31 May 2026 14:39:04 +0700 Subject: [PATCH] Fix Nano 7 database: duration samples, HASHAB cbk, duplicate detection, Locations.itdb - Fix avformat_info duration: store in SAMPLES not ms (firmware uses this to determine playback length; wrong units caused tracks to stop at ~3-11 seconds) - Add Locations.itdb insert/delete for each track (ipod_path mapping) - Add sync_locations_cbk() with HASHAB signing via WASM module - Detect FirewireGuid from SysInfo for cbk header signature - Add duplicate detection in add_track (title+artist match, skip if exists) - Add get_orphaned_files() and delete_orphaned_files() for cleanup - Sync cbk after every add/delete operation --- src/ipod_nano7_db.py | 431 ++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 406 insertions(+), 25 deletions(-) diff --git a/src/ipod_nano7_db.py b/src/ipod_nano7_db.py index f969d8c..de64609 100644 --- a/src/ipod_nano7_db.py +++ b/src/ipod_nano7_db.py @@ -9,9 +9,11 @@ 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 mutagen import File as MutagenFile @@ -29,14 +31,53 @@ class Nano7Database: def __init__(self, mountpoint: str): self.mountpoint = mountpoint - self.sqlite_path = os.path.join( - mountpoint, "iPod_Control", "iTunes", "iTunes Library.itlp", "Library.itdb" - ) 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") + # 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], + } + + 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) @@ -55,8 +96,8 @@ class Nano7Database: target = d return target or "F00" - def copy_track_to_ipod(self, filepath: str) -> str: - """Copy track to iPod_Control/Music/FXX/ and return iPod path""" + 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) @@ -72,10 +113,10 @@ class Nano7Database: target_path = os.path.join(self.music_base, target_dir, new_name) shutil.copy2(filepath, target_path) - # Return iPod path format (colon-separated) + 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 ipod_path + return relative_path, ipod_path def get_audio_info(self, filepath: str) -> Dict[str, Any]: """Extract audio information from file""" @@ -232,27 +273,50 @@ class Nano7Database: return info - def add_track(self, filepath: str, **kwargs) -> Dict[str, Any]: - """Add a track to the iPod Nano 7 database""" + 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). + """ if not os.path.exists(filepath): raise FileNotFoundError(f"File not found: {filepath}") - # Copy file to iPod - ipod_path = self.copy_track_to_ipod(filepath) - - # Get audio info + # Get audio info for duplicate detection audio_info = self.get_audio_info(filepath) - - # Apply overrides for key, value in kwargs.items(): audio_info[key] = value + 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") + + # Check for duplicate: same title + artist + album + conn = sqlite3.connect(self.sqlite_path) + cursor = conn.cursor() + cursor.execute( + "SELECT pid, title FROM item WHERE title = ? AND artist = ?", + (title, artist) + ) + existing = cursor.fetchone() + if existing: + logger.info(f"Track already exists: '{existing[1]}' by {artist}, skipping") + conn.close() + return None + + # Copy file to iPod + relative_path, ipod_path = self.copy_track_to_ipod(filepath) + # 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 - conn = sqlite3.connect(self.sqlite_path) - cursor = conn.cursor() cursor.execute("SELECT MAX(physical_order) FROM item") max_order = cursor.fetchone()[0] or 0 physical_order = max_order + 1 @@ -260,11 +324,6 @@ class Nano7Database: # Get current timestamp (mac epoch: seconds since 1904) mac_epoch = int(time.time()) + 2082844800 - # Get or create artist/album entries - artist = audio_info.get("artist", "Unknown Artist") - album = audio_info.get("album", "Unknown Album") - title = audio_info.get("title", os.path.splitext(os.path.basename(filepath))[0]) - # Check if artist exists cursor.execute("SELECT pid FROM artist WHERE name = ?", (artist,)) row = cursor.fetchone() @@ -352,6 +411,10 @@ class Nano7Database: ) # 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 ( @@ -360,15 +423,14 @@ class Nano7Database: gapless_encoding_delay, gapless_encoding_drain, gapless_last_frame_resynch, analysis_inhibit_flags, audio_fingerprint, volume_normalization_energy - ) VALUES (?, 0, ?, ?, ?, ?, ?, 1, 2112, 316, 0, 0, 0, 0) + ) VALUES (?, 0, ?, ?, 0, ?, ?, 1, 2112, 316, 0, 0, 0, 0) """, ( pid, audio_info["audio_format"], audio_info["bitrate"], - audio_info["channels"], audio_info["sample_rate"], - audio_info["duration_ms"], + duration_samples, ), ) @@ -404,6 +466,9 @@ class Nano7Database: 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, @@ -413,6 +478,13 @@ class Nano7Database: "physical_order": physical_order, } logger.info(f"Added track: {title} by {artist}") + + # Sync iTunesCDB so firmware accepts the modified database + self.sync_itunescdb() + + # Regenerate HASHAB-signed checksums for Locations.itdb (Nano 6G/7G) + self.sync_locations_cbk() + return result def get_track_count(self) -> int: @@ -605,12 +677,27 @@ class Nano7Database: 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 + self.sync_itunescdb() + + # Regenerate HASHAB-signed checksums for Locations.itdb (Nano 6G/7G) + self.sync_locations_cbk() + return True except Exception as e: @@ -696,3 +783,297 @@ class Nano7Database: 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 + + # Read current Library.itdb + with open(self.sqlite_path, "rb") as f: + lib_data = f.read() + + # Compute new SHA1 hash + new_hash = hashlib.sha1(lib_data).hexdigest() + + # Read existing iTunesCDB to extract the header (first 244 bytes) + with open(self.cdb_path, "rb") as f: + cdb_data = f.read() + + # Find zlib header (0x78 0x9c is most common, but also check 0x78 0xda and 0x78 0x01) + zlib_offset = -1 + for pattern in [b"\x78\x9c", b"\x78\xda", b"\x78\x01", b"\x78\x5e"]: + zlib_offset = cdb_data.find(pattern) + if zlib_offset >= 0: + break + if zlib_offset == -1: + logger.error("Could not find zlib header in iTunesCDB") + return + + header = cdb_data[:zlib_offset] + + # Compress Library.itdb + compressed = zlib.compress(lib_data) + + # Write new iTunesCDB + with open(self.cdb_path, "wb") as f: + f.write(header + compressed) + + # Update hash in iTunesCDB.ext + 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"Synced iTunesCDB: hash={new_hash}, compressed={len(compressed)} bytes") + + 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 ---- + + 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() + + orphans = [] + 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 + + 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) + orphans.append({ + "file_path": fpath, + "size": size, + "title": file_title or fname, + "artist": file_artist or "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. + """ + 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}") + + return deleted