Unsafe fixes.

mediahive/__init__.py

@@ -1 +1 @@
-"""MediaHive - Media Browser Server"""
+"""MediaHive - Media Browser Server."""

mediahive/__main__.py

@@ -31,7 +31,7 @@ def _derive_name(path: str) -> str:
     return p.name or p.anchor.strip("/\\").lower() or "media"
 
 
-def main():
+def main() -> None:
     _configure_windows_event_loop_policy()
 
     parser = argparse.ArgumentParser(

mediahive/config.py

@@ -1,7 +1,7 @@
-"""Platform-appropriate config persistence for MediaHive.
+r"""Platform-appropriate config persistence for MediaHive.
 
 Config file location:
-  Windows:  %APPDATA%\\mediahive\\config.toml
+  Windows:  %APPDATA%\mediahive\config.toml
   macOS:    ~/Library/Application Support/mediahive/config.toml
   Linux:    $XDG_CONFIG_HOME/mediahive/config.toml  (~/.config/mediahive/config.toml)
 """

mediahive/hivescan/__main__.py

@@ -17,7 +17,7 @@ def _configure_windows_event_loop_policy() -> None:
     asyncio.set_event_loop_policy(policy_cls())
 
 
-def main():
+def main() -> None:
     _configure_windows_event_loop_policy()
 
     parser = argparse.ArgumentParser(

mediahive/hivescan/indexer.py

@@ -400,10 +400,7 @@ async def _process_movies(
         return await find_playable_file(item.path) is not None
 
     # Filter movies with playable files
-    valid_movies = []
-    for item in categories[ContentType.MOVIE]:
-        if await has_playable(item):
-            valid_movies.append(item)
+    valid_movies = [item for item in categories[ContentType.MOVIE] if await has_playable(item)]
     skipped = len(categories[ContentType.MOVIE]) - len(valid_movies)
     if skipped > 0:
         logger.debug(
@@ -429,7 +426,7 @@ async def _process_movies(
             len(categories[ContentType.MOVIE]),
         ) if movie_groups else None
 
-    for idx, (movie_key, items) in enumerate(movie_groups.items(), 1):
+    for idx, (_movie_key, items) in enumerate(movie_groups.items(), 1):
         first_item = items[0]
         logger.debug(
             "    [%d/%d] %s (%s)",
@@ -658,10 +655,7 @@ async def _process_series(
         return len(await find_episode_files(item.path)) > 0
 
     # Filter series with video content
-    valid_series = []
-    for item in categories[ContentType.SERIES]:
-        if await has_video_content(item):
-            valid_series.append(item)
+    valid_series = [item for item in categories[ContentType.SERIES] if await has_video_content(item)]
     skipped = len(categories[ContentType.SERIES]) - len(valid_series)
     if skipped > 0:
         logger.info(
@@ -687,7 +681,7 @@ async def _process_series(
             len(categories[ContentType.SERIES]),
         )
 
-    for idx, (series_key, items) in enumerate(series_groups.items(), 1):
+    for idx, (_series_key, items) in enumerate(series_groups.items(), 1):
         first_item = items[0]
         logger.debug("    [%d/%d] %s", idx, len(series_groups), first_item.title)
 

mediahive/hivescan/scanner.py

@@ -12,6 +12,7 @@ defined in ``.mediahive/scanignore`` (gitignore-style syntax).
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import logging
 import uuid
 from collections.abc import Awaitable, Callable
@@ -59,7 +60,7 @@ class RootScanner:
         root_id: str,
         media_root: Path,
         send: Send,
-    ):
+    ) -> None:
         self.root_id = root_id
         self.media_root = media_root
         self._send = send
@@ -108,10 +109,8 @@ class RootScanner:
             self._rescan_worker_task,
         ):
             if task and not task.done():
-                try:
+                with contextlib.suppress(TimeoutError, asyncio.CancelledError):
                     await asyncio.wait_for(task, timeout=2.0)
-                except TimeoutError, asyncio.CancelledError:
-                    pass
 
     def is_scanning(self) -> bool:
         return self._scan_task is not None and not self._scan_task.done()
@@ -266,7 +265,7 @@ class RootScanner:
         try:
             root_children = await asyncio.to_thread(lambda: list(root_ap.iterdir()))
         except OSError, PermissionError:
-            logger.error("Cannot list media root: %s", self.media_root)
+            logger.exception("Cannot list media root: %s", self.media_root)
             return downloads
 
         for item_async in root_children:
@@ -302,7 +301,7 @@ class RootScanner:
         1. Discover downloads
         2. Categorise → movies / series
         3. Iterate async generators, send each item as Upsert
-        4. Queue showreel tasks
+        4. Queue showreel tasks.
         """
         task_id = f"scan-{uuid.uuid4().hex[:8]}"
         media_root_str = self.media_root.as_posix()
@@ -674,10 +673,8 @@ class RootScanner:
                     "Showreel worker error (queue size=%d)",
                     self._showreel_queue.qsize(),
                 )
