en2zmax/neo-pod-desktop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
neo-pod-desktop/src/ipod_nano7_db.py
Maksim Totmin fba3fe7d05 Add HASHAB signing to iTunesCDB header + fix sync_itunescdb 
- Set hashing_scheme=3 at offset 0x30 (iPod Nano 7G requirement)
- Compute and write HASHAB signature at offset 0xAB
- Import compute_hashab from hashab module
- iTunesCDB.ext creation restored (was removed accidentally)
- 0-byte iTunesDB placeholder (firmware expects this file)
2026-05-31 19:10:44 +07:00

1445 lines
55 KiB
Python
Raw Blame History

#!/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
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
logger = logging.getLogger(__name__)
MASTER_CONTAINER_PID = 203092939621887772 # "iPod" master playlist
MUSIC_CONTAINER_PID = 4872745547565090077 # Music smart playlist
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
);
"""
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],
}
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 = {}
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 (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 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}")
# 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
self.sync_itunescdb()
# Regenerate HASHAB-signed checksums for Locations.itdb (Nano 6G/7G)
self.sync_locations_cbk()
return result
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 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 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):
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
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("<I", header, 4, 244)
struct.pack_into("<I", header, 8, total_size)
struct.pack_into("<I", header, 12, 0x00020002)
# Set hashing_scheme = 3 (HASHAB) at offset 0x30
struct.pack_into("<H", header, 0x30, 3)
# Build full iTunesCDB (header + compressed payload)
itdb_data = bytes(header) + compressed
# Compute HASHAB signature and write it at offset 0xAB
if self.firewire_id and len(self.firewire_id) >= 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 ----
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
Reference in New Issue View Git Blame Copy Permalink