#!/usr/bin/env python3 import os import re import logging import argparse 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"} TRACE = 5 score_threshold = 0.5 logging.addLevelName(TRACE, "TRACE") class CustomLogger(logging.Logger): def trace(self, msg: str, *args, **kwargs) -> None: if self.isEnabledFor(TRACE): self._log(TRACE, msg, args, **kwargs) logging.setLoggerClass(CustomLogger) logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment] parser = argparse.ArgumentParser() parser.add_argument( "-ll", "--log-level", type=str.upper, default="INFO", choices=["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], help="Set log level", ) args = parser.parse_args() LOG_LEVELS = { "TRACE": TRACE, "DEBUG": logging.DEBUG, "INFO": logging.INFO, "WARNING": logging.WARNING, "ERROR": logging.ERROR, "CRITICAL": logging.CRITICAL, } logging.basicConfig( level=LOG_LEVELS[args.log_level.upper()], format="%(asctime)s %(levelname)s %(message)s", ) logger.trace("Test") def words(text): return set(re.findall(r"\b[\w']+\b", text.lower())) def score_match(search_words, candidate_words): intersection = search_words & candidate_words union = search_words | candidate_words if not union: return 0.0 return len(intersection) / len(union) def find_folder(directory: Path, folder: str) -> Path: search_words = words(folder) best_match = Path("") best_score = 0.0 for candidate in directory.iterdir(): # depth 1 only if not candidate.is_dir(): continue candidate_words = words(candidate.name) score = len(search_words & candidate_words) / len(search_words) if score > best_score: best_score = score best_match = candidate return best_match 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 logger.debug(f"Folder Count Index: {folder_count}") try: artist = p.parts[len(WORKDIR.parts) + 1] # /music/Artist/Album/File logger.debug(f"Artist: {artist}") except IndexError: logger.error(f"No Artist found for {p}") return None artist_dir = find_folder(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 search_dir = artist_dir if folder_count >= 2: try: album = p.parts[len(WORKDIR.parts) + 2] # .../Artist/Album/File logger.debug(f"Album: {album}") except IndexError: logger.error(f"No album found for {p}") return None album_dir = find_folder(artist_dir, album) logger.debug(f"Album path: {album_dir}") if not album_dir.exists(): logger.error(f"Album dir for '{album}' does not exist") return None search_dir = album_dir if folder_count >= 3: try: subalbum = p.parts[len(WORKDIR.parts) + 3] # .../Artist/Album/File logger.debug(f"Sub-Album: {subalbum}") except IndexError: logger.error(f"No Sub-Album found for {p}") return None subalbum_dir = find_folder(album_dir, subalbum) logger.debug(f"Sub-Album path: {subalbum_dir}") if not subalbum_dir.exists(): logger.error(f"Sub-Album dir for '{subalbum}' does not exist") return None search_dir = subalbum_dir search_words = words(p.stem) logger.debug(f"Search words {search_words}") best_match = None best_score = 0 for root, _, files in os.walk(search_dir): for filename in files: candidate = Path(root) / filename candidate_parts = candidate.stem.split(" - ") logger.trace(f"Candidate: {candidate}") logger.trace(f"parts: {candidate_parts}") if candidate.suffix.lower() not in AUDIO_EXTS: continue if len(candidate_parts) == 4: logger.trace(f"Lidarr Filename: {len(candidate_parts)} ") candidate_words = words(" ".join(candidate_parts[2:])) score = score_match(search_words, candidate_words) logger.trace( f"search: {search_words} candidates: {candidate_words}, score/best {score} / {best_score}" ) if score > best_score: best_score = score best_match = candidate else: continue logger.trace(f"Fallback Filename: {len(candidate_parts)} ") candidate_words = words(candidate.stem) score = score_match(search_words, candidate_words) logger.trace(f"words: {candidate_words} {score / {best_score}}") if score > best_score: best_score = score best_match = candidate # Require all words to match if best_match and best_score > score_threshold: logger.info(f"Best Match: {best_match}") return str(best_match) logger.error( f"No match found with score greater {score_treshold}, best match: {best_score}:{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.warning(f" -> No match for {path}") 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()