Add filesystem Library Scan (movies/tv/music) and a full Music library

- New admin Library Scan page walks the NAS mounts directly for movies,
  TV, and music not already in Radarr/Sonarr/local storage, and imports
  them with storage_type=scan (fully integrated: streamable, transcodable,
  listed in Admin) — Radarr/Sonarr import flows are untouched.
- New Music section (nav tab, album-grid browse, per-album track list,
  persistent bottom player with queue/seek) backed by a Track model and
  range-request audio streaming endpoint.
- MusicBrainz + Cover Art Archive integration (free, no API key) to fill
  in album art for scanned tracks, paced to their 1 req/sec rate limit.
- Mounted the NAS music share into CT104 (new mp2 LXC mountpoint,
  CIFS read-only) alongside the existing video mount.
- Optional Bluetooth/output-device picker on the player bar via the Audio
  Output Devices API (setSinkId) for already-paired speakers — Chrome/Edge
  only, since browsers can't pair new BT devices directly.
This commit is contained in:
Myron Blair
2026-07-26 20:25:41 -05:00
parent f7626ced2e
commit 514cbad4ab
15 changed files with 975 additions and 40 deletions
+167
View File
@@ -0,0 +1,167 @@
"""Filesystem scanner for content already sitting on the NAS mounts —
covers movies/tv placed outside Radarr/Sonarr, and the music library
(nothing else in StreamHoard touches music). Read-only: never writes files,
just walks paths and reports what it finds. Callers (server.py) cross-check
results against the DB to decide what's new vs. already imported.
"""
import re
from pathlib import Path
from typing import Dict, List, Optional
from mutagen import File as MutagenFile
MOVIE_EXTS = {".mp4", ".mkv", ".avi", ".m4v", ".mov"}
AUDIO_EXTS = {".mp3", ".m4a", ".flac", ".aac", ".wav", ".ogg"}
MOVIES_DIR = Path("/mnt/nas/video/movies")
TV_DIR = Path("/mnt/nas/video/tv")
MUSIC_DIR = Path("/mnt/nas/music/Music")
_YEAR_RE = re.compile(r"[\[\(]\s*(\d{4})\s*[\]\)]")
_SEASON_DIR_RE = re.compile(r"season\s*(\d+)", re.IGNORECASE)
_EPISODE_RE = re.compile(r"[Ss](\d{1,2})[Ee](\d{1,3})")
_TRACK_NUM_RE = re.compile(r"^(\d{1,3})[\s._-]+(.*)$")
def _clean_title(name: str) -> str:
name = _YEAR_RE.sub("", name)
return re.sub(r"\s+", " ", name).strip(" -_.")
def _parse_year(name: str) -> Optional[int]:
m = _YEAR_RE.search(name)
return int(m.group(1)) if m else None
def _largest_video_file(files: List[Path]) -> Optional[Path]:
vids = [f for f in files if f.suffix.lower() in MOVIE_EXTS]
if not vids:
return None
return max(vids, key=lambda f: f.stat().st_size)
def scan_movies() -> List[Dict]:
"""Each movies/ entry is either a standalone video file or a folder containing one."""
if not MOVIES_DIR.is_dir():
return []
out = []
for entry in sorted(MOVIES_DIR.iterdir()):
if entry.name.startswith("@") or entry.name.startswith("#"):
continue # Synology internal dirs (@eaDir, #recycle)
if entry.is_file():
if entry.suffix.lower() not in MOVIE_EXTS:
continue
title_raw = entry.stem
file_path = entry
elif entry.is_dir():
try:
files = [f for f in entry.rglob("*") if f.is_file()]
except OSError:
continue
file_path = _largest_video_file(files)
if not file_path:
continue
title_raw = entry.name
else:
continue
out.append({
"title": _clean_title(title_raw) or title_raw,
"year": _parse_year(title_raw),
"storage_path": str(file_path),
"size_bytes": file_path.stat().st_size,
})
return out
def scan_tv() -> List[Dict]:
"""tv/<Show>/Season N/<episode files>. Groups by show, one entry per episode."""
if not TV_DIR.is_dir():
return []
out = []
for show_dir in sorted(TV_DIR.iterdir()):
if not show_dir.is_dir() or show_dir.name.startswith("@") or show_dir.name.startswith("#"):
continue
show_title = _clean_title(show_dir.name) or show_dir.name
for season_dir in sorted(show_dir.iterdir()):
if not season_dir.is_dir():
continue
m = _SEASON_DIR_RE.search(season_dir.name)
season_number = int(m.group(1)) if m else 1
for f in sorted(season_dir.iterdir()):
if not f.is_file() or f.suffix.lower() not in MOVIE_EXTS:
continue
em = _EPISODE_RE.search(f.stem)
if em:
season_number = int(em.group(1))
episode_number = int(em.group(2))
title = _clean_title(f.stem[em.end():]) or f"Episode {episode_number}"
else:
episode_number = None
title = _clean_title(f.stem)
out.append({
"show_title": show_title,
"season_number": season_number,
"episode_number": episode_number,
"title": title,
"storage_path": str(f),
"size_bytes": f.stat().st_size,
})
return out
def _read_tags(path: Path) -> Dict:
try:
tags = MutagenFile(path, easy=True)
except Exception:
tags = None
duration = 0.0
artist = album = title = track_number = None
if tags is not None:
if getattr(tags, "info", None) is not None:
duration = getattr(tags.info, "length", 0.0) or 0.0
artist = (tags.get("artist") or [None])[0]
album = (tags.get("album") or [None])[0]
title = (tags.get("title") or [None])[0]
tn = (tags.get("tracknumber") or [None])[0]
if tn:
try:
track_number = int(str(tn).split("/")[0])
except ValueError:
track_number = None
return {"duration": duration, "artist": artist, "album": album, "title": title, "track_number": track_number}
def scan_music() -> List[Dict]:
"""Music/<Artist>/<Album>/<NN Title.ext>. Prefers embedded tags, falls back to path/filename."""
if not MUSIC_DIR.is_dir():
return []
out = []
for artist_dir in sorted(MUSIC_DIR.iterdir()):
if not artist_dir.is_dir() or artist_dir.name.startswith("@") or artist_dir.name.startswith("#"):
continue
for album_dir in sorted(artist_dir.iterdir()):
if not album_dir.is_dir():
continue
for f in sorted(album_dir.iterdir()):
if not f.is_file() or f.suffix.lower() not in AUDIO_EXTS:
continue
tags = _read_tags(f)
track_number = tags["track_number"]
title = tags["title"]
if not title:
fm = _TRACK_NUM_RE.match(f.stem)
if fm:
track_number = track_number or int(fm.group(1))
title = fm.group(2)
else:
title = f.stem
out.append({
"artist": tags["artist"] or artist_dir.name,
"album": tags["album"] or album_dir.name,
"title": _clean_title(title) or title,
"track_number": track_number,
"duration_seconds": round(tags["duration"], 1),
"storage_path": str(f),
"size_bytes": f.stat().st_size,
})
return out