From 2a39e1f0ea9710b09772321b0895d4b3591a2ef1 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Fri, 4 Sep 2026 00:00:53 +0000 Subject: [PATCH] Fix scanner rescan bugs and wasted work Bugs fixed: - Partial rescans no longer replace whole index entries: Upsert events now carry the scanned torrent paths and IndexStore merges partial rebuilds, so touching one season no longer drops the others from listings. - Deleted torrents are now detected: discovery reports the full candidate set via a new Sync event after each completed scan, and the store prunes file entries/episodes/seasons/items whose torrents vanished. - Seen mtimes are committed only after a scan completes successfully, so cancelled/failed scans retry their items. - Showreel worker no longer rebroadcasts stale whole items; it sends narrow MovieShowreel/EpisodeReel events that update only reel fields. - Permanently failing reel generations (e.g. DoVi/libplacebo on GPU-less machines) and short-video re-queueing are no longer retried every scan: reel outcomes persist in reel-state.json with exponential backoff. Optimizations: - Persist scan state (scan-state.json) and ffmpeg probe results (probe-cache.json, keyed by mtime+size, failures included) under .mediahive/, written only when changed. A warm restart over an unchanged library drops from ~57 s + 467 queued reel tasks to ~0.2 s with an empty queue (measured on a 163-item root). - Bounded parallelism (semaphores + gather) for cast-profile downloads, TMDb title lookups, and season-detail fetches. - Incremental TMDb-id index bookkeeping instead of rebuilding per upsert; removed dead trigger_scan. See docs/scanning-review.md for the full findings/fixes/measurements report. --- docs/scanning-review.md | 352 +++++++++++++++++++++++++++++ mediahive/hivescan/indexer.py | 98 ++++++-- mediahive/hivescan/scanner.py | 401 ++++++++++++++++++++++++++++----- mediahive/hivescan/showreel.py | 95 +++++++- mediahive/index_store.py | 298 ++++++++++++++++++++++-- mediahive/models/events.py | 41 +++- mediahive/root_registry.py | 31 ++- 7 files changed, 1218 insertions(+), 98 deletions(-) create mode 100644 docs/scanning-review.md diff --git a/docs/scanning-review.md b/docs/scanning-review.md new file mode 100644 index 0000000..0a38e3e --- /dev/null +++ b/docs/scanning-review.md @@ -0,0 +1,352 @@ +# Scanner review — findings, fixes, and measured results + +Date: 2026-09-03 (review); fixes implemented same day. +Scope: `mediahive/hivescan/*`, `mediahive/index_store.py`, `mediahive/root_registry.py`, +`mediahive/models/events.py` +Method: code review plus instrumented runs of the real `RootScanner` (monkeypatched +timers around every ffmpeg invocation, HTTP request, and filesystem primitive). + +Structure of this document: + +- **Section 1** records the findings as measured against the *pre-fix* code + (line references are from that revision). +- **Section 2** describes the fixes that were implemented for each finding. +- **Section 3** gives before/after measurements. +- **Appendix A** (blob storage options) is kept for reference only; it was + explicitly decided **not** to change the on-disk storage format for now. + +Test environment (details omitted intentionally): the media library lives on a +network-mounted filesystem (SMB/CIFS). Two roots were measured: + +- **Subset root**: 163 torrents (121 movies, 40 series entries), stale/empty index. +- **Full library root**: existing index with 1127 movies, 159 series, + 1608 episodes, ~2983 video files, ~35k people records. + +Environment characteristic that dominates several measurements: +`stat`/`scandir`/`exists` on the network mount are ~0.1 ms (attribute caching +works), but **every small file write costs ~0.2 s** (synchronous write-through). +The scanner writes thousands of small files into `.mediahive/` on that mount. + +--- + +## 1. Findings (pre-fix) + +### F1 — Partial rescan replaces whole entries (the "disappearing seasons" bug) + +**This is a correctness bug, not a performance issue.** + +Chain of events: + +1. Discovery mtime-gates per torrent directory/file. Touching one season + directory of a series yields a `downloads` list containing *only that + season*. +2. `_process_series` rebuilds the `Series` object from only the items it was + given. +3. `IndexStore.upsert_series` **replaced the entire entry** and broadcast the + partial series to all connected clients. + +Measured end-to-end on a 5-season series (5 separate season torrents, +110 episodes): + +``` +full scan -> index entry seasons [1,2,3,4,5] (110 episodes) +touch season 4 -> rescan 0.2 s, emits ONE upsert: seasons [4] (22 episodes) + -> index entry is now seasons [4] — seasons 1-3,5 gone +``` + +The missing seasons returned only when a scan happened to include all seasons +again — in practice the next **process restart**, because the seen-mtimes map +(F2) was memory-only and forced a full rediscovery at startup. + +Same bug class, other variants: + +- **Movies**: a touched version directory dropped the other versions of the + same movie from the listing. +- **TMDb dedupe collapse**: when a partial entry arrived under a *different* + item id for an already-known TMDb id, the old **complete** entry was + explicitly deleted. +- **Deletions were never detected**: discovery only ever added to + `_seen_mtimes`; nothing emitted removals. A torrent deleted from disk stayed + in the index forever. + +### F2 — No persistent scan state: every restart was a full reprocess + +All scanner state was process memory: `_seen_mtimes`, the ffmpeg probe cache, +and the episode/playable-file/bluray-probe caches. + +Consequences measured: + +- **Steady-state rescan within one process: 0.2 s** (subset root, nothing + changed) — mtime gating worked fine while the process lived. +- **Warm-restart scan (fresh process caches, all disk caches warm): 56.5 s** + for the same 163 items, of which **52.9 s (94 %) was re-running ffmpeg probes + on all 469 video files** (21 s `ffmpeg -i` + 32 s `showinfo` passes on HDR + files). TMDb was 100 % disk-cache hits and cost 0.8 s total. +- Scaled to the full library: every application restart re-probed **~2983 + files ≈ 6–8 minutes** of sequential ffmpeg, during which the whole index was + re-derived and re-upserted item by item (see F4). + +### F3 — Preview (showreel) generation: restart storms and infinite retries + +- **Every scan that included an item enqueued all of its reel tasks**, whether + or not the reels existed. Existence was only checked later by the serial + worker. A full scan of the subset root enqueued **467** tasks; the full + library would enqueue ~2700. +- **Nothing was persisted about the queue.** A restart before the queue drained + started everything over. +- **Failures were never recorded.** In the drain test, 3 of 18 movies failed + deterministically (DoVi profile 7 titles require `libplacebo` tonemapping, + which fails on GPU-less machines; one file has a matroska demux error). The + same files were retried on every subsequent drain — ~1.5 s of probing plus + crop detection plus an error task broadcast to every client, **forever**. +- The reel-existence check required **all five** reels; short videos + legitimately produce fewer, so they were treated as "missing" and re-queued + on every scan, generating nothing new each time. +- Measured generation pace with software AV1 encoding: ~16–28 s per movie + (5 clips), ~4–5 s per episode clip. +- The reel worker rebroadcast **whole items** built from stale scan data, + clobbering newer store state. + +### F4 — Degraded operation while a full scan is in progress + +- Items were re-upserted one by one as processed; combined with F1, any + partial rescan interleaved with normal use made listings lose data until the + next restart. +- The showreel worker broadcast an **error task for every permanent failure on + every scan** (F3), producing user-visible noise. +- The index snapshot was rewritten every 5 s while dirty; with 1286+ items + that is a ~7.5 MB serialize + write per flush, continuously, for the + duration of a scan. +- `upsert_*` rebuilt the TMDb-id lookup maps on **every** upsert — O(n²) per + scan. Measured negligible; fixed anyway as part of the merge work. + +### F5 — Cold-scan cost was serialized small-file I/O, not TMDb + +Cold scan of the subset root: **975 s for 163 items**. + +| Time | Share | Where | +|---|---|---| +| 763 s | 78 % | `download_cast_profile`: 3494 cast images, strictly serialized; ≈0.22 s each ≈ 0.19 s network-mount write + 0.03 s HTTP | +| ~100 s | 10 % | ffmpeg probes (469 files, incl. 83 HDR `showinfo` passes) | +| 73 s | 7 % | TMDb API layer: 217 uncached requests (27 s HTTP) **plus ~41 s writing per-request cache JSON files** to the network mount | +| ~75 s | 8 % | covers / backdrops / season posters (same small-write cost) | +| 0.3 s | — | filesystem discovery walk | + +Notes: + +- The TMDb disk cache itself is fine: once warm it serves 282 requests in + 0.8 s. Cache *reads* need no optimization. +- The full cast of every title was downloaded sequentially, one tiny file per + person (the full library has ~35k people records). Re-runs are cheap + (exists-check), so this was a cold-scan-only cost — but it made the first + scan of a new root take ~8× longer than everything else combined. + +### F6 — Measured as noise (not worth effort) + +- The 30-second rescan loop's tree walk: 3.5–6 s per pass over the full + library (~1900 directories). Continuous but light. +- `get_directory_size` per torrent: 0.1 s total in the scan. +- TMDb disk-cache reads: sub-second per scan. +- `trigger_scan` was dead code — nothing called it. +- `_seen_mtimes` was updated *before* processing; a cancelled/failed scan + permanently lost that update until the next restart. +- The in-process probe cache was keyed by path only; a replaced file kept + stale probe data until restart. + +--- + +## 2. Implemented fixes + +All proposals P1–P5 from the review were implemented, keeping the existing +on-disk format unchanged (no blob storage — see Appendix A). + +### F1 → merge-semantics upserts + deletion sync + +- The `Upsert` event now carries `scanned: list[str]` — the media-root-relative + torrent paths whose content was (re)scanned to build the item + (`mediahive/models/events.py`; `_process_movies`/`_process_series` yield it). +- `IndexStore.upsert_movie/upsert_series` (`mediahive/index_store.py`) merge a + partial rebuild into the existing entry instead of replacing it: + file entries belonging to scanned torrents are replaced, everything else is + preserved, episodes/seasons emptied by the merge are dropped, same-episode + multi-release files are unioned, and non-None scalar fields from the fresh + scan win. The store broadcasts only when the merged result actually changed. +- The TMDb dedupe collapse now folds duplicates through the same merge (the + complete entry's scalars win), so a partial candidate can no longer delete a + complete entry; the reverse TMDb-id map is fixed up incrementally instead of + rebuilding both maps per upsert (also F4/P5). +- **Deletion sync**: discovery collects the full set of candidate torrent + paths; after a fully completed scan the scanner emits a new `Sync` event and + `IndexStore.sync_torrent_paths` drops file entries whose torrent path is + gone, cascading to empty episodes/seasons/items with proper removals. + +### F2 → persisted scan state and probe cache + +Three small JSON files under `.mediahive/` per root, loaded at scanner start +and written atomically (tmp + rename) **only when changed**: + +- `scan-state.json`: relpath → mtime. A restart over an unchanged library now + discovers "0 new items" and finishes in walk time. Mtimes are committed + **after** the scan completes successfully (fixes the pre-commit nit from F6): + a cancelled/failed scan retries its items. +- `probe-cache.json`: path → {mtime, size, probe fields} for every probed + file, **failures included**. Keying by mtime+size makes it self-invalidating + when a file is replaced (also fixes the stale-probe nit from F6). Non-plain + paths (bluray:/concat: URIs) fail `stat` and stay memory-cached only. +- `reel-state.json`: see F3 below. + +### F3 → reel-state persistence, backoff, and queue gating + +- `reel-state.json` records, per media folder (movies) or per episode + (`folder#SxxEyy`), the video's mtime+size, status (`done`/`failed`), + attempt count, and last-attempt timestamp. +- `_reel_needed` gates both scan-time queueing and the worker: + `done` entries are skipped; `failed` entries back off exponentially + (6 h → 12 h → … capped at 1 week); entries with no record fall back to a + cheap on-disk existence check, and existing reels are silently recorded as + `done` so future scans take the cheap path. This ends both the infinite + retries of unreadable files and the re-queueing of short videos. +- The worker no longer rebroadcasts whole (stale) items: it sends narrow + `MovieShowreel` / `EpisodeReel` events, and the store updates only the reel + fields of the current entry (`set_movie_showreel` / `set_episode_reel`). +- Reel state is persisted at scan finalize, on scanner `stop()`, and as soon + as the reel queue drains (a crash between scans no longer loses records). + +### F5 → bounded parallelism for downloads and TMDb fetches + +- Cast-profile downloads: `asyncio.gather` with a semaphore of 8. +- TMDb title lookups (movies and series) are prefetched in parallel + (semaphore of 4) before the grouping loops, which then read the per-call + caches. +- Season-detail fetches are prefetched in parallel (semaphore of 4) before the + season loop. +- Per-item cover/backdrop/poster logic is unchanged (exists-check-fast when + warm). + +### Housekeeping (P5) + +- Dead `trigger_scan` removed (`is_scanning` kept). +- TMDb-id index bookkeeping is incremental (see F1 above). +- `_rebuild_tmdb_indexes` remains only for snapshot load and dedupe. + +--- + +## 3. Measured results + +Subset root (163 items, 121 movies / 40 series entries, 469 video files), +same network mount: + +| Scenario | Before | After | +|---|---|---| +| Touch one season of a 5-season series | other 4 seasons vanish until next full scan | 0.3 s rescan, one partial upsert (`scanned=[S04]`), store keeps all 5 seasons | +| Steady rescan, nothing changed (same process) | 0.2 s | 0.2 s, 0 upserts, no writes | +| Warm restart, unchanged library (fresh process) | 56.5 s (94 % ffmpeg re-probes) + 467 reel tasks queued | **0.2 s, 0 upserts, reel queue 0** | +| First scan with warm TMDb cache but no probe cache | 56.5 s | 51.6 s once — writes `probe-cache.json` (469 records, 207 KB); subsequent runs skip all probing | +| Permanently unreadable files (3 DoVi/libplacebo movies) | retried on every scan and every startup, error broadcast each time | recorded as failed once, skipped within backoff | +| Cold scan, nothing cached | 975 s | not re-measured end-to-end; the dominant terms are now 8-way parallel (cast images: 3494 downloads measured at 3.7 s when warm) | +| Deleted torrent | stayed in index forever | removed on the next completed scan via `Sync` | + +Unit-level checks (synthetic `IndexStore` + scanner state, no filesystem +library involved): partial upsert preserves untouched seasons and replaces +rescanned torrent files; multi-release episode union; sync removal cascades; +single-episode reel updates; probe-cache save/load roundtrip; reel backoff +math; reel-state persistence roundtrip across scanner instances. + +--- + +## Appendix A — Blob storage options: scan-time vs runtime concerns + +**Status: reference only — not implemented.** The current on-disk format +(one file per artifact) was deliberately kept. This appendix stays as +documentation of the options if write amplification or file counts ever +become an operational issue. + +The two concerns have opposite constraints, so they should be decided +separately: + +- **Scan-time data** is written and read only by the scanner. Nothing in the + server serves it. Storage format is therefore a pure implementation detail + and can be changed freely. +- **Runtime data** is delivered to the frontend as plain files via + `/api/assets/{root}/{movies|series|people}/{path}` (covers, posters, + backdrops, person photos) and `/api/media/{root}/{path}` (media files), with + etag/range streaming. Anything that replaces files here must keep an HTTP + serving story working. + +### A.1 Artifact inventory (measured) + +| Artifact | Class | Avg size | Count (full library) | Total | Written | +|---|---|---|---|---|---| +| TMDb response cache JSON | scan-time | 25 KB (median 9 KB) | ~8,500 | ~215 MB | on every uncached API request | +| Probe results / scan state / reel state | scan-time | ~0.5 KB/record | ~3,000 records | ~1.5 MB | per processed file | +| Person photos | runtime | 11.4 KB | ~22,500 | ~262 MB | once per person | +| `cover.jpg` / season posters | runtime | ~77 KB | ~2,100 | ~164 MB | once per title/season | +| `backdrop.jpg` | runtime | ~147 KB | ~1,460 | ~220 MB | once per title | +| Reel clips (WebM/AV1) | runtime | ~450 KB | ~8,700 | ~4.0 GB | once per title/episode | +| `index.json` snapshot | both | 7.5 MB | 1 | 7.5 MB | debounced, only when dirty | + +Reference point for the write-amplification math: on the network mount one +small-file write costs ~0.2 s, while a single 7.5 MB sequential write costs +the same ~0.2 s (~40 MB/s). So ~31,000 tiny files ≈ 1.7 hours of serialized +write time, versus ~6 s for the same bytes as one bulk dump. + +### A.2 Scan-time blob store (TMDb cache, probe cache, scan state) + +Nothing here is served, so the only requirement is fast lookup + cheap +persistence. Two workable shapes: + +- **RAM map + debounced atomic dump.** Plain dicts keyed by request hash / + file path, dumped as one binary file (length-prefixed msgspec or JSON blob, + optionally zstd-compressed) with tmp-write + rename, on the same + dirty-flag + debounce discipline `index.json` already uses. Effects: the + ~8,500 individual cache writes collapse into a handful of bulk flushes; + warm lookups become dict hits with zero filesystem calls. TMDb JSON + compresses ~10× (215 MB → ~20–25 MB), so a full dump is a sub-second write. + Caveat: holding all responses parsed in RAM costs ~200 MB for the full + library; storing raw response *bytes* and parsing lazily, or capping to + entries referenced by known index items, keeps this modest. +- **SQLite (stdlib, WAL mode).** One database file, incremental commits, crash + safety without full dumps, and kernel page cache instead of explicit RAM + management. Better fit if the cache is allowed to grow unbounded, at the + price of slightly more code. + +Either way, keep the existing cache *semantics* unchanged: cache HTTP-level +failures, never cache network errors. With persisted scan state in place the +TMDb cache becomes write-rarely (new items only), which further lowers the +value of elaborate engineering here — the simple dump is likely enough. + +### A.3 Runtime-served artifacts + +- **Reels, covers, backdrops, season posters: keep as files.** They are few + per title, tens-to-hundreds of KB, written exactly once, and benefit from + the existing etag/range file serving. No write-amplification problem. +- **Person photos** (~22.5k files × 11.4 KB) are the one runtime class where + tiny files hurt at scan time. Three options, in increasing invasiveness: + 1. **Keep files, fix only the scan-time behavior** — bounded-parallel + downloads, optionally capped to top-N billed cast. Zero changes to + serving; the cold-scan cost drops ~8× but the file count stays. + *(This is the option currently implemented.)* + 2. **Blob db + serve from the db.** Person photos move into the same store + as above; the assets handler gains one branch for the `people` asset + type that streams bytes from the db instead of the filesystem. + Eliminates all 22.5k tiny files. If full RAM residency (262 MB) is + undesirable, use SQLite and let the page cache handle it. + 3. **Hybrid lazy materialization.** The db is authoritative at scan time + (no tiny writes during scans); the assets handler writes the photo to the + conventional path on first request and serves it as a file thereafter. + Serving logic and URLs stay unchanged; disk usage appears only for + people actually viewed. + +Note that `index.json` itself is already the right shape: one atomic 7.5 MB +file, rewritten only when dirty. The goal for everything else is simply to +reach the same shape per concern. + +### A.4 Recommendation (if revisited) + +| Concern | Recommended treatment | +|---|---| +| TMDb response cache | RAM map + debounced single-file dump (SQLite if growth matters) | +| Probe cache, seen-mtimes, reel-failure records | same dump mechanism, separate small files *(currently: three small JSON files, written only when dirty)* | +| Person photos | option 1 now (parallel + lazy fetching); option 2 or 3 if file count becomes an operational issue | +| Covers, posters, backdrops, reels | unchanged — plain files | +| `index.json` | unchanged | diff --git a/mediahive/hivescan/indexer.py b/mediahive/hivescan/indexer.py index 4759e4c..c811c9f 100644 --- a/mediahive/hivescan/indexer.py +++ b/mediahive/hivescan/indexer.py @@ -148,20 +148,30 @@ async def _cache_people_profiles( if not info or not info.cast: return info, people + semaphore = asyncio.Semaphore(8) + + async def fetch_profile(cast_credit, person): + async with semaphore: + return cast_credit.id, await download_cast_profile( + person.profile_path, + media_folder, + person.name, + cast_credit.id, + ) + + tasks = [] for cast_credit in info.cast: if cast_credit.id is None: continue person = people.get(cast_credit.id) if person is None or not person.profile_path: continue - downloaded_path = await download_cast_profile( - person.profile_path, - media_folder, - person.name, - cast_credit.id, - ) + tasks.append(fetch_profile(cast_credit, person)) + + for cast_id, downloaded_path in await asyncio.gather(*tasks): if downloaded_path: - people[cast_credit.id] = Person( + person = people[cast_id] + people[cast_id] = Person( name=person.name, profile_path=Path(downloaded_path).name, gender=person.gender, @@ -378,6 +388,25 @@ async def _build_seasons_data( seasons_map[season_num] = {} seasons_map[season_num][episode_num] = files + # Prefetch all missing season details in parallel; the loop below then + # reads them straight from season_cache. + if tmdb_id: + missing = [ + season_num + for season_num in seasons_map + if (tmdb_id, season_num) not in season_cache + ] + if missing: + semaphore = asyncio.Semaphore(4) + + async def prefetch(num: int) -> None: + async with semaphore: + season_cache[tmdb_id, num] = await fetch_season_details( + tmdb_id, num + ) + + await asyncio.gather(*(prefetch(num) for num in missing)) + seasons_data = [] for season_num in sorted(seasons_map.keys()): episodes_in_season = seasons_map[season_num] @@ -442,11 +471,15 @@ async def _process_movies( generate_showreels: bool, media_root: str | None = None, root_id: str | None = None, -) -> AsyncIterator[tuple[str, Movie, tuple[str, Path, str] | None, dict[int, Person]]]: +) -> AsyncIterator[ + tuple[str, Movie, tuple[str, Path, str] | None, dict[int, Person], list[str]] +]: """Async generator that processes all movies. Yields: - Tuples of ``(Movie, showreel_task_or_None)`` as each movie is processed. + Tuples of ``(movie_id, Movie, showreel_task_or_None, people, scanned)`` + as each movie is processed. ``scanned`` lists the media-root-relative + torrent paths whose content was (re)scanned to build the movie. """ _ = root_id @@ -499,6 +532,21 @@ async def _process_movies( len(categories[ContentType.MOVIE]), ) if movie_groups else None + # Prefetch TMDb lookups for all unique titles in parallel; the grouping + # loop below then reads them straight from movie_tmdb_cache. + if movie_groups: + semaphore = asyncio.Semaphore(4) + first_by_key = { + f"{items[0].title.lower()}:{items[0].year}": items[0] + for items in movie_groups.values() + } + + async def prefetch(item: ParsedContent) -> None: + async with semaphore: + await get_movie_tmdb(item.title, item.year) + + await asyncio.gather(*(prefetch(item) for item in first_by_key.values())) + for idx, (_movie_key, items) in enumerate(movie_groups.items(), 1): first_item = items[0] logger.debug( @@ -640,7 +688,8 @@ async def _process_movies( showreel_source_sets=showreel_source_sets or None, files=files, ) - yield item_id, movie, showreel_task, people + scanned = [make_relative_path(item.path.as_posix(), media_root) for item in items] + yield item_id, movie, showreel_task, people, scanned # Process movies without TMDb info for group_data in no_tmdb_movie_groups.values(): @@ -712,7 +761,8 @@ async def _process_movies( showreel_source_sets=showreel_source_sets or None, files=files, ) - yield item_id, movie, showreel_task, {} + scanned = [make_relative_path(item.path.as_posix(), media_root) for item in items] + yield item_id, movie, showreel_task, {}, scanned async def _process_series( @@ -723,12 +773,14 @@ async def _process_series( media_root: str | None = None, root_id: str | None = None, ) -> AsyncIterator[ - tuple[str, Series, list[tuple[str, Path, int, int, str]], dict[int, Person]] + tuple[str, Series, list[tuple[str, Path, int, int, str]], dict[int, Person], list[str]] ]: """Async generator that processes all series. Yields: - Tuples of ``(Series, episode_reel_tasks)`` as each series is processed. + Tuples of ``(series_id, Series, episode_reel_tasks, people, scanned)`` + as each series is processed. ``scanned`` lists the media-root-relative + torrent paths whose content was (re)scanned to build the series. """ _ = root_id @@ -783,6 +835,20 @@ async def _process_series( len(categories[ContentType.SERIES]), ) + # Prefetch TMDb lookups for all unique titles in parallel; the grouping + # loop below then reads them straight from series_tmdb_cache. + if series_groups: + semaphore = asyncio.Semaphore(4) + first_by_key = { + items[0].title.lower(): items[0] for items in series_groups.values() + } + + async def prefetch(item: ParsedContent) -> None: + async with semaphore: + await get_series_tmdb(item.title) + + await asyncio.gather(*(prefetch(item) for item in first_by_key.values())) + for idx, (_series_key, items) in enumerate(series_groups.items(), 1): first_item = items[0] logger.debug(" [%d/%d] %s", idx, len(series_groups), first_item.title) @@ -898,7 +964,8 @@ async def _process_series( backdrop_path=make_relative_path(backdrop_path, media_root), seasons=seasons_data, ) - yield series_id, series, ep_reel_tasks, people + scanned = [make_relative_path(item.path.as_posix(), media_root) for item in items] + yield series_id, series, ep_reel_tasks, people, scanned # Process series without TMDb info for group_data in no_tmdb_groups.values(): @@ -941,4 +1008,5 @@ async def _process_series( cover_path=make_relative_path(cover_path, media_root), seasons=seasons_data, ) - yield series_id, series, ep_reel_tasks, {} + scanned = [make_relative_path(item.path.as_posix(), media_root) for item in items] + yield series_id, series, ep_reel_tasks, {}, scanned diff --git a/mediahive/hivescan/scanner.py b/mediahive/hivescan/scanner.py index 6dd3d0b..877fc16 100644 --- a/mediahive/hivescan/scanner.py +++ b/mediahive/hivescan/scanner.py @@ -13,9 +13,11 @@ from __future__ import annotations import asyncio import contextlib +import json import logging import os import threading +import time import uuid from collections.abc import Awaitable, Callable from pathlib import Path @@ -33,15 +35,25 @@ from mediahive.hivescan.showreel import ( generate_showreel_images, get_existing_episode_reel_sources, get_existing_showreel_source_sets, + load_probe_records, movie_showreels_exist, + probe_records_dirty, + save_probe_records, ) from mediahive.hivescan.tmdb_client import set_cache_dir from mediahive.hivescan.utils import ( DEFAULT_OUTPUT_FOLDER, make_relative_path, ) -from mediahive.models.data import Movie, Series, TaskInfo -from mediahive.models.events import ScanEvent, Task, Upsert +from mediahive.models.data import TaskInfo +from mediahive.models.events import ( + EpisodeReel, + MovieShowreel, + ScanEvent, + Sync, + Task, + Upsert, +) logger = logging.getLogger("hivescan.scanner") @@ -76,6 +88,13 @@ class RootScanner: self._rescan_worker_task: asyncio.Task | None = None self._seen_mtimes: dict[str, int] = {} + # Persisted state (under .mediahive/) + self._scan_state_path = self._output_dir / "scan-state.json" + self._reel_state_path = self._output_dir / "reel-state.json" + self._probe_cache_path = self._output_dir / "probe-cache.json" + self._reel_state: dict[str, dict[str, dict]] = {"movies": {}, "episodes": {}} + self._reel_state_dirty = False + # ------------------------------------------------------------------ # Public lifecycle # ------------------------------------------------------------------ @@ -84,12 +103,16 @@ class RootScanner: """Initialise and start background workers.""" await AsyncPath(self._output_dir).mkdir(parents=True, exist_ok=True) set_cache_dir(self._output_dir / ".tmdb-cache") + await asyncio.to_thread(self._load_scan_state) + await asyncio.to_thread(load_probe_records, self._probe_cache_path) + await asyncio.to_thread(self._load_reel_state) logger.info( - "Scanner started for root %s — path=%s, scanignore=%s", + "Scanner started for root %s — path=%s, scanignore=%s, known-paths=%d", self.root_id, self.media_root, "loaded" if self._scanignore.file_path.exists() else "defaults only", + len(self._seen_mtimes), ) self._showreel_worker_task = asyncio.create_task(self._showreel_worker()) @@ -110,6 +133,11 @@ class RootScanner: if task and not task.done(): with contextlib.suppress(TimeoutError, asyncio.CancelledError): await asyncio.wait_for(task, timeout=2.0) + # Best-effort persistence of scanner state. + with contextlib.suppress(Exception): + await asyncio.to_thread(self._save_scan_state) + await asyncio.to_thread(self._save_reel_state) + await asyncio.to_thread(save_probe_records) def is_scanning(self) -> bool: return self._scan_task is not None and not self._scan_task.done() @@ -117,12 +145,163 @@ class RootScanner: def showreel_queue_size(self) -> int: return self._showreel_queue.qsize() - def trigger_scan(self) -> bool: - """Start a scan. Returns False if one is already running.""" - if self.is_scanning(): - return False - self._start_scan() - return True + # ------------------------------------------------------------------ + # Scan-state persistence + # ------------------------------------------------------------------ + + def _load_scan_state(self) -> None: + try: + data = json.loads(self._scan_state_path.read_text(encoding="utf-8")) + except FileNotFoundError: + return + except OSError, ValueError: + logger.exception("Failed to load scan state from %s", self._scan_state_path) + return + seen = data.get("seen_mtimes") + if isinstance(seen, dict): + self._seen_mtimes = {str(k): int(v) for k, v in seen.items()} + + def _save_scan_state(self) -> None: + try: + payload = json.dumps({"version": 1, "seen_mtimes": self._seen_mtimes}) + tmp = self._scan_state_path.with_suffix(".tmp") + tmp.parent.mkdir(parents=True, exist_ok=True) + tmp.write_text(payload, encoding="utf-8") + tmp.replace(self._scan_state_path) + except OSError, TypeError, ValueError: + logger.exception("Failed to save scan state to %s", self._scan_state_path) + + # ------------------------------------------------------------------ + # Reel-state persistence and gating + # + # ``reel-state.json`` records, per media folder (movies) or per episode, + # whether reel generation succeeded or failed for the current video file + # (keyed by mtime+size). This stops the showreel worker from retrying + # permanently unreadable files on every scan, and stops short videos from + # being re-queued forever because they legitimately have fewer reels than + # the maximum. + # ------------------------------------------------------------------ + + # Retry delay for failed generations: 6h, 12h, 24h, ... capped at a week. + _REEL_RETRY_BASE_HOURS = 6 + _REEL_RETRY_MAX_HOURS = 168 + + def _load_reel_state(self) -> None: + try: + data = json.loads(self._reel_state_path.read_text(encoding="utf-8")) + except FileNotFoundError: + return + except OSError, ValueError: + logger.exception("Failed to load reel state from %s", self._reel_state_path) + return + for bucket in ("movies", "episodes"): + records = data.get(bucket) + if isinstance(records, dict): + self._reel_state[bucket] = records + + def _save_reel_state(self) -> None: + try: + payload = json.dumps({"version": 1, **self._reel_state}) + tmp = self._reel_state_path.with_suffix(".tmp") + tmp.parent.mkdir(parents=True, exist_ok=True) + tmp.write_text(payload, encoding="utf-8") + tmp.replace(self._reel_state_path) + except OSError, TypeError, ValueError: + logger.exception("Failed to save reel state to %s", self._reel_state_path) + + def _reel_state_bucket( + self, + kind: str, + media_folder: Path, + season_num: int | None = None, + episode_num: int | None = None, + ) -> tuple[dict[str, dict], str]: + folder_key = make_relative_path( + media_folder.as_posix(), self.media_root.as_posix() + ) + if kind == "movie": + return self._reel_state["movies"], folder_key + key = f"{folder_key}#S{season_num:02d}E{episode_num:02d}" + return self._reel_state["episodes"], key + + def _record_reel_state( + self, + kind: str, + media_folder: Path, + season_num: int | None, + episode_num: int | None, + video_path: str, + sig: tuple[int, int] | None, + status: str, + error: str | None = None, + ) -> None: + bucket, key = self._reel_state_bucket(kind, media_folder, season_num, episode_num) + previous = bucket.get(key) + attempts = ( + 0 if status == "done" else int((previous or {}).get("attempts", 0)) + 1 + ) + bucket[key] = { + "video": make_relative_path(video_path, self.media_root.as_posix()), + "mtime": sig[0] if sig else None, + "size": sig[1] if sig else None, + "status": status, + "attempts": attempts, + "last": time.time(), + "error": error, + } + self._reel_state_dirty = True + + async def _reel_needed( + self, + kind: str, + video_path: str, + media_folder: Path, + season_num: int | None = None, + episode_num: int | None = None, + ) -> tuple[bool, tuple[int, int] | None]: + """Decide whether a reel-generation task should be queued. + + Returns ``(needed, signature)`` where signature is the video file's + (mtime, size) or None when it cannot be stat'ed. + """ + sig: tuple[int, int] | None = None + try: + st = await AsyncPath(video_path).stat() + sig = (int(st.st_mtime), st.st_size) + except OSError, ValueError: + pass + + bucket, key = self._reel_state_bucket(kind, media_folder, season_num, episode_num) + rec = bucket.get(key) + if ( + rec is not None + and sig is not None + and rec.get("mtime") == sig[0] + and rec.get("size") == sig[1] + ): + if rec.get("status") == "done": + return False, sig + attempts = int(rec.get("attempts", 1)) + delay = min( + self._REEL_RETRY_BASE_HOURS * 2**attempts, + self._REEL_RETRY_MAX_HOURS, + ) * 3600 + if time.time() - float(rec.get("last", 0)) < delay: + return False, sig + + if kind == "movie": + exists = await movie_showreels_exist(media_folder) + else: + exists = await episode_reel_exists(media_folder, season_num, episode_num) + if exists: + if rec is None and sig is not None: + # Reels already on disk (e.g. generated before this feature): + # record success so future scans take the cheap path. + self._record_reel_state( + kind, media_folder, season_num, episode_num, video_path, sig, "done" + ) + return False, sig + return True, sig # ------------------------------------------------------------------ # Internal scan orchestration @@ -143,9 +322,20 @@ class RootScanner: except Exception: logger.exception("Rescan loop error") - async def _discover_downloads(self, task_id: str) -> list[ParsedContent]: - """Recursively walk the media root, respecting scanignore rules.""" + async def _discover_downloads( + self, task_id: str + ) -> tuple[list[ParsedContent], dict[str, int], set[str]]: + """Recursively walk the media root, respecting scanignore rules. + + Returns ``(downloads, found_mtimes, known_paths)``: the new/changed + items to process, the mtimes observed for them (committed to + ``_seen_mtimes`` only after the scan completes successfully), and the + full set of media-root-relative candidate paths seen on disk (used + for deletion detection). + """ downloads: list[ParsedContent] = [] + found_mtimes: dict[str, int] = {} + known_paths: set[str] = set() media_root_str = self.media_root.as_posix() dirs_visited = 0 @@ -251,6 +441,7 @@ class RootScanner: if is_media_container: relpath = make_relative_path(str(directory), media_root_str) + known_paths.add(relpath) try: stat_info = await ap.stat() mtime = int(stat_info.st_mtime) @@ -260,7 +451,7 @@ class RootScanner: relpath not in self._seen_mtimes or self._seen_mtimes[relpath] != mtime ): - self._seen_mtimes[relpath] = mtime + found_mtimes[relpath] = mtime downloads.append(await parse_download(directory)) return @@ -277,6 +468,7 @@ class RootScanner: for child_file in child_files: if child_file.suffix.lower() in video_extensions: relpath = make_relative_path(str(child_file), media_root_str) + known_paths.add(relpath) try: stat_info = await AsyncPath(child_file).stat() mtime = int(stat_info.st_mtime) @@ -287,10 +479,11 @@ class RootScanner: and self._seen_mtimes[relpath] == mtime ): continue - self._seen_mtimes[relpath] = mtime + found_mtimes[relpath] = mtime downloads.append(await parse_download(child_file)) else: relpath = make_relative_path(str(directory), media_root_str) + known_paths.add(relpath) try: stat_info = await ap.stat() mtime = int(stat_info.st_mtime) @@ -298,7 +491,7 @@ class RootScanner: return if relpath in self._seen_mtimes and self._seen_mtimes[relpath] == mtime: return - self._seen_mtimes[relpath] = mtime + found_mtimes[relpath] = mtime downloads.append(await parse_download(directory)) logger.info("Starting filesystem discovery at %s", self.media_root) @@ -316,7 +509,7 @@ class RootScanner: raise except OSError, PermissionError: logger.exception("Cannot list media root: %s", self.media_root) - return downloads + return downloads, found_mtimes, known_paths for item_async in root_children: item = Path(item_async) @@ -329,6 +522,7 @@ class RootScanner: await _walk(item) else: relpath = make_relative_path(str(item), media_root_str) + known_paths.add(relpath) try: stat_info = await AsyncPath(item).stat() mtime = int(stat_info.st_mtime) @@ -336,7 +530,7 @@ class RootScanner: continue if relpath in self._seen_mtimes and self._seen_mtimes[relpath] == mtime: continue - self._seen_mtimes[relpath] = mtime + found_mtimes[relpath] = mtime downloads.append(await parse_download(item)) logger.info( @@ -344,7 +538,31 @@ class RootScanner: len(downloads), dirs_visited, ) - return downloads + return downloads, found_mtimes, known_paths + + async def _finalize_scan( + self, found_mtimes: dict[str, int], known_paths: set[str] + ) -> None: + """Commit discovery state after a fully completed scan. + + Commits observed mtimes (so cancelled/failed scans retry their items), + prunes vanished paths, persists state when anything changed, and sends + the Sync event that lets the store drop deleted torrents. + """ + committed = {k: v for k, v in self._seen_mtimes.items() if k in known_paths} + committed.update(found_mtimes) + state_changed = committed != self._seen_mtimes + self._seen_mtimes = committed + + await self._send(Sync(paths=sorted(known_paths))) + + if state_changed: + await asyncio.to_thread(self._save_scan_state) + if probe_records_dirty(): + await asyncio.to_thread(save_probe_records) + if self._reel_state_dirty: + self._reel_state_dirty = False + await asyncio.to_thread(self._save_reel_state) async def _run_scan(self) -> None: """Full scan pipeline: @@ -370,10 +588,13 @@ class RootScanner: ) ) - downloads = await self._discover_downloads(task_id) + downloads, found_mtimes, known_paths = await self._discover_downloads( + task_id + ) if not downloads: logger.info("No new downloads found (%s)", task_id) + await self._finalize_scan(found_mtimes, known_paths) await self._send( Task( data=TaskInfo( @@ -418,7 +639,7 @@ class RootScanner: ) ) ) - async for movie_id, movie, showreel_task, people in _process_movies( + async for movie_id, movie, showreel_task, people, scanned in _process_movies( categories, self._output_dir, fetch_covers=True, @@ -432,22 +653,34 @@ class RootScanner: id=movie_id, item=movie, people=people or None, + scanned=scanned, ) ) if showreel_task: - await self._showreel_queue.put(( - "movie", - movie_id, - showreel_task, - movie, - )) - logger.info( - "[%d/%d] Movie: %s (showreel queued, queue=%d)", - processed + 1, - total, - movie.title, - self._showreel_queue.qsize(), + needed, sig = await self._reel_needed( + "movie", showreel_task[0], showreel_task[1] ) + if needed: + await self._showreel_queue.put(( + "movie", + movie_id, + showreel_task, + sig, + )) + logger.info( + "[%d/%d] Movie: %s (showreel queued, queue=%d)", + processed + 1, + total, + movie.title, + self._showreel_queue.qsize(), + ) + else: + logger.info( + "[%d/%d] Movie: %s (showreel up to date)", + processed + 1, + total, + movie.title, + ) else: logger.info( "[%d/%d] Movie: %s (no showreel task)", @@ -480,7 +713,7 @@ class RootScanner: ) ) ) - async for series_id, series, ep_reel_tasks, people in _process_series( + async for series_id, series, ep_reel_tasks, people, scanned in _process_series( categories, self._output_dir, fetch_covers=True, @@ -494,22 +727,29 @@ class RootScanner: id=series_id, item=series, people=people or None, + scanned=scanned, ) ) + queued = 0 for task in ep_reel_tasks: - await self._showreel_queue.put(("episode", series_id, task, series)) - if ep_reel_tasks: + needed, sig = await self._reel_needed( + "episode", task[0], task[1], task[2], task[3] + ) + if needed: + await self._showreel_queue.put(("episode", series_id, task, sig)) + queued += 1 + if queued: logger.info( "[%d/%d] Series: %s (%d episode reels queued, queue=%d)", processed + 1, total, series.title, - len(ep_reel_tasks), + queued, self._showreel_queue.qsize(), ) else: logger.info( - "[%d/%d] Series: %s (no reel tasks)", + "[%d/%d] Series: %s (reels up to date)", processed + 1, total, series.title, @@ -527,6 +767,7 @@ class RootScanner: ) ) + await self._finalize_scan(found_mtimes, known_paths) await self._send( Task( data=TaskInfo( @@ -578,16 +819,14 @@ class RootScanner: """Background worker that generates showreels one at a time.""" logger.info("Showreel worker started for root %s", self.root_id) media_root_path = self.media_root - media_root_str = self.media_root.as_posix() while True: try: - kind, item_id, task_data, item = await self._showreel_queue.get() + kind, item_id, task_data, sig = await self._showreel_queue.get() task_id = f"showreel-{uuid.uuid4().hex[:8]}" remaining = self._showreel_queue.qsize() if kind == "movie": - movie: Movie = item video_path, media_folder, title = task_data logger.info( "Showreel dequeued: %s (video=%s, folder=%s, %d remaining)", @@ -596,7 +835,8 @@ class RootScanner: media_folder, remaining, ) - if await movie_showreels_exist(media_folder): + needed, _ = await self._reel_needed("movie", video_path, media_folder) + if not needed: logger.info("Showreel skipped (already exists): %s", title) self._showreel_queue.task_done() continue @@ -620,9 +860,16 @@ class RootScanner: media_folder, media_root=media_root_path ) paths = [sources[0] for sources in source_sets if sources] - movie.showreel_images = paths or None - movie.showreel_source_sets = source_sets or None - await self._send(Upsert(kind="movie", id=item_id, item=movie)) + self._record_reel_state( + "movie", media_folder, None, None, video_path, sig, "done" + ) + await self._send( + MovieShowreel( + id=item_id, + showreel_images=paths or None, + showreel_source_sets=source_sets or None, + ) + ) await self._send( Task( data=TaskInfo( @@ -636,6 +883,16 @@ class RootScanner: ) ) else: + self._record_reel_state( + "movie", + media_folder, + None, + None, + video_path, + sig, + "failed", + error="no reels produced", + ) logger.warning( "Showreel generation returned nothing: %s", title ) @@ -651,7 +908,6 @@ class RootScanner: ) elif kind == "episode": - series: Series = item video_path, media_folder, season_num, episode_num, series_title = ( task_data ) @@ -663,7 +919,10 @@ class RootScanner: video_path, remaining, ) - if await episode_reel_exists(media_folder, season_num, episode_num): + needed, _ = await self._reel_needed( + "episode", video_path, media_folder, season_num, episode_num + ) + if not needed: logger.info( "Episode reel skipped (already exists): %s %s", series_title, @@ -694,20 +953,29 @@ class RootScanner: episode_num, media_root=media_root_path, ) - for season in series.seasons: - if season.season_number == season_num: - for episode in season.episodes: - if episode.episode_number == episode_num: - episode.reel_image = ( - reel_sources[0] - if reel_sources - else make_relative_path( - reel_path, - media_root_str, - ) - ) - episode.reel_sources = reel_sources or None - await self._send(Upsert(kind="series", id=item_id, item=series)) + reel_image = ( + reel_sources[0] + if reel_sources + else make_relative_path(reel_path, media_root_path.as_posix()) + ) + self._record_reel_state( + "episode", + media_folder, + season_num, + episode_num, + video_path, + sig, + "done", + ) + await self._send( + EpisodeReel( + id=item_id, + season=season_num, + episode=episode_num, + reel_image=reel_image, + reel_sources=reel_sources or None, + ) + ) await self._send( Task( data=TaskInfo( @@ -719,6 +987,16 @@ class RootScanner: ) ) else: + self._record_reel_state( + "episode", + media_folder, + season_num, + episode_num, + video_path, + sig, + "failed", + error="no reel produced", + ) logger.warning( "Episode reel generation failed: %s %s", series_title, @@ -736,6 +1014,11 @@ class RootScanner: ) self._showreel_queue.task_done() + if self._showreel_queue.empty() and self._reel_state_dirty: + # Persist as soon as the queue drains; scans may be far + # apart and a crash would otherwise lose the records. + self._reel_state_dirty = False + await asyncio.to_thread(self._save_reel_state) except asyncio.CancelledError: logger.info("Showreel worker shutting down for root %s", self.root_id) diff --git a/mediahive/hivescan/showreel.py b/mediahive/hivescan/showreel.py index b36716f..6a63ac1 100644 --- a/mediahive/hivescan/showreel.py +++ b/mediahive/hivescan/showreel.py @@ -7,13 +7,14 @@ and HDR passthrough. import asyncio import contextlib +import json import logging import re import shlex import subprocess import sys from collections import Counter -from dataclasses import dataclass +from dataclasses import asdict, dataclass from pathlib import Path from typing import Any @@ -425,6 +426,85 @@ class MediaProbeInfo: _media_probe_cache: dict[str, MediaProbeInfo] = {} + +# --------------------------------------------------------------------------- +# Persistent probe records +# +# Probe results are keyed by (path, mtime, size) and persisted to +# ``probe-cache.json`` under the root's .mediahive folder so that process +# restarts do not re-run ffmpeg on unchanged files. The in-RAM structures +# are process-global (keyed by absolute path, so sharing across roots is +# safe); each root loads/saves its own file, merging into the same dict. +# --------------------------------------------------------------------------- + +_probe_records: dict[str, dict] = {} +_probe_records_path: Path | None = None +_probe_records_dirty = False + + +def load_probe_records(path: Path) -> None: + """Load persisted probe records from ``path`` (missing file is fine).""" + global _probe_records_path + _probe_records_path = path + try: + data = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return + except OSError, ValueError: + logger.exception("Failed to load probe cache from %s", path) + return + records = data.get("records") + if isinstance(records, dict): + _probe_records.update(records) + logger.info("Loaded probe cache: %d records from %s", len(records), path) + + +def probe_records_dirty() -> bool: + return _probe_records_dirty + + +def save_probe_records() -> None: + """Persist probe records if any were added since the last save.""" + global _probe_records_dirty + if not _probe_records_dirty or _probe_records_path is None: + return + _probe_records_dirty = False + try: + payload = json.dumps({"version": 1, "records": _probe_records}) + tmp = _probe_records_path.with_suffix(".tmp") + tmp.parent.mkdir(parents=True, exist_ok=True) + tmp.write_text(payload, encoding="utf-8") + tmp.replace(_probe_records_path) + except OSError, TypeError, ValueError: + logger.exception("Failed to save probe cache to %s", _probe_records_path) + + +def _record_probe(video_path: str, stat_info, info: MediaProbeInfo) -> None: + global _probe_records_dirty + if stat_info is None: + return # Non-plain paths (bluray:/concat: URIs) are not persisted + _probe_records[video_path] = { + "mtime": int(stat_info.st_mtime), + "size": stat_info.st_size, + "info": asdict(info), + } + _probe_records_dirty = True + + +def _lookup_probe_record(video_path: str, stat_info) -> MediaProbeInfo | None: + rec = _probe_records.get(video_path) + if rec is None or stat_info is None: + return None + if rec.get("mtime") != int(stat_info.st_mtime): + return None + if rec.get("size") != stat_info.st_size: + return None + try: + return MediaProbeInfo(**rec["info"]) + except TypeError, KeyError: + return None + + _duration_re = re.compile(r"Duration:\s*(\d+):(\d+):(\d+(?:\.\d+)?)") _dimension_re = re.compile(r"(\d{2,5})x(\d{2,5})") _dovi_profile_re = re.compile( @@ -449,11 +529,23 @@ async def probe_media_info(video_path: str) -> MediaProbeInfo: if cached is not None: return cached + # Stat once: used both to validate persisted records and to key new ones. + # Non-plain paths (bluray:/concat: URIs) fail stat and stay memory-cached. + stat_info = None + with contextlib.suppress(OSError, ValueError): + stat_info = await AsyncPath(video_path).stat() + + recorded = _lookup_probe_record(video_path, stat_info) + if recorded is not None: + _media_probe_cache[video_path] = recorded + return recorded + info = MediaProbeInfo() cmd = ["ffmpeg", "-hide_banner", "-i", video_path] ffmpeg_result = await _run_ffmpeg(cmd, timeout_seconds=30, allow_nonzero_exit=True) if ffmpeg_result is None: _media_probe_cache[video_path] = info + _record_probe(video_path, stat_info, info) return info stdout, stderr = ffmpeg_result @@ -552,6 +644,7 @@ async def probe_media_info(video_path: str) -> MediaProbeInfo: info.subtitle_languages = subtitle_languages or None _media_probe_cache[video_path] = info + _record_probe(video_path, stat_info, info) return info diff --git a/mediahive/index_store.py b/mediahive/index_store.py index 2d1ca3b..303b49a 100644 --- a/mediahive/index_store.py +++ b/mediahive/index_store.py @@ -18,10 +18,13 @@ from aiopathlib import AsyncPath from fastapi import WebSocket from mediahive.models.data import ( + Episode, IndexSnapshot, Movie, + Season, Series, TaskInfo, + Torrent, ) from mediahive.models.events import Remove, Task, Upsert from mediahive.models.tmdb import Person @@ -147,6 +150,107 @@ class IndexStore: return None return item.info.tmdb_id + @staticmethod + def _newest_from_files(files: dict[str, Torrent]) -> int | None: + timestamps = [t.added_at for t in files.values() if t.added_at] + return max(timestamps) if timestamps else None + + # ------------------------------------------------------------------ + # Merge helpers (partial rescan support) + # ------------------------------------------------------------------ + + def _merge_movie(self, existing: Movie, new: Movie, scanned: set[str]) -> Movie: + """Merge a partially rebuilt movie into the existing entry. + + File entries belonging to torrents in ``scanned`` are replaced by the + new data; everything else is preserved. + """ + files = {k: v for k, v in existing.files.items() if k not in scanned} + files.update(new.files) + return Movie( + title=new.title or existing.title, + info=new.info or existing.info, + year=new.year if new.year is not None else existing.year, + newest=( + self._newest_from_files(files) + or max(filter(None, [existing.newest, new.newest]), default=None) + ), + cover_path=new.cover_path or existing.cover_path, + backdrop_path=new.backdrop_path or existing.backdrop_path, + showreel_images=new.showreel_images or existing.showreel_images, + showreel_source_sets=new.showreel_source_sets + or existing.showreel_source_sets, + files=files, + ) + + def _merge_series( + self, existing: Series, new: Series, scanned: set[str] + ) -> Series: + """Merge a partially rebuilt series into the existing entry. + + File entries belonging to torrents in ``scanned`` are replaced by the + new data; seasons/episodes/files from torrents that were not rescanned + are preserved. Episodes and seasons left without files are dropped. + """ + seasons: dict[int, Season] = {} + for season in existing.seasons: + episodes: dict[int, Episode] = {} + for ep in season.episodes: + files = {k: v for k, v in ep.files.items() if k not in scanned} + if files: + episodes[ep.episode_number] = msgspec.structs.replace( + ep, files=files + ) + if episodes: + seasons[season.season_number] = msgspec.structs.replace( + season, + episodes=list(episodes.values()), + episode_count=len(episodes), + ) + + for season in new.seasons: + current = seasons.get(season.season_number) + if current is None: + seasons[season.season_number] = season + continue + episodes = {ep.episode_number: ep for ep in current.episodes} + for ep in season.episodes: + old = episodes.get(ep.episode_number) + if old is None: + episodes[ep.episode_number] = ep + continue + # Same episode from an unscanned torrent too: union the files, + # prefer fresh metadata/reel info from the new scan. + files = dict(old.files) + files.update(ep.files) + episodes[ep.episode_number] = msgspec.structs.replace( + ep, + files=files, + reel_image=ep.reel_image or old.reel_image, + reel_sources=ep.reel_sources or old.reel_sources, + ) + ordered = [episodes[k] for k in sorted(episodes)] + seasons[season.season_number] = msgspec.structs.replace( + season, + episodes=ordered, + episode_count=len(ordered), + poster_path=season.poster_path or current.poster_path, + ) + + alt_titles = sorted( + set(existing.alternative_titles or []) + | set(new.alternative_titles or []) + ) + return Series( + title=new.title or existing.title, + info=new.info or existing.info, + alternative_titles=alt_titles or None, + newest=max(filter(None, [existing.newest, new.newest]), default=None), + cover_path=new.cover_path or existing.cover_path, + backdrop_path=new.backdrop_path or existing.backdrop_path, + seasons=[seasons[k] for k in sorted(seasons)], + ) + def _rebuild_tmdb_indexes(self) -> None: """Rebuild TMDb id lookup maps from the current in-memory items.""" self._movie_tmdb_ids.clear() @@ -189,22 +293,34 @@ class IndexStore: self._rebuild_tmdb_indexes() def _collapse_movie_tmdb_duplicates(self, tmdb_id: int, keep_id: str) -> None: - """Remove other movie entries that share a TMDb id.""" + """Fold other entries that share a TMDb id into the kept one.""" for item_id, movie in list(self.movies.items()): if item_id == keep_id: continue - if self._get_tmdb_id(movie) == tmdb_id: - self.movies.pop(item_id, None) - self._rebuild_tmdb_indexes() + if self._get_tmdb_id(movie) != tmdb_id: + continue + kept = self.movies.get(keep_id) + if kept is not None: + # Preserve any file versions the duplicate alone carried. + self.movies[keep_id] = self._merge_movie(movie, kept, set()) + self.movies.pop(item_id, None) + if self._movie_tmdb_ids.get(tmdb_id) == item_id: + self._movie_tmdb_ids[tmdb_id] = keep_id def _collapse_series_tmdb_duplicates(self, tmdb_id: int, keep_id: str) -> None: - """Remove other series entries that share a TMDb id.""" + """Fold other entries that share a TMDb id into the kept one.""" for item_id, series in list(self.series.items()): if item_id == keep_id: continue - if self._get_tmdb_id(series) == tmdb_id: - self.series.pop(item_id, None) - self._rebuild_tmdb_indexes() + if self._get_tmdb_id(series) != tmdb_id: + continue + kept = self.series.get(keep_id) + if kept is not None: + # Preserve any seasons/episodes the duplicate alone carried. + self.series[keep_id] = self._merge_series(series, kept, set()) + self.series.pop(item_id, None) + if self._series_tmdb_ids.get(tmdb_id) == item_id: + self._series_tmdb_ids[tmdb_id] = keep_id async def _write_snapshot(self) -> None: """Write current index to disk (called from debounce task).""" @@ -300,8 +416,14 @@ class IndexStore: item_id: str, item: Movie, people: dict[int, Person] | None = None, + scanned: list[str] | None = None, ) -> bool: - """Insert or update a movie. Returns True if it was a real change.""" + """Insert or update a movie. Returns True if it was a real change. + + When ``scanned`` is given, the item is a partial rebuild covering only + those torrent paths; it is merged into the existing entry instead of + replacing it. + """ tmdb_id = self._get_tmdb_id(item) existing_id = self._movie_tmdb_ids.get(tmdb_id) if tmdb_id is not None else None if existing_id is not None and existing_id != item_id: @@ -311,6 +433,10 @@ class IndexStore: if tmdb_id is not None: self._movie_tmdb_ids[tmdb_id] = item_id self._collapse_movie_tmdb_duplicates(tmdb_id, item_id) + existing = self.movies.get(item_id) + + if existing is not None and scanned is not None: + item = self._merge_movie(existing, item, set(scanned)) if existing is not None: if msgspec.json.encode(existing) == msgspec.json.encode(item): @@ -334,8 +460,14 @@ class IndexStore: item_id: str, item: Series, people: dict[int, Person] | None = None, + scanned: list[str] | None = None, ) -> bool: - """Insert or update a series. Returns True if it was a real change.""" + """Insert or update a series. Returns True if it was a real change. + + When ``scanned`` is given, the item is a partial rebuild covering only + those torrent paths; it is merged into the existing entry instead of + replacing it. + """ tmdb_id = self._get_tmdb_id(item) existing_id = ( self._series_tmdb_ids.get(tmdb_id) if tmdb_id is not None else None @@ -347,6 +479,10 @@ class IndexStore: if tmdb_id is not None: self._series_tmdb_ids[tmdb_id] = item_id self._collapse_series_tmdb_duplicates(tmdb_id, item_id) + existing = self.series.get(item_id) + + if existing is not None and scanned is not None: + item = self._merge_series(existing, item, set(scanned)) if existing is not None: if msgspec.json.encode(existing) == msgspec.json.encode(item): @@ -367,22 +503,148 @@ class IndexStore: def remove_movie(self, item_id: str) -> None: """Remove a movie from the index and broadcast.""" - self.movies.pop(item_id, None) - for tmdb_id, mapped_id in list(self._movie_tmdb_ids.items()): - if mapped_id == item_id: - self._movie_tmdb_ids.pop(tmdb_id, None) + movie = self.movies.pop(item_id, None) + if movie is None: + return + tmdb_id = self._get_tmdb_id(movie) + if tmdb_id is not None and self._movie_tmdb_ids.get(tmdb_id) == item_id: + self._movie_tmdb_ids.pop(tmdb_id, None) self._schedule_snapshot() self._broadcast(Remove(kind="movie", id=item_id)) def remove_series(self, item_id: str) -> None: """Remove a series from the index and broadcast.""" - self.series.pop(item_id, None) - for tmdb_id, mapped_id in list(self._series_tmdb_ids.items()): - if mapped_id == item_id: - self._series_tmdb_ids.pop(tmdb_id, None) + series = self.series.pop(item_id, None) + if series is None: + return + tmdb_id = self._get_tmdb_id(series) + if tmdb_id is not None and self._series_tmdb_ids.get(tmdb_id) == item_id: + self._series_tmdb_ids.pop(tmdb_id, None) self._schedule_snapshot() self._broadcast(Remove(kind="series", id=item_id)) + # ------------------------------------------------------------------ + # Scanner-driven maintenance + # ------------------------------------------------------------------ + + def sync_torrent_paths(self, paths: set[str]) -> None: + """Drop file entries whose torrent path no longer exists on disk. + + ``paths`` is the complete set of media-root-relative torrent paths the + scanner found during a fully completed discovery pass. Episodes and + seasons left without files are dropped; items left without any files + are removed entirely. + """ + for item_id, movie in list(self.movies.items()): + kept = {k: v for k, v in movie.files.items() if k in paths} + if len(kept) == len(movie.files): + continue + if not kept: + self.remove_movie(item_id) + continue + updated = msgspec.structs.replace( + movie, files=kept, newest=self._newest_from_files(kept) + ) + self.movies[item_id] = updated + self._schedule_snapshot() + self._broadcast(Upsert(kind="movie", id=item_id, item=updated)) + + for item_id, series in list(self.series.items()): + removed_any = False + new_seasons: list[Season] = [] + for season in series.seasons: + new_episodes: list[Episode] = [] + for ep in season.episodes: + files = {k: v for k, v in ep.files.items() if k in paths} + if len(files) < len(ep.files): + removed_any = True + if files: + new_episodes.append( + msgspec.structs.replace(ep, files=files) + ) + else: + removed_any = True + if not new_episodes: + removed_any = True + continue + if len(new_episodes) < len(season.episodes): + season = msgspec.structs.replace( + season, + episodes=new_episodes, + episode_count=len(new_episodes), + ) + new_seasons.append(season) + if not removed_any: + continue + if not new_seasons: + self.remove_series(item_id) + continue + updated_series = msgspec.structs.replace(series, seasons=new_seasons) + self.series[item_id] = updated_series + self._schedule_snapshot() + self._broadcast(Upsert(kind="series", id=item_id, item=updated_series)) + + def set_movie_showreel( + self, + item_id: str, + showreel_images: list[str] | None, + showreel_source_sets: list[list[str]] | None, + ) -> None: + """Update only the showreel fields of a movie (reel worker callback).""" + movie = self.movies.get(item_id) + if movie is None: + return + updated = msgspec.structs.replace( + movie, + showreel_images=showreel_images, + showreel_source_sets=showreel_source_sets, + ) + if msgspec.json.encode(updated) == msgspec.json.encode(movie): + return + self.movies[item_id] = updated + self._schedule_snapshot() + self._broadcast(Upsert(kind="movie", id=item_id, item=updated)) + + def set_episode_reel( + self, + item_id: str, + season_num: int, + episode_num: int, + reel_image: str | None, + reel_sources: list[str] | None, + ) -> None: + """Update only the reel fields of one episode (reel worker callback).""" + series = self.series.get(item_id) + if series is None: + return + for season in series.seasons: + if season.season_number != season_num: + continue + for ep in season.episodes: + if ep.episode_number != episode_num: + continue + if ep.reel_image == reel_image and ep.reel_sources == reel_sources: + return + new_episodes = [ + msgspec.structs.replace( + e, reel_image=reel_image, reel_sources=reel_sources + ) + if e is ep + else e + for e in season.episodes + ] + new_seasons = [ + msgspec.structs.replace(s, episodes=new_episodes) + if s is season + else s + for s in series.seasons + ] + updated = msgspec.structs.replace(series, seasons=new_seasons) + self.series[item_id] = updated + self._schedule_snapshot() + self._broadcast(Upsert(kind="series", id=item_id, item=updated)) + return + # ------------------------------------------------------------------ # WebSocket management # ------------------------------------------------------------------ diff --git a/mediahive/models/events.py b/mediahive/models/events.py index 76fa578..e36416c 100644 --- a/mediahive/models/events.py +++ b/mediahive/models/events.py @@ -14,12 +14,20 @@ from .tmdb import Person class Upsert(msgspec.Struct, tag="upsert"): - """Single item inserted or updated.""" + """Single item inserted or updated. + + ``scanned`` lists the media-root-relative torrent paths whose content was + (re)scanned to build this item. When present, the store merges the item + into the existing entry instead of replacing it wholesale: only data + belonging to the scanned torrents is replaced. ``None`` means full + replacement (legacy behaviour). + """ kind: str # "movie" or "series" id: str item: Movie | Series people: dict[int, Person] | None = None + scanned: list[str] | None = None class Remove(msgspec.Struct, tag="remove"): @@ -29,6 +37,35 @@ class Remove(msgspec.Struct, tag="remove"): id: str +class Sync(msgspec.Struct, tag="sync"): + """Full set of media-root-relative torrent paths currently on disk. + + Sent by the scanner after a successfully completed discovery pass so the + store can drop entries whose files no longer exist. Internal only — + never forwarded to WebSocket clients. + """ + + paths: list[str] + + +class MovieShowreel(msgspec.Struct, tag="movie-showreel"): + """Reel worker result for a movie (internal, scanner → store).""" + + id: str + showreel_images: list[str] | None = None + showreel_source_sets: list[list[str]] | None = None + + +class EpisodeReel(msgspec.Struct, tag="episode-reel"): + """Reel worker result for one episode (internal, scanner → store).""" + + id: str + season: int + episode: int + reel_image: str | None = None + reel_sources: list[str] | None = None + + class Task(msgspec.Struct, tag="task"): """Task progress broadcast.""" @@ -36,4 +73,4 @@ class Task(msgspec.Struct, tag="task"): # Union of scan events (scanner → server) and WS broadcast messages -ScanEvent = Upsert | Task +ScanEvent = Upsert | Sync | MovieShowreel | EpisodeReel | Task diff --git a/mediahive/root_registry.py b/mediahive/root_registry.py index 302cb4b..10dccc8 100644 --- a/mediahive/root_registry.py +++ b/mediahive/root_registry.py @@ -11,7 +11,14 @@ import msgspec from mediahive.config import load_config, save_config from mediahive.index_store import IndexStore -from mediahive.models.events import ScanEvent, Task, Upsert +from mediahive.models.events import ( + EpisodeReel, + MovieShowreel, + ScanEvent, + Sync, + Task, + Upsert, +) logger = logging.getLogger("mediahive.root_registry") @@ -158,9 +165,27 @@ class RootContext: event = await self._events.get() if isinstance(event, Upsert): if event.kind == "movie": - self.store.upsert_movie(event.id, event.item, event.people) + self.store.upsert_movie( + event.id, event.item, event.people, event.scanned + ) else: - self.store.upsert_series(event.id, event.item, event.people) + self.store.upsert_series( + event.id, event.item, event.people, event.scanned + ) + elif isinstance(event, Sync): + self.store.sync_torrent_paths(set(event.paths)) + elif isinstance(event, MovieShowreel): + self.store.set_movie_showreel( + event.id, event.showreel_images, event.showreel_source_sets + ) + elif isinstance(event, EpisodeReel): + self.store.set_episode_reel( + event.id, + event.season, + event.episode, + event.reel_image, + event.reel_sources, + ) elif isinstance(event, Task): self.store.broadcast_task(event.data) except asyncio.CancelledError: