From 626fb9a0ae7ae9e8fc452bcb425098788db65eef Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Sun, 31 May 2026 01:43:56 +0000 Subject: [PATCH] Unify playback-state read/write via /api/meta/playback-state --- frontend/src/App.vue | 8 +- frontend/src/api.ts | 10 +- frontend/src/components/MediaDetail.vue | 5 +- mediahive/models/protocol.py | 8 + mediahive/server.py | 291 ++++++++++++++++++++++-- mediahive/winmain.py | 196 +++++++++++----- 6 files changed, 435 insertions(+), 83 deletions(-) diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 464f6e6..9ea4503 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -207,7 +207,6 @@ import { openFolder, isMpcBeReachable, fetchResumePositions, - normalizeMediaPath, getPlayerStatus, } from "./api" import { useSettings } from "./composables/useSettings" @@ -547,10 +546,9 @@ async function refreshPlayerStatus() { } } -function hasResumePosition(filePath: string | null) { - if (!filePath) return false - const normalizedPath = normalizeMediaPath(filePath) - return Number(resumePositions.value[normalizedPath] || 0) > 0 +function hasResumePosition(mediaId: string | null) { + if (!mediaId) return false + return Number(resumePositions.value[mediaId] || 0) > 0 } function startMpcBePolling() { diff --git a/frontend/src/api.ts b/frontend/src/api.ts index b24a57e..52c7d0a 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -126,7 +126,15 @@ export async function fetchResumePositions(): Promise> { if (!positions || typeof positions !== "object") { return {} } - return positions as Record + const normalized: Record = {} + for (const [slug, value] of Object.entries(positions as Record)) { + if (!value || typeof value !== "object") continue + const pos = (value as { pos?: unknown }).pos + if (typeof pos === "number" && Number.isFinite(pos) && pos > 0) { + normalized[slug] = pos + } + } + return normalized } catch { return {} } diff --git a/frontend/src/components/MediaDetail.vue b/frontend/src/components/MediaDetail.vue index 443db5c..7a46c09 100644 --- a/frontend/src/components/MediaDetail.vue +++ b/frontend/src/components/MediaDetail.vue @@ -256,7 +256,7 @@ const props = defineProps<{ item: MediaItem allMovies: MovieUi[] focusEpisode?: { seasonNumber: number; episodeNumber: number } | null - hasResumePosition: (filePath: string | null) => boolean + hasResumePosition: (mediaId: string | null) => boolean getRootName: (rootId: string | null | undefined) => string | null }>() const emit = defineEmits<{ @@ -828,7 +828,8 @@ function closeVersionActionMenu() { } function getPlayLabel(filePath: string | null): string { - return props.hasResumePosition(filePath) ? "Continue" : "Play" + if (!filePath || props.item.type !== "movies") return "Play" + return props.hasResumePosition(props.item.id) ? "Continue" : "Play" } function handlePlayVersion(filePath: string | null) { diff --git a/mediahive/models/protocol.py b/mediahive/models/protocol.py index e4f7b3e..3d9fb8d 100644 --- a/mediahive/models/protocol.py +++ b/mediahive/models/protocol.py @@ -117,6 +117,14 @@ class RootsRequest(msgspec.Struct): roots: dict[str, str] +class PlaybackStateUpdateRequest(msgspec.Struct): + """POST /api/meta/playback-state body.""" + + root_id: str + file_path: str + pos: int | None = None + + class RootEntryResponse(msgspec.Struct): """Single root entry in responses.""" diff --git a/mediahive/server.py b/mediahive/server.py index 17c4c52..0f703cb 100644 --- a/mediahive/server.py +++ b/mediahive/server.py @@ -16,10 +16,13 @@ import os import re import subprocess import sys +import threading import time import urllib.error import urllib.request from contextlib import asynccontextmanager, suppress +from dataclasses import dataclass, field +from datetime import datetime from pathlib import Path import aiofiles @@ -43,6 +46,7 @@ from mediahive.hivescan.tmdb_client import close_http_client from mediahive.models.events import Remove, Task, Upsert from mediahive.models.protocol import ( OpenFolderRequest, + PlaybackStateUpdateRequest, PlayMediaRequest, RootsRequest, WsInit, @@ -83,6 +87,218 @@ if sys.platform == "win32": from ctypes import wintypes +@dataclass +class _PlaybackEntry: + """Single resume position entry with timestamp.""" + + pos: int + ts: datetime + + def to_dict(self) -> dict: + return { + "pos": self.pos, + "ts": self.ts, + } + + @staticmethod + def from_dict(data: dict) -> _PlaybackEntry | None: + if not isinstance(data, dict): + return None + pos = data.get("pos") + ts = data.get("ts") + if not isinstance(pos, int) or pos < 0: + return None + if isinstance(ts, str): + try: + ts = datetime.fromisoformat(ts) + except ValueError, TypeError: + return None + elif isinstance(ts, datetime): + pass + else: + return None + return _PlaybackEntry(pos=pos, ts=ts) + + +@dataclass +class _PlaybackRootSnapshot: + file_path: Path + signature: tuple[bool, int, int] | None = None + entries: dict[str, _PlaybackEntry] = field(default_factory=dict) + + +class PlaybackStateCache: + """Background cache for merged playback-state across all active roots. + + Stores resume positions by movie slug with timestamps. When merging + across roots, picks the most recent entry for each slug. + """ + + def __init__(self, poll_interval: float = 60.0) -> None: + self._poll_interval = poll_interval + self._roots: dict[str, _PlaybackRootSnapshot] = {} + self._merged_entries: dict[str, _PlaybackEntry] = {} + self._lock = threading.RLock() + self._stop = threading.Event() + self._thread: threading.Thread | None = None + + def start(self) -> None: + if self._thread is not None and self._thread.is_alive(): + return + self._stop.clear() + self._thread = threading.Thread( + target=self._run, + name="mediahive-playback-state-cache", + daemon=True, + ) + self._thread.start() + + def stop(self) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=max(2.0, self._poll_interval + 1.0)) + self._thread = None + + def get_merged_entries(self) -> dict[str, _PlaybackEntry]: + """Return merged resume entries keyed by slug (most recent wins).""" + with self._lock: + return { + slug: _PlaybackEntry(e.pos, e.ts) + for slug, e in self._merged_entries.items() + } + + def update_resume_position( + self, + root_id: str, + root_path: Path, + slug: str, + pos: int | None, + ) -> None: + """Read-modify-write one root file and refresh the in-memory cache immediately.""" + file_path = root_path / ".mediahive" / "playback-state.json" + entries = self._read_resume_entries(file_path) + + if pos is None: + entries.pop(slug, None) + else: + entries[slug] = _PlaybackEntry(pos=pos, ts=datetime.now()) + + self._write_resume_entries(file_path, entries) + + snapshot = _PlaybackRootSnapshot( + file_path=file_path, + signature=self._signature(file_path), + entries=entries, + ) + with self._lock: + self._roots[root_id] = snapshot + self._merged_entries = self._build_merged_entries(self._roots) + + def _run(self) -> None: + while not self._stop.is_set(): + try: + self._refresh_once() + except Exception: + logger.exception("Playback-state cache refresh failed") + if self._stop.wait(self._poll_interval): + break + + def _refresh_once(self) -> None: + contexts = supervisor.all_contexts() + + with self._lock: + previous = self._roots + + next_roots: dict[str, _PlaybackRootSnapshot] = {} + merged: dict[str, _PlaybackEntry] = {} + + for root_id, ctx in contexts.items(): + file_path = ctx.root_path / ".mediahive" / "playback-state.json" + snapshot = previous.get(root_id) + if snapshot is None or snapshot.file_path != file_path: + snapshot = _PlaybackRootSnapshot(file_path=file_path) + + signature = self._signature(file_path) + if signature != snapshot.signature: + snapshot.signature = signature + snapshot.entries = self._read_resume_entries(file_path) + + next_roots[root_id] = snapshot + + # Merge: for each slug, keep the entry with the most recent timestamp + for slug, entry in snapshot.entries.items(): + existing = merged.get(slug) + if existing is None or entry.ts > existing.ts: + merged[slug] = entry + + with self._lock: + self._roots = next_roots + self._merged_entries = merged + + @staticmethod + def _signature(path: Path) -> tuple[bool, int, int]: + try: + stat = path.stat() + return (True, stat.st_mtime_ns, stat.st_size) + except OSError: + return (False, 0, 0) + + @staticmethod + def _read_resume_entries(path: Path) -> dict[str, _PlaybackEntry]: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except OSError, json.JSONDecodeError: + return {} + + if not isinstance(raw, dict): + return {} + + positions = raw.get("resume_positions") + if not isinstance(positions, dict): + return {} + + entries: dict[str, _PlaybackEntry] = {} + for slug, data in positions.items(): + entry = _PlaybackEntry.from_dict(data) + if entry is not None: + entries[str(slug)] = entry + + return entries + + @staticmethod + def _write_resume_entries(path: Path, entries: dict[str, _PlaybackEntry]) -> None: + data = { + "resume_positions": { + slug: entry.to_dict() for slug, entry in sorted(entries.items()) + } + } + try: + path.parent.mkdir(parents=True, exist_ok=True) + tmp_path = path.with_suffix(f"{path.suffix}.tmp") + tmp_path.write_text( + json.dumps(data, indent=2, sort_keys=True, default=str), + encoding="utf-8", + ) + tmp_path.replace(path) + except OSError: + logger.exception("Failed to write playback-state: %s", path) + + @staticmethod + def _build_merged_entries( + roots: dict[str, _PlaybackRootSnapshot], + ) -> dict[str, _PlaybackEntry]: + merged: dict[str, _PlaybackEntry] = {} + for snapshot in roots.values(): + for slug, entry in snapshot.entries.items(): + existing = merged.get(slug) + if existing is None or entry.ts > existing.ts: + merged[slug] = entry + return merged + + +playback_state_cache = PlaybackStateCache() + + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- @@ -95,6 +311,37 @@ def _get_context(root_id: str): return ctx +def _normalize_media_path_value(path: str) -> str: + return path.replace("\\", "/").lstrip("/") + + +def _expand_torrent_playable_path(file_key: str, playable_file: str | None) -> str: + if not playable_file: + return file_key + if playable_file.startswith("concat:") or "://" in playable_file: + return playable_file + if playable_file.startswith(f"{file_key}/"): + return playable_file + if playable_file.startswith("/"): + return playable_file.lstrip("/") + return f"{file_key}/{playable_file}" + + +def _resolve_movie_slug_for_file_path(ctx, file_path: str) -> str | None: + target = _normalize_media_path_value(file_path) + for movie_id, movie in ctx.store.movies.items(): + for file_key, torrent in movie.files.items(): + normalized_key = _normalize_media_path_value(file_key) + if normalized_key == target: + return movie_id + playable_path = _expand_torrent_playable_path( + file_key, torrent.playable_file + ) + if _normalize_media_path_value(playable_path) == target: + return movie_id + return None + + def _load_root_metadata(root_path: Path, meta_key: str): """Load allowed per-root metadata values from .mediahive.""" key = meta_key.strip().lower().strip("/") @@ -437,6 +684,7 @@ async def _activate_all_roots() -> None: @asynccontextmanager async def lifespan(_app: FastAPI): await frontend.load() + playback_state_cache.start() # Defer root activation to a background task so the server starts # immediately and macOS permission dialogs do not block startup. @@ -448,6 +696,8 @@ async def lifespan(_app: FastAPI): try: yield finally: + playback_state_cache.stop() + activation_task.cancel() with suppress(asyncio.CancelledError): await activation_task @@ -723,26 +973,31 @@ async def root_metadata(root_id: str, meta_key: str): @app.get("/api/meta/playback-state") async def merged_playback_state(): - """Return merged playback-state resume positions from all active roots.""" - merged: dict[str, float] = {} - for ctx in supervisor.all_contexts().values(): - try: - data = _load_root_metadata(ctx.root_path, "playback-state") - except HTTPException as exc: - if exc.status_code == 404: - continue - raise + """Return merged playback-state resume positions from in-memory cache. - if not isinstance(data, dict): - continue - positions = data.get("resume_positions") - if not isinstance(positions, dict): - continue - for key, value in positions.items(): - if isinstance(value, int | float): - merged[str(key)] = float(value) + Merges across all roots, preferring the most recent timestamp for each slug. + Format: {"key": "playback-state", "data": {"resume_positions": {slug: {pos, ts}}}} + """ + entries = playback_state_cache.get_merged_entries() + positions = {slug: entry.to_dict() for slug, entry in entries.items()} + return {"key": "playback-state", "data": {"resume_positions": positions}} - return {"key": "playback-state", "data": {"resume_positions": merged}} + +@app.post("/api/meta/playback-state") +async def write_playback_state(request: Request): + """Update one playback-state entry via backend-managed read-modify-write.""" + req = msgspec.json.decode(await request.body(), type=PlaybackStateUpdateRequest) + ctx = _get_context(req.root_id) + slug = _resolve_movie_slug_for_file_path(ctx, req.file_path) + if slug is None: + raise HTTPException( + status_code=404, + detail=f"Movie not found for file path: {req.file_path}", + ) + + pos = None if req.pos is None or req.pos <= 0 else int(req.pos) + playback_state_cache.update_resume_position(req.root_id, ctx.root_path, slug, pos) + return {"status": "ok", "slug": slug, "pos": pos} # --- MPC-BE / Player status --- diff --git a/mediahive/winmain.py b/mediahive/winmain.py index b07c892..d877ec5 100644 --- a/mediahive/winmain.py +++ b/mediahive/winmain.py @@ -120,59 +120,127 @@ _DURATION_RE = re.compile(r'

(\d+)

') def _default_playback_state() -> dict[str, object]: return { "current": None, - "resume_positions": {}, } -def _load_playback_state(path: Path) -> dict[str, object]: +def _normalize_media_path(path: str) -> str: + return path.replace("\\", "/").lstrip("/") + + +def _expand_playable_file(file_key: str, playable_file: str | None) -> str: + if not playable_file: + return file_key + if playable_file.startswith("concat:") or "://" in playable_file: + return playable_file + if playable_file.startswith(f"{file_key}/"): + return playable_file + if playable_file.startswith("/"): + return playable_file.lstrip("/") + return f"{file_key}/{playable_file}" + + +def _fetch_resume_positions(backend_url: str) -> dict[str, int]: + req = urllib.request.Request( + url=f"{backend_url}/api/meta/playback-state", + method="GET", + ) try: - raw = json.loads(path.read_text(encoding="utf-8")) + with urllib.request.urlopen(req, timeout=2) as resp: + raw = json.loads(resp.read().decode("utf-8")) + except OSError, TimeoutError, urllib.error.URLError, json.JSONDecodeError: + return {} + + data = raw.get("data") if isinstance(raw, dict) else None + positions = data.get("resume_positions") if isinstance(data, dict) else None + if not isinstance(positions, dict): + return {} + + cleaned: dict[str, int] = {} + for slug, value in positions.items(): + if not isinstance(slug, str) or not isinstance(value, dict): + continue + pos = value.get("pos") + if isinstance(pos, int) and pos > 0: + cleaned[slug] = pos * 1000 + return cleaned + + +def _post_resume_position( + backend_url: str, + root_id: str, + file_path: str, + pos: int | None, +) -> bool: + body = json.dumps({ + "root_id": root_id, + "file_path": file_path, + "pos": pos, + }).encode("utf-8") + req = urllib.request.Request( + url=f"{backend_url}/api/meta/playback-state", + data=body, + method="POST", + headers={"Content-Type": "application/json"}, + ) + try: + with urllib.request.urlopen(req, timeout=2): + return True + except OSError, TimeoutError, urllib.error.URLError: + logger.warning("Failed to post playback-state update for %s", file_path) + return False + + +def _load_movie_slug_map(index_path: Path) -> dict[str, str]: + try: + raw = json.loads(index_path.read_text(encoding="utf-8")) except OSError, TypeError, json.JSONDecodeError: - return _default_playback_state() + return {} - if not isinstance(raw, dict): - return _default_playback_state() + movies = raw.get("movies") if isinstance(raw, dict) else None + if not isinstance(movies, dict): + return {} - current = raw.get("current") - resume_positions = raw.get("resume_positions") - normalized: dict[str, object] = { - "current": current if isinstance(current, dict) else None, - "resume_positions": {}, - } - - if isinstance(resume_positions, dict): - cleaned_positions: dict[str, int] = {} - for key, value in resume_positions.items(): - if isinstance(key, str) and isinstance(value, (int, float)): - cleaned_positions[key] = max(0, int(value)) - normalized["resume_positions"] = cleaned_positions - - return normalized - - -def _save_playback_state(path: Path, state: dict[str, object]) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - tmp_path = path.with_suffix(f"{path.suffix}.tmp") - tmp_path.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8") - tmp_path.replace(path) + mapping: dict[str, str] = {} + for movie_id, movie in movies.items(): + if not isinstance(movie_id, str) or not isinstance(movie, dict): + continue + files = movie.get("files") + if not isinstance(files, dict): + continue + for file_key, torrent in files.items(): + if not isinstance(file_key, str): + continue + normalized_key = _normalize_media_path(file_key) + mapping[normalized_key] = movie_id + playable_file = ( + torrent.get("playable_file") if isinstance(torrent, dict) else None + ) + expanded = _expand_playable_file( + file_key, playable_file if isinstance(playable_file, str) else None + ) + mapping[_normalize_media_path(expanded)] = movie_id + return mapping def _media_key_for_filepath( - filepath: str, roots: list[Path] -) -> tuple[str, Path] | None: - """Resolve a filepath to a (relative_key, matched_root) tuple.""" - for root in roots: + filepath: str, roots: dict[str, Path] +) -> tuple[str | None, str, str] | None: + """Resolve a filepath to a (movie_slug, root_id, relative_key) tuple.""" + for root_id, root in roots.items(): try: relative = Path(filepath).resolve().relative_to(root.resolve()) - return relative.as_posix(), root + relative_key = relative.as_posix() + index_path = root / ".mediahive" / "index.json" + movie_slug = _load_movie_slug_map(index_path).get( + _normalize_media_path(relative_key) + ) + return movie_slug, root_id, relative_key except OSError, RuntimeError, ValueError: continue return None def _should_clear_resume(position_ms: int, duration_ms: int) -> bool: - if position_ms <= MPC_BE_RESUME_CLEAR_MARGIN_MS: - return True if duration_ms <= 0: return False return duration_ms - position_ms <= MPC_BE_RESUME_CLEAR_MARGIN_MS @@ -248,7 +316,7 @@ def _mpcbe_fetch_status() -> tuple[str, int, int, int] | None: def _start_gamepad_remote( - stop_event: threading.Event, roots: list[Path] + stop_event: threading.Event, roots: dict[str, Path], backend_url: str ) -> threading.Thread: """Start background XInput polling and send mapped commands to MPC-BE.""" get_state = _load_xinput_get_state() @@ -278,18 +346,11 @@ def _start_gamepad_remote( status_updated_at = 0.0 status_miss_count = 0 - # Use the first root's playback state path as primary - primary_root = roots[0] if roots else Path.cwd() - playback_state_path = primary_root / ".mediahive" / "playback-state.json" - playback_state = _load_playback_state(playback_state_path) - resume_positions = playback_state["resume_positions"] - if not isinstance(resume_positions, dict): - resume_positions = {} - playback_state["resume_positions"] = resume_positions - if playback_state.get("current") is not None: - playback_state["current"] = None - _save_playback_state(playback_state_path, playback_state) + playback_state = _default_playback_state() + resume_positions = _fetch_resume_positions(backend_url) tracked_media_key: str | None = None + tracked_root_id: str | None = None + tracked_relative_path = "" tracked_filepath = "" resume_applied_for_key: str | None = None last_playback_state_flush_at = 0.0 @@ -302,12 +363,11 @@ def _start_gamepad_remote( request_pool.submit(_seek_mpcbe_to_position, position_ms) ) - def flush_playback_state() -> None: - _save_playback_state(playback_state_path, playback_state) - def clear_tracked_current(*, clear_resume_applied: bool) -> None: nonlocal \ tracked_media_key, \ + tracked_root_id, \ + tracked_relative_path, \ tracked_filepath, \ last_playback_state_flush_at, \ resume_applied_for_key @@ -316,38 +376,56 @@ def _start_gamepad_remote( resume_applied_for_key = None return tracked_media_key = None + tracked_root_id = None + tracked_relative_path = "" tracked_filepath = "" playback_state["current"] = None last_playback_state_flush_at = 0.0 if clear_resume_applied: resume_applied_for_key = None - flush_playback_state() def finalize_tracked_current() -> None: nonlocal \ tracked_media_key, \ + tracked_root_id, \ + tracked_relative_path, \ tracked_filepath, \ resume_applied_for_key, \ last_playback_state_flush_at if tracked_media_key is None: if playback_state.get("current") is not None: playback_state["current"] = None - flush_playback_state() return position_ms = player_position_ms or 0 duration_ms = player_duration_ms or 0 if _should_clear_resume(position_ms, duration_ms): resume_positions.pop(tracked_media_key, None) + if tracked_root_id and tracked_relative_path: + _post_resume_position( + backend_url, tracked_root_id, tracked_relative_path, None + ) + elif position_ms <= MPC_BE_RESUME_CLEAR_MARGIN_MS: + # Ignore brief starts; keep the previous saved resume position. + pass else: + position_seconds = max(0, position_ms // 1000) resume_positions[tracked_media_key] = position_ms + if tracked_root_id and tracked_relative_path: + _post_resume_position( + backend_url, + tracked_root_id, + tracked_relative_path, + position_seconds, + ) tracked_media_key = None + tracked_root_id = None + tracked_relative_path = "" tracked_filepath = "" playback_state["current"] = None resume_applied_for_key = None last_playback_state_flush_at = 0.0 - flush_playback_state() def persist_tracked_current(now: float, *, force: bool = False) -> None: nonlocal last_playback_state_flush_at @@ -367,7 +445,6 @@ def _start_gamepad_remote( "updated_at": int(time.time()), } last_playback_state_flush_at = now - flush_playback_state() def maybe_apply_resume(now: float) -> None: nonlocal player_position_ms, resume_applied_for_key @@ -388,7 +465,6 @@ def _start_gamepad_remote( if _should_clear_resume(saved_position, player_duration_ms): resume_positions.pop(tracked_media_key, None) resume_applied_for_key = tracked_media_key - flush_playback_state() return if len(pending_requests) >= MPC_BE_MAX_INFLIGHT_REQUESTS: return @@ -409,6 +485,8 @@ def _start_gamepad_remote( status_updated_at, \ status_miss_count, \ tracked_media_key, \ + tracked_root_id, \ + tracked_relative_path, \ tracked_filepath, \ resume_applied_for_key if status_future is None or not status_future.done(): @@ -437,6 +515,8 @@ def _start_gamepad_remote( filepath, position_ms, duration_ms, state = status resolved = _media_key_for_filepath(filepath, roots) if filepath else None media_key = resolved[0] if resolved else None + root_id = resolved[1] if resolved else None + relative_path = resolved[2] if resolved else "" if tracked_media_key is not None and media_key != tracked_media_key: finalize_tracked_current() @@ -445,6 +525,8 @@ def _start_gamepad_remote( clear_tracked_current(clear_resume_applied=True) elif tracked_media_key != media_key: tracked_media_key = media_key + tracked_root_id = root_id + tracked_relative_path = relative_path tracked_filepath = filepath resume_applied_for_key = None @@ -900,7 +982,7 @@ def winmain() -> None: poll_thread: threading.Thread | None = None # Resolve all root paths for gamepad remote - gamepad_roots = [Path(p) for p in initial_roots.values()] + gamepad_roots = {root_id: Path(p) for root_id, p in initial_roots.items()} def on_shown() -> None: api._window = window @@ -913,7 +995,7 @@ def winmain() -> None: nonlocal poll_thread if poll_thread is None and _supports_gamepad_remote(): - poll_thread = _start_gamepad_remote(poll_stop, gamepad_roots) + poll_thread = _start_gamepad_remote(poll_stop, gamepad_roots, backend_url) threading.Thread( target=_activate_initial_roots,