diff --git a/mediahive/__main__.py b/mediahive/__main__.py index dfe3dfa..bad8c8a 100644 --- a/mediahive/__main__.py +++ b/mediahive/__main__.py @@ -10,95 +10,42 @@ DEVMODE = bool(os.getenv("MEDIAHIVE_FRONTEND_URL")) def main(): parser = argparse.ArgumentParser(description="MediaHive - Media scanning, indexing, and streaming") - subparsers = parser.add_subparsers(dest='command', required=True, help='Available commands') - - # Server subcommand - server_parser = subparsers.add_parser('server', help='Run the MediaHive streaming server') - server_parser.add_argument( + parser.add_argument( "media_folder", nargs="?", help="Path to the media folder (default: MEDIAHIVE_PATH or current directory)", ) - server_parser.add_argument( + parser.add_argument( "-l", "--listen", action="append", help=(f"Endpoint (default: localhost:{DEFAULT_PORT})."), ) - # Scan subcommand - scan_parser = subparsers.add_parser('scan', help='Run the Hivescan media scanning server', - description="Hivescan server — continuous media scanning with live WS updates.", - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" -Examples: - mediahive scan /path/to/torrents/* # Scan paths, auto-detect common root - mediahive scan /mnt/disk1/* /mnt/disk2/* # Scan multiple locations - mediahive scan /torrents/* -o /srv/media # Override output directory - mediahive scan /torrents/* --port 9000 # Custom port - -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) - """) - scan_parser.add_argument( - "paths", - nargs="+", - help="Folders or glob patterns to scan for downloads", - ) - scan_parser.add_argument( - "-o", - "--output-dir", - metavar="DIR", - help="Output directory for index and covers (default: .mediahive at common root)", - ) - scan_parser.add_argument( - "--host", - default="0.0.0.0", - help="Host to bind to (default: 0.0.0.0)", - ) - scan_parser.add_argument( - "--port", - type=int, - default=8421, - help="Port to listen on (default: 8421)", - ) - args = parser.parse_args() - if args.command == 'server': - # Determine media folder - match Path( - args.media_folder or os.environ.get("MEDIAHIVE_PATH") or Path.cwd() - ).parts: - case (*rest, ".mediahive", "index.json"): - ... - case (*rest, ".mediahive"): - ... - case rest: - ... - mediaroot = Path(*rest).resolve() - if not mediaroot.exists() or not mediaroot.is_dir(): - print(f"Error: Folder does not exist: {mediaroot}") - exit(1) - os.environ["MEDIAHIVE_PATH"] = mediaroot.as_posix() - dev = {"reload": True, "reload_dirs": ["mediahive"]} - server.run( - "mediahive.server:app", - listen=args.listen, - default_port=DEFAULT_PORT, - **(dev if DEVMODE else {}), - ) - elif args.command == 'scan': - # Pass configuration via environment variables (read by server.py lifespan) - os.environ["HIVESCAN_PATHS"] = os.pathsep.join(args.paths) - if args.output_dir: - os.environ["HIVESCAN_OUTPUT"] = args.output_dir - - from mediahive.hivescan.server import run - run(host=args.host, port=args.port) + # Determine media folder + match Path( + args.media_folder or os.environ.get("MEDIAHIVE_PATH") or Path.cwd() + ).parts: + case (*rest, ".mediahive", "index.json"): + ... + case (*rest, ".mediahive"): + ... + case rest: + ... + mediaroot = Path(*rest).resolve() + if not mediaroot.exists() or not mediaroot.is_dir(): + print(f"Error: Folder does not exist: {mediaroot}") + exit(1) + os.environ["MEDIAHIVE_PATH"] = mediaroot.as_posix() + dev = {"reload": True, "reload_dirs": ["mediahive"]} + server.run( + "mediahive.server:app", + listen=args.listen, + default_port=DEFAULT_PORT, + **(dev if DEVMODE else {}), + ) if __name__ == "__main__": diff --git a/mediahive/hivescan/__init__.py b/mediahive/hivescan/__init__.py index a90acde..4303863 100644 --- a/mediahive/hivescan/__init__.py +++ b/mediahive/hivescan/__init__.py @@ -1,13 +1,13 @@ """ -Hivescan - Continuous media scanning server with live WebSocket updates. +Hivescan - Continuous media scanning with live WebSocket updates. -Usage: - hivescan /path/to/torrents/* # Start scanning server - hivescan /path/* --port 9000 # Custom port +Usage as a module: + python -m mediahive.hivescan /path/to/torrents/* + python -m mediahive.hivescan /path/* --port 9000 Or as a library: - from mediahive.models.data import Movie, Series, TaskInfo - from mediahive.hivescan.server import app + from mediahive.hivescan.scanning import scan_downloads, categorize_downloads + from mediahive.hivescan.indexer import _process_movies, _process_series """ from mediahive.hivescan.models import ContentType, ContentHash, ParsedContent @@ -17,9 +17,9 @@ from mediahive.hivescan.scanning import ( find_playable_file, find_episode_files, ) -from mediahive.hivescan.index_store import IndexStore from mediahive.hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root from mediahive.hivescan.showreel import generate_showreel_images, generate_episode_reel +from mediahive.hivescan import scanner from mediahive.models.data import ( Episode, IndexSnapshot, @@ -47,8 +47,6 @@ __all__ = [ "categorize_downloads", "find_playable_file", "find_episode_files", - # Index store - "IndexStore", # Struct types "CastMember", "Episode", diff --git a/mediahive/hivescan/__main__.py b/mediahive/hivescan/__main__.py new file mode 100644 index 0000000..c6aa03c --- /dev/null +++ b/mediahive/hivescan/__main__.py @@ -0,0 +1,88 @@ +import argparse +import asyncio +import glob +import logging +import os +from pathlib import Path + +from mediahive.hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root + + +def main(): + parser = argparse.ArgumentParser( + description="Hivescan server — continuous media scanning with live WS updates.", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + python -m mediahive.hivescan /path/to/torrents/* # Scan paths, auto-detect common root + python -m mediahive.hivescan /mnt/disk1/* /mnt/disk2/* # Scan multiple locations + python -m mediahive.hivescan /torrents/* -o /srv/media # Override output directory + python -m mediahive.hivescan /torrents/* --port 9000 # Custom port + +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) + """ + ) + parser.add_argument( + "paths", + nargs="+", + help="Folders or glob patterns to scan for downloads", + ) + parser.add_argument( + "-o", + "--output-dir", + metavar="DIR", + help="Output directory for index and covers (default: .mediahive at common root)", + ) + parser.add_argument( + "--host", + default="0.0.0.0", + help="Host to bind to (default: 0.0.0.0)", + ) + parser.add_argument( + "--port", + type=int, + default=8421, + help="Port to listen on (default: 8421)", + ) + + args = parser.parse_args() + + # Derive the media root so we can set MEDIAHIVE_PATH + all_paths: list[Path] = [] + for pattern in args.paths: + expanded = glob.glob(pattern) + if expanded: + all_paths.extend(Path(p) for p in expanded) + else: + all_paths.append(Path(pattern)) + + if args.output_dir: + media_root = Path(args.output_dir).parent + else: + media_root = asyncio.run(find_common_root(all_paths)) + if media_root is None: + print("Error: Cannot determine common root; use -o to set output directory") + exit(1) + + # Configure environment for the mediahive server + scanner + os.environ["MEDIAHIVE_PATH"] = str(media_root.resolve()) + os.environ["HIVESCAN_PATHS"] = os.pathsep.join(args.paths) + if args.output_dir: + os.environ["HIVESCAN_OUTPUT"] = args.output_dir + + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + datefmt="%H:%M:%S", + ) + + import uvicorn + uvicorn.run("mediahive.server:app", host=args.host, port=args.port, log_level="info") + + +if __name__ == "__main__": + main() diff --git a/mediahive/hivescan/scanner.py b/mediahive/hivescan/scanner.py new file mode 100644 index 0000000..fe44797 --- /dev/null +++ b/mediahive/hivescan/scanner.py @@ -0,0 +1,339 @@ +""" +Scan orchestration — background tasks for continuous media scanning. + +All scanning logic lives here in hivescan. Communication with the mediahive +server happens exclusively through an async ``send`` callable that pushes +:class:`~mediahive.models.protocol.ScanEvent` messages (``EvUpsert`` / +``EvTask``) onto an :class:`asyncio.Queue` owned by the caller. +""" + +import asyncio +import glob +import logging +import os +import uuid +from pathlib import Path +from typing import Awaitable, Callable, List, Optional + +from aiopathlib import AsyncPath + +from mediahive.hivescan.indexer import _process_movies, _process_series +from mediahive.hivescan.models import ContentType, ParsedContent +from mediahive.hivescan.parsing import parse_download +from mediahive.hivescan.scanning import categorize_downloads +from mediahive.hivescan.showreel import ( + episode_reel_exists, + generate_episode_reel, + generate_showreel_images, + get_expected_showreel_paths, + movie_showreels_exist, +) +from mediahive.hivescan.tmdb_client import set_cache_dir +from mediahive.hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root, make_relative_path +from mediahive.models.data import Movie, Series, TaskInfo +from mediahive.models.protocol import EvTask, EvUpsert, ScanEvent + +logger = logging.getLogger("hivescan.scanner") + +# Type alias for the send callable +Send = Callable[[ScanEvent], Awaitable[None]] + + +# --------------------------------------------------------------------------- +# Configuration (populated by ``start``) +# --------------------------------------------------------------------------- + +_scan_paths: List[str] = [] +_output_dir: Optional[Path] = None +_media_root: Optional[Path] = None + +# Runtime state +_send: Optional[Send] = 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 +_seen_mtimes: dict[str, int] = {} + + +# --------------------------------------------------------------------------- +# Public lifecycle +# --------------------------------------------------------------------------- + + +async def start(send: Send) -> None: + """ + Initialise and start the scanner. + + Reads ``HIVESCAN_PATHS`` / ``HIVESCAN_OUTPUT`` from the environment, + expands globs, sets up the TMDb cache, and starts background workers. + """ + global _send, _output_dir, _media_root, _scan_paths + global _showreel_worker_task, _rescan_worker_task + + _send = send + + raw_paths = os.environ.get("HIVESCAN_PATHS", "") + if not raw_paths: + logger.error("HIVESCAN_PATHS environment variable must be set") + return + + all_paths: List[Path] = [] + for pattern in raw_paths.split(os.pathsep): + pattern = pattern.strip() + if not pattern: + continue + expanded = await asyncio.to_thread(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 = await find_common_root(all_paths) + if _media_root is None: + logger.error("Cannot determine common root; set HIVESCAN_OUTPUT") + return + _output_dir = _media_root / DEFAULT_OUTPUT_FOLDER + + await AsyncPath(_output_dir).mkdir(parents=True, exist_ok=True) + set_cache_dir(_output_dir / ".tmdb-cache") + + logger.info( + "Scanner started — %d scan paths, output=%s", + len(_scan_paths), _output_dir, + ) + + _showreel_worker_task = asyncio.create_task(_showreel_worker()) + _rescan_worker_task = asyncio.create_task(_rescan_loop()) + + +async def stop() -> None: + """Cancel all background tasks.""" + for task in (_scan_task, _showreel_worker_task, _rescan_worker_task): + if task and not task.done(): + task.cancel() + + +def is_scanning() -> bool: + return _scan_task is not None and not _scan_task.done() + + +def showreel_queue_size() -> int: + return _showreel_queue.qsize() + + +def trigger_scan(paths: Optional[List[str]] = None) -> bool: + """Start a scan. Returns False if one is already running.""" + if is_scanning(): + return False + _start_scan(paths) + return True + + +# --------------------------------------------------------------------------- +# Internal scan orchestration +# --------------------------------------------------------------------------- + + +def _start_scan(paths: Optional[List[str]] = None): + global _scan_task + _scan_task = asyncio.create_task(_run_scan(paths)) + + +async def _rescan_loop(): + try: + while True: + _start_scan() + if _scan_task: + await _scan_task + await asyncio.sleep(1) + except asyncio.CancelledError: + return + except Exception: + logger.exception("Rescan loop error") + + +async def _discover_downloads(paths_to_scan: List[str]) -> List[ParsedContent]: + """Walk the filesystem and parse all downloads, skipping unchanged torrents.""" + downloads: List[ParsedContent] = [] + media_root_str = str(_media_root) if _media_root else None + for pattern in paths_to_scan: + p = Path(pattern) + ap = AsyncPath(p) + if await ap.is_dir(): + for item in ap.iterdir(): + if not Path(item).name.startswith("."): + relpath = make_relative_path(str(item), media_root_str) + stat_info = await AsyncPath(item).stat() + mtime = int(stat_info.st_mtime) + if relpath in _seen_mtimes and _seen_mtimes[relpath] == mtime: + continue + _seen_mtimes[relpath] = mtime + downloads.append(await parse_download(Path(item))) + elif await ap.exists(): + relpath = make_relative_path(str(p), media_root_str) + stat_info = await ap.stat() + mtime = int(stat_info.st_mtime) + if relpath in _seen_mtimes and _seen_mtimes[relpath] == mtime: + continue + _seen_mtimes[relpath] = mtime + downloads.append(await parse_download(p)) + return downloads + + +async def _run_scan(override_paths: Optional[List[str]] = None): + """ + Full scan pipeline: + 1. Walk filesystem, parse torrents + 2. Categorise → movies / series + 3. Iterate async generators, send each item as EvUpsert + 4. Queue showreel tasks + """ + task_id = f"scan-{uuid.uuid4().hex[:8]}" + paths_to_scan = override_paths or _scan_paths + media_root_str = str(_media_root) if _media_root else None + + try: + downloads = await _discover_downloads(paths_to_scan) + + if downloads: + logger.info("Scan started (%s)", task_id) + await _send(EvTask(data=TaskInfo( + id=task_id, status="running", progress=0, detail="Scanning filesystem...", + ))) + logger.info("Found %d items to process", len(downloads)) + + categories = categorize_downloads(downloads) + total = len(categories[ContentType.MOVIE]) + len(categories[ContentType.SERIES]) + processed = 0 + + # Process movies + await _send(EvTask(data=TaskInfo( + id=task_id, status="running", progress=0, detail="Processing movies...", + ))) + async for movie, showreel_task in _process_movies( + categories, _output_dir, + fetch_covers=True, generate_showreels=True, media_root=media_root_str, + ): + await _send(EvUpsert(kind="movie", item=movie)) + if showreel_task: + await _showreel_queue.put(("movie", showreel_task, movie)) + processed += 1 + progress = round(processed / total, 3) if total else 1 + await _send(EvTask(data=TaskInfo( + id=task_id, status="running", progress=progress, detail=movie.title, + ))) + + # Process series + await _send(EvTask(data=TaskInfo( + id=task_id, status="running", + progress=processed / total if total else 0.5, + detail="Processing series...", + ))) + async for series, ep_reel_tasks in _process_series( + categories, _output_dir, + fetch_covers=True, generate_showreels=True, media_root=media_root_str, + ): + await _send(EvUpsert(kind="series", item=series)) + for task in ep_reel_tasks: + await _showreel_queue.put(("episode", task, series)) + processed += 1 + progress = round(processed / total, 3) if total else 1 + await _send(EvTask(data=TaskInfo( + id=task_id, status="running", progress=progress, detail=series.title, + ))) + + await _send(EvTask(data=TaskInfo( + id=task_id, status="completed", progress=1, detail="Scan complete", + ))) + if downloads: + logger.info("Scan complete (%s)", task_id) + + except asyncio.CancelledError: + await _send(EvTask(data=TaskInfo( + id=task_id, status="cancelled", progress=0, detail="Scan cancelled", + ))) + logger.info("Scan cancelled (%s)", task_id) + except Exception: + logger.exception("Scan failed (%s)", task_id) + await _send(EvTask(data=TaskInfo( + id=task_id, status="error", progress=0, detail="Scan error", + ))) + + +# --------------------------------------------------------------------------- +# Showreel worker +# --------------------------------------------------------------------------- + + +async def _showreel_worker(): + """Background worker that generates showreels one at a time.""" + logger.info("Showreel worker started") + media_root_path = Path(_media_root) if _media_root else None + media_root_str = str(_media_root) if _media_root else None + + while True: + try: + kind, task_data, item = await _showreel_queue.get() + task_id = f"showreel-{uuid.uuid4().hex[:8]}" + + if kind == "movie": + movie: Movie = item + video_path, media_folder, title = task_data + if await movie_showreels_exist(media_folder): + _showreel_queue.task_done() + continue + await _send(EvTask(data=TaskInfo( + id=task_id, status="running", progress=0, detail=f"Showreel: {title}", + ))) + await generate_showreel_images(video_path, media_folder, title=title) + paths = get_expected_showreel_paths(media_folder, media_root=media_root_path) + movie.showreel_images = paths if paths else None + await _send(EvUpsert(kind="movie", item=movie)) + await _send(EvTask(data=TaskInfo( + id=task_id, status="completed", progress=1, detail=f"Showreel: {title}", + ))) + + elif kind == "episode": + series: Series = item + video_path, media_folder, season_num, episode_num, series_title = task_data + if await episode_reel_exists(media_folder, season_num, episode_num): + _showreel_queue.task_done() + continue + ep_code = f"S{season_num:02d}E{episode_num:02d}" + await _send(EvTask(data=TaskInfo( + id=task_id, status="running", progress=0, detail=f"Reel: {series_title} {ep_code}", + ))) + reel_path = await generate_episode_reel( + video_path, media_folder, season_num, episode_num, + ) + if reel_path: + for season in series.seasons: + if season.season_number == season_num: + for episode in season.episodes: + if episode.episode_number == episode_num: + episode.reel_image = make_relative_path( + reel_path, media_root_str, + ) + await _send(EvUpsert(kind="series", item=series)) + await _send(EvTask(data=TaskInfo( + id=task_id, status="completed", progress=1, detail=f"Reel: {series_title} {ep_code}", + ))) + + _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 diff --git a/mediahive/hivescan/server.py b/mediahive/hivescan/server.py deleted file mode 100644 index c2d72e9..0000000 --- a/mediahive/hivescan/server.py +++ /dev/null @@ -1,497 +0,0 @@ -""" -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 aiopathlib import AsyncPath -from mediahive.hivescan.index_store import IndexStore -from mediahive.hivescan.indexer import _process_movies, _process_series -from mediahive.models.data import TaskInfo -from mediahive.models.protocol import MsgspecResponse, ScanRequest, StatusResponse -from mediahive.hivescan.models import ContentType, ParsedContent -from mediahive.hivescan.parsing import parse_download -from mediahive.hivescan.scanning import categorize_downloads -from mediahive.hivescan.showreel import ( - generate_episode_reel, - generate_showreel_images, - episode_reel_exists, - movie_showreels_exist, -) -from mediahive.hivescan.tmdb_client import set_cache_dir -from mediahive.hivescan.utils import DEFAULT_OUTPUT_FOLDER, find_common_root, make_relative_path - -logger = logging.getLogger("hivescan.server") - -# --------------------------------------------------------------------------- -# Configuration (from environment) -# --------------------------------------------------------------------------- - -SCAN_PATHS: List[str] = [] # set in lifespan from mediahive.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 -_seen_mtimes: dict[str, int] = {} - - -# --------------------------------------------------------------------------- -# 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 = await asyncio.to_thread(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 = await 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 - - await AsyncPath(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)) - await 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 _discover_downloads(paths_to_scan: List[str]) -> List[ParsedContent]: - """Walk the filesystem and parse all downloads, skipping unchanged torrents.""" - downloads: List[ParsedContent] = [] - media_root_str = str(MEDIA_ROOT) if MEDIA_ROOT else None - for pattern in paths_to_scan: - p = Path(pattern) - ap = AsyncPath(p) - if await ap.is_dir(): - for item in ap.iterdir(): - if not Path(item).name.startswith("."): - relpath = make_relative_path(str(item), media_root_str) - stat_info = await AsyncPath(item).stat() - mtime = int(stat_info.st_mtime) - if relpath in _seen_mtimes and _seen_mtimes[relpath] == mtime: - continue # skip unchanged - _seen_mtimes[relpath] = mtime - downloads.append(await parse_download(Path(item))) - elif await ap.exists(): - relpath = make_relative_path(str(p), media_root_str) - stat_info = await ap.stat() - mtime = int(stat_info.st_mtime) - if relpath in _seen_mtimes and _seen_mtimes[relpath] == mtime: - continue - _seen_mtimes[relpath] = mtime - downloads.append(await parse_download(p)) - return downloads - - -async def _run_scan(override_paths: Optional[List[str]] = None): - """ - Full scan pipeline: - 1. Walk filesystem, parse torrents (in thread) - 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]}" - - paths_to_scan = override_paths or SCAN_PATHS - media_root_str = str(MEDIA_ROOT) if MEDIA_ROOT else None - - try: - # 1. Discover downloads (now async) - downloads = await _discover_downloads(paths_to_scan) - - if downloads: - logger.info("Scan started (%s)", task_id) - - store.broadcast_task( - TaskInfo( - id=task_id, status="running", progress=0, detail="Scanning filesystem..." - ) - ) - - logger.info("Found %d items to process", len(downloads)) - categories = categorize_downloads(downloads) - total = len(categories[ContentType.MOVIE]) + len(categories[ContentType.SERIES]) - processed = 0 - changed = 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, - ): - was_changed = store.upsert_movie(movie) - if was_changed: - changed += 1 - logger.info(" Updated movie: %s", movie.title) - if showreel_task: - await _showreel_queue.put(("movie", showreel_task, movie.id)) - processed += 1 - progress = processed / total if total else 1 - if was_changed: - 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, - ): - was_changed = store.upsert_series(series) - if was_changed: - changed += 1 - logger.info(" Updated series: %s", series.title) - for task in ep_reel_tasks: - await _showreel_queue.put(("episode", task, series.id)) - processed += 1 - progress = processed / total if total else 1 - if was_changed: - 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") - ) - if downloads: - logger.info( - "Scan complete (%s): %d movies, %d series (%d changed)", - task_id, - len(store.movies), - len(store.series), - changed, - ) - - 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 await 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 mediahive.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 await episode_reel_exists(media_folder, season_num, episode_num): - _showreel_queue.task_done() - continue - ep_code = f"S{season_num:02d}E{episode_num:02d}" - 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/mediahive/hivescan/index_store.py b/mediahive/index_store.py similarity index 96% rename from mediahive/hivescan/index_store.py rename to mediahive/index_store.py index f64c917..b207bf0 100644 --- a/mediahive/hivescan/index_store.py +++ b/mediahive/index_store.py @@ -33,7 +33,7 @@ from mediahive.models.protocol import ( WsUpsert, ) -logger = logging.getLogger("hivescan.index_store") +logger = logging.getLogger("mediahive.index_store") # Debounce interval for writing snapshots to disk (seconds) SNAPSHOT_DEBOUNCE = 5.0 @@ -90,7 +90,7 @@ class IndexStore: ) 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_movie_versions = sum(len(m.torrents) for m in movies_list) total_series_episodes = sum( sum(len(season.episodes) for season in s.seasons) for s in series_list ) @@ -224,6 +224,10 @@ class IndexStore: """Broadcast a task progress message to all WS clients.""" self._broadcast(WsTask(data=task_info)) + def broadcast(self, msg: object) -> None: + """Broadcast an already-encoded message to all WS clients.""" + self._broadcast(msg) + # ------------------------------------------------------------------ # Read helpers # ------------------------------------------------------------------ @@ -235,7 +239,7 @@ class IndexStore: ) 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_movie_versions = sum(len(m.torrents) for m in movies_list) total_series_episodes = sum( sum(len(season.episodes) for season in s.seasons) for s in series_list ) diff --git a/mediahive/models/protocol.py b/mediahive/models/protocol.py index eb6e729..5a73ab5 100644 --- a/mediahive/models/protocol.py +++ b/mediahive/models/protocol.py @@ -54,6 +54,27 @@ class WsTask(msgspec.Struct, tag="task"): WsMessage = WsInit | WsUpsert | WsRemove | WsTask +# --------------------------------------------------------------------------- +# Scan events (scanner → server, via async queue) +# --------------------------------------------------------------------------- + + +class EvUpsert(msgspec.Struct, tag="upsert"): + """Scanner produced or updated a media item.""" + + kind: str # "movie" or "series" + item: Movie | Series + + +class EvTask(msgspec.Struct, tag="task"): + """Scanner progress update.""" + + data: TaskInfo + + +ScanEvent = EvUpsert | EvTask + + # --------------------------------------------------------------------------- # API request / response types # --------------------------------------------------------------------------- diff --git a/mediahive/server.py b/mediahive/server.py index fa27638..226c2ec 100644 --- a/mediahive/server.py +++ b/mediahive/server.py @@ -1,9 +1,12 @@ """ FastAPI server for MediaHive. -Replaces Tauri backend with async HTTP server. + +Serves media files, the Vue frontend, and — when ``HIVESCAN_PATHS`` is set — +also runs the continuous scanning pipeline with live WebSocket updates. """ -import json +import asyncio +import logging import mimetypes import os import subprocess @@ -13,32 +16,106 @@ from pathlib import Path import aiofiles import msgspec -from fastapi import FastAPI, HTTPException, Request +from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, StreamingResponse from fastapi_vue import Frontend -from mediahive.models.protocol import PlayMediaRequest, OpenFolderRequest +from mediahive.index_store import IndexStore +from mediahive.models.protocol import ( + EvTask, + EvUpsert, + MsgspecResponse, + PlayMediaRequest, + OpenFolderRequest, + ScanEvent, + ScanRequest, + StatusResponse, + WsTask, +) from mediahive.__main__ import DEVMODE +logger = logging.getLogger("mediahive.server") + # Vue Frontend static files frontend = Frontend(Path(__file__).with_name("frontend-build"), cached=["/assets/"]) - # Media root path (initialized in lifespan) MEDIAROOT = None +# In-memory index store (initialized in lifespan) +store: IndexStore | None = None + +# Whether the scanner subsystem is active +_scanner_active = False + +# Queue for scanner → server events +_scan_events: asyncio.Queue[ScanEvent] = asyncio.Queue() +_consumer_task: asyncio.Task | None = None + + +async def _send_event(event: ScanEvent) -> None: + """Push a scan event onto the queue (passed to hivescan as *send*).""" + await _scan_events.put(event) + + +async def _consume_scan_events() -> None: + """Background task: apply incoming scan events to the IndexStore.""" + while True: + try: + event = await _scan_events.get() + if isinstance(event, EvUpsert): + if event.kind == "movie": + store.upsert_movie(event.item) + else: + store.upsert_series(event.item) + elif isinstance(event, EvTask): + store.broadcast_task(event.data) + except asyncio.CancelledError: + return + except Exception: + logger.exception("Error processing scan event") + @asynccontextmanager async def lifespan(app: FastAPI): - global MEDIAROOT + global MEDIAROOT, store, _scanner_active, _consumer_task + if not os.environ.get("MEDIAHIVE_PATH"): raise RuntimeError("MEDIAHIVE_PATH environment variable must be set") + MEDIAROOT = Path(os.environ["MEDIAHIVE_PATH"]) await frontend.load() + + # Initialise the in-memory index store + snapshot_path = MEDIAROOT / ".mediahive" / "index.json" + store = IndexStore(snapshot_path, media_root=str(MEDIAROOT)) + await store.load_snapshot() + logger.info( + "Index store ready: %d movies, %d series", + len(store.movies), len(store.series), + ) + + # If scan paths are configured, start the scanner subsystem + if os.environ.get("HIVESCAN_PATHS"): + from mediahive.hivescan.scanner import start as start_scanner, stop as stop_scanner + + _consumer_task = asyncio.create_task(_consume_scan_events()) + await start_scanner(_send_event) + _scanner_active = True + yield + # Shutdown + if _scanner_active: + from mediahive.hivescan.scanner import stop as stop_scanner + + await stop_scanner() + if _consumer_task: + _consumer_task.cancel() + await store.flush_snapshot() + app = FastAPI(title="MediaHive Server", lifespan=lifespan, debug=DEVMODE) @@ -52,11 +129,6 @@ app.add_middleware( ) -def linux_to_windows_path(path: str) -> str: - """Normalize media path (now relative paths are kept as is).""" - return path - - def normalize_path(url_path: str) -> Path: """ Convert URL path to filesystem path. @@ -77,70 +149,66 @@ async def health_check(): @app.get("/api/index") -async def load_media_index(): - """ - Load and return the media index from disk. - Converts Linux paths to Windows paths. - """ - index_path = MEDIAROOT / ".mediahive" / "index.json" - if not index_path.exists(): - raise HTTPException( - status_code=404, detail=f"Index file not found: {index_path}" - ) +async def get_index(): + """Return the full media index from the in-memory store.""" + return MsgspecResponse(store.get_full_index()) + +# --------------------------------------------------------------------------- +# Scanning API (active when HIVESCAN_PATHS is configured) +# --------------------------------------------------------------------------- + + +@app.websocket("/ws") +async def ws_endpoint(ws: WebSocket): + """Live index updates and task progress.""" + await store.connect(ws) try: - async with aiofiles.open(index_path, "r", encoding="utf-8") as f: - content = await f.read() + while True: + await ws.receive_text() + except WebSocketDisconnect: + store.disconnect(ws) + except Exception: + store.disconnect(ws) - index = json.loads(content) - # Convert all Linux paths to Windows paths - for movie in index.get("movies", []): - if movie.get("cover_path"): - movie["cover_path"] = linux_to_windows_path(movie["cover_path"]) - if movie.get("backdrop_path"): - movie["backdrop_path"] = linux_to_windows_path(movie["backdrop_path"]) - if movie.get("showreel_images"): - movie["showreel_images"] = [ - linux_to_windows_path(p) for p in movie["showreel_images"] - ] - for version in movie.get("versions", []): - version["path"] = linux_to_windows_path(version["path"]) - if version.get("playable_file"): - version["playable_file"] = linux_to_windows_path( - version["playable_file"] - ) - if version.get("torrent_path"): - version["torrent_path"] = linux_to_windows_path( - version["torrent_path"] - ) +@app.post("/api/scan") +async def trigger_scan(request: Request): + """Trigger a new scan. Returns 409 if a scan is already running.""" + if not _scanner_active: + raise HTTPException(status_code=503, detail="Scanner not configured") + from mediahive.hivescan.scanner import trigger_scan as _trigger - for series in index.get("series", []): - if series.get("cover_path"): - series["cover_path"] = linux_to_windows_path(series["cover_path"]) - if series.get("backdrop_path"): - series["backdrop_path"] = linux_to_windows_path(series["backdrop_path"]) - for season in series.get("seasons", []): - if season.get("poster_path"): - season["poster_path"] = linux_to_windows_path(season["poster_path"]) - for episode in season.get("episodes", []): - if episode.get("reel_image"): - episode["reel_image"] = linux_to_windows_path( - episode["reel_image"] - ) - for release in episode.get("releases", []): - release["path"] = linux_to_windows_path(release["path"]) - if release.get("playable_file"): - release["playable_file"] = linux_to_windows_path( - release["playable_file"] - ) + body_bytes = await request.body() + req = ( + msgspec.json.decode(body_bytes, type=ScanRequest) + if body_bytes + else ScanRequest() + ) + started = _trigger(req.paths if req.paths else None) + return {"status": "started" if started else "already_running"} - return index - except json.JSONDecodeError as e: - raise HTTPException(status_code=500, detail=f"Failed to parse index file: {e}") - except Exception as e: - raise HTTPException(status_code=500, detail=f"Failed to read index file: {e}") +@app.get("/api/status") +async def server_status(): + """Return current server status.""" + if _scanner_active: + from mediahive.hivescan.scanner import is_scanning, showreel_queue_size + + return MsgspecResponse( + StatusResponse( + scanning=is_scanning(), + movies=len(store.movies), + series=len(store.series), + showreel_queue=showreel_queue_size(), + ) + ) + return MsgspecResponse( + StatusResponse( + movies=len(store.movies), + series=len(store.series), + ) + ) @app.post("/api/play")