diff --git a/mediahive/__init__.py b/mediahive/__init__.py index 4a3914f..6ea254b 100644 --- a/mediahive/__init__.py +++ b/mediahive/__init__.py @@ -1,2 +1 @@ """MediaHive - Media Browser Server""" - diff --git a/mediahive/__main__.py b/mediahive/__main__.py index bad8c8a..b15ff84 100644 --- a/mediahive/__main__.py +++ b/mediahive/__main__.py @@ -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="?", diff --git a/mediahive/hivescan/__init__.py b/mediahive/hivescan/__init__.py index 4303863..ed05e1f 100644 --- a/mediahive/hivescan/__init__.py +++ b/mediahive/hivescan/__init__.py @@ -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", ] diff --git a/mediahive/hivescan/__main__.py b/mediahive/hivescan/__main__.py index c6aa03c..ebf0acc 100644 --- a/mediahive/hivescan/__main__.py +++ b/mediahive/hivescan/__main__.py @@ -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__": diff --git a/mediahive/hivescan/images.py b/mediahive/hivescan/images.py index 81f8ffa..a8cc5cd 100644 --- a/mediahive/hivescan/images.py +++ b/mediahive/hivescan/images.py @@ -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") - diff --git a/mediahive/hivescan/indexer.py b/mediahive/hivescan/indexer.py index edb5327..3b9febe 100644 --- a/mediahive/hivescan/indexer.py +++ b/mediahive/hivescan/indexer.py @@ -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) diff --git a/mediahive/hivescan/models.py b/mediahive/hivescan/models.py index b313c84..ced825c 100644 --- a/mediahive/hivescan/models.py +++ b/mediahive/hivescan/models.py @@ -51,4 +51,3 @@ class ParsedContent: is_directory: bool = False raw_parsed: dict = field(default_factory=dict) content_hash: Optional[ContentHash] = None - diff --git a/mediahive/hivescan/parsing.py b/mediahive/hivescan/parsing.py index 9fcec83..817f86a 100644 --- a/mediahive/hivescan/parsing.py +++ b/mediahive/hivescan/parsing.py @@ -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 - diff --git a/mediahive/hivescan/scanner.py b/mediahive/hivescan/scanner.py index fe44797..16f771e 100644 --- a/mediahive/hivescan/scanner.py +++ b/mediahive/hivescan/scanner.py @@ -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() diff --git a/mediahive/hivescan/scanning.py b/mediahive/hivescan/scanning.py index 7979f9f..45c83af 100644 --- a/mediahive/hivescan/scanning.py +++ b/mediahive/hivescan/scanning.py @@ -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 - diff --git a/mediahive/hivescan/showreel.py b/mediahive/hivescan/showreel.py index 871570c..f19fec9 100644 --- a/mediahive/hivescan/showreel.py +++ b/mediahive/hivescan/showreel.py @@ -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 - diff --git a/mediahive/hivescan/tmdb_client.py b/mediahive/hivescan/tmdb_client.py index ceb589b..9497dae 100644 --- a/mediahive/hivescan/tmdb_client.py +++ b/mediahive/hivescan/tmdb_client.py @@ -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"), diff --git a/mediahive/hivescan/utils.py b/mediahive/hivescan/utils.py index b668da4..c90c90e 100644 --- a/mediahive/hivescan/utils.py +++ b/mediahive/hivescan/utils.py @@ -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, ) - diff --git a/mediahive/index_store.py b/mediahive/index_store.py index b207bf0..d69e21e 100644 --- a/mediahive/index_store.py +++ b/mediahive/index_store.py @@ -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() diff --git a/mediahive/models/data.py b/mediahive/models/data.py index 68f3d71..44d4574 100644 --- a/mediahive/models/data.py +++ b/mediahive/models/data.py @@ -8,7 +8,7 @@ from __future__ import annotations import msgspec -from .tmdb import CastMember, EpisodeInfo, Info, SimilarMedia +from .tmdb import Info # --------------------------------------------------------------------------- diff --git a/mediahive/models/events.py b/mediahive/models/events.py new file mode 100644 index 0000000..d5fa4d3 --- /dev/null +++ b/mediahive/models/events.py @@ -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 diff --git a/mediahive/models/protocol.py b/mediahive/models/protocol.py index 5a73ab5..95a23f2 100644 --- a/mediahive/models/protocol.py +++ b/mediahive/models/protocol.py @@ -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) -c:\mediahive\mediahive\models\protocol.py + return msgspec.json.encode(content) diff --git a/mediahive/models/tmdb.py b/mediahive/models/tmdb.py index bb1cc37..2314165 100644 --- a/mediahive/models/tmdb.py +++ b/mediahive/models/tmdb.py @@ -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 -c:\mediahive\mediahive\models\tmdb.py + networks: list[str] | None = None diff --git a/mediahive/server.py b/mediahive/server.py index 226c2ec..9a1efa8 100644 --- a/mediahive/server.py +++ b/mediahive/server.py @@ -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)