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

Fix scanner rescan bugs and wasted work

Browse Source 
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.
This commit is contained in:
LeoVasanko
2026-09-04 00:00:53 +00:00
parent d3addadf14
commit 2a39e1f0ea
7 changed files with 1218 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/scanning-review.md
+352
View File
@@ -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 |
mediahive/hivescan/indexer.py
+83 -15
View File
@@ -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
mediahive/hivescan/scanner.py
+342 -59
View File
@@ -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)
mediahive/hivescan/showreel.py
+94 -1
View File
@@ -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
mediahive/index_store.py
+280 -18
View File
@@ -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
# ------------------------------------------------------------------
mediahive/models/events.py
+39 -2
View File
@@ -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
mediahive/root_registry.py
+28 -3
View File
@@ -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: