#!/usr/bin/env python3 """ iPod Nano 7 Database Manager Handles SQLite database and iTunesCDB for iPod Nano 7 (firmware 2.x+) """ import os 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 from mutagen.easyid3 import EasyID3 from mutagen.mp4 import MP4 from mutagen.mp3 import MP3 logger = logging.getLogger(__name__) MASTER_CONTAINER_PID = 203092939621887772 # "iPod" master playlist class Nano7Database: """Manages iPod Nano 7 SQLite database and iTunesCDB""" 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") # 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) 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) 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 = {} 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: 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 else: info["bitrate"] = 256 # Channels 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] 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] 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 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"): 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] 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"), ]: try: if tag_name in tags: val = tags[tag_name] if isinstance(val, list) and val: info[file_key] = str(val[0]) 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 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). """ 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 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 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}") # 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: """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() 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, "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 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 self.sync_itunescdb() # Regenerate HASHAB-signed checksums for Locations.itdb (Nano 6G/7G) self.sync_locations_cbk() return True except Exception as e: logger.error(f"Failed to delete track pid={pid}: {e}") return False 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): 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: 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 # 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