Merge cont'd.
This commit is contained in:
+24
-77
@@ -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__":
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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")
|
||||
@@ -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
|
||||
)
|
||||
@@ -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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
+136
-68
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user