mirror of
https://github.com/myronblair/kino-app
synced 2026-07-27 19:24:45 -05:00
Myron Blair
43340c0f49
- Navbar: drop the Users link (already reachable from Admin's quick-links) - Admin quick-links: drop Settings (already reachable from the navbar for every user) - library_cleanup.py: parallelize the missing-file stat checks across a small thread pool instead of one file at a time — sequential NAS round trips were taking minutes for a library of any real size, even after moving them off the event loop.
210 lines
11 KiB
Python
210 lines
11 KiB
Python
"""Library integrity scan — missing files, orphaned records/disk usage, stuck jobs, duplicates.
|
|
Read-only by design: scan() only reports findings; server.py's /library/cleanup/fix endpoint
|
|
performs the actual deletion/reset the admin picked, one category at a time.
|
|
"""
|
|
import shutil
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Dict, List
|
|
|
|
import asyncio
|
|
|
|
# Movie/episode/track files often live on an NFS/CIFS-mounted NAS, where every is_file() stat
|
|
# is a real network round trip. Checking hundreds of them one at a time (even off the event
|
|
# loop) can take minutes; a bounded worker pool checks them concurrently instead.
|
|
_STAT_WORKERS = 24
|
|
|
|
|
|
def _file_path(storage_type: str, storage_path: str, videos_dir: Path) -> Path:
|
|
if storage_type in ("radarr", "sonarr", "scan"):
|
|
return Path(storage_path)
|
|
return videos_dir / storage_path
|
|
|
|
|
|
def _scan_filesystem(movies: List[Dict], episodes: List[Dict], tracks: List[Dict],
|
|
videos_dir: Path, hls_dir: Path, live_ids: set) -> Dict[str, List[Dict]]:
|
|
"""Every filesystem stat call in the whole scan lives here, run off the event loop via
|
|
asyncio.to_thread, with the individual stat calls themselves parallelized across a small
|
|
worker pool — sequential one-at-a-time NAS round trips were taking minutes for a library
|
|
of any real size."""
|
|
out: Dict[str, List[Dict]] = {"missing_files": [], "orphaned_hls": []}
|
|
|
|
candidates = [] # (finding_dict, path_to_check)
|
|
for m in movies:
|
|
if m.get("storage_type") not in ("local", "radarr", "scan") or not m.get("storage_path"):
|
|
continue
|
|
candidates.append((
|
|
{"id": m["id"], "content_type": "movie", "label": m.get("title", "Untitled"), "detail": m["storage_path"]},
|
|
_file_path(m["storage_type"], m["storage_path"], videos_dir),
|
|
))
|
|
for e in episodes:
|
|
if e.get("_orphaned"):
|
|
continue # already reported as orphaned_episodes; don't also report a missing file
|
|
if e.get("storage_type") not in ("local", "sonarr", "scan") or not e.get("storage_path"):
|
|
continue
|
|
label = f"S{e.get('season_number')}E{e.get('episode_number')} {e.get('title') or ''}".strip()
|
|
candidates.append((
|
|
{"id": e["id"], "content_type": "episode", "label": label, "detail": e["storage_path"]},
|
|
_file_path(e["storage_type"], e["storage_path"], videos_dir),
|
|
))
|
|
for t in tracks:
|
|
candidates.append((
|
|
{"id": t["id"], "content_type": "track", "label": f"{t.get('artist','')} — {t.get('title','Untitled')}", "detail": t["storage_path"]},
|
|
Path(t["storage_path"]),
|
|
))
|
|
|
|
with ThreadPoolExecutor(max_workers=_STAT_WORKERS) as pool:
|
|
exists = pool.map(lambda pair: pair[1].is_file(), candidates)
|
|
for (finding, _path), is_present in zip(candidates, exists):
|
|
if not is_present:
|
|
out["missing_files"].append(finding)
|
|
|
|
if hls_dir.is_dir():
|
|
for d in hls_dir.iterdir():
|
|
if d.is_dir() and d.name not in live_ids:
|
|
size = sum(f.stat().st_size for f in d.rglob("*") if f.is_file())
|
|
out["orphaned_hls"].append({"id": d.name, "content_type": "hls_dir", "label": d.name, "detail": f"{round(size / 1_000_000)} MB reclaimable"})
|
|
|
|
return out
|
|
|
|
|
|
async def scan(db, videos_dir: Path, hls_dir: Path) -> Dict[str, List[Dict]]:
|
|
findings: Dict[str, List[Dict]] = {
|
|
"missing_files": [], "orphaned_episodes": [], "orphaned_hls": [],
|
|
"stuck_jobs": [], "duplicate_paths": [], "orphaned_subtitles": [], "dangling_refs": [],
|
|
}
|
|
|
|
movies = await db.movies.find({}, {"_id": 0, "id": 1, "title": 1, "storage_type": 1, "storage_path": 1}).to_list(10000)
|
|
episodes = await db.episodes.find({}, {"_id": 0, "id": 1, "title": 1, "show_id": 1, "season_number": 1, "episode_number": 1, "storage_type": 1, "storage_path": 1}).to_list(20000)
|
|
tracks = await db.tracks.find({}, {"_id": 0, "id": 1, "title": 1, "artist": 1, "storage_path": 1}).to_list(20000)
|
|
show_ids = set(await db.shows.distinct("id"))
|
|
|
|
for e in episodes:
|
|
label = f"S{e.get('season_number')}E{e.get('episode_number')} {e.get('title') or ''}".strip()
|
|
if e.get("show_id") not in show_ids:
|
|
findings["orphaned_episodes"].append({"id": e["id"], "content_type": "episode", "label": label, "detail": f"show_id {e.get('show_id')} not found"})
|
|
e["_orphaned"] = True
|
|
|
|
live_ids = {m["id"] for m in movies} | {e["id"] for e in episodes}
|
|
fs_findings = await asyncio.to_thread(_scan_filesystem, movies, episodes, tracks, videos_dir, hls_dir, live_ids)
|
|
findings["missing_files"] = fs_findings["missing_files"]
|
|
findings["orphaned_hls"] = fs_findings["orphaned_hls"]
|
|
|
|
# --- Stuck transcode jobs (running with no progress for 2+ hours — likely dead from a restart) ---
|
|
now = datetime.now(timezone.utc)
|
|
running = await db.transcode_queue.find({"status": "running"}, {"_id": 0}).to_list(500)
|
|
for j in running:
|
|
started = j.get("started_at")
|
|
if not started:
|
|
continue
|
|
try:
|
|
age_hours = (now - datetime.fromisoformat(started)).total_seconds() / 3600
|
|
except ValueError:
|
|
continue
|
|
if age_hours > 2:
|
|
findings["stuck_jobs"].append({"id": j["id"], "content_type": "job", "label": j.get("movie_title") or j["movie_id"], "detail": f"running {round(age_hours, 1)}h with no completion"})
|
|
|
|
# --- Duplicate storage_path within the same collection (same file imported twice) ---
|
|
for coll_name, coll, label_fn in (
|
|
("movies", db.movies, lambda d: d.get("title", "Untitled")),
|
|
("episodes", db.episodes, lambda d: d.get("title") or d.get("id")),
|
|
("tracks", db.tracks, lambda d: d.get("title", "Untitled")),
|
|
):
|
|
docs = await coll.find({"storage_path": {"$ne": None}}, {"_id": 0}).sort("created_at", 1).to_list(20000)
|
|
seen = {}
|
|
for d in docs:
|
|
key = d.get("storage_path")
|
|
if not key:
|
|
continue
|
|
if key in seen:
|
|
findings["duplicate_paths"].append({"id": d["id"], "content_type": coll_name[:-1], "label": label_fn(d), "detail": f"duplicate of {seen[key]}"})
|
|
else:
|
|
seen[key] = d["id"]
|
|
|
|
# --- Orphaned subtitles (reference a movie/episode that no longer exists) ---
|
|
movie_ids = {m["id"] for m in movies}
|
|
episode_ids = {e["id"] for e in episodes}
|
|
subs = await db.subtitles.find({}, {"_id": 0}).to_list(5000)
|
|
for s in subs:
|
|
mid, eid = s.get("movie_id") or "", s.get("episode_id") or ""
|
|
if mid and mid not in movie_ids:
|
|
findings["orphaned_subtitles"].append({"id": s["id"], "content_type": "subtitle", "label": s.get("label", "Subtitle"), "detail": f"movie_id {mid} not found"})
|
|
elif eid and eid not in episode_ids:
|
|
findings["orphaned_subtitles"].append({"id": s["id"], "content_type": "subtitle", "label": s.get("label", "Subtitle"), "detail": f"episode_id {eid} not found"})
|
|
|
|
# --- Dangling watchlist/progress rows (reference a deleted movie) ---
|
|
for coll_name, coll in (("watchlist", db.watchlist), ("progress", db.progress)):
|
|
rows = await coll.find({}, {"_id": 0}).to_list(20000)
|
|
for r in rows:
|
|
if r.get("movie_id") and r["movie_id"] not in movie_ids:
|
|
findings["dangling_refs"].append({"id": r["id"] if "id" in r else f"{r['profile_id']}:{r['movie_id']}", "content_type": coll_name, "label": r["movie_id"], "detail": f"references deleted movie {r['movie_id']}", "profile_id": r.get("profile_id"), "movie_id": r.get("movie_id")})
|
|
|
|
# --- Dangling episode_progress rows (reference a deleted episode) ---
|
|
ep_rows = await db.episode_progress.find({}, {"_id": 0}).to_list(20000)
|
|
for r in ep_rows:
|
|
if r.get("episode_id") and r["episode_id"] not in episode_ids:
|
|
findings["dangling_refs"].append({"id": f"{r['profile_id']}:{r['episode_id']}", "content_type": "episode_progress", "label": r["episode_id"], "detail": f"references deleted episode {r['episode_id']}", "profile_id": r.get("profile_id"), "episode_id": r.get("episode_id")})
|
|
|
|
return findings
|
|
|
|
|
|
async def _delete_content(db, content_type: str, item_id: str) -> None:
|
|
"""Deletes a movie/episode/track and every row that references it, so a Fix never
|
|
leaves orphans behind for the next scan to catch on a second pass."""
|
|
if content_type == "movie":
|
|
await db.movies.delete_one({"id": item_id})
|
|
await db.watchlist.delete_many({"movie_id": item_id})
|
|
await db.progress.delete_many({"movie_id": item_id})
|
|
await db.subtitles.delete_many({"movie_id": item_id})
|
|
elif content_type == "episode":
|
|
await db.episodes.delete_one({"id": item_id})
|
|
await db.episode_progress.delete_many({"episode_id": item_id})
|
|
await db.subtitles.delete_many({"episode_id": item_id})
|
|
elif content_type == "track":
|
|
await db.tracks.delete_one({"id": item_id})
|
|
else:
|
|
raise ValueError(f"Unknown content_type: {content_type}")
|
|
|
|
|
|
async def fix_one(db, videos_dir: Path, hls_dir: Path, category: str, item: Dict) -> None:
|
|
content_type = item.get("content_type")
|
|
item_id = item.get("id")
|
|
|
|
if category == "missing_files":
|
|
await _delete_content(db, content_type, item_id)
|
|
|
|
elif category == "orphaned_episodes":
|
|
await _delete_content(db, "episode", item_id)
|
|
|
|
elif category == "orphaned_hls":
|
|
shutil.rmtree(hls_dir / item_id, ignore_errors=True)
|
|
|
|
elif category == "stuck_jobs":
|
|
await db.transcode_queue.update_one({"id": item_id}, {"$set": {
|
|
"status": "failed", "error": "Reset by library cleanup (stuck with no progress)",
|
|
"finished_at": datetime.now(timezone.utc).isoformat(),
|
|
}})
|
|
|
|
elif category == "duplicate_paths":
|
|
await _delete_content(db, content_type, item_id)
|
|
|
|
elif category == "orphaned_subtitles":
|
|
sub = await db.subtitles.find_one({"id": item_id}, {"_id": 0})
|
|
if sub and sub.get("storage_path"):
|
|
try:
|
|
(Path(str(videos_dir.parent / "subtitles")) / sub["storage_path"]).unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
await db.subtitles.delete_one({"id": item_id})
|
|
|
|
elif category == "dangling_refs":
|
|
if content_type == "episode_progress":
|
|
await db.episode_progress.delete_one({"profile_id": item.get("profile_id"), "episode_id": item.get("episode_id")})
|
|
else:
|
|
coll = {"watchlist": db.watchlist, "progress": db.progress}[content_type]
|
|
await coll.delete_one({"profile_id": item.get("profile_id"), "movie_id": item.get("movie_id")})
|
|
|
|
else:
|
|
raise ValueError(f"Unknown cleanup category: {category}")
|