LeoVasanko/mediahive
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

Improve scanner cancellation and offload heavy index I/O

Browse Source
This commit is contained in:
LeoVasanko
2026-05-29 21:48:23 +00:00
parent 359ca0c5b4
commit 25e9bab57b
10 changed files with 379 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mediahive/hivescan/images.py
+8
View File
@@ -29,6 +29,14 @@ def _get_image_client() -> httpx.AsyncClient:
return _image_client
async def close_image_client() -> None:
"""Close the persistent image HTTP client if it was created."""
global _image_client
if _image_client is not None:
await _image_client.aclose()
_image_client = None
async def _download_image(url: str, output_path: Path, description: str) -> str | None:
"""Download an image from URL to output path."""
ap = AsyncPath(output_path)
mediahive/hivescan/indexer.py
+7 -9
View File
@@ -62,7 +62,10 @@ async def _build_torrent_info(
probe_info = await probe_media_info(str(probe_target))
if item.content_hash and item.content_hash.size == 0:
item.content_hash.size = await get_directory_size(item.content_hash.path)
item.content_hash.size = await asyncio.to_thread(
get_directory_size,
item.content_hash.path,
)
size = item.content_hash.size if item.content_hash else None
added_at = await get_added_timestamp(item.path)
@@ -189,8 +192,9 @@ async def _collect_episode_files(
)
if not already_added:
if item.content_hash and item.content_hash.size == 0:
item.content_hash.size = await get_directory_size(
item.content_hash.path
item.content_hash.size = await asyncio.to_thread(
get_directory_size,
item.content_hash.path,
)
size = item.content_hash.size if item.content_hash else 0
all_episode_files[key].append({
@@ -441,9 +445,6 @@ async def _process_movies(
)
tmdb_info = await get_movie_tmdb(first_item.title, first_item.year)
# Yield to event loop so HTTP requests stay responsive
if idx % 20 == 0:
await asyncio.sleep(0)
if tmdb_info and tmdb_info.tmdb_id:
if tmdb_info.tmdb_id not in tmdb_movie_groups:
@@ -694,9 +695,6 @@ async def _process_series(
logger.debug(" [%d/%d] %s", idx, len(series_groups), first_item.title)
tmdb_info = await get_series_tmdb(first_item.title)
# Yield to event loop so HTTP requests stay responsive
if idx % 20 == 0:
await asyncio.sleep(0)
if tmdb_info and tmdb_info.tmdb_id:
if tmdb_info.tmdb_id not in tmdb_groups:
mediahive/hivescan/scanner.py
+79 -29
View File
@@ -14,6 +14,8 @@ from __future__ import annotations
import asyncio
import contextlib
import logging
import os
import threading
import uuid
from collections.abc import Awaitable, Callable
from pathlib import Path
@@ -95,19 +97,16 @@ class RootScanner:
async def stop(self) -> None:
"""Cancel all background tasks."""
for task in (
tasks = [
self._scan_task,
self._showreel_worker_task,
self._rescan_worker_task,
):
]
for task in tasks:
if task and not task.done():
task.cancel()
# Wait briefly for graceful shutdown
for task in (
self._scan_task,
self._showreel_worker_task,
self._rescan_worker_task,
):
# Wait briefly for graceful shutdown to avoid lingering scanner tasks.
for task in tasks:
if task and not task.done():
with contextlib.suppress(TimeoutError, asyncio.CancelledError):
await asyncio.wait_for(task, timeout=2.0)
@@ -150,6 +149,47 @@ class RootScanner:
media_root_str = self.media_root.as_posix()
dirs_visited = 0
def _collect_children(
directory: Path,
stop_event: threading.Event,
) -> tuple[list[Path], list[Path], bool]:
child_dirs: list[Path] = []
child_files: list[Path] = []
is_media_container = False
with os.scandir(directory) as entries:
for entry in entries:
if stop_event.is_set():
return child_dirs, child_files, is_media_container
name = entry.name
if name.startswith("."):
continue
item = Path(entry.path)
try:
is_dir = entry.is_dir(follow_symlinks=False)
except OSError:
continue
if is_dir:
if name.upper() in media_container_dirs:
is_media_container = True
child_dirs.append(item)
else:
child_files.append(item)
return child_dirs, child_files, is_media_container
def _collect_root_children(
directory: Path,
stop_event: threading.Event,
) -> list[Path]:
items: list[Path] = []
with os.scandir(directory) as entries:
for entry in entries:
if stop_event.is_set():
return items
if entry.name.startswith("."):
continue
items.append(Path(entry.path))
return items
async def _report(detail: str) -> None:
await self._send(
Task(
@@ -182,29 +222,33 @@ class RootScanner:
if not await ap.is_dir():
return
child_dirs: list[Path] = []
child_files: list[Path] = []
is_media_container = False
try:
entries = await asyncio.to_thread(lambda: list(ap.iterdir()))
for item_async in entries:
item = Path(item_async)
if item.name.startswith("."):
continue
if self._scanignore and self._scanignore.is_excluded(item):
continue
if await AsyncPath(item).is_dir():
if item.name.upper() in media_container_dirs:
is_media_container = True
child_dirs.append(item)
else:
child_files.append(item)
stop_event = threading.Event()
child_dirs, child_files, is_media_container = await asyncio.to_thread(
_collect_children,
directory,
stop_event,
)
except asyncio.CancelledError:
stop_event.set()
raise
except OSError, PermissionError:
logger.debug("Cannot list directory: %s", directory)
return
# Apply ignore rules after fast scandir classification.
if self._scanignore:
child_dirs = [
item
for item in child_dirs
if not self._scanignore.is_excluded(item)
]
child_files = [
item
for item in child_files
if not self._scanignore.is_excluded(item)
]
if is_media_container:
relpath = make_relative_path(str(directory), media_root_str)
try:
@@ -230,7 +274,6 @@ class RootScanner:
logger.info("Scanning: %s (%d found so far)", rel, len(downloads))
for child in child_dirs:
await _walk(child)
await asyncio.sleep(0)
for child_file in child_files:
if child_file.suffix.lower() in video_extensions:
relpath = make_relative_path(str(child_file), media_root_str)
@@ -261,9 +304,16 @@ class RootScanner:
logger.info("Starting filesystem discovery at %s", self.media_root)
await _report(f"Scanning: {self.media_root}")
root_ap = AsyncPath(self.media_root)
try:
root_children = await asyncio.to_thread(lambda: list(root_ap.iterdir()))
stop_event = threading.Event()
root_children = await asyncio.to_thread(
_collect_root_children,
self.media_root,
stop_event,
)
except asyncio.CancelledError:
stop_event.set()
raise
except OSError, PermissionError:
logger.exception("Cannot list media root: %s", self.media_root)
return downloads
mediahive/hivescan/scanning.py
+157 -40
View File
@@ -3,6 +3,8 @@
import asyncio
import glob
import operator
import os
import threading
from collections import defaultdict
from pathlib import Path
@@ -32,6 +34,51 @@ _playable_file_cache: dict[str, str | None] = {}
_bluray_probe_file_cache: dict[str, str | None] = {}
def _scandir_split(
directory: Path,
stop_event: threading.Event,
) -> tuple[list[Path], list[Path]]:
"""Return child directories and files, checking stop_event each iteration."""
child_dirs: list[Path] = []
child_files: list[Path] = []
with os.scandir(directory) as entries:
for entry in entries:
if stop_event.is_set():
return child_dirs, child_files
try:
is_dir = entry.is_dir(follow_symlinks=False)
except OSError:
continue
p = Path(entry.path)
if is_dir:
child_dirs.append(p)
else:
child_files.append(p)
return child_dirs, child_files
def _scandir_files_with_suffix(
directory: Path,
suffixes: set[str],
stop_event: threading.Event,
) -> list[Path]:
"""Return files in directory with a matching suffix, cancellable via stop_event."""
files: list[Path] = []
with os.scandir(directory) as entries:
for entry in entries:
if stop_event.is_set():
return files
try:
if entry.is_dir(follow_symlinks=False):
continue
except OSError:
continue
p = Path(entry.path)
if p.suffix.lower() in suffixes:
files.append(p)
return files
async def scan_downloads(base_pattern: str) -> list[ParsedContent]:
"""Scan download directories matching the pattern.
@@ -107,22 +154,37 @@ async def find_episode_files(
_episode_files_cache[cache_key] = episodes
return episodes
stack = [path]
while stack:
current = stack.pop()
stop_event = threading.Event()
try:
for f in ap.rglob("*"):
af = AsyncPath(f)
if await af.is_file() and Path(f).suffix.lower() in VIDEO_EXTENSIONS:
if "sample" in Path(f).name.lower():
child_dirs, child_files = await asyncio.to_thread(
_scandir_split,
current,
stop_event,
)
except asyncio.CancelledError:
stop_event.set()
raise
except OSError, PermissionError:
continue
ep_info = parse_episode_from_filename(Path(f).name)
stack.extend(child_dirs)
for f in child_files:
if f.suffix.lower() not in VIDEO_EXTENSIONS:
continue
if "sample" in f.name.lower():
continue
af = AsyncPath(f)
try:
ep_info = parse_episode_from_filename(f.name)
if ep_info:
if ep_info not in episodes:
episodes[ep_info] = []
episodes[ep_info].append((
Path(f).as_posix(),
(await af.stat()).st_size,
))
episodes[ep_info].append((f.as_posix(), (await af.stat()).st_size))
except OSError, PermissionError:
pass
continue
_episode_files_cache[cache_key] = episodes
return episodes
@@ -174,10 +236,20 @@ async def find_playable_file(path: Path) -> str | None:
# Check nested Blu-ray structure (e.g., MovieName/DISC1/BDMV/)
try:
entries = await asyncio.to_thread(lambda: list(ap.iterdir()))
for subdir in entries:
if await AsyncPath(subdir).is_dir():
nested_bdmv_dir = Path(subdir) / "BDMV"
stop_event = threading.Event()
child_dirs, _ = await asyncio.to_thread(
_scandir_split,
path,
stop_event,
)
except asyncio.CancelledError:
stop_event.set()
raise
except OSError, PermissionError:
child_dirs = []
for subdir in child_dirs:
nested_bdmv_dir = subdir / "BDMV"
nested_movieobject = nested_bdmv_dir / "MovieObject.bdmv"
nested_index = nested_bdmv_dir / "index.bdmv"
@@ -191,27 +263,43 @@ async def find_playable_file(path: Path) -> str | None:
_playable_file_cache[cache_key] = result
return result
nested_video_ts_dir = Path(subdir) / "VIDEO_TS"
nested_video_ts_dir = subdir / "VIDEO_TS"
nested_video_ts_ifo = nested_video_ts_dir / "VIDEO_TS.IFO"
if await AsyncPath(nested_video_ts_ifo).exists():
result = nested_video_ts_ifo.as_posix()
_playable_file_cache[cache_key] = result
return result
except OSError, PermissionError:
pass
# Find largest video file
video_files = []
stack = [path]
while stack:
current = stack.pop()
stop_event = threading.Event()
try:
for f in ap.rglob("*"):
af = AsyncPath(f)
if await af.is_file() and Path(f).suffix.lower() in VIDEO_EXTENSIONS:
if "sample" in Path(f).name.lower():
continue
video_files.append((Path(f).as_posix(), (await af.stat()).st_size))
child_dirs, child_files = await asyncio.to_thread(
_scandir_split,
current,
stop_event,
)
except asyncio.CancelledError:
stop_event.set()
raise
except OSError, PermissionError:
pass
continue
stack.extend(child_dirs)
for f in child_files:
if f.suffix.lower() not in VIDEO_EXTENSIONS:
continue
if "sample" in f.name.lower():
continue
af = AsyncPath(f)
try:
video_files.append((f.as_posix(), (await af.stat()).st_size))
except OSError, PermissionError:
continue
if not video_files:
_playable_file_cache[cache_key] = None
@@ -254,19 +342,32 @@ async def find_metadata_probe_file(playable_path: str | None) -> str | None:
# Group VOBs by title set (VTS_XX_Y.VOB)
title_sets: dict[str, list[tuple[str, int]]] = defaultdict(list)
try:
for f in AsyncPath(video_ts_dir).glob("*.vob"):
af = AsyncPath(f)
if not await af.is_file():
continue
name = Path(f).name.upper()
if name.startswith("VTS_") and len(name) >= 10:
ts_num = name[4:6]
size = (await af.stat()).st_size
title_sets[ts_num].append((Path(f).as_posix(), size))
stop_event = threading.Event()
vob_files = await asyncio.to_thread(
_scandir_files_with_suffix,
video_ts_dir,
{".vob"},
stop_event,
)
except asyncio.CancelledError:
stop_event.set()
raise
except OSError, PermissionError:
_bluray_probe_file_cache[cache_key] = None
return None
for f in vob_files:
af = AsyncPath(f)
name = f.name.upper()
if not name.startswith("VTS_") or len(name) < 10:
continue
try:
size = (await af.stat()).st_size
except OSError, PermissionError:
continue
ts_num = name[4:6]
title_sets[ts_num].append((f.as_posix(), size))
if not title_sets:
_bluray_probe_file_cache[cache_key] = None
return None
@@ -299,15 +400,31 @@ async def find_metadata_probe_file(playable_path: str | None) -> str | None:
return None
candidates: list[tuple[str, int]] = []
stack = [stream_dir]
while stack:
current = stack.pop()
stop_event = threading.Event()
try:
for f in ap_stream.rglob("*.m2ts"):
af = AsyncPath(f)
if not await af.is_file():
continue
candidates.append((Path(f).as_posix(), (await af.stat()).st_size))
child_dirs, child_files = await asyncio.to_thread(
_scandir_split,
current,
stop_event,
)
except asyncio.CancelledError:
stop_event.set()
raise
except OSError, PermissionError:
_bluray_probe_file_cache[cache_key] = None
return None
continue
stack.extend(child_dirs)
for f in child_files:
if f.suffix.lower() != ".m2ts":
continue
af = AsyncPath(f)
try:
candidates.append((f.as_posix(), (await af.stat()).st_size))
except OSError, PermissionError:
continue
if not candidates:
_bluray_probe_file_cache[cache_key] = None
mediahive/hivescan/showreel.py
+3 -3
View File
@@ -78,7 +78,7 @@ async def _run_ffmpeg(
await _kill_proc(proc)
try:
stdout, stderr = await proc.communicate()
except OSError, asyncio.SubprocessError:
except OSError, subprocess.SubprocessError:
stdout, stderr = b"", b""
logger.exception(
"ffmpeg command timed out. cmd=%s stderr=%s",
@@ -110,7 +110,7 @@ async def _run_ffmpeg(
except asyncio.CancelledError:
await _kill_proc(proc)
raise
except OSError, asyncio.SubprocessError:
except OSError, subprocess.SubprocessError:
logger.exception("Unexpected error running ffmpeg command: %s", shlex.join(cmd))
return None
@@ -120,7 +120,7 @@ async def _kill_proc(proc: asyncio.subprocess.Process | None) -> None:
if proc is not None and proc.returncode is None:
proc.kill()
with contextlib.suppress(Exception):
await asyncio.wait_for(proc.wait(), timeout=2)
await asyncio.wait_for(proc.wait(), timeout=0.5)
# Showreel timestamp positions in seconds (5, 10, 15, 20, 25 minutes)
mediahive/hivescan/tmdb_client.py
+8
View File
@@ -58,6 +58,14 @@ def _get_http_client() -> httpx.AsyncClient:
return _http_client
async def close_http_client() -> None:
"""Close the persistent TMDb HTTP client if it was created."""
global _http_client
if _http_client is not None:
await _http_client.aclose()
_http_client = None
# Sentinel value to distinguish "cached None" from "not in cache"
_NOT_FOUND = object()
mediahive/hivescan/utils.py
+32 -14
View File
@@ -1,5 +1,6 @@
"""Utility functions for paths, sizes, and timestamps."""
import os
import time
from pathlib import Path
@@ -113,28 +114,45 @@ async def get_added_timestamp(path: Path) -> int | None:
return int(atime)
async def get_directory_size(path: Path) -> int:
"""Calculate total size of a directory recursively."""
ap = AsyncPath(path)
total = 0
def get_directory_size(path: Path) -> int:
"""Calculate total size using scandir recursion in a sync worker."""
try:
if await ap.is_file():
return (await ap.stat()).st_size
for item in ap.rglob("*"):
if await AsyncPath(item).is_file():
total += (await AsyncPath(item).stat()).st_size
if path.is_file():
return path.stat().st_size
except OSError, PermissionError:
pass
return 0
total = 0
stack = [path]
while stack:
current = stack.pop()
try:
with os.scandir(current) as entries:
for entry in entries:
try:
if entry.is_dir(follow_symlinks=False):
stack.append(Path(entry.path))
continue
except OSError:
continue
try:
total += entry.stat(follow_symlinks=False).st_size
except OSError:
continue
except OSError, PermissionError:
continue
return total
def format_size(size_bytes: int) -> str:
"""Format size in human-readable format."""
size = float(size_bytes)
for unit in ["B", "KB", "MB", "GB", "TB"]:
if size_bytes < 1024:
return f"{size_bytes:.2f} {unit}"
size_bytes /= 1024
return f"{size_bytes:.2f} PB"
if size < 1024:
return f"{size:.2f} {unit}"
size /= 1024
return f"{size:.2f} PB"
async def find_common_root(paths: list[Path]) -> Path | None:
mediahive/index_store.py
+29 -16
View File
@@ -9,7 +9,6 @@ debounced background task.
import asyncio
import contextlib
import logging
import os
from datetime import datetime
from pathlib import Path
@@ -170,17 +169,22 @@ class IndexStore:
async def _write_snapshot(self) -> None:
"""Write current index to disk (called from debounce task)."""
snapshot = self._build_snapshot()
await AsyncPath(self.snapshot_path.parent).mkdir(parents=True, exist_ok=True)
tmp = self.snapshot_path.with_suffix(".tmp")
await AsyncPath(tmp).write_bytes(
msgspec.json.format(msgspec.json.encode(snapshot), indent=2)
)
# os.replace is atomic and overwrites on all platforms (unlike rename on Windows)
await asyncio.to_thread(os.replace, tmp, self.snapshot_path)
# Copy values on the event loop thread, then do full snapshot build + disk I/O
# in a worker thread to keep the loop responsive.
movies = list(self.movies.values())
series = list(self.series.values())
await asyncio.to_thread(self._write_snapshot_sync, movies, series)
logger.debug("Snapshot written to %s", self.snapshot_path)
def _write_snapshot_sync(self, movies: list[Movie], series: list[Series]) -> None:
"""Build and write snapshot synchronously in a worker thread."""
snapshot = self._build_snapshot_from_lists(movies, series)
self.snapshot_path.parent.mkdir(parents=True, exist_ok=True)
tmp = self.snapshot_path.with_suffix(".tmp")
tmp.write_bytes(msgspec.json.format(msgspec.json.encode(snapshot), indent=2))
# Path.replace is atomic and overwrites on all platforms.
tmp.replace(self.snapshot_path)
def _schedule_snapshot(self) -> None:
"""Schedule a debounced snapshot write."""
self._snapshot_dirty = True
@@ -300,12 +304,14 @@ class IndexStore:
# Read helpers
# ------------------------------------------------------------------
def _build_snapshot(self) -> IndexSnapshot:
"""Build a sorted IndexSnapshot with computed stats."""
movies_list = sorted(
self.movies.values(), key=lambda x: (x.title.lower(), x.year or 0)
)
series_list = sorted(self.series.values(), key=lambda x: x.title.lower())
def _build_snapshot_from_lists(
self,
movies: list[Movie],
series: list[Series],
) -> IndexSnapshot:
"""Build a sorted IndexSnapshot with computed stats from list copies."""
movies_list = sorted(movies, key=lambda x: (x.title.lower(), x.year or 0))
series_list = sorted(series, key=lambda x: x.title.lower())
total_movie_versions = sum(len(m.torrents) for m in movies_list)
total_series_episodes = sum(
@@ -325,6 +331,13 @@ class IndexStore:
series=series_list,
)
def _build_snapshot(self) -> IndexSnapshot:
"""Build a sorted IndexSnapshot with computed stats."""
return self._build_snapshot_from_lists(
list(self.movies.values()),
list(self.series.values()),
)
def get_full_index(self) -> IndexSnapshot:
"""Return the full index as an IndexSnapshot."""
return self._build_snapshot()
mediahive/server.py
+7
View File
@@ -30,7 +30,9 @@ from fastapi_vue import Frontend
from mediahive.__main__ import DEVMODE
from mediahive.config import load_config
from mediahive.hivescan.images import close_image_client
from mediahive.hivescan.scanner import RootScanner
from mediahive.hivescan.tmdb_client import close_http_client
from mediahive.models.protocol import (
MsgspecResponse,
OpenFolderRequest,
@@ -369,6 +371,11 @@ async def lifespan(_app: FastAPI):
await supervisor.shutdown()
with suppress(Exception):
await close_http_client()
with suppress(Exception):
await close_image_client()
app = FastAPI(title="MediaHive Server", lifespan=lifespan, debug=DEVMODE)
mediahive/winmain.py
+28 -2
View File
@@ -6,6 +6,7 @@ Or from PyInstaller: MediaHive.exe [media_folder]
import argparse
import asyncio
import contextlib
import ctypes
import html
import json
@@ -839,6 +840,7 @@ def winmain() -> None:
port=backend_port,
loop="asyncio",
log_level="warning",
timeout_graceful_shutdown=0,
)
server = uvicorn.Server(config)
backend_thread = threading.Thread(
@@ -898,14 +900,38 @@ def winmain() -> None:
name="mediahive-initial-roots-activation",
).start()
def on_closing() -> None:
# Begin backend shutdown as soon as the window starts closing so that
# by the time webview.start() returns the backend is already done.
server.should_exit = True
window.events.closing += on_closing
webview.start(func=on_shown, icon=_icon_path(), **_webview_start_kwargs())
# Ensure backend shutdown has been requested (in case closing event
# was not fired or we are on a platform that does not support it).
server.should_exit = True
backend_thread.join(timeout=2)
poll_stop.set()
if poll_thread is not None:
poll_thread.join(timeout=1)
server.should_exit = True
backend_thread.join(timeout=10)
# Close log file handles so mediahive.log is not left locked.
if getattr(sys, "frozen", False):
logging.shutdown()
for handler in logging.root.handlers[:]:
handler.close()
logging.root.removeHandler(handler)
if sys.stdout is not sys.__stdout__:
with contextlib.suppress(Exception):
sys.stdout.close()
if sys.stderr is not sys.__stderr__:
with contextlib.suppress(Exception):
sys.stderr.close()
sys.stdout = sys.__stdout__
sys.stderr = sys.__stderr__
if __name__ == "__main__":