"""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 datetime import datetime, timezone from pathlib import Path from typing import Dict, List 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 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": [], } # --- Missing files --- movies = await db.movies.find({}, {"_id": 0, "id": 1, "title": 1, "storage_type": 1, "storage_path": 1}).to_list(10000) for m in movies: if m.get("storage_type") not in ("local", "radarr", "scan") or not m.get("storage_path"): continue if not _file_path(m["storage_type"], m["storage_path"], videos_dir).is_file(): findings["missing_files"].append({"id": m["id"], "content_type": "movie", "label": m.get("title", "Untitled"), "detail": m["storage_path"]}) 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) 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"}) continue if e.get("storage_type") not in ("local", "sonarr", "scan") or not e.get("storage_path"): continue if not _file_path(e["storage_type"], e["storage_path"], videos_dir).is_file(): findings["missing_files"].append({"id": e["id"], "content_type": "episode", "label": label, "detail": e["storage_path"]}) tracks = await db.tracks.find({}, {"_id": 0, "id": 1, "title": 1, "artist": 1, "storage_path": 1}).to_list(20000) for t in tracks: if not Path(t["storage_path"]).is_file(): findings["missing_files"].append({"id": t["id"], "content_type": "track", "label": f"{t.get('artist','')} — {t.get('title','Untitled')}", "detail": t["storage_path"]}) # --- Orphaned HLS output dirs (no matching movie or episode id) --- live_ids = {m["id"] for m in movies} | {e["id"] for e in episodes} 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()) findings["orphaned_hls"].append({"id": d.name, "content_type": "hls_dir", "label": d.name, "detail": f"{round(size / 1_000_000)} MB reclaimable"}) # --- 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}")