Reduce cognitive complexity in WebRTC + REST hot paths (S3776)

Each of the flagged ``S3776`` functions was split into smaller named
helpers — same behaviour, smaller per-function complexity. No public
API changed; tests still pass headless + WS + TLS.

webrtc_host.py:
- _handle_ctrl_message: dispatch dict + per-message handlers
  (_handle_input_message / _handle_send_sas_message /
  _handle_annotate / _handle_renegotiate_answer) plus a shared
  _safe_audit_log so the rate-limit logging isn't duplicated.
- _async_apply_renegotiate_answer: split track re-subscription into
  _maybe_resubscribe_viewer_video / _audio with a shared
  _receiver_track helper.
- _handle_auth: split into _reject_auth + _auto_approve_via_trust /
  _whitelist with early-return semantics.
- _snapshot_remote_ip: extract _extract_remote_ip so the nested
  candidate-pair walk reads top-down.

webrtc_viewer.py:
- _handle_ctrl_message: dispatch dict and per-handler methods
  matching the host side; _handle_inbox_op_result covers the two
  inbox response variants.

webrtc_stats.py:
- _sample: factor candidate-pair / remote-inbound-rtp branches
  into _absorb_entry / _absorb_remote_inbound.

adaptive_bitrate.py:
- on_stats: split into _react_to_hard_cap and _react_to_quality;
  downscale / upscale logic each got their own helper plus
  _should_downscale / _should_upscale predicates.

hw_codec.py:
- install_hardware_codec: extract _open_codec_context (codec init
  + libx264 fallback) and _shape_changed; the patched closure now
  reads as a thin coordinator.

webrtc_dialogs.py:
- _refresh: extract _populate_row + _is_stale.
- _on_import: extract _prompt_import_data, _import_one,
  _extract_fingerprints, _confirm_overwrite.

webrtc_panel.py:
- _read_webrtc_config: introduce _checked_or and _read_region
  helpers to eliminate the long hasattr ladder.
- _on_sessions_context_menu: split into _trust_session_viewer +
  _copy_session_id_to_clipboard.

webrtc_workers.py:
- HostPublishLoopWorker.run: extract _publish_one_session,
  _handle_signaling_error, _safe_stop_session_if so the run loop
  reads as a one-line state machine.

host_service.py:
- main: replace the long if/elif chain on args.command with a
  module-level _COMMAND_DISPATCH dict and per-command helper
  functions.

address_book.py:
- upsert: extract _find_entry_locked, _refresh_entry_locked,
  _build_entry. The early return / merge / append flow now reads
  linearly.

web_viewer/index.html:
- handleControlMessage: dispatch object + handleAuthOk /
  verifyFingerprint / rememberFingerprint helpers replace the
  if/else-if cascade.
- setLanguage: replace nested ternary with explicit if/elif chain.

Author: JE-Chen

je_auto_control/gui/remote_desktop/webrtc_dialogs.py

@@ -366,28 +366,37 @@ def _refresh(self) -> None:
         entries = self._known.list_entries()
         self._table.setRowCount(len(entries))
         for row, (host_id, fps) in enumerate(sorted(entries.items())):
