Compare commits
3 Commits
fba3fe7d05
...
0f12c4985f
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0f12c4985f | ||
|
|
4f7bd2709c | ||
|
|
4c4a19640d |
@ -543,9 +543,10 @@ class IPodDevice:
|
||||
except ImportError:
|
||||
self.logger.warning("ipod_nano7_db not available, using simple file copy")
|
||||
except Exception as e:
|
||||
self.logger.error(f"Nano7 database transfer failed: {e}, falling back to simple copy")
|
||||
self.logger.error(f"Nano7 database transfer failed: {e}")
|
||||
return False # Don't fall back — file was already copied to F00/
|
||||
|
||||
# Fallback: simple file copy (won't show in iPod interface)
|
||||
# Fallback: simple file copy — only on ImportError
|
||||
if relative_path:
|
||||
dest_dir = os.path.join(self.mount_point, relative_path)
|
||||
else:
|
||||
|
||||
982
src/ipod_mhbd_writer.py
Normal file
982
src/ipod_mhbd_writer.py
Normal file
@ -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("<I", buf, offset + 4, header_len)
|
||||
struct.pack_into("<I", buf, offset + 8, total_or_count)
|
||||
|
||||
|
||||
def _gen_id() -> 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("<IIII", 1, string_len, 1, 0)
|
||||
return header + sub + string_data
|
||||
|
||||
|
||||
def _write_mhod100_position(position: int) -> 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("<I", position) + b"\x00" * 16
|
||||
return header + body
|
||||
|
||||
|
||||
def _write_playlist_prefs_mhod100() -> 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("<I", data, 0x18, 0)
|
||||
struct.pack_into("<I", data, 0x1C, 0)
|
||||
struct.pack_into("<I", data, 0x20, 0)
|
||||
struct.pack_into("<I", data, 0x24, 0)
|
||||
struct.pack_into("<I", data, 0x28, 0)
|
||||
struct.pack_into("<I", data, 0x2C, 0)
|
||||
struct.pack_into("<I", data, 0x30, 0x010084)
|
||||
struct.pack_into("<I", data, 0x34, 0x05)
|
||||
struct.pack_into("<I", data, 0x38, 0x09)
|
||||
struct.pack_into("<I", data, 0x3C, 0x03)
|
||||
struct.pack_into("<I", data, 0x40, 0x120001)
|
||||
struct.pack_into("<I", data, 0x44, 0)
|
||||
struct.pack_into("<I", data, 0x48, 0)
|
||||
struct.pack_into("<I", data, 0x4C, 0x640014)
|
||||
struct.pack_into("<I", data, 0x50, 0x01)
|
||||
struct.pack_into("<I", data, 0x54, 0)
|
||||
struct.pack_into("<I", data, 0x58, 0)
|
||||
struct.pack_into("<I", data, 0x5C, 0x320014)
|
||||
struct.pack_into("<I", data, 0x60, 0x01)
|
||||
struct.pack_into("<I", data, 0x64, 0)
|
||||
struct.pack_into("<I", data, 0x68, 0)
|
||||
struct.pack_into("<I", data, 0x6C, 0x5a0014)
|
||||
struct.pack_into("<I", data, 0x70, 0x01)
|
||||
struct.pack_into("<I", data, 0x74, 0)
|
||||
struct.pack_into("<I", data, 0x78, 0)
|
||||
struct.pack_into("<I", data, 0x7C, 0x500014)
|
||||
struct.pack_into("<I", data, 0x80, 0x01)
|
||||
struct.pack_into("<I", data, 0x84, 0)
|
||||
struct.pack_into("<I", data, 0x88, 0)
|
||||
struct.pack_into("<I", data, 0x8C, 0x7d0015)
|
||||
struct.pack_into("<I", data, 0x90, 0x01)
|
||||
return bytes(data)
|
||||
|
||||
|
||||
def _item_mk_sort_indicators(item: dict) -> 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("<I", header, 0x0C, mhod_count) # child_count
|
||||
struct.pack_into("<I", header, 0x10, track_id) # track_id
|
||||
struct.pack_into("<I", header, 0x14, 1) # visible
|
||||
struct.pack_into("<I", header, 0x18, track.filetype_code) # filetype
|
||||
struct.pack_into("B", header, 0x1C, 1 if track.vbr else 0) # vbr_flag
|
||||
struct.pack_into("B", header, 0x1D, 1 if ft == "mp3" else 0) # mp3_flag
|
||||
struct.pack_into("B", header, 0x1E, 1 if track.compilation else 0)
|
||||
struct.pack_into("B", header, 0x1F, track.rating) # rating 0-100
|
||||
struct.pack_into("<I", header, 0x20, mac_ts) # last_modified
|
||||
struct.pack_into("<I", header, 0x24, track.size) # size
|
||||
struct.pack_into("<I", header, 0x28, track.length_ms) # length (ms)
|
||||
struct.pack_into("<I", header, 0x2C, track.track_number)
|
||||
struct.pack_into("<I", header, 0x30, track.track_count)
|
||||
struct.pack_into("<I", header, 0x34, track.year)
|
||||
struct.pack_into("<I", header, 0x38, track.bitrate)
|
||||
struct.pack_into("<I", header, 0x3C, _sample_rate_to_fixed(track.sample_rate))
|
||||
struct.pack_into("<i", header, 0x40, track.volume) # signed, -255..255
|
||||
struct.pack_into("<I", header, 0x44, 0) # start_time
|
||||
struct.pack_into("<I", header, 0x48, 0) # stop_time
|
||||
struct.pack_into("<I", header, 0x4C, 0) # sound_check
|
||||
struct.pack_into("<I", header, 0x50, track.play_count)
|
||||
struct.pack_into("<I", header, 0x54, 0) # play_count_2
|
||||
struct.pack_into("<I", header, 0x58, 0) # last_played
|
||||
struct.pack_into("<I", header, 0x5C, track.disc_number)
|
||||
struct.pack_into("<I", header, 0x60, track.disc_count)
|
||||
struct.pack_into("<I", header, 0x64, 0) # user_id
|
||||
struct.pack_into("<I", header, 0x68, mac_added) # date_added
|
||||
struct.pack_into("<I", header, 0x6C, 0) # bookmark_time
|
||||
struct.pack_into("<Q", header, 0x70, track.db_track_id) # db_track_id
|
||||
struct.pack_into("B", header, 0x78, 0) # checked_flag
|
||||
struct.pack_into("B", header, 0x79, 0) # app_rating
|
||||
struct.pack_into("<H", header, 0x7A, 0) # bpm
|
||||
struct.pack_into("<H", header, 0x7C, track.art_count) # artwork_count
|
||||
struct.pack_into("<H", header, 0x7E, track.audio_format_flag)
|
||||
struct.pack_into("<I", header, 0x80, track.art_size) # artwork_size
|
||||
struct.pack_into("<I", header, 0x84, 0) # unk0x84
|
||||
struct.pack_into("<f", header, 0x88, float(track.sample_rate)) # sample_rate_2
|
||||
struct.pack_into("<I", header, 0x8C, 0) # date_released
|
||||
struct.pack_into("<H", header, 0x90, track.mpeg_audio_type)
|
||||
struct.pack_into("B", header, 0x92, track.explicit)
|
||||
struct.pack_into("B", header, 0x93, 1 if track.purchased_aac else 0)
|
||||
struct.pack_into("<I", header, 0x94, 0) # unk0x94
|
||||
struct.pack_into("<I", header, 0x98, 0) # genius_category_id
|
||||
# Extended fields
|
||||
struct.pack_into("<I", header, 0x9C, 0) # skip_count
|
||||
struct.pack_into("<I", header, 0xA0, 0) # last_skipped
|
||||
struct.pack_into("B", header, 0xA4, 1 if track.art_count > 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("<Q", header, 0xA8, track.db_track_id) # db_track_id_2
|
||||
struct.pack_into("B", header, 0xB0, 0) # lyrics_flag
|
||||
struct.pack_into("B", header, 0xB1, 0) # movie_flag
|
||||
struct.pack_into("B", header, 0xB2, 1 if track.play_count > 0 else 2) # played/not
|
||||
struct.pack_into("B", header, 0xB3, 0) # unk0xB3
|
||||
struct.pack_into("<I", header, 0xB4, 0) # unk0xB4
|
||||
struct.pack_into("<I", header, 0xB8, 0) # pregap
|
||||
struct.pack_into("<Q", header, 0xBC, 0) # sample_count
|
||||
struct.pack_into("<I", header, 0xC4, 0) # unk0xC4
|
||||
struct.pack_into("<I", header, 0xC8, 0) # postgap
|
||||
struct.pack_into("<I", header, 0xCC, 0) # encoder
|
||||
struct.pack_into("<I", header, 0xD0, 1) # media_type (audio)
|
||||
struct.pack_into("<I", header, 0xD4, 0) # season_number
|
||||
struct.pack_into("<I", header, 0xD8, 0) # episode_number
|
||||
struct.pack_into("<I", header, 0xDC, 0) # date_added_to_itunes
|
||||
struct.pack_into("<I", header, 0xE0, 0) # store_track_id
|
||||
struct.pack_into("<I", header, 0xE4, 0) # store_encoder_version
|
||||
struct.pack_into("<I", header, 0xE8, 0) # store_artist_id
|
||||
struct.pack_into("<I", header, 0xEC, 0) # unk0xEC
|
||||
struct.pack_into("<I", header, 0xF0, 0) # store_album_id
|
||||
struct.pack_into("<I", header, 0xF4, 0) # store_content_flag
|
||||
struct.pack_into("<I", header, 0xF8, 0) # gapless_payload_size
|
||||
struct.pack_into("<I", header, 0xFC, 0) # unk0xFC
|
||||
struct.pack_into("<H", header, 0x100, 0) # gapless_track_flag
|
||||
struct.pack_into("<H", header, 0x102, 0) # gapless_album_flag
|
||||
# hash_0x104: 20 bytes at 0x104 → 0x118
|
||||
struct.pack_into("<I", header, 0x118, 0) # unk0x118
|
||||
struct.pack_into("<I", header, 0x11C, 0) # unk0x11C
|
||||
struct.pack_into("<I", header, 0x120, track.album_id) # album_id
|
||||
struct.pack_into("<Q", header, 0x124, db_id_2) # db_id_2_ref
|
||||
struct.pack_into("<I", header, 0x12C, track.size) # size_2
|
||||
struct.pack_into("<I", header, 0x130, 0) # unk0x130
|
||||
header[0x134:0x13C] = _item_mk_sort_indicators(
|
||||
{"sort_title": track.sort_title, "sort_album": track.sort_album,
|
||||
"sort_artist": track.sort_artist, "sort_album_artist": track.sort_album_artist})
|
||||
# 0x13C..0x15F: zero padding
|
||||
struct.pack_into("<I", header, 0x160, track.mhii_link) # artwork_id_ref
|
||||
struct.pack_into("<I", header, 0x168, 1) # unk0x168
|
||||
# 0x16C..0x1DF: zero padding
|
||||
struct.pack_into("<I", header, 0x1E0, track.artist_id) # artist_id_ref
|
||||
# 0x1E4..0x1F3: zero padding
|
||||
struct.pack_into("<I", header, 0x1F4, 0) # composer_id
|
||||
|
||||
return bytes(header) + mhod_data
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MHLT writer
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _write_mhlt(tracks, start_track_id: int, db_id_2: int) -> 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("<I", header, 0x0C, 1) # child_count
|
||||
struct.pack_into("<H", header, 0x10, pod_group_flag) # podcast_group_flag
|
||||
struct.pack_into("<H", header, 0x12, 0) # unk0x12
|
||||
struct.pack_into("<I", header, 0x14, mhip_id) # group_id
|
||||
struct.pack_into("<I", header, 0x18, track_id) # track_id
|
||||
struct.pack_into("<I", header, 0x1C, 0) # timestamp
|
||||
struct.pack_into("<I", header, 0x20, pod_group_ref) # group_id_ref
|
||||
struct.pack_into("<Q", header, 0x2C, track_db_id) # track_persistent_id
|
||||
struct.pack_into("<Q", header, 0x3C, mhip_persist_id) # mhip_persistent_id
|
||||
return bytes(header) + mhod_pos
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MHYP writer — master playlist
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _write_master_playlist(track_ids: list[int], db_id_2: int,
|
||||
playlist_id: int = 0) -> 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("<I", header, 0x0C, mhod_count)
|
||||
struct.pack_into("<I", header, 0x10, mhip_count)
|
||||
struct.pack_into("B", header, 0x14, 1) # master_flag
|
||||
struct.pack_into("B", header, 0x15, 0)
|
||||
struct.pack_into("B", header, 0x16, 0)
|
||||
struct.pack_into("B", header, 0x17, 0)
|
||||
struct.pack_into("<I", header, 0x18, _to_mac_epoch(ts))
|
||||
struct.pack_into("<Q", header, 0x1C, pid) # playlist_id
|
||||
struct.pack_into("<I", header, 0x24, 0) # unk0x24
|
||||
struct.pack_into("<H", header, 0x28, 1) # string_mhod_child_count
|
||||
struct.pack_into("<H", header, 0x2A, 0) # podcast_flag
|
||||
struct.pack_into("<I", header, 0x2C, 5) # sort_order (artist)
|
||||
# Extended fields (master=True skips db_id_2/playlist_id_2 per libgpod)
|
||||
struct.pack_into("<I", header, 0x58, _to_mac_epoch(ts)) # timestamp_2
|
||||
|
||||
return bytes(header) + mhod_title + mhod_prefs + mhips
|
||||
|
||||
|
||||
def _write_empty_smart_mhlp() -> 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("<I", data, 32, 0) # limitValue
|
||||
struct.pack_into("B", data, 36, 0) # matchCheckedOnly
|
||||
struct.pack_into("B", data, 37, 0) # reverseSort
|
||||
return bytes(data)
|
||||
|
||||
|
||||
def _write_slst_empty() -> 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("<I", header, 0x0C, mhod_count)
|
||||
struct.pack_into("<I", header, 0x10, mhip_count)
|
||||
struct.pack_into("B", header, 0x14, 1) # master=True (built-in category for ds5)
|
||||
struct.pack_into("B", header, 0x15, 0)
|
||||
struct.pack_into("B", header, 0x16, 0)
|
||||
struct.pack_into("B", header, 0x17, 0)
|
||||
struct.pack_into("<I", header, 0x18, _to_mac_epoch(ts))
|
||||
playlist_id = _gen_id()
|
||||
struct.pack_into("<Q", header, 0x1C, playlist_id)
|
||||
struct.pack_into("<I", header, 0x24, 0) # unk0x24
|
||||
struct.pack_into("<H", header, 0x28, 1) # string_mhod_child_count
|
||||
struct.pack_into("<H", header, 0x2A, 0) # podcast_flag
|
||||
struct.pack_into("<I", header, 0x2C, 5) # sort_order (artist)
|
||||
# mhsd5_type = 4 (Music) at offsets 0x50, 0x52
|
||||
struct.pack_into("<H", header, 0x50, 4) # mhsd5_type
|
||||
struct.pack_into("<H", header, 0x52, 4) # mhsd5_type_2
|
||||
struct.pack_into("<I", header, 0x58, _to_mac_epoch(ts))
|
||||
|
||||
return (bytes(header) + mhod_title + mhod_prefs +
|
||||
mhod_splpref + mhod_slst)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MHSD writer
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _write_mhsd(dataset_type: int, child_data: bytes) -> 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("<I", header, 0x0C, dataset_type)
|
||||
return bytes(header) + child_data
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MHIA / MHLA writer (albums)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _write_mhia(album_id: int, album_name: str, album_artist: str) -> 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("<I", header, 0x0C, child_count)
|
||||
struct.pack_into("<I", header, 0x10, album_id)
|
||||
struct.pack_into("<Q", header, 0x14, _gen_id())
|
||||
struct.pack_into("<H", header, 0x1C, 2)
|
||||
struct.pack_into("<H", header, 0x1E, 0)
|
||||
return bytes(header) + bytes(children)
|
||||
|
||||
|
||||
def _write_mhla(tracks) -> 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("<I", header, 0x0C, child_count)
|
||||
struct.pack_into("<I", header, 0x10, artist_id)
|
||||
struct.pack_into("<Q", header, 0x14, _gen_id())
|
||||
struct.pack_into("<I", header, 0x1C, 2)
|
||||
return bytes(header) + bytes(children)
|
||||
|
||||
|
||||
def _write_mhli(tracks) -> 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("<I", header, 0x0C, 1) # compressed (1=uncompressed, 2=compressed)
|
||||
struct.pack_into("<I", header, 0x10, 0x4F) # version (iTunes 10+)
|
||||
struct.pack_into("<I", header, 0x14, child_count)
|
||||
struct.pack_into("<Q", header, 0x18, db_id)
|
||||
struct.pack_into("<H", header, 0x20, 2) # platform (Windows)
|
||||
struct.pack_into("<H", header, 0x22, 611) # unk0x22
|
||||
struct.pack_into("<Q", header, 0x24, db_id_2)
|
||||
struct.pack_into("<I", header, 0x2C, 0) # unk0x2c
|
||||
# hashing_scheme = 4 (HASHAB) — set BEFORE hash computation
|
||||
struct.pack_into("<H", header, 0x30, 4)
|
||||
# unk0x32: 20 bytes of zeros (already zero from bytearray init)
|
||||
header[0x46:0x48] = language.encode("utf-8")[:2].ljust(2, b"\x00")
|
||||
struct.pack_into("<Q", header, 0x48, lib_pid)
|
||||
struct.pack_into("<I", header, 0x50, 1) # unk0x50
|
||||
struct.pack_into("<I", header, 0x54, 15) # unk0x54
|
||||
# hash58: 20 bytes of zeros (0x58..0x6C)
|
||||
struct.pack_into("<i", header, 0x6C, timezone_offset)
|
||||
struct.pack_into("<H", header, 0x70, 4) # hash_type_indicator (HASHAB)
|
||||
# hash72: 46 bytes of zeros (0x72..0xA0)
|
||||
struct.pack_into("<h", header, 0xA0, -1) # audio_language
|
||||
struct.pack_into("<h", header, 0xA2, -1) # subtitle_language
|
||||
# unk0xa4, unk0xa6: zeros
|
||||
struct.pack_into("<H", header, 0xA8, 0) # cdb_flag (0=uncompressed)
|
||||
# hashAB: 57 bytes of zeros (0xAB..0xE4)
|
||||
|
||||
return bytes(header) + datasets
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# HASHAB signing + file writing
|
||||
# ===========================================================================
|
||||
|
||||
def write_itunesdb(ipod_mount: str, tracks: list[TrackInfo],
|
||||
firewire_id: bytes = b"",
|
||||
backup: bool = True) -> 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("<H", itdb_data, 0x30, 4)
|
||||
|
||||
# Zero fields for hash computation
|
||||
itdb_data[0x18:0x20] = b"\x00" * 8 # db_id
|
||||
itdb_data[0x32:0x46] = b"\x00" * 20 # unk_0x32
|
||||
itdb_data[0x58:0x6C] = b"\x00" * 20 # hash58
|
||||
itdb_data[0x72:0xA0] = b"\x00" * 46 # hash72
|
||||
itdb_data[0xAB:0xE4] = b"\x00" * 57 # hashAB
|
||||
|
||||
sha1_digest = hashlib.sha1(bytes(itdb_data)).digest()
|
||||
signature = compute_hashab(sha1_digest, firewire_id[:8])
|
||||
|
||||
if len(signature) != 57:
|
||||
logger.error("HASHAB returned %d bytes (expected 57)", len(signature))
|
||||
return False
|
||||
|
||||
# Restore backed-up fields
|
||||
itdb_data[0x18:0x20] = backup_db_id
|
||||
itdb_data[0x32:0x46] = backup_unk32
|
||||
# Write HASHAB signature
|
||||
itdb_data[0xAB:0xAB + 57] = signature
|
||||
|
||||
# Keep scheme=4 (matching what was used during SHA1 computation)
|
||||
struct.pack_into("<H", itdb_data, 0x30, 4)
|
||||
logger.info("HASHAB signature written (scheme=4)")
|
||||
else:
|
||||
logger.warning("No FireWire GUID — database will NOT pass verification!")
|
||||
|
||||
# ---- Write BOTH iTunesDB (binary) and iTunesCDB (compressed SQLite) ----
|
||||
import shutil as _shutil
|
||||
if backup and os.path.exists(itdb_path):
|
||||
_shutil.copy2(itdb_path, itdb_path + ".backup")
|
||||
|
||||
# 1. Write proper binary iTunesDB
|
||||
temp = itdb_path + ".tmp"
|
||||
with open(temp, "wb") as f:
|
||||
f.write(itdb_data)
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
os.replace(temp, itdb_path)
|
||||
logger.info("Wrote iTunesDB: %d bytes", len(itdb_data))
|
||||
|
||||
# 2. Also write iTunesCDB: compressed Library.itdb with mhbd header.
|
||||
# This mirrors the original approach (firmware validates via this).
|
||||
lib_path = os.path.join(itunes_dir, "iTunes Library.itlp", "Library.itdb")
|
||||
cdb_path = os.path.join(itunes_dir, "iTunesCDB")
|
||||
if os.path.exists(lib_path):
|
||||
with open(lib_path, "rb") as f:
|
||||
lib_data = f.read()
|
||||
compressed = zlib.compress(lib_data, 1)
|
||||
|
||||
# Build mhbd header for the compressed SQLite payload (cdb_flag=1)
|
||||
total_size = MHBD_HEADER_SIZE + len(compressed)
|
||||
cdb_header = bytearray(MHBD_HEADER_SIZE)
|
||||
_write_generic_header(cdb_header, 0, b"mhbd", MHBD_HEADER_SIZE, total_size)
|
||||
struct.pack_into("<I", cdb_header, 0x0C, 2) # compressed=2 (iTunesCDB)
|
||||
struct.pack_into("<I", cdb_header, 0x10, 0x4F) # version
|
||||
struct.pack_into("<I", cdb_header, 0x14, 0) # child_count=0
|
||||
struct.pack_into("<Q", cdb_header, 0x18, _gen_id()) # db_id
|
||||
struct.pack_into("<H", cdb_header, 0x20, 2) # platform
|
||||
struct.pack_into("<H", cdb_header, 0x22, 611) # unk0x22
|
||||
struct.pack_into("<Q", cdb_header, 0x24, _gen_id()) # db_id_2
|
||||
struct.pack_into("<H", cdb_header, 0x30, 4) # hashing_scheme
|
||||
cdb_header[0x46:0x48] = b"ru"
|
||||
struct.pack_into("<Q", cdb_header, 0x48, _gen_id()) # lib_pid
|
||||
struct.pack_into("<I", cdb_header, 0x50, 1) # unk0x50
|
||||
struct.pack_into("<I", cdb_header, 0x54, 15) # unk0x54
|
||||
struct.pack_into("<H", cdb_header, 0x70, 4) # hash_type_indicator
|
||||
struct.pack_into("<h", cdb_header, 0xA0, -1) # audio_language
|
||||
struct.pack_into("<h", cdb_header, 0xA2, -1) # subtitle_language
|
||||
struct.pack_into("<H", cdb_header, 0xA8, 1) # cdb_flag=1 (compressed)
|
||||
|
||||
cdb_data = bytes(cdb_header) + compressed
|
||||
|
||||
# HASHAB sign the compressed database
|
||||
if firewire_id and len(firewire_id) >= 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("<H", data_for_hash, 0x30, 4)
|
||||
sha1_cdb = hashlib.sha1(bytes(data_for_hash)).digest()
|
||||
sig = compute_hashab(sha1_cdb, firewire_id[:8])
|
||||
cdb_header[0xAB:0xAB + 57] = sig
|
||||
struct.pack_into("<H", cdb_header, 0x30, 4)
|
||||
cdb_data = bytes(cdb_header) + compressed
|
||||
|
||||
cdb_temp = cdb_path + ".tmp"
|
||||
with open(cdb_temp, "wb") as f:
|
||||
f.write(cdb_data)
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
os.replace(cdb_temp, cdb_path)
|
||||
logger.info("Wrote iTunesCDB (compressed SQLite): %d bytes", len(cdb_data))
|
||||
|
||||
# 3. Write iTunesCDB.ext with SHA1 of Library.itdb (SQLite)
|
||||
ext_path = os.path.join(itunes_dir, "iTunesCDB.ext")
|
||||
if os.path.exists(lib_path):
|
||||
with open(lib_path, "rb") as f:
|
||||
lib_hash = hashlib.sha1(f.read()).hexdigest()
|
||||
else:
|
||||
lib_hash = hashlib.sha1(bytes(itdb_data)).hexdigest()
|
||||
ext_content = f"[iTunesCDB]\nitunesdb_hash={lib_hash}\n"
|
||||
with open(ext_path, "w") as f:
|
||||
f.write(ext_content)
|
||||
logger.info("Updated iTunesCDB.ext: hash=%s", lib_hash)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# SQLite → TrackInfo conversion
|
||||
# ===========================================================================
|
||||
|
||||
def tracks_from_sqlite(mountpoint: str) -> 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
|
||||
1832
src/ipod_nano7_db.py
1832
src/ipod_nano7_db.py
File diff suppressed because it is too large
Load Diff
Loading…
x
Reference in New Issue
Block a user