Harden ffmpeg subprocess handling and Windows loop policy

mediahive/__main__.py

@@ -1,4 +1,5 @@
 import argparse
+import asyncio
 import json
 import os
 import sys
@@ -10,6 +11,16 @@ DEFAULT_PORT = 8420
 DEVMODE = os.getenv("MEDIAHIVE_DEV") == "1"
 
 
+def _configure_windows_event_loop_policy() -> None:
+    """Ensure Windows uses Proactor loop so asyncio subprocess APIs are available."""
+    if sys.platform != "win32":
+        return
+    policy_cls = getattr(asyncio, "WindowsProactorEventLoopPolicy", None)
+    if policy_cls is None:
+        return
+    asyncio.set_event_loop_policy(policy_cls())
+
+
 def _derive_name(path: str) -> str:
     """Derive a root name from a path."""
     p = Path(path)
@@ -17,6 +28,8 @@ def _derive_name(path: str) -> str:
 
 
 def main():
+    _configure_windows_event_loop_policy()
+
     parser = argparse.ArgumentParser(
         description="MediaHive - Media scanning, indexing, and streaming"
     )

mediahive/hivescan/__main__.py

@@ -1,11 +1,25 @@
 import argparse
+import asyncio
 import json
 import logging
 import os
+import sys
 from pathlib import Path
 
 
+def _configure_windows_event_loop_policy() -> None:
+    """Ensure Windows uses Proactor loop so asyncio subprocess APIs are available."""
+    if sys.platform != "win32":
+        return
+    policy_cls = getattr(asyncio, "WindowsProactorEventLoopPolicy", None)
+    if policy_cls is None:
+        return
+    asyncio.set_event_loop_policy(policy_cls())
+
+
 def main():
