Server-scripts/media handling/reference_playlists_to_library.py

#!/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()