6f551ba439
Dockerized media library scanner for NAS drives. Identifies movies and series from filenames, fetches posters via the public IMDB suggestion API (no key required), and exposes a dark-themed web UI with multilingual search support. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
479 lines
17 KiB
Python
479 lines
17 KiB
Python
"""
|
|
scanner.py — Core media scanning logic.
|
|
Identifies movies and TV series from NAS volumes using parse-torrent-name.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import unicodedata
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import PTN
|
|
|
|
from config import NasConfig, Settings
|
|
from database import (
|
|
Episode, Movie, NasSource, ScanHistory, Season, Series,
|
|
UnmatchedFile, get_session,
|
|
)
|
|
from metadata import find_local_poster, get_movie_metadata, get_series_metadata
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Patterns that indicate a directory is a season folder
|
|
_SEASON_DIR_RE = re.compile(r"(?i)(?:season|saison|^s)\s*(\d+)$")
|
|
# Episode patterns in filename
|
|
_EPISODE_FILE_RE = re.compile(r"(?i)S\d{2}E\d{2}|\d{1,2}x\d{2}")
|
|
# Articles to strip during normalization
|
|
_LEADING_ARTICLES = re.compile(r"^(?:the|le|la|les|a|an)\s+", re.IGNORECASE)
|
|
# Characters to keep during normalization (letters, digits, space, hyphen)
|
|
_KEEP_CHARS = re.compile(r"[^\w\s\-]", re.UNICODE)
|
|
|
|
|
|
# ─────────────────────────────────────────
|
|
# Title normalization
|
|
# ─────────────────────────────────────────
|
|
|
|
def normalize_title(title: str) -> str:
|
|
title = title.lower().strip()
|
|
title = unicodedata.normalize("NFKD", title)
|
|
title = title.encode("ascii", "ignore").decode("ascii")
|
|
title = _KEEP_CHARS.sub("", title)
|
|
title = _LEADING_ARTICLES.sub("", title)
|
|
title = re.sub(r"\s+", " ", title).strip()
|
|
return title
|
|
|
|
|
|
# ─────────────────────────────────────────
|
|
# Series vs Movie detection
|
|
# ─────────────────────────────────────────
|
|
|
|
def _is_season_directory(path: Path) -> Optional[int]:
|
|
"""Return season number if any parent dir indicates a season folder, else None."""
|
|
for part in path.parts:
|
|
m = _SEASON_DIR_RE.search(part)
|
|
if m:
|
|
return int(m.group(1))
|
|
return None
|
|
|
|
|
|
def classify_file(file_path: Path) -> tuple[bool, dict]:
|
|
"""
|
|
Parse the filename and classify as series (True) or movie (False).
|
|
Returns (is_series, ptn_info).
|
|
"""
|
|
info = PTN.parse(file_path.stem)
|
|
|
|
# Priority 1: season directory name
|
|
season_from_dir = _is_season_directory(file_path)
|
|
if season_from_dir is not None:
|
|
if "season" not in info:
|
|
info["season"] = season_from_dir
|
|
return True, info
|
|
|
|
# Priority 2: SxxExx pattern in filename
|
|
if _EPISODE_FILE_RE.search(file_path.stem):
|
|
return True, info
|
|
|
|
# Priority 3: PTN returned both season and episode
|
|
if "season" in info and "episode" in info:
|
|
return True, info
|
|
|
|
return False, info
|
|
|
|
|
|
# ─────────────────────────────────────────
|
|
# Database helpers
|
|
# ─────────────────────────────────────────
|
|
|
|
def _upsert_nas_source(session, nas: NasConfig) -> NasSource:
|
|
record = session.query(NasSource).filter_by(path=nas.container_path).first()
|
|
if record is None:
|
|
record = NasSource(name=nas.name, path=nas.container_path)
|
|
session.add(record)
|
|
session.flush()
|
|
logger.info("Registered new NAS source: %s → %s", nas.name, nas.container_path)
|
|
else:
|
|
record.name = nas.name # update name if changed in .env
|
|
return record
|
|
|
|
|
|
def _get_or_create_series(session, title: str, year: Optional[int]) -> Series:
|
|
normalized = normalize_title(title)
|
|
record = session.query(Series).filter_by(title_normalized=normalized).first()
|
|
if record is None:
|
|
record = Series(title=title, title_normalized=normalized, year=year)
|
|
session.add(record)
|
|
session.flush()
|
|
return record
|
|
|
|
|
|
def _get_or_create_season(session, series: Series, season_number: int, nas_source: NasSource) -> Season:
|
|
record = (
|
|
session.query(Season)
|
|
.filter_by(series_id=series.id, season_number=season_number, nas_id=nas_source.id)
|
|
.first()
|
|
)
|
|
if record is None:
|
|
record = Season(series_id=series.id, season_number=season_number, nas_id=nas_source.id)
|
|
session.add(record)
|
|
session.flush()
|
|
return record
|
|
|
|
|
|
# ─────────────────────────────────────────
|
|
# File processing
|
|
# ─────────────────────────────────────────
|
|
|
|
def _local_poster(file_path: Path) -> Optional[str]:
|
|
local = find_local_poster(file_path)
|
|
return f"local:{local}" if local else None
|
|
|
|
|
|
def _fetch_posters_parallel(session) -> None:
|
|
"""
|
|
Fetch missing posters and/or IMDB IDs in parallel (10 workers).
|
|
- Items without poster_url → fetch both poster + imdb_id
|
|
- Items with poster_url but without imdb_id → fetch imdb_id only
|
|
"""
|
|
from sqlalchemy import or_
|
|
all_items = (
|
|
[(m.id, "movie", m.title, m.year, m.poster_url is None)
|
|
for m in session.query(Movie).filter(
|
|
or_(Movie.poster_url.is_(None), Movie.imdb_id.is_(None))
|
|
).all()] +
|
|
[(s.id, "series", s.title, s.year, s.poster_url is None)
|
|
for s in session.query(Series).filter(
|
|
or_(Series.poster_url.is_(None), Series.imdb_id.is_(None))
|
|
).all()]
|
|
)
|
|
if not all_items:
|
|
return
|
|
|
|
logger.info("Fetching metadata for %d items in parallel…", len(all_items))
|
|
|
|
def fetch_one(task):
|
|
item_id, kind, title, year, need_poster = task
|
|
if kind == "movie":
|
|
meta = get_movie_metadata(title, year)
|
|
else:
|
|
meta = get_series_metadata(title, year)
|
|
return item_id, kind, meta, need_poster
|
|
|
|
posters_fetched = 0
|
|
with ThreadPoolExecutor(max_workers=10) as pool:
|
|
futures = {pool.submit(fetch_one, t): t for t in all_items}
|
|
for future in as_completed(futures):
|
|
try:
|
|
item_id, kind, meta, need_poster = future.result()
|
|
updates = {}
|
|
if need_poster and meta.poster_url:
|
|
updates["poster_url"] = meta.poster_url
|
|
posters_fetched += 1
|
|
if meta.imdb_id:
|
|
updates["imdb_id"] = meta.imdb_id
|
|
if updates:
|
|
model = Movie if kind == "movie" else Series
|
|
session.query(model).filter_by(id=item_id).update(updates)
|
|
except Exception as exc:
|
|
logger.debug("Metadata fetch error: %s", exc)
|
|
|
|
session.commit()
|
|
logger.info("Metadata pass done — new posters: %d, total processed: %d",
|
|
posters_fetched, len(all_items))
|
|
|
|
|
|
|
|
def _process_file(
|
|
session,
|
|
file_path: Path,
|
|
nas_source: NasSource,
|
|
settings: Settings,
|
|
counters: dict,
|
|
poster_cache: dict,
|
|
) -> None:
|
|
str_path = str(file_path)
|
|
|
|
try:
|
|
file_size = file_path.stat().st_size
|
|
except OSError:
|
|
file_size = None
|
|
|
|
try:
|
|
is_series, info = classify_file(file_path)
|
|
except Exception as exc:
|
|
reason = f"PTN parse error: {exc}"
|
|
logger.warning("Unmatched %s — %s", str_path, reason)
|
|
_record_unmatched(session, str_path, nas_source.id, reason)
|
|
counters["unmatched"] += 1
|
|
return
|
|
|
|
title = info.get("title")
|
|
if not title:
|
|
reason = "No title extracted by PTN"
|
|
_record_unmatched(session, str_path, nas_source.id, reason)
|
|
counters["unmatched"] += 1
|
|
return
|
|
|
|
year = info.get("year")
|
|
quality = info.get("quality")
|
|
codec = info.get("codec")
|
|
|
|
# Fallback: extract year from title like "Black Bag (2025)" when PTN misses it
|
|
if not year:
|
|
m = re.search(r'\((\d{4})\)\s*$', title)
|
|
if m:
|
|
year = int(m.group(1))
|
|
title = title[:m.start()].strip()
|
|
|
|
if is_series:
|
|
season_num = info.get("season", 1)
|
|
episode_num = info.get("episode")
|
|
if episode_num is None:
|
|
reason = "Series detected but no episode number found"
|
|
_record_unmatched(session, str_path, nas_source.id, reason)
|
|
counters["unmatched"] += 1
|
|
return
|
|
|
|
series = _get_or_create_series(session, title, year)
|
|
|
|
if series.poster_url is None:
|
|
local = _local_poster(file_path)
|
|
if local:
|
|
series.poster_url = local
|
|
|
|
season = _get_or_create_season(session, series, int(season_num), nas_source)
|
|
|
|
existing = session.query(Episode).filter_by(file_path=str_path).first()
|
|
if existing is None:
|
|
episode = Episode(
|
|
season_id=season.id,
|
|
episode_number=int(episode_num),
|
|
title=info.get("episode_title"),
|
|
quality=quality,
|
|
codec=codec,
|
|
file_path=str_path,
|
|
file_size=file_size,
|
|
)
|
|
session.add(episode)
|
|
counters["added"] += 1
|
|
else:
|
|
existing.quality = quality
|
|
existing.codec = codec
|
|
existing.file_size = file_size
|
|
existing.updated_at = datetime.utcnow()
|
|
else:
|
|
existing = session.query(Movie).filter_by(file_path=str_path).first()
|
|
if existing is None:
|
|
movie = Movie(
|
|
title=title,
|
|
year=year,
|
|
quality=quality,
|
|
codec=codec,
|
|
file_path=str_path,
|
|
file_size=file_size,
|
|
poster_url=_local_poster(file_path),
|
|
nas_id=nas_source.id,
|
|
)
|
|
session.add(movie)
|
|
counters["added"] += 1
|
|
else:
|
|
existing.title = title
|
|
existing.year = year
|
|
existing.quality = quality
|
|
existing.codec = codec
|
|
existing.file_size = file_size
|
|
existing.updated_at = datetime.utcnow()
|
|
if existing.poster_url is None:
|
|
local = _local_poster(file_path)
|
|
if local:
|
|
existing.poster_url = local
|
|
|
|
counters["found"] += 1
|
|
|
|
|
|
def _record_unmatched(session, file_path: str, nas_id: int, reason: str) -> None:
|
|
record = UnmatchedFile(file_path=file_path, nas_id=nas_id, reason=reason)
|
|
session.add(record)
|
|
|
|
|
|
# ─────────────────────────────────────────
|
|
# Staleness checks (removal of missing files)
|
|
# ─────────────────────────────────────────
|
|
|
|
def _remove_stale_movies(session, nas_source: NasSource) -> int:
|
|
removed = 0
|
|
movies = session.query(Movie).filter_by(nas_id=nas_source.id).all()
|
|
for movie in movies:
|
|
if not Path(movie.file_path).exists():
|
|
logger.info("Removing stale movie: %s", movie.file_path)
|
|
session.delete(movie)
|
|
removed += 1
|
|
return removed
|
|
|
|
|
|
def _remove_stale_episodes(session, nas_source: NasSource) -> int:
|
|
removed = 0
|
|
seasons = session.query(Season).filter_by(nas_id=nas_source.id).all()
|
|
for season in seasons:
|
|
for episode in list(season.episodes):
|
|
if not Path(episode.file_path).exists():
|
|
logger.info("Removing stale episode: %s", episode.file_path)
|
|
session.delete(episode)
|
|
removed += 1
|
|
# Prune empty seasons
|
|
session.flush()
|
|
if not season.episodes:
|
|
# Re-query to check after flush
|
|
refreshed = session.query(Season).get(season.id)
|
|
if refreshed and not refreshed.episodes:
|
|
session.delete(refreshed)
|
|
# Prune series with no remaining seasons
|
|
for series in session.query(Series).all():
|
|
session.flush()
|
|
refreshed = session.query(Series).get(series.id)
|
|
if refreshed and not refreshed.seasons:
|
|
session.delete(refreshed)
|
|
return removed
|
|
|
|
|
|
# ─────────────────────────────────────────
|
|
# Public scan functions
|
|
# ─────────────────────────────────────────
|
|
|
|
def run_full_scan(settings: Settings, scan_id: int) -> None:
|
|
logger.info("Starting full scan (scan_id=%d)", scan_id)
|
|
session = get_session()
|
|
counters = {"found": 0, "added": 0, "removed": 0, "unmatched": 0}
|
|
poster_cache: dict = {}
|
|
|
|
try:
|
|
# Clear unmatched from previous scans
|
|
session.query(UnmatchedFile).delete()
|
|
|
|
for nas in settings.nas_list:
|
|
if not os.path.isdir(nas.container_path):
|
|
logger.warning("NAS path not accessible: %s — skipping", nas.container_path)
|
|
continue
|
|
|
|
nas_source = _upsert_nas_source(session, nas)
|
|
session.commit()
|
|
|
|
logger.info("Scanning NAS: %s (%s)", nas.name, nas.container_path)
|
|
|
|
for root, _dirs, files in os.walk(nas.container_path):
|
|
for filename in files:
|
|
ext = Path(filename).suffix.lower()
|
|
if ext not in settings.video_extensions:
|
|
continue
|
|
file_path = Path(root) / filename
|
|
_process_file(session, file_path, nas_source, settings, counters, poster_cache)
|
|
|
|
removed = _remove_stale_movies(session, nas_source)
|
|
removed += _remove_stale_episodes(session, nas_source)
|
|
counters["removed"] += removed
|
|
|
|
nas_source.last_scan = datetime.utcnow()
|
|
session.commit()
|
|
|
|
_fetch_posters_parallel(session)
|
|
_finish_scan(session, scan_id, "completed", counters)
|
|
logger.info("Full scan complete: %s", counters)
|
|
|
|
except Exception as exc:
|
|
logger.exception("Full scan failed: %s", exc)
|
|
_finish_scan(session, scan_id, "failed", counters, str(exc))
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def run_incremental_scan(settings: Settings, scan_id: int) -> None:
|
|
logger.info("Starting incremental scan (scan_id=%d)", scan_id)
|
|
session = get_session()
|
|
counters = {"found": 0, "added": 0, "removed": 0, "unmatched": 0}
|
|
poster_cache: dict = {}
|
|
|
|
try:
|
|
for nas in settings.nas_list:
|
|
if not os.path.isdir(nas.container_path):
|
|
logger.warning("NAS path not accessible: %s — skipping", nas.container_path)
|
|
continue
|
|
|
|
nas_source = _upsert_nas_source(session, nas)
|
|
session.commit()
|
|
|
|
since = nas_source.last_scan
|
|
logger.info(
|
|
"Incremental scan NAS: %s — changes since %s",
|
|
nas.name,
|
|
since.isoformat() if since else "never",
|
|
)
|
|
|
|
since_ts = since.timestamp() if since else 0.0
|
|
|
|
for root, _dirs, files in os.walk(nas.container_path):
|
|
for filename in files:
|
|
ext = Path(filename).suffix.lower()
|
|
if ext not in settings.video_extensions:
|
|
continue
|
|
file_path = Path(root) / filename
|
|
try:
|
|
mtime = file_path.stat().st_mtime
|
|
except OSError:
|
|
continue
|
|
if mtime > since_ts:
|
|
_process_file(session, file_path, nas_source, settings, counters, poster_cache)
|
|
|
|
removed = _remove_stale_movies(session, nas_source)
|
|
removed += _remove_stale_episodes(session, nas_source)
|
|
counters["removed"] += removed
|
|
|
|
nas_source.last_scan = datetime.utcnow()
|
|
session.commit()
|
|
|
|
_fetch_posters_parallel(session)
|
|
_finish_scan(session, scan_id, "completed", counters)
|
|
logger.info("Incremental scan complete: %s", counters)
|
|
|
|
except Exception as exc:
|
|
logger.exception("Incremental scan failed: %s", exc)
|
|
_finish_scan(session, scan_id, "failed", counters, str(exc))
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _finish_scan(session, scan_id: int, status: str, counters: dict, error: str = None) -> None:
|
|
record = session.query(ScanHistory).get(scan_id)
|
|
if record:
|
|
record.status = status
|
|
record.finished_at = datetime.utcnow()
|
|
record.files_found = counters["found"]
|
|
record.files_added = counters["added"]
|
|
record.files_removed = counters["removed"]
|
|
record.files_unmatched = counters["unmatched"]
|
|
record.error_message = error
|
|
session.commit()
|
|
|
|
|
|
def create_scan_record(scan_type: str) -> int:
|
|
"""Insert a new running scan record and return its ID."""
|
|
session = get_session()
|
|
try:
|
|
record = ScanHistory(scan_type=scan_type, status="running", started_at=datetime.utcnow())
|
|
session.add(record)
|
|
session.commit()
|
|
return record.id
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def has_any_scan_completed() -> bool:
|
|
session = get_session()
|
|
try:
|
|
return session.query(ScanHistory).filter_by(status="completed").count() > 0
|
|
finally:
|
|
session.close()
|