kaitai
This commit is contained in:
1 parent
f804c327de
commit
6a32066069
21 files changed
+2028
-1
No files matched your search
+5
-1
@@ -19,4 +19,8 @@ surf.sublime-workspace
|
||||
maple.sublime-workspace
|
||||
*.secret.yaml
|
||||
|
||||
etc/sldl-macos.yaml
|
||||
etc/sldl-macos.yaml
|
||||
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
.venv/
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1 @@
|
||||
3.14
|
||||
Whitespace-only changes.
@@ -0,0 +1,21 @@
|
||||
[project]
|
||||
name = "maple"
|
||||
version = "0.1.0"
|
||||
description = "Add your description here"
|
||||
readme = "README.md"
|
||||
authors = [
|
||||
{ name = "Nathan Rashleigh", email = "nathan.rashleigh@switchdin.com" }
|
||||
]
|
||||
requires-python = ">=3.14"
|
||||
dependencies = [
|
||||
"mutagen>=1.47.0",
|
||||
]
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[tool.uv]
|
||||
dev-dependencies = [
|
||||
"pytest>=9.0.3",
|
||||
]
|
||||
@@ -0,0 +1,2 @@
|
||||
def hello() -> str:
|
||||
return "Hello from maple!"
|
||||
@@ -0,0 +1,3 @@
|
||||
from maple.cli import main
|
||||
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,99 @@
|
||||
"""maple CLI: scan a directory of audio + m3u8 playlists and build a rekordbox USB."""
|
||||
import argparse
|
||||
import sys
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
from maple.copy import stage_audio
|
||||
from maple.meta import read_tags
|
||||
from maple.model import Album, Artist, Database, PlaylistEntry, PlaylistNode, Track
|
||||
from maple.pdb.writer import write_pdb
|
||||
from maple.scan import scan_audio, scan_playlists
|
||||
|
||||
|
||||
def build(input_dir: Path, usb_root: Path) -> Database:
|
||||
db = Database()
|
||||
today = date.today().isoformat()
|
||||
|
||||
artist_index: dict[str, int] = {}
|
||||
album_index: dict[tuple[int, str], int] = {}
|
||||
track_by_src: dict[Path, int] = {}
|
||||
|
||||
audio_files = scan_audio(input_dir)
|
||||
for src in audio_files:
|
||||
meta = read_tags(src)
|
||||
artist_id = artist_index.get(meta.artist)
|
||||
if artist_id is None:
|
||||
artist_id = len(db.artists) + 1
|
||||
artist_index[meta.artist] = artist_id
|
||||
db.artists.append(Artist(id=artist_id, name=meta.artist))
|
||||
|
||||
album_key = (artist_id, meta.album)
|
||||
album_id = album_index.get(album_key)
|
||||
if album_id is None:
|
||||
album_id = len(db.albums) + 1
|
||||
album_index[album_key] = album_id
|
||||
db.albums.append(Album(id=album_id, name=meta.album, artist_id=artist_id))
|
||||
|
||||
_, file_path = stage_audio(src, usb_root, meta.artist, meta.album)
|
||||
track_id = len(db.tracks) + 1
|
||||
db.tracks.append(Track(
|
||||
id=track_id,
|
||||
title=meta.title,
|
||||
artist_id=artist_id,
|
||||
album_id=album_id,
|
||||
filename=Path(file_path).name,
|
||||
file_path=file_path,
|
||||
file_size=meta.file_size,
|
||||
sample_rate=meta.sample_rate,
|
||||
sample_depth=meta.sample_depth,
|
||||
bitrate=meta.bitrate,
|
||||
duration=meta.duration,
|
||||
track_number=meta.track_number,
|
||||
year=meta.year,
|
||||
date_added=today,
|
||||
source_path=src,
|
||||
))
|
||||
track_by_src[src.resolve()] = track_id
|
||||
|
||||
playlists = scan_playlists(input_dir)
|
||||
for sort_order, (name, paths) in enumerate(playlists.items(), start=1):
|
||||
playlist_id = len(db.playlist_nodes) + 1
|
||||
db.playlist_nodes.append(PlaylistNode(
|
||||
id=playlist_id,
|
||||
parent_id=0,
|
||||
sort_order=sort_order,
|
||||
name=name,
|
||||
is_folder=False,
|
||||
))
|
||||
for entry_index, p in enumerate(paths, start=1):
|
||||
tid = track_by_src.get(p.resolve())
|
||||
if tid is None:
|
||||
print(f"warning: playlist {name!r} references missing file {p}", file=sys.stderr)
|
||||
continue
|
||||
db.playlist_entries.append(PlaylistEntry(
|
||||
playlist_id=playlist_id,
|
||||
track_id=tid,
|
||||
entry_index=entry_index,
|
||||
))
|
||||
|
||||
return db
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(prog="maple", description="Build a rekordbox USB from a folder of audio + m3u8 playlists.")
|
||||
parser.add_argument("input_dir", type=Path)
|
||||
parser.add_argument("output_root", nargs="?", type=Path, default=Path("./data/usb"))
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
if not args.input_dir.is_dir():
|
||||
parser.error(f"input_dir {args.input_dir} is not a directory")
|
||||
|
||||
db = build(args.input_dir, args.output_root)
|
||||
|
||||
pdb_path = args.output_root / "PIONEER" / "rekordbox" / "export.pdb"
|
||||
write_pdb(db, pdb_path)
|
||||
|
||||
print(f"wrote {len(db.tracks)} tracks, {len(db.artists)} artists, {len(db.albums)} albums, "
|
||||
f"{len(db.playlist_nodes)} playlists ({len(db.playlist_entries)} entries) to {pdb_path}")
|
||||
return 0
|
||||
@@ -0,0 +1,24 @@
|
||||
"""Copy audio files into the USB Contents layout."""
|
||||
import re
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
_BAD = re.compile(r'[\\/:*?"<>|\x00-\x1f]')
|
||||
|
||||
|
||||
def sanitise(component: str) -> str:
|
||||
s = _BAD.sub("_", component).strip(" .")
|
||||
return s or "_"
|
||||
|
||||
|
||||
def stage_audio(src: Path, usb_root: Path, artist: str, album: str) -> tuple[Path, str]:
|
||||
artist_dir = sanitise(artist)
|
||||
album_dir = sanitise(album)
|
||||
filename = sanitise(src.name)
|
||||
dest_rel = Path("Contents") / artist_dir / album_dir / filename
|
||||
dest_abs = usb_root / dest_rel
|
||||
dest_abs.parent.mkdir(parents=True, exist_ok=True)
|
||||
if not dest_abs.exists():
|
||||
shutil.copy2(src, dest_abs)
|
||||
file_path = "/" + dest_rel.as_posix()
|
||||
return dest_abs, file_path
|
||||
@@ -0,0 +1,84 @@
|
||||
"""Audio metadata extraction via mutagen."""
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from mutagen import File as MutagenFile
|
||||
|
||||
|
||||
@dataclass
|
||||
class TrackMeta:
|
||||
title: str
|
||||
artist: str
|
||||
album: str
|
||||
track_number: int
|
||||
year: int
|
||||
duration: int
|
||||
bitrate: int
|
||||
sample_rate: int
|
||||
sample_depth: int
|
||||
file_size: int
|
||||
|
||||
|
||||
def _first(value) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
if isinstance(value, list):
|
||||
return str(value[0]) if value else ""
|
||||
return str(value)
|
||||
|
||||
|
||||
def _int(value) -> int:
|
||||
s = _first(value).strip()
|
||||
if not s:
|
||||
return 0
|
||||
# Handle "3/12" track numbers
|
||||
s = s.split("/")[0]
|
||||
try:
|
||||
return int(s)
|
||||
except ValueError:
|
||||
return 0
|
||||
|
||||
|
||||
def read_tags(path: Path) -> TrackMeta:
|
||||
file_size = path.stat().st_size
|
||||
audio = MutagenFile(path, easy=True)
|
||||
if audio is None:
|
||||
return TrackMeta(
|
||||
title=path.stem,
|
||||
artist="Unknown Artist",
|
||||
album="Unknown Album",
|
||||
track_number=0,
|
||||
year=0,
|
||||
duration=0,
|
||||
bitrate=0,
|
||||
sample_rate=44100,
|
||||
sample_depth=16,
|
||||
file_size=file_size,
|
||||
)
|
||||
|
||||
tags = audio.tags or {}
|
||||
title = _first(tags.get("title")) or path.stem
|
||||
artist = _first(tags.get("artist")) or "Unknown Artist"
|
||||
album = _first(tags.get("album")) or "Unknown Album"
|
||||
track_number = _int(tags.get("tracknumber"))
|
||||
year = _int(tags.get("date") or tags.get("year"))
|
||||
|
||||
info = getattr(audio, "info", None)
|
||||
duration = int(getattr(info, "length", 0) or 0)
|
||||
bitrate_bps = int(getattr(info, "bitrate", 0) or 0)
|
||||
bitrate = bitrate_bps // 1000 if bitrate_bps > 1000 else bitrate_bps
|
||||
sample_rate = int(getattr(info, "sample_rate", 44100) or 44100)
|
||||
sample_depth = int(getattr(info, "bits_per_sample", 16) or 16)
|
||||
|
||||
return TrackMeta(
|
||||
title=title,
|
||||
artist=artist,
|
||||
album=album,
|
||||
track_number=track_number,
|
||||
year=year,
|
||||
duration=duration,
|
||||
bitrate=bitrate,
|
||||
sample_rate=sample_rate,
|
||||
sample_depth=sample_depth,
|
||||
file_size=file_size,
|
||||
)
|
||||
@@ -0,0 +1,67 @@
|
||||
"""Domain model for the in-memory database that becomes a .pdb file."""
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@dataclass
|
||||
class Artist:
|
||||
id: int
|
||||
name: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class Album:
|
||||
id: int
|
||||
name: str
|
||||
artist_id: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class Track:
|
||||
id: int
|
||||
title: str
|
||||
artist_id: int
|
||||
album_id: int
|
||||
filename: str
|
||||
file_path: str
|
||||
file_size: int = 0
|
||||
sample_rate: int = 44100
|
||||
sample_depth: int = 16
|
||||
bitrate: int = 0
|
||||
duration: int = 0
|
||||
track_number: int = 0
|
||||
year: int = 0
|
||||
disc_number: int = 0
|
||||
tempo: int = 0
|
||||
comment: str = ""
|
||||
date_added: str = ""
|
||||
release_date: str = ""
|
||||
mix_name: str = ""
|
||||
analyze_path: str = ""
|
||||
analyze_date: str = ""
|
||||
source_path: Path | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class PlaylistNode:
|
||||
id: int
|
||||
parent_id: int
|
||||
sort_order: int
|
||||
name: str
|
||||
is_folder: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
class PlaylistEntry:
|
||||
playlist_id: int
|
||||
track_id: int
|
||||
entry_index: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class Database:
|
||||
tracks: list[Track] = field(default_factory=list)
|
||||
artists: list[Artist] = field(default_factory=list)
|
||||
albums: list[Album] = field(default_factory=list)
|
||||
playlist_nodes: list[PlaylistNode] = field(default_factory=list)
|
||||
playlist_entries: list[PlaylistEntry] = field(default_factory=list)
|
||||
Whitespace-only changes.
@@ -0,0 +1,88 @@
|
||||
"""Page builder for the rekordbox .pdb format."""
|
||||
from struct import pack
|
||||
|
||||
PAGE_SIZE = 4096
|
||||
HEADER_SIZE = 40
|
||||
ROW_GROUP_SIZE = 36
|
||||
MAX_ROWS_PER_PAGE = 16 # one row group per page, matching rekordcrate
|
||||
|
||||
|
||||
def _align4(offset: int) -> int:
|
||||
return (offset + 3) & ~3
|
||||
|
||||
|
||||
class PageBuilder:
|
||||
def __init__(self, page_index: int, page_type: int):
|
||||
self.page_index = page_index
|
||||
self.page_type = page_type
|
||||
self.next_page = page_index # caller patches before finalize
|
||||
self.row_bytes: list[bytes] = []
|
||||
self.row_offsets: list[int] = [] # relative to heap_pos
|
||||
self.heap_cursor = 0 # position relative to heap_pos
|
||||
|
||||
def try_add(self, row: bytes) -> bool:
|
||||
if len(self.row_bytes) >= MAX_ROWS_PER_PAGE:
|
||||
return False
|
||||
aligned = _align4(self.heap_cursor)
|
||||
end = aligned + len(row)
|
||||
heap_capacity = PAGE_SIZE - HEADER_SIZE - ROW_GROUP_SIZE
|
||||
if end > heap_capacity:
|
||||
return False
|
||||
self.row_offsets.append(aligned)
|
||||
self.row_bytes.append(row)
|
||||
self.heap_cursor = end
|
||||
return True
|
||||
|
||||
def num_rows(self) -> int:
|
||||
return len(self.row_bytes)
|
||||
|
||||
def finalize(self) -> bytes:
|
||||
n = len(self.row_bytes)
|
||||
if n == 0:
|
||||
used_size = 0
|
||||
heap_padding = PAGE_SIZE - HEADER_SIZE
|
||||
else:
|
||||
used_size = self.heap_cursor
|
||||
heap_padding = PAGE_SIZE - HEADER_SIZE - ROW_GROUP_SIZE
|
||||
free_size = heap_padding - used_size
|
||||
if free_size < 0:
|
||||
raise ValueError("page overflow")
|
||||
|
||||
page_flags = 0x00 if n > 0 else 0x44
|
||||
|
||||
header = b""
|
||||
header += pack("<I", 0) # magic
|
||||
header += pack("<I", self.page_index)
|
||||
header += pack("<I", self.page_type)
|
||||
header += pack("<I", self.next_page)
|
||||
header += pack("<I", 0) # unknown1
|
||||
header += pack("<I", 0) # unknown2
|
||||
header += pack("<BBBB", n & 0xFF, 0, 0, page_flags)
|
||||
header += pack("<HH", free_size, used_size)
|
||||
header += pack("<H", 0) # unknown5
|
||||
header += pack("<H", n) # num_rows_large
|
||||
header += pack("<H", 0) # unknown6
|
||||
header += pack("<H", 0) # unknown7
|
||||
assert len(header) == HEADER_SIZE
|
||||
|
||||
heap = bytearray(heap_padding)
|
||||
if n > 0:
|
||||
for ofs, row in zip(self.row_offsets, self.row_bytes):
|
||||
heap[ofs:ofs + len(row)] = row
|
||||
|
||||
if n == 0:
|
||||
return header + bytes(heap)
|
||||
|
||||
# Row group footer (36 bytes), placed at end of page
|
||||
offsets_padded = self.row_offsets + [0] * (16 - n)
|
||||
presence = 0
|
||||
for i in range(n):
|
||||
presence |= 1 << i
|
||||
rg = b""
|
||||
for ofs in reversed(offsets_padded):
|
||||
rg += pack("<H", ofs)
|
||||
rg += pack("<H", presence)
|
||||
rg += pack("<H", 16) # unknown (rekordcrate default)
|
||||
assert len(rg) == ROW_GROUP_SIZE
|
||||
|
||||
return header + bytes(heap) + rg
|
||||
@@ -0,0 +1,147 @@
|
||||
"""Row serialisers for the rekordbox export.pdb format."""
|
||||
from struct import pack
|
||||
|
||||
from maple.model import Album, Artist, PlaylistEntry, PlaylistNode, Track
|
||||
from maple.pdb.strings import encode_devicesql
|
||||
|
||||
TRACK_FIXED_PREFIX = 0x5E
|
||||
TRACK_OFS_STRINGS_START = TRACK_FIXED_PREFIX
|
||||
TRACK_HEADER_LEN = TRACK_OFS_STRINGS_START + 21 * 2
|
||||
|
||||
|
||||
def _pad2(b: bytes) -> bytes:
|
||||
return b + b"\x00" if len(b) & 1 else b
|
||||
|
||||
|
||||
def serialize_track(t: Track) -> bytes:
|
||||
strings = [
|
||||
"", # 0 isrc
|
||||
"", # 1 texter
|
||||
"1", # 2 unknown_string_2
|
||||
"2", # 3 unknown_string_3
|
||||
"", # 4 unknown_string_4
|
||||
"", # 5 message
|
||||
"ON", # 6 kuvo_public
|
||||
"ON", # 7 autoload_hot_cues
|
||||
"", # 8 unknown_string_5
|
||||
"", # 9 unknown_string_6
|
||||
t.date_added, # 10 date_added
|
||||
t.release_date, # 11 release_date
|
||||
t.mix_name, # 12 mix_name
|
||||
"", # 13 unknown_string_7
|
||||
t.analyze_path, # 14 analyze_path
|
||||
t.analyze_date, # 15 analyze_date
|
||||
t.comment, # 16 comment
|
||||
t.title, # 17 title
|
||||
"", # 18 unknown_string_8
|
||||
t.filename, # 19 filename
|
||||
t.file_path, # 20 file_path
|
||||
]
|
||||
|
||||
encoded = [encode_devicesql(s) for s in strings]
|
||||
|
||||
offsets = []
|
||||
cursor = TRACK_HEADER_LEN
|
||||
blob_parts = []
|
||||
for enc in encoded:
|
||||
offsets.append(cursor)
|
||||
blob_parts.append(enc)
|
||||
cursor += len(enc)
|
||||
blob = b"".join(blob_parts)
|
||||
|
||||
prefix = b""
|
||||
prefix += pack("<HH", 0x24, 0) # subtype, index_shift
|
||||
prefix += pack("<I", 788224) # bitmask
|
||||
prefix += pack("<I", t.sample_rate) # sample_rate
|
||||
prefix += pack("<I", 0) # composer_id
|
||||
prefix += pack("<I", t.file_size) # file_size
|
||||
prefix += pack("<I", 0) # unknown
|
||||
prefix += pack("<H", 19048) # unknown ("always 19048?")
|
||||
prefix += pack("<H", 30967) # unknown ("always 30967?")
|
||||
prefix += pack("<I", 0) # artwork_id
|
||||
prefix += pack("<I", 0) # key_id
|
||||
prefix += pack("<I", 0) # original_artist_id
|
||||
prefix += pack("<I", 0) # label_id
|
||||
prefix += pack("<I", 0) # remixer_id
|
||||
prefix += pack("<I", t.bitrate) # bitrate
|
||||
prefix += pack("<I", t.track_number) # track_number
|
||||
prefix += pack("<I", t.tempo) # tempo
|
||||
prefix += pack("<I", 0) # genre_id
|
||||
prefix += pack("<I", t.album_id) # album_id
|
||||
prefix += pack("<I", t.artist_id) # artist_id
|
||||
prefix += pack("<I", t.id) # id
|
||||
prefix += pack("<H", t.disc_number) # disc_number
|
||||
prefix += pack("<H", 0) # play_count
|
||||
prefix += pack("<H", t.year) # year
|
||||
prefix += pack("<H", t.sample_depth) # sample_depth
|
||||
prefix += pack("<H", t.duration) # duration
|
||||
prefix += pack("<H", 41) # unknown ("always 41?")
|
||||
prefix += pack("<BB", 0, 0) # color_id, rating
|
||||
prefix += pack("<H", 1) # unknown ("always 1?")
|
||||
prefix += pack("<H", 3) # unknown ("alternating 2 or 3")
|
||||
|
||||
assert len(prefix) == TRACK_FIXED_PREFIX, len(prefix)
|
||||
|
||||
ofs_block = b"".join(pack("<H", o) for o in offsets)
|
||||
row = prefix + ofs_block + blob
|
||||
return _pad2(row)
|
||||
|
||||
|
||||
def serialize_artist(a: Artist) -> bytes:
|
||||
name = encode_devicesql(a.name)
|
||||
near_ofs = 0x0A
|
||||
far_threshold = 0x100
|
||||
if near_ofs < far_threshold:
|
||||
subtype = 0x60
|
||||
row = b""
|
||||
row += pack("<HH", subtype, a.id & 0xFFFF) # subtype, index_shift
|
||||
row += pack("<I", a.id) # id
|
||||
row += pack("<B", 0x03) # always 0x03
|
||||
row += pack("<B", near_ofs) # ofs_name_near
|
||||
assert len(row) == near_ofs
|
||||
row += name
|
||||
else:
|
||||
subtype = 0x64
|
||||
# Far form: fixed prefix + u2 ofs_name_far at row+0x0a
|
||||
# Layout: subtype u2, index_shift u2, id u4, 0x03 u1, ofs_name_near u1, ofs_name_far u2 -> name follows at offset 0x0e
|
||||
far = 0x0E
|
||||
row = b""
|
||||
row += pack("<HH", subtype, a.id & 0xFFFF)
|
||||
row += pack("<I", a.id)
|
||||
row += pack("<B", 0x03)
|
||||
row += pack("<B", 0)
|
||||
row += pack("<H", far)
|
||||
row += name
|
||||
return _pad2(row)
|
||||
|
||||
|
||||
def serialize_album(a: Album) -> bytes:
|
||||
name = encode_devicesql(a.name)
|
||||
near_ofs = 0x16
|
||||
subtype = 0x80
|
||||
row = b""
|
||||
row += pack("<HH", subtype, a.id & 0xFFFF) # subtype, index_shift
|
||||
row += pack("<I", 0) # unknown
|
||||
row += pack("<I", a.artist_id) # artist_id
|
||||
row += pack("<I", a.id) # id
|
||||
row += pack("<I", 0) # unknown
|
||||
row += pack("<B", 0x03) # always 0x03
|
||||
row += pack("<B", near_ofs) # ofs_name_near
|
||||
assert len(row) == near_ofs
|
||||
row += name
|
||||
return _pad2(row)
|
||||
|
||||
|
||||
def serialize_playlist_tree(n: PlaylistNode) -> bytes:
|
||||
row = b""
|
||||
row += pack("<I", n.parent_id)
|
||||
row += pack("<I", 0)
|
||||
row += pack("<I", n.sort_order)
|
||||
row += pack("<I", n.id)
|
||||
row += pack("<I", 1 if n.is_folder else 0)
|
||||
row += encode_devicesql(n.name)
|
||||
return _pad2(row)
|
||||
|
||||
|
||||
def serialize_playlist_entry(e: PlaylistEntry) -> bytes:
|
||||
return pack("<III", e.entry_index, e.track_id, e.playlist_id)
|
||||
@@ -0,0 +1,19 @@
|
||||
"""DeviceSQL variable-length string encoding."""
|
||||
|
||||
EMPTY = b"\x03"
|
||||
|
||||
|
||||
def encode_devicesql(s: str) -> bytes:
|
||||
if not s:
|
||||
return EMPTY
|
||||
try:
|
||||
ascii_bytes = s.encode("ascii")
|
||||
except UnicodeEncodeError:
|
||||
ascii_bytes = None
|
||||
if ascii_bytes is not None and len(ascii_bytes) < 127:
|
||||
length = len(ascii_bytes) + 1
|
||||
length_and_kind = ((length << 1) | 1) & 0xFF
|
||||
return bytes([length_and_kind]) + ascii_bytes
|
||||
payload = s.encode("utf-16-le")
|
||||
total_length = len(payload) + 4
|
||||
return bytes([0x90]) + total_length.to_bytes(2, "little") + b"\x00" + payload
|
||||
@@ -0,0 +1,113 @@
|
||||
"""Top-level rekordbox .pdb file writer."""
|
||||
from pathlib import Path
|
||||
from struct import pack
|
||||
|
||||
from maple.model import Database
|
||||
from maple.pdb.pager import PAGE_SIZE, PageBuilder
|
||||
from maple.pdb.rows import (
|
||||
serialize_album,
|
||||
serialize_artist,
|
||||
serialize_playlist_entry,
|
||||
serialize_playlist_tree,
|
||||
serialize_track,
|
||||
)
|
||||
|
||||
PAGE_TYPE_TRACKS = 0
|
||||
PAGE_TYPE_ARTISTS = 2
|
||||
PAGE_TYPE_ALBUMS = 3
|
||||
PAGE_TYPE_PLAYLIST_TREE = 7
|
||||
PAGE_TYPE_PLAYLIST_ENTRIES = 8
|
||||
|
||||
FILE_HEADER_SIZE = 28 # 0,page_size,num_tables,next_unused,unknown,sequence,gap = 7*u4
|
||||
TABLE_ENTRY_SIZE = 16 # page_type,empty_candidate,first_page,last_page = 4*u4
|
||||
|
||||
|
||||
def _build_chain(page_type: int, row_iter, start_index: int) -> tuple[list[bytes], int, int]:
|
||||
"""Return (page_bytes_list, first_page, last_page)."""
|
||||
pages: list[PageBuilder] = []
|
||||
page_idx = start_index
|
||||
current = PageBuilder(page_idx, page_type)
|
||||
|
||||
for row in row_iter:
|
||||
if not current.try_add(row):
|
||||
pages.append(current)
|
||||
page_idx += 1
|
||||
current = PageBuilder(page_idx, page_type)
|
||||
if not current.try_add(row):
|
||||
raise ValueError(f"row of size {len(row)} too large for an empty page")
|
||||
|
||||
pages.append(current)
|
||||
|
||||
# If the chain has no rows at all, still emit one empty page
|
||||
first_page = start_index
|
||||
last_page = page_idx
|
||||
|
||||
# Patch next_page pointers: each non-last page points to the next; last points to itself
|
||||
out: list[bytes] = []
|
||||
for i, p in enumerate(pages):
|
||||
if i == len(pages) - 1:
|
||||
p.next_page = p.page_index
|
||||
else:
|
||||
p.next_page = p.page_index + 1
|
||||
out.append(p.finalize())
|
||||
|
||||
return out, first_page, last_page
|
||||
|
||||
|
||||
def write_pdb(db: Database, path: str | Path) -> None:
|
||||
# Layout: page 0 is the file header page. Data pages begin at page_index 1.
|
||||
chains: list[tuple[int, list[bytes], int, int]] = [] # (page_type, pages, first, last)
|
||||
next_page_idx = 1
|
||||
|
||||
serialized_tracks = [serialize_track(t) for t in db.tracks]
|
||||
serialized_artists = [serialize_artist(a) for a in db.artists]
|
||||
serialized_albums = [serialize_album(a) for a in db.albums]
|
||||
serialized_pl_tree = [serialize_playlist_tree(n) for n in db.playlist_nodes]
|
||||
serialized_pl_entries = [serialize_playlist_entry(e) for e in db.playlist_entries]
|
||||
|
||||
for ptype, rows in [
|
||||
(PAGE_TYPE_TRACKS, serialized_tracks),
|
||||
(PAGE_TYPE_ARTISTS, serialized_artists),
|
||||
(PAGE_TYPE_ALBUMS, serialized_albums),
|
||||
(PAGE_TYPE_PLAYLIST_TREE, serialized_pl_tree),
|
||||
(PAGE_TYPE_PLAYLIST_ENTRIES, serialized_pl_entries),
|
||||
]:
|
||||
pages, first, last = _build_chain(ptype, rows, next_page_idx)
|
||||
chains.append((ptype, pages, first, last))
|
||||
next_page_idx = last + 1
|
||||
|
||||
num_tables = len(chains)
|
||||
next_unused_page = next_page_idx
|
||||
|
||||
file_header = b""
|
||||
file_header += pack("<I", 0) # magic
|
||||
file_header += pack("<I", PAGE_SIZE)
|
||||
file_header += pack("<I", num_tables)
|
||||
file_header += pack("<I", next_unused_page)
|
||||
file_header += pack("<I", 0) # unknown
|
||||
file_header += pack("<I", 1) # sequence
|
||||
file_header += pack("<I", 0) # gap
|
||||
assert len(file_header) == FILE_HEADER_SIZE
|
||||
|
||||
tables_block = b""
|
||||
for ptype, _pages, first, last in chains:
|
||||
tables_block += pack("<I", ptype)
|
||||
tables_block += pack("<I", 0) # empty_candidate
|
||||
tables_block += pack("<I", first)
|
||||
tables_block += pack("<I", last)
|
||||
assert len(tables_block) == num_tables * TABLE_ENTRY_SIZE
|
||||
|
||||
header_full = file_header + tables_block
|
||||
pad_to_page = PAGE_SIZE - len(header_full)
|
||||
if pad_to_page < 0:
|
||||
raise ValueError("too many tables for one page")
|
||||
page0 = header_full + b"\x00" * pad_to_page
|
||||
|
||||
out_path = Path(path)
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with out_path.open("wb") as f:
|
||||
f.write(page0)
|
||||
for _ptype, pages, _first, _last in chains:
|
||||
for p in pages:
|
||||
assert len(p) == PAGE_SIZE
|
||||
f.write(p)
|
||||
Whitespace-only changes.
@@ -0,0 +1,34 @@
|
||||
"""Discover audio files and m3u8 playlists in an input directory."""
|
||||
from pathlib import Path
|
||||
|
||||
AUDIO_EXTS = {".mp3", ".flac", ".wav", ".aiff", ".aif", ".m4a"}
|
||||
|
||||
|
||||
def scan_audio(input_dir: Path) -> list[Path]:
|
||||
out = []
|
||||
for p in sorted(input_dir.rglob("*")):
|
||||
if p.is_file() and p.suffix.lower() in AUDIO_EXTS:
|
||||
out.append(p)
|
||||
return out
|
||||
|
||||
|
||||
def parse_m3u8(playlist_path: Path) -> list[Path]:
|
||||
base = playlist_path.parent
|
||||
entries = []
|
||||
with playlist_path.open("r", encoding="utf-8", errors="replace") as f:
|
||||
for line in f:
|
||||
s = line.strip()
|
||||
if not s or s.startswith("#"):
|
||||
continue
|
||||
p = Path(s)
|
||||
if not p.is_absolute():
|
||||
p = (base / p).resolve()
|
||||
entries.append(p)
|
||||
return entries
|
||||
|
||||
|
||||
def scan_playlists(input_dir: Path) -> dict[str, list[Path]]:
|
||||
out: dict[str, list[Path]] = {}
|
||||
for p in sorted(input_dir.rglob("*.m3u8")):
|
||||
out[p.stem] = parse_m3u8(p)
|
||||
return out
|
||||
@@ -0,0 +1,30 @@
|
||||
from maple.pdb.strings import EMPTY, encode_devicesql
|
||||
|
||||
|
||||
def test_empty():
|
||||
assert encode_devicesql("") == EMPTY
|
||||
|
||||
|
||||
def test_short_ascii():
|
||||
out = encode_devicesql("ON")
|
||||
# length=3 -> length_and_kind = (3<<1)|1 = 7
|
||||
assert out == bytes([7]) + b"ON"
|
||||
|
||||
|
||||
def test_longer_ascii():
|
||||
s = "Hello, World!"
|
||||
out = encode_devicesql(s)
|
||||
length = len(s) + 1
|
||||
assert out[0] == ((length << 1) | 1)
|
||||
assert out[1:] == s.encode("ascii")
|
||||
|
||||
|
||||
def test_utf16():
|
||||
s = "ЛаBа"
|
||||
out = encode_devicesql(s)
|
||||
assert out[0] == 0x90
|
||||
# Total length stored = payload_len + 4
|
||||
payload = s.encode("utf-16-le")
|
||||
assert out[1:3] == (len(payload) + 4).to_bytes(2, "little")
|
||||
assert out[3] == 0
|
||||
assert out[4:] == payload
|
||||
@@ -0,0 +1,29 @@
|
||||
from pathlib import Path
|
||||
|
||||
from maple.model import Album, Artist, Database, PlaylistEntry, PlaylistNode, Track
|
||||
from maple.pdb.writer import write_pdb
|
||||
|
||||
|
||||
def test_smoke(tmp_path: Path):
|
||||
db = Database()
|
||||
db.artists.append(Artist(id=1, name="AceMo"))
|
||||
db.albums.append(Album(id=1, name="All My Life", artist_id=1))
|
||||
db.tracks.append(Track(
|
||||
id=1, title="Where They At", artist_id=1, album_id=1,
|
||||
filename="ace.mp3", file_path="/Contents/AceMo/All My Life/ace.mp3",
|
||||
file_size=14547692, sample_rate=44100, sample_depth=16, bitrate=320,
|
||||
duration=436, track_number=7, year=2019, date_added="2025-01-04",
|
||||
))
|
||||
db.playlist_nodes.append(PlaylistNode(id=1, parent_id=0, sort_order=1, name="MyList"))
|
||||
db.playlist_entries.append(PlaylistEntry(playlist_id=1, track_id=1, entry_index=1))
|
||||
|
||||
out = tmp_path / "export.pdb"
|
||||
write_pdb(db, out)
|
||||
assert out.exists()
|
||||
data = out.read_bytes()
|
||||
assert len(data) % 4096 == 0
|
||||
assert data[:4] == b"\x00\x00\x00\x00"
|
||||
# page_size at offset 4
|
||||
assert int.from_bytes(data[4:8], "little") == 4096
|
||||
# num_tables at offset 8
|
||||
assert int.from_bytes(data[8:12], "little") == 5
|
||||
Generated
+91
@@ -0,0 +1,91 @@
|
||||
version = 1
|
||||
requires-python = ">=3.14"
|
||||
|
||||
[[package]]
|
||||
name = "colorama"
|
||||
version = "0.4.6"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697 }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335 },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "iniconfig"
|
||||
version = "2.3.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503 }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484 },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "maple"
|
||||
version = "0.1.0"
|
||||
source = { editable = "." }
|
||||
dependencies = [
|
||||
{ name = "mutagen" },
|
||||
]
|
||||
|
||||
[package.dev-dependencies]
|
||||
dev = [
|
||||
{ name = "pytest" },
|
||||
]
|
||||
|
||||
[package.metadata]
|
||||
requires-dist = [{ name = "mutagen", specifier = ">=1.47.0" }]
|
||||
|
||||
[package.metadata.requires-dev]
|
||||
dev = [{ name = "pytest", specifier = ">=9.0.3" }]
|
||||
|
||||
[[package]]
|
||||
name = "mutagen"
|
||||
version = "1.47.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/81/e6/64bc71b74eef4b68e61eb921dcf72dabd9e4ec4af1e11891bbd312ccbb77/mutagen-1.47.0.tar.gz", hash = "sha256:719fadef0a978c31b4cf3c956261b3c58b6948b32023078a2117b1de09f0fc99", size = 1274186 }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/b0/7a/620f945b96be1f6ee357d211d5bf74ab1b7fe72a9f1525aafbfe3aee6875/mutagen-1.47.0-py3-none-any.whl", hash = "sha256:edd96f50c5907a9539d8e5bba7245f62c9f520aef333d13392a79a4f70aca719", size = 194391 },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "packaging"
|
||||
version = "26.2"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/d7/f1/e7a6dd94a8d4a5626c03e4e99c87f241ba9e350cd9e6d75123f992427270/packaging-26.2.tar.gz", hash = "sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661", size = 228134 }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195 },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pluggy"
|
||||
version = "1.6.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412 }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538 },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pygments"
|
||||
version = "2.20.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991 }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151 },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pytest"
|
||||
version = "9.0.3"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "colorama", marker = "sys_platform == 'win32'" },
|
||||
{ name = "iniconfig" },
|
||||
{ name = "packaging" },
|
||||
{ name = "pluggy" },
|
||||
{ name = "pygments" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/7d/0d/549bd94f1a0a402dc8cf64563a117c0f3765662e2e668477624baeec44d5/pytest-9.0.3.tar.gz", hash = "sha256:b86ada508af81d19edeb213c681b1d48246c1a91d304c6c81a427674c17eb91c", size = 1572165 }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249 },
|
||||
]
|
||||
Reference in new issue
Block a user