From 2aa5c9eee775e151041642d573bdcb80d277416a Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:44:48 -0700 Subject: [PATCH] refactor(plugins): contextlib.suppress for try/except-pass, fold None guards --- plugins/context_engine/__init__.py | 4 +--- plugins/cron_providers/__init__.py | 9 +++------ plugins/disk-cleanup/__init__.py | 5 ++--- plugins/disk-cleanup/disk_cleanup.py | 9 +++------ plugins/google_meet/_jsonfile.py | 5 ++--- plugins/google_meet/audio_bridge.py | 5 ++--- plugins/google_meet/cli.py | 5 ++--- plugins/google_meet/process_manager.py | 9 +++------ plugins/google_meet/realtime/openai_client.py | 5 ++--- plugins/plugin_loader.py | 5 ++--- 10 files changed, 22 insertions(+), 39 deletions(-) diff --git a/plugins/context_engine/__init__.py b/plugins/context_engine/__init__.py index 56c84d5acb..57f30054ef 100644 --- a/plugins/context_engine/__init__.py +++ b/plugins/context_engine/__init__.py @@ -40,9 +40,7 @@ def _load_engine_from_dir(engine_dir: Path) -> Optional["ContextEngine"]: # noq mod = _loader.load_plugin_module( f"plugins.context_engine.{name}", engine_dir, parents=("plugins", "plugins.context_engine"), logger=logger) - if mod is None: - return None - return _loader.instance_from_module( + return mod and _loader.instance_from_module( mod, collector=_EngineCollector(engine_name=name), collected_attr="engine", base_cls=ContextEngine, name=name, logger=logger) diff --git a/plugins/cron_providers/__init__.py b/plugins/cron_providers/__init__.py index 4031c8c83b..dc40b10a97 100644 --- a/plugins/cron_providers/__init__.py +++ b/plugins/cron_providers/__init__.py @@ -53,9 +53,8 @@ def find_provider_dir(name: str) -> Optional[Path]: if bundled.is_dir() and (bundled / "__init__.py").exists(): return bundled user_dir = _loader.user_plugins_dir() - if user_dir and (user_dir / name).is_dir() and _is_cron_provider_dir(user_dir / name): - return user_dir / name - return None + user = user_dir / name if user_dir else None + return user if user and user.is_dir() and _is_cron_provider_dir(user) else None def discover_cron_schedulers() -> List[Tuple[str, str, bool]]: @@ -85,9 +84,7 @@ def _load_provider_from_dir(provider_dir: Path) -> Optional["CronScheduler"]: # mod = _loader.load_plugin_module( module_name, provider_dir, parents=("plugins", "plugins.cron_providers"), logger=logger, synthetic_namespace=None if is_bundled else _USER_NAMESPACE) - if mod is None: - return None - return _loader.instance_from_module( + return mod and _loader.instance_from_module( mod, collector=_ProviderCollector(), collected_attr="provider", base_cls=CronScheduler, name=name, logger=logger) diff --git a/plugins/disk-cleanup/__init__.py b/plugins/disk-cleanup/__init__.py index 6ab289c709..b641fab455 100644 --- a/plugins/disk-cleanup/__init__.py +++ b/plugins/disk-cleanup/__init__.py @@ -7,6 +7,7 @@ from __future__ import annotations +import contextlib import logging import re import shlex @@ -38,10 +39,8 @@ def _extract_paths_from_terminal(args: Dict[str, Any], result: str) -> Set[str]: paths: Set[str] = set() cmd = args.get("command") or "" if isinstance(cmd, str) and cmd: - try: # tokenise — catches `touch /tmp/hermes-x/test_foo.py` + with contextlib.suppress(ValueError): # tokenise — catches `touch /tmp/hermes-x/test_foo.py` paths.update(tok for tok in shlex.split(cmd, posix=True) if tok.startswith(("/", "~"))) - except ValueError: - pass # Only scan the result text if it's a reasonable size (avoid 50KB dumps). if isinstance(result, str) and len(result) < 4096: paths.update(_TERMINAL_PATH_REGEX.findall(result)) diff --git a/plugins/disk-cleanup/disk_cleanup.py b/plugins/disk-cleanup/disk_cleanup.py index 16f53509b6..15db509b49 100755 --- a/plugins/disk-cleanup/disk_cleanup.py +++ b/plugins/disk-cleanup/disk_cleanup.py @@ -8,6 +8,7 @@ Scope: strictly HERMES_HOME and /tmp/hermes-*; never ~/.hermes/logs/ or system d from __future__ import annotations +import contextlib import json import logging import shutil @@ -37,11 +38,9 @@ def _state_file(name: str) -> Path: def is_safe_path(path: Path) -> bool: """Accept only paths under HERMES_HOME or ``/tmp/hermes-*`` (rejects /mnt/c etc.).""" - try: + with contextlib.suppress(ValueError, OSError): path.resolve().relative_to(get_hermes_home()) return True - except (ValueError, OSError): - pass parts = path.parts return len(parts) >= 3 and parts[1] == "tmp" and parts[2].startswith("hermes-") @@ -284,13 +283,11 @@ def _sweep_empty_dirs(hermes_home: Path) -> int: while stack: dirpath, visited = stack.pop() if visited: - try: + with contextlib.suppress(OSError): if not any(dirpath.iterdir()): dirpath.rmdir() removed += 1 _log(f"DELETED: {dirpath} (empty dir)") - except OSError: - pass continue stack.append((dirpath, True)) stack.extend((child, False) for child in _subdirs(dirpath, _EMPTY_DIR_SWEEP_PRUNE_DIRS)) diff --git a/plugins/google_meet/_jsonfile.py b/plugins/google_meet/_jsonfile.py index 9758e57e7e..dd3a5549b3 100644 --- a/plugins/google_meet/_jsonfile.py +++ b/plugins/google_meet/_jsonfile.py @@ -2,6 +2,7 @@ from __future__ import annotations +import contextlib import json from pathlib import Path from typing import Any, Optional @@ -24,8 +25,6 @@ def write_json_atomic(path: Path, data: Any, mode: Optional[int] = None) -> None tmp = path.with_suffix(".json.tmp") tmp.write_text(json.dumps(data, indent=2), encoding="utf-8") if mode is not None: - try: + with contextlib.suppress(OSError, NotImplementedError): # best-effort on non-POSIX filesystems tmp.chmod(mode) - except (OSError, NotImplementedError): # best-effort on non-POSIX filesystems - pass tmp.replace(path) diff --git a/plugins/google_meet/audio_bridge.py b/plugins/google_meet/audio_bridge.py index 2c7d350092..c69513e94a 100644 --- a/plugins/google_meet/audio_bridge.py +++ b/plugins/google_meet/audio_bridge.py @@ -7,6 +7,7 @@ Linux: pactl creates a null-sink plus a virtual source on the sink's monitor; ca from __future__ import annotations +import contextlib import platform import subprocess from typing import Optional @@ -56,10 +57,8 @@ class AudioBridge: return if self._platform == "linux": for mod_id in reversed(self._module_ids): # virtual-source before null-sink - try: + with contextlib.suppress(Exception): _pactl("unload-module", str(mod_id), check=False) - except Exception: - pass self._module_ids = [] self._torn_down = True diff --git a/plugins/google_meet/cli.py b/plugins/google_meet/cli.py index 9ebd2a731c..912a7abfe4 100644 --- a/plugins/google_meet/cli.py +++ b/plugins/google_meet/cli.py @@ -10,6 +10,7 @@ from __future__ import annotations import argparse +import contextlib import importlib.util import json import platform @@ -220,10 +221,8 @@ def _cmd_auth() -> int: context = browser.new_context() page = context.new_page() page.goto("https://accounts.google.com/", wait_until="domcontentloaded") - try: + with contextlib.suppress(EOFError): input("press Enter after you've signed in ... ") - except EOFError: - pass context.storage_state(path=str(path)) browser.close() except Exception as e: diff --git a/plugins/google_meet/process_manager.py b/plugins/google_meet/process_manager.py index 30e9030e33..c415796254 100644 --- a/plugins/google_meet/process_manager.py +++ b/plugins/google_meet/process_manager.py @@ -8,6 +8,7 @@ across turns can find the bot. The bot is a detached subprocess reached via file from __future__ import annotations +import contextlib import json import os import signal @@ -47,10 +48,8 @@ def _active_pid() -> int: def _kill(pid: int, sig) -> None: - try: + with contextlib.suppress(ProcessLookupError): os.kill(pid, sig) - except ProcessLookupError: - pass _NO_ACTIVE = {"ok": False, "reason": "no active meeting"} @@ -72,10 +71,8 @@ def start(url: str, *, out_dir: Optional[Path] = None, headed: bool = False, out.mkdir(parents=True, exist_ok=True) # Wipe stale files from a previous run of this meeting id. for name in ("transcript.txt", "status.json"): - try: + with contextlib.suppress(OSError): (out / name).unlink() - except OSError: - pass env = {**os.environ, "HERMES_MEET_URL": url, "HERMES_MEET_OUT_DIR": str(out), "HERMES_MEET_GUEST_NAME": guest_name} for value, var in ( diff --git a/plugins/google_meet/realtime/openai_client.py b/plugins/google_meet/realtime/openai_client.py index 1f14f3933b..0eb8cc94bb 100644 --- a/plugins/google_meet/realtime/openai_client.py +++ b/plugins/google_meet/realtime/openai_client.py @@ -7,6 +7,7 @@ Chrome's fake mic. One sync WebSocket per session; ``websockets`` is imported la from __future__ import annotations import base64 +import contextlib import json import threading import time @@ -69,10 +70,8 @@ class RealtimeSession: def close(self) -> None: if self._ws is not None: - try: + with contextlib.suppress(Exception): self._ws.close() - except Exception: - pass self._ws = None def speak(self, text: str, timeout: float = 30.0) -> dict: diff --git a/plugins/plugin_loader.py b/plugins/plugin_loader.py index 49b719f75b..7f9d3594d1 100644 --- a/plugins/plugin_loader.py +++ b/plugins/plugin_loader.py @@ -5,6 +5,7 @@ or an ABC-subclass fallback.""" from __future__ import annotations +import contextlib import importlib.machinery import importlib.util import logging @@ -150,10 +151,8 @@ def instance_from_module(mod: Any, *, collector: Any, collected_attr: str, base_ for attr_name in dir(mod): attr = getattr(mod, attr_name, None) if isinstance(attr, type) and issubclass(attr, base_cls) and attr is not base_cls: - try: + with contextlib.suppress(Exception): return attr() - except Exception: - pass return None