import os import re import shutil from lyra_worker.types import MBTarget _ILLEGAL = re.compile(r'[<>:"/\\|?*\x00-\x1f]') def safe_name(s: str) -> str: """A filesystem-safe path component (single segment, no separators).""" cleaned = _ILLEGAL.sub("_", s).strip().strip(".").strip() return cleaned or "Unknown" def album_dir(dest_root: str, target: MBTarget) -> str: """`{dest_root}/{safe artist}/{safe album}[ (year)]`.""" album = safe_name(target.album) if target.year: album = f"{album} ({target.year})" return f"{dest_root}/{safe_name(target.artist)}/{album}" _AUDIO_EXT = {".flac", ".mp3", ".m4a", ".opus", ".ogg", ".aac", ".wav"} _COVER_STEMS = {"cover", "folder", "front"} _COVER_EXT = {".jpg", ".jpeg", ".png"} def staging_dir(dest_root: str, job_id: str) -> str: """A per-job download staging directory, on the same filesystem as the library.""" return f"{dest_root}/.staging/{job_id}" def clear_staging_root(dest_root: str) -> None: """Remove every staging dir (called at worker startup to sweep crash orphans).""" shutil.rmtree(f"{dest_root}/.staging", ignore_errors=True) def _find_cover(staging: str) -> str | None: for root, _dirs, files in os.walk(staging): for name in sorted(files): stem, ext = os.path.splitext(name) if ext.lower() in _COVER_EXT and stem.lower() in _COVER_STEMS: return os.path.join(root, name) return None def import_album(staging: str, final: str) -> str: """Move the album's audio files (+ one cover image) from `staging` into a fresh `final`, replacing any existing folder there. Non-audio artifacts and nested subfolders are left behind (the caller removes `staging`). Tolerates a missing `staging`. Returns `final`.""" if os.path.isdir(final): shutil.rmtree(final) os.makedirs(final, exist_ok=True) if not os.path.isdir(staging): return final for name in sorted(os.listdir(staging)): src = os.path.join(staging, name) if os.path.isfile(src) and os.path.splitext(name)[1].lower() in _AUDIO_EXT: shutil.move(src, os.path.join(final, name)) cover = _find_cover(staging) if cover is not None: shutil.move(cover, os.path.join(final, "cover.jpg")) return final