-                try:
+                with contextlib.suppress(ValueError):
                     self._showreel_queue.task_done()
-                except ValueError:
-                    pass
 
 
 # ---------------------------------------------------------------------------

mediahive/hivescan/scanning.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import glob
+import operator
 from collections import defaultdict
 from pathlib import Path
 
@@ -216,7 +217,7 @@ async def find_playable_file(path: Path) -> str | None:
         _playable_file_cache[cache_key] = None
         return None
 
-    video_files.sort(key=lambda x: x[1], reverse=True)
+    video_files.sort(key=operator.itemgetter(1), reverse=True)
     result = video_files[0][0]
     _playable_file_cache[cache_key] = result
     return result
@@ -312,7 +313,7 @@ async def find_metadata_probe_file(playable_path: str | None) -> str | None:
         _bluray_probe_file_cache[cache_key] = None
         return None
 
-    candidates.sort(key=lambda x: x[1], reverse=True)
+    candidates.sort(key=operator.itemgetter(1), reverse=True)
     result = candidates[0][0]
     _bluray_probe_file_cache[cache_key] = result
     return result

mediahive/hivescan/showreel.py

@@ -6,6 +6,7 @@ and HDR passthrough.
 """
 
 import asyncio
+import contextlib
 import logging
 import re
 import shlex
@@ -75,7 +76,7 @@ async def _run_ffmpeg(
                 stdout, stderr = await proc.communicate()
             except Exception:
                 stdout, stderr = b"", b""
-            logger.error(
+            logger.exception(
                 "ffmpeg command timed out. cmd=%s stderr=%s",
                 shlex.join(cmd),
                 _decode_stderr(stderr),
@@ -113,10 +114,8 @@ 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:
         proc.kill()
-        try:
+        with contextlib.suppress(Exception):
             await asyncio.wait_for(proc.wait(), timeout=2)
-        except Exception:
-            pass
 
 
 # Showreel timestamp positions in seconds (5, 10, 15, 20, 25 minutes)
@@ -763,7 +762,7 @@ async def generate_showreel_images(
 
     # Detect Dolby Vision profile for tonemapping (profiles 5/7 need conversion)
     dovi_profile = await detect_dovi_profile(ffmpeg_input)
-    needs_tonemap = dovi_profile is not None and dovi_profile in (5, 7)
+    needs_tonemap = dovi_profile is not None and dovi_profile in {5, 7}
     if needs_tonemap:
         logger.info("    DoVi profile %d detected, will convert to HDR10", dovi_profile)
 
@@ -924,7 +923,7 @@ async def generate_episode_reel(
 
     # Detect Dolby Vision profile for tonemapping (profiles 5/7 need conversion)
     dovi_profile = await detect_dovi_profile(ffmpeg_input)
-    needs_tonemap = dovi_profile is not None and dovi_profile in (5, 7)
+    needs_tonemap = dovi_profile is not None and dovi_profile in {5, 7}
     if needs_tonemap:
         logger.info("    DoVi profile %d detected, will convert to HDR10", dovi_profile)
 

mediahive/hivescan/tmdb_client.py

@@ -85,14 +85,11 @@ async def _load_from_cache(cache_path: Path):
         return _NOT_FOUND
 
 
-async def _save_to_cache(cache_path: Path, data: dict | None):
+async def _save_to_cache(cache_path: Path, data: dict | None) -> None:
     """Save response to cache."""
     try:
         await AsyncPath(_get_cache_dir()).mkdir(parents=True, exist_ok=True)
-        if data is None:
-            text = json.dumps({"_cached_none": True})
-        else:
-            text = json.dumps(data)
+        text = json.dumps({"_cached_none": True}) if data is None else json.dumps(data)
         await AsyncPath(cache_path).write_text(text, encoding="utf-8")
     except Exception:
         pass  # Cache write failures are not critical
@@ -144,20 +141,18 @@ async def tmdb_api_request(
 async def fetch_movie_details(movie_id: int) -> dict | None:
     """Fetch detailed movie info including credits, similar, keywords, and alternative titles."""
     # Use append_to_response to get multiple data in one request
-    data = await tmdb_api_request(
+    return await tmdb_api_request(
         f"/movie/{movie_id}",
         {"append_to_response": "credits,similar,keywords,alternative_titles"},
     )
-    return data
 
 
 async def fetch_series_details(series_id: int) -> dict | None:
     """Fetch detailed TV series info including credits, similar, and keywords."""
     # Use append_to_response to get multiple data in one request
-    data = await tmdb_api_request(
+    return await tmdb_api_request(
         f"/tv/{series_id}", {"append_to_response": "credits,similar,keywords"}
     )
-    return data
 
 
 async def fetch_season_details(series_id: int, season_number: int) -> SeasonInfo | None:
@@ -238,12 +233,10 @@ def _generate_title_variants(words: list[str], min_words: int = 2) -> list[str]:
     variants.append(" ".join(words))
 
     # Then try removing from end (most common: edition names at end)
-    for num_words in range(len(words) - 1, min_words - 1, -1):
-        variants.append(" ".join(words[:num_words]))
+    variants.extend(" ".join(words[:num_words]) for num_words in range(len(words) - 1, min_words - 1, -1))
 
     # Then try removing from start (garbage at beginning)
-    for start in range(1, len(words) - min_words + 1):
-        variants.append(" ".join(words[start:]))
+    variants.extend(" ".join(words[start:]) for start in range(1, len(words) - min_words + 1))
 
     # Finally try middle portions (remove from both ends)
     for start in range(1, len(words) - min_words):

mediahive/hivescan/utils.py

@@ -184,7 +184,7 @@ async def find_common_root(paths: list[Path]) -> Path | None:
 
     # Find common prefix
     common_parts = []
-    for parts in zip(*all_parts):
+    for parts in zip(*all_parts, strict=False):
         if len(set(parts)) == 1:
             common_parts.append(parts[0])
         else:
@@ -216,8 +216,7 @@ def sanitize_filename(name: str) -> str:
     """Sanitize a string for use as a filename."""
     for char in ["/", "\\", ":", "*", "?", '"', "<", ">", "|"]:
         name = name.replace(char, "_")
-    name = name.strip(". ")
-    return name
+    return name.strip(". ")
 
 
 def get_media_folder_name(title: str, year: int | None, media_type: str) -> str:

mediahive/index_store.py

@@ -7,6 +7,7 @@ debounced background task.
 """
 
 import asyncio
+import contextlib
 import logging
 import os
 from datetime import datetime
@@ -51,7 +52,7 @@ class IndexStore:
         snapshot_path: Path,
         media_root: str | None = None,
         root_id: str | None = None,
-    ):
+    ) -> None:
         self.snapshot_path = snapshot_path
         self.media_root = media_root
         self.root_id = root_id
@@ -205,10 +206,8 @@ class IndexStore:
         """Force-write a snapshot immediately (e.g. on shutdown)."""
         if self._snapshot_task and not self._snapshot_task.done():
             self._snapshot_task.cancel()
-            try:
+            with contextlib.suppress(asyncio.CancelledError):
                 await self._snapshot_task
-            except asyncio.CancelledError:
-                pass
         await self._write_snapshot()
 
     # ------------------------------------------------------------------

mediahive/root_registry.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import hashlib
 import logging
 from pathlib import Path
@@ -86,7 +87,7 @@ class RootEntry(msgspec.Struct):
 class RootContext:
     """Runtime container for a single media root."""
 
-    def __init__(self, root_id: str, root_path: Path, name: str | None = None):
+    def __init__(self, root_id: str, root_path: Path, name: str | None = None) -> None:
         self.root_id = root_id
         self.name = name or root_id
         self.root_path = root_path
@@ -134,10 +135,8 @@ class RootContext:
 
         if self._consumer_task and not self._consumer_task.done():
             self._consumer_task.cancel()
