kino-app/backend/library_scan.py

"""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>. Filename/path-derived only — no file reads, so this
    stays fast even over a network share with a large library. Embedded tags (better titles, real
    duration) are read later, only for the tracks actually selected for import — see tags_for_file().
    """
    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
                track_number = None
                title = f.stem
                fm = _TRACK_NUM_RE.match(f.stem)
                if fm:
                    track_number = int(fm.group(1))
                    title = fm.group(2)
                out.append({
                    "artist": artist_dir.name,
                    "album": album_dir.name,
                    "title": _clean_title(title) or title,
                    "track_number": track_number,
                    "duration_seconds": 0,
                    "storage_path": str(f),
                    "size_bytes": f.stat().st_size,
                })
    return out


def tags_for_file(path_str: str) -> Dict:
    """Embedded-tag lookup for a single file — used at import time, not during scan."""
    return _read_tags(Path(path_str))