BLE001 except Exception replaced for more specific types.

mediahive/config.py

@@ -52,7 +52,7 @@ def load_config() -> Config:
         try:
             cfg = msgspec.toml.decode(path.read_bytes(), type=Config)
             return _migrate_legacy_media_folder(cfg)
-        except Exception:
+        except OSError, msgspec.DecodeError, msgspec.ValidationError:
             return Config()
     return Config()
 

mediahive/hivescan/images.py

@@ -42,7 +42,7 @@ async def _download_image(url: str, output_path: Path, description: str) -> str
         await AsyncPath(output_path.parent).mkdir(parents=True, exist_ok=True)
         await ap.write_bytes(response.content)
         return output_path.as_posix()
-    except Exception as e:
+    except (httpx.HTTPError, OSError) as e:
         print(f"    Failed to download {description}: {e}")
         return None
 

mediahive/hivescan/indexer.py

@@ -400,7 +400,9 @@ async def _process_movies(
         return await find_playable_file(item.path) is not None
 
     # Filter movies with playable files
-    valid_movies = [item for item in categories[ContentType.MOVIE] if await has_playable(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(
@@ -655,7 +657,9 @@ async def _process_series(
         return len(await find_episode_files(item.path)) > 0
 
     # Filter series with video content
-    valid_series = [item for item in categories[ContentType.SERIES] if await has_video_content(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(

mediahive/hivescan/showreel.py

@@ -74,7 +74,7 @@ async def _run_ffmpeg(
             await _kill_proc(proc)
             try:
                 stdout, stderr = await proc.communicate()
-            except Exception:
+            except OSError, asyncio.SubprocessError:
                 stdout, stderr = b"", b""
             logger.exception(
                 "ffmpeg command timed out. cmd=%s stderr=%s",
@@ -105,7 +105,7 @@ async def _run_ffmpeg(
     except asyncio.CancelledError:
         await _kill_proc(proc)
         raise
-    except Exception:
+    except OSError, asyncio.SubprocessError:
         logger.exception("Unexpected error running ffmpeg command: %s", shlex.join(cmd))
         return None
 
@@ -519,7 +519,7 @@ async def detect_dovi_profile(video_path: str) -> int | None:
     """
     try:
         return (await probe_media_info(video_path)).dovi_profile
-    except Exception as e:
+    except (OSError, ValueError, RuntimeError) as e:
         logger.warning("    DoVi detection error: %s", e)
         return None
 
@@ -545,7 +545,7 @@ async def is_hdr_video(video_path: str) -> bool:
     """
     try:
         return (await probe_media_info(video_path)).is_hdr
-    except Exception:
+    except OSError, ValueError, RuntimeError:
         return False
 
 
@@ -674,7 +674,7 @@ async def get_video_duration(video_path: str) -> float | None:
     """Get the duration of a video file in seconds using ffmpeg probe output."""
     try:
         return (await probe_media_info(video_path)).duration
-    except Exception:
+    except OSError, ValueError, RuntimeError:
         return None
 
 

mediahive/hivescan/tmdb_client.py

@@ -81,7 +81,7 @@ async def _load_from_cache(cache_path: Path):
         if data.get("_cached_none"):
             return None
         return data
-    except Exception:
+    except OSError, TypeError, json.JSONDecodeError:
         return _NOT_FOUND
 
 
@@ -91,7 +91,7 @@ async def _save_to_cache(cache_path: Path, data: dict | None) -> None:
         await AsyncPath(_get_cache_dir()).mkdir(parents=True, exist_ok=True)
         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:
+    except OSError, TypeError, ValueError:
         pass  # Cache write failures are not critical
 
 
@@ -133,7 +133,7 @@ async def tmdb_api_request(
         # Cache the failure (None) to avoid retrying
         await _save_to_cache(cache_path, None)
         return None
-    except Exception:
+    except httpx.HTTPError:
         # Don't cache network errors - they may be transient
         return None
 
@@ -233,10 +233,15 @@ 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)
-    variants.extend(" ".join(words[:num_words]) for num_words in range(len(words) - 1, min_words - 1, -1))
+    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)
-    variants.extend(" ".join(words[start:]) for start in range(1, len(words) - min_words + 1))
+    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/index_store.py

@@ -290,7 +290,7 @@ class IndexStore:
         """Send data to a WS client; mark as dead on failure."""
         try:
             await ws.send_bytes(data)
-        except Exception:
+        except OSError, RuntimeError:
             dead.append(ws)
 
     def broadcast_task(self, task_info: TaskInfo) -> None:

mediahive/server.py

@@ -76,7 +76,7 @@ def _load_resume_positions(root_path: Path) -> dict[str, int]:
     playback_state_path = root_path / ".mediahive" / "playback-state.json"
     try:
         raw = json.loads(playback_state_path.read_text(encoding="utf-8"))
-    except Exception:
+    except OSError, TypeError, json.JSONDecodeError:
         return {}
 
     resume_positions = raw.get("resume_positions") if isinstance(raw, dict) else None
@@ -472,7 +472,7 @@ async def ws_endpoint(ws: WebSocket, root_id: str) -> None:
             await ws.receive_text()
     except WebSocketDisconnect:
         ctx.store.disconnect(ws)
-    except Exception:
+    except OSError, RuntimeError:
         ctx.store.disconnect(ws)
 
 
@@ -492,7 +492,7 @@ async def play_media(root_id: str, request: Request):
     try:
         _open_with_default_app(file_path)
         return {"status": "ok"}
-    except Exception as e:
+    except (OSError, subprocess.SubprocessError, RuntimeError, ValueError) as e:
         raise HTTPException(status_code=500, detail=f"Failed to play media: {e}")
 
 
@@ -527,7 +527,7 @@ async def open_folder(root_id: str, request: Request):
             subprocess.Popen(["xdg-open", str(folder)])
 
         return {"status": "ok"}
-    except Exception as e:
+    except (OSError, subprocess.SubprocessError, RuntimeError, ValueError) as e:
         raise HTTPException(status_code=500, detail=f"Failed to open folder: {e}")
 
 

mediahive/winmain.py

@@ -125,7 +125,7 @@ def _default_playback_state() -> dict[str, object]:
 def _load_playback_state(path: Path) -> dict[str, object]:
     try:
         raw = json.loads(path.read_text(encoding="utf-8"))
-    except Exception:
+    except OSError, TypeError, json.JSONDecodeError:
         return _default_playback_state()
 
     if not isinstance(raw, dict):
@@ -163,7 +163,7 @@ def _media_key_for_filepath(
         try:
             relative = Path(filepath).resolve().relative_to(root.resolve())
             return relative.as_posix(), root
-        except Exception:
+        except OSError, RuntimeError, ValueError:
             continue
     return None
 
@@ -186,7 +186,7 @@ def _load_xinput_get_state():
             fn.argtypes = [ctypes.c_uint, ctypes.POINTER(_XINPUT_STATE)]
             fn.restype = ctypes.c_ulong
             return fn
-        except Exception:
+        except AttributeError, OSError:
             continue
     raise RuntimeError("XInput DLL not found")
 
@@ -412,7 +412,7 @@ def _start_gamepad_remote(
 
         try:
             status = status_future.result()
-        except Exception:
+        except OSError, RuntimeError, ValueError:
             status = None
         status_future = None
 
@@ -662,7 +662,7 @@ def _wait_for_backend(timeout: int | None = None) -> bool:
         try:
             with urllib.request.urlopen(url, timeout=BACKEND_HEALTH_REQUEST_TIMEOUT):
                 return True
-        except Exception:
+        except urllib.error.URLError, TimeoutError, OSError:
             time.sleep(BACKEND_HEALTH_POLL_SECONDS)
 
 
@@ -835,7 +835,7 @@ def winmain() -> None:
         try:
             with urllib.request.urlopen(req, timeout=10):
                 logger.info("Requested initial roots activation")
-        except Exception as exc:
+        except (urllib.error.URLError, TimeoutError, OSError) as exc:
             logger.warning("Initial roots activation request failed: %s", exc)
 
     if not _wait_for_backend(timeout=HEALTH_TIMEOUT):
@@ -863,7 +863,7 @@ def winmain() -> None:
             user_agent = window.evaluate_js("navigator.userAgent")
             if isinstance(user_agent, str):
                 logger.info("Embedded webview user agent: %s", user_agent)
-        except Exception as exc:
+        except (OSError, RuntimeError, ValueError) as exc:
             logger.warning("Could not read embedded user agent: %s", exc)
 
         nonlocal poll_thread

rtorrent_client.py

@@ -48,7 +48,9 @@ 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") -> None:
+    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(
@@ -60,7 +62,7 @@ class RTorrentClient:
         try:
             downloads = self.proxy.download_list("")
             return {h.upper() for h in downloads}
-        except Exception as e:
+        except (OSError, xmlrpc.client.Error) as e:
             print(f"Error getting loaded torrents: {e}")
             return set()
 
@@ -83,7 +85,7 @@ class RTorrentClient:
                 "", str(torrent_path), f'd.directory.set="{download_dir}"'
             )
             return True
-        except Exception as e:
+        except (OSError, xmlrpc.client.Error) as e:
             print(f"Error loading torrent {torrent_path}: {e}")
             return False
 
@@ -105,7 +107,7 @@ class RTorrentClient:
                 "base_path": base_path,  # Full path to data (file or folder)
                 "is_multi_file": is_multi_file,
             }
-        except Exception as e:
+        except (OSError, xmlrpc.client.Error) as e:
             print(f"Error getting torrent info for {info_hash}: {e}")
             return None
 
@@ -129,9 +131,9 @@ class RTorrentClient:
                         info = self.get_torrent_info(info_hash)
                         if info:
                             unregistered.append(info)
-                except Exception:
+                except OSError, xmlrpc.client.Error:
                     continue
-        except Exception as e:
+        except (OSError, xmlrpc.client.Error) as e:
             print(f"Error scanning for unregistered torrents: {e}")
         return unregistered
 
@@ -154,6 +156,6 @@ class RTorrentClient:
                 # Just remove from rtorrent, keep files
                 self.proxy.d.erase(info_hash)
             return True
-        except Exception as e:
+        except (OSError, xmlrpc.client.Error) as e:
             print(f"Error removing torrent {info_hash}: {e}")
             return False

scripts/guibuild.py

@@ -256,7 +256,7 @@ def main() -> None:
 
         print(f"✓ Built successfully: {zip_path}")
         print(f"  Size: {zip_path.stat().st_size / (1024 * 1024):.1f} MB")
-    except Exception as e:
+    except (FileNotFoundError, OSError, RuntimeError, ValueError) as e:
         print(f"✗ Build failed: {e}", file=sys.stderr)
         sys.exit(1)
 

scripts/release.py

@@ -222,7 +222,7 @@ def main() -> None:
         print("\nDone. To publish to PyPI, run:")
         print("  uv publish")
 
-    except Exception as e:
+    except (FileNotFoundError, OSError, RuntimeError, ValueError, httpx.HTTPError) as e:
         print(f"✗ Release failed: {e}", file=sys.stderr)
         sys.exit(1)
 

scripts/rtorrent-manager.py

@@ -88,7 +88,7 @@ def parse_torrent(filepath: Path) -> TorrentInfo | None:
     try:
         with Path(filepath).open("rb") as f:
             data = bencodepy.decode(f.read())
-    except Exception as e:
+    except (OSError, ValueError, TypeError) as e:
         print(f"Error parsing {filepath}: {e}")
         return None
 
@@ -381,7 +381,7 @@ Examples:
                         try:
                             torrent_file.unlink()
                             removed_torrent_files += 1
-                        except Exception:
+                        except OSError:
                             pass
 
                 # Delete the downloaded files
@@ -393,7 +393,7 @@ Examples:
                             download_path.unlink()
                         removed_downloads += 1
                         print(f"  [DEL] {download_path}")
-                    except Exception as e:
+                    except OSError as e:
                         print(f"  [ERR] {download_path}: {e}")
                 else:
                     print(f"  [DEL] {torrent_info['name']} (no data)")
@@ -442,7 +442,7 @@ Examples:
                 try:
                     torrent.path.unlink()
                     print(f"Removed: {torrent.path}")
-                except Exception as e:
+                except OSError as e:
                     print(f"Failed to remove {torrent.path}: {e}")