-            try:
+            with contextlib.suppress(asyncio.CancelledError):
                 await self._consumer_task
-            except asyncio.CancelledError:
-                pass
 
         try:
             await self.store.flush_snapshot()
@@ -175,7 +174,7 @@ class RootContext:
 class Supervisor:
     """Manages the active set of RootContexts and handles atomic replacement."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         # Active contexts keyed by root_id
         self._contexts: dict[str, RootContext] = {}
         self._lock = asyncio.Lock()
@@ -211,7 +210,7 @@ class Supervisor:
         total_movie_versions = 0
         total_series_episodes = 0
         for ctx in self._contexts.values():
-            if ctx.status != "ready" and ctx.status != "scanning":
+            if ctx.status not in {"ready", "scanning"}:
                 continue
             movies.extend(ctx.store.movies.values())
             series.extend(ctx.store.series.values())

mediahive/server.py

@@ -18,7 +18,7 @@ import subprocess
 import sys
 import urllib.error
 import urllib.request
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, suppress
 from pathlib import Path
 
 import aiofiles
@@ -248,10 +248,7 @@ def _etag_matches_if_none_match(if_none_match: str | None, etag: str) -> bool:
         return v
 
     wanted = normalize(etag)
-    for candidate in if_none_match.split(","):
-        if normalize(candidate) == wanted:
-            return True
-    return False
+    return any(normalize(candidate) == wanted for candidate in if_none_match.split(","))
 
 
 def _validate_root_paths(roots: dict[str, str]) -> dict[str, str]:
@@ -283,7 +280,7 @@ def _validate_root_paths(roots: dict[str, str]) -> dict[str, str]:
 async def _attach_scanners() -> None:
     """Ensure every active root context has a running scanner."""
     for ctx in supervisor.all_contexts().values():
-        if ctx.scanner is None and ctx.status in ("ready", "loading"):
+        if ctx.scanner is None and ctx.status in {"ready", "loading"}:
             try:
                 scanner = RootScanner(ctx.root_id, ctx.root_path, ctx.send_event)
                 await scanner.start()
@@ -361,10 +358,8 @@ async def lifespan(app: FastAPI):
     yield
 
     activation_task.cancel()
-    try:
+    with suppress(asyncio.CancelledError):
         await activation_task
-    except asyncio.CancelledError:
-        pass
 
     await supervisor.shutdown()
 
@@ -464,7 +459,7 @@ async def trigger_root_scan(root_id: str):
 
 
 @app.websocket("/api/roots/{root_id}/ws")
-async def ws_endpoint(ws: WebSocket, root_id: str):
+async def ws_endpoint(ws: WebSocket, root_id: str) -> None:
     """Live index updates and task progress for a single root."""
     ctx = supervisor.get(root_id)
     if ctx is None:

rtorrent_client.py

@@ -9,7 +9,7 @@ from pathlib import Path
 class SCGITransport(xmlrpc.client.Transport):
     """SCGI transport for communicating with rtorrent via Unix socket."""
 
-    def __init__(self, socket_path: str):
+    def __init__(self, socket_path: str) -> None:
         super().__init__()
         self.socket_path = socket_path
 
@@ -48,7 +48,7 @@ class SCGITransport(xmlrpc.client.Transport):
 class RTorrentClient:
     """Client for communicating with rtorrent via XMLRPC over SCGI socket."""
 
-    def __init__(self, socket_path: str = "/home/user/rtorrent/.session/rpc.socket"):
+    def __init__(self, socket_path: str = "/home/user/rtorrent/.session/rpc.socket") -> None:
         self.socket_path = socket_path
         transport = SCGITransport(socket_path)
         self.proxy = xmlrpc.client.ServerProxy(
@@ -59,7 +59,7 @@ class RTorrentClient:
         """Get set of info hashes for all currently loaded torrents."""
         try:
             downloads = self.proxy.download_list("")
-            return set(h.upper() for h in downloads)
+            return {h.upper() for h in downloads}
         except Exception as e:
             print(f"Error getting loaded torrents: {e}")
             return set()

scripts/devserver.py

@@ -49,7 +49,7 @@ async def run_devserver(
         await pg.spawn(*vite, cwd=front)
 
 
-def main():
+def main() -> None:
     parser = argparse.ArgumentParser(
         description="Run Vite and FastAPI development servers",
         formatter_class=argparse.RawDescriptionHelpFormatter,

scripts/fastapi-vue/build-frontend.py

@@ -10,6 +10,6 @@ from buildutil import build
 
 
 class CustomBuildHook(BuildHookInterface):
-    def initialize(self, version, build_data):
+    def initialize(self, version, build_data) -> None:
         super().initialize(version, build_data)
         build("frontend")

scripts/fastapi-vue/buildutil.py

@@ -178,7 +178,7 @@ def build(folder: str = "frontend") -> None:
         logger.warning(e)
         raise SystemExit(1)
 
-    def run(cmd):
+    def run(cmd) -> None:
         display_cmd = [Path(cmd[0]).stem, *cmd[1:]]
         logger.info("### %s", " ".join(display_cmd))
         subprocess.run(cmd, check=True, cwd=folder)

scripts/fastapi-vue/devutil.py

@@ -16,7 +16,7 @@ from fastapi_vue.hostutil import parse_endpoint
 class ProcessGroup:
     """Manage async subprocesses with automatic cleanup, like TaskGroup for processes."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         self._procs: list[asyncio.subprocess.Process] = []
         self._cmds: dict[int, str] = {}  # pid -> command name
 