-            items = [
-                QTableWidgetItem(host_id),
-                QTableWidgetItem(_short_fp(fps.get("app_fp"))),
-                QTableWidgetItem(_short_fp(fps.get("dtls_fp"))),
-                QTableWidgetItem(_format_last_seen(fps.get("last_seen"))),
-            ]
-            is_stale = False
-            last_seen = fps.get("last_seen")
-            if last_seen:
-                try:
-                    dt = datetime.fromisoformat(last_seen)
-                    if now - dt > stale_after:
-                        is_stale = True
-                except (TypeError, ValueError):
-                    pass
-            if is_stale:
-                tip = _t("rd_webrtc_kh_stale_tip")
-                for it in items:
-                    it.setForeground(stale_color)
-                    it.setToolTip(tip)
-            for col, item in enumerate(items):
-                self._table.setItem(row, col, item)
+            self._populate_row(row, host_id, fps,
+                               now=now, stale_after=stale_after,
+                               stale_color=stale_color)
+
+    def _populate_row(self, row: int, host_id: str, fps: dict, *,
+                      now, stale_after, stale_color) -> None:
+        items = [
+            QTableWidgetItem(host_id),
+            QTableWidgetItem(_short_fp(fps.get("app_fp"))),
+            QTableWidgetItem(_short_fp(fps.get("dtls_fp"))),
+            QTableWidgetItem(_format_last_seen(fps.get("last_seen"))),
+        ]
+        if self._is_stale(fps.get("last_seen"), now=now,
+                          stale_after=stale_after):
+            tip = _t("rd_webrtc_kh_stale_tip")
+            for it in items:
+                it.setForeground(stale_color)
+                it.setToolTip(tip)
+        for col, item in enumerate(items):
+            self._table.setItem(row, col, item)
+
+    @staticmethod
+    def _is_stale(last_seen, *, now, stale_after) -> bool:
+        if not last_seen:
+            return False
+        from datetime import datetime
+        try:
+            dt = datetime.fromisoformat(last_seen)
+        except (TypeError, ValueError):
+            return False
+        return now - dt > stale_after
 
     def _on_forget(self) -> None:
         rows = sorted(
@@ -454,57 +463,74 @@ def _on_export(self) -> None:
             QMessageBox.warning(self, "WebRTC", str(error))
 
     def _on_import(self) -> None:
+        data = self._prompt_import_data()
+        if data is None:
+            return
+        existing = self._known.list_entries()
+        added = 0
+        skipped = 0
+        for host_id, value in data.items():
+            outcome = self._import_one(host_id, value, existing)
+            if outcome == "added":
+                added += 1
+            elif outcome == "skipped":
+                skipped += 1
+        QMessageBox.information(
+            self, "WebRTC",
+            _t("rd_webrtc_kh_import_done").format(added=added, skipped=skipped),
+        )
+        self._refresh()
+
+    def _prompt_import_data(self):
         import json
         path, _filter = QFileDialog.getOpenFileName(
             self, _t("rd_webrtc_kh_import"), "", "JSON (*.json);;All (*)",
         )
         if not path:
-            return
+            return None
         try:
             with open(path, "r", encoding="utf-8") as fh:
                 data = json.load(fh)
         except (OSError, json.JSONDecodeError) as error:
             QMessageBox.warning(self, "WebRTC", str(error))
-            return
+            return None
         if not isinstance(data, dict):
             QMessageBox.warning(
                 self, "WebRTC", _t("rd_webrtc_kh_import_bad"),
             )
-            return
-        existing = self._known.list_entries()
-        added = 0
-        skipped = 0
-        for host_id, value in data.items():
-            if not isinstance(host_id, str):
-                continue
-            app_fp = None
-            dtls_fp = None
-            if isinstance(value, str):
-                app_fp = value
-            elif isinstance(value, dict):
-                app_fp = value.get("app_fp")
-                dtls_fp = value.get("dtls_fp")
-            else:
-                continue
-            if host_id in existing:
-                result = QMessageBox.question(
-                    self, "WebRTC",
-                    _t("rd_webrtc_kh_import_overwrite").format(host=host_id),
-                    QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
-                )
-                if result != QMessageBox.StandardButton.Yes:
-                    skipped += 1
-                    continue
-            if isinstance(app_fp, str) and app_fp:
-                self._known.remember(host_id, app_fp)
-            if isinstance(dtls_fp, str) and dtls_fp:
-                self._known.remember_dtls_fingerprint(host_id, dtls_fp)
-            added += 1
-        QMessageBox.information(
+            return None
+        return data
+
+    def _import_one(self, host_id, value, existing) -> str:
+        """Return ``"added"``, ``"skipped"``, or ``"ignored"`` per entry."""
+        if not isinstance(host_id, str):
+            return "ignored"
+        app_fp, dtls_fp = self._extract_fingerprints(value)
+        if app_fp is None and dtls_fp is None:
+            return "ignored"
+        if host_id in existing and not self._confirm_overwrite(host_id):
+            return "skipped"
+        if isinstance(app_fp, str) and app_fp:
+            self._known.remember(host_id, app_fp)
+        if isinstance(dtls_fp, str) and dtls_fp:
+            self._known.remember_dtls_fingerprint(host_id, dtls_fp)
+        return "added"
+
+    @staticmethod
+    def _extract_fingerprints(value):
+        if isinstance(value, str):
+            return value, None
+        if isinstance(value, dict):
+            return value.get("app_fp"), value.get("dtls_fp")
+        return None, None
+
+    def _confirm_overwrite(self, host_id: str) -> bool:
+        result = QMessageBox.question(
             self, "WebRTC",
-            _t("rd_webrtc_kh_import_done").format(added=added, skipped=skipped),
+            _t("rd_webrtc_kh_import_overwrite").format(host=host_id),
+            QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
         )
-        self._refresh()
+        return result == QMessageBox.StandardButton.Yes
 
     def _on_forget_stale(self) -> None:
         from datetime import datetime, timedelta, timezone

je_auto_control/gui/remote_desktop/webrtc_panel.py

@@ -148,73 +148,59 @@ def _build_advanced_group(panel: TranslatableMixin,
     return group
 
 
+def _checked_or(panel, attr: str, default: bool = False) -> bool:
+    """Return ``panel.<attr>.isChecked()`` if the widget exists, else default."""
+    widget = getattr(panel, attr, None)
+    return widget.isChecked() if widget is not None else default
+
+
+def _read_region(panel) -> Optional[tuple]:
+    edit = getattr(panel, "_region_edit", None)
+    if edit is None:
+        return None
+    text = edit.text().strip()
+    if not text:
+        return None
+    try:
+        parts = [int(p.strip()) for p in text.split(",")]
+    except (ValueError, TypeError):
+        return None
+    return tuple(parts) if len(parts) == 4 else None
+
+
 def _read_webrtc_config(panel) -> WebRTCConfig:
     """Build a WebRTCConfig from the advanced group + monitor/fps fields."""
     from je_auto_control.utils.remote_desktop.webrtc_transport import (
         _DEFAULT_STUN_SERVERS,
     )
     stun_field = panel._stun_edit.text().strip()
-    if stun_field:
-        ice_servers = [stun_field]
-    else:
-        ice_servers = list(_DEFAULT_STUN_SERVERS)
-    turn_url = panel._turn_edit.text().strip() or None
-    show_cursor = (panel._cursor_check.isChecked()
-                   if hasattr(panel, "_cursor_check") else True)
-    accept_viewer_video = (
-        panel._accept_viewer_video_check.isChecked()
-        if hasattr(panel, "_accept_viewer_video_check") else False
-    )
-    accept_opus_audio = (
-        panel._accept_opus_audio_check.isChecked()
-        if hasattr(panel, "_accept_opus_audio_check") else False
-    )
-    share_my_screen = (
-        panel._share_my_screen_check.isChecked()
-        if hasattr(panel, "_share_my_screen_check") else False
-    )
-    share_opus_mic = (
-        panel._share_opus_mic_check.isChecked()
-        if hasattr(panel, "_share_opus_mic_check") else False
+    ice_servers = [stun_field] if stun_field else list(_DEFAULT_STUN_SERVERS)
+    monitor = (
+        int(panel._monitor_combo.currentData() or _DEFAULT_MONITOR)
+        if hasattr(panel, "_monitor_combo") else _DEFAULT_MONITOR
     )
+    fps = (int(panel._fps_spin.value())
+           if hasattr(panel, "_fps_spin") else _DEFAULT_FPS)
     max_bitrate = (
         int(panel._max_bitrate_spin.value())
         if hasattr(panel, "_max_bitrate_spin") else 0
     )
-    region = None
-    if hasattr(panel, "_region_edit"):
-        text = panel._region_edit.text().strip()
-        if text:
-            try:
-                parts = [int(p.strip()) for p in text.split(",")]
-                if len(parts) == 4:
-                    region = tuple(parts)
-            except (ValueError, TypeError):
-                region = None
-    config = WebRTCConfig(
+    return WebRTCConfig(
         ice_servers=ice_servers,
-        turn_url=turn_url,
+        turn_url=panel._turn_edit.text().strip() or None,
         turn_username=panel._turn_user_edit.text().strip() or None,
         turn_credential=panel._turn_cred_edit.text() or None,
-        monitor_index=(
-            int(panel._monitor_combo.currentData() or _DEFAULT_MONITOR)
-            if hasattr(panel, "_monitor_combo") else _DEFAULT_MONITOR
-        ),
-        fps=int(panel._fps_spin.value()) if hasattr(panel, "_fps_spin")
-        else _DEFAULT_FPS,
-        show_cursor=show_cursor,
-        accept_viewer_video=accept_viewer_video,
-        accept_viewer_audio_opus=accept_opus_audio,
-        share_my_screen=share_my_screen,
-        share_my_audio_opus=share_opus_mic,
+        monitor_index=monitor,
+        fps=fps,
+        show_cursor=_checked_or(panel, "_cursor_check", default=True),
+        accept_viewer_video=_checked_or(panel, "_accept_viewer_video_check"),
+        accept_viewer_audio_opus=_checked_or(panel, "_accept_opus_audio_check"),
+        share_my_screen=_checked_or(panel, "_share_my_screen_check"),
+        share_my_audio_opus=_checked_or(panel, "_share_opus_mic_check"),
         max_bitrate_kbps=max_bitrate,
-        region=region,
-        host_voice=(
-            panel._host_voice_check.isChecked()
-            if hasattr(panel, "_host_voice_check") else False
-        ),
+        region=_read_region(panel),
+        host_voice=_checked_or(panel, "_host_voice_check"),
     )
-    return config
 
 
 class _WebRTCHostPanel(TranslatableMixin, QWidget):
@@ -995,7 +981,7 @@ def _refresh_sessions_table(self) -> None:
             self._sessions_table.setItem(row, 4, QTableWidgetItem(connected))
 
     def _on_sessions_context_menu(self, position) -> None:
-        from PySide6.QtWidgets import QApplication, QMenu
+        from PySide6.QtWidgets import QMenu
         if self._multi_host is None:
             return
         row = self._sessions_table.rowAt(position.y())
@@ -1019,20 +1005,27 @@ def _on_sessions_context_menu(self, position) -> None:
         if chosen is disc:
             self._on_disconnect_selected()
         elif chosen is trust and viewer_id:
-            try:
-                # find the actual full viewer_id from the host
-                with self._multi_host._lock:
-                    host = self._multi_host._sessions.get(sid)
-                full_vid = host.pending_viewer_id if host is not None else None
-                if full_vid:
-                    self._trust_list.add(full_vid, label=f"sess {sid[:6]}")
-                    self._refresh_trusted_list()
-            except (RuntimeError, OSError, ValueError) as error:
-                autocontrol_logger.warning("trust viewer: %r", error)
+            self._trust_session_viewer(sid)
         elif chosen is copy_id and sid:
-            clip = QApplication.clipboard()
-            if clip is not None:
-                clip.setText(sid)
+            self._copy_session_id_to_clipboard(sid)
+
+    def _trust_session_viewer(self, sid: str) -> None:
+        try:
+            with self._multi_host._lock:
+                host = self._multi_host._sessions.get(sid)
+            full_vid = host.pending_viewer_id if host is not None else None
+            if full_vid:
+                self._trust_list.add(full_vid, label=f"sess {sid[:6]}")
+                self._refresh_trusted_list()
+        except (RuntimeError, OSError, ValueError) as error:
+            autocontrol_logger.warning("trust viewer: %r", error)
+
+    @staticmethod
+    def _copy_session_id_to_clipboard(sid: str) -> None:
+        from PySide6.QtWidgets import QApplication
+        clip = QApplication.clipboard()
+        if clip is not None:
+            clip.setText(sid)
 
     def _on_disconnect_selected(self) -> None:
         if self._multi_host is None: