refactor(all): walrus guards, comprehension for attachment caching

plugins/platforms/line/adapter.py

@@ -342,6 +342,7 @@ _OUTBOUND_MEDIA = {
 _INBOUND_MEDIA_EXT = {"image": ".jpg", "audio": ".m4a", "video": ".mp4", "file": ".bin"}
 _INBOUND_AV_CACHERS = {"audio": cache_audio_from_bytes, "video": cache_video_from_bytes}
 _LIFECYCLE_EVENTS = frozenset({"follow", "unfollow", "join", "leave"})
+_ENV_SEED_KEYS = (("LINE_HOST", "host"), ("LINE_PUBLIC_URL", "public_url"), ("LINE_HOME_CHANNEL", "home_channel"))
 
 
 class LineAdapter(BasePlatformAdapter):
@@ -868,9 +869,7 @@ def _env_enablement() -> Optional[Dict[str, Any]]:
     if os.getenv("LINE_PORT"):
         with contextlib.suppress(ValueError):
             seeded["port"] = int(os.environ["LINE_PORT"])
-    for env, key in (("LINE_HOST", "host"), ("LINE_PUBLIC_URL", "public_url"), ("LINE_HOME_CHANNEL", "home_channel")):
-        if os.getenv(env):
-            seeded[key] = os.environ[env]
+    seeded.update({key: os.environ[env] for env, key in _ENV_SEED_KEYS if os.getenv(env)})
     return seeded
 
 

plugins/platforms/simplex/adapter.py

@@ -247,8 +247,7 @@ class SimplexAdapter(BasePlatformAdapter):
     async def _on_rcv_file_descr_ready(self, resp: dict) -> None:
         """XFTP files fire this before newChatItems; start the download now, the chat item arrives later."""
         rcv_file = resp.get("rcvFileTransfer", {}) or {}
-        file_id = rcv_file.get("fileId") if isinstance(rcv_file, dict) else None
-        if file_id is not None:
+        if (file_id := rcv_file.get("fileId") if isinstance(rcv_file, dict) else None) is not None:
             logger.debug("SimpleX: rcvFileDescrReady for fileId=%s — sending /freceive", file_id)
             await self._send_fire_and_forget(f"/freceive {file_id}")
 
@@ -610,14 +609,11 @@ def _env_enablement() -> Optional[dict]:
     if not ws_url:
         return None
     seed: dict = {"ws_url": ws_url}
-    auto_accept = _get_scoped_secret("SIMPLEX_AUTO_ACCEPT", "").strip().lower()
-    if auto_accept:
+    if auto_accept := _get_scoped_secret("SIMPLEX_AUTO_ACCEPT", "").strip().lower():
         seed["auto_accept"] = auto_accept not in {"0", "false", "no"}
-    group_allowed = _get_scoped_secret("SIMPLEX_GROUP_ALLOWED", "").strip()
-    if group_allowed:
+    if group_allowed := _get_scoped_secret("SIMPLEX_GROUP_ALLOWED", "").strip():
         seed["group_allowed"] = group_allowed
-    home = _get_scoped_secret("SIMPLEX_HOME_CHANNEL", "").strip()
-    if home:
+    if home := _get_scoped_secret("SIMPLEX_HOME_CHANNEL", "").strip():
         seed["home_channel"] = {"chat_id": home, "name": _get_scoped_secret("SIMPLEX_HOME_CHANNEL_NAME", "").strip() or home}
     return seed
 

plugins/platforms/teams/adapter.py

@@ -169,11 +169,9 @@ def _env_enablement() -> dict | None:
     port = coerce_port(os.getenv("TEAMS_PORT", "").strip(), None)
     if port is not None:
         seed["port"] = port
-    service_url = os.getenv("TEAMS_SERVICE_URL", "").strip()
-    if service_url:
+    if service_url := os.getenv("TEAMS_SERVICE_URL", "").strip():
         seed["service_url"] = service_url
-    home = os.getenv("TEAMS_HOME_CHANNEL", "").strip()
-    if home:
+    if home := os.getenv("TEAMS_HOME_CHANNEL", "").strip():
         seed["home_channel"] = {"chat_id": home, "name": os.getenv("TEAMS_HOME_CHANNEL_NAME", "Home")}
     return seed
 
@@ -470,12 +468,8 @@ class TeamsAdapter(BasePlatformAdapter):
             user_id=str(user_id),
             user_name=getattr(from_account, "name", None) or "",
             guild_id=getattr(conv, "tenant_id", None) or self._tenant_id)
-        media: list = []  # (path, media_type, kind)
-        for att in getattr(activity, "attachments", None) or []:
-            cached = await self._cache_attachment(att)
-            if cached:
-                media.append(cached)
-        media_kinds = [kind for _, _, kind in media]
+        media: list = [m for m in [await self._cache_attachment(a) for a in getattr(activity, "attachments", None) or []] if m]
+        media_kinds = [kind for _, _, kind in media]  # media items are (path, media_type, kind)
         msg_type = next((t for kind, t in _MEDIA_KIND_PRECEDENCE if kind in media_kinds), MessageType.TEXT)
         await self.handle_message(MessageEvent(
             text=text, source=source, message_type=msg_type, message_id=msg_id,