script to reference moved song in playlists

media handling/reference_playlists_to_library.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+
+import os
+import re
+import logging
+
+from pathlib import Path
+
+dry_run = True
+WORKDIR = Path("/data/navidrome/music/")
+
+PLAYLIST_DIR = WORKDIR / "playlists"
+MUSIC_DIR = WORKDIR / "music"
+
+AUDIO_EXTS = {
+    ".mp3", ".flac", ".m4a", ".ogg", ".opus",
+    ".aac", ".wav", ".ape"
+}
+logger = logging.getLogger(__name__)
+logging.basicConfig(
+    level=logging.DEBUG,
+    format="%(asctime)s %(levelname)s %(message)s",
+    handlers=[
+        #logging.FileHandler("reference_playlists.log"),
+        logging.StreamHandler()
+    ]
+)
+logger.info('Started')
+
+
+def words(text):
+    return set(re.findall(r"\b[\w']+\b", text.lower()))
+
+
+def find_replacement(old_path):
+    p = Path(old_path)
+    # Find the second "music" directory
+    music_idx = p.parts.index("music", p.parts.index("music") + 1)
+    folder_count = len(p.parts) - music_idx - 2
+         
+    try:
+        artist = p.parts[-(folder_count + 1)]  # /music/Artist/Album/File
+        logger.debug(f"Artist: {artist}")
+    except IndexError:
+        logger.error(f"No Artist found for {p}")
+        return None
+
+    artist_dir = Path(MUSIC_DIR) / artist
+    logger.debug(f"Artist path: {artist_dir}")
+    
+    if not artist_dir.exists():
+        logger.error(f"Artist dir {artist_dir} dors not exist")
+        return None
+    if folder_count > 1:
+        try:
+            album = p.parts[-(folder_count)]  # .../Artist/Album/File
+            logger.debug(f"Album: {album}")
+        except IndexError:
+            logger.error(f"No album found for {p}")
+            return None
+        
+        album_dir = Path(MUSIC_DIR).rglob(album)
+    
+        try:
+            album_dir = next(album_dir)
+            logger.debug(f"Album path: {album_dir}")
+        except StopIteration:
+            logger.error(f"Album dir for '{album}' does not exist")
+            return None
+    
+    search_words = words(p.stem)
+
+    best_match = None
+    best_score = 0
+
+    for root, _, files in os.walk(artist_dir):
+        for filename in files:
+
+            candidate = Path(root) / filename
+                
+            candidate_parts = candidate.stem.split(" - ")
+            
+            logger.debug(f"Candidate: {candidate}")
+            logger.debug(f"parts: {candidate_parts}")
+            if candidate.suffix.lower() not in AUDIO_EXTS:
+                continue
+
+            if len(candidate_parts) == 4:
+                logger.debug(f"Lidarr Filename: {len(candidate_parts)} ")
+                candidate_words = words(" ".join(candidate_parts[2:]))
+                logger.debug(f"Go on with: {candidate_words} ")
+                score = len(search_words & candidate_words)
+
+                if score > best_score:
+                    best_score = score
+                    best_match = candidate
+            else:
+                continue
+
+                logger.debug(f"Fallback Filename: {len(candidate_parts)} ")
+                candidate_words = words(candidate.stem)
+
+                logger.debug(f"words: {candidate_words}")
+                score = len(search_words & candidate_words)
+
+                if score > best_score:
+                    best_score = score
+                    best_match = candidate
+
+    # Require all words to match
+    if best_match and best_score == len(search_words):
+        return str(best_match)
+
+    return None
+
+def main():
+    for playlist in Path(PLAYLIST_DIR).glob("*.m3u"):
+    
+        logger.info(f"\nProcessing {playlist}")
+    
+        changed = False
+        output = []
+    
+        with open(playlist, "r", encoding="utf-8", errors="ignore") as f:
+            lines = f.readlines()
+    
+        for line in lines:
+            if not line or line.startswith("#"):
+                output.append(line)
+                continue
+
+            path = WORKDIR / Path(line.strip()).relative_to("/")
+
+            if not path.is_absolute() or path.exists():
+                output.append(path)
+                continue 
+            
+            logger.info(f"Missing: {path}")
+            replacement = find_replacement(path)
+    
+            if replacement:
+                logger.info(f"  -> Found: {replacement}")
+                output.append(replacement + "\n")
+                changed = True
+            else:
+                logger.info("  -> No match")
+                output.append(line)
+                exit() 
+        if changed and not dry_run:
+            
+            backup = playlist.with_suffix(playlist.suffix + ".bak")
+            if not backup.exists():
+                backup.write_text(
+                    playlist.read_text(encoding="utf-8"),
+                    encoding="utf-8"
+                )
+    
+            with open(playlist, "w", encoding="utf-8") as f:
+                f.writelines(output)
+    
+            logger.info("Updated playlist")
+
+if __name__ == "__main__":
+    main()