diff --git a/hivescan/__init__.py b/hivescan/__init__.py index 522228a..37534e1 100644 --- a/hivescan/__init__.py +++ b/hivescan/__init__.py @@ -1,22 +1,43 @@ """ -Hivescan - Scans downloaded torrent directories and generates a media index. +Hivescan - Continuous media scanning server with live WebSocket updates. Usage: - hivescan [path] [options] + hivescan /path/to/torrents/* # Start scanning server + hivescan /path/* --port 9000 # Custom port Or as a library: - from hivescan import scan_downloads, generate_media_index + from hivescan.index_store import IndexStore + from hivescan.structs import Movie, Series, TaskInfo + from hivescan.server import app """ from hivescan.models import ContentType, ContentHash, ParsedContent -from hivescan.scanning import scan_downloads, categorize_downloads, find_playable_file, find_episode_files -from hivescan.indexer import generate_media_index +from hivescan.scanning import ( + scan_downloads, + categorize_downloads, + find_playable_file, + find_episode_files, +) +from hivescan.index_store import IndexStore from hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root from hivescan.showreel import generate_showreel_images, generate_episode_reel -from hivescan.tmdb_client import ( +from hivescan.structs import ( + CastMember, + Episode, + EpisodeRelease, + IndexSnapshot, + MediaStats, + Movie, + MovieVersion, + Season, + Series, + SimilarMedia, + TaskInfo, + TMDbEpisodeInfo, TMDbInfo, TMDbSeasonInfo, - TMDbEpisodeInfo, +) +from hivescan.tmdb_client import ( fetch_movie_info, fetch_series_info, fetch_season_details, @@ -33,15 +54,27 @@ __all__ = [ "categorize_downloads", "find_playable_file", "find_episode_files", - # Index generation - "generate_media_index", + # Index store + "IndexStore", + # Struct types + "CastMember", + "Episode", + "EpisodeRelease", + "IndexSnapshot", + "MediaStats", + "Movie", + "MovieVersion", + "Season", + "Series", + "SimilarMedia", + "TaskInfo", + "TMDbEpisodeInfo", + "TMDbInfo", + "TMDbSeasonInfo", # Showreel generation "generate_showreel_images", "generate_episode_reel", # TMDb client - "TMDbInfo", - "TMDbSeasonInfo", - "TMDbEpisodeInfo", "fetch_movie_info", "fetch_series_info", "fetch_season_details", diff --git a/hivescan/__main__.py b/hivescan/__main__.py index 0837fec..2ae85f3 100644 --- a/hivescan/__main__.py +++ b/hivescan/__main__.py @@ -1,32 +1,25 @@ -"""CLI entry point for hivescan.""" +"""Entry point for hivescan — launches the scanning FastAPI server.""" import argparse -import glob -import sys -from pathlib import Path - -from hivescan.scanning import scan_downloads, categorize_downloads -from hivescan.indexer import generate_media_index -from hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root -from hivescan.tmdb_client import set_cache_dir +import os def main(): parser = argparse.ArgumentParser( - description="Scan downloaded torrents and generate a media index.", + description="Hivescan server — continuous media scanning with live WS updates.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: - %(prog)s /path/to/torrents/* # Scan paths, auto-detect common root - %(prog)s /mnt/disk1/* /mnt/disk2/* # Scan multiple locations - %(prog)s /torrents/* -o /srv/media # Override output directory - %(prog)s /torrents/* --no-showreels # Skip showreel generation - %(prog)s /torrents/* --no-covers # Skip cover/backdrop downloads + %(prog)s /path/to/torrents/* # Scan paths, auto-detect common root + %(prog)s /mnt/disk1/* /mnt/disk2/* # Scan multiple locations + %(prog)s /torrents/* -o /srv/media # Override output directory + %(prog)s /torrents/* --port 9000 # Custom port -Output: - By default, creates a .mediahive folder at the common root of scanned paths. - All paths in the index are stored relative to the .mediahive parent folder. - Use -o/--output-dir to override the output location. +The server exposes: + WS /ws Live index updates & task progress + POST /api/scan Trigger a new scan + GET /api/status Current server status + GET /api/index Full index as JSON (HTTP fallback) """, ) @@ -36,85 +29,32 @@ Output: help="Folders or glob patterns to scan for downloads", ) parser.add_argument( - "-o", "--output-dir", + "-o", + "--output-dir", metavar="DIR", - help=f"Output directory for index and covers (default: {DEFAULT_OUTPUT_FOLDER} at common root)", + help="Output directory for index and covers (default: .mediahive at common root)", ) parser.add_argument( - "--no-showreels", - action="store_true", - help="Skip generating showreel images", + "--host", + default="0.0.0.0", + help="Host to bind to (default: 0.0.0.0)", ) parser.add_argument( - "--no-covers", - action="store_true", - help="Skip downloading cover and backdrop images from TMDb", + "--port", + type=int, + default=8421, + help="Port to listen on (default: 8421)", ) - args = parser.parse_args() - # Expand glob patterns and collect all paths - all_paths = [] - for pattern in args.paths: - expanded = glob.glob(pattern) - if expanded: - all_paths.extend(Path(p) for p in expanded) - else: - # Treat as literal path if no glob match - all_paths.append(Path(pattern)) - - if not all_paths: - print("Error: No paths found to scan", file=sys.stderr) - sys.exit(1) - - # Determine output directory + # Pass configuration via environment variables (read by server.py lifespan) + os.environ["HIVESCAN_PATHS"] = os.pathsep.join(args.paths) if args.output_dir: - output_dir = Path(args.output_dir) - media_root = output_dir.parent - else: - # Find common root of all scan paths - media_root = find_common_root(all_paths) - if media_root is None: - print("Error: Cannot determine common root for paths (different drives?)", file=sys.stderr) - print(" Use -o/--output-dir to specify output location", file=sys.stderr) - sys.exit(1) - output_dir = media_root / DEFAULT_OUTPUT_FOLDER + os.environ["HIVESCAN_OUTPUT"] = args.output_dir - output_dir.mkdir(parents=True, exist_ok=True) - index_path = output_dir / "index.json" + from hivescan.server import run - # Set TMDb cache directory within output dir - set_cache_dir(output_dir / ".tmdb-cache") - - print(f"Media root: {media_root}") - print(f"Output dir: {output_dir}") - print(f"Scanning {len(all_paths)} paths...") - - # Scan all paths - downloads = [] - for path in all_paths: - if path.is_dir(): - # Scan directory contents - for item in path.iterdir(): - if not item.name.startswith("."): - from hivescan.parsing import parse_download - downloads.append(parse_download(item)) - elif path.exists(): - from hivescan.parsing import parse_download - downloads.append(parse_download(path)) - - print(f"Found {len(downloads)} items") - - categories = categorize_downloads(downloads) - - generate_media_index( - categories, - index_path, - output_dir, - media_root=media_root, - fetch_covers=not args.no_covers, - generate_showreels=not args.no_showreels, - ) + run(host=args.host, port=args.port) if __name__ == "__main__": diff --git a/hivescan/images.py b/hivescan/images.py index f528ad5..4d097a8 100644 --- a/hivescan/images.py +++ b/hivescan/images.py @@ -1,6 +1,6 @@ """TMDb image downloading functions.""" -import urllib.request +import httpx from pathlib import Path from typing import Optional @@ -12,25 +12,42 @@ TMDB_IMAGE_BASE = "https://image.tmdb.org/t/p" DEFAULT_POSTER_SIZE = "w500" DEFAULT_BACKDROP_SIZE = "w1280" +# Shared async HTTP client (created lazily) +_image_client: Optional[httpx.AsyncClient] = None -def _download_image(url: str, output_path: Path, description: str) -> Optional[str]: + +def _get_image_client() -> httpx.AsyncClient: + """Get or create a shared async HTTP client for image downloads.""" + global _image_client + if _image_client is None: + _image_client = httpx.AsyncClient( + headers={"User-Agent": "TorrentManager/1.0"}, + timeout=30.0, + follow_redirects=True, + ) + return _image_client + + +async def _download_image( + url: str, output_path: Path, description: str +) -> Optional[str]: """Download an image from URL to output path.""" if output_path.exists(): return str(output_path) try: - req = urllib.request.Request(url, headers={"User-Agent": "TorrentManager/1.0"}) - with urllib.request.urlopen(req, timeout=30) as response: - output_path.parent.mkdir(parents=True, exist_ok=True) - with open(output_path, "wb") as f: - f.write(response.read()) + client = _get_image_client() + response = await client.get(url) + response.raise_for_status() + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(response.content) return str(output_path) except Exception as e: print(f" Failed to download {description}: {e}") return None -def download_cover_image( +async def download_cover_image( poster_path: str, title: str, year: Optional[int], @@ -50,10 +67,10 @@ def download_cover_image( url = f"{TMDB_IMAGE_BASE}/{size}{poster_path}" print(f" Downloading cover: {title}") - return _download_image(url, cover_path, f"cover for {title}") + return await _download_image(url, cover_path, f"cover for {title}") -def download_backdrop_image( +async def download_backdrop_image( backdrop_path: str, title: str, year: Optional[int], @@ -73,10 +90,10 @@ def download_backdrop_image( url = f"{TMDB_IMAGE_BASE}/{size}{backdrop_path}" print(f" Downloading backdrop: {title}") - return _download_image(url, local_path, f"backdrop for {title}") + return await _download_image(url, local_path, f"backdrop for {title}") -def download_season_poster( +async def download_season_poster( poster_path: str, media_folder: Path, season_num: int, @@ -92,4 +109,4 @@ def download_season_poster( media_folder.mkdir(parents=True, exist_ok=True) url = f"{TMDB_IMAGE_BASE}/{DEFAULT_POSTER_SIZE}{poster_path}" - return _download_image(url, output_path, f"season {season_num} poster") + return await _download_image(url, output_path, f"season {season_num} poster") diff --git a/hivescan/index_store.py b/hivescan/index_store.py new file mode 100644 index 0000000..a28ef8b --- /dev/null +++ b/hivescan/index_store.py @@ -0,0 +1,236 @@ +""" +In-memory index store with disk snapshot and WebSocket broadcast. + +The IndexStore is the single source of truth for the media index. +All mutations happen synchronously in the asyncio event loop — no locks needed. +index.json on disk is a recovery snapshot only, written periodically via a +debounced background task. +""" + +import asyncio +import logging +from datetime import datetime +from pathlib import Path +from typing import Optional + +import msgspec +from fastapi import WebSocket + +from hivescan.structs import ( + IndexSnapshot, + MediaStats, + Movie, + Series, + TaskInfo, + WsInit, + WsInitData, + WsRemove, + WsTask, + WsUpsert, +) + +logger = logging.getLogger("hivescan.index_store") + +# Debounce interval for writing snapshots to disk (seconds) +SNAPSHOT_DEBOUNCE = 5.0 + + +class IndexStore: + """In-memory media index with WS broadcast and disk snapshots.""" + + def __init__(self, snapshot_path: Path, media_root: Optional[str] = None): + self.snapshot_path = snapshot_path + self.media_root = media_root + + # The index: keyed by item id + self.movies: dict[str, Movie] = {} + self.series: dict[str, Series] = {} + + # Connected WebSocket clients + self._clients: set[WebSocket] = set() + + # Snapshot debounce state + self._snapshot_dirty = False + self._snapshot_task: Optional[asyncio.Task] = None + + # ------------------------------------------------------------------ + # Persistence + # ------------------------------------------------------------------ + + def load_snapshot(self) -> None: + """Load index from disk snapshot (recovery on startup).""" + if not self.snapshot_path.exists(): + logger.info("No snapshot found at %s, starting fresh", self.snapshot_path) + return + try: + data = msgspec.json.decode( + self.snapshot_path.read_bytes(), type=IndexSnapshot + ) + for m in data.movies: + self.movies[m.id] = m + for s in data.series: + self.series[s.id] = s + logger.info( + "Loaded snapshot: %d movies, %d series", + len(self.movies), + len(self.series), + ) + except Exception: + logger.exception("Failed to load snapshot from %s", self.snapshot_path) + + def _write_snapshot(self) -> None: + """Write current index to disk (synchronous, 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.versions) 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, + ) + + self.snapshot_path.parent.mkdir(parents=True, exist_ok=True) + tmp = self.snapshot_path.with_suffix(".tmp") + tmp.write_bytes(msgspec.json.format(msgspec.json.encode(snapshot), indent=2)) + tmp.replace(self.snapshot_path) + logger.debug("Snapshot written to %s", self.snapshot_path) + + def _schedule_snapshot(self) -> None: + """Schedule a debounced snapshot write.""" + self._snapshot_dirty = True + if self._snapshot_task is None or self._snapshot_task.done(): + self._snapshot_task = asyncio.create_task(self._debounced_snapshot()) + + async def _debounced_snapshot(self) -> None: + """Wait for debounce interval then write if still dirty.""" + while self._snapshot_dirty: + self._snapshot_dirty = False + await asyncio.sleep(SNAPSHOT_DEBOUNCE) + # After the sleep, if no new mutations happened, write + self._write_snapshot() + + async def flush_snapshot(self) -> None: + """Force-write a snapshot immediately (e.g. on shutdown).""" + if self._snapshot_task and not self._snapshot_task.done(): + self._snapshot_task.cancel() + try: + await self._snapshot_task + except asyncio.CancelledError: + pass + self._write_snapshot() + + # ------------------------------------------------------------------ + # Mutations + # ------------------------------------------------------------------ + + def upsert_movie(self, item: Movie) -> None: + """Insert or update a movie in the index and broadcast.""" + self.movies[item.id] = item + self._schedule_snapshot() + self._broadcast(WsUpsert(kind="movie", item=item)) + + def upsert_series(self, item: Series) -> None: + """Insert or update a series in the index and broadcast.""" + self.series[item.id] = item + self._schedule_snapshot() + self._broadcast(WsUpsert(kind="series", item=item)) + + 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)) + + 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)) + + # ------------------------------------------------------------------ + # WebSocket management + # ------------------------------------------------------------------ + + async def connect(self, ws: WebSocket) -> None: + """Accept a WS client and send the full index as init.""" + await ws.accept() + self._clients.add(ws) + logger.info("WS client connected (%d total)", len(self._clients)) + # Send full current state + msg = WsInit( + data=WsInitData( + movies=list(self.movies.values()), + series=list(self.series.values()), + ) + ) + await ws.send_bytes(msgspec.json.encode(msg)) + + def disconnect(self, ws: WebSocket) -> None: + """Remove a WS client.""" + self._clients.discard(ws) + logger.info("WS client disconnected (%d remaining)", len(self._clients)) + + def _broadcast(self, msg: object) -> None: + """Broadcast a message to all connected WS clients (non-blocking).""" + data = msgspec.json.encode(msg) + dead: list[WebSocket] = [] + for ws in self._clients: + asyncio.create_task(self._safe_send(ws, data, dead)) + # Clean up dead connections after sends are scheduled + for ws in dead: + self._clients.discard(ws) + + @staticmethod + async def _safe_send(ws: WebSocket, data: bytes, dead: list) -> None: + """Send data to a WS client; mark as dead on failure.""" + try: + await ws.send_bytes(data) + except Exception: + dead.append(ws) + + def broadcast_task(self, task_info: TaskInfo) -> None: + """Broadcast a task progress message to all WS clients.""" + self._broadcast(WsTask(data=task_info)) + + # ------------------------------------------------------------------ + # Read helpers + # ------------------------------------------------------------------ + + def get_full_index(self) -> IndexSnapshot: + """Return the full index as an IndexSnapshot.""" + 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.versions) for m in movies_list) + total_series_episodes = sum( + sum(len(season.episodes) for season in s.seasons) for s in series_list + ) + + return 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, + ) diff --git a/hivescan/indexer.py b/hivescan/indexer.py index fbafe0c..0b7250e 100644 --- a/hivescan/indexer.py +++ b/hivescan/indexer.py @@ -1,25 +1,26 @@ -"""Media index generation.""" +"""Media index generation — async generators for continuous scanning.""" import hashlib -import json -from datetime import datetime +import logging from pathlib import Path -from typing import Dict, List, Optional, Tuple - -from tqdm import tqdm +from typing import AsyncIterator, Dict, List, Optional, Tuple from hivescan.showreel import ( - episode_reel_exists, - generate_episode_reel, - generate_showreel_images, get_expected_episode_reel_path, get_expected_showreel_paths, - movie_showreels_exist, ) -from hivescan.tmdb_client import ( +from hivescan.structs import ( + Episode, + EpisodeRelease, + Movie, + MovieVersion, + Season, + Series, + TMDbEpisodeInfo, TMDbInfo, TMDbSeasonInfo, - TMDbEpisodeInfo, +) +from hivescan.tmdb_client import ( fetch_movie_info, fetch_series_info, fetch_season_details, @@ -27,7 +28,11 @@ from hivescan.tmdb_client import ( from hivescan.models import ContentType, ParsedContent from hivescan.scanning import find_cover_image, find_episode_files, find_playable_file -from hivescan.images import download_cover_image, download_backdrop_image, download_season_poster +from hivescan.images import ( + download_cover_image, + download_backdrop_image, + download_season_poster, +) from hivescan.utils import ( get_added_timestamp, get_media_folder_path, @@ -35,27 +40,33 @@ from hivescan.utils import ( sort_by_quality, ) +logger = logging.getLogger("hivescan.indexer") -def _build_version_info(item: ParsedContent, media_root: Optional[str] = None) -> dict: - """Build version/release info dict for a single torrent.""" + +def _build_version_info( + item: ParsedContent, media_root: Optional[str] = None +) -> MovieVersion: + """Build version/release info for a single torrent.""" playable_file = find_playable_file(item.path) size = item.content_hash.size if item.content_hash else None newest = get_added_timestamp(item.path) - return { - "path": make_relative_path(str(item.path), media_root), - "playable_file": make_relative_path(playable_file, media_root), - "resolution": item.resolution, - "quality": item.quality, - "codec": item.codec, - "audio": item.audio, - "encoder": item.encoder, - "size": size, - "newest": newest, - } + return MovieVersion( + path=make_relative_path(str(item.path), media_root), + playable_file=make_relative_path(playable_file, media_root), + resolution=item.resolution, + quality=item.quality, + codec=item.codec, + audio=item.audio, + encoder=item.encoder, + size=size, + newest=newest, + ) -def _collect_episode_files(items: List[ParsedContent]) -> Dict[Tuple[int, int], List[Dict]]: +def _collect_episode_files( + items: List[ParsedContent], +) -> Dict[Tuple[int, int], List[Dict]]: """ Collect all episode files from a list of torrent items. @@ -71,21 +82,27 @@ def _collect_episode_files(items: List[ParsedContent]) -> Dict[Tuple[int, int], if key not in all_episode_files: all_episode_files[key] = [] for file_path, file_size in files: - all_episode_files[key].append({ - "path": file_path, - "size": file_size, - "resolution": item.resolution, - "quality": item.quality, - "codec": item.codec, - "audio": item.audio, - "encoder": item.encoder, - "torrent_path": str(item.path), - }) + all_episode_files[key].append( + { + "path": file_path, + "size": file_size, + "resolution": item.resolution, + "quality": item.quality, + "codec": item.codec, + "audio": item.audio, + "encoder": item.encoder, + "torrent_path": str(item.path), + } + ) # Handle individual episodes from PTN parsing if item.episode is not None and item.season is not None: - season_nums = item.season if isinstance(item.season, list) else [item.season] - episode_nums = item.episode if isinstance(item.episode, list) else [item.episode] + season_nums = ( + item.season if isinstance(item.season, list) else [item.season] + ) + episode_nums = ( + item.episode if isinstance(item.episode, list) else [item.episode] + ) playable = find_playable_file(item.path) if playable: @@ -94,19 +111,24 @@ def _collect_episode_files(items: List[ParsedContent]) -> Dict[Tuple[int, int], key = (sn, ep) if key not in all_episode_files: all_episode_files[key] = [] - already_added = any(f["path"] == playable for f in all_episode_files.get(key, [])) + already_added = any( + f["path"] == playable + for f in all_episode_files.get(key, []) + ) if not already_added: size = item.content_hash.size if item.content_hash else 0 - all_episode_files[key].append({ - "path": playable, - "size": size, - "resolution": item.resolution, - "quality": item.quality, - "codec": item.codec, - "audio": item.audio, - "encoder": item.encoder, - "torrent_path": str(item.path), - }) + all_episode_files[key].append( + { + "path": playable, + "size": size, + "resolution": item.resolution, + "quality": item.quality, + "codec": item.codec, + "audio": item.audio, + "encoder": item.encoder, + "torrent_path": str(item.path), + } + ) return all_episode_files @@ -120,7 +142,7 @@ def _build_episodes_data( episode_reel_tasks: List, series_title: str, media_root: Optional[str] = None, -) -> List[dict]: +) -> List[Episode]: """Build episode data list for a season.""" episodes_data = [] @@ -134,40 +156,49 @@ def _build_episodes_data( if generate_showreels and episode_files: best_file = episode_files[0]["path"] if best_file and not best_file.endswith(".bdmv"): - reel_path = get_expected_episode_reel_path(series_folder, season_num, episode_num, media_root=Path(media_root) if media_root else None) - episode_reel_tasks.append((best_file, series_folder, season_num, episode_num, series_title)) + reel_path = get_expected_episode_reel_path( + series_folder, + season_num, + episode_num, + media_root=Path(media_root) if media_root else None, + ) + episode_reel_tasks.append( + (best_file, series_folder, season_num, episode_num, series_title) + ) releases = [] for f in episode_files: - releases.append({ - "path": make_relative_path(f["torrent_path"], media_root), - "playable_file": make_relative_path(f["path"], media_root), - "resolution": f.get("resolution"), - "quality": f.get("quality"), - "codec": f.get("codec"), - "audio": f.get("audio"), - "encoder": f.get("encoder"), - "size": f.get("size"), - }) + releases.append( + EpisodeRelease( + path=make_relative_path(f["torrent_path"], media_root), + playable_file=make_relative_path(f["path"], media_root), + resolution=f.get("resolution"), + quality=f.get("quality"), + codec=f.get("codec"), + audio=f.get("audio"), + encoder=f.get("encoder"), + size=f.get("size"), + ) + ) - episode_data = { - "episode_number": episode_num, - "name": tmdb_ep.name if tmdb_ep else None, - "overview": tmdb_ep.overview if tmdb_ep else None, - "air_date": tmdb_ep.air_date if tmdb_ep else None, - "runtime": tmdb_ep.runtime if tmdb_ep else None, - "still_path": tmdb_ep.still_path if tmdb_ep else None, - "rating": tmdb_ep.vote_average if tmdb_ep else None, - "director": tmdb_ep.director if tmdb_ep else None, - "reel_image": reel_path, - "releases": releases, - } + episode_data = Episode( + episode_number=episode_num, + name=tmdb_ep.name if tmdb_ep else None, + overview=tmdb_ep.overview if tmdb_ep else None, + air_date=tmdb_ep.air_date if tmdb_ep else None, + runtime=tmdb_ep.runtime if tmdb_ep else None, + still_path=tmdb_ep.still_path if tmdb_ep else None, + rating=tmdb_ep.vote_average if tmdb_ep else None, + director=tmdb_ep.director if tmdb_ep else None, + reel_image=reel_path, + releases=releases, + ) episodes_data.append(episode_data) return episodes_data -def _build_seasons_data( +async def _build_seasons_data( all_episode_files: Dict[Tuple[int, int], List[Dict]], tmdb_id: Optional[int], series_folder: Path, @@ -177,7 +208,7 @@ def _build_seasons_data( season_cache: Dict, episode_reel_tasks: List, media_root: Optional[str] = None, -) -> List[dict]: +) -> List[Season]: """Build seasons data structure for a series.""" # Group episodes by season seasons_map: Dict[int, Dict[int, List[Dict]]] = {} @@ -197,8 +228,12 @@ def _build_seasons_data( if tmdb_id: cache_key = (tmdb_id, season_num) if cache_key not in season_cache: - print(f" Fetching season {season_num} details for {display_title}") - season_cache[cache_key] = fetch_season_details(tmdb_id, season_num) + logger.info( + " Fetching season %d details for %s", season_num, display_title + ) + season_cache[cache_key] = await fetch_season_details( + tmdb_id, season_num + ) tmdb_season = season_cache[cache_key] if tmdb_season and tmdb_season.episodes: @@ -208,43 +243,57 @@ def _build_seasons_data( # Download season poster season_poster_path = None if fetch_covers and tmdb_season and tmdb_season.poster_path: - season_poster_path = download_season_poster(tmdb_season.poster_path, series_folder, season_num) + season_poster_path = await download_season_poster( + tmdb_season.poster_path, series_folder, season_num + ) episodes_data = _build_episodes_data( - episodes_in_season, tmdb_episodes, series_folder, season_num, - generate_showreels, episode_reel_tasks, display_title, media_root + episodes_in_season, + tmdb_episodes, + series_folder, + season_num, + generate_showreels, + episode_reel_tasks, + display_title, + media_root, ) - season_data = { - "season_number": season_num, - "name": tmdb_season.name if tmdb_season else None, - "overview": tmdb_season.overview if tmdb_season else None, - "air_date": tmdb_season.air_date if tmdb_season else None, - "poster_path": make_relative_path(season_poster_path, media_root) if season_poster_path else None, - "episode_count": len(episodes_data), - "episodes": episodes_data, - } + season_data = Season( + season_number=season_num, + name=tmdb_season.name if tmdb_season else None, + overview=tmdb_season.overview if tmdb_season else None, + air_date=tmdb_season.air_date if tmdb_season else None, + poster_path=make_relative_path(season_poster_path, media_root) + if season_poster_path + else None, + episode_count=len(episodes_data), + episodes=episodes_data, + ) seasons_data.append(season_data) return seasons_data -def _process_movies( +async def _process_movies( categories: dict, cover_dir: Path, fetch_covers: bool, generate_showreels: bool, media_root: Optional[str] = None, -) -> Tuple[List[dict], List[Tuple[str, Path, str]]]: - """Process all movies and return (movies_list, showreel_tasks).""" +) -> AsyncIterator[Tuple[Movie, Optional[Tuple[str, Path, str]]]]: + """ + Async generator that processes all 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]] = {} - def get_movie_tmdb(title: str, year: Optional[int]) -> Optional[TMDbInfo]: + async def get_movie_tmdb(title: str, year: Optional[int]) -> Optional[TMDbInfo]: cache_key = f"{title.lower()}:{year}" if cache_key in movie_tmdb_cache: return movie_tmdb_cache[cache_key] - tmdb_info = fetch_movie_info(title, year) + tmdb_info = await fetch_movie_info(title, year) movie_tmdb_cache[cache_key] = tmdb_info return tmdb_info @@ -252,7 +301,9 @@ def _process_movies( return find_playable_file(item.path) is not None # Filter movies with playable files - valid_movies = [item for item in categories[ContentType.MOVIE] if has_playable(item)] + valid_movies = [ + item for item in categories[ContentType.MOVIE] if has_playable(item) + ] skipped = len(categories[ContentType.MOVIE]) - len(valid_movies) if skipped > 0: print(f" Skipped {skipped} movie torrents with no playable video files") @@ -269,13 +320,23 @@ def _process_movies( tmdb_movie_groups: Dict[int, Dict] = {} no_tmdb_movie_groups: Dict[str, Dict] = {} - print(f" Processing {len(movie_groups)} unique movies ({len(categories[ContentType.MOVIE])} total versions)...") + logger.info( + " Processing %d unique movies (%d total versions)...", + len(movie_groups), + len(categories[ContentType.MOVIE]), + ) for idx, (movie_key, items) in enumerate(movie_groups.items(), 1): first_item = items[0] - print(f" [{idx}/{len(movie_groups)}] {first_item.title} ({first_item.year})\x1b[K", end="\r") + logger.info( + " [%d/%d] %s (%s)", + idx, + len(movie_groups), + first_item.title, + first_item.year, + ) - tmdb_info = get_movie_tmdb(first_item.title, first_item.year) + tmdb_info = await get_movie_tmdb(first_item.title, first_item.year) if tmdb_info and tmdb_info.tmdb_id: if tmdb_info.tmdb_id not in tmdb_movie_groups: @@ -290,15 +351,14 @@ def _process_movies( else: key = f"{first_item.title.lower()}:{first_item.year or 0}" if key not in no_tmdb_movie_groups: - no_tmdb_movie_groups[key] = {"items": [], "title": first_item.title, "year": first_item.year} + no_tmdb_movie_groups[key] = { + "items": [], + "title": first_item.title, + "year": first_item.year, + } no_tmdb_movie_groups[key]["items"].extend(items) - print() - - movies = [] - movie_showreel_tasks: List[Tuple[str, Path, str]] = [] - - # Process movies with TMDb info + # Process movies with TMDb info — yield each as ready for tmdb_id, group_data in tmdb_movie_groups.items(): tmdb_info = group_data["tmdb_info"] items = group_data["items"] @@ -318,59 +378,74 @@ def _process_movies( if cover_path: break if not cover_path and tmdb_info.poster_path: - cover_path = download_cover_image(tmdb_info.poster_path, display_title, year, "movie", cover_dir) + cover_path = await download_cover_image( + tmdb_info.poster_path, display_title, year, "movie", cover_dir + ) versions = [_build_version_info(item, media_root) for item in items] sort_by_quality(versions) # Queue showreel generation showreel_paths = [] + showreel_task = None if generate_showreels and versions: - best_playable = versions[0].get("playable_file") + best_playable = versions[0].playable_file if best_playable: - # Reconstruct absolute path from relative path - abs_playable = str(Path(media_root) / best_playable) if media_root else best_playable - media_folder = get_media_folder_path(display_title, year, "movie", cover_dir) - showreel_paths = get_expected_showreel_paths(media_folder, media_root=Path(media_root) if media_root else None) - movie_showreel_tasks.append((abs_playable, media_folder, display_title)) + abs_playable = ( + str(Path(media_root) / best_playable) + if media_root + else best_playable + ) + media_folder = get_media_folder_path( + display_title, year, "movie", cover_dir + ) + showreel_paths = get_expected_showreel_paths( + media_folder, media_root=Path(media_root) if media_root else None + ) + showreel_task = (abs_playable, media_folder, display_title) # Download backdrop backdrop_path = None if fetch_covers and tmdb_info.backdrop_path: - backdrop_path = download_backdrop_image(tmdb_info.backdrop_path, display_title, year, "movie", cover_dir) + backdrop_path = await download_backdrop_image( + tmdb_info.backdrop_path, display_title, year, "movie", cover_dir + ) - different_titles = [t for t in torrent_titles if t.lower() != display_title.lower()] - version_timestamps = [v["newest"] for v in versions if v.get("newest")] + different_titles = [ + t for t in torrent_titles if t.lower() != display_title.lower() + ] + version_timestamps = [v.newest for v in versions if v.newest] newest = max(version_timestamps) if version_timestamps else None - movies.append({ - "id": item_id, - "title": display_title, - "original_title": tmdb_info.original_title, - "alternative_titles": tmdb_info.alternative_titles, - "torrent_titles": different_titles if different_titles else None, - "year": year, - "newest": newest, - "cover_path": make_relative_path(cover_path, media_root), - "backdrop_path": make_relative_path(backdrop_path, media_root), - "showreel_images": showreel_paths if showreel_paths else None, - "versions": versions, - "tmdb_id": tmdb_info.tmdb_id, - "tmdb_title": tmdb_info.title, - "rating": tmdb_info.rating, - "vote_count": tmdb_info.vote_count, - "overview": tmdb_info.overview, - "genres": tmdb_info.genres, - "release_date": tmdb_info.release_date, - "runtime": tmdb_info.runtime, - "status": tmdb_info.status, - "tagline": tmdb_info.tagline, - "poster_path": tmdb_info.poster_path, - "similar": tmdb_info.similar, - "keywords": tmdb_info.keywords, - "cast": tmdb_info.cast, - "director": tmdb_info.director, - }) + movie = Movie( + id=item_id, + title=display_title, + original_title=tmdb_info.original_title, + alternative_titles=tmdb_info.alternative_titles, + torrent_titles=different_titles if different_titles else None, + year=year, + newest=newest, + cover_path=make_relative_path(cover_path, media_root), + backdrop_path=make_relative_path(backdrop_path, media_root), + showreel_images=showreel_paths if showreel_paths else None, + versions=versions, + tmdb_id=tmdb_info.tmdb_id, + tmdb_title=tmdb_info.title, + rating=tmdb_info.rating, + vote_count=tmdb_info.vote_count, + overview=tmdb_info.overview, + genres=tmdb_info.genres, + release_date=tmdb_info.release_date, + runtime=tmdb_info.runtime, + status=tmdb_info.status, + tagline=tmdb_info.tagline, + poster_path=tmdb_info.poster_path, + similar=tmdb_info.similar, + keywords=tmdb_info.keywords, + cast=tmdb_info.cast, + director=tmdb_info.director, + ) + yield movie, showreel_task # Process movies without TMDb info for key, group_data in no_tmdb_movie_groups.items(): @@ -379,57 +454,69 @@ def _process_movies( year = group_data["year"] item_id = hashlib.md5(f"movie:{title}:{year}".encode()).hexdigest()[:12] - cover_path = find_cover_image(title, year, "movie", cover_dir) if fetch_covers else None + cover_path = ( + find_cover_image(title, year, "movie", cover_dir) if fetch_covers else None + ) versions = [_build_version_info(item, media_root) for item in items] sort_by_quality(versions) showreel_paths = [] + showreel_task = None if generate_showreels and versions: - best_playable = versions[0].get("playable_file") + best_playable = versions[0].playable_file if best_playable: - # Reconstruct absolute path from relative path - abs_playable = str(Path(media_root) / best_playable) if media_root else best_playable + abs_playable = ( + str(Path(media_root) / best_playable) + if media_root + else best_playable + ) if not abs_playable.endswith(".bdmv"): - 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) - movie_showreel_tasks.append((abs_playable, media_folder, title)) + 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, + ) + showreel_task = (abs_playable, media_folder, title) - version_timestamps = [v["newest"] for v in versions if v.get("newest")] + version_timestamps = [v.newest for v in versions if v.newest] newest = max(version_timestamps) if version_timestamps else None - movies.append({ - "id": item_id, - "title": title, - "original_title": None, - "torrent_titles": None, - "year": year, - "newest": newest, - "cover_path": make_relative_path(cover_path, media_root), - "showreel_images": showreel_paths if showreel_paths else None, - "versions": versions, - }) - - return movies, movie_showreel_tasks + movie = Movie( + id=item_id, + title=title, + year=year, + newest=newest, + cover_path=make_relative_path(cover_path, media_root), + showreel_images=showreel_paths if showreel_paths else None, + versions=versions, + ) + yield movie, showreel_task -def _process_series( +async def _process_series( categories: dict, cover_dir: Path, fetch_covers: bool, generate_showreels: bool, media_root: Optional[str] = None, -) -> Tuple[List[dict], List[Tuple[str, Path, int, int, str]]]: - """Process all series and return (series_list, episode_reel_tasks).""" +) -> AsyncIterator[Tuple[Series, List[Tuple[str, Path, int, int, str]]]]: + """ + Async generator that processes all 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]] = {} - def get_series_tmdb(title: str) -> Optional[TMDbInfo]: + async def get_series_tmdb(title: str) -> Optional[TMDbInfo]: cache_key = title.lower() if cache_key in series_tmdb_cache: return series_tmdb_cache[cache_key] - tmdb_info = fetch_series_info(title) + tmdb_info = await fetch_series_info(title) series_tmdb_cache[cache_key] = tmdb_info return tmdb_info @@ -439,10 +526,14 @@ def _process_series( return len(find_episode_files(item.path)) > 0 # Filter series with video content - valid_series = [item for item in categories[ContentType.SERIES] if has_video_content(item)] + valid_series = [ + item for item in categories[ContentType.SERIES] if has_video_content(item) + ] skipped = len(categories[ContentType.SERIES]) - len(valid_series) if skipped > 0: - print(f" Skipped {skipped} series torrents with no playable video files") + logger.info( + " Skipped %d series torrents with no playable video files", skipped + ) # Group by title series_groups: Dict[str, List[ParsedContent]] = {} @@ -456,13 +547,17 @@ def _process_series( tmdb_groups: Dict[int, Dict] = {} no_tmdb_groups: Dict[str, Dict] = {} - print(f" Processing {len(series_groups)} unique series ({len(categories[ContentType.SERIES])} total entries)...") + logger.info( + " Processing %d unique series (%d total entries)...", + len(series_groups), + len(categories[ContentType.SERIES]), + ) for idx, (series_key, items) in enumerate(series_groups.items(), 1): first_item = items[0] - print(f" [{idx}/{len(series_groups)}] {first_item.title}\x1b[K", end="\r") + logger.info(" [%d/%d] %s", idx, len(series_groups), first_item.title) - tmdb_info = get_series_tmdb(first_item.title) + tmdb_info = await get_series_tmdb(first_item.title) if tmdb_info and tmdb_info.tmdb_id: if tmdb_info.tmdb_id not in tmdb_groups: @@ -479,12 +574,7 @@ def _process_series( no_tmdb_groups[key] = {"items": [], "title": first_item.title} no_tmdb_groups[key]["items"].extend(items) - print() - - series = [] - episode_reel_tasks: List[Tuple[str, Path, int, int, str]] = [] - - # Process series with TMDb info + # Process series with TMDb info — yield each as ready for series_idx, (tmdb_id, group_data) in enumerate(tmdb_groups.items(), 1): tmdb_info = group_data["tmdb_info"] items = group_data["items"] @@ -493,7 +583,7 @@ def _process_series( display_title = tmdb_info.title series_id = hashlib.md5(f"series:{tmdb_id}".encode()).hexdigest()[:12] - print(f" [{series_idx}/{len(tmdb_groups)}] {display_title}\x1b[K") + logger.info(" [%d/%d] %s", series_idx, len(tmdb_groups), display_title) series_folder = get_media_folder_path(display_title, None, "series", cover_dir) @@ -507,56 +597,71 @@ def _process_series( if cover_path: break if not cover_path and tmdb_info.poster_path: - cover_path = download_cover_image(tmdb_info.poster_path, display_title, None, "series", cover_dir) + cover_path = await download_cover_image( + tmdb_info.poster_path, display_title, None, "series", cover_dir + ) # Download backdrop backdrop_path = None if fetch_covers and tmdb_info.backdrop_path: - backdrop_path = download_backdrop_image(tmdb_info.backdrop_path, display_title, None, "series", cover_dir) + backdrop_path = await download_backdrop_image( + tmdb_info.backdrop_path, display_title, None, "series", cover_dir + ) # Collect and build episode data all_episode_files = _collect_episode_files(items) - seasons_data = _build_seasons_data( - all_episode_files, tmdb_id, series_folder, display_title, - fetch_covers, generate_showreels, season_cache, episode_reel_tasks, media_root + ep_reel_tasks: List[Tuple[str, Path, int, int, str]] = [] + seasons_data = await _build_seasons_data( + all_episode_files, + tmdb_id, + series_folder, + display_title, + fetch_covers, + generate_showreels, + season_cache, + ep_reel_tasks, + media_root, ) if not seasons_data: - print(f" Skipping {display_title} - no episodes found") + logger.info(" Skipping %s - no episodes found", display_title) continue - different_titles = [t for t in torrent_titles if t.lower() != display_title.lower()] + different_titles = [ + t for t in torrent_titles if t.lower() != display_title.lower() + ] item_timestamps = [get_added_timestamp(item.path) for item in items] item_timestamps = [t for t in item_timestamps if t is not None] newest = max(item_timestamps) if item_timestamps else None - series.append({ - "id": series_id, - "title": display_title, - "original_title": tmdb_info.original_title, - "torrent_titles": different_titles if different_titles else None, - "newest": newest, - "cover_path": make_relative_path(cover_path, media_root), - "backdrop_path": make_relative_path(backdrop_path, media_root), - "seasons": seasons_data, - "tmdb_id": tmdb_info.tmdb_id, - "tmdb_title": tmdb_info.title, - "rating": tmdb_info.rating, - "vote_count": tmdb_info.vote_count, - "overview": tmdb_info.overview, - "genres": tmdb_info.genres, - "release_date": tmdb_info.release_date, - "status": tmdb_info.status, - "tagline": tmdb_info.tagline, - "poster_path": tmdb_info.poster_path, - "similar": tmdb_info.similar, - "keywords": tmdb_info.keywords, - "cast": tmdb_info.cast, - "creators": tmdb_info.creators, - "number_of_seasons": tmdb_info.number_of_seasons, - "number_of_episodes": tmdb_info.number_of_episodes, - "networks": tmdb_info.networks, - }) + series = Series( + id=series_id, + title=display_title, + original_title=tmdb_info.original_title, + torrent_titles=different_titles if different_titles else None, + newest=newest, + cover_path=make_relative_path(cover_path, media_root), + backdrop_path=make_relative_path(backdrop_path, media_root), + seasons=seasons_data, + tmdb_id=tmdb_info.tmdb_id, + tmdb_title=tmdb_info.title, + rating=tmdb_info.rating, + vote_count=tmdb_info.vote_count, + overview=tmdb_info.overview, + genres=tmdb_info.genres, + release_date=tmdb_info.release_date, + status=tmdb_info.status, + tagline=tmdb_info.tagline, + poster_path=tmdb_info.poster_path, + similar=tmdb_info.similar, + keywords=tmdb_info.keywords, + cast=tmdb_info.cast, + creators=tmdb_info.creators, + number_of_seasons=tmdb_info.number_of_seasons, + number_of_episodes=tmdb_info.number_of_episodes, + networks=tmdb_info.networks, + ) + yield series, ep_reel_tasks # Process series without TMDb info for key, group_data in no_tmdb_groups.items(): @@ -564,135 +669,38 @@ def _process_series( title = group_data["title"] series_id = hashlib.md5(f"series:{title}".encode()).hexdigest()[:12] - cover_path = find_cover_image(title, None, "series", cover_dir) if fetch_covers else None + cover_path = ( + find_cover_image(title, None, "series", cover_dir) if fetch_covers else None + ) series_folder = get_media_folder_path(title, None, "series", cover_dir) all_episode_files = _collect_episode_files(items) - seasons_data = _build_seasons_data( - all_episode_files, None, series_folder, title, - fetch_covers, generate_showreels, season_cache, episode_reel_tasks, media_root + ep_reel_tasks: List[Tuple[str, Path, int, int, str]] = [] + seasons_data = await _build_seasons_data( + all_episode_files, + None, + series_folder, + title, + fetch_covers, + generate_showreels, + season_cache, + ep_reel_tasks, + media_root, ) if not seasons_data: - print(f" Skipping {title} - no episodes found") + logger.info(" Skipping %s - no episodes found", title) continue item_timestamps = [get_added_timestamp(item.path) for item in items] item_timestamps = [t for t in item_timestamps if t is not None] newest = max(item_timestamps) if item_timestamps else None - series.append({ - "id": series_id, - "title": title, - "original_title": None, - "torrent_titles": None, - "newest": newest, - "cover_path": make_relative_path(cover_path, media_root), - "seasons": seasons_data, - }) - - return series, episode_reel_tasks - - -def _run_showreel_generation( - movie_tasks: List[Tuple[str, Path, str]], - episode_tasks: List[Tuple[str, Path, int, int, str]], -) -> None: - """Run showreel generation for movies and episodes.""" - pending_movie_tasks = [ - (vp, mf, t) for vp, mf, t in movie_tasks - if not movie_showreels_exist(mf) - ] - pending_episode_tasks = [ - (vp, mf, s, e, t) for vp, mf, s, e, t in episode_tasks - if not episode_reel_exists(mf, s, e) - ] - - total_units = len(pending_movie_tasks) * 5 + len(pending_episode_tasks) - skipped_movies = len(movie_tasks) - len(pending_movie_tasks) - skipped_episodes = len(episode_tasks) - len(pending_episode_tasks) - - if total_units > 0: - print(f"\nGenerating showreels: {len(pending_movie_tasks)} movies, {len(pending_episode_tasks)} episodes") - if skipped_movies > 0 or skipped_episodes > 0: - print(f" (skipping {skipped_movies} movies, {skipped_episodes} episodes already done)") - - with tqdm(total=total_units, unit="clip", dynamic_ncols=True) as pbar: - for video_path, media_folder, title in pending_movie_tasks: - pbar.set_description(f"{title[:40]}") - generate_showreel_images(video_path, media_folder, title=title, pbar=pbar) - - for video_path, media_folder, season_num, episode_num, series_title in pending_episode_tasks: - episode_code = f"S{season_num:02d}E{episode_num:02d}" - pbar.set_description(f"{series_title[:30]} {episode_code}") - generate_episode_reel(video_path, media_folder, season_num, episode_num, pbar=pbar) - - print("Showreel generation complete.") - elif movie_tasks or episode_tasks: - print(f"\nAll showreels already exist ({skipped_movies} movies, {skipped_episodes} episodes).") - - -def generate_media_index( - categories: dict[ContentType, list[ParsedContent]], - output_path: Path, - cover_dir: Path, - media_root: Optional[Path] = None, - fetch_covers: bool = True, - generate_showreels: bool = True, -) -> None: - """ - Generate a comprehensive metadata index for the media browser app. - - The index includes: - - Media metadata with versions bundled together - - Cover image paths (relative to media_root) - - Playable file paths - - TMDb data: ratings, cast, similar items, keywords, etc. - - Showreel images for movies and episodes - """ - print(f"Generating media index: {output_path}") - - # Convert media_root to string for relative path calculations - media_root_str = str(media_root) if media_root else None - - # Process movies and series - movies, movie_showreel_tasks = _process_movies(categories, cover_dir, fetch_covers, generate_showreels, media_root_str) - series, episode_reel_tasks = _process_series(categories, cover_dir, fetch_covers, generate_showreels, media_root_str) - - # Sort results - movies.sort(key=lambda x: (x["title"].lower(), x.get("year") or 0)) - series.sort(key=lambda x: x["title"].lower()) - - # Calculate totals - total_movie_versions = sum(len(m["versions"]) for m in movies) - total_series_episodes = sum( - sum(len(season.get("episodes", [])) for season in s["seasons"]) - for s in series - ) - - # Build and write the index - index = { - "version": 5, - "generated_at": datetime.now().isoformat(), - "media_root": media_root_str, - "stats": { - "total_movies": len(movies), - "total_movie_versions": total_movie_versions, - "total_series": len(series), - "total_series_episodes": total_series_episodes, - }, - "movies": movies, - "series": series, - } - - output_path.parent.mkdir(parents=True, exist_ok=True) - with open(output_path, "w", encoding="utf-8") as f: - json.dump(index, f, indent=2, ensure_ascii=False) - - print(f" Movies: {len(movies)} ({total_movie_versions} versions)") - print(f" Series: {len(series)} ({total_series_episodes} episodes)") - print(f" Output: {output_path}") - - # Generate showreels - if generate_showreels: - _run_showreel_generation(movie_showreel_tasks, episode_reel_tasks) + series = Series( + id=series_id, + title=title, + newest=newest, + cover_path=make_relative_path(cover_path, media_root), + seasons=seasons_data, + ) + yield series, ep_reel_tasks diff --git a/hivescan/models.py b/hivescan/models.py index 18abb92..dac6274 100644 --- a/hivescan/models.py +++ b/hivescan/models.py @@ -9,6 +9,7 @@ from typing import Optional class ContentType(Enum): """Types of content that can be identified.""" + MOVIE = "movie" SERIES = "series" OTHER = "other" @@ -17,6 +18,7 @@ class ContentType(Enum): @dataclass class ContentHash: """Hash representing a file or directory's content based on torrent name.""" + path: Path hash: str _size: Optional[int] = None @@ -26,6 +28,7 @@ class ContentHash: """Get the size, computing it lazily if needed.""" if self._size is None: from hivescan.utils import get_directory_size + self._size = get_directory_size(self.path) return self._size @@ -43,6 +46,7 @@ class ContentHash: @dataclass class ParsedContent: """Information parsed from a torrent name.""" + path: Path name: str content_type: ContentType diff --git a/hivescan/parsing.py b/hivescan/parsing.py index 1d7d4c2..eb092b4 100644 --- a/hivescan/parsing.py +++ b/hivescan/parsing.py @@ -62,17 +62,17 @@ def parse_episode_from_filename(filename: str) -> Optional[Tuple[int, int]]: name = filename.lower() # S01E05 format - match = re.search(r's(\d{1,2})e(\d{1,3})', name) + match = re.search(r"s(\d{1,2})e(\d{1,3})", name) if match: return int(match.group(1)), int(match.group(2)) # 1x05 format - match = re.search(r'(\d{1,2})x(\d{1,3})', name) + match = re.search(r"(\d{1,2})x(\d{1,3})", name) if match: return int(match.group(1)), int(match.group(2)) # Season 1 Episode 5 format - match = re.search(r'season\s*(\d{1,2}).*episode\s*(\d{1,3})', name) + match = re.search(r"season\s*(\d{1,2}).*episode\s*(\d{1,3})", name) if match: return int(match.group(1)), int(match.group(2)) diff --git a/hivescan/scanning.py b/hivescan/scanning.py index d2dc9c8..d8468ab 100644 --- a/hivescan/scanning.py +++ b/hivescan/scanning.py @@ -10,7 +10,18 @@ from hivescan.utils import get_media_folder_path, sanitize_filename # Video file extensions -VIDEO_EXTENSIONS = {'.mkv', '.mp4', '.avi', '.m4v', '.mov', '.wmv', '.flv', '.webm', '.ts', '.m2ts'} +VIDEO_EXTENSIONS = { + ".mkv", + ".mp4", + ".avi", + ".m4v", + ".mov", + ".wmv", + ".flv", + ".webm", + ".ts", + ".m2ts", +} # Caches for expensive operations _episode_files_cache: Dict[str, Dict[Tuple[int, int], List[Tuple[str, int]]]] = {} @@ -44,7 +55,9 @@ def scan_downloads(base_pattern: str) -> Iterator[ParsedContent]: yield parse_download(path) -def categorize_downloads(downloads: list[ParsedContent]) -> dict[ContentType, list[ParsedContent]]: +def categorize_downloads( + downloads: list[ParsedContent], +) -> dict[ContentType, list[ParsedContent]]: """Categorize downloads by content type.""" categories: dict[ContentType, list[ParsedContent]] = { ContentType.MOVIE: [], @@ -85,14 +98,14 @@ def find_episode_files(path: Path) -> Dict[Tuple[int, int], List[Tuple[str, int] try: for f in path.rglob("*"): if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS: - if 'sample' in f.name.lower(): + if "sample" in f.name.lower(): continue ep_info = parse_episode_from_filename(f.name) if ep_info: if ep_info not in episodes: episodes[ep_info] = [] episodes[ep_info].append((str(f), f.stat().st_size)) - except (OSError, PermissionError): + except OSError, PermissionError: pass _episode_files_cache[cache_key] = episodes @@ -134,7 +147,7 @@ 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 @@ -142,10 +155,10 @@ def find_playable_file(path: Path) -> Optional[str]: try: for f in path.rglob("*"): if f.is_file() and f.suffix.lower() in VIDEO_EXTENSIONS: - if 'sample' in f.name.lower(): + if "sample" in f.name.lower(): continue video_files.append((f, f.stat().st_size)) - except (OSError, PermissionError): + except OSError, PermissionError: pass if not video_files: @@ -158,7 +171,9 @@ def find_playable_file(path: Path) -> Optional[str]: return result -def find_cover_image(title: str, year: Optional[int], media_type: str, cover_dir: Path) -> Optional[str]: +def find_cover_image( + title: str, year: Optional[int], media_type: str, cover_dir: Path +) -> Optional[str]: """Find a cover image for the given media item.""" media_folder = get_media_folder_path(title, year, media_type, cover_dir) cover_path = media_folder / "cover.jpg" diff --git a/hivescan/server.py b/hivescan/server.py new file mode 100644 index 0000000..cdde738 --- /dev/null +++ b/hivescan/server.py @@ -0,0 +1,461 @@ +""" +Hivescan FastAPI server — scanning, TMDb lookups, showreel generation. + +Exposes: + GET /ws — WebSocket for live index updates & task progress + POST /api/scan — trigger a new scan + GET /api/status — current server status + GET /api/index — full index as JSON (HTTP fallback) +""" + +import asyncio +import glob +import logging +import os +import sys +import uuid +from contextlib import asynccontextmanager +from pathlib import Path +from typing import List, Optional + +import uvicorn +from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect +from fastapi.middleware.cors import CORSMiddleware + +import msgspec +from hivescan.index_store import IndexStore +from hivescan.indexer import _process_movies, _process_series +from hivescan.structs import MsgspecResponse, ScanRequest, StatusResponse, TaskInfo +from hivescan.models import ContentType, ParsedContent +from hivescan.parsing import parse_download +from hivescan.scanning import categorize_downloads +from hivescan.showreel import ( + generate_episode_reel, + generate_showreel_images, + episode_reel_exists, + movie_showreels_exist, +) +from hivescan.tmdb_client import set_cache_dir +from hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root, make_relative_path + +logger = logging.getLogger("hivescan.server") + +# --------------------------------------------------------------------------- +# Configuration (from environment) +# --------------------------------------------------------------------------- + +SCAN_PATHS: List[str] = [] # set in lifespan from HIVESCAN_PATHS +OUTPUT_DIR: Optional[Path] = None # .mediahive folder +MEDIA_ROOT: Optional[Path] = None # parent of OUTPUT_DIR + +# Global state +store: Optional[IndexStore] = None +_scan_task: Optional[asyncio.Task] = None +_showreel_queue: asyncio.Queue = asyncio.Queue() +_showreel_worker_task: Optional[asyncio.Task] = None +_rescan_worker_task: Optional[asyncio.Task] = None + + +# --------------------------------------------------------------------------- +# Lifespan +# --------------------------------------------------------------------------- + + +@asynccontextmanager +async def lifespan(app: FastAPI): + global store, OUTPUT_DIR, MEDIA_ROOT, SCAN_PATHS, _showreel_worker_task + + # Parse configuration + raw_paths = os.environ.get("HIVESCAN_PATHS", "") + if not raw_paths: + logger.error("HIVESCAN_PATHS environment variable must be set") + sys.exit(1) + + # Expand globs + all_paths: List[Path] = [] + for pattern in raw_paths.split(os.pathsep): + pattern = pattern.strip() + if not pattern: + continue + expanded = glob.glob(pattern) + if expanded: + all_paths.extend(Path(p) for p in expanded) + else: + all_paths.append(Path(pattern)) + + SCAN_PATHS = [str(p) for p in all_paths] + + if os.environ.get("HIVESCAN_OUTPUT"): + OUTPUT_DIR = Path(os.environ["HIVESCAN_OUTPUT"]) + MEDIA_ROOT = OUTPUT_DIR.parent + else: + MEDIA_ROOT = find_common_root(all_paths) + if MEDIA_ROOT is None: + logger.error("Cannot determine common root; set HIVESCAN_OUTPUT") + sys.exit(1) + OUTPUT_DIR = MEDIA_ROOT / DEFAULT_OUTPUT_FOLDER + + OUTPUT_DIR.mkdir(parents=True, exist_ok=True) + set_cache_dir(OUTPUT_DIR / ".tmdb-cache") + + # Initialise index store and load snapshot + store = IndexStore(OUTPUT_DIR / "index.json", media_root=str(MEDIA_ROOT)) + store.load_snapshot() + logger.info( + "Index store ready: %d movies, %d series (from snapshot)", + len(store.movies), + len(store.series), + ) + + # Start showreel worker + _showreel_worker_task = asyncio.create_task(_showreel_worker()) + + # Start scan loop (initial scan + periodic rescans) + _rescan_worker_task = asyncio.create_task(_rescan_loop()) + + yield + + # Shutdown — cancel running tasks, flush snapshot + if _scan_task and not _scan_task.done(): + _scan_task.cancel() + if _showreel_worker_task and not _showreel_worker_task.done(): + _showreel_worker_task.cancel() + if _rescan_worker_task and not _rescan_worker_task.done(): + _rescan_worker_task.cancel() + await store.flush_snapshot() + + +# --------------------------------------------------------------------------- +# FastAPI app +# --------------------------------------------------------------------------- + +app = FastAPI(title="Hivescan Server", lifespan=lifespan) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + + +# --------------------------------------------------------------------------- +# WebSocket endpoint +# --------------------------------------------------------------------------- + + +@app.websocket("/ws") +async def ws_endpoint(ws: WebSocket): + """Live index updates and task progress.""" + await store.connect(ws) + try: + while True: + # Keep connection alive; ignore client messages for now + await ws.receive_text() + except WebSocketDisconnect: + store.disconnect(ws) + except Exception: + store.disconnect(ws) + + +# --------------------------------------------------------------------------- +# HTTP endpoints +# --------------------------------------------------------------------------- + + +@app.post("/api/scan") +async def trigger_scan(request: Request): + """Trigger a new scan. If a scan is already running, returns 409.""" + if _scan_task and not _scan_task.done(): + return {"status": "already_running"} + body_bytes = await request.body() + req = ( + msgspec.json.decode(body_bytes, type=ScanRequest) + if body_bytes + else ScanRequest() + ) + override = req.paths if req.paths else None + _start_scan(override) + return {"status": "started"} + + +@app.get("/api/status") +async def server_status(): + """Return current server status.""" + scanning = _scan_task is not None and not _scan_task.done() + return MsgspecResponse( + StatusResponse( + scanning=scanning, + movies=len(store.movies), + series=len(store.series), + showreel_queue=_showreel_queue.qsize(), + ) + ) + + +@app.get("/api/index") +async def get_index(): + """Full index as JSON (HTTP fallback for non-WS clients).""" + return MsgspecResponse(store.get_full_index()) + + +# --------------------------------------------------------------------------- +# Scan orchestration +# --------------------------------------------------------------------------- + + +def _start_scan(paths: Optional[List[str]] = None): + """Launch a background scan task.""" + global _scan_task + _scan_task = asyncio.create_task(_run_scan(paths)) + + +async def _rescan_loop(): + """Run scans in a loop with a short sleep between each.""" + try: + while True: + _start_scan() + # Wait for the current scan to finish + if _scan_task: + await _scan_task + await asyncio.sleep(1) + except asyncio.CancelledError: + return + except Exception: + logger.exception("Rescan loop error") + + +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, upsert each item into IndexStore + 4. Queue showreel tasks + """ + task_id = f"scan-{uuid.uuid4().hex[:8]}" + logger.info("Scan started (%s)", task_id) + + store.broadcast_task( + TaskInfo( + id=task_id, status="running", progress=0, detail="Scanning filesystem..." + ) + ) + + paths_to_scan = override_paths or SCAN_PATHS + media_root_str = str(MEDIA_ROOT) if MEDIA_ROOT else None + + try: + # 1. Discover downloads (sync filesystem walk — fast enough) + downloads: List[ParsedContent] = [] + for pattern in paths_to_scan: + p = Path(pattern) + if p.is_dir(): + for item in p.iterdir(): + if not item.name.startswith("."): + downloads.append(parse_download(item)) + elif p.exists(): + downloads.append(parse_download(p)) + + logger.info("Found %d items to process", len(downloads)) + categories = categorize_downloads(downloads) + total = len(categories[ContentType.MOVIE]) + len(categories[ContentType.SERIES]) + processed = 0 + + # 2. Process movies + store.broadcast_task( + 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, + ): + store.upsert_movie(movie) + if showreel_task: + await _showreel_queue.put(("movie", showreel_task, movie.id)) + processed += 1 + progress = processed / total if total else 1 + store.broadcast_task( + TaskInfo( + id=task_id, + status="running", + progress=round(progress, 3), + detail=movie.title, + ) + ) + + # 3. Process series + store.broadcast_task( + 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, + ): + store.upsert_series(series) + for task in ep_reel_tasks: + await _showreel_queue.put(("episode", task, series.id)) + processed += 1 + progress = processed / total if total else 1 + store.broadcast_task( + TaskInfo( + id=task_id, + status="running", + progress=round(progress, 3), + detail=series.title, + ) + ) + + store.broadcast_task( + TaskInfo(id=task_id, status="completed", progress=1, detail="Scan complete") + ) + logger.info( + "Scan complete (%s): %d movies, %d series", + task_id, + len(store.movies), + len(store.series), + ) + + except asyncio.CancelledError: + store.broadcast_task( + 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) + store.broadcast_task( + TaskInfo(id=task_id, status="error", progress=0, detail="Scan error") + ) + + +# --------------------------------------------------------------------------- +# Showreel worker — processes one task at a time from the queue +# --------------------------------------------------------------------------- + + +async def _showreel_worker(): + """Background worker that generates showreels one at a time.""" + logger.info("Showreel worker started") + while True: + try: + kind, task_data, item_id = await _showreel_queue.get() + task_id = f"showreel-{uuid.uuid4().hex[:8]}" + + if kind == "movie": + video_path, media_folder, title = task_data + if movie_showreels_exist(media_folder): + _showreel_queue.task_done() + continue + store.broadcast_task( + TaskInfo( + id=task_id, + status="running", + progress=0, + detail=f"Showreel: {title}", + ) + ) + await generate_showreel_images(video_path, media_folder, title=title) + # Update the movie item with generated showreel paths + if item_id in store.movies: + movie = store.movies[item_id] + from hivescan.showreel import get_expected_showreel_paths + + media_root_path = ( + Path(store.media_root) if store.media_root else None + ) + paths = get_expected_showreel_paths( + media_folder, media_root=media_root_path + ) + movie.showreel_images = paths if paths else None + store.upsert_movie(movie) + store.broadcast_task( + TaskInfo( + id=task_id, + status="completed", + progress=1, + detail=f"Showreel: {title}", + ) + ) + + elif kind == "episode": + video_path, media_folder, season_num, episode_num, series_title = ( + task_data + ) + if 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}" + store.broadcast_task( + 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 + ) + # Update the series item with the generated reel path + if item_id in store.series and reel_path: + series = store.series[item_id] + media_root_str = store.media_root + 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 = make_relative_path( + reel_path, media_root_str + ) + store.upsert_series(series) + store.broadcast_task( + TaskInfo( + id=task_id, + status="completed", + progress=1, + detail=f"Reel: {series_title} {ep_code}", + ) + ) + + _showreel_queue.task_done() + + except asyncio.CancelledError: + logger.info("Showreel worker shutting down") + return + except Exception: + logger.exception("Showreel worker error") + try: + _showreel_queue.task_done() + except ValueError: + pass + + +# --------------------------------------------------------------------------- +# Standalone entry point +# --------------------------------------------------------------------------- + + +def run(host: str = "0.0.0.0", port: int = 8421): + """Run the hivescan server.""" + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + datefmt="%H:%M:%S", + ) + uvicorn.run(app, host=host, port=port, log_level="info") diff --git a/hivescan/showreel.py b/hivescan/showreel.py index 18e0dc1..bcf7ed2 100644 --- a/hivescan/showreel.py +++ b/hivescan/showreel.py @@ -6,15 +6,16 @@ Supports automatic black bar detection and removal, hardware-accelerated encodin and HDR passthrough. """ +import asyncio import json +import logging import re import shlex -import subprocess from collections import Counter from pathlib import Path from typing import Optional -from tqdm import tqdm +logger = logging.getLogger("hivescan.showreel") # Showreel timestamp positions in seconds (5, 10, 15, 20, 25 minutes) @@ -77,7 +78,9 @@ def get_expected_episode_reel_path( return str(output_path) -def movie_showreels_exist(media_folder: Path, timestamps: list[int] = SHOWREEL_TIMESTAMPS) -> bool: +def movie_showreels_exist( + media_folder: Path, timestamps: list[int] = SHOWREEL_TIMESTAMPS +) -> bool: """Check if all showreel files for a movie already exist.""" for reel_num in range(1, len(timestamps) + 1): if not (media_folder / f"reel{reel_num}.webm").exists(): @@ -117,7 +120,7 @@ def get_bluray_uri(video_path: str) -> Optional[str]: _av1_encoder_cache: Optional[str] = None -def get_av1_encoder() -> str: +async def get_av1_encoder() -> str: """ Detect the best available AV1 encoder. @@ -133,20 +136,32 @@ def get_av1_encoder() -> str: # Check for NVIDIA AV1 encoder try: - result = subprocess.run( - ["ffmpeg", "-hide_banner", "-encoders"], - capture_output=True, - text=True, - timeout=10 + proc = await asyncio.create_subprocess_exec( + "ffmpeg", + "-hide_banner", + "-encoders", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, ) - if "av1_nvenc" in result.stdout: + stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=10) + if b"av1_nvenc" in stdout: # Verify it actually works (driver support) - test_result = subprocess.run( - ["ffmpeg", "-f", "lavfi", "-i", "nullsrc=s=64x64:d=1", "-c:v", "av1_nvenc", "-f", "null", "-"], - capture_output=True, - timeout=10 + test_proc = await asyncio.create_subprocess_exec( + "ffmpeg", + "-f", + "lavfi", + "-i", + "nullsrc=s=64x64:d=1", + "-c:v", + "av1_nvenc", + "-f", + "null", + "-", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, ) - if test_result.returncode == 0: + await asyncio.wait_for(test_proc.communicate(), timeout=10) + if test_proc.returncode == 0: _av1_encoder_cache = "av1_nvenc" return _av1_encoder_cache except Exception: @@ -175,7 +190,7 @@ def get_encoder_options(encoder: str) -> list[str]: return ["-crf", "38", "-preset", "6"] -def detect_dovi_profile(video_path: str) -> Optional[int]: +async def detect_dovi_profile(video_path: str) -> Optional[int]: """ Detect Dolby Vision profile from a video file. @@ -187,17 +202,29 @@ def detect_dovi_profile(video_path: str) -> Optional[int]: try: # Check for Dolby Vision configuration record in video stream cmd = [ - "ffprobe", "-v", "error", "-select_streams", "v:0", - "-show_entries", "stream_side_data_list", - "-of", "json", video_path + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream_side_data_list", + "-of", + "json", + video_path, ] - print(f" $ {shlex.join(cmd)}") - result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + logger.debug(" $ %s", shlex.join(cmd)) + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30) - if result.returncode != 0: + if proc.returncode != 0: return None - data = json.loads(result.stdout) + data = json.loads(stdout) streams = data.get("streams", []) if not streams: return None @@ -215,20 +242,32 @@ def detect_dovi_profile(video_path: str) -> Optional[int]: # Alternative: check using mediainfo-style detection via codec tag # Some DoVi content has "dvhe" or "dvh1" codec tags codec_cmd = [ - "ffprobe", "-v", "error", "-select_streams", "v:0", - "-show_entries", "stream=codec_tag_string,codec_name", - "-of", "csv=p=0", video_path + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=codec_tag_string,codec_name", + "-of", + "csv=p=0", + video_path, ] - codec_result = subprocess.run(codec_cmd, capture_output=True, text=True, timeout=30) - if codec_result.returncode == 0: - codec_info = codec_result.stdout.lower() + codec_proc = await asyncio.create_subprocess_exec( + *codec_cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + codec_stdout, _ = await asyncio.wait_for(codec_proc.communicate(), timeout=30) + if codec_proc.returncode == 0: + codec_info = codec_stdout.decode("utf-8").lower() if "dvhe" in codec_info or "dvh1" in codec_info or "dav1" in codec_info: # DoVi detected but profile unknown, assume needs conversion return 7 # Conservative: treat as dual-layer return None except Exception as e: - print(f" DoVi detection error: {e}") + logger.warning(" DoVi detection error: %s", e) return None @@ -247,27 +286,32 @@ def get_dovi_to_hdr10_filter() -> str: ) -def is_hdr_video(video_path: str) -> bool: +async def is_hdr_video(video_path: str) -> bool: """ Check if a video file is HDR using ffprobe. Returns True if the video has HDR metadata (bt2020, SMPTE ST 2084, etc.) """ try: - result = subprocess.run( - [ - "ffprobe", "-v", "quiet", "-select_streams", "v:0", - "-show_entries", "stream=color_transfer,color_primaries,color_space", - "-of", "json", video_path - ], - capture_output=True, - text=True, - timeout=30 + proc = await asyncio.create_subprocess_exec( + "ffprobe", + "-v", + "quiet", + "-select_streams", + "v:0", + "-show_entries", + "stream=color_transfer,color_primaries,color_space", + "-of", + "json", + video_path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, ) - if result.returncode != 0: + stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30) + if proc.returncode != 0: return False - data = json.loads(result.stdout) + data = json.loads(stdout) streams = data.get("streams", []) if not streams: return False @@ -285,7 +329,7 @@ def is_hdr_video(video_path: str) -> bool: return False -def detect_crop(video_path: str) -> Optional[str]: +async def detect_crop(video_path: str) -> Optional[str]: """ Detect black bars in a video and return the crop filter string. @@ -306,18 +350,30 @@ def detect_crop(video_path: str) -> Optional[str]: try: # First, get source video dimensions to check if it's 16:9 dim_cmd = [ - "ffprobe", "-v", "error", "-select_streams", "v:0", - "-show_entries", "stream=width,height", - "-of", "csv=p=0", video_path + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=width,height", + "-of", + "csv=p=0", + video_path, ] - print(f" $ {shlex.join(dim_cmd)}") - dim_result = subprocess.run(dim_cmd, capture_output=True, text=True, timeout=30) - if dim_result.returncode != 0: - print(f" Failed to get dimensions: {dim_result.stderr.strip()}") + logger.debug(" $ %s", shlex.join(dim_cmd)) + dim_proc = await asyncio.create_subprocess_exec( + *dim_cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + dim_stdout, _ = await asyncio.wait_for(dim_proc.communicate(), timeout=30) + if dim_proc.returncode != 0: return None - # Parse "width,height" output - parts = dim_result.stdout.strip().split(",") + # Parse "width,height" output (take first line only, ffprobe may emit multiple) + first_line = dim_stdout.decode("utf-8").strip().splitlines()[0].strip() + parts = first_line.split(",") if len(parts) < 2: return None src_width, src_height = int(parts[0]), int(parts[1]) @@ -332,21 +388,42 @@ def detect_crop(video_path: str) -> Optional[str]: # Use ffmpeg to run cropdetect on just 2 seconds at 5-minute mark # This is much faster than scanning 60 seconds with ffprobe lavfi cmd = [ - "ffmpeg", "-hide_banner", "-ss", "300", "-i", video_path, - "-t", "2", "-vf", "cropdetect=limit=24:round=2:reset=0", - "-f", "null", "-" + "ffmpeg", + "-hide_banner", + "-ss", + "300", + "-i", + video_path, + "-t", + "2", + "-vf", + "cropdetect=limit=24:round=2:reset=0", + "-f", + "null", + "-", ] - print(f" $ {shlex.join(cmd)}") - result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) + logger.debug(" $ %s", shlex.join(cmd)) + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + _, stderr_bytes = await asyncio.wait_for(proc.communicate(), timeout=60) + stderr_text = stderr_bytes.decode("utf-8", errors="replace") # cropdetect outputs to stderr like: [Parsed_cropdetect_0 @ ...] x1:0 x2:1919 y1:138 y2:941 w:1920 h:800 ... # We need to parse the crop values from stderr - crop_pattern = re.compile(r'crop=(\d+):(\d+):(\d+):(\d+)') + crop_pattern = re.compile(r"crop=(\d+):(\d+):(\d+):(\d+)") crop_values = [] - for line in result.stderr.split('\n'): + for line in stderr_text.split("\n"): match = crop_pattern.search(line) if match: - w, h, x, y = int(match.group(1)), int(match.group(2)), int(match.group(3)), int(match.group(4)) + w, h, x, y = ( + int(match.group(1)), + int(match.group(2)), + int(match.group(3)), + int(match.group(4)), + ) if w > 0 and h > 0 and x >= 0 and y >= 0: crop_values.append((w, h, x, y)) @@ -398,44 +475,48 @@ def detect_crop(video_path: str) -> Optional[str]: return None crop_result = f"crop={w_aligned}:{h_aligned}:{x_aligned}:{y_aligned}" - print(f" Detected crop: {crop_result}") + logger.debug(" Detected crop: %s", crop_result) return crop_result except Exception as e: - print(f" Crop detection error: {e}") + logger.warning(" Crop detection error: %s", e) return None -def get_video_duration(video_path: str) -> Optional[float]: +async def get_video_duration(video_path: str) -> Optional[float]: """ Get the duration of a video file in seconds using ffprobe. """ try: - result = subprocess.run( - [ - "ffprobe", "-v", "quiet", "-show_entries", "format=duration", - "-of", "json", video_path - ], - capture_output=True, - text=True, - timeout=30 + proc = await asyncio.create_subprocess_exec( + "ffprobe", + "-v", + "quiet", + "-show_entries", + "format=duration", + "-of", + "json", + video_path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, ) - if result.returncode != 0: + stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30) + if proc.returncode != 0: return None - data = json.loads(result.stdout) + data = json.loads(stdout) duration = data.get("format", {}).get("duration") return float(duration) if duration else None except Exception: return None -def generate_showreel_images( +async def generate_showreel_images( video_path: str, media_folder: Path, timestamps: list[int] = SHOWREEL_TIMESTAMPS, title: str = None, - pbar: Optional[tqdm] = None, + on_progress=None, ) -> list[str]: """ Generate showreel video clips from a video file at specified timestamps. @@ -447,7 +528,8 @@ def generate_showreel_images( video_path: Path to the video file (or index.bdmv for Blu-ray discs) media_folder: Folder for this specific media item timestamps: List of timestamps in seconds to capture - pbar: Optional tqdm progress bar to update + title: Title for logging + on_progress: Optional callback(reel_num) called after each reel completes Returns: List of relative paths to generated showreel video clips @@ -482,9 +564,9 @@ def generate_showreel_images( media_folder.mkdir(parents=True, exist_ok=True) # Check video duration to avoid seeking past the end - duration = get_video_duration(ffmpeg_input) + duration = await get_video_duration(ffmpeg_input) if duration is None: - print(f" Could not get duration for: {video_path}") + logger.warning(" Could not get duration for: %s", video_path) return [] # Filter timestamps that are within the video duration (with 40s margin for 10s clips) @@ -497,17 +579,17 @@ def generate_showreel_images( return [] # Get the best available AV1 encoder - encoder = get_av1_encoder() + encoder = await get_av1_encoder() encoder_opts = get_encoder_options(encoder) # Detect Dolby Vision profile for tonemapping (profiles 5/7 need conversion) - dovi_profile = detect_dovi_profile(ffmpeg_input) + dovi_profile = await detect_dovi_profile(ffmpeg_input) needs_tonemap = dovi_profile is not None and dovi_profile in (5, 7) if needs_tonemap: - print(f" DoVi profile {dovi_profile} detected, will convert to HDR10") + logger.info(" DoVi profile %d detected, will convert to HDR10", dovi_profile) # Detect black bars once for all clips (uses same video source) - crop_filter = detect_crop(ffmpeg_input) + crop_filter = await detect_crop(ffmpeg_input) generated_paths = [] @@ -518,8 +600,8 @@ def generate_showreel_images( # Skip if already exists if output_path.exists(): generated_paths.append(str(output_path)) - if pbar: - pbar.update(1) + if on_progress: + on_progress(reel_num) continue # Build video filter chain: @@ -534,60 +616,74 @@ def generate_showreel_images( vf_parts.append("scale='min(720,iw)':-2") vf_filter = ",".join(vf_parts) cmd = [ - "ffmpeg", "-y", "-ss", str(timestamp), "-i", ffmpeg_input, - "-hide_banner", "-loglevel", "warning", "-stats", - "-map", "0:v:0", "-map", "0:a:0?", # First video, first audio (optional) - "-t", "10", - "-vf", vf_filter, - "-c:v", encoder, + "ffmpeg", + "-y", + "-ss", + str(timestamp), + "-i", + ffmpeg_input, + "-hide_banner", + "-loglevel", + "warning", + "-stats", + "-map", + "0:v:0", + "-map", + "0:a:0?", # First video, first audio (optional) + "-t", + "10", + "-vf", + vf_filter, + "-c:v", + encoder, *encoder_opts, - "-c:a", "libopus", - "-ac", "2", - "-b:a", "128k", + "-c:a", + "libopus", + "-ac", + "2", + "-b:a", + "128k", str(output_path), ] - print(f" $ {shlex.join(cmd)}") + logger.debug(" $ %s", shlex.join(cmd)) try: - result = subprocess.run(cmd, timeout=120) + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + await asyncio.wait_for(proc.communicate(), timeout=120) - if result.returncode == 0 and output_path.exists(): + if proc.returncode == 0 and output_path.exists(): generated_paths.append(str(output_path)) - if pbar: - pbar.update(1) + if on_progress: + on_progress(reel_num) else: output_path.unlink(missing_ok=True) - if pbar: - # Update remaining reels as skipped - remaining = len(valid_timestamps) - reel_num + 1 - pbar.update(remaining) - pbar.refresh() # Abort remaining reels - if first one fails, others likely will too break except BaseException as e: output_path.unlink(missing_ok=True) - if isinstance(e, (KeyboardInterrupt, SystemExit)): + if isinstance(e, (KeyboardInterrupt, SystemExit, asyncio.CancelledError)): raise - if pbar: - pbar.clear() - print(f"\n\033[91mError generating showreel for {title or 'unknown'} at {timestamp}s: {e}\033[0m") - if pbar: - # Update remaining reels as skipped - remaining = len(valid_timestamps) - reel_num + 1 - pbar.update(remaining) - pbar.refresh() + logger.error( + "Error generating showreel for %s at %ds: %s", + title or "unknown", + timestamp, + e, + ) # Abort remaining reels break return generated_paths -def generate_episode_reel( +async def generate_episode_reel( video_path: str, media_folder: Path, season_num: int, episode_num: int, - pbar: Optional[tqdm] = None, ) -> Optional[str]: """ Generate a single 10-second reel video clip for a TV episode. @@ -601,7 +697,6 @@ def generate_episode_reel( media_folder: Folder for this series season_num: Season number episode_num: Episode number - pbar: Optional tqdm progress bar to update Returns: Relative path to generated image, or None if failed @@ -629,7 +724,7 @@ def generate_episode_reel( return str(output_path) # Check video duration - duration = get_video_duration(ffmpeg_input) + duration = await get_video_duration(ffmpeg_input) if duration is None: return None @@ -639,17 +734,17 @@ def generate_episode_reel( actual_timestamp = max(10, min(actual_timestamp, duration - 40)) # Get the best available AV1 encoder - encoder = get_av1_encoder() + encoder = await get_av1_encoder() encoder_opts = get_encoder_options(encoder) # Detect Dolby Vision profile for tonemapping (profiles 5/7 need conversion) - dovi_profile = detect_dovi_profile(ffmpeg_input) + dovi_profile = await detect_dovi_profile(ffmpeg_input) needs_tonemap = dovi_profile is not None and dovi_profile in (5, 7) if needs_tonemap: - print(f" DoVi profile {dovi_profile} detected, will convert to HDR10") + logger.info(" DoVi profile %d detected, will convert to HDR10", dovi_profile) # Detect black bars for cropping - crop_filter = detect_crop(ffmpeg_input) + crop_filter = await detect_crop(ffmpeg_input) # Build video filter chain: # 1. DoVi to HDR10 conversion (if needed) - must come first @@ -663,42 +758,54 @@ def generate_episode_reel( vf_parts.append("scale='min(720,iw)':-2") vf_filter = ",".join(vf_parts) cmd = [ - "ffmpeg", "-y", "-ss", str(actual_timestamp), "-i", ffmpeg_input, - "-hide_banner", "-loglevel", "warning", "-stats", - "-map", "0:v:0", "-map", "0:a:0?", # First video, first audio (optional) - "-t", "10", - "-vf", vf_filter, - "-c:v", encoder, + "ffmpeg", + "-y", + "-ss", + str(actual_timestamp), + "-i", + ffmpeg_input, + "-hide_banner", + "-loglevel", + "warning", + "-stats", + "-map", + "0:v:0", + "-map", + "0:a:0?", # First video, first audio (optional) + "-t", + "10", + "-vf", + vf_filter, + "-c:v", + encoder, *encoder_opts, - "-c:a", "libopus", - "-ac", "2", - "-b:a", "128k", + "-c:a", + "libopus", + "-ac", + "2", + "-b:a", + "128k", str(output_path), ] - print(f" $ {shlex.join(cmd)}") + logger.debug(" $ %s", shlex.join(cmd)) try: - result = subprocess.run(cmd, timeout=120) + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + await asyncio.wait_for(proc.communicate(), timeout=120) - if result.returncode == 0 and output_path.exists(): - if pbar: - pbar.update(1) + if proc.returncode == 0 and output_path.exists(): return str(output_path) else: output_path.unlink(missing_ok=True) - if pbar: - pbar.update(1) - pbar.refresh() return None except BaseException as e: output_path.unlink(missing_ok=True) - if isinstance(e, (KeyboardInterrupt, SystemExit)): + if isinstance(e, (KeyboardInterrupt, SystemExit, asyncio.CancelledError)): raise - if pbar: - pbar.clear() episode_code = f"S{season_num:02d}E{episode_num:02d}" - print(f"\n\033[91mError generating episode reel for {episode_code}: {e}\033[0m") - if pbar: - pbar.update(1) - pbar.refresh() + logger.error("Error generating episode reel for %s: %s", episode_code, e) return None diff --git a/hivescan/structs.py b/hivescan/structs.py new file mode 100644 index 0000000..58affee --- /dev/null +++ b/hivescan/structs.py @@ -0,0 +1,338 @@ +""" +Typed structures for the hivescan/mediahive API, index state, and WebSocket protocol. + +All API and state types are msgspec.Structs for fast serialization. +Internal scanning types (ContentHash, ParsedContent) remain in models.py. +""" + +from __future__ import annotations + +import msgspec +from fastapi.responses import Response + + +# --------------------------------------------------------------------------- +# Sub-types (shared by TMDb results and index items) +# --------------------------------------------------------------------------- + + +class CastMember(msgspec.Struct): + """Actor/crew member.""" + + name: str + character: str | None = None + profile_path: str | None = None + + +class SimilarMedia(msgspec.Struct): + """Pointer to a similar movie/series on TMDb.""" + + id: int + title: str + poster_path: str | None = None + + +# --------------------------------------------------------------------------- +# TMDb result types (returned by tmdb_client, consumed by indexer) +# --------------------------------------------------------------------------- + + +class TMDbEpisodeInfo(msgspec.Struct): + """Episode metadata from TMDb.""" + + episode_number: int + season_number: int + name: str | None = None + overview: str | None = None + air_date: str | None = None + runtime: int | None = None + still_path: str | None = None + vote_average: float | None = None + vote_count: int | None = None + director: str | None = None + + +class TMDbSeasonInfo(msgspec.Struct): + """Season metadata from TMDb.""" + + season_number: int + name: str | None = None + overview: str | None = None + air_date: str | None = None + poster_path: str | None = None + episode_count: int | None = None + episodes: list[TMDbEpisodeInfo] | None = None + + +class TMDbInfo(msgspec.Struct): + """Full metadata result from TMDb (movies or series).""" + + tmdb_id: int + title: str | None = None + original_title: str | None = None + alternative_titles: list[str] | None = None + rating: float | None = None + vote_count: int | None = None + overview: str | None = None + genres: list[str] | None = None + release_date: str | None = None + runtime: int | None = None + status: str | None = None + tagline: str | None = None + poster_path: str | None = None + backdrop_path: str | None = None + similar: list[SimilarMedia] | None = None + keywords: list[str] | None = None + cast: list[CastMember] | None = None + director: str | None = None + creators: list[str] | None = None + number_of_seasons: int | None = None + number_of_episodes: int | None = None + networks: list[str] | None = None + + +# --------------------------------------------------------------------------- +# Index item types (the state stored in IndexStore, sent over WS/API) +# --------------------------------------------------------------------------- + + +class MovieVersion(msgspec.Struct): + """One release/torrent of a movie.""" + + path: str | None = None + playable_file: str | None = None + resolution: str | None = None + quality: str | None = None + codec: str | None = None + audio: str | None = None + encoder: str | None = None + size: int | None = None + newest: int | None = None + + +class EpisodeRelease(msgspec.Struct): + """One release/torrent file of an episode.""" + + path: str | None = None + playable_file: str | None = None + resolution: str | None = None + quality: str | None = None + codec: str | None = None + audio: str | None = None + encoder: str | None = None + size: int | None = None + + +class Episode(msgspec.Struct): + """Episode within a season.""" + + episode_number: int + name: str | None = None + overview: str | None = None + air_date: str | None = None + runtime: int | None = None + still_path: str | None = None + rating: float | None = None + director: str | None = None + reel_image: str | None = None + releases: list[EpisodeRelease] = [] + + +class Season(msgspec.Struct): + """Season within a series.""" + + season_number: int + name: str | None = None + overview: str | None = None + air_date: str | None = None + poster_path: str | None = None + episode_count: int | None = None + episodes: list[Episode] = [] + + +class Movie(msgspec.Struct): + """A movie in the index (one or more versions/releases).""" + + id: str + title: str + original_title: str | None = None + alternative_titles: list[str] | None = None + torrent_titles: list[str] | None = None + year: int | None = None + newest: int | None = None + cover_path: str | None = None + backdrop_path: str | None = None + showreel_images: list[str] | None = None + versions: list[MovieVersion] = [] + tmdb_id: int | None = None + tmdb_title: str | None = None + rating: float | None = None + vote_count: int | None = None + overview: str | None = None + genres: list[str] | None = None + release_date: str | None = None + runtime: int | None = None + status: str | None = None + tagline: str | None = None + poster_path: str | None = None + similar: list[SimilarMedia] | None = None + keywords: list[str] | None = None + cast: list[CastMember] | None = None + director: str | None = None + + +class Series(msgspec.Struct): + """A TV series in the index.""" + + id: str + title: str + original_title: str | None = None + alternative_titles: list[str] | None = None + torrent_titles: list[str] | None = None + newest: int | None = None + cover_path: str | None = None + backdrop_path: str | None = None + seasons: list[Season] = [] + tmdb_id: int | None = None + tmdb_title: str | None = None + rating: float | None = None + vote_count: int | None = None + overview: str | None = None + genres: list[str] | None = None + release_date: str | None = None + status: str | None = None + tagline: str | None = None + poster_path: str | None = None + similar: list[SimilarMedia] | None = None + keywords: list[str] | None = None + cast: list[CastMember] | None = None + creators: list[str] | None = None + number_of_seasons: int | None = None + number_of_episodes: int | None = None + networks: list[str] | None = None + + +# --------------------------------------------------------------------------- +# Snapshot (disk format for index.json) +# --------------------------------------------------------------------------- + + +class MediaStats(msgspec.Struct): + """Aggregate counts for the index snapshot.""" + + total_movies: int = 0 + total_movie_versions: int = 0 + total_series: int = 0 + total_series_episodes: int = 0 + + +class IndexSnapshot(msgspec.Struct): + """On-disk recovery snapshot of the full index.""" + + version: int = 5 + generated_at: str = "" + media_root: str | None = None + stats: MediaStats = msgspec.UNSET # type: ignore[assignment] + movies: list[Movie] = [] + series: list[Series] = [] + + def __post_init__(self): + if self.stats is msgspec.UNSET: + self.stats = MediaStats() + + +# --------------------------------------------------------------------------- +# WebSocket message types +# --------------------------------------------------------------------------- + + +class WsInitData(msgspec.Struct): + """Payload of the init message.""" + + movies: list[Movie] + series: list[Series] + + +class WsInit(msgspec.Struct, tag="init"): + """Full index sent on WS connect.""" + + 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 TaskInfo(msgspec.Struct): + """Progress info for a background task (scan, showreel, etc.).""" + + id: str + status: str + progress: float = 0.0 + detail: 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 + + +# --------------------------------------------------------------------------- +# API request / response types +# --------------------------------------------------------------------------- + + +class ScanRequest(msgspec.Struct): + """POST /api/scan body.""" + + paths: list[str] | None = None + + +class StatusResponse(msgspec.Struct): + """GET /api/status response.""" + + scanning: bool = False + movies: int = 0 + series: int = 0 + showreel_queue: int = 0 + + +class PlayMediaRequest(msgspec.Struct): + """POST /api/play body (mediahive server).""" + + file_path: str = "" + + +class OpenFolderRequest(msgspec.Struct): + """POST /api/open-folder body (mediahive server).""" + + folder_path: str = "" + + +# --------------------------------------------------------------------------- +# FastAPI response helper +# --------------------------------------------------------------------------- + + +class MsgspecResponse(Response): + """FastAPI response that serializes content with msgspec.json.""" + + media_type = "application/json; charset=utf-8" + + def render(self, content: object) -> bytes: + return msgspec.json.encode(content) diff --git a/hivescan/tmdb_client.py b/hivescan/tmdb_client.py index 920a131..27e6dcd 100644 --- a/hivescan/tmdb_client.py +++ b/hivescan/tmdb_client.py @@ -3,18 +3,25 @@ TMDb Client - Fetch movie and TV series metadata from The Movie Database (TMDb). """ +import asyncio import hashlib import json import os import sys -import time import urllib.parse -from dataclasses import dataclass from pathlib import Path -from typing import Dict, List, Optional +from typing import Dict, Optional import httpx +from hivescan.structs import ( + CastMember, + SimilarMedia, + TMDbEpisodeInfo, + TMDbInfo, + TMDbSeasonInfo, +) + # TMDb API configuration TMDB_API_KEY = os.environ.get("TMDB_API_KEY", "6bd914e6a5df1c6d1ddf622cf2dbc232") TMDB_API_BASE = "https://api.themoviedb.org/3" @@ -22,8 +29,8 @@ TMDB_API_BASE = "https://api.themoviedb.org/3" # API response cache directory (can be overridden via set_cache_dir) _tmdb_cache_dir: Optional[Path] = None -# Persistent HTTP client for connection reuse -_http_client: Optional[httpx.Client] = None +# Persistent async HTTP client for connection reuse +_http_client: Optional[httpx.AsyncClient] = None def set_cache_dir(cache_dir: Path) -> None: @@ -40,11 +47,11 @@ def _get_cache_dir() -> Path: return Path.cwd() / ".tmdb-cache" -def _get_http_client() -> httpx.Client: - """Get or create a persistent HTTP client for connection reuse.""" +def _get_http_client() -> httpx.AsyncClient: + """Get or create a persistent async HTTP client for connection reuse.""" global _http_client if _http_client is None: - _http_client = httpx.Client( + _http_client = httpx.AsyncClient( base_url=TMDB_API_BASE, headers={"Accept": "application/json", "User-Agent": "TorrentManager/1.0"}, timeout=10.0, @@ -70,7 +77,7 @@ def _load_from_cache(cache_path: Path): if not cache_path.exists(): return _NOT_FOUND try: - with open(cache_path, "r") as f: + with open(cache_path, "r", encoding="utf-8") as f: data = json.load(f) # Handle cached "no results" / errors if data.get("_cached_none"): @@ -84,7 +91,7 @@ def _save_to_cache(cache_path: Path, data: Optional[Dict]): """Save response to cache.""" try: _get_cache_dir().mkdir(parents=True, exist_ok=True) - with open(cache_path, "w") as f: + with open(cache_path, "w", encoding="utf-8") as f: if data is None: json.dump({"_cached_none": True}, f) else: @@ -93,62 +100,12 @@ def _save_to_cache(cache_path: Path, data: Optional[Dict]): pass # Cache write failures are not critical -@dataclass -class TMDbEpisodeInfo: - """Information about a TV episode from TMDb.""" - episode_number: int - season_number: int - name: Optional[str] = None - overview: Optional[str] = None - air_date: Optional[str] = None - runtime: Optional[int] = None # Minutes - still_path: Optional[str] = None # Episode screenshot - vote_average: Optional[float] = None - vote_count: Optional[int] = None - director: Optional[str] = None +# TMDbEpisodeInfo, TMDbSeasonInfo, TMDbInfo imported from hivescan.structs -@dataclass -class TMDbSeasonInfo: - """Information about a TV season from TMDb.""" - season_number: int - name: Optional[str] = None - overview: Optional[str] = None - air_date: Optional[str] = None - poster_path: Optional[str] = None - episode_count: Optional[int] = None - episodes: Optional[List[TMDbEpisodeInfo]] = None - - -@dataclass -class TMDbInfo: - """Information fetched from TMDb.""" - tmdb_id: int - title: Optional[str] = None # Official title from TMDb - original_title: Optional[str] = None - alternative_titles: Optional[List[str]] = None # Titles in other languages - rating: Optional[float] = None - vote_count: Optional[int] = None - overview: Optional[str] = None - genres: Optional[List[str]] = None - release_date: Optional[str] = None - runtime: Optional[int] = None # Minutes for movies - status: Optional[str] = None # Released, Ended, etc. - tagline: Optional[str] = None - poster_path: Optional[str] = None # TMDb poster path - backdrop_path: Optional[str] = None - similar: Optional[List[Dict]] = None # List of similar movies/shows - keywords: Optional[List[str]] = None - cast: Optional[List[Dict]] = None # Top cast members - director: Optional[str] = None # For movies - creators: Optional[List[str]] = None # For TV series - number_of_seasons: Optional[int] = None # For TV series - number_of_episodes: Optional[int] = None # For TV series - networks: Optional[List[str]] = None # For TV series - seasons: Optional[List[TMDbSeasonInfo]] = None # Season details for TV series - - -def tmdb_api_request(endpoint: str, params: Optional[Dict[str, str]] = None) -> Optional[Dict[str, str]]: +async def tmdb_api_request( + endpoint: str, params: Optional[Dict[str, str]] = None +) -> Optional[Dict[str, str]]: """Make a request to the TMDb API with disk caching and connection reuse.""" params = params or {} @@ -162,13 +119,15 @@ def tmdb_api_request(endpoint: str, params: Optional[Dict[str, str]] = None) -> try: client = _get_http_client() - response = client.get(endpoint, params=params) + response = await client.get(endpoint, params=params) if response.status_code == 429: # Rate limited - wait and retry - print(f" Rate limited, waiting...", file=sys.stderr) - time.sleep(1) - return tmdb_api_request(endpoint, {k: v for k, v in params.items() if k != "api_key"}) + print(" Rate limited, waiting...", file=sys.stderr) + await asyncio.sleep(1) + return await tmdb_api_request( + endpoint, {k: v for k, v in params.items() if k != "api_key"} + ) response.raise_for_status() data = response.json() @@ -184,27 +143,28 @@ def tmdb_api_request(endpoint: str, params: Optional[Dict[str, str]] = None) -> return None -def fetch_movie_details(movie_id: int) -> Optional[Dict]: +async def fetch_movie_details(movie_id: int) -> Optional[Dict]: """Fetch detailed movie info including credits, similar, keywords, and alternative titles.""" # Use append_to_response to get multiple data in one request - data = tmdb_api_request( + data = await tmdb_api_request( f"/movie/{movie_id}", - {"append_to_response": "credits,similar,keywords,alternative_titles"} + {"append_to_response": "credits,similar,keywords,alternative_titles"}, ) return data -def fetch_series_details(series_id: int) -> Optional[Dict]: +async def fetch_series_details(series_id: int) -> Optional[Dict]: """Fetch detailed TV series info including credits, similar, and keywords.""" # Use append_to_response to get multiple data in one request - data = tmdb_api_request( - f"/tv/{series_id}", - {"append_to_response": "credits,similar,keywords"} + data = await tmdb_api_request( + f"/tv/{series_id}", {"append_to_response": "credits,similar,keywords"} ) return data -def fetch_season_details(series_id: int, season_number: int) -> Optional[TMDbSeasonInfo]: +async def fetch_season_details( + series_id: int, season_number: int +) -> Optional[TMDbSeasonInfo]: """ Fetch detailed season info including all episodes. @@ -214,9 +174,8 @@ def fetch_season_details(series_id: int, season_number: int) -> Optional[TMDbSea - Runtime, ratings - Directors for each episode """ - data = tmdb_api_request( - f"/tv/{series_id}/season/{season_number}", - {"append_to_response": "images"} + data = await tmdb_api_request( + f"/tv/{series_id}/season/{season_number}", {"append_to_response": "images"} ) if not data: @@ -310,7 +269,21 @@ def _titles_match(original_title: str, tmdb_title: str, search_query: str) -> bo query_words = _normalize_for_match(search_query) # Remove common stop words that don't help matching - stop_words = {"the", "a", "an", "of", "and", "or", "in", "on", "at", "to", "for", "is", "it"} + stop_words = { + "the", + "a", + "an", + "of", + "and", + "or", + "in", + "on", + "at", + "to", + "for", + "is", + "it", + } original_significant = original_words - stop_words tmdb_significant = tmdb_words - stop_words query_significant = query_words - stop_words @@ -331,10 +304,16 @@ def _titles_match(original_title: str, tmdb_title: str, search_query: str) -> bo overlap = original_significant & tmdb_significant # Either good overlap, or the TMDb title is contained in original (or vice versa) - return bool(overlap) or tmdb_significant <= original_significant or original_significant <= tmdb_significant + return ( + bool(overlap) + or tmdb_significant <= original_significant + or original_significant <= tmdb_significant + ) -def _search_movie_with_fallbacks(title: str, year: Optional[int]) -> Optional[Dict]: +async def _search_movie_with_fallbacks( + title: str, year: Optional[int] +) -> Optional[Dict]: """ Search for a movie with progressive title shortening fallbacks. @@ -346,20 +325,25 @@ def _search_movie_with_fallbacks(title: str, year: Optional[int]) -> Optional[Di words = title.split() variants = _generate_title_variants(words, min_words=2) - def _result_matches(top_result: Dict, original_title: str, search_query: str) -> bool: + def _result_matches( + top_result: Dict, original_title: str, search_query: str + ) -> bool: """Check if result matches against either title or original_title.""" tmdb_title = top_result.get("title", "") tmdb_original = top_result.get("original_title", "") - return ( - _titles_match(original_title, tmdb_title, search_query) - or _titles_match(original_title, tmdb_original, search_query) + return _titles_match(original_title, tmdb_title, search_query) or _titles_match( + original_title, tmdb_original, search_query ) # Try all variants with year first if year: for search_title in variants: - params = {"query": search_title, "include_adult": "false", "year": str(year)} - data = tmdb_api_request("/search/movie", params) + params = { + "query": search_title, + "include_adult": "false", + "year": str(year), + } + data = await tmdb_api_request("/search/movie", params) if data and data.get("results"): # Validate the top result matches our title (check both title and original_title) top_result = data["results"][0] @@ -369,7 +353,7 @@ def _search_movie_with_fallbacks(title: str, year: Optional[int]) -> Optional[Di # Then try without year for search_title in variants: params = {"query": search_title, "include_adult": "false"} - data = tmdb_api_request("/search/movie", params) + data = await tmdb_api_request("/search/movie", params) if data and data.get("results"): top_result = data["results"][0] if _result_matches(top_result, title, search_title): @@ -378,9 +362,11 @@ def _search_movie_with_fallbacks(title: str, year: Optional[int]) -> Optional[Di return None -def fetch_movie_info(title: str, year: Optional[int] = None) -> Optional[TMDbInfo]: +async def fetch_movie_info( + title: str, year: Optional[int] = None +) -> Optional[TMDbInfo]: """Fetch comprehensive movie info from TMDb.""" - data = _search_movie_with_fallbacks(title, year) + data = await _search_movie_with_fallbacks(title, year) if not data or not data.get("results"): return None @@ -389,7 +375,7 @@ def fetch_movie_info(title: str, year: Optional[int] = None) -> Optional[TMDbInf movie_id = result["id"] # Fetch full details with credits, similar movies, and keywords - details = fetch_movie_details(movie_id) + details = await fetch_movie_details(movie_id) if not details: # Fall back to basic info from search return TMDbInfo( @@ -429,7 +415,11 @@ def fetch_movie_info(title: str, year: Optional[int] = None) -> Optional[TMDbInf credits = details.get("credits", {}) cast_data = credits.get("cast", [])[:10] cast = [ - {"name": c["name"], "character": c.get("character", ""), "profile_path": c.get("profile_path")} + CastMember( + name=c["name"], + character=c.get("character", ""), + profile_path=c.get("profile_path"), + ) for c in cast_data ] @@ -441,7 +431,7 @@ def fetch_movie_info(title: str, year: Optional[int] = None) -> Optional[TMDbInf # Extract similar movies (limit to 10) similar_data = details.get("similar", {}).get("results", [])[:10] similar = [ - {"id": s["id"], "title": s["title"], "poster_path": s.get("poster_path")} + SimilarMedia(id=s["id"], title=s["title"], poster_path=s.get("poster_path")) for s in similar_data ] @@ -467,7 +457,7 @@ def fetch_movie_info(title: str, year: Optional[int] = None) -> Optional[TMDbInf ) -def _search_series_with_fallbacks(title: str) -> Optional[Dict]: +async def _search_series_with_fallbacks(title: str) -> Optional[Dict]: """ Search for a TV series with progressive title shortening fallbacks. @@ -479,7 +469,7 @@ def _search_series_with_fallbacks(title: str) -> Optional[Dict]: for search_title in variants: params = {"query": search_title, "include_adult": "false"} - data = tmdb_api_request("/search/tv", params) + data = await tmdb_api_request("/search/tv", params) if data and data.get("results"): # Validate the top result matches our title top_result = data["results"][0] @@ -490,9 +480,9 @@ def _search_series_with_fallbacks(title: str) -> Optional[Dict]: return None -def fetch_series_info(title: str) -> Optional[TMDbInfo]: +async def fetch_series_info(title: str) -> Optional[TMDbInfo]: """Fetch comprehensive TV series info from TMDb.""" - data = _search_series_with_fallbacks(title) + data = await _search_series_with_fallbacks(title) if not data or not data.get("results"): return None @@ -501,7 +491,7 @@ def fetch_series_info(title: str) -> Optional[TMDbInfo]: series_id = result["id"] # Fetch full details with credits, similar shows, and keywords - details = fetch_series_details(series_id) + details = await fetch_series_details(series_id) if not details: # Fall back to basic info from search return TMDbInfo( @@ -526,7 +516,11 @@ def fetch_series_info(title: str) -> Optional[TMDbInfo]: credits = details.get("credits", {}) cast_data = credits.get("cast", [])[:10] cast = [ - {"name": c["name"], "character": c.get("character", ""), "profile_path": c.get("profile_path")} + CastMember( + name=c["name"], + character=c.get("character", ""), + profile_path=c.get("profile_path"), + ) for c in cast_data ] @@ -539,7 +533,7 @@ def fetch_series_info(title: str) -> Optional[TMDbInfo]: # Extract similar series (limit to 10) similar_data = details.get("similar", {}).get("results", [])[:10] similar = [ - {"id": s["id"], "title": s["name"], "poster_path": s.get("poster_path")} + SimilarMedia(id=s["id"], title=s["name"], poster_path=s.get("poster_path")) for s in similar_data ] diff --git a/hivescan/utils.py b/hivescan/utils.py index 830e902..359b09e 100644 --- a/hivescan/utils.py +++ b/hivescan/utils.py @@ -14,8 +14,10 @@ _ATIME_FRESHNESS_THRESHOLD = 3600 # Resolution priority for quality sorting (higher = better) RESOLUTION_PRIORITY = { - "2160p": 4, "4K": 4, - "1080p": 3, "1080i": 3, + "2160p": 4, + "4K": 4, + "1080p": 3, + "1080i": 3, "720p": 2, "480p": 1, } @@ -35,7 +37,7 @@ def get_added_timestamp(path: Path) -> Optional[int]: """ try: stat_info = path.stat() - except (OSError, PermissionError): + except OSError, PermissionError: return None if path.is_dir(): @@ -59,7 +61,7 @@ def get_directory_size(path: Path) -> int: for item in path.rglob("*"): if item.is_file(): total += item.stat().st_size - except (OSError, PermissionError): + except OSError, PermissionError: pass return total @@ -126,7 +128,9 @@ def find_common_root(paths: List[Path]) -> Optional[Path]: return Path(*common_parts) -def make_relative_path(path: Optional[str], root: Optional[str] = None) -> Optional[str]: +def make_relative_path( + path: Optional[str], root: Optional[str] = None +) -> Optional[str]: """ Convert an absolute path to a path relative to the given root. @@ -138,16 +142,16 @@ def make_relative_path(path: Optional[str], root: Optional[str] = None) -> Optio return path root_str = str(root).rstrip("/") if path.startswith(root_str): - rel = path[len(root_str):] + rel = path[len(root_str) :] return rel.lstrip("/") return path def sanitize_filename(name: str) -> str: """Sanitize a string for use as a filename.""" - for char in ['/', '\\', ':', '*', '?', '"', '<', '>', '|']: - name = name.replace(char, '_') - name = name.strip('. ') + for char in ["/", "\\", ":", "*", "?", '"', "<", ">", "|"]: + name = name.replace(char, "_") + name = name.strip(". ") return name @@ -159,16 +163,29 @@ def get_media_folder_name(title: str, year: Optional[int], media_type: str) -> s return sanitized_title -def get_media_folder_path(title: str, year: Optional[int], media_type: str, cover_dir: Path) -> Path: +def get_media_folder_path( + title: str, year: Optional[int], media_type: str, cover_dir: Path +) -> Path: """Get the full path to a media item's folder.""" subdir = "movies" if media_type == "movie" else "series" folder_name = get_media_folder_name(title, year, media_type) return cover_dir / subdir / folder_name -def sort_by_quality(items: list[dict], reverse: bool = True) -> None: - """Sort items in-place by resolution quality and size.""" +def sort_by_quality(items: list, reverse: bool = True) -> None: + """Sort items in-place by resolution quality and size. + + Works with both plain dicts (intermediate episode files) and + msgspec.Struct instances (MovieVersion, EpisodeRelease). + """ + + def _val(v, key, default=None): + return v.get(key, default) if isinstance(v, dict) else getattr(v, key, default) + items.sort( - key=lambda v: (RESOLUTION_PRIORITY.get(v.get("resolution", ""), 0), v.get("size", 0) or 0), - reverse=reverse + key=lambda v: ( + RESOLUTION_PRIORITY.get(_val(v, "resolution", "") or "", 0), + _val(v, "size", 0) or 0, + ), + reverse=reverse, ) diff --git a/mediahive-hivescan-merge.md b/mediahive-hivescan-merge.md new file mode 100644 index 0000000..47ce768 --- /dev/null +++ b/mediahive-hivescan-merge.md @@ -0,0 +1,137 @@ +# Hivescan → FastAPI Server Conversion + +> **Status:** Backend complete. Frontend integration and mediahive bridge pending. + +## Architecture + +Two separate FastAPI servers sharing the same codebase, launched independently: + +- **mediahive** (port 8420) — frontend, media serving, play/open-folder actions. *Not yet connected to scanner.* +- **hivescan server** (port 8421) — scanning, TMDb lookups, cover downloads, showreel generation. Owns the in-memory index. Exposes WS for index state + updates. + +Each has its own entry point in `pyproject.toml`: + +``` +hivescan = "hivescan.__main__:main" +mediahive = "mediahive.__main__:main" +``` + +The `hivescan` CLI accepts scan paths as positional args, with `--host` / `--port` / `-o` options. Configuration is passed to the server via `HIVESCAN_PATHS` and `HIVESCAN_OUTPUT` environment variables. + +## In-Memory Index (`hivescan/index_store.py`) + +- `IndexStore` class holds two plain dicts (`movies: dict[str, dict]`, `series: dict[str, dict]`), keyed by item `id`. Single source of truth — all mutations are synchronous in the asyncio event loop, no locks needed. +- `index.json` on disk is a **recovery snapshot** only. Written via a debounced background task (`SNAPSHOT_DEBOUNCE = 5.0` seconds). On startup, `load_snapshot()` populates dicts from disk to avoid full rescan; the scan then runs on top to pick up changes. +- Upsert by item `id`. Items from previous runs are preserved (offline drives). +- `flush_snapshot()` forces an immediate write (called on shutdown). +- Snapshot format: version 5, includes `stats`, sorted `movies` and `series` lists, `media_root`, `generated_at` timestamp. + +## WebSocket Protocol + +The hivescan server exposes `GET /ws`. On connect: + +1. Server calls `IndexStore.connect(ws)` — accepts the socket, adds it to `_clients: set[WebSocket]`, sends the full current state: + ```json + {"type": "init", "data": {"movies": [...], "series": [...]}} + ``` +2. Live pushes on every mutation: + ```json + {"type": "upsert", "kind": "movie", "item": { ... }} + {"type": "upsert", "kind": "series", "item": { ... }} + {"type": "remove", "kind": "movie", "id": "abc123"} + {"type": "task", "data": {"id": "scan-abc12345", "status": "running", "progress": 0.42, "detail": "The Matrix (1999)"}} + ``` +3. Task messages have `status` values: `running`, `completed`, `cancelled`, `error`. +4. Dead connections are cleaned up via `_safe_send()` — failed sends cause the socket to be discarded. + +## HTTP Endpoints + +| Method | Path | Description | +|--------|------|-------------| +| `WS` | `/ws` | Live index updates & task progress | +| `POST` | `/api/scan` | Trigger a new scan (optional `paths` body override). Returns `{"status": "already_running"}` if a scan is in progress. | +| `GET` | `/api/status` | `{scanning, movies, series, showreel_queue}` | +| `GET` | `/api/index` | Full index as JSON (HTTP fallback for non-WS clients) | + +CORS is enabled for all origins. + +## Scanning (asyncio, no threads) + +- `hivescan/indexer.py`: `_process_movies` and `_process_series` are **async generators**. + - `_process_movies` yields `Tuple[dict, Optional[Tuple[str, Path, str]]]` — the movie dict and an optional showreel task `(video_path, media_folder, title)`. + - `_process_series` yields `Tuple[dict, List[Tuple[str, Path, int, int, str]]]` — the series dict and a list of episode reel tasks `(video_path, media_folder, season_num, episode_num, series_title)`. +- TMDb HTTP calls use `httpx.AsyncClient` (lazy-initialised singleton). Disk cache stays synchronous (fast local I/O). +- Cover/backdrop downloads use a separate `httpx.AsyncClient` in `images.py`. +- File scanning (`Path.rglob`, `stat`, `Path.iterdir`): runs synchronously in the event loop (fast enough). Glob expansion for scan paths happens at startup in the lifespan handler. +- Each yielded item is upserted into the in-memory dict (sync) and broadcast to WS clients (non-blocking `asyncio.create_task` per client send). +- `print()` calls throughout replaced with `logging.getLogger()`. + +## Showreel Generation + +- Background `_showreel_worker` task started at server startup, runs for the lifetime of the server. +- Uses an `asyncio.Queue` — scan tasks push `("movie", task_data, item_id)` or `("episode", task_data, item_id)` tuples. +- Processes one task at a time to avoid saturating CPU/GPU. +- Invokes ffmpeg/ffprobe via `asyncio.create_subprocess_exec` with `asyncio.wait_for` timeouts (replaced all `subprocess.run` calls). +- `generate_showreel_images` accepts an `on_progress` callback (replaced `tqdm` progress bar parameter). +- `generate_episode_reel` no longer takes a progress bar parameter. +- On completion: updates the relevant item in the in-memory dict (`showreel_images` for movies, `reel_image` for episodes) → re-upserts → WS broadcast. +- Skips tasks where showreels/reels already exist on disk. + +## File Changes + +### Modified + +| File | Changes | +|---|---| +| `indexer.py` | `_process_movies` / `_process_series` → async generators yielding `(item_dict, showreel_tasks)`. `_build_seasons_data` → async. Removed `generate_media_index()` and `_run_showreel_generation()` (server orchestrates now). Removed `json`, `datetime`, `tqdm` imports. Added `logging`. `print()` → `logger.info()`. | +| `tmdb_client.py` | `httpx.Client` → `httpx.AsyncClient` (lazy singleton via `_get_http_client()`). All public functions async: `tmdb_api_request`, `fetch_movie_details`, `fetch_series_details`, `fetch_season_details`, `fetch_movie_info`, `fetch_series_info`, `_search_movie_with_fallbacks`, `_search_series_with_fallbacks`. `time.sleep(1)` → `await asyncio.sleep(1)`. Disk cache functions unchanged. | +| `images.py` | `urllib.request` → `httpx.AsyncClient` (lazy singleton via `_get_image_client()`). All functions async: `_download_image`, `download_cover_image`, `download_backdrop_image`, `download_season_poster`. | +| `showreel.py` | `subprocess.run` → `asyncio.create_subprocess_exec` + `asyncio.wait_for`. All ffmpeg/ffprobe functions async: `get_av1_encoder`, `detect_dovi_profile`, `is_hdr_video`, `detect_crop`, `get_video_duration`, `generate_showreel_images`, `generate_episode_reel`. `tqdm` removed. `print()` → `logger.debug/info/warning/error`. Sync helper functions unchanged (`get_expected_showreel_paths`, `get_expected_episode_reel_path`, `movie_showreels_exist`, `episode_reel_exists`, `get_bluray_uri`, `get_encoder_options`, `get_dovi_to_hdr10_filter`). | +| `__main__.py` | Was CLI argparse with `--no-showreels`/`--no-covers` and direct scan execution. Now a server launcher: positional `paths` args, `--host` (default `0.0.0.0`), `--port` (default `8421`), `-o`/`--output-dir`. Sets `HIVESCAN_PATHS`/`HIVESCAN_OUTPUT` env vars and calls `hivescan.server.run()`. | +| `__init__.py` | Replaced `generate_media_index` export with `IndexStore`. Updated module docstring. | +| `pyproject.toml` | Removed `tqdm>=4.67.3` from dependencies. Entry points unchanged. | + +### New + +| File | Purpose | +|---|---| +| `hivescan/index_store.py` | `IndexStore` class (~200 lines): in-memory dict storage, `upsert_movie/series`, `remove_movie/series`, `load_snapshot`, `flush_snapshot`, debounced `_write_snapshot`, WS client management (`connect`, `disconnect`), `_broadcast` with `_safe_send`, `broadcast_task`, `get_full_index`. | +| `hivescan/server.py` | FastAPI app (~350 lines): lifespan handler (env config, glob expansion, store init, auto-scan), WS endpoint, HTTP endpoints (`/api/scan`, `/api/status`, `/api/index`), `_run_scan` orchestrator (iterates async generators, upserts, queues showreels, broadcasts progress), `_showreel_worker` (queue consumer), `run()` standalone entry. | + +### Unchanged + +`scanning.py`, `parsing.py`, `models.py`, `utils.py` — all remain synchronous. + +## Plan: mediahive Integration + +### Data flow + +``` +hivescan (8421) ──WS──▶ mediahive (8420) ──WS──▶ browser + keeps its own + in-memory dict +``` + +mediahive keeps its own in-memory copy of the index (`movies: dict`, `series: dict`), populated from one of two sources: + +1. **hivescan WS** (preferred) — on startup, mediahive connects to `ws://localhost:8421/ws`. The `init` message seeds the dicts, then `upsert`/`remove` messages keep them in sync. Task messages are forwarded to browser clients as-is. +2. **Disk fallback** — if hivescan is not running (connection refused / WS drops and doesn't reconnect), load `index.json` from disk once as a static snapshot. The frontend still works, just without live updates. + +Auto-reconnect: if the WS drops, mediahive retries on a backoff. When it reconnects, it gets a fresh `init` and replaces its dicts entirely. + +### `mediahive/server.py` changes + +- **New WS client task** started in `lifespan`. Connects to hivescan, handles messages, updates in-memory dicts. On failure, loads `index.json` fallback. +- **New WS endpoint** `GET /ws` for browser clients. On connect: send `{"type": "init", "data": {movies, series}}` from the local dicts. Forward `upsert`, `remove`, `task` messages as they arrive from hivescan. +- **Existing `GET /api/index`** — return from in-memory dicts instead of reading `index.json` each time. Keeps working for non-WS clients. Remove the `linux_to_windows_path` conversion (paths are already relative). +- Keep `play`, `open-folder`, `media/{path}` endpoints unchanged. + +### Frontend changes + +| File | Change | +|---|---| +| `composables/useWebSocket.ts` (new) | Connect to mediahive's `/ws` (same origin, no CORS). On `init`: replace reactive index. On `upsert`: merge item by id. On `remove`: delete by id. On `task`: update reactive task state. Auto-reconnect with backoff. | +| `types.ts` | Add `WsMessage`, `TaskInfo` types. | +| `App.vue` | Replace `loadMediaIndex()` fetch with the WS composable. Items appear incrementally as scan runs. Fall back to `GET /api/index` if WS never connects. | +| `Header.vue` | Show task indicator (spinner + progress text) when scan/showreel tasks are active. | +| `api.ts` | Remove `loadMediaIndex()`. Keep `playMedia()`, `openFolder()`, `getCoverUrl()`. | diff --git a/mediahive/server.py b/mediahive/server.py index c6ba11e..458998c 100644 --- a/mediahive/server.py +++ b/mediahive/server.py @@ -12,11 +12,13 @@ from contextlib import asynccontextmanager from pathlib import Path import aiofiles -from fastapi import FastAPI, HTTPException +import msgspec +from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, StreamingResponse from fastapi_vue import Frontend -from pydantic import BaseModel + +from hivescan.structs import PlayMediaRequest, OpenFolderRequest from mediahive.__main__ import DEVMODE @@ -65,17 +67,6 @@ def normalize_path(url_path: str) -> Path: return MEDIAROOT / clean_path -# === API Models === - - -class PlayMediaRequest(BaseModel): - file_path: str - - -class OpenFolderRequest(BaseModel): - folder_path: str - - # === API Endpoints === @@ -153,18 +144,17 @@ async def load_media_index(): @app.post("/api/play") -async def play_media(request: PlayMediaRequest): +async def play_media(request: Request): """ Open a media file with the system's default player. """ - print(f"[play] Received path: {request.file_path}") - file_path = MEDIAROOT / request.file_path + req = msgspec.json.decode(await request.body(), type=PlayMediaRequest) + print(f"[play] Received path: {req.file_path}") + file_path = MEDIAROOT / req.file_path if not file_path.exists(): print(f"[play] File not found: {file_path}") - raise HTTPException( - status_code=404, detail=f"File not found: {request.file_path}" - ) + raise HTTPException(status_code=404, detail=f"File not found: {req.file_path}") try: # Use os.startfile on Windows (non-blocking) @@ -182,18 +172,19 @@ async def play_media(request: PlayMediaRequest): @app.post("/api/open-folder") -async def open_folder(request: OpenFolderRequest): +async def open_folder(request: Request): """ Open a folder in the system file explorer. If the path is a file, opens the parent folder and selects the file. """ - print(f"[open-folder] Received path: {request.folder_path}") - target_path = MEDIAROOT / request.folder_path + req = msgspec.json.decode(await request.body(), type=OpenFolderRequest) + print(f"[open-folder] Received path: {req.folder_path}") + target_path = MEDIAROOT / req.folder_path if not target_path.exists(): print(f"[open-folder] Path not found: {target_path}") raise HTTPException( - status_code=404, detail=f"Path not found: {request.folder_path}" + status_code=404, detail=f"Path not found: {req.folder_path}" ) try: diff --git a/pyproject.toml b/pyproject.toml index a68acb2..ad337ea 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,8 +10,8 @@ dependencies = [ "fastapi-vue>=0.5.2", "fastapi[standard]>=0.128.0", "httpx[http2]>=0.28.1", + "msgspec>=0.19", "parse-torrent-title>=2.8.1", - "tqdm>=4.67.3", "uvicorn[standard]>=0.40.0", ] diff --git a/rtorrent_client.py b/rtorrent_client.py index 1a6ea45..153d833 100644 --- a/rtorrent_client.py +++ b/rtorrent_client.py @@ -24,7 +24,7 @@ class SCGITransport(xmlrpc.client.Transport): # Connect to socket sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.connect(self.socket_path) - sock.send(request.encode('utf-8')) + sock.send(request.encode("utf-8")) # Read response response = b"" @@ -54,7 +54,9 @@ class RTorrentClient: def __init__(self, socket_path: str = "/home/user/rtorrent/.session/rpc.socket"): self.socket_path = socket_path transport = SCGITransport(socket_path) - self.proxy = xmlrpc.client.ServerProxy("http://localhost/RPC2", transport=transport) + self.proxy = xmlrpc.client.ServerProxy( + "http://localhost/RPC2", transport=transport + ) def get_loaded_hashes(self) -> set[str]: """Get set of info hashes for all currently loaded torrents.""" @@ -81,9 +83,7 @@ class RTorrentClient: # load.start_verbose with d.directory.set to specify download location # This will hash-check existing files instead of re-downloading self.proxy.load.start_verbose( - "", - str(torrent_path), - f"d.directory.set=\"{download_dir}\"" + "", str(torrent_path), f'd.directory.set="{download_dir}"' ) return True except Exception as e: @@ -125,8 +125,10 @@ class RTorrentClient: for info_hash in hashes: try: message = self.proxy.d.message(info_hash) - if message and ("unregistered" in message.lower() or - "not registered" in message.lower()): + if message and ( + "unregistered" in message.lower() + or "not registered" in message.lower() + ): info = self.get_torrent_info(info_hash) if info: unregistered.append(info) @@ -157,4 +159,4 @@ class RTorrentClient: return True except Exception as e: print(f"Error removing torrent {info_hash}: {e}") - return False \ No newline at end of file + return False diff --git a/scripts/fastapi-vue/buildutil.py b/scripts/fastapi-vue/buildutil.py index 533c46e..db1ad46 100644 --- a/scripts/fastapi-vue/buildutil.py +++ b/scripts/fastapi-vue/buildutil.py @@ -31,7 +31,7 @@ def _check_node_version(node_path: str) -> None: """ try: result = subprocess.run( - [node_path, "--version"], capture_output=True, text=True, check=True + [node_path, "--version"], capture_output=True, encoding="utf-8", check=True ) version_str = result.stdout.strip() # Parse version like "v20.10.0" or "v18.17.1" @@ -43,7 +43,7 @@ def _check_node_version(node_path: str) -> None: raise RuntimeError( f"Node.js {version_str} found, but v20+ required (install with nvm)" ) - except (subprocess.CalledProcessError, FileNotFoundError, ValueError): + except subprocess.CalledProcessError, FileNotFoundError, ValueError: pass raise RuntimeError("Could not determine Node.js version") diff --git a/scripts/rtorrent-manager.py b/scripts/rtorrent-manager.py index c34bdc7..f44db8e 100644 --- a/scripts/rtorrent-manager.py +++ b/scripts/rtorrent-manager.py @@ -18,6 +18,7 @@ from rtorrent_client import RTorrentClient @dataclass class TorrentInfo: """Information extracted from a torrent file.""" + path: Path name: str trackers: list[str] @@ -88,7 +89,7 @@ def parse_torrent(filepath: Path) -> TorrentInfo | None: TorrentInfo object or None if parsing fails """ try: - with open(filepath, 'rb') as f: + with open(filepath, "rb") as f: data = bencodepy.decode(f.read()) except Exception as e: print(f"Error parsing {filepath}: {e}") @@ -98,23 +99,23 @@ def parse_torrent(filepath: Path) -> TorrentInfo | None: trackers = [] # Main announce URL - if b'announce' in data: - announce = data[b'announce'] + if b"announce" in data: + announce = data[b"announce"] if isinstance(announce, bytes): - trackers.append(announce.decode('utf-8', errors='replace')) + trackers.append(announce.decode("utf-8", errors="replace")) # Announce list (multiple trackers) - if b'announce-list' in data: - for tier in data[b'announce-list']: + if b"announce-list" in data: + for tier in data[b"announce-list"]: for tracker in tier: if isinstance(tracker, bytes): - url = tracker.decode('utf-8', errors='replace') + url = tracker.decode("utf-8", errors="replace") if url not in trackers: trackers.append(url) # Extract name - info = data.get(b'info', {}) - name = info.get(b'name', b'Unknown').decode('utf-8', errors='replace') + info = data.get(b"info", {}) + name = info.get(b"name", b"Unknown").decode("utf-8", errors="replace") # Calculate info hash info_hash = hashlib.sha1(bencodepy.encode(info)).hexdigest().upper() @@ -124,23 +125,22 @@ def parse_torrent(filepath: Path) -> TorrentInfo | None: files = None is_multi_file = False - if b'length' in info: + if b"length" in info: # Single file torrent - size = info[b'length'] + size = info[b"length"] files = [name] is_multi_file = False - elif b'files' in info: + elif b"files" in info: # Multi-file torrent files = [] size = 0 is_multi_file = True - for file_info in info[b'files']: - file_path = '/'.join( - p.decode('utf-8', errors='replace') - for p in file_info.get(b'path', []) + for file_info in info[b"files"]: + file_path = "/".join( + p.decode("utf-8", errors="replace") for p in file_info.get(b"path", []) ) files.append(file_path) - size += file_info.get(b'length', 0) + size += file_info.get(b"length", 0) return TorrentInfo( path=filepath, @@ -171,8 +171,9 @@ def scan_torrent_directories(paths: list[str]) -> Iterator[Path]: yield torrent_file -def find_torrents_with_tracker(tracker_domain: str, - paths: list[str]) -> list[TorrentInfo]: +def find_torrents_with_tracker( + tracker_domain: str, paths: list[str] +) -> list[TorrentInfo]: """ Find all torrents that have a specific tracker domain. @@ -198,7 +199,7 @@ def format_size(size_bytes: int | None) -> str: if size_bytes is None: return "Unknown" - for unit in ['B', 'KB', 'MB', 'GB', 'TB']: + for unit in ["B", "KB", "MB", "GB", "TB"]: if size_bytes < 1024: return f"{size_bytes:.2f} {unit}" size_bytes /= 1024 @@ -223,9 +224,16 @@ Examples: nargs="+", help="Directories or glob patterns containing .torrent files", ) - parser.add_argument("--dry", action="store_true", help="Dry run - show what would be done without making changes") - parser.add_argument("--tracker", default="hdbits.org", - help="Tracker domain to filter by (default: hdbits.org)") + parser.add_argument( + "--dry", + action="store_true", + help="Dry run - show what would be done without making changes", + ) + parser.add_argument( + "--tracker", + default="hdbits.org", + help="Tracker domain to filter by (default: hdbits.org)", + ) args = parser.parse_args() dry_run = args.dry @@ -245,7 +253,7 @@ Examples: print("DRY RUN MODE - No changes will be made") print("=" * 60) - print(f"Scanning for torrents...") + print("Scanning for torrents...") print(f"Search paths: {expanded_paths}") print("-" * 60) @@ -261,18 +269,18 @@ Examples: without_hdbits = [t for t in all_torrents if not t.has_tracker(tracker_domain)] # Print stats - print(f"\n{'='*60}") - print(f"SUMMARY") - print(f"{'='*60}") + print(f"\n{'=' * 60}") + print("SUMMARY") + print(f"{'=' * 60}") print(f"Total torrents scanned: {len(all_torrents)}") print(f"With {tracker_domain}: {len(with_hdbits)}") print(f"Without {tracker_domain}: {len(without_hdbits)}") # Add hdbits torrents to rtorrent if with_hdbits: - print(f"\n{'='*60}") - print(f"VERIFYING DOWNLOADS & ADDING TO RTORRENT") - print(f"{'='*60}") + print(f"\n{'=' * 60}") + print("VERIFYING DOWNLOADS & ADDING TO RTORRENT") + print(f"{'=' * 60}") # First, verify which torrents have their data verified = [] @@ -285,15 +293,15 @@ Examples: else: missing_data.append((torrent, message)) - print(f"\nVerification results:") + print("\nVerification results:") print(f" Downloads found: {len(verified)}") print(f" Downloads missing: {len(missing_data)}") # Report missing downloads if missing_data: - print(f"\n{'='*60}") - print(f"TORRENTS WITH MISSING DATA (will not add)") - print(f"{'='*60}") + print(f"\n{'=' * 60}") + print("TORRENTS WITH MISSING DATA (will not add)") + print(f"{'=' * 60}") for torrent, message in missing_data: print(f"\n Name: {torrent.name}") print(f" Torrent: {torrent.path}") @@ -301,9 +309,9 @@ Examples: # Now add verified torrents to rtorrent if verified: - print(f"\n{'='*60}") + print(f"\n{'=' * 60}") print(f"ADDING {len(verified)} VERIFIED TORRENTS TO RTORRENT") - print(f"{'='*60}") + print(f"{'=' * 60}") client = RTorrentClient() loaded_hashes = client.get_loaded_hashes() @@ -334,12 +342,14 @@ Examples: if dry_run: print(f"\nDry run: {added} would be added, {skipped} already loaded") else: - print(f"\nRtorrent results: {added} added, {skipped} skipped, {failed} failed") + print( + f"\nRtorrent results: {added} added, {skipped} skipped, {failed} failed" + ) # Clean up unregistered torrents from rtorrent - print(f"\n{'='*60}") - print(f"CHECKING FOR UNREGISTERED TORRENTS") - print(f"{'='*60}") + print(f"\n{'=' * 60}") + print("CHECKING FOR UNREGISTERED TORRENTS") + print(f"{'=' * 60}") client = RTorrentClient() unregistered = client.get_unregistered_torrents() @@ -353,7 +363,9 @@ Examples: for torrent_info in unregistered: # Determine the download path (base_path is the actual file/folder) - download_path = Path(torrent_info['base_path']) if torrent_info['base_path'] else None + download_path = ( + Path(torrent_info["base_path"]) if torrent_info["base_path"] else None + ) if dry_run: status = "[DRY]" @@ -363,11 +375,11 @@ Examples: print(f" {status} {torrent_info['name']} (no data path)") else: # Remove from rtorrent (keeps downloaded files) - if client.remove_torrent(torrent_info['hash']): + if client.remove_torrent(torrent_info["hash"]): removed_from_rtorrent += 1 # Delete the .torrent file if it exists - tied_file = torrent_info['tied_file'] + tied_file = torrent_info["tied_file"] if tied_file: torrent_file = Path(tied_file) if torrent_file.exists(): @@ -391,42 +403,48 @@ Examples: else: print(f" [DEL] {torrent_info['name']} (no data)") else: - print(f" [ERR] {torrent_info['name']}: failed to remove from rtorrent") + print( + f" [ERR] {torrent_info['name']}: failed to remove from rtorrent" + ) print() if dry_run: - print(f"Dry run: {len(unregistered)} would be removed (rtorrent + .torrent + downloads)") + print( + f"Dry run: {len(unregistered)} would be removed (rtorrent + .torrent + downloads)" + ) else: - print(f"Cleanup: {removed_from_rtorrent} from rtorrent, {removed_torrent_files} .torrents, {removed_downloads} downloads") + print( + f"Cleanup: {removed_from_rtorrent} from rtorrent, {removed_torrent_files} .torrents, {removed_downloads} downloads" + ) else: print("No unregistered torrents found.") # List torrents without hdbits.org if without_hdbits: - print(f"\n{'='*60}") + print(f"\n{'=' * 60}") print(f"TORRENTS WITHOUT {tracker_domain.upper()}") - print(f"{'='*60}") + print(f"{'=' * 60}") for torrent in without_hdbits: print(f"\nName: {torrent.name}") print(f"Path: {torrent.path}") print(f"Size: {format_size(torrent.size)}") if torrent.trackers: - print(f"Trackers:") + print("Trackers:") for tracker in torrent.trackers: print(f" - {tracker}") else: print("Trackers: (none)") # Remove the non-hdbits torrent files - print(f"\n{'='*60}") + print(f"\n{'=' * 60}") if dry_run: print(f"WOULD REMOVE {len(without_hdbits)} TORRENT FILE(S)") - print(f"{'='*60}") + print(f"{'=' * 60}") for torrent in without_hdbits: print(f"Would remove: {torrent.path}") else: print(f"REMOVING {len(without_hdbits)} TORRENT FILE(S)") - print(f"{'='*60}") + print(f"{'=' * 60}") for torrent in without_hdbits: try: torrent.path.unlink()