diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6dd5859 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.DS_Store +__pycache__/ +*.pyc +*.xml diff --git a/README.md b/README.md index 946fef5..10e875a 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,20 @@ -# serato2rekordbox v1.3 +# serato2rekordbox v1.4 -This command line tool converts your Serato DJ Pro library (playlists, tracks, metadata, beatgrids, and hot cues) into a Rekordbox XML file that can be imported into Rekordbox DJ (tested on 6.8.5 and should work on older/newer versions) and can be exported easily to a USB or just used in Rekordbox in HID mode. +This command line tool converts your Serato DJ Pro library (playlists, tracks, metadata, beatgrids, and hot cues) into a Rekordbox XML file that can be imported into Rekordbox DJ (tested on 6.8.5 and should work on older/newer versions) and can be exported easily to a USB or just used in Rekordbox in HID mode. ## Changelog +### v1.4: + +- **FLAC hot cues and beatgrids** -- Serato writes hot cues and beatgrids as double-base64-encoded Vorbis comments (`SERATO_MARKERS_V2`, `SERATO_BEATGRID`) in FLAC files. The extractor now correctly decodes these, matching the upstream staddle/serato2rekordbox implementation. ID3 GEOB frames are tried as a fallback for third-party tools +- **USB drive / path remapping support** -- point at a USB drive containing both `_Serato_/` and your music files with a single path; the script auto-discovers crates and remaps track paths from the original library location to the USB +- **Interactive crate selector** -- pick which crates to convert before processing (curses-based on macOS/Linux, text-based on Windows) +- **New format support** -- `.flac`, `.alac`, `.aiff` files are now fully supported (metadata, hot cues, beatgrids) +- **CLI arguments** -- `--serato`, `--music`, `--output`, `--skip-crate-selection` for full control without editing code +- **Missing beatgrid is no longer fatal** -- tracks without Serato beatgrid data are still included in the XML (Rekordbox will analyse on import) +- **FLAC metadata fix** -- ID3 tags are now loaded correctly (previously crashed on all `.flac` files) +- **Code cleanup** -- removed unused imports, consolidated format dispatch, improved error messages + ### v1.3: - Fixed issue where subcrates inside subcrates were not processed at all @@ -38,17 +49,19 @@ I couldn't find any other free software that could convert it so I decided to ma * **Track Metadata:** Transfers essential metadata including Title, Artist, BPM, and Key. * **Hot Cue Transfer:** Extracts and transfers hot cues. * **Accurate Beatgrids:** Extracts the Serato beatgrid data directly from the audio files to extract the *first beat position* from the audio file's beatgrid data and includes it in the XML. This tells Rekordbox exactly where the first beat is, allowing it to correctly align the entire beatgrid without needing to re-analyse it itself. -* **Automatic Serato Folder Detection:** Automatically attempts to find your Serato `_Serato_` folder on standard Windows, macOS and Linux locations. +* **USB Drive Support:** Run the script against a USB drive that contains both the Serato metadata folder and your music files. Track paths stored in the original library location are automatically remapped to the USB via a fast filename index. +* **Interactive Crate Selector:** Choose which crates to convert via an in-terminal checkbox UI (macOS/Linux) or numbered menu (Windows). Skip with `--skip-crate-selection`. +* **Automatic Serato Folder Detection:** Automatically attempts to find your Serato `_Serato_` folder on standard Windows, macOS and Linux locations, or from within a USB drive root. * **Detailed Error Reporting:** Collects and reports errors (missing files, unsupported formats, processing errors, crate reading issues) in a clear, grouped summary at the end. Failed tracks are excluded from the output XML. -* **File Support:** Supports conversion for `.mp3`, `.m4a` and `.wav` audio files found in your Serato library. +* **File Support:** Supports conversion for `.mp3`, `.m4a`, `.alac`, `.wav`, `.flac`, and `.aiff` audio files. * Normal crates and subcrates are supported. ## Prerequisites Before running the script, ensure you have the following: -1. **Python 3:** The script is written in Python 3. You can download it from [python.org](https://www.python.org/downloads/). -2. **Serato DJ Pro:** Serato must be installed and run at least once on the computer where you run this script, as the script needs to access the Serato `_Serato_` folder and the music files. +1. **Python 3:** The script is written in Python 3. You can download it from [python.org](https://python.org/downloads/). +2. **Serato DJ Pro:** Serato must be installed and run at least once on the computer where you run this script (or your Serato metadata folder must be available on a USB drive). 3. **Rekordbox DJ:** You will need Rekordbox DJ (version 6 recommended) to import the generated XML file. ## Installation @@ -60,34 +73,75 @@ Before running the script, ensure you have the following: ``` Or download the ZIP file and extract it. -2. **Install Dependencies:** Open your terminal or command prompt, navigate to the script's directory, and install the required Python package: +2. **Install Dependencies:** Open your terminal or command prompt, navigate to the script's directory, and install the required Python packages: ```bash pip install tqdm mutagen ``` ## Usage -1. **Open Terminal/Command Prompt:** Navigate to the directory where you saved the script files. -2. **Run the Script:** Execute the script using Python 3: - ```bash - python3 serato2rekordbox.py - ``` -3. **Let the magic happen:** The script will attempt to auto-detect your Serato folder, read your crates, process your tracks, structure playlists, and finally write the XML file. It shouldn't take longer than a few seconds to complete unless you have a huge library. -4. **Review Output:** After completion, the script will print a summary of successful/unsuccessful conversions and list any items that could not be processed, grouped by the type of error. +### Quick start (auto-detect everything) + +If your Serato library is in the default location and music files are where Serato expects them: + +```bash +python3 serato2rekordbox.py +``` + +### USB drive (one command) + +Your USB drive contains `_Serato_/` (metadata) plus your music folders: + +```bash +python3 serato2rekordbox.py /Volumes/YourUSB +``` + +The script will: +1. Find `_Serato_/` inside the USB for crate files +2. Index all music files in sibling folders (skipping `_Serato_`, `_Serato_Backup`, system folders) +3. Remap crate paths from the original library location to the USB via filename lookup +4. Open the crate selector so you can pick what to convert + +### Separating Serato metadata from music files + +```bash +python3 serato2rekordbox.py --serato ~/Music/_Serato_ --music /Volumes/USB/Music +``` + +### Command-line options + +| Flag | Description | +|---|---| +| `root` | (positional) Path to a USB drive or folder containing both `_Serato_/` and music files | +| `--serato PATH` | Path to the Serato library folder (contains `subcrates/`). Auto-detected when using positional `root`; otherwise defaults to `~/Music/_Serato_` | +| `--music PATH` | Root path of the music files. Auto-detected when using positional `root`; otherwise defaults to the Serato folder | +| `--output PATH` | Output XML file path. Defaults to `serato2rekordbox.xml` beside the script | +| `--skip-crate-selection` | Skip the interactive crate selector and import all crates | + +### Crate selector + +On macOS and Linux a full-screen checkbox interface opens: +- **Arrow keys** navigate, **Space** toggles, **A** selects all, **N** deselects all, **Enter** confirms, **q** cancels + +On Windows (or when curses is unavailable) a numbered menu appears: +- Type comma-separated numbers to toggle crates (e.g. `1,3,5`), **a** for all, **n** for none, **Enter** to confirm, **q** to cancel + +Press **Enter** immediately to accept all crates and skip selection. -The output file `serato2rekordbox.xml` will be generated in the same directory as the script. +### Output -**Note:** There are no configuration variables (`base_dir`, `serato_folder_path`) to change at the top of the script anymore, as it attempts to find the Serato folder automatically. +After completion the script prints a summary of successful / unsuccessful conversions. The output file `serato2rekordbox.xml` is generated in the script's directory (or at the path specified by `--output`). ## Importing into Rekordbox Once `serato2rekordbox.xml` is generated: 1. Open Rekordbox DJ. -2. Go to Settings (Gear icon at top) > Advanced > rekordbox xml and select the generated XML file. -3. In the sidebar, drag the imported playlists into your USB and wait for Rekordbox to finish processing. +2. **Add your tracks first** -- drag the music folders from your USB into Rekordbox so the tracks appear in the library. This is required because Rekordbox only resolves playlist references for tracks it already knows about. +3. Go to Settings (Gear icon at top) > Advanced > rekordbox xml and select the generated XML file. +4. In the sidebar, the imported playlists will appear under the **Collection** tab. Drag them onto your USB export and wait for Rekordbox to finish processing. -Rekordbox will import the playlists and tracks. The tracks should appear with their correct metadata (Title, Artist, BPM, Key), Hot Cues, and the accurate Beatgrids based on the first beat position provided in the XML. Rekordbox may still perform some background analysis (like waveform drawing), but it should respect the imported beatgrid and cue data. +Rekordbox will import the playlists with their track ordering, Hot Cues, and the accurate Beatgrids based on the first beat position provided in the XML. Rekordbox may still perform some background analysis (like waveform drawing), but it should respect the imported beatgrid and cue data. ## Limitations @@ -95,21 +149,18 @@ Rekordbox will import the playlists and tracks. The tracks should appear with th * Some tracks may not have the correct beatgrid data or key. * Smart crates are not supported. * Some beatgrids in Rekordbox appear to be slightly off beat even though perfectly on beat in Serato. -* You may have to manually adjust the Serato folder path in the code if it isn't auto detected. * There may be occasional tracks which are not correctly processed for some unknown reason, however across my library of almost 4000 tracks there's only been a handful which have been problematic for me. ## Notes * The script has been tested on my own Serato library which contains almost 4000 tracks and performs as intended. It was able to process the entire library in around 20 seconds and rekordbox exporting to my USB (HFS+) only took around 15 minutes. -* Rekordbox 6.8.5 is recommended at the moment as Rekordbox 7 is reported to export to USB much slower. +* Rekordbox 6.8.5 is recommended at the moment as Rekordbox 7 is reported to export to USB much slower. * `HFS+` seems to be the fastest USB file format but `FAT32` is more compatible with older Pioneer hardware. -* `.wav` files have not been extensively tested but appear to work fine. ## Future improvements -- Support other file formats eg. `.flac`, `.alac`, `.aiff` - only `.mp3`, `.m4a` and `.wav` are supported at the moment. - Make a GUI? -- I thought about trying to reverse engineer the USB export structure so the program could directly export to a USB itself without needing Rekordbox at all, however the USB structure (analysis, database etc) is extremely complex, would require alot of effort and likely wouldn't be as reliable. +- I thought about trying to reverse engineer the USB export structure so the program could directly export to a USB itself without needing Rekordbox at all, however the USB structure (analysis, database etc) is extremely complex, would require a lot of effort and likely wouldn't be as reliable. ## Contributing diff --git a/extract_aiff.py b/extract_aiff.py new file mode 100644 index 0000000..27d16ae --- /dev/null +++ b/extract_aiff.py @@ -0,0 +1,186 @@ +import io +import logging +import struct + +import mutagen +from mutagen.aiff import AIFF + +from utils import convert_key_to_camelot + + +def extract_metadata(file_path: str) -> dict: + """Extract metadata, hot cues, and beatgrid from an AIFF file. + + Serato embeds ID3v2 tags in AIFF files (same GEOB structure as WAV/FLAC). + """ + results = { + "metadata": { + "title": "Unknown", + "artist": "Unknown", + "bpm": 0.0, + "key": "Unknown", + "duration_sec": 0.0, + "sample_rate": 0, + }, + "hot_cues": [], + "beatgrid": {"markers": {"non_terminal": [], "terminal": None}}, + } + + try: + audio = AIFF(file_path) + except Exception as e: + logging.error("Failed to open AIFF file '%s': %s", file_path, e) + return results + + # Audio info + if audio.info: + results["metadata"]["duration_sec"] = round(audio.info.length, 3) + results["metadata"]["sample_rate"] = audio.info.sample_rate or 0 + + # Serato writes ID3 tags to AIFF — same structure as WAV + tags = audio.tags + if not tags: + # Try loading as generic mutagen file for ID3 + try: + tagfile = mutagen.File(file_path) + if tagfile and tagfile.tags: + tags = tagfile.tags + except Exception: + pass + + if tags: + # Standard metadata + tit2 = tags.get("TIT2") + if tit2 and hasattr(tit2, "text") and tit2.text: + results["metadata"]["title"] = str(tit2.text[0]).strip() + + tpe1 = tags.get("TPE1") + if tpe1 and hasattr(tpe1, "text") and tpe1.text: + results["metadata"]["artist"] = str(tpe1.text[0]).strip() + + tbpm = tags.get("TBPM") + if tbpm and hasattr(tbpm, "text") and tbpm.text: + try: + results["metadata"]["bpm"] = float(str(tbpm.text[0]).strip()) + except (ValueError, TypeError): + pass + + tkey = tags.get("TKEY", tags.get("initialkey")) + if tkey and hasattr(tkey, "text") and tkey.text: + results["metadata"]["key"] = convert_key_to_camelot(str(tkey.text[0]).strip()) + + # Hot cues and beatgrid from GEOB frames + from mutagen.id3 import GEOB + + for tag in tags.values(): + if isinstance(tag, GEOB): + desc = getattr(tag, "desc", "") + if desc == "Serato Markers2": + try: + results["hot_cues"] = parse_serato_hot_cues(tag.data) + except Exception as e: + logging.warning("Error reading hot cues from '%s': %s", file_path, e) + elif desc == "Serato BeatGrid": + try: + results["beatgrid"] = parse_beatgrid(tag.data) + except Exception as e: + logging.warning("Error reading beatgrid from '%s': %s", file_path, e) + + return results + + +def parse_serato_hot_cues(base64_data: bytes) -> list: + """Parse Serato Markers2 GEOB payload (shared with MP3/WAV/FLAC).""" + import base64 + import re + + if isinstance(base64_data, str): + base64_data = base64_data.encode("utf-8") + + clean_data = re.sub(rb"[^a-zA-Z0-9+/=]", b"", base64_data) + padding_needed = 4 - (len(clean_data) % 4) + if padding_needed != 4: + clean_data += b"=" * padding_needed + + try: + data = base64.b64decode(clean_data) + except Exception: + return [] + + hot_cues = [] + idx = 0 + total = len(data) + + while idx < total: + nxt = data[idx:].find(b"\x00") + if nxt == -1: + break + + entry_type = data[idx : idx + nxt].decode("utf-8", errors="replace") + idx += nxt + 1 + + if idx + 4 > total: + break + + entry_len = struct.unpack(">I", data[idx : idx + 4])[0] + idx += 4 + + if entry_type == "CUE" and idx + entry_len <= total: + raw = data[idx : idx + entry_len] + if len(raw) >= 12: + cue_index = raw[1] + position_ms = struct.unpack(">I", raw[2:6])[0] + r, g, b = raw[7:10] + color = "#{:02X}{:02X}{:02X}".format(r, g, b) + label = raw[12:].rstrip(b"\x00").decode("utf-8", errors="replace") + + hot_cues.append({ + "index": cue_index, + "position_ms": position_ms, + "color": color, + "name": label, + }) + + idx += entry_len + + return hot_cues + + +def parse_beatgrid(raw_data: bytes) -> dict: + """Parse Serato BeatGrid binary struct from a GEOB frame.""" + result = {"markers": {"non_terminal": [], "terminal": None}} + + fp = io.BytesIO(raw_data) + version = struct.unpack("BB", fp.read(2)) + if version != (0x01, 0x00): + logging.warning("Unsupported beatgrid version: %s", version) + + num_markers = struct.unpack(">I", fp.read(4))[0] + markers = [] + + for i in range(num_markers): + pos = struct.unpack(">f", fp.read(4))[0] + data = fp.read(4) + if i == num_markers - 1: + bpm = struct.unpack(">f", data)[0] + markers.append(("terminal", pos, bpm)) + else: + beats = struct.unpack(">I", data)[0] + markers.append(("non_terminal", pos, beats)) + + fp.read(1) # trailing byte + + if not markers: + return result + + non_terminal = [] + terminal = None + for m in markers: + if m[0] == "terminal": + terminal = {"position": m[1], "bpm": m[2]} + else: + non_terminal.append({"position": m[1], "beats_till_next_marker": m[2]}) + + result["markers"]["non_terminal"] = non_terminal + result["markers"]["terminal"] = terminal + return result diff --git a/extract_flac.py b/extract_flac.py new file mode 100644 index 0000000..8e343e6 --- /dev/null +++ b/extract_flac.py @@ -0,0 +1,406 @@ +import base64 +import io +import logging +import re +import struct + +from mutagen.flac import FLAC + +from utils import convert_key_to_camelot + +# ── Vorbis comment value extraction ────────────────────────────────────────── + + +def get_vorbis_value(vc, key): + """Safely get a single Vorbis comment value, always returned as bytes.""" + if key not in vc: + return None + vals = vc[key] + if not vals: + return None + val = vals[0] + if hasattr(val, "data"): + val = val.data + if isinstance(val, str): + return val.encode("utf-8") + return val + + +# ── Double-base64 decode matching staddle/serato2rekordbox ─────────────────── + + +def _first_decode(raw_value: bytes) -> bytes: + """Run the first base64 layer and strip the MIME / descriptor header. + + Returns the inner base64-encoded payload (still as raw bytes). + """ + # Clean — remove '=' entirely so _pad_base64 adds it only in valid positions + clean = re.sub(rb"[^a-zA-Z0-9+/]", b"", raw_value) + padded = _pad_base64(clean) + + data = base64.b64decode(padded) + + if not data.startswith(b"application/octet-stream\x00"): + raise ValueError("Missing MIME header") + + fieldname_endpos = data.index(b"\x00", 26) + return data[fieldname_endpos + 1:] + + +def decode_serato_hot_cues_payload(raw_value: bytes) -> bytes: + """Decode SERATO_MARKERS_V2 to raw binary cue data.""" + fielddata = _first_decode(raw_value) + fielddata = fielddata.replace(b"\n", b"") + + # Skip the first 2 bytes (descriptor prefix) and take up to the next null + null_pos = fielddata.find(b"\x00") + inner_b64 = fielddata[2:null_pos] if null_pos >= 2 else b"" + + try: + return base64.b64decode(_pad_base64(inner_b64)) + except Exception as e1: + logging.debug("Cue inner b64 decode failed, retrying truncated: %s", e1) + truncated = inner_b64[: (len(inner_b64) // 4) * 4] + return base64.b64decode(_pad_base64(truncated)) + + +def decode_serato_beatgrid_payload(raw_value: bytes) -> bytes: + """Decode SERATO_BEATGRID to raw binary beatgrid struct.""" + fielddata = _first_decode(raw_value) + + # Clean — remove '=' entirely so _pad_base64 adds it only in valid terminal positions + clean = re.sub(rb"[^a-zA-Z0-9+/]", b"", fielddata) + try: + return base64.b64decode(_pad_base64(clean)) + except Exception: + # Truncate to multiple-of-4 and retry + truncated = clean[: (len(clean) // 4) * 4] + return base64.b64decode(_pad_base64(truncated)) + + +def _pad_base64(b64_bytes: bytes) -> bytes: + """Pad base64 bytes, truncating to last valid 4-char boundary if needed.""" + missing = len(b64_bytes) % 4 + if missing: + # 1 mod 4 is invalid for base64 — truncate the trailing byte + if missing == 1: + b64_bytes = b64_bytes[: len(b64_bytes) - 1] + return b64_bytes # now 0 mod 4 + b64_bytes += b"=" * (4 - missing) + return b64_bytes + + +# ── Hot cue parsing ────────────────────────────────────────────────────────── + + +def parse_serato_hot_cues(binary_data: bytes) -> list: + """Parse raw binary cue entries from the decoded payload. + + Format: version (2 bytes 0x01 0x01), then null-terminated entry name + + 4-byte big-endian length + entry data, repeated until empty name. + """ + hot_cues = [] + fp = io.BytesIO(binary_data) + + try: + version = struct.unpack("BB", fp.read(2)) + except struct.error: + return hot_cues + + if version != (0x01, 0x01): + # Maybe no version header — try simple null-separated parsing + return _parse_hot_cues_simple(binary_data) + + while True: + name_bytes = _read_null_terminated(fp) + if not name_bytes: + break + name = name_bytes.decode("utf-8", errors="replace") + + try: + entry_len = struct.unpack(">I", fp.read(4))[0] + except struct.error: + break + + if entry_len <= 0: + break + + entry_data = fp.read(entry_len) + + if name == "CUE" and len(entry_data) >= 12: + cue_index = entry_data[1] + position_ms = struct.unpack(">I", entry_data[2:6])[0] + r, g, b = entry_data[7:10] + color = "#{:02X}{:02X}{:02X}".format(r, g, b) + label = entry_data[12:].rstrip(b"\x00").decode("utf-8", errors="replace") + + hot_cues.append({ + "index": cue_index, + "position_ms": position_ms, + "color": color, + "name": label, + }) + + return hot_cues + + +def _parse_hot_cues_simple(data: bytes) -> list: + """Fallback: parse null-separated CUE entries without version header.""" + hot_cues = [] + idx = 0 + total = len(data) + + while idx < total: + nxt = data[idx:].find(b"\x00") + if nxt == -1: + break + entry_type = data[idx : idx + nxt].decode("utf-8", errors="replace") + idx += nxt + 1 + + if idx + 4 > total: + break + entry_len = struct.unpack(">I", data[idx : idx + 4])[0] + idx += 4 + + if entry_type == "CUE" and idx + entry_len <= total: + raw = data[idx : idx + entry_len] + if len(raw) >= 12: + hot_cues.append({ + "index": raw[1], + "position_ms": struct.unpack(">I", raw[2:6])[0], + "color": "#{:02X}{:02X}{:02X}".format(*raw[7:10]), + "name": raw[12:].rstrip(b"\x00").decode("utf-8", errors="replace"), + }) + idx += entry_len + + return hot_cues + + +def _read_null_terminated(fp: io.BytesIO) -> bytes: + chunks = [] + while True: + b = fp.read(1) + if not b or b == b"\x00": + break + chunks.append(b) + return b"".join(chunks) + + +# ── Beatgrid parsing ───────────────────────────────────────────────────────── + + +def parse_beatgrid(binary_data: bytes) -> dict: + """Parse the Serato BeatGrid binary struct. + + Format: version (2 bytes), marker count (4 bytes BE), markers (8 bytes + each for non-terminal, 8 bytes for terminal), trailing byte. + """ + result = {"markers": {"non_terminal": [], "terminal": None}} + fp = io.BytesIO(binary_data) + + try: + version = struct.unpack("BB", fp.read(2)) + num_markers = struct.unpack(">I", fp.read(4))[0] + except struct.error: + return result + + if version != (0x01, 0x00): + logging.warning("Unsupported beatgrid version: %s", version) + + markers = [] + for i in range(num_markers): + pos_bytes = fp.read(4) + data_bytes = fp.read(4) + if len(pos_bytes) < 4 or len(data_bytes) < 4: + break + pos = struct.unpack(">f", pos_bytes)[0] + if i == num_markers - 1: + bpm = struct.unpack(">f", data_bytes)[0] + markers.append(("terminal", pos, bpm)) + else: + beats = struct.unpack(">I", data_bytes)[0] + markers.append(("non_terminal", pos, beats)) + + fp.read(1) # trailing byte + + if not markers: + return result + + non_terminal = [] + terminal = None + for m in markers: + if m[0] == "terminal": + terminal = {"position": m[1], "bpm": m[2]} + else: + non_terminal.append({"position": m[1], "beats_till_next_marker": m[2]}) + + result["markers"]["non_terminal"] = non_terminal + result["markers"]["terminal"] = terminal + return result + + +# ── ID3 GEOB fallback ──────────────────────────────────────────────────────── + + +def try_id3_geob(file_path: str, need_cues: bool, need_beatgrid: bool) -> tuple: + """Try to read Serato hot cues and beatgrid from ID3 GEOB frames in a FLAC. + + Returns (hot_cues_list, beatgrid_dict). + """ + hot_cues = [] + beatgrid = {"markers": {"non_terminal": [], "terminal": None}} + + try: + from mutagen.id3 import ID3, GEOB + + id3_tags = ID3(file_path) + except Exception: + return hot_cues, beatgrid + + for tag in id3_tags.values(): + if not isinstance(tag, GEOB): + continue + desc = getattr(tag, "desc", "") + if desc == "Serato Markers2" and need_cues: + hot_cues = _parse_geob_markers(tag.data) + elif desc == "Serato BeatGrid" and need_beatgrid: + beatgrid = parse_beatgrid(tag.data) + + return hot_cues, beatgrid + + +def _parse_geob_markers(raw_data: bytes) -> list: + """Parse GEOB Serato Markers2 data (single base64, no MIME wrapper).""" + clean = re.sub(rb"[^a-zA-Z0-9+/]", b"", raw_data) + clean = clean.rstrip(b"=") + missing = len(clean) % 4 + if missing: + clean += b"=" * (4 - missing) + + try: + data = base64.b64decode(clean) + except Exception: + return [] + + return _parse_hot_cues_simple(data) + + +# ── Main entry point ───────────────────────────────────────────────────────── + + +def extract_metadata(file_path: str) -> dict: + """Extract metadata, hot cues, and beatgrid from a FLAC file.""" + results = { + "metadata": { + "title": "Unknown", + "artist": "Unknown", + "bpm": 0.0, + "key": "Unknown", + "duration_sec": 0.0, + "sample_rate": 0, + }, + "hot_cues": [], + "beatgrid": {"markers": {"non_terminal": [], "terminal": None}}, + } + + try: + audio = FLAC(file_path) + except Exception as e: + logging.error("Failed to open FLAC file '%s': %s", file_path, e) + return results + + if audio.info: + results["metadata"]["duration_sec"] = round(audio.info.length, 3) + results["metadata"]["sample_rate"] = audio.info.sample_rate or 0 + + # ── Vorbis comments ────────────────────────────────────────────────── + + vc = audio.tags + if vc: + if "TITLE" in vc: + results["metadata"]["title"] = str(vc["TITLE"][0]).strip() + if "ARTIST" in vc: + results["metadata"]["artist"] = str(vc["ARTIST"][0]).strip() + if "BPM" in vc: + try: + results["metadata"]["bpm"] = float(vc["BPM"][0]) + except (ValueError, TypeError, IndexError): + pass + if results["metadata"]["key"] == "Unknown": + for key_tag in ("KEY", "INITIALKEY", "MUSICBRAINZ_TRACKKEY", "TBPM_KEY"): + if key_tag in vc: + results["metadata"]["key"] = convert_key_to_camelot(str(vc[key_tag][0])) + break + + # Hot cues + raw = get_vorbis_value(vc, "SERATO_MARKERS_V2") + if raw is not None: + try: + binary = decode_serato_hot_cues_payload(raw) + cues = parse_serato_hot_cues(binary) + if cues: + results["hot_cues"] = cues + except Exception as e: + logging.warning("Failed to decode hot cues from '%s': %s", file_path, e) + + # Beatgrid + raw = get_vorbis_value(vc, "SERATO_BEATGRID") + if raw is not None and results["beatgrid"]["markers"]["terminal"] is None: + try: + binary = decode_serato_beatgrid_payload(raw) + bg = parse_beatgrid(binary) + if bg["markers"]["terminal"] is not None: + results["beatgrid"] = bg + except Exception as e: + logging.warning("Failed to decode beatgrid from '%s': %s", file_path, e) + + # ── ID3 GEOB fallback (only if Vorbis didn't provide cues or beatgrid) ─ + + need_cues = not results["hot_cues"] + need_bg = results["beatgrid"]["markers"]["terminal"] is None + + has_id3 = False + try: + from mutagen.id3 import ID3 + + ID3(file_path) + has_id3 = True + except Exception: + pass + + if has_id3 and (need_cues or need_bg): + cues, bg = try_id3_geob(file_path, need_cues, need_bg) + if cues: + results["hot_cues"] = cues + if bg["markers"]["terminal"] is not None: + results["beatgrid"] = bg + + # Fill missing basic metadata from ID3 if Vorbis was empty + if has_id3 and (results["metadata"]["title"] == "Unknown" or results["metadata"]["artist"] == "Unknown"): + try: + from mutagen.id3 import ID3 + + id3_tags = ID3(file_path) + if results["metadata"]["title"] == "Unknown": + tit2 = id3_tags.get("TIT2") + if tit2 and hasattr(tit2, "text") and tit2.text: + results["metadata"]["title"] = str(tit2.text[0]).strip() + if results["metadata"]["artist"] == "Unknown": + tpe1 = id3_tags.get("TPE1") + if tpe1 and hasattr(tpe1, "text") and tpe1.text: + results["metadata"]["artist"] = str(tpe1.text[0]).strip() + if results["metadata"]["bpm"] == 0.0: + tbpm = id3_tags.get("TBPM") + if tbpm and hasattr(tbpm, "text") and tbpm.text: + try: + results["metadata"]["bpm"] = float(str(tbpm.text[0]).strip()) + except (ValueError, TypeError): + pass + if results["metadata"]["key"] == "Unknown": + tkey = id3_tags.get("TKEY") + if tkey and hasattr(tkey, "text") and tkey.text: + results["metadata"]["key"] = convert_key_to_camelot(str(tkey.text[0]).strip()) + except Exception: + pass + + return results diff --git a/extract_m4a.py b/extract_m4a.py index bce3e0c..a1619bd 100644 --- a/extract_m4a.py +++ b/extract_m4a.py @@ -1,17 +1,15 @@ import base64 +import binascii import io import logging import re import struct from pathlib import Path -import binascii -import json from mutagen.mp4 import MP4 -from mutagen.id3 import ID3 import mutagen.mp4 -from utils import major_key_conversion, minor_key_conversion, convert_key_to_camelot +from utils import convert_key_to_camelot def read_null_terminated(fp: io.BytesIO) -> bytes: chunks = [] @@ -119,10 +117,11 @@ def parse_serato_hot_cues(tag_data) -> list: tag_data = tag_data.encode("utf-8") clean = tag_data.replace(b"\n", b"") - clean = re.sub(rb"[^a-zA-Z0-9+/=]", b"", clean) + clean = re.sub(rb"[^a-zA-Z0-9+/]", b"", clean) - if len(clean) % 4 != 0: - clean += b"=" * ((4 - len(clean) % 4)) + missing = len(clean) % 4 + if missing: + clean += b"=" * (4 - missing) try: outer_decoded = base64.b64decode(clean) @@ -183,17 +182,23 @@ def parse_serato_hot_cues(tag_data) -> list: return cues def extract_metadata(file_path: str) -> dict: - results = {"metadata": {}, "hot_cues": [], "beatgrid":[]} + results = {"metadata": {}, "hot_cues": [], "beatgrid": []} track = Path(file_path) if not track.exists(): logging.error("File not found: %s", file_path) return results - results["beatgrid"] = get_beatgrid(file_path) + try: + results["beatgrid"] = get_beatgrid(file_path) + except ValueError: + # Beatgrid tag not present or malformed — not fatal + pass + except Exception as e: + logging.warning("Error reading beatgrid from '%s': %s", file_path, e) candidates = [] - if track.suffix.lower() == ".m4a": + if track.suffix.lower() in (".m4a", ".alac"): audio = MP4(str(track)) results["metadata"]["title"] = audio.get("\xa9nam", ["Unknown Title"])[0] results["metadata"]["artist"] = audio.get("\xa9ART", ["Unknown Artist"])[0] @@ -215,7 +220,7 @@ def extract_metadata(file_path: str) -> dict: for tag_key in candidates: tag_data = None - if track.suffix.lower() == ".m4a": + if track.suffix.lower() in (".m4a", ".alac"): tag_data = audio.get(tag_key, [None])[0] else: @@ -244,7 +249,9 @@ def extract_metadata(file_path: str) -> dict: def decode_beatgrid(value): raw_bytes = bytes(value) - cleaned = raw_bytes.replace(b'\n', b'') + # Remove newlines and all '=' to avoid invalid-position padding + cleaned = re.sub(rb"[^a-zA-Z0-9+/=]", b"", raw_bytes) + cleaned = cleaned.rstrip(b"=") try: potential = cleaned[:-1] @@ -253,7 +260,7 @@ def decode_beatgrid(value): if missing_padding: potential += b'=' * (4 - missing_padding) - decoded = base64.b64decode(potential, validate=True) + decoded = base64.b64decode(potential) except binascii.Error: missing_padding = len(cleaned) % 4 @@ -261,7 +268,7 @@ def decode_beatgrid(value): if missing_padding: cleaned += b'=' * (4 - missing_padding) - decoded = base64.b64decode(cleaned, validate=True) + decoded = base64.b64decode(cleaned) return decoded @@ -274,7 +281,7 @@ def process_grid_data(grid_data): expected_length = 6 + (marker_count * 8) + 1 if len(grid_data) != expected_length: - print(f"Warning: Decoded grid data length ({len(grid_data)}) does not match expected length ({expected_length}).") + logging.debug("Grid data length (%d) differs from expected (%d) — parsing anyway", len(grid_data), expected_length) markers_block = grid_data[6:6 + marker_count * 8] markers = [] diff --git a/extract_mp3.py b/extract_mp3.py index 7dd8260..3e5b8f5 100644 --- a/extract_mp3.py +++ b/extract_mp3.py @@ -1,19 +1,15 @@ -import os -import re -import struct import base64 -import logging -from pathlib import Path import io -import json -import sys +import logging +import re +import struct from collections import namedtuple import mutagen from mutagen.id3 import ID3, GEOB from mutagen.mp3 import MP3 -from utils import major_key_conversion, minor_key_conversion, convert_key_to_camelot +from utils import convert_key_to_camelot NonTerminalBeatgridMarker = namedtuple("NonTerminalBeatgridMarker", ["position", "beats_till_next_marker"]) TerminalBeatgridMarker = namedtuple("TerminalBeatgridMarker", ["position", "bpm"]) @@ -121,6 +117,15 @@ def extract_metadata(input_file: str) -> dict: key = convert_key_to_camelot(key) + beatgrid = {} + try: + beatgrid = get_beatgrid(input_file) + except ValueError: + # Beatgrid tag not present — not fatal + pass + except Exception as e: + logging.warning("Error reading beatgrid from '%s': %s", input_file, e) + return { "metadata": { "title": audio_metadata.get("TIT2", "Unknown"), @@ -130,7 +135,7 @@ def extract_metadata(input_file: str) -> dict: "duration_sec": audio_metadata.get("TotalTime", 0) }, "hot_cues": hot_cues, - "beatgrid": get_beatgrid(input_file) + "beatgrid": beatgrid } def parse_beatgrid_markers(fp): diff --git a/extract_wav.py b/extract_wav.py index a46c6f3..432a230 100644 --- a/extract_wav.py +++ b/extract_wav.py @@ -1,24 +1,18 @@ -import os -import re -import struct import base64 -import logging -from pathlib import Path import io -import json -import sys +import logging +import re +import struct from collections import namedtuple import mutagen -from mutagen.id3 import GEOB +from mutagen.id3 import GEOB -from utils import major_key_conversion, minor_key_conversion, convert_key_to_camelot +from utils import convert_key_to_camelot NonTerminalBeatgridMarker = namedtuple("NonTerminalBeatgridMarker", ["position", "beats_till_next_marker"]) TerminalBeatgridMarker = namedtuple("TerminalBeatgridMarker", ["position", "bpm"]) -logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') - def parse_serato_hot_cues(base64_data): if not base64_data: logging.debug("No hot cue data provided.") diff --git a/serato2rekordbox.py b/serato2rekordbox.py index d6bd7fc..5c15573 100644 --- a/serato2rekordbox.py +++ b/serato2rekordbox.py @@ -1,102 +1,439 @@ -print(r''' - _ ___ _ _ _ - | | |__ \ | | | | | +print(r''' + _ ___ _ _ _ + | | |__ \ | | | | | ___ ___ _ __ __ _| |_ ___ ) |_ __ ___| | _____ _ __ __| | |__ _____ __ / __|/ _ \ '__/ _` | __/ _ \ / /| '__/ _ \ |/ / _ \| '__/ _` | '_ \ / _ \ \/ / - \__ \ __/ | | (_| | || (_) / /_| | | __/ < (_) | | | (_| | |_) | (_) > < + \__ \ __/ | | (_| | || (_) / /_| | | __/ < (_) | | | (_| | |_) | (_) > < |___/\___|_| \__,_|\__\___/____|_| \___|_|\_\___/|_| \__,_|_.__/ \___/_/\_\ ''') -current_version = "serato2rekordbox v1.3" -print("\nVersion 1.3\n\n") +__version__ = "serato2rekordbox v1.5" +print(f"\n{__version__}\n") +_IS_LOCAL_BUILD = True # disable update checker (not on the original repo yet) + +import argparse +import base64 +import curses +import logging import os +import platform import re +import ssl import struct -from xml.etree.ElementTree import Element, SubElement, tostring +import urllib.parse +import urllib.request +from collections import OrderedDict, defaultdict from xml.dom import minidom +from xml.etree.ElementTree import Element, SubElement, tostring + from tqdm import tqdm -import platform -import urllib.parse -from collections import defaultdict -from collections import OrderedDict -import extract_mp3 +import extract_flac import extract_m4a +import extract_mp3 import extract_wav +import extract_aiff -import urllib.request -import ssl +# ── Constants ──────────────────────────────────────────────────────────────── -try: - url = "https://raw.githubusercontent.com/BytePhoenixCoding/serato2rekordbox/main/README.md" - context = ssl._create_unverified_context() # <- disable SSL verification - with urllib.request.urlopen(url, timeout=5, context=context) as response: - content = response.read().decode('utf-8') - - if current_version not in content: - print("──────────────────────────────────────────────────────────") - print("⚠️ A new version of serato2rekordbox is available!") - print("🔗 Please update here: https://github.com/BytePhoenixCoding/serato2rekordbox") - print("──────────────────────────────────────────────────────────\n") - else: - print("✅ serato2rekordbox is up to date.") -except Exception as e: - print(f"(Update check skipped: {e})") +SUPPORTED_EXTENSIONS = { + ".mp3", ".m4a", ".alac", ".wav", ".flac", ".aiff", +} + +EXTRACTOR_MAP = { + ".mp3": extract_mp3, + ".m4a": extract_m4a, + ".alac": extract_m4a, + ".wav": extract_wav, + ".flac": extract_flac, + ".aiff": extract_aiff, +} + +KIND_MAP = { + ".mp3": "MP3 File", + ".m4a": "M4A File", + ".alac": "ALAC File", + ".wav": "WAV File", + ".flac": "FLAC File", + ".aiff": "AIFF File", +} -START_MARKER = b'ptrk' +START_MARKER = b"ptrk" PATH_LENGTH_OFFSET = 4 START_MARKER_FULL_LENGTH = len(START_MARKER) + PATH_LENGTH_OFFSET M4A_BEATGRID_OFFSET = 0.07 M4A_HOTCUE_OFFSET = 0.03 -unsuccessfulConversions = [] +# ── Helpers ────────────────────────────────────────────────────────────────── + def prettify(elem): - rough_string = tostring(elem, 'utf-8') - reparsed = minidom.parseString(rough_string) - return reparsed.toprettyxml(indent=" ") + rough = tostring(elem, "utf-8") + return minidom.parseString(rough).toprettyxml(indent=" ") -def find_serato_folder(): - home_dir = os.path.expanduser('~') - potential_paths = [] +def check_for_updates(): + try: + url = "https://raw.githubusercontent.com/BytePhoenixCoding/serato2rekordbox/main/README.md" + ctx = ssl._create_unverified_context() + with urllib.request.urlopen(url, timeout=5, context=ctx) as resp: + content = resp.read().decode("utf-8") + if __version__ not in content: + print("──────────────────────────────────────────────────────────") + print("A new version of serato2rekordbox is available!") + print("Please update: https://github.com/BytePhoenixCoding/serato2rekordbox") + print("──────────────────────────────────────────────────────────\n") + else: + print("serato2rekordbox is up to date.") + except Exception as e: + print(f"(Update check skipped: {e})") + + +def find_serato_folder(): + """Auto-detect the Serato library folder (used as default for --serato).""" + home = os.path.expanduser("~") + candidates = [] if platform.system() == "Windows": - potential_paths = [ - os.path.join(home_dir, 'Music', '_Serato_'), - os.path.join(home_dir, 'Documents', '_Serato_'), - ] - elif platform.system() == "Darwin": - potential_paths = [ - os.path.join(home_dir, 'Music', '_Serato_'), - ] - else: - potential_paths = [ - os.path.join(home_dir, 'Music', '_Serato_'), + candidates = [ + os.path.join(home, "Music", "_Serato_"), + os.path.join(home, "Documents", "_Serato_"), ] + else: + candidates = [os.path.join(home, "Music", "_Serato_")] + + for p in candidates: + if os.path.isdir(p): + return p + return None + + +def resolve_track_path(crate_path, music_root, path_cache): + """Resolve a file path stored in a .crate file to an actual filesystem path. + + Strategy: + 1. If the path exists as-is, return it. + 2. Look up the cache (built from music_root) by filename for O(1) lookup. + """ + # Normalise separators + norm = crate_path.replace("\\", os.sep) + if platform.system() != "Windows" and not norm.startswith(os.sep): + norm = os.sep + norm + + # Already seen? + if norm in path_cache: + return path_cache[norm] - for path in potential_paths: - if os.path.exists(path) and os.path.isdir(path): - print(f"✅ Found Serato folder at: {path}") - return path + # Exists verbatim? + if os.path.isfile(norm): + path_cache[norm] = norm + return norm - print("Error: Serato '_Serato_' folder not found in common locations.") - print("Please ensure Serato DJ Pro has been run at least once.") + # Fallback: look up by basename in the prebuilt cache + basename = os.path.basename(norm) + resolved = path_cache.get(("__basename__", basename)) + if resolved: + path_cache[norm] = resolved + return resolved + + # Not found return None -def generate_rekordbox_xml(processed_data, all_tracks_in_tracks): + +def build_path_cache(music_root): + """Walk music_root and build a {filename: full_path} lookup dict. + + Returns a dict where: + - ( '__basename__', 'song.mp3' ) → '/Volumes/USB/Music/song.mp3' + Only the first occurrence of each filename is kept (matching original dedup). + """ + cache = {} + if not music_root or not os.path.isdir(music_root): + return cache + + for dirpath, _dirs, files in os.walk(music_root): + for fname in files: + key = ("__basename__", fname) + if key not in cache: + cache[key] = os.path.join(dirpath, fname) + return cache + + +# ── Crate I/O ──────────────────────────────────────────────────────────────── + + +def find_serato_crates(subcrates_path): + """Recursively find all .crate files under subcrates_path.""" + crates = [] + if not os.path.isdir(subcrates_path): + print(f"Error: subcrates folder not found: {subcrates_path}") + return crates + + print(f"Searching for .crate files in: {subcrates_path}") + for root, _dirs, files in os.walk(subcrates_path): + for f in files: + if f.endswith(".crate"): + crates.append(os.path.join(root, f)) + print(f"Found {len(crates)} crate file(s).\n") + return crates + + +def extract_file_paths_from_crate(crate_path, encoding="utf-16-be"): + """Parse raw track paths from a .crate binary file.""" + paths = [] + seen = set() + errors = [] + + try: + with open(crate_path, "rb") as fh: + blob = fh.read() + + blob_len = len(blob) + i = 0 + + while i < blob_len - START_MARKER_FULL_LENGTH: + idx = blob.find(START_MARKER, i) + if idx == -1: + break + i = idx + len(START_MARKER) + + if i + PATH_LENGTH_OFFSET > blob_len: + errors.append(("parse_eof", crate_path, f"Unexpected EOF after marker at byte {idx}")) + break + + path_len = struct.unpack(">I", blob[i : i + PATH_LENGTH_OFFSET])[0] + i += PATH_LENGTH_OFFSET + + if i + path_len > blob_len: + errors.append(("parse_overflow", crate_path, f"Path size {path_len} exceeds data at byte {i}")) + break + + raw = blob[i : i + path_len] + i += path_len + + try: + abs_path = raw.decode(encoding).strip() + except UnicodeDecodeError: + errors.append(("decode", crate_path, f"UTF-16 decode failure at byte {i - path_len}")) + continue + + if abs_path not in seen: + paths.append(abs_path) + seen.add(abs_path) + + except FileNotFoundError: + print(f"Error: Crate file not found: {crate_path}") + except Exception as exc: + errors.append(("read", crate_path, str(exc))) + + return paths, errors + + +def format_crate_name(raw_name): + """Turn 'Folder%%Subfolder%%ID' into 'Folder [Subfolder] [ID]'.""" + segments = raw_name.split("%%") + if len(segments) == 1: + return segments[0] + return segments[0] + "".join(f" [{s}]" for s in segments[1:]) + + +# ── Crate selector ─────────────────────────────────────────────────────────── + + +def select_crates(stdscr, crate_list): + """Curses-based crate selector (macOS / Linux). + + Returns a list of (crate_path, display_name) tuples the user selected. + """ + curses.curs_set(0) + stdscr.keypad(True) + stdscr.nodelay(False) + + height, width = stdscr.getmaxyx() + if height < 5 or width < 40: + curses.echo() + curses.endwin() + print("Terminal too small for interactive selector. Converting all crates.") + return [(p, format_crate_name(os.path.basename(p)[:-6])) for p in crate_list] + + title = " Select crates to convert (Arrows=move, Space=toggle, A=all, N=none, Enter=go) " + footer_sel = " Selected: {sel}/{total} | q=cancel" + + selected = [True] * len(crate_list) + cursor = 0 + scroll_offset = 0 + + def draw(): + stdscr.erase() + h, w = stdscr.getmaxyx() + + stdscr.attron(curses.A_BOLD | curses.A_REVERSE) + stdscr.addnstr(0, 0, title.center(w), w - 1) + stdscr.attroff(curses.A_BOLD | curses.A_REVERSE) + + stdscr.addnstr(1, 0, " " * (w - 1), curses.A_DIM) + + visible = h - 4 + + nonlocal scroll_offset + if cursor < scroll_offset: + scroll_offset = cursor + elif cursor >= scroll_offset + visible: + scroll_offset = cursor - visible + 1 + scroll_offset = max(0, min(scroll_offset, len(crate_list) - visible)) + + for i in range(visible): + idx = scroll_offset + i + row = i + 2 + if idx >= len(crate_list): + break + + name = crate_list[idx][1] + checked = " X " if selected[idx] else " " + prefix = f"[{checked}] " + + max_name_len = w - len(prefix) - 2 + displayed = name[:max_name_len] if len(name) > max_name_len else name + + attr = curses.A_REVERSE if idx == cursor else curses.A_NORMAL + try: + stdscr.addstr(row, 0, prefix, attr) + stdscr.addstr(row, len(prefix), displayed, attr) + except curses.error: + pass + + sel_count = sum(selected) + footer = footer_sel.format(sel=sel_count, total=len(crate_list)).ljust(width - 1) + stdscr.attron(curses.A_BOLD | curses.A_REVERSE) + stdscr.addnstr(h - 1, 0, footer, width - 1) + stdscr.attroff(curses.A_BOLD | curses.A_REVERSE) + stdscr.refresh() + + try: + while True: + draw() + ch = stdscr.getch() + + if ch in (curses.KEY_ENTER, 10, 13): + break + elif ch == curses.KEY_UP: + cursor = max(0, cursor - 1) + elif ch == curses.KEY_DOWN: + cursor = min(len(crate_list) - 1, cursor + 1) + elif ch == curses.KEY_PPAGE: + cursor = max(0, cursor - (height - 4)) + elif ch == curses.KEY_NPAGE: + cursor = min(len(crate_list) - 1, cursor + (height - 4)) + elif ch == curses.KEY_HOME: + cursor = 0 + elif ch == curses.KEY_END: + cursor = len(crate_list) - 1 + elif ch in (32,): + selected[cursor] = not selected[cursor] + elif ch in (97, 65): + selected = [True] * len(crate_list) + elif ch in (110, 78): + selected = [False] * len(crate_list) + elif ch in (113, 27): + curses.echo() + curses.endwin() + print("Crate selection cancelled.") + return None + + finally: + curses.echo() + curses.endwin() + + return [ + (crate_list[i][0], crate_list[i][1]) + for i in range(len(crate_list)) + if selected[i] + ] + + +def select_crates_fallback(crate_list): + """Plain-text crate selector for Windows or when curses is unavailable. + + Shows a numbered list; user types comma-separated numbers to toggle. + Returns a list of (crate_path, display_name) tuples the user selected. + """ + print("\n Select crates to convert (all selected by default)\n") + for i, (_path, name) in enumerate(crate_list, 1): + print(f" [{chr(9732)}] {i:3d}. {name}") + + print(f"\n {len(crate_list)} crate(s) found.") + print(" Commands: toggle (e.g. 1,3,5)") + print(" select all") + print(" deselect all") + print(" confirm and start conversion") + print(" cancel\n") + + selected = [True] * len(crate_list) + + while True: + try: + choice = input(">>> ").strip().lower() + except (EOFError, KeyboardInterrupt): + print("\nCancelled.") + return None + + if choice == "": + break + if choice == "q": + print("Crate selection cancelled.") + return None + if choice == "a": + selected = [True] * len(crate_list) + print("All selected.") + continue + if choice == "n": + selected = [False] * len(crate_list) + print("All deselected.") + continue + + # Parse comma-separated numbers + parts = [p.strip() for p in choice.split(",")] + for part in parts: + try: + idx = int(part) - 1 + if 0 <= idx < len(crate_list): + selected[idx] = not selected[idx] + name = crate_list[idx][1] + state = "selected" if selected[idx] else "deselected" + print(f" {name} -> {state}") + else: + print(f" Invalid number: {part}") + except ValueError: + print(f" Invalid input: {part}") + + sel_count = sum(selected) + print(f" ({sel_count}/{len(crate_list)} selected)\n") + + return [ + (crate_list[i][0], crate_list[i][1]) + for i in range(len(crate_list)) + if selected[i] + ] + + +# ── XML generation ─────────────────────────────────────────────────────────── + + +def generate_rekordbox_xml(playlists, tracks): + """Build a Rekordbox-compatible XML document and write to disk.""" root = Element("DJ_PLAYLISTS", Version="1.0.0") SubElement(root, "PRODUCT", Name="rekordbox", Version="6.0.0", Company="AlphaTheta") - collection = SubElement(root, "COLLECTION", Entries=str(len(all_tracks_in_tracks))) + collection = SubElement(root, "COLLECTION", Entries=str(len(tracks))) playlists_elem = SubElement(root, "PLAYLISTS") root_playlist = SubElement(playlists_elem, "NODE", Type="0", Name="ROOT", Count="0") track_id_map = {} - current_track_id = 1 + next_id = 1 - for path, data in tqdm(all_tracks_in_tracks.items(), desc="⚙️ (4/4) Adding tracks"): - track_id_map[path] = current_track_id + for path, data in tqdm(tracks.items(), desc="(4/4) Adding tracks"): + track_id_map[path] = next_id + # Build file URI if platform.system() == "Windows": uri_path = path.replace("\\", "/") if re.match(r"^[A-Za-z]:", uri_path): @@ -105,21 +442,26 @@ def generate_rekordbox_xml(processed_data, all_tracks_in_tracks): else: uri = "file://localhost/" + urllib.parse.quote(path.lstrip("/")) - kind = "MP3 File" if path.lower().endswith(".mp3") else "M4A File" if path.lower().endswith(".m4a") else "WAV File" - - tr = SubElement(collection, "TRACK", - TrackID=str(current_track_id), - Name=data["title"].strip(), - Artist=data["artist"].strip(), - Kind=kind, - Location=uri, - AverageBpm=f"{data['bpm']:.2f}", - Tonality=data["key"], - TotalTime=f"{data['totalTime_sec']:.3f}") + ext = os.path.splitext(path)[1].lower() + kind = KIND_MAP.get(ext, "File") + + tr = SubElement( + collection, "TRACK", + TrackID=str(next_id), + Name=data["title"].strip(), + Artist=data["artist"].strip(), + Kind=kind, + Location=uri, + AverageBpm=f"{data['bpm']:.2f}", + Tonality=data["key"], + TotalTime=f"{data['totalTime_sec']:.3f}", + ) - is_m4a = path.lower().endswith(".m4a") + is_m4a = ext in (".m4a", ".alac") sr = data.get("sample_rate", 0) delay = (2 * 1024 / sr) if (is_m4a and sr) else 0.0 + + # Beatgrid TEMPO elements raw_grid = data.get("beatgrid") seg_positions, seg_bpms = [], [] @@ -139,312 +481,366 @@ def generate_rekordbox_xml(processed_data, all_tracks_in_tracks): seg_positions.append(float(terminal["position"])) seg_bpms.append(float(terminal.get("bpm", data["bpm"]))) - else: seg_positions, seg_bpms = [data.get("first_beat_pos_sec") or 0.0], [data["bpm"]] - elif isinstance(raw_grid, list) and raw_grid: seg_positions, seg_bpms = [float(raw_grid[0])], [data["bpm"]] - else: seg_positions, seg_bpms = [0.0], [data["bpm"]] for pos, bpm_val in zip(seg_positions, seg_bpms): if is_m4a: pos += M4A_BEATGRID_OFFSET - pos += delay / 1000.0 SubElement(tr, "TEMPO", Inizio=f"{pos:.3f}", Bpm=f"{bpm_val:.2f}", Battito="1") + # Hot cue POSITION_MARK elements for cue in data.get("hot_cues", []): sec = cue["position_ms"] / 1000.0 - if is_m4a: sec += M4A_HOTCUE_OFFSET - r, g, b = (int(cue["color"][i:i + 2], 16) for i in (1, 3, 5)) - - SubElement(tr, "POSITION_MARK", - Name=cue["name"], Type="0", - Start=f"{sec:.3f}", Num=str(cue["index"]), - Red=str(r), Green=str(g), Blue=str(b)) - current_track_id += 1 - - root_playlist.set("Count", str(len(processed_data))) - - for plist_name, tracks in processed_data.items(): - pnode = SubElement(root_playlist, "NODE", Name=plist_name, Type="1", KeyType="0", Entries=str(len(tracks))) - - for t in tracks: + r, g, b = (int(cue["color"][i : i + 2], 16) for i in (1, 3, 5)) + SubElement( + tr, "POSITION_MARK", + Name=cue["name"], Type="0", + Start=f"{sec:.3f}", Num=str(cue["index"]), + Red=str(r), Green=str(g), Blue=str(b), + ) + + next_id += 1 + + root_playlist.set("Count", str(len(playlists))) + + for plist_name, plist_tracks in playlists.items(): + pnode = SubElement( + root_playlist, "NODE", + Name=plist_name, Type="1", KeyType="0", Entries=str(len(plist_tracks)), + ) + for t in plist_tracks: tid = track_id_map.get(t["file_location"]) - if tid: SubElement(pnode, "TRACK", Key=str(tid)) - with open("serato2rekordbox.xml", "w", encoding="utf-8") as f: - f.write(prettify(root)) - -def find_serato_crates(serato_subcrates_path): - crate_file_paths = [] - - if not os.path.exists(serato_subcrates_path): - print(f"Error: Serato subcrates folder path not found: {serato_subcrates_path}") - return [] - - print(f"✅ Searching for .crate files in: {serato_subcrates_path}") - for root, dirs, files in os.walk(serato_subcrates_path): - for file in files: - if file.endswith('.crate'): - full_path = os.path.join(root, file) - crate_file_paths.append(full_path) - print(f"✅ Found {len(crate_file_paths)} crate files.\n") - return crate_file_paths - -def extract_file_paths_from_crate(crate_file_path, encoding: str = "utf-16-be"): - paths: list[str] = [] - seen: set[str] = set() - - try: - with open(crate_file_path, "rb") as f: - blob = f.read() - - blob_len = len(blob) - i = 0 - - while i < blob_len - START_MARKER_FULL_LENGTH: - marker_idx = blob.find(START_MARKER, i) - if marker_idx == -1: - break - - i = marker_idx + len(START_MARKER) - - # read 4-byte BE length - if i + PATH_LENGTH_OFFSET > blob_len: - unsuccessfulConversions.append({ - "type": "crate_parse_error", - "path": crate_file_path, - "error": f"Unexpected EOF after marker at byte {marker_idx}" - }) - break - - path_len = struct.unpack(">I", blob[i : i + PATH_LENGTH_OFFSET])[0] - i += PATH_LENGTH_OFFSET - - if i + path_len > blob_len: - unsuccessfulConversions.append({ - "type": "crate_parse_error", - "path": crate_file_path, - "error": f"Path size {path_len} exceeds remaining data at byte {i}" - }) - break - - raw_path = blob[i : i + path_len] - i += path_len # advance for next loop - + xml_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "serato2rekordbox.xml") + with open(xml_path, "w", encoding="utf-8") as fh: + fh.write(prettify(root)) + + return xml_path + + +# ── Main ───────────────────────────────────────────────────────────────────── + + +def main(): + parser = argparse.ArgumentParser( + description="Convert a Serato DJ library to Rekordbox XML.", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog="""\ +Examples: + # Everything on a USB drive — one command: + python serato2rekordbox.py /Volumes/Gianmarco + + # Auto-detect Serato folder on this Mac: + python serato2rekordbox.py + + # Serato metadata in one place, music files on a USB: + python serato2rekordbox.py --serato ~/Music/_Serato_ --music /Volumes/USB + + # Skip crate selection and convert everything: + python serato2rekordbox.py /Volumes/Gianmarco --skip-crate-selection + + # Custom output file: + python serato2rekordbox.py /Volumes/Gianmarco --output ~/Desktop/my_library.xml +""", + ) + parser.add_argument( + "root", + nargs="?", + default=None, + help="Single path to a USB drive or folder containing both " + "_Serato_/ (metadata) and your music files. " + "If used, --serato and --music are auto-detected from this path.", + ) + parser.add_argument( + "--serato", + default=None, + help="Path to the Serato library folder (contains subcrates/). " + "Auto-detected when using positional , otherwise defaults " + "to ~/Music/_Serato_.", + ) + parser.add_argument( + "--music", + default=None, + help="Root path of the music files (e.g. a USB folder with songs). " + "Auto-detected when using positional , otherwise defaults " + "to the Serato folder (original library location).", + ) + parser.add_argument( + "--output", + default=None, + help="Output XML file path. Defaults to serato2rekordbox.xml beside this script.", + ) + parser.add_argument( + "--skip-crate-selection", + action="store_true", + help="Skip the interactive crate selector and import all crates " + "(default behaviour when omitted is to open the selector).", + ) + + args = parser.parse_args() + + # Check for updates (non-blocking) — skip for local/dev builds + if not _IS_LOCAL_BUILD: + check_for_updates() + + # ── Resolve paths ──────────────────────────────────────────────────── + + serato_path = args.serato + music_root = args.music + + # If a positional root was given, auto-detect serato + music from it + if args.root: + root = os.path.expanduser(args.root) + if not os.path.isdir(root): + print(f"Error: Root path not found: {root}") + return + + # Find _Serato_ inside root (or root itself if it has subcrates/) + if not serato_path: + serato_in_root = os.path.join(root, "_Serato_") + if os.path.isdir(serato_in_root): + serato_path = serato_in_root + elif os.path.isdir(os.path.join(root, "subcrates")): + serato_path = root + + # Find music folders: all top-level dirs in root except _Serato_, + # _Serato_Backup, and hidden/system dirs + if not music_root: + excluded = {"_Serato_", "_Serato_Backup", ".Spotlight-V100", + ".fseventsd", "Contents", "System Volume Information", + ".Trashes", ".com.apple.timemachine.donotpresent"} + candidates = [] try: - abs_path = raw_path.decode(encoding).strip() - except UnicodeDecodeError: - unsuccessfulConversions.append({ - "type": "crate_decode_error", - "path": crate_file_path, - "error": f"Failed UTF-16 decode at byte {i - path_len}" - }) - continue - - # keep only the first occurrence of any duplicate - if abs_path not in seen: - paths.append(abs_path) - seen.add(abs_path) - - except FileNotFoundError: - print(f"Error: Crate file not found: {crate_file_path}") - except Exception as exc: - unsuccessfulConversions.append({ - "type": "crate_read_error", - "path": crate_file_path, - "error": str(exc) - }) - - return paths - -### Main script ### + for entry in os.scandir(root): + if entry.is_dir() and entry.name not in excluded and not entry.name.startswith("."): + candidates.append(entry.path) + except PermissionError: + pass + + if candidates: + # Build cache from all candidate dirs (union) + _combined_cache = {} + _music_labels = [] + for c in sorted(candidates): + _music_labels.append(os.path.basename(c)) + _combined_cache.update(build_path_cache(c)) + # Temporarily stash; we'll rebuild properly below + music_root = root # signal to build from all siblings + print(f"Found music folders in {root}: {', '.join(_music_labels)}") + else: + music_root = root + else: + serato_path = serato_path or find_serato_folder() -serato_base_path = find_serato_folder() + if not serato_path or not os.path.isdir(serato_path): + print("Error: Serato folder not found. Specify the path or use:") + print(" python serato2rekordbox.py /Volumes/YourUSB") + return -serato_subcrates_path = os.path.join(serato_base_path, 'subcrates') + if not music_root: + music_root = serato_path -serato_crate_paths = find_serato_crates(serato_subcrates_path) + print(f"Serato library: {serato_path}") + if music_root != serato_path: + print(f"Music files root: {music_root} (path remapping enabled)") + else: + print("Music files: same location as Serato library") -if not serato_crate_paths: - print("⚠️ No .crate files found in the subcrates folder.") - exit(0) + subcrates_path = os.path.join(serato_path, "subcrates") -track_to_crates = defaultdict(list) -all_track_paths_from_crates = set() + # ── Find crates ────────────────────────────────────────────────────── -for path in tqdm(serato_crate_paths, desc="⚙️ (1/4) Reading crate contents"): - crate_name = os.path.basename(path)[:-6] + crate_paths = find_serato_crates(subcrates_path) + if not crate_paths: + print("No .crate files found. Nothing to do.") + return - try: - formatted_crate_name = crate_name.split('%%')[0] + " [" + crate_name.split('%%')[1] + "]" - except IndexError: - formatted_crate_name = crate_name - except Exception as e: - unsuccessfulConversions.append({'type': 'crate_name_format_error', 'path': path, 'error': f"Error formatting crate name: {e}"}) - formatted_crate_name = crate_name + # Build (path, display_name) list for selector + crate_entries = [ + (p, format_crate_name(os.path.basename(p)[:-6])) + for p in crate_paths + ] - paths_in_crate = extract_file_paths_from_crate(path) - for track_path in paths_in_crate: + # ── Crate selection ────────────────────────────────────────────────── - normalized_path = track_path.replace('\\', os.sep) - if platform.system() != "Windows" and not normalized_path.startswith(os.sep): - lookup_path = os.sep + normalized_path + if not args.skip_crate_selection: + if platform.system() == "Windows": + # curses is unreliable on Windows — use plain-text fallback + selected = select_crates_fallback(crate_entries) else: - lookup_path = normalized_path + try: + selected = curses.wrapper(select_crates, crate_entries) + except Exception as e: + # curses failed (e.g. macOS Secure Terminal) — use fallback + print(f"(curses unavailable: {e} — using simple selector)") + selected = select_crates_fallback(crate_entries) + + if selected is None: + print("Aborted by user.") + return + if not selected: + print("No crates selected. Nothing to do.") + return + print(f"\nConverting {len(selected)} crate(s)...\n") + crate_entries_to_process = selected + else: + crate_entries_to_process = crate_entries + + # ── Phase 1: Read crate contents ───────────────────────────────────── + + # Build a fast filename → resolved path cache + # When a positional root was given and --music wasn't set, scan all + # sibling folders (excluding _Serato_, hidden dirs, etc.) + excluded_dirs = {"_Serato_", "_Serato_Backup", ".Spotlight-V100", + ".fseventsd", "Contents", "System Volume Information", + ".Trashes", ".com.apple.timemachine.donotpresent"} + if args.root and not args.music: + path_cache = {} + try: + for entry in os.scandir(args.root): + if entry.is_dir() and entry.name not in excluded_dirs and not entry.name.startswith("."): + path_cache.update(build_path_cache(entry.path)) + except PermissionError: + pass + else: + path_cache = build_path_cache(music_root) - track_to_crates[lookup_path].append(formatted_crate_name) - all_track_paths_from_crates.add(lookup_path) + if path_cache: + print(f"Path cache built: {len(path_cache)} file(s) indexed\n") -all_tracks_in_tracks = {} + track_to_crates = defaultdict(list) + all_track_paths = set() + crate_errors = [] -for full_system_path in tqdm(all_track_paths_from_crates, desc="⚙️ (2/4) Processing tracks"): - if not os.path.exists(full_system_path): - unsuccessfulConversions.append({'type': 'file_not_found', 'path': full_system_path, 'error': 'File not found'}) - continue + for crate_path, display_name in tqdm(crate_entries_to_process, desc="(1/4) Reading crates"): + raw_paths, errors = extract_file_paths_from_crate(crate_path) + crate_errors.extend(errors) - try: - file_extension = os.path.splitext(full_system_path)[1].lower() - extracted_data = None + for raw in raw_paths: + resolved = resolve_track_path(raw, music_root, path_cache) + if resolved: + track_to_crates[resolved].append(display_name) + all_track_paths.add(resolved) - if file_extension == '.mp3': - extracted_data = extract_mp3.extract_metadata(full_system_path) + if not all_track_paths: + print("No tracks resolved from selected crates. Check --music path.") + return - elif file_extension == '.m4a': - extracted_data = extract_m4a.extract_metadata(full_system_path) + # ── Phase 2: Extract metadata ──────────────────────────────────────── - elif file_extension == '.wav': - extracted_data = extract_wav.extract_metadata(full_system_path) + all_tracks = {} + errors = list(crate_errors) - else: - unsuccessfulConversions.append({'type': 'unsupported_format', 'path': full_system_path, 'error': f"Unsupported format: {file_extension}"}) + for path in tqdm(all_track_paths, desc="(2/4) Processing tracks"): + ext = os.path.splitext(path)[1].lower() + if ext not in EXTRACTOR_MAP: + errors.append(("unsupported_format", path, f"Unsupported format: {ext}")) continue - metadata = extracted_data.get('metadata', {}) - hot_cues = extracted_data.get('hot_cues', []) - beatgrid = extracted_data.get('beatgrid') - - all_tracks_in_tracks[full_system_path] = { - 'file_location': full_system_path, - 'title': metadata.get('title', os.path.basename(full_system_path)), - 'artist': metadata.get('artist', 'Unknown Artist'), - 'bpm': metadata.get('bpm', 0.0), - 'key': metadata.get('key', 'Unknown'), - 'totalTime_sec': metadata.get('duration_sec', 0), - 'hot_cues': hot_cues, - 'beatgrid': beatgrid, - 'sample_rate': metadata.get('sample_rate', 0) - } - - except Exception as e: - unsuccessfulConversions.append({'type': 'processing_error', 'path': full_system_path, 'error': f"{e}"}) - - -processedSeratoFiles: "OrderedDict[str, list]" = OrderedDict() - -for crate_path in tqdm(serato_crate_paths, - desc="⚙️ (3/4) Structuring Playlists"): - raw_name = os.path.basename(crate_path)[:-6] # strip ".crate" - - segments = raw_name.split("%%") - if len(segments) == 1: - crate_display_name = segments[0] # flat crate - else: - crate_display_name = segments[0] + "".join( - f" [{seg}]" for seg in segments[1:] - ) - - processedSeratoFiles[crate_display_name] = [] - - # Re-read paths *in crate order* so the playlist keeps Serato's sequence. - ordered_paths = extract_file_paths_from_crate(crate_path) - - for p in ordered_paths: - # normalise path exactly the same way as earlier - norm = p.replace("\\", os.sep) - if platform.system() != "Windows" and not norm.startswith(os.sep): - norm = os.sep + norm - - track_data = all_tracks_in_tracks.get(norm) - if track_data: - processedSeratoFiles[crate_display_name].append(track_data) + try: + extracted = EXTRACTOR_MAP[ext].extract_metadata(path) + except Exception as e: + errors.append(("processing_error", path, str(e))) + continue -# strip out any empty crates -processedSeratoFiles = { - name: tracks for name, tracks in processedSeratoFiles.items() if tracks -} + meta = extracted.get("metadata", {}) + all_tracks[path] = { + "file_location": path, + "title": meta.get("title", os.path.basename(path)), + "artist": meta.get("artist", "Unknown Artist"), + "bpm": meta.get("bpm", 0.0), + "key": meta.get("key", "Unknown"), + "totalTime_sec": meta.get("duration_sec", 0), + "hot_cues": extracted.get("hot_cues", []), + "beatgrid": extracted.get("beatgrid"), + "sample_rate": meta.get("sample_rate", 0), + } -if processedSeratoFiles: - generate_rekordbox_xml(processedSeratoFiles, all_tracks_in_tracks) + # ── Phase 3: Build playlists (preserving crate order) ──────────────── -else: - print("\nNo tracks were successfully processed. XML file not generated.") + playlists: "OrderedDict[str, list]" = OrderedDict() + for crate_path, display_name in tqdm( + [c for c in crate_entries_to_process if c[1] in + {dn for _, dn in crate_entries_to_process}], + desc="(3/4) Structuring playlists", + ): + playlists[display_name] = [] + raw_paths, _ = extract_file_paths_from_crate(crate_path) -print("\n") -print(f"✅ Found {len(all_track_paths_from_crates)} unique tracks across all crates.") -print(f'✅ {str(len(all_track_paths_from_crates) - len(unsuccessfulConversions))} / {str(len(all_track_paths_from_crates))} tracks successfully converted.') -print("\n") + for raw in raw_paths: + resolved = resolve_track_path(raw, music_root, path_cache) + if resolved and resolved in all_tracks: + playlists[display_name].append(all_tracks[resolved]) -if unsuccessfulConversions: - print(f"⚠️ {len(unsuccessfulConversions)} Unsuccessful Conversions ({len(all_track_paths_from_crates) - len(all_tracks_in_tracks)} tracks failed).") - print("⚠️ The following items could not be processed and have not been included in the XML file:") + # Drop empty crates + playlists = {n: t for n, t in playlists.items() if t} - grouped_errors = {} - for item in unsuccessfulConversions: - error_type = item.get('type', 'unknown') - if error_type not in grouped_errors: - grouped_errors[error_type] = [] - grouped_errors[error_type].append(item) + # ── Phase 4: Generate XML ──────────────────────────────────────────── - error_type_titles = { - 'file_not_found': "Files Not Found:", - 'unsupported_format': "Unsupported File Formats:", - 'processing_error': "Errors During Track Processing:", - 'beatgrid_parse_error': "Errors Parsing Beatgrid Data:", - 'crate_read_error': "Errors Reading Crate Files:", - 'crate_parse_error': "Errors Parsing Crate File Contents:", - 'crate_decode_error': "Errors Decoding Paths in Crate Files:", - 'crate_name_format_error': "Errors Formatting Crate/Playlist Names:", - 'unknown': "Other Errors:" - } + if not playlists: + print("\nNo tracks successfully processed. XML not generated.") + return - sorted_error_types = sorted(grouped_errors.keys(), key=lambda x: list(error_type_titles.keys()).index(x) if x in error_type_titles else len(error_type_titles)) + xml_path = generate_rekordbox_xml(playlists, all_tracks) - for error_type in sorted_error_types: - title = error_type_titles.get(error_type, error_type + ":") - print(f"\n{title}") - for item in grouped_errors[error_type]: - item_path = item.get('path', 'N/A') - item_error = item.get('error', 'No details') + # ── Summary ────────────────────────────────────────────────────────── - if error_type in ['file_not_found', 'unsupported_format', 'processing_error', 'beatgrid_parse_error']: + print() + print(f"Found {len(all_track_paths)} unique track(s) across selected crates.") + print(f"{len(all_tracks)} / {len(all_track_paths)} track(s) converted.") - filename = os.path.basename(item_path) + if args.output: + # Move XML to requested path + import shutil + dst = os.path.abspath(args.output) + shutil.move(xml_path, dst) + print(f"XML written to: {dst}") + else: + print(f"XML written to: {xml_path}") - crates_for_file = track_to_crates.get(item_path, []) - crate_display = ", ".join(crates_for_file) if crates_for_file else "Unknown Crate" + print() - if "not a MP4 file" in item_error: - item_error = "File appears to be invalid or corrupt" + if errors: + print(f"{len(errors)} issue(s) encountered:") - print(f'- "{filename}" ({crate_display}): {item_error}') + grouped = defaultdict(list) + for err_type, err_path, err_msg in errors: + grouped[err_type].append((err_path, err_msg)) - elif error_type in ['crate_read_error', 'crate_parse_error', 'crate_decode_error', 'crate_name_format_error']: + titles = { + "file_not_found": "Files Not Found", + "unsupported_format": "Unsupported Formats", + "processing_error": "Processing Errors", + "parse_eof": "Crate Parse Errors", + "parse_overflow": "Crate Parse Errors", + "decode": "Crate Decode Errors", + "read": "Crate Read Errors", + } - crate_filename = os.path.basename(item_path) - print(f'- Crate "{crate_filename}": {item_error}') + for etype in sorted(grouped, key=lambda t: list(titles.keys()).index(t) if t in titles else 999): + print(f"\n {titles.get(etype, etype)}:") + for epath, emsg in grouped[etype]: + if etype in ("file_not_found", "unsupported_format", "processing_error"): + fname = os.path.basename(epath) + crates = track_to_crates.get(epath, ["?"]) + print(f" - {fname} [{', '.join(crates)}]: {emsg}") + else: + crate_fname = os.path.basename(epath) + print(f" - Crate {crate_fname}: {emsg}") + else: + print("All tracks processed successfully.") - else: - print(f'- Item "{item_path}": {item_error}') -else: - print("\n✅ All tracks successfully processed.") \ No newline at end of file +if __name__ == "__main__": + main()