diff --git a/extract_flac.py b/extract_flac.py new file mode 100644 index 0000000..51f33e2 --- /dev/null +++ b/extract_flac.py @@ -0,0 +1,164 @@ +import base64 +import io +import logging +import re +import struct +from pathlib import Path + +import mutagen +from mutagen.flac import FLAC + +from utils import convert_key_to_camelot + +# The FLAC Serato markers format is identical to M4A: double base64 +# with an "application/octet-stream" outer wrapper and "Serato Markers2" +# header. We re-use the robust m4a parser. +from extract_m4a import parse_serato_hot_cues as _parse_markers_v2 +from extract_mp3 import parse_beatgrid_markers + +logger = logging.getLogger(__name__) + + +def _b64_clean(raw) -> bytes: + """Strip whitespace/newlines and ensure proper base64 padding.""" + if isinstance(raw, str): + raw = raw.encode("utf-8") + clean = re.sub(rb"[^a-zA-Z0-9+/=]", b"", raw) + pad = (-len(clean)) % 4 + if pad: + clean += b"=" * pad + return clean + + +def get_beatgrid_flac(audio: FLAC) -> dict: + """ + Read the Serato BeatGrid from a FLAC file's Vorbis comments. + Serato stores it under 'SERATO_BEATGRID' as base64. + The decoded payload has the same structure as the MP3 GEOB beatgrid. + """ + result = {"markers": {"non_terminal": [], "terminal": None}} + + raw = None + for key in ("SERATO_BEATGRID", "serato_beatgrid", + "SERATO_BEATGRID_BINARY", "serato_beatgrid_binary"): + raw = audio.tags.get(key) + if raw: + break + + if not raw: + return result + + if isinstance(raw, list): + raw = raw[0] + + try: + decoded = base64.b64decode(_b64_clean(raw)) + except Exception as e: + logger.warning(f"Failed to decode beatgrid base64: {e}") + return result + + # The decoded data has an "application/octet-stream\x00\x00Serato BeatGrid\x00" prefix + # followed by the raw grid data (version + markers), same as MP3 GEOB. + marker_str = b"Serato BeatGrid\x00" + idx = decoded.find(marker_str) + if idx >= 0: + grid_data = decoded[idx + len(marker_str):] + else: + # Fallback: try without the prefix + grid_data = decoded + + try: + markers = parse_beatgrid_markers(io.BytesIO(grid_data)) + except Exception as e: + logger.warning(f"Failed to parse beatgrid markers: {e}") + return result + + if not markers: + return result + + non_terminal = [] + terminal = None + + for m in markers: + # TerminalBeatgridMarker and NonTerminalBeatgridMarker are namedtuples + if hasattr(m, 'bpm'): + terminal = {"position": m.position, "bpm": m.bpm} + else: + non_terminal.append({ + "position": m.position, + "beats_till_next_marker": m.beats_till_next_marker + }) + + result["markers"]["non_terminal"] = non_terminal + result["markers"]["terminal"] = terminal + return result + + +def extract_metadata(file_path: str) -> dict: + results = { + "metadata": {}, + "hot_cues": [], + "beatgrid": {"markers": {"non_terminal": [], "terminal": None}} + } + track = Path(file_path) + + if not track.exists(): + logging.error("File not found: %s", file_path) + return results + + try: + audio = FLAC(str(track)) + except Exception as e: + logging.error(f"Could not open FLAC file {file_path}: {e}") + return results + + # --- Basic metadata --- + results["metadata"]["title"] = audio.get("title", ["Unknown Title"])[0] + results["metadata"]["artist"] = audio.get("artist", ["Unknown Artist"])[0] + + # BPM — Serato stores it as 'BPM' or 'TBPM' Vorbis comment + bpm_str = audio.get("bpm", audio.get("tbpm", ["0"]))[0] + try: + results["metadata"]["bpm"] = float(re.sub(r"[^0-9.]", "", str(bpm_str))) if bpm_str else 0.0 + except (ValueError, TypeError): + results["metadata"]["bpm"] = 0.0 + + # Key — try Serato's INITIALKEY, then standard KEY tag + classical_key = (audio.get("initialkey", [None])[0] or + audio.get("key", [None])[0] or + "Unknown") + results["metadata"]["key"] = convert_key_to_camelot(classical_key) if classical_key != "Unknown" else "Unknown" + + # Duration + results["metadata"]["duration_sec"] = round(audio.info.length, 3) + + # Sample rate + results["metadata"]["sample_rate"] = audio.info.sample_rate + + # --- Hot Cues --- + # Serato stores markers in Vorbis comments under SERATO_MARKERS_V2. + # The format is double-base64, identical to M4A. + markers_raw = None + for key in ("SERATO_MARKERS_V2", "serato_markers_v2", + "SERATO_MARKERS", "serato_markers"): + markers_raw = audio.tags.get(key) + if markers_raw: + break + + if markers_raw: + try: + # The m4a parser expects the raw (outer) base64 bytes + raw_bytes = markers_raw[0] if isinstance(markers_raw, list) else markers_raw + if isinstance(raw_bytes, str): + raw_bytes = raw_bytes.encode("utf-8") + results["hot_cues"] = _parse_markers_v2(raw_bytes) + except Exception as e: + logging.warning(f"Error parsing hot cues from {file_path}: {e}") + + # --- Beatgrid --- + try: + results["beatgrid"] = get_beatgrid_flac(audio) + except Exception as e: + logging.warning(f"Error reading beatgrid from {file_path}: {e}") + + return results diff --git a/serato2rekordbox.py b/serato2rekordbox.py index d6bd7fc..fcae745 100644 --- a/serato2rekordbox.py +++ b/serato2rekordbox.py @@ -24,6 +24,7 @@ import extract_mp3 import extract_m4a import extract_wav +import extract_flac import urllib.request import ssl @@ -105,7 +106,10 @@ 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" + kind = ("MP3 File" if path.lower().endswith(".mp3") + else "M4A File" if path.lower().endswith(".m4a") + else "FLAC File" if path.lower().endswith(".flac") + else "WAV File") tr = SubElement(collection, "TRACK", TrackID=str(current_track_id), @@ -324,6 +328,9 @@ def extract_file_paths_from_crate(crate_file_path, encoding: str = "utf-16-be"): elif file_extension == '.wav': extracted_data = extract_wav.extract_metadata(full_system_path) + elif file_extension == '.flac': + extracted_data = extract_flac.extract_metadata(full_system_path) + else: unsuccessfulConversions.append({'type': 'unsupported_format', 'path': full_system_path, 'error': f"Unsupported format: {file_extension}"}) continue