165 lines
4.7 KiB
Python
Executable File
165 lines
4.7 KiB
Python
Executable File
#!/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()
|