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

Cleanup

Browse Source
This commit is contained in:
LeoVasanko
2026-02-09 09:10:39 +00:00
parent a3833a0f5c
commit 4347a8d11d
19 changed files with 357 additions and 279 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mediahive/__init__.py
-1
View File
@@ -1,2 +1 @@
"""MediaHive - Media Browser Server"""
mediahive/__main__.py
+5 -1
View File
@@ -9,7 +9,11 @@ DEVMODE = bool(os.getenv("MEDIAHIVE_FRONTEND_URL"))
def main():
parser = argparse.ArgumentParser(description="MediaHive - Media scanning, indexing, and streaming")
parser = argparse.ArgumentParser(
description="MediaHive - Media scanning, indexing, and streaming"
)
# TODO: Accept .mediahive root folder directly from CLI.
# Future: use gitignore-style system (file in .mediahive folder) for path determination.
parser.add_argument(
"media_folder",
nargs="?",
mediahive/hivescan/__init__.py
+5 -59
View File
@@ -1,75 +1,21 @@
"""
Hivescan - Continuous media scanning with live WebSocket updates.
Usage as a module:
python -m mediahive.hivescan /path/to/torrents/*
python -m mediahive.hivescan /path/* --port 9000
Or as a library:
Import from submodules directly:
from mediahive.hivescan.scanner import start, stop
from mediahive.hivescan.scanning import scan_downloads, categorize_downloads
from mediahive.hivescan.indexer import _process_movies, _process_series
from mediahive.hivescan.models import ContentType, ParsedContent
from mediahive.hivescan.tmdb_client import fetch_movie_info, fetch_series_info
"""
# Minimal public API - prefer importing from submodules directly
from mediahive.hivescan.models import ContentType, ContentHash, ParsedContent
from mediahive.hivescan.scanning import (
scan_downloads,
categorize_downloads,
find_playable_file,
find_episode_files,
)
from mediahive.hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root
from mediahive.hivescan.showreel import generate_showreel_images, generate_episode_reel
from mediahive.hivescan import scanner
from mediahive.models.data import (
Episode,
IndexSnapshot,
MediaStats,
Movie,
Season,
Series,
TaskInfo,
Torrent,
)
from mediahive.hivescan.tmdb_client import (
fetch_movie_info,
fetch_series_info,
fetch_season_details,
set_cache_dir,
)
__all__ = [
# Models
"ContentType",
"ContentHash",
"ParsedContent",
# Scanning
"scan_downloads",
"categorize_downloads",
"find_playable_file",
"find_episode_files",
# Struct types
"CastMember",
"Episode",
"IndexSnapshot",
"MediaStats",
"Movie",
"Season",
"Series",
"SimilarMedia",
"TaskInfo",
"Torrent",
"EpisodeInfo",
"Info",
"SeasonInfo",
# Showreel generation
"generate_showreel_images",
"generate_episode_reel",
# TMDb client
"fetch_movie_info",
"fetch_series_info",
"fetch_season_details",
"set_cache_dir",
# Utilities
"DEFAULT_OUTPUT_FOLDER",
"find_common_root",
]
mediahive/hivescan/__main__.py
+22 -14
View File
@@ -5,7 +5,7 @@ import logging
import os
from pathlib import Path
from mediahive.hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root
from mediahive.hivescan.utils import find_common_root
def main():
@@ -24,7 +24,7 @@ The server exposes:
POST /api/scan Trigger a new scan
GET /api/status Current server status
GET /api/index Full index as JSON (HTTP fallback)
"""
""",
)
parser.add_argument(
"paths",
@@ -51,25 +51,30 @@ The server exposes:
args = parser.parse_args()
# Derive the media root so we can set MEDIAHIVE_PATH
all_paths: list[Path] = []
for pattern in args.paths:
expanded = glob.glob(pattern)
if expanded:
all_paths.extend(Path(p) for p in expanded)
else:
all_paths.append(Path(pattern))
# TODO: Take .mediahive root folder from CLI directly.
# Future: use gitignore-style system (file in .mediahive folder) for path determination.
# Derive media_root only if no explicit output-dir is given
if args.output_dir:
media_root = Path(args.output_dir).parent
media_root = Path(args.output_dir).parent.resolve()
else:
# Expand globs once to find common root
all_paths: list[Path] = []
for pattern in args.paths:
expanded = glob.glob(pattern)
if expanded:
all_paths.extend(Path(p) for p in expanded)
else:
all_paths.append(Path(pattern))
media_root = asyncio.run(find_common_root(all_paths))
if media_root is None:
print("Error: Cannot determine common root; use -o to set output directory")
exit(1)
media_root = media_root.resolve()
# Configure environment for the mediahive server + scanner
os.environ["MEDIAHIVE_PATH"] = str(media_root.resolve())
# Configure environment - scanner will re-expand patterns from HIVESCAN_PATHS
os.environ["MEDIAHIVE_PATH"] = str(media_root)
os.environ["HIVESCAN_PATHS"] = os.pathsep.join(args.paths)
if args.output_dir:
os.environ["HIVESCAN_OUTPUT"] = args.output_dir
@@ -81,7 +86,10 @@ The server exposes:
)
import uvicorn
uvicorn.run("mediahive.server:app", host=args.host, port=args.port, log_level="info")
uvicorn.run(
"mediahive.server:app", host=args.host, port=args.port, log_level="info"
)
if __name__ == "__main__":
mediahive/hivescan/images.py
-1
View File
@@ -113,4 +113,3 @@ async def download_season_poster(
await AsyncPath(media_folder).mkdir(parents=True, exist_ok=True)
url = f"{TMDB_IMAGE_BASE}/{DEFAULT_POSTER_SIZE}{poster_path}"
return await _download_image(url, output_path, f"season {season_num} poster")
mediahive/hivescan/indexer.py
+47 -26
View File
@@ -17,6 +17,7 @@ from mediahive.models.data import (
Series,
Torrent,
)
from mediahive.models.tmdb import EpisodeInfo, Info, SeasonInfo
from mediahive.hivescan.tmdb_client import (
fetch_movie_info,
fetch_series_info,
@@ -24,7 +25,11 @@ from mediahive.hivescan.tmdb_client import (
)
from mediahive.hivescan.models import ContentType, ParsedContent
from mediahive.hivescan.scanning import find_cover_image, find_episode_files, find_playable_file
from mediahive.hivescan.scanning import (
find_cover_image,
find_episode_files,
find_playable_file,
)
from mediahive.hivescan.images import (
download_cover_image,
download_backdrop_image,
@@ -119,7 +124,9 @@ async def _collect_episode_files(
)
if not already_added:
if item.content_hash and item.content_hash.size == 0:
item.content_hash.size = await get_directory_size(item.content_hash.path)
item.content_hash.size = await get_directory_size(
item.content_hash.path
)
size = item.content_hash.size if item.content_hash else 0
all_episode_files[key].append(
{
@@ -291,9 +298,9 @@ async def _process_movies(
Yields (Movie, showreel_task_or_None) for each movie as it is processed.
"""
# In-memory cache for TMDb lookups
movie_tmdb_cache: Dict[str, Optional[TMDbInfo]] = {}
movie_tmdb_cache: Dict[str, Optional[Info]] = {}
async def get_movie_tmdb(title: str, year: Optional[int]) -> Optional[TMDbInfo]:
async def get_movie_tmdb(title: str, year: Optional[int]) -> Optional[Info]:
cache_key = f"{title.lower()}:{year}"
if cache_key in movie_tmdb_cache:
return movie_tmdb_cache[cache_key]
@@ -311,7 +318,9 @@ async def _process_movies(
valid_movies.append(item)
skipped = len(categories[ContentType.MOVIE]) - len(valid_movies)
if skipped > 0:
logger.debug(" Skipped %d movie torrents with no playable video files", skipped)
logger.debug(
" Skipped %d movie torrents with no playable video files", skipped
)
# Group by title+year
movie_groups: Dict[str, List[ParsedContent]] = {}
@@ -402,13 +411,16 @@ async def _process_movies(
# Queue showreel generation
showreel_paths = []
showreel_task = None
if generate_showreels and versions:
if generate_showreels and torrents:
# Find the best version for showreel (highest quality)
best_relpath = max(versions.keys(), key=lambda k: (
RESOLUTION_PRIORITY.get(versions[k].resolution or "", 0),
versions[k].size or 0,
k
))
best_relpath = max(
torrents.keys(),
key=lambda k: (
RESOLUTION_PRIORITY.get(torrents[k].resolution or "", 0),
torrents[k].size or 0,
k,
),
)
best_version = torrents[best_relpath]
if best_version.playable_file:
abs_playable = (
@@ -455,7 +467,9 @@ async def _process_movies(
item_id = hashlib.md5(f"movie:{title}:{year}".encode()).hexdigest()[:12]
cover_path = (
await find_cover_image(title, year, "movie", cover_dir) if fetch_covers else None
await find_cover_image(title, year, "movie", cover_dir)
if fetch_covers
else None
)
torrents = {}
@@ -470,21 +484,24 @@ async def _process_movies(
showreel_task = None
if generate_showreels and torrents:
# Find the best version for showreel (highest quality)
best_relpath = max(torrents.keys(), key=lambda k: (
RESOLUTION_PRIORITY.get(torrents[k].resolution or "", 0),
torrents[k].size or 0,
k
))
best_relpath = max(
torrents.keys(),
key=lambda k: (
RESOLUTION_PRIORITY.get(torrents[k].resolution or "", 0),
torrents[k].size or 0,
k,
),
)
best_version = torrents[best_relpath]
if best_version.playable_file and not best_version.playable_file.endswith(".bdmv"):
if best_version.playable_file and not best_version.playable_file.endswith(
".bdmv"
):
abs_playable = (
str(Path(media_root) / best_version.playable_file)
if media_root
else best_version.playable_file
)
media_folder = get_media_folder_path(
title, year, "movie", cover_dir
)
media_folder = get_media_folder_path(title, year, "movie", cover_dir)
showreel_paths = get_expected_showreel_paths(
media_folder,
media_root=Path(media_root) if media_root else None,
@@ -519,10 +536,10 @@ async def _process_series(
Yields (Series, episode_reel_tasks) for each series as it is processed.
"""
# In-memory cache for TMDb lookups
series_tmdb_cache: Dict[str, Optional[TMDbInfo]] = {}
season_cache: Dict[Tuple[int, int], Optional[TMDbSeasonInfo]] = {}
series_tmdb_cache: Dict[str, Optional[Info]] = {}
season_cache: Dict[Tuple[int, int], Optional[SeasonInfo]] = {}
async def get_series_tmdb(title: str) -> Optional[TMDbInfo]:
async def get_series_tmdb(title: str) -> Optional[Info]:
cache_key = title.lower()
if cache_key in series_tmdb_cache:
return series_tmdb_cache[cache_key]
@@ -605,7 +622,9 @@ async def _process_series(
# Find/download cover
cover_path = None
if fetch_covers:
cover_path = await find_cover_image(display_title, None, "series", cover_dir)
cover_path = await find_cover_image(
display_title, None, "series", cover_dir
)
if not cover_path:
for tt in torrent_titles:
cover_path = await find_cover_image(tt, None, "series", cover_dir)
@@ -668,7 +687,9 @@ async def _process_series(
series_id = hashlib.md5(f"series:{title}".encode()).hexdigest()[:12]
cover_path = (
await find_cover_image(title, None, "series", cover_dir) if fetch_covers else None
await find_cover_image(title, None, "series", cover_dir)
if fetch_covers
else None
)
series_folder = get_media_folder_path(title, None, "series", cover_dir)
mediahive/hivescan/models.py
-1
View File
@@ -51,4 +51,3 @@ class ParsedContent:
is_directory: bool = False
raw_parsed: dict = field(default_factory=dict)
content_hash: Optional[ContentHash] = None
mediahive/hivescan/parsing.py
-1
View File
@@ -78,4 +78,3 @@ def parse_episode_from_filename(filename: str) -> Optional[Tuple[int, int]]:
return int(match.group(1)), int(match.group(2))
return None
mediahive/hivescan/scanner.py
+157 -56
View File
@@ -3,8 +3,8 @@ Scan orchestration — background tasks for continuous media scanning.
All scanning logic lives here in hivescan. Communication with the mediahive
server happens exclusively through an async ``send`` callable that pushes
:class:`~mediahive.models.protocol.ScanEvent` messages (``EvUpsert`` /
``EvTask``) onto an :class:`asyncio.Queue` owned by the caller.
:class:`~mediahive.models.events.ScanEvent` messages (``Upsert`` /
``Task``) onto an :class:`asyncio.Queue` owned by the caller.
"""
import asyncio
@@ -29,9 +29,13 @@ from mediahive.hivescan.showreel import (
movie_showreels_exist,
)
from mediahive.hivescan.tmdb_client import set_cache_dir
from mediahive.hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root, make_relative_path
from mediahive.hivescan.utils import (
DEFAULT_OUTPUT_FOLDER,
find_common_root,
make_relative_path,
)
from mediahive.models.data import Movie, Series, TaskInfo
from mediahive.models.protocol import EvTask, EvUpsert, ScanEvent
from mediahive.models.events import ScanEvent, Task, Upsert
logger = logging.getLogger("hivescan.scanner")
@@ -106,7 +110,8 @@ async def start(send: Send) -> None:
logger.info(
"Scanner started — %d scan paths, output=%s",
len(_scan_paths), _output_dir,
len(_scan_paths),
_output_dir,
)
_showreel_worker_task = asyncio.create_task(_showreel_worker())
@@ -192,7 +197,7 @@ async def _run_scan(override_paths: Optional[List[str]] = None):
Full scan pipeline:
1. Walk filesystem, parse torrents
2. Categorise → movies / series
3. Iterate async generators, send each item as EvUpsert
3. Iterate async generators, send each item as Upsert
4. Queue showreel tasks
"""
task_id = f"scan-{uuid.uuid4().hex[:8]}"
@@ -204,9 +209,16 @@ async def _run_scan(override_paths: Optional[List[str]] = None):
if downloads:
logger.info("Scan started (%s)", task_id)
await _send(EvTask(data=TaskInfo(
id=task_id, status="running", progress=0, detail="Scanning filesystem...",
)))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="running",
progress=0,
detail="Scanning filesystem...",
)
)
)
logger.info("Found %d items to process", len(downloads))
categories = categorize_downloads(downloads)
@@ -214,57 +226,110 @@ async def _run_scan(override_paths: Optional[List[str]] = None):
processed = 0
# Process movies
await _send(EvTask(data=TaskInfo(
id=task_id, status="running", progress=0, detail="Processing movies...",
)))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="running",
progress=0,
detail="Processing movies...",
)
)
)
async for movie, showreel_task in _process_movies(
categories, _output_dir,
fetch_covers=True, generate_showreels=True, media_root=media_root_str,
categories,
_output_dir,
fetch_covers=True,
generate_showreels=True,
media_root=media_root_str,
):
await _send(EvUpsert(kind="movie", item=movie))
await _send(Upsert(kind="movie", item=movie))
if showreel_task:
await _showreel_queue.put(("movie", showreel_task, movie))
processed += 1
progress = round(processed / total, 3) if total else 1
await _send(EvTask(data=TaskInfo(
id=task_id, status="running", progress=progress, detail=movie.title,
)))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="running",
progress=progress,
detail=movie.title,
)
)
)
# Process series
await _send(EvTask(data=TaskInfo(
id=task_id, status="running",
progress=processed / total if total else 0.5,
detail="Processing series...",
)))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="running",
progress=processed / total if total else 0.5,
detail="Processing series...",
)
)
)
async for series, ep_reel_tasks in _process_series(
categories, _output_dir,
fetch_covers=True, generate_showreels=True, media_root=media_root_str,
categories,
_output_dir,
fetch_covers=True,
generate_showreels=True,
media_root=media_root_str,
):
await _send(EvUpsert(kind="series", item=series))
await _send(Upsert(kind="series", item=series))
for task in ep_reel_tasks:
await _showreel_queue.put(("episode", task, series))
processed += 1
progress = round(processed / total, 3) if total else 1
await _send(EvTask(data=TaskInfo(
id=task_id, status="running", progress=progress, detail=series.title,
)))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="running",
progress=progress,
detail=series.title,
)
)
)
await _send(EvTask(data=TaskInfo(
id=task_id, status="completed", progress=1, detail="Scan complete",
)))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="completed",
progress=1,
detail="Scan complete",
)
)
)
if downloads:
logger.info("Scan complete (%s)", task_id)
except asyncio.CancelledError:
await _send(EvTask(data=TaskInfo(
id=task_id, status="cancelled", progress=0, detail="Scan cancelled",
)))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="cancelled",
progress=0,
detail="Scan cancelled",
)
)
)
logger.info("Scan cancelled (%s)", task_id)
except Exception:
logger.exception("Scan failed (%s)", task_id)
await _send(EvTask(data=TaskInfo(
id=task_id, status="error", progress=0, detail="Scan error",
)))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="error",
progress=0,
detail="Scan error",
)
)
)
# ---------------------------------------------------------------------------
@@ -289,29 +354,57 @@ async def _showreel_worker():
if await movie_showreels_exist(media_folder):
_showreel_queue.task_done()
continue
await _send(EvTask(data=TaskInfo(
id=task_id, status="running", progress=0, detail=f"Showreel: {title}",
)))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="running",
progress=0,
detail=f"Showreel: {title}",
)
)
)
await generate_showreel_images(video_path, media_folder, title=title)
paths = get_expected_showreel_paths(media_folder, media_root=media_root_path)
paths = get_expected_showreel_paths(
media_folder, media_root=media_root_path
)
movie.showreel_images = paths if paths else None
await _send(EvUpsert(kind="movie", item=movie))
await _send(EvTask(data=TaskInfo(
id=task_id, status="completed", progress=1, detail=f"Showreel: {title}",
)))
await _send(Upsert(kind="movie", item=movie))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="completed",
progress=1,
detail=f"Showreel: {title}",
)
)
)
elif kind == "episode":
series: Series = item
video_path, media_folder, season_num, episode_num, series_title = task_data
video_path, media_folder, season_num, episode_num, series_title = (
task_data
)
if await episode_reel_exists(media_folder, season_num, episode_num):
_showreel_queue.task_done()
continue
ep_code = f"S{season_num:02d}E{episode_num:02d}"
await _send(EvTask(data=TaskInfo(
id=task_id, status="running", progress=0, detail=f"Reel: {series_title} {ep_code}",
)))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="running",
progress=0,
detail=f"Reel: {series_title} {ep_code}",
)
)
)
reel_path = await generate_episode_reel(
video_path, media_folder, season_num, episode_num,
video_path,
media_folder,
season_num,
episode_num,
)
if reel_path:
for season in series.seasons:
@@ -319,12 +412,20 @@ async def _showreel_worker():
for episode in season.episodes:
if episode.episode_number == episode_num:
episode.reel_image = make_relative_path(
reel_path, media_root_str,
reel_path,
media_root_str,
)
await _send(EvUpsert(kind="series", item=series))
await _send(EvTask(data=TaskInfo(
id=task_id, status="completed", progress=1, detail=f"Reel: {series_title} {ep_code}",
)))
await _send(Upsert(kind="series", item=series))
await _send(
Task(
data=TaskInfo(
id=task_id,
status="completed",
progress=1,
detail=f"Reel: {series_title} {ep_code}",
)
)
)
_showreel_queue.task_done()
mediahive/hivescan/scanning.py
+6 -5
View File
@@ -79,7 +79,9 @@ def categorize_downloads(
return categories
async def find_episode_files(path: Path) -> Dict[Tuple[int, int], List[Tuple[str, int]]]:
async def find_episode_files(
path: Path,
) -> Dict[Tuple[int, int], List[Tuple[str, int]]]:
"""
Find all episode video files in a directory.
@@ -115,7 +117,7 @@ async def find_episode_files(path: Path) -> Dict[Tuple[int, int], List[Tuple[str
if ep_info not in episodes:
episodes[ep_info] = []
episodes[ep_info].append((str(f), (await af.stat()).st_size))
except (OSError, PermissionError):
except OSError, PermissionError:
pass
_episode_files_cache[cache_key] = episodes
@@ -159,7 +161,7 @@ async def find_playable_file(path: Path) -> Optional[str]:
result = str(nested_bdmv)
_playable_file_cache[cache_key] = result
return result
except (OSError, PermissionError):
except OSError, PermissionError:
pass
# Find largest video file
@@ -171,7 +173,7 @@ async def find_playable_file(path: Path) -> Optional[str]:
if "sample" in Path(f).name.lower():
continue
video_files.append((str(f), (await af.stat()).st_size))
except (OSError, PermissionError):
except OSError, PermissionError:
pass
if not video_files:
@@ -205,4 +207,3 @@ async def find_cover_image(
return str(legacy_path)
return None
mediahive/hivescan/showreel.py
+6 -3
View File
@@ -90,9 +90,13 @@ async def movie_showreels_exist(
return True
async def episode_reel_exists(media_folder: Path, season_num: int, episode_num: int) -> bool:
async def episode_reel_exists(
media_folder: Path, season_num: int, episode_num: int
) -> bool:
"""Check if an episode reel file already exists."""
return await AsyncPath(media_folder / f"S{season_num:02d}E{episode_num:02d}.webm").exists()
return await AsyncPath(
media_folder / f"S{season_num:02d}E{episode_num:02d}.webm"
).exists()
def get_bluray_uri(video_path: str) -> Optional[str]:
@@ -811,4 +815,3 @@ async def generate_episode_reel(
episode_code = f"S{season_num:02d}E{episode_num:02d}"
logger.error("Error generating episode reel for %s: %s", episode_code, e)
return None
mediahive/hivescan/tmdb_client.py
+10 -10
View File
@@ -16,9 +16,11 @@ import httpx
from aiopathlib import AsyncPath
from mediahive.models.tmdb import (
CastMember,
EpisodeInfo,
Info,
SeasonInfo,
SimilarMedia,
)
# TMDb API configuration
@@ -163,7 +165,7 @@ async def fetch_series_details(series_id: int) -> Optional[Dict]:
async def fetch_season_details(
series_id: int, season_number: int
) -> Optional[TMDbSeasonInfo]:
) -> Optional[SeasonInfo]:
"""
Fetch detailed season info including all episodes.
@@ -204,7 +206,7 @@ async def fetch_season_details(
)
episodes.append(episode)
return TMDbSeasonInfo(
return SeasonInfo(
season_number=data.get("season_number", season_number),
name=data.get("name"),
overview=data.get("overview"),
@@ -361,9 +363,7 @@ async def _search_movie_with_fallbacks(
return None
async def fetch_movie_info(
title: str, year: Optional[int] = None
) -> Optional[TMDbInfo]:
async def fetch_movie_info(title: str, year: Optional[int] = None) -> Optional[Info]:
"""Fetch comprehensive movie info from TMDb."""
data = await _search_movie_with_fallbacks(title, year)
@@ -377,7 +377,7 @@ async def fetch_movie_info(
details = await fetch_movie_details(movie_id)
if not details:
# Fall back to basic info from search
return TMDbInfo(
return Info(
tmdb_id=movie_id,
title=result.get("title"),
original_title=result.get("original_title"),
@@ -434,7 +434,7 @@ async def fetch_movie_info(
for s in similar_data
]
return TMDbInfo(
return Info(
tmdb_id=movie_id,
title=details.get("title"),
original_title=details.get("original_title"),
@@ -479,7 +479,7 @@ async def _search_series_with_fallbacks(title: str) -> Optional[Dict]:
return None
async def fetch_series_info(title: str) -> Optional[TMDbInfo]:
async def fetch_series_info(title: str) -> Optional[Info]:
"""Fetch comprehensive TV series info from TMDb."""
data = await _search_series_with_fallbacks(title)
@@ -493,7 +493,7 @@ async def fetch_series_info(title: str) -> Optional[TMDbInfo]:
details = await fetch_series_details(series_id)
if not details:
# Fall back to basic info from search
return TMDbInfo(
return Info(
tmdb_id=series_id,
title=result.get("name"),
original_title=result.get("original_name"),
@@ -539,7 +539,7 @@ async def fetch_series_info(title: str) -> Optional[TMDbInfo]:
# Get first air date
first_air_date = details.get("first_air_date")
return TMDbInfo(
return Info(
tmdb_id=series_id,
title=details.get("name"),
original_title=details.get("original_name"),
mediahive/hivescan/utils.py
+11 -6
View File
@@ -1,6 +1,5 @@
"""Utility functions for paths, sizes, and timestamps."""
import os
import time
from pathlib import Path
from typing import Optional, List
@@ -40,7 +39,7 @@ async def get_added_timestamp(path: Path) -> Optional[int]:
ap = AsyncPath(path)
try:
stat_info = await ap.stat()
except (OSError, PermissionError):
except OSError, PermissionError:
return None
if await ap.is_dir():
@@ -65,7 +64,7 @@ async def get_directory_size(path: Path) -> int:
for item in ap.rglob("*"):
if await AsyncPath(item).is_file():
total += (await AsyncPath(item).stat()).st_size
except (OSError, PermissionError):
except OSError, PermissionError:
pass
return total
@@ -99,7 +98,10 @@ async def find_common_root(paths: List[Path]) -> Optional[Path]:
for p in resolved:
# Find the first existing parent to get device info
check_path = p
while not await AsyncPath(check_path).exists() and check_path.parent != check_path:
while (
not await AsyncPath(check_path).exists()
and check_path.parent != check_path
):
check_path = check_path.parent
if await AsyncPath(check_path).exists():
devices.add((await AsyncPath(check_path).stat()).st_dev)
@@ -113,7 +115,11 @@ async def find_common_root(paths: List[Path]) -> Optional[Path]:
# Find common path prefix
if len(resolved) == 1:
# Single path - use its parent as root
return resolved[0].parent if await AsyncPath(resolved[0]).is_file() else resolved[0]
return (
resolved[0].parent
if await AsyncPath(resolved[0]).is_file()
else resolved[0]
)
# Get parts of each path
all_parts = [p.parts for p in resolved]
@@ -194,4 +200,3 @@ def sort_by_quality(items: list, reverse: bool = True) -> None:
),
reverse=reverse,
)
mediahive/index_store.py
+26 -41
View File
@@ -25,12 +25,10 @@ from mediahive.models.data import (
Series,
TaskInfo,
)
from mediahive.models.events import Remove, Task, Upsert
from mediahive.models.protocol import (
WsInit,
WsInitData,
WsRemove,
WsTask,
WsUpsert,
)
logger = logging.getLogger("mediahive.index_store")
@@ -40,7 +38,15 @@ SNAPSHOT_DEBOUNCE = 5.0
class IndexStore:
"""In-memory media index with WS broadcast and disk snapshots."""
"""In-memory media index with WS broadcast and disk snapshots.
# TODO: Decouple WS transport from data storage
# Consider splitting WS broadcasting into a separate Broadcaster class that:
# - Owns per-connection queues instead of direct WebSocket references
# - Each WS connection gets its own asyncio.Queue for messages
# - Scanner pushes events to all queues; each WS reader drains its own queue
# This would make IndexStore testable without FastAPI's WebSocket.
"""
def __init__(self, snapshot_path: Path, media_root: Optional[str] = None):
self.snapshot_path = snapshot_path
@@ -68,9 +74,7 @@ class IndexStore:
logger.info("No snapshot found at %s, starting fresh", self.snapshot_path)
return
try:
data = msgspec.json.decode(
await ap.read_bytes(), type=IndexSnapshot
)
data = msgspec.json.decode(await ap.read_bytes(), type=IndexSnapshot)
for m in data.movies:
self.movies[m.id] = m
for s in data.series:
@@ -85,32 +89,13 @@ class IndexStore:
async def _write_snapshot(self) -> None:
"""Write current index to disk (called from debounce task)."""
movies_list = sorted(
self.movies.values(), key=lambda x: (x.title.lower(), x.year or 0)
)
series_list = sorted(self.series.values(), key=lambda x: x.title.lower())
total_movie_versions = sum(len(m.torrents) for m in movies_list)
total_series_episodes = sum(
sum(len(season.episodes) for season in s.seasons) for s in series_list
)
snapshot = IndexSnapshot(
generated_at=datetime.now().isoformat(),
media_root=self.media_root,
stats=MediaStats(
total_movies=len(movies_list),
total_movie_versions=total_movie_versions,
total_series=len(series_list),
total_series_episodes=total_series_episodes,
),
movies=movies_list,
series=series_list,
)
snapshot = self._build_snapshot()
await AsyncPath(self.snapshot_path.parent).mkdir(parents=True, exist_ok=True)
tmp = self.snapshot_path.with_suffix(".tmp")
await AsyncPath(tmp).write_bytes(msgspec.json.format(msgspec.json.encode(snapshot), indent=2))
await AsyncPath(tmp).write_bytes(
msgspec.json.format(msgspec.json.encode(snapshot), indent=2)
)
# os.replace is atomic and overwrites on all platforms (unlike rename on Windows)
await asyncio.to_thread(os.replace, tmp, self.snapshot_path)
logger.debug("Snapshot written to %s", self.snapshot_path)
@@ -153,7 +138,7 @@ class IndexStore:
return False
self.movies[item.id] = item
self._schedule_snapshot()
self._broadcast(WsUpsert(kind="movie", item=item))
self._broadcast(Upsert(kind="movie", item=item))
return True
def upsert_series(self, item: Series) -> bool:
@@ -164,20 +149,20 @@ class IndexStore:
return False
self.series[item.id] = item
self._schedule_snapshot()
self._broadcast(WsUpsert(kind="series", item=item))
self._broadcast(Upsert(kind="series", item=item))
return True
def remove_movie(self, item_id: str) -> None:
"""Remove a movie from the index and broadcast."""
self.movies.pop(item_id, None)
self._schedule_snapshot()
self._broadcast(WsRemove(kind="movie", id=item_id))
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)
self._schedule_snapshot()
self._broadcast(WsRemove(kind="series", id=item_id))
self._broadcast(Remove(kind="series", id=item_id))
# ------------------------------------------------------------------
# WebSocket management
@@ -222,18 +207,14 @@ class IndexStore:
def broadcast_task(self, task_info: TaskInfo) -> None:
"""Broadcast a task progress message to all WS clients."""
self._broadcast(WsTask(data=task_info))
def broadcast(self, msg: object) -> None:
"""Broadcast an already-encoded message to all WS clients."""
self._broadcast(msg)
self._broadcast(Task(data=task_info))
# ------------------------------------------------------------------
# Read helpers
# ------------------------------------------------------------------
def get_full_index(self) -> IndexSnapshot:
"""Return the full index as an IndexSnapshot."""
def _build_snapshot(self) -> IndexSnapshot:
"""Build a sorted IndexSnapshot with computed stats."""
movies_list = sorted(
self.movies.values(), key=lambda x: (x.title.lower(), x.year or 0)
)
@@ -256,3 +237,7 @@ class IndexStore:
movies=movies_list,
series=series_list,
)
def get_full_index(self) -> IndexSnapshot:
"""Return the full index as an IndexSnapshot."""
return self._build_snapshot()
mediahive/models/data.py
+1 -1
View File
@@ -8,7 +8,7 @@ from __future__ import annotations
import msgspec
from .tmdb import CastMember, EpisodeInfo, Info, SimilarMedia
from .tmdb import Info
# ---------------------------------------------------------------------------
mediahive/models/events.py
+37
View File
@@ -0,0 +1,37 @@
"""
Event types shared between scanner and WebSocket.
These types are used as:
- Internal scan events (scanner → server queue)
- WebSocket messages (server → clients)
"""
from __future__ import annotations
import msgspec
from .data import Movie, Series, TaskInfo
class Upsert(msgspec.Struct, tag="upsert"):
"""Single item inserted or updated."""
kind: str # "movie" or "series"
item: Movie | Series
class Remove(msgspec.Struct, tag="remove"):
"""Single item removed."""
kind: str
id: str
class Task(msgspec.Struct, tag="task"):
"""Task progress broadcast."""
data: TaskInfo
# Union of scan events (scanner → server) and WS broadcast messages
ScanEvent = Upsert | Task
mediahive/models/protocol.py
+14 -43
View File
@@ -9,7 +9,8 @@ from __future__ import annotations
import msgspec
from fastapi.responses import Response
from .data import Movie, Series, TaskInfo
from .data import Movie, Series
from .events import Remove, ScanEvent, Task, Upsert
# ---------------------------------------------------------------------------
@@ -30,49 +31,20 @@ class WsInit(msgspec.Struct, tag="init"):
data: WsInitData
class WsUpsert(msgspec.Struct, tag="upsert"):
"""Single item inserted or updated."""
kind: str
item: Movie | Series
class WsRemove(msgspec.Struct, tag="remove"):
"""Single item removed."""
kind: str
id: str
class WsTask(msgspec.Struct, tag="task"):
"""Task progress broadcast."""
data: TaskInfo
# Union of all outbound WS messages (for documentation / future decoding)
WsMessage = WsInit | WsUpsert | WsRemove | WsTask
WsMessage = WsInit | Upsert | Remove | Task
# ---------------------------------------------------------------------------
# Scan events (scanner → server, via async queue)
# ---------------------------------------------------------------------------
class EvUpsert(msgspec.Struct, tag="upsert"):
"""Scanner produced or updated a media item."""
kind: str # "movie" or "series"
item: Movie | Series
class EvTask(msgspec.Struct, tag="task"):
"""Scanner progress update."""
data: TaskInfo
ScanEvent = EvUpsert | EvTask
# Re-export unified types for backward compatibility
__all__ = [
"Remove",
"ScanEvent",
"Task",
"Upsert",
"WsInit",
"WsInitData",
"WsMessage",
]
# ---------------------------------------------------------------------------
@@ -118,5 +90,4 @@ class MsgspecResponse(Response):
media_type = "application/json; charset=utf-8"
def render(self, content: object) -> bytes:
return msgspec.json.encode(content)</content>
<parameter name="filePath">c:\mediahive\mediahive\models\protocol.py
return msgspec.json.encode(content)
mediahive/models/tmdb.py
+1 -2
View File
@@ -86,5 +86,4 @@ class Info(msgspec.Struct):
creators: list[str] | None = None
number_of_seasons: int | None = None
number_of_episodes: int | None = None
networks: list[str] | None = None</content>
<parameter name="filePath">c:\mediahive\mediahive\models\tmdb.py
networks: list[str] | None = None
mediahive/server.py
+9 -8
View File
@@ -22,16 +22,13 @@ from fastapi.responses import FileResponse, StreamingResponse
from fastapi_vue import Frontend
from mediahive.index_store import IndexStore
from mediahive.models.events import ScanEvent, Task, Upsert
from mediahive.models.protocol import (
EvTask,
EvUpsert,
MsgspecResponse,
PlayMediaRequest,
OpenFolderRequest,
ScanEvent,
ScanRequest,
StatusResponse,
WsTask,
)
from mediahive.__main__ import DEVMODE
@@ -65,12 +62,12 @@ async def _consume_scan_events() -> None:
while True:
try:
event = await _scan_events.get()
if isinstance(event, EvUpsert):
if isinstance(event, Upsert):
if event.kind == "movie":
store.upsert_movie(event.item)
else:
store.upsert_series(event.item)
elif isinstance(event, EvTask):
elif isinstance(event, Task):
store.broadcast_task(event.data)
except asyncio.CancelledError:
return
@@ -94,12 +91,16 @@ async def lifespan(app: FastAPI):
await store.load_snapshot()
logger.info(
"Index store ready: %d movies, %d series",
len(store.movies), len(store.series),
len(store.movies),
len(store.series),
)
# If scan paths are configured, start the scanner subsystem
if os.environ.get("HIVESCAN_PATHS"):
from mediahive.hivescan.scanner import start as start_scanner, stop as stop_scanner
from mediahive.hivescan.scanner import (
start as start_scanner,
stop as stop_scanner,
)
_consumer_task = asyncio.create_task(_consume_scan_events())
await start_scanner(_send_event)