reference all songs in playlist after Lidarr update
This commit is contained in:
lucas committed
1 parent
045861ada6
commit
2eb7b8d6d1
1 file changed
-243
@@ -1,243 +0,0 @@
|
||||
#!/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()
|
||||
Reference in new issue
Block a user