Add FLAC (.flac) file support

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

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