68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
from pathlib import Path
|
|
from typing import Dict, Iterator, Optional, Tuple
|
|
|
|
from serato_doctor.models.database import DatabaseTrack, SeratoDatabase
|
|
|
|
|
|
TEXT_FIELDS = {
|
|
b"tsng": "title",
|
|
b"tart": "artist",
|
|
b"talb": "album",
|
|
b"tgen": "genre",
|
|
}
|
|
|
|
|
|
def iter_records(data: bytes) -> Iterator[Tuple[bytes, bytes]]:
|
|
"""Yield complete big-endian tag-length-value records."""
|
|
|
|
offset = 0
|
|
while offset + 8 <= len(data):
|
|
tag = data[offset : offset + 4]
|
|
length = int.from_bytes(data[offset + 4 : offset + 8], "big")
|
|
payload_start = offset + 8
|
|
payload_end = payload_start + length
|
|
if payload_end > len(data):
|
|
break
|
|
yield tag, data[payload_start:payload_end]
|
|
offset = payload_end
|
|
|
|
|
|
def decode_text(payload: bytes) -> Optional[str]:
|
|
value = payload.decode("utf-16-be", errors="ignore").strip("\x00").strip()
|
|
return value or None
|
|
|
|
|
|
def normalize_database_path(value: str) -> Path:
|
|
if value.startswith(("Users/", "Volumes/")):
|
|
value = "/" + value
|
|
return Path(value)
|
|
|
|
|
|
def parse_track(payload: bytes) -> Optional[DatabaseTrack]:
|
|
fields: Dict[str, Optional[str]] = {}
|
|
path = None
|
|
for tag, value in iter_records(payload):
|
|
if tag == b"pfil":
|
|
decoded_path = decode_text(value)
|
|
if decoded_path:
|
|
path = normalize_database_path(decoded_path)
|
|
elif tag in TEXT_FIELDS:
|
|
fields[TEXT_FIELDS[tag]] = decode_text(value)
|
|
if path is None:
|
|
return None
|
|
return DatabaseTrack(path=path, filename=path.name, **fields)
|
|
|
|
|
|
def parse_database(database_path: Path) -> SeratoDatabase:
|
|
data = database_path.read_bytes()
|
|
version = None
|
|
tracks = []
|
|
for tag, payload in iter_records(data):
|
|
if tag == b"vrsn":
|
|
version = decode_text(payload)
|
|
elif tag == b"otrk":
|
|
track = parse_track(payload)
|
|
if track is not None:
|
|
tracks.append(track)
|
|
return SeratoDatabase(database_path, version, tuple(tracks))
|