hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 29 Packages Projects Releases Wiki Activity

refactor(plugins): contextlib.suppress for try/except-pass, fold None guards

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:44:48 -07:00
parent 7f1a19993d
commit 2aa5c9eee7
10 changed files with 22 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
plugins/context_engine/__init__.py
View File

@@ -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)

9
plugins/cron_providers/__init__.py
View File

@@ -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)

5
plugins/disk-cleanup/__init__.py
View File

@@ -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))

9
plugins/disk-cleanup/disk_cleanup.py
View File

@@ -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))

5
plugins/google_meet/_jsonfile.py
View File

@@ -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)

5
plugins/google_meet/audio_bridge.py
View File

@@ -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

5
plugins/google_meet/cli.py
View File

@@ -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:

9
plugins/google_meet/process_manager.py
View File

@@ -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 (

5
plugins/google_meet/realtime/openai_client.py
View File

@@ -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:

5
plugins/plugin_loader.py
View File

@@ -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