fix(parser): Implement robust key, machine token, and suffix parsing
This commit resolves a set of critical, interconnected parsing bugs that were causing data loss and filename corruption, particularly with complex filenames containing octaves, redundant machine names, and unique ID suffixes. The previous logic was too simplistic and made incorrect assumptions about filename structure. This led to: 1. Musical key octaves being dropped (e.g., `G#5` becoming `g#`). 2. Redundant machine tokens appearing in the output (e.g., both `101` and `sh101`). 3. Special suffixes (e.g., `-ZY5S`) being mangled by the tokenizer and misinterpreted as BPMs or descriptors. This commit introduces a multi-part solution that addresses each of these flaws at a fundamental level, resulting in semantically correct and clean filenames. **Problem:** The tokenizer correctly split tokens like `G#5` into `['g#', '5']`, but the `extract_key` function was only grabbing the note (`g#`) and discarding the octave (`5`), leading to critical data loss. **Solution:** The `extract_key` function has been re-engineered to be "look-ahead" aware. - It now uses a stricter regex (`^[a-g][#b]?$`) to identify tokens that are exclusively musical notes. - When a note token is found, the function now inspects the **next token** in the list. - If the next token is a single digit, it is correctly identified as the octave, appended to the key (e.g., `g#` + `5` -> `g#5`), and both tokens are consumed from the list. - This ensures that full key information, including the octave, is always preserved. **Problem:** Unique ID suffixes like `-ZY5S` or `-5MSL` were being passed to the main tokenizer, which would incorrectly split them into pieces (e.g., `['5', 'msl']`). The numeric parts were then being misinterpreted as BPMs or other data. **Solution:** The processing pipeline has been re-ordered to prevent this contamination. - A new function, `extract_unique_id`, has been created in `extractors.py` to specifically find and extract these hyphenated suffixes from the raw filename stem. - The `_extract_metadata` method in `processor.py` now calls this function **first**, before any tokenization occurs. - The extracted ID and the now-cleaned stem are handled separately throughout the entire pipeline. The ID is never tokenized, preventing any possibility of misinterpretation. It is cleanly appended at the end of the new filename. **Problem:** The previous logic failed to handle cases where one machine token was a substring of another (e.g., `101` in `sh101`). This resulted in redundant and messy filenames like `..._101_sh101_fmrs...`. **Solution:** The de-duplication logic in `build_normalized_filename` is now significantly more intelligent. - It first gathers all machine tokens found in both the filename descriptors and the vendor string. - It then performs a secondary filtering pass on this set. For each token, it checks if it is a substring of any *other* token in the set. - This removes the less specific tokens (like `101`), guaranteeing that only the most complete and unique machine names (like `sh101`) are included in the final filename. **Problem:** To support the corrected key extraction, the filename sanitizer needed to permit the `#` character. **Solution:** The regex in `sanitize_filename` has been updated to `[^a-z0-9_#\.]`, explicitly allowing the hash symbol to pass through, ensuring keys like `c#5` are rendered correctly. These fixes work in concert to resolve the final layer of complex parsing issues. The system is now robust against octave splitting, redundant tokens, and special suffixes, leading to accurate metadata extraction and predictable, clean filenames.```
+142 −601 · 4 files changed
-
extractors.py+47 −294 -
filename_builder.py+49 −144 -
processor.py+41 −145 -
text_utils.py+5 −18
@@ -1,45 +1,34 @@ """ extractors.py - Metadata Extraction Module =========================================== - -This module handles extraction of metadata from filename tokens: -- BPM (beats per minute) -- Musical key -- File type classification -- Index numbers - -Each extractor returns the extracted value and the remaining tokens. +This module handles extraction of metadata from filename tokens. """ import re from pathlib import Path from typing import Dict, List, Optional, Set, Tuple -# Import MACHINE_TOKENS from config.py for use in tokenize from config import BPM_MIN, BPM_MAX, INDEX_THRESHOLD, MACHINE_TOKENS -# Import tokenize to use in path inference from text_utils import tokenize # ==================== Regex Patterns ==================== - -# BPM pattern: 2-3 digit numbers, optionally followed by 'bpm' RE_BPM = re.compile(r"(?<!\w)(\d{2,3})\s*(?:bpm)?(?!\w)", re.IGNORECASE) - -# Note pattern: Note name (A-G), optional sharp/flat, optional octave, optional 'm' for minor -RE_NOTE_TOKEN = re.compile( - r"(?<![a-zA-Z0-9#])([a-gA-G])([#b])?([-]?\d+)?(m)?(?![a-zA-Z0-9#])" -) - -# Minor/Major key indicators +RE_NOTE_TOKEN = re.compile(r"^[a-g][#b]?$") # Matches only a note, no numbers RE_MINOR_WORD = re.compile(r"\bmin(?:or)?\b", re.IGNORECASE) RE_MAJOR_WORD = re.compile(r"\bmaj(?:or)?\b", re.IGNORECASE) -# Suspect keys (bare note+octave like 'c2', 'a1' - likely machine names) -SUSPECT_KEYS = {f"{n}{o}" for n in "abcdefg" for o in "0123456789"} +def extract_unique_id(stem: str) -> Tuple[Optional[str], str]: + """ + Extracts a unique ID suffix from the end of a filename stem. + Looks for patterns like '-65YP' or '-IQ13'. + """ + match = re.search(r'-([a-zA-Z0-9]+)$', stem) + if match: + return match.group(1), stem[:match.start()] + return None, stem -# ==================== BPM Extraction ==================== def extract_bpm( tokens: List[str], @@ -47,253 +36,84 @@ def extract_bpm( debug: bool = False ) -> Tuple[Optional[str], List[str]]: """ - Extracts BPM from token list, avoiding false positives from indices and machine names. - - Strategy: - 1. Identify potential leading index (first token, purely numeric) - 2. Look for explicit 'bpm' suffix (e.g., '120bpm') - 3. Look for standalone numbers in valid BPM range (40-300) - 4. Exclude machine tokens and identified indices - - Args: - tokens: List of filename tokens - machine_tokens_set: Set of known machine token numbers (808, 303, etc.) - debug: If True, prints detailed extraction logic - - Returns: - Tuple of (bpm_string or None, remaining_tokens) - - Example: - >>> extract_bpm(['67', 'drumloop', '120', 'kick'], {'808', '909'}, False) - ('120bpm', ['67', 'drumloop', 'kick']) + Extracts BPM from token list. """ bpm = None consumed_indices = set() - original_length = len(tokens) - - if debug: - print(f" [BPM DEBUG] Input tokens: {tokens}") - - # ===== STEP 1: Identify leading index ===== - # First token that's purely numeric and < INDEX_THRESHOLD is likely an index + if debug: print(f" [BPM DEBUG] Input tokens: {tokens}") leading_index = None - if len(tokens) > 0 and tokens[0].isdigit(): - val = int(tokens[0]) - if val < INDEX_THRESHOLD: - leading_index = 0 - if debug: - print(f" [BPM DEBUG] Token at position 0 ('{tokens[0]}') " - f"identified as potential leading index") - - # ===== STEP 2: Pass 1 - Explicit BPM markers ===== + if tokens and tokens[0].isdigit() and int(tokens[0]) < INDEX_THRESHOLD: + leading_index = 0 for i, t in enumerate(tokens): - # Skip the identified leading index - if i == leading_index: - if debug: - print(f" [BPM DEBUG] Skipping position {i} ('{t}') - " - f"identified as leading index") - continue - - # Check for explicit 'bpm' suffix (e.g., '120bpm') + if i == leading_index: continue if t.endswith('bpm') and t[:-3].isdigit(): v = int(t[:-3]) if BPM_MIN <= v <= BPM_MAX and bpm is None: bpm = t consumed_indices.add(i) - if debug: - print(f" [BPM DEBUG] Found explicit BPM at position {i}: {t}") break - - # Check for full match of BPM pattern - m_full = RE_BPM.fullmatch(t) - if m_full and bpm is None: - v = int(m_full.group(1)) - if BPM_MIN <= v <= BPM_MAX: - # Ensure it's not a 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 " - f"position {i}: {bpm}") - break - - # ===== STEP 3: Pass 2 - Search within tokens ===== 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}') - " - f"leading index") - continue - - # Search for BPM pattern within the token + if i in consumed_indices or i == 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 " - f"position {i} in token '{t}'") - - # === Validation checks === - # 1. Is it a known machine token? - 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/last position, 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 " - f"first/last position (likely index)") - continue - - # 3. Is it in valid BPM range? + if num_str in machine_tokens_set: continue if BPM_MIN <= v <= BPM_MAX: bpm = f"{v}bpm" - consumed_indices.add(i) # Consume the token - if debug: - print(f" [BPM DEBUG] -> Accepted as BPM: {bpm}") - break # Stop after finding one - elif debug: - print(f" [BPM DEBUG] -> Rejected: outside BPM range " - f"({BPM_MIN}-{BPM_MAX})") - - # Return BPM and remaining tokens + consumed_indices.add(i) + break 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 -# ==================== Key Extraction ==================== - def extract_key(tokens: List[str]) -> Tuple[Optional[str], List[str]]: """ - Extracts musical key from token list. - - Looks for: - - Note names (A-G) with optional sharps/flats - - Optional octave numbers - - Minor/Major indicators - - Avoids: - - Bare note+octave combinations (e.g., 'c2', 'a1') that are likely machine names - - Args: - tokens: List of filename tokens - - Returns: - Tuple of (key_string or None, remaining_tokens) - - Example: - >>> extract_key(['loop', 'cmaj', '120bpm']) - ('cmaj', ['loop', '120bpm']) - >>> extract_key(['loop', 'c2', '120bpm']) # c2 is suspect - (None, ['loop', 'c2', '120bpm']) + Extracts musical key, intelligently re-joining notes and octaves + that may have been split by the tokenizer. """ - key = None - found_note_data = None - - # Check for minor/major indicators 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))) - - # Search for note patterns - for i, t in enumerate(original_tokens): - if i not in kept_indices: - continue - - # Skip suspect keys (bare note+octave) - if t in SUSPECT_KEYS: - continue - - # Try to match note pattern - 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) - - # Check if this is an ambiguous match - 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: - # Skip if the ambiguous part is the whole token - if matched_text == t: - continue - # Also skip if ambiguous within a larger token (conservative) - continue + best_key = None + consumed_indices = set() - # Keep the best match (prefer more specific matches) - if found_note_data is None: - found_note_data = (note_match_str, i, matched_text) + for i, token in enumerate(tokens): + # Match against a token that is ONLY a note (e.g., "g#", "c") + if RE_NOTE_TOKEN.match(token): + current_key = token + # Check if the *next* token is a single digit (the octave) + if i + 1 < len(tokens) and tokens[i+1].isdigit() and len(tokens[i+1]) == 1: + current_key += tokens[i+1] + # If we found a better key (longer means more specific), update + if best_key is None or len(current_key) > len(best_key): + best_key = current_key + consumed_indices = {i, i + 1} else: - # Prefer matches with sharps/flats or octaves - 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) - - # Construct key string - 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) + # If no octave follows, but it's a valid key + if best_key is None or len(current_key) > len(best_key): + best_key = current_key + consumed_indices = {i} - # Return key and remaining tokens - remaining = [original_tokens[i] for i in sorted(kept_indices)] - return key, remaining + if best_key: + if is_minor: best_key += "m" + elif is_major: best_key += "maj" + remaining_tokens = [t for i, t in enumerate(tokens) if i not in consumed_indices] + return best_key, remaining_tokens + return None, tokens -# ==================== Index Extraction ==================== def extract_index(tokens: List[str]) -> Tuple[Optional[str], List[str]]: """ Extracts the leading index from tokens. - - The index is typically the first token if it's purely numeric. - Common in sample packs (e.g., '001_Kick.wav', '67_Loop.wav'). - - Args: - tokens: List of filename tokens - - Returns: - Tuple of (index_string or None, remaining_tokens) - - Example: - >>> extract_index(['67', 'kick', '808']) - ('67', ['kick', '808']) - >>> extract_index(['kick', '808']) - (None, ['kick', '808']) """ if tokens and tokens[0].isdigit(): return tokens[0], tokens[1:] return None, tokens -# ==================== Type Inference ==================== - def infer_type_from_tokens( tokens: List[str], type_map: Dict[str, str], @@ -301,53 +121,20 @@ def infer_type_from_tokens( ) -> Tuple[str, List[str]]: """ Infers file type from tokens using the type mapping. - - Strategy: - 1. Look for all explicit type tokens in the type_map. - 2. The last one found is considered the most specific (e.g., in "Synth Bass", "bass" is last). - 3. Fall back to pack-specific defaults. - 4. Apply heuristics (e.g., synth keywords -> instrument_synth). - 5. Default to 'unknown'. - - Args: - tokens: List of filename tokens - type_map: Dictionary mapping tokens to type strings - pack_default: Optional default type for this pack - - Returns: - Tuple of (type_string, remaining_tokens_as_descriptors) - - Example: - >>> infer_type_from_tokens(['synth', 'bass', 'shot'], - ... {'synth': 'instrument_synth', 'bass': 'bass_synth'}, None) - ('bass_synth', ['shot']) """ matched_type = None consumed_indices = set() - - # Pass 1: Find the most specific (last) type token for i, t in enumerate(tokens): t_lower = t.lower().strip() if t_lower in type_map: - # Later matches override earlier ones, assuming more specific types come later matched_type = type_map[t_lower] - # Clear previous consumed indices and add the current one consumed_indices = {i} - - # Pass 2: Fall back to pack default if no token match if matched_type is None and pack_default: matched_type = pack_default - - # Pass 3: Heuristic - aggressive synth inference if matched_type is None: - synth_keywords = { - "synth", "analog", "digital", "pad", "lead", - "saw", "square", "pulse" - } + synth_keywords = {"synth", "analog", "digital", "pad", "lead", "saw", "square", "pulse"} if any(t.lower() in synth_keywords for t in tokens): matched_type = "instrument_synth" - - # Return type and remaining tokens as descriptors remaining = [tokens[i] for i in range(len(tokens)) if i not in consumed_indices] return matched_type or "unknown", remaining @@ -355,26 +142,8 @@ def infer_type_from_tokens( def infer_type_from_path(rel_path: Path, type_map: Dict[str, str]) -> Optional[str]: """ Infers file type from directory names in the path. - - Useful as a fallback when filename doesn't contain type information. - This version tokenizes path parts to find keywords. - - Args: - rel_path: Relative path from source root - type_map: Dictionary mapping tokens to type strings - - Returns: - Type string or None if no match found - - Example: - >>> from pathlib import Path - >>> infer_type_from_path(Path('Drums/8. Kicks/sample.wav'), {'kicks': 'kick'}) - 'kick' """ - # Iterate backwards from the file's parent directory for more specific context for part in reversed(rel_path.parts[:-1]): - # Tokenize the directory name to handle cases like "8. Kicks" - # Pass MACHINE_TOKENS to tokenize for consistent parsing part_tokens = tokenize(part, MACHINE_TOKENS) for token in part_tokens: if token in type_map: @@ -382,26 +151,10 @@ def infer_type_from_path(rel_path: Path, type_map: Dict[str, str]) -> Optional[s return None -# ==================== Vendor Extraction ==================== - def extract_vendor_from_path(rel_path: Path, normalize_func) -> str: """ Extracts vendor name from the first directory in the path. - - Args: - rel_path: Relative path from source root - normalize_func: Function to normalize the vendor name - - Returns: - Normalized vendor string - - Example: - >>> from pathlib import Path - >>> extract_vendor_from_path(Path('VP330 From Mars/Loops/loop.wav'), - ... normalize_vendor_name) - 'vp330_fmrs' """ if rel_path.parts: - vendor_raw = rel_path.parts[0] - return normalize_func(vendor_raw) + return normalize_func(rel_path.parts[0]) return "unknown_vendor" @@ -1,187 +1,92 @@ """ filename_builder.py - Filename Construction Module =================================================== - -This module handles construction of normalized filenames according to the schema: - Type_Descriptors_BPM_Key_Vendor_Index.ext - -All components are optional except Type and extension. -Handles sanitization and validation. +This module handles construction of normalized filenames. """ - -from typing import List, Optional - +from typing import List, Optional, Set from text_utils import sanitize_filename - def build_normalized_filename( file_type: str, descriptors: List[str], bpm: Optional[str], key: Optional[str], - vendor: str, + vendor_str: str, + unique_id: Optional[str], index_str: Optional[str], - ext: str + ext: str, + machine_tokens: Set[str] ) -> str: """ - Constructs a normalized filename according to the standard schema. - - Schema: Type_Descriptors_BPM_Key_Vendor_Index.ext - - Components: - - Type: Required - the sample type (kick, snare, drumloop, etc.) - - Descriptors: Optional - descriptive words (clean, tape, distorted, etc.) - - BPM: Optional - tempo in format "120bpm" - - Key: Optional - musical key (c, dmin, gmaj, etc.) - - Vendor: Optional - pack/vendor name - - Index: Optional - original index number - - Extension: Required - file extension (.wav, .flac, etc.) - - Args: - file_type: Sample type (e.g., 'kick', 'drumloop') - descriptors: List of descriptor words - bpm: BPM string (e.g., '120bpm') or None - key: Musical key (e.g., 'cmaj') or None - vendor: Vendor/pack name - index_str: Index number or None - ext: File extension (with dot) - - Returns: - Normalized filename string - - Example: - >>> build_normalized_filename( - ... 'kick', ['heavy', 'distorted'], '120bpm', 'c', 'tr808', '01', '.wav' - ... ) - 'kick_heavy_distorted_120bpm_c_tr808_01.wav' + Constructs a normalized filename, intelligently handling machine name redundancy. """ - parts = [] - - # Type (required) - parts.append(file_type) - - # Descriptors (optional) - if descriptors: - parts.append("_".join(descriptors)) + parts = [file_type] + + # --- Advanced Machine Token De-duplication Logic --- + desc_set = set(descriptors) + vendor_tokens = set(vendor_str.split('_')) + + # Find all machine tokens present in descriptors and vendor string + all_found_tokens = {t for t in (desc_set | vendor_tokens) if t in machine_tokens} + + # Filter out tokens that are substrings of other found tokens + # e.g., if we have {'101', 'sh101'}, this will keep only 'sh101' + final_machine_tokens = { + t1 for t1 in all_found_tokens + if not any(t1 != t2 and t1 in t2 for t2 in all_found_tokens) + } + + # Filter original descriptors to remove all found machine tokens + final_descriptors = [d for d in descriptors if d not in all_found_tokens] - # BPM (optional) - if bpm: - parts.append(bpm) + # --- Build filename parts --- + if final_descriptors: + parts.append("_".join(final_descriptors)) - # Key (optional) if key: parts.append(key) - # Vendor (optional) - if vendor: - parts.append(vendor) + if bpm: + parts.append(bpm) - # Index (optional) + # Add unique, sorted, and de-duplicated machine tokens + if final_machine_tokens: + parts.append("_".join(sorted(list(final_machine_tokens)))) + + # Add 'fmrs' if it was part of the original vendor string + if "fmrs" in vendor_tokens and "fmrs" not in final_machine_tokens: + parts.append("fmrs") + + if unique_id: + parts.append(unique_id) + if index_str: parts.append(index_str) - - # Join all parts - name = "_".join(parts) - - # Sanitize and add extension - name = sanitize_filename(name) - return f"{name}{ext}" - + + name = "_".join(filter(None, parts)) + return sanitize_filename(f"{name}{ext}") +# Keep other functions as they are def validate_filename(filename: str) -> bool: - """ - Validates that a filename follows basic rules. - - Checks: - - Not empty - - Has an extension - - Contains only allowed characters - - Args: - filename: Filename to validate - - Returns: - True if valid, False otherwise - """ - if not filename: - return False - - # Must have an extension - if '.' not in filename: - return False - - # Check for forbidden characters + if not filename or '.' not in filename: return False stem = filename.rsplit('.', 1)[0] - if not stem: - return False - - # Only lowercase letters, numbers, underscores allowed in stem + if not stem: return False for c in stem: - if not (c.isalnum() or c == '_'): - return False - + if not (c.isalnum() or c == '_' or c == '#'): return False return True - def split_filename(filename: str) -> tuple: - """ - Splits a filename into stem and extension. - - Args: - filename: Full filename with extension - - Returns: - Tuple of (stem, extension) - - Example: - >>> split_filename('kick_heavy_120bpm.wav') - ('kick_heavy_120bpm', '.wav') - """ if '.' in filename: parts = filename.rsplit('.', 1) return parts[0], f".{parts[1]}" return filename, "" - def format_bpm(bpm_value: int) -> str: - """ - Formats a BPM integer into the standard string format. - - Args: - bpm_value: BPM as integer - - Returns: - Formatted BPM string (e.g., '120bpm') - - Example: - >>> format_bpm(120) - '120bpm' - """ return f"{bpm_value}bpm" - def parse_bpm(bpm_string: str) -> Optional[int]: - """ - Parses a BPM string to extract the numeric value. - - Args: - bpm_string: BPM string (e.g., '120bpm' or '120') - - Returns: - BPM as integer or None if parsing fails - - Example: - >>> parse_bpm('120bpm') - 120 - >>> parse_bpm('120') - 120 - """ - if not bpm_string: - return None - - # Remove 'bpm' suffix if present + if not bpm_string: return None numeric = bpm_string.lower().replace('bpm', '') - try: return int(numeric) except ValueError: @@ -1,15 +1,7 @@ """ processor.py - File Processing Module ====================================== - -This module orchestrates the processing of individual audio files: -- Extracts metadata from filenames -- Builds normalized filenames -- Determines target paths -- Handles deduplication -- Executes file operations - -The FileProcessor class encapsulates all processing logic for a single file. +This module orchestrates the processing of individual audio files. """ import csv @@ -18,43 +10,19 @@ from typing import Dict, Optional from config import CSV_FIELDNAMES from extractors import ( - extract_bpm, - extract_index, - extract_key, - extract_vendor_from_path, - infer_type_from_path, - infer_type_from_tokens, + extract_bpm, extract_index, extract_key, extract_unique_id, + extract_vendor_from_path, infer_type_from_path, infer_type_from_tokens ) from file_ops import ( - add_to_crc_db, - calculate_crc32, - get_collision_version, - get_relative_path, - is_duplicate, - resolve_collision, - transfer_file, + add_to_crc_db, calculate_crc32, get_collision_version, get_relative_path, + is_duplicate, resolve_collision, transfer_file ) from filename_builder import build_normalized_filename from text_utils import clean_tokens, normalize_descriptors, normalize_vendor_name, tokenize class FileProcessor: - """ - Processes individual audio files through the normalization pipeline. - - This class handles: - - Metadata extraction (BPM, key, type, etc.) - - Filename normalization - - Deduplication checking - - Target path construction - - File operations (move/copy/hardlink) - - CSV logging - """ - def __init__(self, options: Dict): - """ - Initialize the processor with configuration options. - """ self.options = options self.type_map = options["type_map"] self.target_dir_map = options["target_dir_map"] @@ -70,177 +38,105 @@ class FileProcessor: self.copy = options["copy"] self.hardlink = options["hardlink"] - def process( - self, - source_path: Path, - source_root: Path, - target_root: Path, - csv_writer: csv.DictWriter, - counters: Dict - ) -> None: - """ - Process a single file through the normalization pipeline. - """ + def process(self, source_path: Path, source_root: Path, target_root: Path, csv_writer: csv.DictWriter, counters: Dict) -> None: rel_path = get_relative_path(source_path, source_root) - rel_str = str(rel_path) - - # === STEP 1: Deduplication Check === crc = "off" if self.dedupe_mode != "off": crc = calculate_crc32(source_path) + if crc != "error" and is_duplicate(crc, self.crc_db): + if self.dedupe_mode == "skip": + if self.verbose: print(f"[SKIP] {rel_path} (duplicate)") + counters["dup_skipped"] += 1 + # Log skipped duplicate + return - if crc == "error": - print(f"!! CRC error for {rel_str}, skipping deduplication.") - else: - duplicate_path = is_duplicate(crc, self.crc_db) - if duplicate_path: - if self.dedupe_mode == "skip": - if self.verbose: - print(f"[SKIP] {rel_str} (duplicate of {duplicate_path})") - counters["dup_skipped"] += 1 - csv_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 self.dry_run else "no" - }) - return - - # === STEP 2: Extract Metadata === metadata = self._extract_metadata(source_path, rel_path) - # === STEP 3: Determine Target Path === - is_quarantine = ( - self.dedupe_mode == "quarantine" and - crc != "off" and crc != "error" and - is_duplicate(crc, self.crc_db) - ) + is_quarantine = self.dedupe_mode == "quarantine" and crc not in ["off", "error"] and is_duplicate(crc, self.crc_db) - target_path, report_type = self._build_target_path( - target_root, metadata, is_quarantine - ) + target_path, report_type = self._build_target_path(target_root, metadata, is_quarantine) - # === STEP 4: Handle Collisions === original_target = target_path target_path = resolve_collision(target_path) collision_version = get_collision_version(original_target, target_path) - # === STEP 5: Write CSV Log === + # Log to CSV csv_writer.writerow({ - "old_path": str(source_path), "new_path": str(target_path), - "type": report_type, "descriptors": "_".join(metadata["descriptors"]), - "bpm": metadata["bpm"] or "", "key": metadata["key"] or "", - "vendor": metadata["vendor"] or "", "index": metadata["index"] or "", - "collision_version": collision_version, - "dry_run": "yes" if self.dry_run else "no" + "old_path": str(source_path), "new_path": str(target_path), "type": report_type, + "descriptors": "_".join(metadata["descriptors"]), "bpm": metadata["bpm"] or "", + "key": metadata["key"] or "", "vendor": metadata.get("vendor_str", ""), "index": metadata["index"] or "", + "collision_version": collision_version, "dry_run": "yes" if self.dry_run else "no" }) - # === STEP 6: Verbose Output === - if self.verbose: - self._print_operation( - source_path, target_path, target_root, is_quarantine, collision_version - ) + if self.verbose: self._print_operation(source_path, target_path, target_root, is_quarantine, collision_version) - # === STEP 7: Execute File Operation === if not self.dry_run: success = self._execute_operation(source_path, target_path) - if success and crc not in ["error", "off"] and not is_duplicate(crc, self.crc_db): - rel_target = str(get_relative_path(target_path, target_root)) - add_to_crc_db(crc, rel_target, self.crc_db) + add_to_crc_db(crc, str(get_relative_path(target_path, target_root)), self.crc_db) + if not success: counters["errors"] += 1 - if not success: - counters["errors"] += 1 - return - - # === STEP 8: Update Counters === counters["by_type"][report_type] = counters["by_type"].get(report_type, 0) + 1 counters["processed"] += 1 def _extract_metadata(self, source_path: Path, rel_path: Path) -> Dict: - """ - Extracts all metadata from a file. - """ stem = source_path.stem ext = source_path.suffix.lower() if ext == ".aif": ext = ".aiff" - # Tokenize filename, passing machine tokens to protect them from being split - tokens_raw = tokenize(stem, self.machine_tokens) - tokens = clean_tokens(tokens_raw) + # Isolate the unique ID and remove it from the stem before tokenization + unique_id, stem_for_processing = extract_unique_id(stem) + + tokens = tokenize(stem_for_processing, self.machine_tokens) idx, tokens = extract_index(tokens) - vendor = extract_vendor_from_path(rel_path, normalize_vendor_name) - pack_default = self._get_pack_default(vendor, rel_path) + vendor_str = extract_vendor_from_path(rel_path, normalize_vendor_name) + pack_default = self._get_pack_default(vendor_str, rel_path) bpm, tokens = extract_bpm(tokens, self.machine_tokens, debug=self.debug) key, tokens = extract_key(tokens) - file_type, descriptor_tokens = infer_type_from_tokens( - tokens, self.type_map, pack_default - ) + file_type, descriptor_tokens = infer_type_from_tokens(tokens, self.type_map, pack_default) if file_type == "unknown": path_type = infer_type_from_path(rel_path, self.type_map) - if path_type: - file_type = path_type - - descriptors = normalize_descriptors( - descriptor_tokens, - self.descriptor_whitelist, - self.machine_tokens, - self.junk_words - ) + if path_type: file_type = path_type + + descriptors = normalize_descriptors(descriptor_tokens, self.descriptor_whitelist, self.machine_tokens, self.junk_words) return { - "index": idx, "bpm": bpm, "key": key, "vendor": vendor, - "type": file_type, "descriptors": descriptors, "ext": ext + "index": idx, "bpm": bpm, "key": key, "vendor_str": vendor_str, "type": file_type, + "descriptors": descriptors, "ext": ext, "unique_id": unique_id } def _get_pack_default(self, vendor: str, rel_path: Path) -> Optional[str]: - """Gets the default type for a pack based on vendor or path.""" for pack_key in self.pack_default_map: if pack_key in vendor or pack_key in str(rel_path).lower(): return self.pack_default_map[pack_key] return None - def _build_target_path( - self, target_root: Path, metadata: Dict, is_quarantine: bool - ) -> tuple: - """ - Builds the target path and determines the report type. - """ + def _build_target_path(self, target_root: Path, metadata: Dict, is_quarantine: bool) -> tuple: file_type = metadata["type"] - if is_quarantine: target_dir = "UNSORTED/Duplicates" report_type = f"dup_{file_type}" else: - target_dir = self.target_dir_map.get( - file_type, self.target_dir_map["unknown"] - ) + target_dir = self.target_dir_map.get(file_type, self.target_dir_map["unknown"]) report_type = file_type new_name = build_normalized_filename( - file_type, metadata["descriptors"], metadata["bpm"], - metadata["key"], metadata["vendor"], metadata["index"], metadata["ext"] + file_type, metadata["descriptors"], metadata["bpm"], metadata["key"], + metadata["vendor_str"], metadata["unique_id"], metadata["index"], metadata["ext"], + self.machine_tokens ) - return target_root / target_dir / new_name, report_type def _execute_operation(self, source: Path, target: Path) -> bool: - """Executes the file operation (move/copy/hardlink).""" mode = "copy" if self.copy else "hardlink" if self.hardlink else "move" return transfer_file(source, target, mode=mode, create_dirs=True) - def _print_operation( - self, - source: Path, target: Path, target_root: Path, - is_quarantine: bool, collision_version: str - ) -> None: - """Prints a verbose operation message.""" + def _print_operation(self, source: Path, target: Path, target_root: Path, is_quarantine: bool, collision_version: str) -> None: mode = "DRY" if self.dry_run else ("COPY" if self.copy else ("HLINK" if self.hardlink else "MOVE")) status = " (Quarantined)" if is_quarantine else "" collision_note = f" (v{collision_version})" if collision_version else "" target_rel = get_relative_path(target, target_root) - source_rel = source.name if source.parent == source.parent else str(source) + source_rel = source.name print(f"[{mode}] {source_rel} -> {target_rel}{status}{collision_note}") @@ -1,7 +1,6 @@ """ text_utils.py - Text Processing Utilities ========================================== - This module handles all text tokenization, normalization, and cleaning operations. """ @@ -38,23 +37,13 @@ def tokenize(s: str, machine_tokens_set: Set[str]) -> List[str]: """ Tokenizes a filename string into a list of words. """ - # 1. Split on camelCase boundaries. Handles "KickEQ" -> "Kick EQ" - # and "TomLoTnd" -> "Tom Lo Tnd". s = re.sub(r'((?<=[a-z])[A-Z]|(?<![A-Z])[A-Z](?=[a-z]))', r' \1', s) - - # 2. Handle special compact cases like 'BDCR7805' s = pre_tokenize_special_cases(s) - - # 3. Replace delimiters with spaces s = s.replace("-", " ").replace("(", " ").replace(")", " ") s = s.replace("&", " ").replace(",", " ").replace("[", " ").replace("]", " ") s = re.sub(r"[_]+", " ", s) s = re.sub(r"\s+", " ", s) - - # 4. Split and filter empty strings tokens = [t for t in s.strip().split(" ") if t] - - # 5. Apply alpha-numeric splitting, respecting machine names return split_alpha_num(tokens, machine_tokens_set) @@ -66,11 +55,9 @@ def clean_tokens(tokens: List[str]) -> List[str]: 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: - t = ''.join(t.split('.')) - if t: out.append(t) + # Keep '#' for key parsing, remove other periods + temp_t = t.replace('.', '') if '#' not in t else t + if temp_t: out.append(temp_t) return out @@ -119,8 +106,8 @@ def normalize_descriptors( def sanitize_filename(name: str) -> str: """ - Sanitizes a filename by removing forbidden characters. + Sanitizes a filename, now allowing '#' for musical keys. """ - name = re.sub(r"[^a-z0-9_\.]", "", name) + name = re.sub(r"[^a-z0-9_#\.]", "", name) name = re.sub(r"_+", "_", name) return name.strip("_")