main file
+878 · 1 file changed
-
music_sort_renamer-new-7.py+878 −0
@@ -0,0 +1,878 @@ +#!/usr/bin/env python3 +""" +Music Sample Sorter & Renamer (Robust) - v7 +=========================================== + +CHANGES IN v7: +- Fixed BPM parsing: Leading numeric tokens (indices) are now properly excluded +- Improved BPM detection: Better heuristics to distinguish indices from BPM +- Added token position tracking for more accurate parsing +- Enhanced logging for BPM detection (with --debug flag) +- Code cleanup and better documentation + +Normalizes names AND sorts files from a vendor pack (source) into your canonical +library (target) folder structure. Incorporates specific rules and mappings +to reduce unknowns from abstract/synth packs. Includes refined BPM detection +to avoid misinterpreting indices/names, and improved descriptor retention. + +Core filename schema: + Type_Descriptors_BPM_Key_Vendor_Index.ext +All lowercase, underscore-separated, with optional parts omitted cleanly. + +Key features: +- Moves/copies/hardlinks files from --source-root to --target-root based on inferred type +- CRC32 de-duplication (skip or quarantine) +- Generalized vendor detection with specific naming for "From Mars" packs (e.g., 'lindrum_fmrs') +- Improved type inference (more synth/perc/fx keywords, aggressive synth guessing, path fallbacks, pack defaults) +- Alpha-numeric splitting for better token matching (e.g., MicroMoog23 -> micromoog, 23) +- Cleaner descriptors (whitelisted + machine tokens; junk filtered + better word retention) +- **FIXED BPM parsing**: Properly excludes leading index numbers from being misidentified as BPM +- Safer key parsing (ignores bare note+octave like "c2") +- Externalizable maps via --config JSON (though many common ones now built-in) +- Dry-run by default, CSV audit log, collision-safe naming, summary counts + +Usage (dry-run first): + python music_sort_renamer-v7.py \ + --source-root "/path/to/Vendor Pack" \ + --target-root "/100-Sound-Library/MUSIC" \ + --csv report.csv \ + --dedupe quarantine \ + --copy \ + --verbose + +Apply for real (move): + python music_sort_renamer-v7.py \ + --source-root "/path/to/Vendor Pack" \ + --target-root "/100-Sound-Library/MUSIC" \ + --csv report.csv \ + --dedupe skip \ + --write \ + --verbose +""" + +import argparse +import csv +import json +import os +import re +import shutil +import sys +import zlib +from pathlib import Path +from typing import Dict, List, Optional, Set, Tuple + +# ---------------------------- Defaults & Maps ---------------------------- + +VALID_EXTS: Set[str] = {".wav", ".flac", ".aiff", ".aif", ".mp3"} + +# Where files should land inside --target-root, per inferred "type" +TARGET_DIR_MAP: Dict[str, str] = { + # One-shots + "kick": "ONE-SHOTS/Drum_Hits/Kicks", + "snare": "ONE-SHOTS/Drum_Hits/Snares", + "hat_closed": "ONE-SHOTS/Drum_Hits/Hats-Closed", + "hat_open": "ONE-SHOTS/Drum_Hits/Hats-Open", + "clap": "ONE-SHOTS/Drum_Hits/Claps", + "tom": "ONE-SHOTS/Drum_Hits/Toms", + "rim": "ONE-SHOTS/Drum_Hits/Rims", + "cowbell": "ONE-SHOTS/Drum_Hits/Cowbells", + "cymbal_crash": "ONE-SHOTS/Cymbals/Crashes", + "cymbal_ride": "ONE-SHOTS/Cymbals/Rides", + "cymbal_splash": "ONE-SHOTS/Cymbals/Splashes", + "perc": "ONE-SHOTS/Percussion/Misc", # Default perc folder + "stab_synth": "ONE-SHOTS/Stabs/Synth", + "stab_piano": "ONE-SHOTS/Stabs/Piano", + "stab_guitar": "ONE-SHOTS/Stabs/Guitar", + "stab_brass": "ONE-SHOTS/Stabs/Brass", + "stab_string": "ONE-SHOTS/Stabs/Strings", + "vocalcut": "ONE-SHOTS/Vocals/Cuts", + "fx": "ONE-SHOTS/FX/Misc", # Default FX folder + "fx_riser": "ONE-SHOTS/FX/Risers", + "fx_impact": "ONE-SHOTS/FX/Impacts", + "fx_noise": "ONE-SHOTS/FX/Noise", + "bass_808": "ONE-SHOTS/Bass/808", + "bass_synth": "ONE-SHOTS/Bass/Synth", + "bass_guitar": "ONE-SHOTS/Bass/Guitar", + + # Loops + "drumloop": "LOOPS/Drums", + "toploop": "LOOPS/Drums/Tops", + "percloop": "LOOPS/Percussion", + "bassloop": "LOOPS/Bass", + "melodicloop": "LOOPS/Melodic", + "fxloop": "LOOPS/FX", + "vocalloop": "LOOPS/Vocals", + "drum_fill": "LOOPS/Drums/Fills", + + # Instruments (for multi-sampled patches) + "instrument_key": "INSTRUMENTS/Keys", + "instrument_synth": "INSTRUMENTS/Synths", + "instrument_string": "INSTRUMENTS/Strings", + "instrument_brass": "INSTRUMENTS/Brass_Winds", + "instrument_world": "INSTRUMENTS/World", + "instrument_vocal": "INSTRUMENTS/Vocals", + "instrument_perc": "INSTRUMENTS/Percussion", + "instrument_guitar": "INSTRUMENTS/Guitars", + + # Fallback + "unknown": "UNSORTED/_UnknownType", +} + +# Type tokens from filenames (Expanded v5) +TYPE_MAP: Dict[str, str] = { + # Drums + "bd": "kick", "kick": "kick", + "sd": "snare", "snare": "snare", + "ch": "hat_closed", "chh": "hat_closed", "hh": "hat_closed", "hat_closed": "hat_closed", "closedhat": "hat_closed", + "oh": "hat_open", "ohh": "hat_open", "hat_open": "hat_open", "openhat": "hat_open", + "rim": "rim", "rimshot": "rim", "sidestick": "rim", + "clap": "clap", + "tom": "tom", "racktom": "tom", "floortom": "tom", + "cowbell": "cowbell", + "crash": "cymbal_crash", "ride": "cymbal_ride", "splash": "cymbal_splash", "cymbal": "cymbal_crash", "china": "cymbal_crash", + # Percussion + "perc": "perc", "percussion": "perc", + "shaker": "perc", "maracas": "perc", + "conga": "perc", "bongo": "perc", "timbale": "perc", "djembe": "perc", "udu": "perc", + "clave": "perc", "agogo": "perc", "triangle": "perc", "bell": "perc", "gong": "perc", + "tamb": "perc", "tambourine": "perc", + "cabasa": "perc", + "quijada": "perc", + "woodblock": "perc", "block": "perc", + "guiro": "perc", + "click": "perc", + "snap": "perc", + "wendel": "perc", # From reducer config + # Other Drums + "fill": "drum_fill", "drum_fill": "drum_fill", + # Vocals + "vox": "vocalcut", "vocal": "vocalcut", "choir": "vocalcut", "voice": "vocalcut", "vocalcut": "vocalcut", "chant": "vocalcut", + # Stabs + "stab": "stab_synth", # Default to synth + "stab_synth": "stab_synth", "stab_piano": "stab_piano", "stab_guitar": "stab_guitar", + "stab_brass": "stab_brass", "stab_string": "stab_string", + "brass": "stab_brass", "horn": "stab_brass", "trumpet": "stab_brass", "sax": "stab_brass", "trombone": "stab_brass", + "string": "stab_string", "violin": "stab_string", "cello": "stab_string", + "piano": "stab_piano", "keys": "stab_piano", "rhodes": "stab_piano", "organ": "stab_piano", + "guitar": "stab_guitar", + # Bass + "bass": "bass_synth", "sub": "bass_synth", + "808": "bass_808", "tr808": "bass_808", + # FX + "fx": "fx", "sfx": "fx", "effect": "fx", "effects": "fx", + "riser": "fx_riser", "rise": "fx_riser", "sweep": "fx_riser", "buildup": "fx_riser", "sweeper": "fx_riser", "upsweep": "fx_riser", + "impact": "fx_impact", "hit": "fx_impact", "boom": "fx_impact", "slam": "fx_impact", + "noise": "fx_noise", "white": "fx_noise", "pink": "fx_noise", "static": "fx_noise", + "whoosh": "fx", "transition": "fx", "reverse": "fx", "rev": "fx", + "laser": "fx", "zap": "fx", "glitch": "fx", + "ambience": "fx", "ambient": "fx", "atmos": "fx", "atmosphere": "fx", "texture": "fx", "drone": "fx", + # Loops + "loop": "drumloop", # Default to drum loop + "drumloop": "drumloop", "beat": "drumloop", "breakbeat": "drumloop", "break": "drumloop", "groove": "drumloop", + "top": "toploop", "toploop": "toploop", "hats": "toploop", "hihat": "toploop", + "percloop": "percloop", "percussionloop": "percloop", + "bassloop": "bassloop", "bassline": "bassloop", + "melodicloop": "melodicloop", "melodic": "melodicloop", "melody": "melodicloop", "chord": "melodicloop", "progression": "melodicloop", + "fxloop": "fxloop", + "vocalloop": "vocalloop", + # Synth-related + "synth": "instrument_synth", "analog": "instrument_synth", "digital": "instrument_synth", + "pad": "instrument_synth", "lead": "instrument_synth", "arp": "instrument_synth", "seq": "instrument_synth", + "saw": "instrument_synth", "square": "instrument_synth", "sine": "instrument_synth", "triangle": "instrument_synth", + "pulse": "instrument_synth", "pwm": "instrument_synth", + # Instrument categories + "key": "instrument_key", "keyed": "instrument_key", + "world": "instrument_world", "ethnic": "instrument_world", +} + +# Machine token sets (Expanded v5) +MACHINE_TOKENS: Set[str] = { + "101","202","303","505","606","707","808","909","727","626", + "dmx","oberheim","linn","lm1","lm2","lindrum","linndrum", + "dr55","dr550","dr660","sp1200","sp12","sp808","mpc","mpc60","mpc3000", + "rx5","rx7","rx11","rx15","rx21","rx21l", + "dr110","tr77","r8","r70","dd20","dd30","dd50", + "dr202","dr770","dr880", + "tempest","tanzbar","pulsar","electribe","volca", + "machinedrum","rytm","analog","analogfour","digitakt","digitone", + "drumbrute","impact", + "cr78","cr68","cr8000","fr2l","fr3l","fr5","fr6","fr7","fr8","fr9", + "srm","ace","acetone","rhythm77","drumulator","simmons","simmonssds","sds5","sds7","sds8","sds9", + "fairlight","e-mu","emu","sp12000","sp-12","quasimidi","alesis","boss","roland","korg","yamaha", + "sp16","s612","s950","s760","s770","s5000", + "dx7","d50","jd800","jx3p","jx8p","matrix","matrix1000","xpander","ems","modular","moog", + "eko","crumar","vermona","doepfer","cwejman","buchla","serge","arp","arp2600","prophet5","prophet10","cs80","minibrute", + "luna","lyra","lumatone","polybrute","matrixbrute", + "ddd1","polyvoks", "polivoks", "altair","estradin","aelita", "formanta", + "minipops","mks","juno","jp8","jp6","jp4","ms20","pro-one","prophet","odyssey", + "mirage", + "303", "tb303", + "m1", + "synare", + "viscount", + "dr bohm", "bohm", + "micromoog", "minimoog", + "vp330", # Added VP330 +} + +# Whitelisted descriptors (Expanded v5) +DESCRIPTOR_WHITELIST: Set[str] = { + "clean","color","mod","modded","tape","vinyl","sat","saturated","dist","distorted","crushed","crush", + "bright","dark","dull","long","short","decay","verb","reverb","gate","gated","thin","thick","dirt", + "res","resonant","resonance","grit","gritty","lofi","hi","lo","mid","combo","classic","orig", + "pitched","pitch","bend","bendy","ringmod","ring","noise","alien","glitch","static","hollow", + "room","wide","mono","stereo","deep","punchy","click","snap","airy","woody","snappy","fat","warm", + "acoustic", "analog", "digital", + "accent", "flam", "mute", "open", "closed", + "dry", "wet", + "sub", + "square", "saw", "sine", "triangle", "pulse", "pwm", "fm", + "organ", "lead", "pad", "arp", "poly", "mono", + "tube", "echo", "space", "drone", "texture", "atmos", "atmosphere" +} + +# Unhelpful tokens to drop (Expanded v5) +JUNK_WORDS: Set[str] = { + "various","repaired","fix","fixed","sample","samples","one","shot","oneshot","hit","hits", + "drum","drums","loop","loops","from","mars","the","and","of","in","mix","mixed","stem","stems", + "wav", "aif", "aiff", "flac", "mp3", + "kit", "kits", "individual", "hits", + "legacy", "snacks", "essential", "reserve", "creations", "fragments", + "misc", + "instrument", "instruments", "presets", + "di", + "main", "full", + "sfm", "sfm-", + "micro", "mini", # Keep pack names out of descriptors + "pack", "preset", "presets", + "fragment", "fragments", + "oddities", "oddity", "modern", + "private", "reserve", + "micromoog", "minimoog", # Keep machine names out unless explicitly whitelisted (handled in normalize_descriptors) +} + +# Pack default type fallbacks (Suggestion #2) +PACK_DEFAULT_TYPE: Dict[str, str] = { + "mini": "instrument_synth", + "micro": "instrument_synth", + "pulsar": "instrument_synth", + "modular": "instrument_synth", # Applies to Modular Drums and Modular Creations + "found": "fx", + "databenders": "fx", + "ekko": "fx", + "fragments": "fx", # Applies to Tape Fragments, Trumpet Fragments + "oddities": "fx", + "wendel": "perc", + "acid": "instrument_synth", # Bias Acid pack unknowns towards synth + "viscount": "instrument_key", + "360": "instrument_key", + "arp omni": "instrument_string", + "vp330": "instrument_string", + "mirage": "instrument_key", + "s612": "instrument_key", + "s950": "instrument_key", + "synare": "fx", + # Add more defaults as needed based on remaining unknowns +} + + +# Vendor naming logic (Corrected v3 - Keep this version) +def normalize_vendor_name(name: str) -> str: + """Cleans up vendor name, shortens 'From Mars' variations.""" + s = name.strip().lower().replace("-", " ").replace(".", " ").replace("_", " ") + s = s.replace("samples", "").replace("essential wav", "").replace("drum machines", "").replace("drums", "") + s = s.replace("synths", "").replace("loops", "") + s = re.sub(r"\s+", " ", s).strip() + + if "from mars" in s: + parts = s.split(" from mars") + specific_name = parts[0].strip() + specific_name = re.sub(r"[^a-z0-9\s]", "", specific_name).strip() + specific_name = re.sub(r"\s+", "_", specific_name) + # Ensure it doesn't result in just "_fmrs" if specific_name was empty + return f"{specific_name}_fmrs" if specific_name else "frommars" + else: + s = re.sub(r"[^a-z0-9\s]", "", s).strip() + s = re.sub(r"\s+", "_", s) + return s if s else "unknown_vendor" + +# ---------------------------- Regex Helpers ----------------------------- + +RE_BPM = re.compile(r"(?<!\w)(\d{2,3})\s*(?:bpm)?(?!\w)", re.IGNORECASE) +RE_NOTE_TOKEN = re.compile(r"(?<![a-zA-Z0-9#])([a-gA-G])([#b])?([-]?\d+)?(m)?(?![a-zA-Z0-9#])") +RE_MINOR_WORD = re.compile(r"\bmin(?:or)?\b", re.IGNORECASE) +RE_MAJOR_WORD = re.compile(r"\bmaj(?:or)?\b", re.IGNORECASE) +FORBIDDEN_CHARS = re.compile(r"[^a-z0-9_\.]") +SUSPECT_KEYS = {f"{n}{o}" for n in "abcdefg" for o in "0123456789"} + +# ---------------------------- Utilities --------------------------------- + +def split_alpha_num(tokens: List[str]) -> List[str]: + """Splits tokens like 'MicroMoog23' into ['micromoog', '23'].""" + out = [] + for t in tokens: + split_parts = re.sub(r'(?<=\D)(?=\d)|(?<=\d)(?=\D)', ' ', t).split() + out.extend(p.lower() for p in split_parts if p) + return out + +def tokenize(s: str) -> List[str]: + # Basic tokenization + s = s.replace("-", " ").replace("(", " ").replace(")", " ") + s = s.replace("&", " ").replace(",", " ").replace("[", " ").replace("]", " ") + s = re.sub(r"[_]+", " ", s) + s = re.sub(r"\s+", " ", s) + tokens = [t for t in s.strip().split(" ") if t] + # Apply alpha-numeric splitting here + return split_alpha_num(tokens) + +def clean_tokens(tokens: List[str]) -> List[str]: + out = [] + for t in tokens: + t = t.strip().lower() + if not t: continue + if '#' not in t and '.' in t and not re.match(r'^\d+\.\d+$', t): + t = t.replace(".", "") + elif '#' in t: + parts = t.split('.') + t = ''.join(parts) + if t: out.append(t) + return out + + +def file_crc32(p: Path, chunk=1024*256) -> str: + crc = 0 + try: + with p.open("rb") as f: + while True: + b = f.read(chunk) + if not b: break + crc = zlib.crc32(b, crc) + return f"{crc:08x}" + except Exception as e: + print(f"!! Error calculating CRC for {p}: {e}", file=sys.stderr) + return "error" + + +# ------------------------- Extraction Functions ------------------------- + +# v7: FIXED BPM Extraction - properly excludes leading indices +def extract_bpm(tokens: List[str], machine_tokens_set: Set[str], debug: bool = False) -> Tuple[Optional[str], List[str]]: + """ + Extracts BPM, prioritizing explicit 'bpm' suffix or standalone numbers, + while avoiding numbers that are likely machine names or indices. + + v7 FIX: Properly identifies and excludes leading index numbers + """ + bpm = None + consumed_indices = set() + original_length = len(tokens) + + if debug: + print(f" [BPM DEBUG] Input tokens: {tokens}") + + # Identify potential leading index (first token that's purely numeric and < 200) + leading_index = None + if len(tokens) > 0 and tokens[0].isdigit(): + val = int(tokens[0]) + # Common indices are typically 1-999, but BPM is 40-300 + # If it's in BPM range but is the ONLY number, it's likely an index + if val < 1000: + leading_index = 0 + if debug: + print(f" [BPM DEBUG] Token at position 0 ('{tokens[0]}') identified as potential leading index") + + # Pass 1: Explicit 'bpm' suffix or full match standalone number + for i, t in enumerate(tokens): + # Skip if this is the identified leading index + if i == leading_index: + if debug: + print(f" [BPM DEBUG] Skipping position {i} ('{t}') - identified as leading index") + continue + + if t.endswith('bpm') and t[:-3].isdigit(): + v = int(t[:-3]) + if 40 <= v <= 300 and bpm is None: + bpm = t + consumed_indices.add(i) + if debug: + print(f" [BPM DEBUG] Found explicit BPM at position {i}: {t}") + break + + m_full = RE_BPM.fullmatch(t) + if m_full and bpm is None: + v = int(m_full.group(1)) + if 40 <= v <= 300: + # Check if this number is also a known machine token + if str(v) not in machine_tokens_set: + bpm = f"{v}bpm" + consumed_indices.add(i) + if debug: + print(f" [BPM DEBUG] Found BPM via fullmatch at position {i}: {bpm}") + break + + # Pass 2: Search within tokens, but with stricter checks + if bpm is None: + for i, t in enumerate(tokens): + # Skip if already consumed or is the leading index + if i in consumed_indices or i == leading_index: + if debug and i == leading_index: + print(f" [BPM DEBUG] Skipping position {i} ('{t}') - leading index") + continue + + m_search = RE_BPM.search(t) + if m_search: + num_str = m_search.group(1) + v = int(num_str) + + if debug: + print(f" [BPM DEBUG] Found number '{num_str}' at position {i} in token '{t}'") + + # --- Stricter Checks --- + # 1. Is it a known machine token number? + if num_str in machine_tokens_set: + if debug: + print(f" [BPM DEBUG] -> Rejected: matches machine token") + continue + + # 2. Is it likely an index (first or last token, purely numeric)? + is_first_or_last = (i == 0 or i == original_length - 1) + if is_first_or_last and t == num_str: + if debug: + print(f" [BPM DEBUG] -> Rejected: pure number at first/last position (likely index)") + continue + + # 3. If it passes checks and is in range: + if 40 <= v <= 300: + bpm = f"{v}bpm" + if debug: + print(f" [BPM DEBUG] -> Accepted as BPM: {bpm}") + break + elif debug: + print(f" [BPM DEBUG] -> Rejected: outside BPM range (40-300)") + + keep = [tokens[i] for i in range(len(tokens)) if i not in consumed_indices] + + if debug: + print(f" [BPM DEBUG] Final BPM: {bpm}") + print(f" [BPM DEBUG] Remaining tokens: {keep}") + + return bpm, keep + + +def extract_key(tokens: List[str]) -> Tuple[Optional[str], List[str]]: + key = None + found_note_data = None + is_minor = any(RE_MINOR_WORD.search(t) for t in tokens) + is_major = any(RE_MAJOR_WORD.search(t) for t in tokens) + + original_tokens = list(tokens) + kept_indices = set(range(len(tokens))) + + for i, t in enumerate(original_tokens): + if i not in kept_indices: continue + if t in SUSPECT_KEYS: continue # Skip 'c2', 'a1' etc. + + m = RE_NOTE_TOKEN.search(t) + if m: + note_match_str = (m.group(1) + (m.group(2) or "")).lower() + octave_part = m.group(3) + minor_flag = m.group(4) + matched_text = m.group(0) + + # Refined Ambiguity Check: Ensure matched text isn't just note+digit + # Also check if the *token itself* is suspicious if match isn't whole token + is_ambiguous_match = (matched_text in SUSPECT_KEYS or + (len(note_match_str) == 1 and octave_part and not m.group(2) and not minor_flag)) + if is_ambiguous_match: + # If ambiguous part *is* the whole token, definitely skip + if matched_text == t: continue + # If ambiguous part is *within* a larger token, maybe still allow? + # Let's keep skipping for now for safety. + continue + + + if found_note_data is None: + found_note_data = (note_match_str, i, matched_text) + else: + # If we found a more specific match (with sharps/flats/octave), prefer it + prev_note, prev_i, prev_matched = found_note_data + if len(note_match_str) > len(prev_note) or octave_part: + found_note_data = (note_match_str, i, matched_text) + + if found_note_data: + note_str, idx, matched = found_note_data + if is_minor: key = f"{note_str}m" + elif is_major: key = f"{note_str}maj" + else: key = note_str + kept_indices.discard(idx) + + remaining = [original_tokens[i] for i in sorted(kept_indices)] + return key, remaining + + +def normalize_descriptors(descriptors: List[str], whitelist: Set[str], machine_toks: Set[str], junk: Set[str]) -> List[str]: + """Normalize and filter descriptors, keeping machine tokens + whitelisted words, removing junk.""" + out = [] + for d in descriptors: + d_lower = d.lower().strip() + if not d_lower or len(d_lower) < 2: continue + if d_lower in junk: continue + if d_lower in machine_toks or d_lower in whitelist: + out.append(d_lower) + elif d_lower.isdigit(): # Keep small numeric descriptors (e.g., velocity layer numbers?) + # Filter out pure indices (larger than some threshold) + v = int(d_lower) + if v < 100: # Arbitrary threshold for "descriptive" numbers vs indices + out.append(d_lower) + else: + # Keep words that are 3+ chars and not obviously junk (heuristic) + if len(d_lower) >= 3: + out.append(d_lower) + return out + + +def infer_type_from_tokens(tokens: List[str], type_map: Dict[str, str], pack_default: str) -> Tuple[str, List[str]]: + """ + Infer file type from tokens. Returns (type, remaining_tokens_as_descriptors). + Prioritizes specific keywords first, falls back to pack defaults and heuristics. + """ + matched_type = None + consumed_indices = set() + + # Pass 1: Look for explicit type tokens + for i, t in enumerate(tokens): + t_lower = t.lower().strip() + if t_lower in type_map: + matched_type = type_map[t_lower] + consumed_indices.add(i) + break # Use first match + + # Pass 2: If no match and pack has a default type for unknowns + if matched_type is None and pack_default: + matched_type = pack_default + + # Pass 3: Aggressive synth inference for abstract packs + if matched_type is None: + # Look for synth-like keywords + synth_keywords = {"synth", "analog", "digital", "pad", "lead", "bass", "saw", "square", "pulse"} + if any(t.lower() in synth_keywords for t in tokens): + matched_type = "instrument_synth" + + remaining = [tokens[i] for i in range(len(tokens)) if i not in consumed_indices] + return matched_type or "unknown", remaining + + +def infer_type_from_path(rel_path: Path, type_map: Dict[str, str]) -> Optional[str]: + """ + Infer type from directory names in the relative path. + """ + for part in rel_path.parts: + part_lower = part.lower().strip() + if part_lower in type_map: + return type_map[part_lower] + return None + + +def extract_vendor_from_path(rel_path: Path) -> str: + """ + Extract vendor name from the first directory in the relative path. + """ + if rel_path.parts: + vendor_raw = rel_path.parts[0] + return normalize_vendor_name(vendor_raw) + return "unknown_vendor" + + +def build_normalized_name( + file_type: str, + descriptors: List[str], + bpm: Optional[str], + key: Optional[str], + vendor: str, + index_str: Optional[str], + ext: str +) -> str: + """ + Constructs the normalized filename in the schema: + Type_Descriptors_BPM_Key_Vendor_Index.ext + """ + parts = [file_type] + + if descriptors: + parts.append("_".join(descriptors)) + + if bpm: + parts.append(bpm) + + if key: + parts.append(key) + + if vendor: + parts.append(vendor) + + if index_str: + parts.append(index_str) + + name = "_".join(parts) + # Sanitize: remove any remaining forbidden characters + name = FORBIDDEN_CHARS.sub("", name) + # Remove multiple underscores + name = re.sub(r"_+", "_", name) + name = name.strip("_") + + return f"{name}{ext}" + + +# ------------------------- Main Processing Function ------------------------- + +def process_file( + source_path: Path, + source_root: Path, + target_root: Path, + writer: csv.DictWriter, + opts: Dict, + counters: Dict +): + """ + Process a single file: extract metadata, build target path, handle duplicates, write to CSV, and execute action. + """ + try: + rel_path = source_path.relative_to(source_root) + except ValueError: + rel_path = source_path + rel_str = str(rel_path) + + # --- CRC Check --- + crc = "off" + if opts["dedupe_mode"] != "off": + crc = file_crc32(source_path) + if crc == "error": + print(f"!! CRC error for {rel_str}, skipping deduplication.", file=sys.stderr) + elif crc in opts["crc_db"]: + dup_path = opts["crc_db"][crc] + if opts["dedupe_mode"] == "skip": + if opts["verbose"]: + print(f"[SKIP] {rel_str} (duplicate of {dup_path})") + counters["dup_skipped"] += 1 + writer.writerow({"old_path": str(source_path), "new_path": "(skipped duplicate)", "type": "dup_skip", "descriptors": "", "bpm": "", "key": "", "vendor": "", "index": "", "collision_version": "", "dry_run": "yes" if opts["dry_run"] else "no"}) + return + elif opts["dedupe_mode"] == "quarantine": + # Mark for quarantine but continue processing + pass + + # --- Extract metadata --- + stem = source_path.stem + ext = source_path.suffix.lower() + if ext == ".aif": ext = ".aiff" + + # Tokenize filename + tokens_raw = tokenize(stem) + tokens = clean_tokens(tokens_raw) + + # Identify potential index (leading numeric token) + idx = None + if tokens and tokens[0].isdigit(): + idx = tokens[0] + tokens = tokens[1:] # Remove index from tokens for further processing + + # Extract vendor from path + vendor = extract_vendor_from_path(rel_path) + + # Get pack-specific defaults + pack_default = None + for pack_key in opts["pack_default_map"]: + if pack_key in vendor or pack_key in rel_str.lower(): + pack_default = opts["pack_default_map"][pack_key] + break + + # Extract BPM (v7: with improved logic) + bpm, tokens = extract_bpm(tokens, opts["machine_tokens"], debug=opts.get("debug", False)) + + # Extract Key + key, tokens = extract_key(tokens) + + # Infer type from tokens + file_type, descriptor_tokens = infer_type_from_tokens(tokens, opts["type_map"], pack_default) + + # Fallback: infer type from path + if file_type == "unknown": + path_type = infer_type_from_path(rel_path, opts["type_map"]) + if path_type: + file_type = path_type + + # Normalize descriptors + descriptors = normalize_descriptors(descriptor_tokens, opts["descriptor_whitelist"], opts["machine_tokens"], opts["junk_words"]) + + # Build target path + target_dir = opts["target_dir_map"].get(file_type, opts["target_dir_map"]["unknown"]) + + # Handle duplicates going to quarantine + quarantine = False + if opts["dedupe_mode"] == "quarantine" and crc != "off" and crc != "error" and crc in opts["crc_db"]: + target_dir = "UNSORTED/Duplicates" + quarantine = True + report_type = f"dup_{file_type}" + else: + report_type = file_type + + # Build normalized filename + new_name = build_normalized_name(file_type, descriptors, bpm, key, vendor, idx, ext) + + target_path = target_root / target_dir / new_name + + # --- Collision handling --- + collision_v = 0 + collision_v_str = "" + temp_target_path = target_path + while temp_target_path.exists(): + collision_v += 1 + collision_v_str = f"{collision_v:02d}" + name_part = target_path.stem + new_stem = f"{name_part}_v{collision_v_str}" + temp_target_path = target_path.parent / f"{new_stem}{ext}" + if collision_v_str: target_path = temp_target_path + + # --- Write CSV Row --- + writer.writerow({"old_path": str(source_path), "new_path": str(target_path), "type": report_type, "descriptors": "_".join(descriptors), "bpm": bpm or "", "key": key or "", "vendor": vendor or "", "index": idx or "", "collision_version": collision_v_str, "dry_run": "yes" if opts["dry_run"] else "no"}) + + # --- Verbose Output --- + if opts["verbose"]: + mode = "DRY" if opts["dry_run"] else ("COPY" if opts["copy"] else ("HLINK" if opts["hardlink"] else "MOVE")) + status = " (Quarantined)" if quarantine else "" + collision_note = f" (v{collision_v_str})" if collision_v_str else "" + try: target_rel_path = target_path.relative_to(target_root) + except ValueError: target_rel_path = target_path + print(f"[{mode}] {rel_str} -> {target_rel_path}{status}{collision_note}") + + # --- Perform Action --- + if not opts["dry_run"]: + try: + target_path.parent.mkdir(parents=True, exist_ok=True); operation_done = False + if opts["copy"]: shutil.copy2(source_path, target_path); operation_done = True + elif opts["hardlink"]: + try: + if not target_path.parent.exists(): target_path.parent.mkdir(parents=True, exist_ok=True) + source_dev = os.stat(source_path).st_dev; target_dev = os.stat(target_path.parent).st_dev + if source_dev == target_dev: + if target_path.exists(): target_path.unlink() + os.link(source_path, target_path); operation_done = True + else: print(f"!! Hardlink failed: {source_path.name} different filesystem. Copying.", file=sys.stderr); shutil.copy2(source_path, target_path); operation_done = True + except Exception as e_link: print(f"!! Hardlink failed for {source_path.name}: {e_link}. Copying.", file=sys.stderr); shutil.copy2(source_path, target_path); operation_done = True + else: shutil.move(str(source_path), str(target_path)); operation_done = True + + if operation_done and crc != "error" and opts["dedupe_mode"] != 'off' and crc not in opts["crc_db"]: + try: crc_rel_path = str(target_path.relative_to(target_root)) + except ValueError: crc_rel_path = str(target_path) + opts["crc_db"][crc] = crc_rel_path + except Exception as e: print(f"!! Failed action {mode} for {source_path.name} -> {target_path}: {e}", file=sys.stderr); counters["errors"] += 1; return + + counters["by_type"][report_type] = counters["by_type"].get(report_type, 0) + 1 + counters["processed"] += 1 + + +# --- main() function --- +def main(): + ap = argparse.ArgumentParser(description="Sort & rename music samples.", formatter_class=argparse.RawTextHelpFormatter) + ap.add_argument("--source-root", required=True, help="Vendor/source folder") + ap.add_argument("--target-root", required=True, help="Canonical MUSIC root") + ap.add_argument("--csv", default="sort_report.csv", help="CSV report path") + ap.add_argument("--include-ext", default="wav,flac,aiff,aif,mp3", help="Extensions") + ap.add_argument("--dedupe", choices=["off","skip","quarantine"], default="off", help="Duplicate handling:\n off: No check\n skip: Log and skip\n quarantine: Move to UNSORTED/Duplicates") + ap.add_argument("--crc-db", default="crc_cache.json", help="CRC DB JSON path") + ap.add_argument("--config", help="JSON config override file") + ap.add_argument("--write", action="store_true", help="Apply changes (default: dry-run)") + ap.add_argument("--copy", action="store_true", help="Copy instead of move") + ap.add_argument("--hardlink", action="store_true", help="Hardlink instead of move/copy") + ap.add_argument("--verbose", action="store_true", help="Verbose output") + ap.add_argument("--debug", action="store_true", help="Debug mode - shows detailed BPM extraction logic") + args = ap.parse_args() + + # --- Setup & Load Config/DB --- + source_root = Path(args.source_root).resolve() + target_root = Path(args.target_root).resolve() + if not source_root.is_dir(): sys.exit(f"Source root error: {source_root}") + try: target_root.mkdir(parents=True, exist_ok=True) + except Exception as e: sys.exit(f"Target directory error {target_root}: {e}") + exts = {"." + e.strip().lower().lstrip(".") for e in args.include_ext.split(",") if e.strip()} or VALID_EXTS.copy() + print(f"Including extensions: {', '.join(sorted(list(exts)))}") + crc_db = {} + crc_db_path = Path(args.crc_db).resolve() + if args.dedupe != "off": + if crc_db_path.exists(): + print(f"Loading CRC DB from: {crc_db_path}") + try: + with crc_db_path.open("r", encoding="utf-8") as fh: crc_db = json.load(fh) + print(f"Loaded {len(crc_db)} existing CRC entries.") + except Exception as e: print(f"Warning: Could not load CRC DB ({e}). Starting fresh.", file=sys.stderr) + else: print(f"CRC DB will be created at: {crc_db_path}") + + # --- Prepare Maps/Sets --- + current_type_map = TYPE_MAP.copy() + current_target_dir_map = TARGET_DIR_MAP.copy() + current_descriptor_whitelist = DESCRIPTOR_WHITELIST.copy() + current_machine_tokens = MACHINE_TOKENS.copy() + current_junk_words = JUNK_WORDS.copy() + current_pack_default_map = PACK_DEFAULT_TYPE.copy() + + if args.config: + config_path = Path(args.config).resolve() + if config_path.exists(): + print(f"Loading config overrides from: {config_path}") + try: + with config_path.open("r", encoding="utf-8") as fh: cfg = json.load(fh) + current_type_map.update(cfg.get("TYPE_MAP", {})) + current_target_dir_map.update(cfg.get("TARGET_DIR_MAP", {})) + current_descriptor_whitelist.update(cfg.get("DESCRIPTOR_WHITELIST", [])) + current_machine_tokens.update(cfg.get("MACHINE_TOKENS", [])) + current_junk_words.update(cfg.get("JUNK_WORDS", [])) + current_pack_default_map.update(cfg.get("PACK_DEFAULT_TYPE", {})) + print("Config loaded successfully.") + except Exception as e: print(f"!! Failed to load or parse config JSON: {e}", file=sys.stderr) + else: print(f"Warning: Config file specified but not found: {config_path}", file=sys.stderr) + + # --- Prepare Options & Counters --- + dry = not args.write + operation = "COPY" if args.copy else ("HLINK" if args.hardlink else "MOVE") + mode_str = f"{'DRY-RUN of ' if dry else ''}{operation}" + opts = {"exts": exts, "dry_run": dry, "copy": args.copy, "hardlink": args.hardlink, "verbose": args.verbose, "debug": args.debug, "dedupe_mode": args.dedupe, "crc_db": crc_db, "type_map": current_type_map, "target_dir_map": current_target_dir_map, "descriptor_whitelist": current_descriptor_whitelist, "machine_tokens": current_machine_tokens, "junk_words": current_junk_words, "pack_default_map": current_pack_default_map} + counters = {"processed": 0, "errors": 0, "dup_skipped": 0, "by_type": {}} + csv_path = Path(args.csv).resolve() + + # --- Print Header --- + print("-" * 30); print(f"Source: {source_root}"); print(f"Target: {target_root}"); print(f"Mode: {mode_str}"); print(f"Dedupe: {args.dedupe}"); print(f"Report: {csv_path}") + if args.debug: print(f"Debug: ENABLED") + print("-" * 30) + + # --- Processing Loop --- + try: + with csv_path.open("w", newline="", encoding="utf-8") as fh: + fieldnames = ["old_path","new_path","type","descriptors","bpm","key","vendor","index","collision_version","dry_run"] + writer = csv.DictWriter(fh, fieldnames=fieldnames); writer.writeheader() + print("Scanning files..."); all_files = list(source_root.rglob("*")); total_items = len(all_files); print(f"Found {total_items} total items. Processing relevant files...") + relevant_files_found = 0 + for i, p in enumerate(all_files): + if (i + 1) % 500 == 0 or i == total_items - 1: print(f" Scanning progress: {i+1}/{total_items} ({(i + 1) / total_items * 100:.1f}%)", end='\r', flush=True) + if not p.is_file(): continue + ext_lower = p.suffix.lower(); e = ".aiff" if ext_lower == ".aif" else ext_lower + if e not in exts: continue + relevant_files_found += 1 + process_file(p, source_root, target_root, writer, opts, counters) + print(f"\nFinished scanning. Processed {relevant_files_found} relevant files.") + except KeyboardInterrupt: print("\n!! Processing interrupted by user.", file=sys.stderr) + except Exception as e: print(f"\n!! An unexpected error occurred: {e}", file=sys.stderr); import traceback; traceback.print_exc() + finally: + # --- Save CRC DB & Summary --- + if args.dedupe != "off": + try: + with crc_db_path.open("w", encoding="utf-8") as fh_final: json.dump(dict(sorted(opts['crc_db'].items())), fh_final, indent=2) + print(f"\nCRC DB saved to {crc_db_path} ({len(opts['crc_db'])} entries).") + except Exception as e: print(f"!! Failed to save CRC DB: {e}", file=sys.stderr) + print("\n" + "=" * 30); print("Run Summary"); print("=" * 30); print(f" Mode: {mode_str}"); print(f" Source Files Found: {relevant_files_found}"); print(f" Files Processed: {counters['processed']}"); print(f" Duplicates Skipped: {counters['dup_skipped']}"); print(f" Errors Encountered: {counters['errors']}") + print("\n Counts by detected type (includes duplicates if quarantined):") + if counters['by_type']: + for k, v in sorted(counters['by_type'].items(), key=lambda kv: (-kv[1], kv[0])): print(f" {k.replace('dup_', '(dup) '):<20s} {v:>6d}") + else: print(" No files were categorized.") + print("-" * 30); print(f"CSV report saved to: {csv_path}"); print("=" * 30) + +if __name__ == "__main__": + main()