+    _configure_windows_event_loop_policy()
+
     parser = argparse.ArgumentParser(
         description="Hivescan server — continuous media scanning with live WS updates.",
         formatter_class=argparse.RawDescriptionHelpFormatter,

mediahive/hivescan/showreel.py

@@ -22,6 +22,7 @@ from aiopathlib import AsyncPath
 from mediahive.hivescan.utils import classify_resolution_from_dimensions
 
 logger = logging.getLogger("hivescan.showreel")
+_ffmpeg_not_found_logged = False
 
 
 # Suppress console windows when spawning subprocesses on Windows
@@ -32,6 +33,69 @@ async def _subprocess_exec(*args, **kwargs):
     return await asyncio.create_subprocess_exec(*args, **kwargs)
 
 
+def _decode_stderr(stderr: bytes | None) -> str:
+    if not stderr:
+        return "(no stderr output)"
+    text = stderr.decode("utf-8", errors="replace").strip()
+    return text[:1500] if text else "(no stderr output)"
+
+
+def _log_ffmpeg_not_found_once(cmd: list[str]) -> None:
+    global _ffmpeg_not_found_logged
+    if _ffmpeg_not_found_logged:
+        return
+    _ffmpeg_not_found_logged = True
+    logger.error(
+        "ffmpeg executable was not found on PATH. Install ffmpeg and restart MediaHive. Command: %s",
+        shlex.join(cmd),
+    )
+
+
+async def _run_ffmpeg(cmd: list[str], timeout: float) -> tuple[bytes, bytes] | None:
+    """Run ffmpeg with consistent timeout/crash/not-found handling and logging."""
+    proc: asyncio.subprocess.Process | None = None
+    try:
+        logger.debug("    $ %s", shlex.join(cmd))
+        proc = await _subprocess_exec(
+            *cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        try:
+            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+        except asyncio.TimeoutError:
+            await _kill_proc(proc)
+            try:
+                stdout, stderr = await proc.communicate()
+            except Exception:
+                stdout, stderr = b"", b""
+            logger.error(
+                "ffmpeg command timed out. cmd=%s stderr=%s",
+                shlex.join(cmd),
+                _decode_stderr(stderr),
+            )
+            return None
+
+        if proc.returncode != 0:
+            logger.error(
+                "ffmpeg command failed. cmd=%s stderr=%s",
+                shlex.join(cmd),
+                _decode_stderr(stderr),
+            )
+            return None
+
+        return stdout, stderr
+    except FileNotFoundError:
+        _log_ffmpeg_not_found_once(cmd)
+        return None
+    except asyncio.CancelledError:
+        await _kill_proc(proc)
+        raise
+    except Exception:
+        logger.exception("Unexpected error running ffmpeg command: %s", shlex.join(cmd))
+        return None
+
+
 async def _kill_proc(proc: asyncio.subprocess.Process | None) -> None:
     """Kill a subprocess immediately if it is still running."""
     if proc is not None and proc.returncode is None:
@@ -282,18 +346,11 @@ async def get_av1_encoder() -> str:
         return _av1_encoder_cache
 
     # Check for NVIDIA AV1 encoder
-    try:
-        proc = await _subprocess_exec(
-            "ffmpeg",
-            "-hide_banner",
-            "-encoders",
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-        stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=10)
-        if b"av1_nvenc" in stdout:
+    encoders = await _run_ffmpeg(["ffmpeg", "-hide_banner", "-encoders"], timeout=10)
+    if encoders and b"av1_nvenc" in encoders[0]:
         # Verify it actually works (driver support)
-            test_proc = await _subprocess_exec(
+        test_run = await _run_ffmpeg(
+            [
                 "ffmpeg",
                 "-f",
                 "lavfi",
@@ -304,15 +361,12 @@ async def get_av1_encoder() -> str:
                 "-f",
                 "null",
                 "-",
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
+            ],
+            timeout=10,
         )
-            await asyncio.wait_for(test_proc.communicate(), timeout=10)
-            if test_proc.returncode == 0:
+        if test_run is not None:
             _av1_encoder_cache = "av1_nvenc"
             return _av1_encoder_cache
-    except Exception:
-        pass
 
     # Default to libsvtav1
     _av1_encoder_cache = "libsvtav1"
@@ -373,15 +427,13 @@ async def probe_media_info(video_path: str) -> MediaProbeInfo:
         return cached
 
     info = MediaProbeInfo()
-    try:
     cmd = ["ffmpeg", "-hide_banner", "-i", video_path]
-        logger.debug("    $ %s", shlex.join(cmd))
-        proc = await _subprocess_exec(
-            *cmd,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30)
+    ffmpeg_result = await _run_ffmpeg(cmd, timeout=30)
+    if ffmpeg_result is None:
+        _media_probe_cache[video_path] = info
+        return info
+
+    stdout, stderr = ffmpeg_result
     text = (stderr + stdout).decode("utf-8", errors="replace")
     lower_text = text.lower()
 
@@ -439,8 +491,6 @@ async def probe_media_info(video_path: str) -> MediaProbeInfo:
         if lang and lang not in subtitle_languages:
             subtitle_languages.append(lang)
     info.subtitle_languages = subtitle_languages or None
-    except Exception as e:
-        logger.warning("    ffmpeg probe error: %s", e)
 
     _media_probe_cache[video_path] = info
     return info
@@ -507,7 +557,6 @@ async def detect_crop(video_path: str) -> Optional[str]:
         Crop filter string like "crop=1920:800:0:140" if black bars detected,
         or None if no cropping needed or detection failed.
     """
-    try:
     # First, get source video dimensions to check if it's 16:9
     probe_info = await probe_media_info(video_path)
     if not probe_info.width or not probe_info.height:
@@ -538,13 +587,10 @@ async def detect_crop(video_path: str) -> Optional[str]:
         "null",
         "-",
     ]
-        logger.debug("    $ %s", shlex.join(cmd))
-        proc = await _subprocess_exec(
-            *cmd,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-        _, stderr_bytes = await asyncio.wait_for(proc.communicate(), timeout=60)
+    ffmpeg_result = await _run_ffmpeg(cmd, timeout=60)
+    if ffmpeg_result is None:
+        return None
+    _, stderr_bytes = ffmpeg_result
     stderr_text = stderr_bytes.decode("utf-8", errors="replace")
 
     # cropdetect outputs to stderr like: [Parsed_cropdetect_0 @ ...] x1:0 x2:1919 y1:138 y2:941 w:1920 h:800 ...
@@ -614,10 +660,6 @@ async def detect_crop(video_path: str) -> Optional[str]:
     logger.debug("    Detected crop: %s", crop_result)
     return crop_result
 
-    except Exception as e:
-        logger.warning("    Crop detection error: %s", e)
-        return None
-
 
 async def get_video_duration(video_path: str) -> Optional[float]:
     """
@@ -633,7 +675,7 @@ async def generate_showreel_images(
     video_path: str,
     media_folder: Path,
     timestamps: list[int] = SHOWREEL_TIMESTAMPS,
-    title: str = None,
+    title: str | None = None,
     on_progress=None,
 ) -> list[str]:
     """
@@ -770,17 +812,12 @@ async def generate_showreel_images(
             output_path.as_posix(),
         ]
 
-        logger.debug("    $ %s", shlex.join(cmd))
-        proc = None
-        try:
-            proc = await _subprocess_exec(
-                *cmd,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-            )
-            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=120)
+        ffmpeg_result = await _run_ffmpeg(cmd, timeout=120)
+        if ffmpeg_result is None:
+            await AsyncPath(output_path).unlink(missing_ok=True)
+            break
 
-            if proc.returncode == 0 and await AsyncPath(output_path).exists():
+        if await AsyncPath(output_path).exists():
             generated_paths.append(output_path.as_posix())
             logger.info(
                 "    Showreel reel%d generated for %s",
@@ -790,29 +827,13 @@ async def generate_showreel_images(
             if on_progress:
                 on_progress(reel_num)
         else:
-                stderr_text = stderr.decode(errors="replace").strip() if stderr else ""
             logger.error(
-                    "    Showreel reel%d failed (rc=%s) for %s: %s",
+                "    Showreel reel%d failed for %s: output file was not created. cmd=%s",
                 reel_num,
-                    proc.returncode,
                 title or "unknown",
-                    stderr_text[:500] if stderr_text else "(no output)",
+                shlex.join(cmd),
             )
             await AsyncPath(output_path).unlink(missing_ok=True)
-                # Abort remaining reels - if first one fails, others likely will too
-                break
-        except BaseException as e:
-            await _kill_proc(proc)
-            await AsyncPath(output_path).unlink(missing_ok=True)
-            if isinstance(e, (KeyboardInterrupt, SystemExit, asyncio.CancelledError)):
-                raise
-            logger.error(
-                "Error generating showreel for %s at %ds: %s",
-                title or "unknown",
-                timestamp,
-                e,
-            )
-            # Abort remaining reels
             break
 
     if generated_paths:
@@ -939,33 +960,19 @@ async def generate_episode_reel(
         output_path.as_posix(),
     ]
 
-    logger.debug("    $ %s", shlex.join(cmd))
-    proc = None
-    try:
-        proc = await _subprocess_exec(
-            *cmd,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=120)
+    ffmpeg_result = await _run_ffmpeg(cmd, timeout=120)
+    if ffmpeg_result is None:
+        await AsyncPath(output_path).unlink(missing_ok=True)
+        return None
 
-        if proc.returncode == 0 and await AsyncPath(output_path).exists():
+    if await AsyncPath(output_path).exists():
         logger.info("    Episode reel generated: %s", ep_code)
         return output_path.as_posix()
-        else:
-            stderr_text = stderr.decode(errors="replace").strip() if stderr else ""
+
     logger.error(
-                "    Episode reel %s failed (rc=%s): %s",
+        "    Episode reel %s failed: output file was not created. cmd=%s",
         ep_code,
-                proc.returncode,
-                stderr_text[:500] if stderr_text else "(no output)",
+        shlex.join(cmd),
     )
     await AsyncPath(output_path).unlink(missing_ok=True)
     return None
-    except BaseException as e:
-        await _kill_proc(proc)
-        await AsyncPath(output_path).unlink(missing_ok=True)
-        if isinstance(e, (KeyboardInterrupt, SystemExit, asyncio.CancelledError)):
-            raise
-        logger.error("Error generating episode reel for %s: %s", ep_code, e)
-        return None

mediahive/winmain.py

@@ -5,6 +5,7 @@ Or from PyInstaller: MediaHive.exe [media_folder]
 """
 
 import argparse
+import asyncio
 import ctypes
 import html
 import json
@@ -752,7 +753,19 @@ def _reserve_backend_port() -> int:
         return int(sock.getsockname()[1])
 
 
+def _configure_windows_event_loop_policy() -> None:
+    """Ensure Windows uses Proactor loop so asyncio subprocess APIs are available."""
+    if sys.platform != "win32":
+        return
+    policy_cls = getattr(asyncio, "WindowsProactorEventLoopPolicy", None)
+    if policy_cls is None:
+        return
+    asyncio.set_event_loop_policy(policy_cls())
+
+
 def winmain() -> None:
+    _configure_windows_event_loop_policy()
+
     parser = argparse.ArgumentParser(description="MediaHive")
     parser.add_argument(
         "media_folder",