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

refactor(gateway/platforms): webhook/msgraph/keyboards/chunked_upload — walrus guards, fold small helpers

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:13:26 -07:00
parent 873405895f
commit 4009eeb59f
5 changed files with 24 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/platforms/msgraph_webhook.py
+4 -8
View File
@@ -116,8 +116,7 @@ class MSGraphWebhookAdapter(BasePlatformAdapter):
self._notification_scheduler: Optional[NotificationScheduler] = None
self._seen_receipts: set[str] = set()
self._seen_receipt_order: deque[str] = deque()
self._accepted_count = 0
self._duplicate_count = 0
self._accepted_count = self._duplicate_count = 0
def set_notification_scheduler(self, scheduler: Optional[NotificationScheduler]) -> None:
self._notification_scheduler = scheduler
@@ -176,8 +175,7 @@ class MSGraphWebhookAdapter(BasePlatformAdapter):
are rejected so the endpoint can't be enumerated."""
if not self._source_ip_allowed(request):
return web.Response(status=403)
validation_token = request.query.get("validationToken", "")
if not validation_token:
if not (validation_token := request.query.get("validationToken", "")):
return web.Response(status=400)
return web.Response(text=validation_token, content_type="text/plain")
@@ -192,8 +190,7 @@ class MSGraphWebhookAdapter(BasePlatformAdapter):
# Bad clientState is an auth failure: a fully forged batch gets 403 so the sender stops
# retrying; legitimate Graph retries carry a valid clientState → accepted/duplicate paths.
return "auth"
explicit_id = str(notification.get("id") or "").strip()
receipt_key = f"id:{explicit_id}" if explicit_id else None
receipt_key = f"id:{explicit_id}" if (explicit_id := str(notification.get("id") or "").strip()) else None
if receipt_key is not None:
if receipt_key in self._seen_receipts:
return "duplicate"
@@ -206,8 +203,7 @@ class MSGraphWebhookAdapter(BasePlatformAdapter):
if not self._source_ip_allowed(request):
return web.Response(status=403)
# Graph never sends validationToken on POST, but tolerate clients replaying it in-band.
validation_token = request.query.get("validationToken", "")
if validation_token:
if validation_token := request.query.get("validationToken", ""):
return web.Response(text=validation_token, content_type="text/plain")
status, notifications = await self._read_notifications(request)
if status:
gateway/platforms/qqbot/chunked_upload.py
+2 -4
View File
@@ -35,8 +35,7 @@ _MD5_10M_SIZE = 10_002_432 # first N bytes used for the ``md5_10m`` hash (per Q
class _UploadError(Exception):
def __init__(self, file_name: str, file_size: int, message: str) -> None:
self.file_name = file_name
self.file_size = file_size
self.file_name, self.file_size = file_name, file_size
super().__init__(message)
@property
@@ -84,8 +83,7 @@ class _PrepareResult:
def _parse_prepare_response(raw: Dict[str, Any]) -> _PrepareResult:
"""Parse upload_prepare response (either bare or wrapped in ``data``)."""
src = raw.get("data") if isinstance(raw.get("data"), dict) else raw
upload_id = str(src.get("upload_id", ""))
if not upload_id:
if not (upload_id := str(src.get("upload_id", ""))):
raise ValueError(f"upload_prepare response missing upload_id: {str(raw)[:200]}")
block_size = int(src.get("block_size", 0))
raw_parts = src.get("parts") or src.get("part_list") or []
gateway/platforms/qqbot/keyboards.py
+3 -6
View File
@@ -79,14 +79,12 @@ class InlineKeyboard(_Serializable):
def parse_approval_button_data(button_data: str) -> Optional[tuple[str, str]]:
"""Parse approval ``button_data`` into ``(session_key, decision)`` or ``None``."""
m = _APPROVAL_DATA_RE.match(button_data or "")
return m.groups() if m else None
return m.groups() if (m := _APPROVAL_DATA_RE.match(button_data or "")) else None
def parse_update_prompt_button_data(button_data: str) -> Optional[str]:
"""Parse update-prompt ``button_data`` into ``'y'`` / ``'n'`` or ``None``."""
m = _UPDATE_PROMPT_RE.match(button_data or "")
return m.group(1) if m else None
return m.group(1) if (m := _UPDATE_PROMPT_RE.match(button_data or "")) else None
def _single_row_keyboard(group_id: str, *buttons: tuple) -> InlineKeyboard:
@@ -183,8 +181,7 @@ _SCENE_NAMES = {0: "guild", 1: "group", 2: "c2c"}
def parse_interaction_event(raw: Dict[str, Any]) -> InteractionEvent:
"""Parse a raw ``INTERACTION_CREATE`` dispatch payload (``d``)."""
data_raw = raw.get("data") or {}
resolved = data_raw.get("resolved") or {}
scene_code = int(raw.get("chat_type", 0) or 0)
resolved, scene_code = data_raw.get("resolved") or {}, int(raw.get("chat_type", 0) or 0)
return InteractionEvent(
id=str(raw.get("id", "")), type=int(data_raw.get("type", 0) or 0), chat_type=scene_code,
scene=_SCENE_NAMES.get(scene_code, ""), group_openid=str(raw.get("group_openid", "")),
gateway/platforms/signal.py
+9 -15
View File
@@ -74,8 +74,7 @@ def _guess_extension(data: bytes) -> str:
return ext
if len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP":
return ".webp"
container = sniff_container(data)
if container is not None:
if (container := sniff_container(data)) is not None:
return CONTAINER_TO_EXT[container]
return ".zip" if data[:2] == b"PK" else ".bin"
@@ -225,8 +224,7 @@ class SignalAdapter(BasePlatformAdapter):
if not self.http_url or not self.account:
logger.error("Signal: SIGNAL_HTTP_URL and SIGNAL_ACCOUNT are required")
return False
# Scoped lock prevents duplicate Signal listeners for the same phone.
lock_acquired = False
lock_acquired = False # scoped lock prevents duplicate Signal listeners for the same phone
try:
if not self._acquire_platform_lock('signal-phone', self.account, 'Signal account'):
return False
@@ -337,8 +335,7 @@ class SignalAdapter(BasePlatformAdapter):
await asyncio.sleep(HEALTH_CHECK_INTERVAL)
if not self._running:
break
elapsed = time.time() - self._last_sse_activity
if elapsed <= HEALTH_CHECK_STALE_THRESHOLD:
if (elapsed := time.time() - self._last_sse_activity) <= HEALTH_CHECK_STALE_THRESHOLD:
continue
logger.warning("Signal: SSE idle for %.0fs, checking daemon health", elapsed)
try:
@@ -510,17 +507,15 @@ class SignalAdapter(BasePlatformAdapter):
def _remember_recipient_identifiers(self, number: Optional[str], service_id: Optional[str]) -> None:
"""Cache any number↔UUID mapping observed from Signal envelopes."""
if not number or not service_id or not _is_signal_service_id(service_id):
return
self._recipient_uuid_by_number[number] = service_id
self._recipient_number_by_uuid[service_id] = number
if number and service_id and _is_signal_service_id(service_id):
self._recipient_uuid_by_number[number] = service_id
self._recipient_number_by_uuid[service_id] = number
@staticmethod
def _extract_quote_author(quote_data: Any) -> Optional[str]:
"""Return the best available Signal sender identifier from quote metadata."""
if not isinstance(quote_data, dict):
return None
return next((str(quote_data[k]) for k in _QUOTE_AUTHOR_KEYS if quote_data.get(k)), None)
keys = _QUOTE_AUTHOR_KEYS if isinstance(quote_data, dict) else ()
return next((str(quote_data[k]) for k in keys if quote_data.get(k)), None)
def _quote_references_own_message(self, reply_to_id: Optional[str], reply_to_author: Optional[str]) -> bool:
"""True when a Signal quote points at this adapter's outbound message."""
@@ -538,8 +533,7 @@ class SignalAdapter(BasePlatformAdapter):
if timestamp is None:
return
key = str(timestamp)
# Re-insert to mark most-recently-used so eviction drops genuinely old entries.
self._sent_message_timestamps.pop(key, None)
self._sent_message_timestamps.pop(key, None) # re-insert as most-recently-used so eviction drops old ones
self._sent_message_timestamps[key] = None
while len(self._sent_message_timestamps) > self._max_sent_message_timestamps:
self._sent_message_timestamps.popitem(last=False)
gateway/platforms/webhook.py
+6 -11
View File
@@ -105,8 +105,7 @@ def _json_error(message: str, status: int) -> "web.Response":
def _peek_session_id(store, session_key: str):
"""Prefer the store's lock-held accessor; the private-path fallback is for older stores / test doubles."""
peek = getattr(store, "peek_session_id", None)
if callable(peek):
if callable(peek := getattr(store, "peek_session_id", None)):
return peek(session_key)
if hasattr(store, "_ensure_loaded"):
with suppress(Exception):
@@ -285,8 +284,7 @@ class WebhookAdapter(BasePlatformAdapter):
def _record_rate_limit_hit(self, route_name: str, now: float) -> bool:
"""Return True if route is still within limit after recording this hit."""
window = self._rate_counts.get(route_name)
if not isinstance(window, deque):
if not isinstance(window := self._rate_counts.get(route_name), deque):
window = self._rate_counts[route_name] = deque(window or ())
cutoff = now - _RATE_WINDOW_SECONDS
while window and window[0] < cutoff:
@@ -298,8 +296,7 @@ class WebhookAdapter(BasePlatformAdapter):
def _record_delivery_id(self, delivery_id: str, now: float) -> bool:
"""Return True when this delivery should be processed."""
seen_at = self._seen_deliveries.get(delivery_id)
if seen_at is not None and now - seen_at < self._idempotency_ttl:
if (seen_at := self._seen_deliveries.get(delivery_id)) is not None and now - seen_at < self._idempotency_ttl:
return False
if seen_at is not None:
self._seen_deliveries.pop(delivery_id, None)
@@ -462,9 +459,8 @@ class WebhookAdapter(BasePlatformAdapter):
logger.exception("[webhook] direct-deliver failed route=%s delivery=%s", route_name, delivery_id)
return web.json_response(failed, status=502)
if result.success:
return web.json_response(
{"status": "delivered", "route": route_name, "target": delivery["deliver"], "delivery_id": delivery_id},
status=200)
return web.json_response({"status": "delivered", "route": route_name, "target": delivery["deliver"],
"delivery_id": delivery_id}, status=200)
# Target rejected it — 502 with a generic error (don't leak adapter detail).
logger.warning("[webhook] direct-deliver target rejected route=%s target=%s error=%s", route_name,
delivery["deliver"], result.error)
@@ -756,8 +752,7 @@ class WebhookAdapter(BasePlatformAdapter):
target_platform = Platform(platform_name)
except ValueError:
return SendResult(success=False, error=f"Unknown platform: {platform_name}")
adapter = self._find_adapter(target_platform)
if not adapter:
if not (adapter := self._find_adapter(target_platform)):
return SendResult(success=False, error=f"Platform {platform_name} not connected")
extra = delivery.get("deliver_extra", {})
chat_id = extra.get("chat_id", "")