From fd91e7d6788304842e4f68befdc24c933c001b7e Mon Sep 17 00:00:00 2001 From: Lucas Media Date: Thu, 15 Oct 2026 09:42:46 +0000 Subject: [PATCH] python script to reference media files to playlist --- .../reference_playlists_to_library.py | 193 ++++++++++++------ 1 file changed, 136 insertions(+), 57 deletions(-) diff --git a/media handling/reference_playlists_to_library.py b/media handling/reference_playlists_to_library.py index 60334b5..91b975d 100755 --- a/media handling/reference_playlists_to_library.py +++ b/media handling/reference_playlists_to_library.py @@ -3,127 +3,207 @@ 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 -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() - ] +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", ) -logger.info('Started') + +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[-(folder_count + 1)] # /music/Artist/Album/File + 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 = Path(MUSIC_DIR) / artist + 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 - if folder_count > 1: + + search_dir = artist_dir + + if folder_count >= 2: try: - album = p.parts[-(folder_count)] # .../Artist/Album/File + 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 = Path(MUSIC_DIR).rglob(album) - - try: - album_dir = next(album_dir) - logger.debug(f"Album path: {album_dir}") - except StopIteration: + + 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_words = words(p.stem) + 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(artist_dir): + for root, _, files in os.walk(search_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}") + + 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.debug(f"Lidarr Filename: {len(candidate_parts)} ") + logger.trace(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) + 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.debug(f"Fallback Filename: {len(candidate_parts)} ") + logger.trace(f"Fallback Filename: {len(candidate_parts)} ") candidate_words = words(candidate.stem) - logger.debug(f"words: {candidate_words}") - score = len(search_words & candidate_words) + 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 == len(search_words): + 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) @@ -133,32 +213,31 @@ def main(): if not path.is_absolute() or path.exists(): output.append(path) - continue - + 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") + logger.warning(f" -> No match for {path}") output.append(line) - exit() + 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" + 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()