@@ -59,7 +59,7 @@ class ProcessGroup:
         """Wait for one process to exit, terminate others, then wait for all."""
         await self._cleanup(immediate=exc_type is not None)
 
-    async def _cleanup(self, immediate: bool = False):
+    async def _cleanup(self, immediate: bool = False) -> None:
         running = [p for p in self._procs if p.returncode is None]
         if not running:
             return

scripts/release.py

@@ -84,13 +84,12 @@ def find_dist_files(version: str) -> list[Path]:
     Raises FileNotFoundError listing every missing file if any are absent.
     """
     dist_dir = REPO_ROOT / "dist"
-    ver = re.escape(version)
     wheel = next((p for p in dist_dir.glob(f"mediahive-{version}-*.whl")), None)
     sdist = next(
         (
             p
             for p in dist_dir.glob(f"mediahive-{version}.*")
-            if p.suffix in (".gz", ".zip") and p.name != f"mediahive-{version}.zip"
+            if p.suffix in {".gz", ".zip"} and p.name != f"mediahive-{version}.zip"
         ),
         None,
     )

scripts/rtorrent-manager.py

@@ -164,8 +164,7 @@ def scan_torrent_directories(paths: list[str]) -> Iterator[Path]:
         for dir_path in glob.glob(pattern):
             torrent_dir = Path(dir_path)
             if torrent_dir.is_dir():
-                for torrent_file in torrent_dir.glob("*.torrent"):
-                    yield torrent_file
+                yield from torrent_dir.glob("*.torrent")
 
 
 def find_torrents_with_tracker(
@@ -203,7 +202,7 @@ def format_size(size_bytes: int | None) -> str:
     return f"{size_bytes:.2f} PB"
 
 
-def main():
+def main() -> None:
     """Main entry point for the torrent scanner."""
     parser = argparse.ArgumentParser(
         description="Scan and manage torrent files",