# -*- coding: utf-8 -*- """ AzA Empfang Web-Huelle: eigener Desktop-Prozess (pywebview). Drei Betriebsmodi, klar getrennt: * ``desktop_shell`` Vom AzA-Hauptprogramm (basis14) per subprocess gestartet mit /empfang/shell/launch?token=... (KEIN target=...). Fenstertitel: "AzA-Empfang \xb7 Desktop" Storage: %APPDATA%\\AzA\\EmpfangWebView AUMID: ch.aza-medwork.empfang.shellwebview * ``empfang_chat_shell`` Separat installierbare Huelle (AZA_EmpfangShell.exe via aza_empfang_chat_setup.exe) auf Empfangs-/MPA-Computern. Default-URL beim Doppelklick: https://empfang.aza-medwork.ch/empfang/ ?empfang_chat_shell=1&shell_source=empfang_chat_shell Fenstertitel: "AzA Empfang Chat" Storage: %APPDATA%\\AzA\\EmpfangChatWebView AUMID: ch.aza-medwork.empfang.chatshell * ``minichat`` Kleines Empfang-Popup (?minichat=1). Erbt Storage/AUMID des Parent-Modus, damit die Login-Session erhalten bleibt. Fenstertitel: "AzA MiniChat" WebView2-Profil: pywebview nutzt standardmaessig ``private_mode=True`` (ephemeral) und speichert keine Site-Permissions zwischen Sessions. Wir setzen je Modus einen festen ``storage_path`` und ``private_mode=False``, damit Mikrofon-Erlaubnis etc. fuer die App-URL persistieren (wie im normalen Edge-Profil) ohne den Systembrowser zu nutzen. Getrennte Storage-Pfade verhindern, dass Haupt- und Empfang-Chat-Huelle sich beim parallelen Coldstart gegenseitig sperren ("Keine Rueckmeldung"). """ from __future__ import annotations import json import os import subprocess import sys import threading import time from pathlib import Path from urllib.parse import parse_qs, urlencode, urlparse, urlunparse def _strip_minichat_query(href: str) -> str: """URL wie Browser-Chat: Query-Parameter minichat und mode=minichat entfernen.""" raw = str(href or "").strip() if not raw: return "" try: p = urlparse(raw) qs = parse_qs(p.query, keep_blank_values=True) qs.pop("minichat", None) if "mode" in qs: modes = [str(x).lower() for x in qs.get("mode") or []] if modes == ["minichat"]: qs.pop("mode", None) return urlunparse(p._replace(query=urlencode(qs, doseq=True))) except Exception: return raw def _own_top_level_hwnds() -> list[int]: """Top-Level-HWNDs des EIGENEN Prozesses (Windows; sichtbare Fenster).""" if sys.platform != "win32": return [] try: import ctypes from ctypes import wintypes user32 = ctypes.windll.user32 my_pid = int(os.getpid()) out: list[int] = [] @ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HWND, wintypes.LPARAM) def _enum(hwnd, _lp): try: if not user32.IsWindowVisible(hwnd): return True pid = wintypes.DWORD() user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid)) if int(pid.value) == my_pid: out.append(int(hwnd)) except Exception: pass return True user32.EnumWindows(_enum, 0) return out except Exception: return [] def _apply_win32_topmost(value: bool) -> bool: """Erzwingt TOPMOST/NOTOPMOST per SetWindowPos auf alle eigenen Top-Level-Fenster. Workaround fuer pywebview/WinForms+WebView2: ``window.on_top = True`` setzt das WinForms-TopMost-Flag, greift aber nicht zuverlaessig auf das echte Top-Level-Fenster im WebView2-Host. Ein einmaliger SetWindowPos-Aufruf nach dem Setter sorgt zuverlaessig fuer das tatsaechliche Always-on-top. Kein SetForegroundWindow, kein Loop. """ if sys.platform != "win32": return False try: import ctypes user32 = ctypes.windll.user32 HWND_TOPMOST = -1 HWND_NOTOPMOST = -2 SWP_NOMOVE = 0x0002 SWP_NOSIZE = 0x0001 SWP_NOACTIVATE = 0x0010 flags = SWP_NOMOVE | SWP_NOSIZE | SWP_NOACTIVATE target = HWND_TOPMOST if value else HWND_NOTOPMOST any_ok = False for hwnd in _own_top_level_hwnds(): try: if user32.SetWindowPos(hwnd, target, 0, 0, 0, 0, flags): any_ok = True except Exception: pass return any_ok except Exception: return False def _focus_other_empfang_host_window(title_prefixes: tuple[str, ...] = ("AzA-Empfang",)) -> bool: """Bringt ein anderes AzA-Empfang-Fenster (anderer Prozess) in den Vordergrund. ``title_prefixes`` schraenkt das Match auf den eigenen Modus ein. Default ist die Desktop-Huelle ("AzA-Empfang"/"AzA-Empfang \xb7 Desktop"), damit Cross-Mode- Fokus (Chat-Huelle <-> Desktop-Huelle) NICHT passiert. """ if sys.platform != "win32": return False try: import ctypes from ctypes import wintypes user32 = ctypes.windll.user32 my_pid = int(os.getpid()) handles: list[int] = [] prefixes = tuple(p for p in title_prefixes if p) @ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HWND, wintypes.LPARAM) def _enum(hwnd, _lp): if not user32.IsWindowVisible(hwnd): return True buf = ctypes.create_unicode_buffer(260) user32.GetWindowTextW(hwnd, buf, 260) t = buf.value or "" if not any(t == p or t.startswith(p) for p in prefixes): return True pid = wintypes.DWORD() user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid)) if int(pid.value) != my_pid: handles.append(int(hwnd)) return True user32.EnumWindows(_enum, 0) if not handles: return False hwnd = wintypes.HWND(handles[0]) sw_restore = 9 user32.ShowWindow(hwnd, sw_restore) user32.SetForegroundWindow(hwnd) return True except Exception: return False def _empfang_shell_icon_path() -> str | None: """Windows/pywebview: WinForms laedt Fenstericon aus .ico neben diesem Skript. Im PyInstaller-Bundle liegt logo.ico unter sys._MEIPASS (bzw. neben sys.executable). """ candidates: list[Path] = [] try: if getattr(sys, "frozen", False): candidates.append(Path(sys.executable).resolve().parent / "logo.ico") meip = getattr(sys, "_MEIPASS", "") if meip: candidates.append(Path(meip) / "logo.ico") except Exception: pass candidates.append(Path(__file__).resolve().parent / "logo.ico") for c in candidates: try: if c.is_file(): return str(c) except Exception: continue return None # --------------------------------------------------------------------------- # Modus-Klassifizierung (Desktop-Huelle / separate Empfang-Chat-Huelle / # MiniChat-Popup) - bestimmt Titel, Storage, AppUserModelID. # --------------------------------------------------------------------------- _MODE_DESKTOP_SHELL = "desktop_shell" _MODE_EMPFANG_CHAT_SHELL = "empfang_chat_shell" _MODE_MINICHAT = "minichat" def _classify_shell_mode(start_url: str, *, parent_hint: str = "") -> str: """Bestimmt den Betriebsmodus aus der Start-URL. Eindeutige Marker (per Reihenfolge): 1. ?minichat=1 -> minichat (Parent-Modus folgt unten) 2. target=empfang_chat_shell (im /shell/launch) -> empfang_chat_shell 3. shell_source=aza_desktop / ?desktop_shell=1 -> desktop_shell 4. shell_source=empfang_chat_shell / ?empfang_chat_shell=1 -> empfang_chat_shell 5. /empfang/shell/launch?token=... (ohne target) -> desktop_shell (legacy AzA Office) 6. Default (kein Argument, Doppelklick) -> empfang_chat_shell ``parent_hint`` ist nur fuer MiniChat relevant: minichat erbt den Modus seines Parent, damit das WebView2-Profil (und damit die Login-Session) konsistent bleibt. """ u = (start_url or "").strip() low = u.lower() if "minichat=1" in low or "mode=minichat" in low: return _MODE_MINICHAT if "target=empfang_chat_shell" in low or "target%3dempfang_chat_shell" in low: return _MODE_EMPFANG_CHAT_SHELL if ( "shell_source=aza_desktop" in low or "shell_source%3daza_desktop" in low or "desktop_shell=1" in low ): return _MODE_DESKTOP_SHELL if ( "shell_source=empfang_chat_shell" in low or "shell_source%3dempfang_chat_shell" in low or "empfang_chat_shell=1" in low ): return _MODE_EMPFANG_CHAT_SHELL if "/empfang/shell/launch" in low and "token=" in low: # /shell/launch ohne target=... -> historischer AzA-Desktop-Pfad. return _MODE_DESKTOP_SHELL _ph = (parent_hint or "").strip().lower() if _ph in (_MODE_DESKTOP_SHELL, _MODE_EMPFANG_CHAT_SHELL): return _ph return _MODE_EMPFANG_CHAT_SHELL def _split_argv_for_shell_mode(argv: list[str]) -> tuple[list[str], str]: """Filtert ``--shell-mode=`` aus argv heraus. Wird vom MiniChat-Subprocess gesetzt, damit der Kind-Prozess das gleiche WebView2-Profil/AUMID wie sein Parent verwendet (Login bleibt erhalten). Andere Modi setzen das Flag nicht; sie werden aus der URL klassifiziert. """ rest: list[str] = [] mode = "" for a in argv: s = str(a or "") ls = s.strip() if ls.startswith("--shell-mode="): mode = ls.split("=", 1)[1].strip().lower() continue rest.append(s) return rest, mode def _window_title_for_mode(mode: str) -> str: if mode == _MODE_DESKTOP_SHELL: return "AzA-Empfang \u00b7 Desktop" if mode == _MODE_MINICHAT: return "AzA MiniChat" return "AzA Empfang Chat" def _aumid_for_mode(mode: str) -> str: if mode == _MODE_DESKTOP_SHELL: # Kompatibel zur bisherigen Haupt-Huelle (Taskleisten-Icon). return "ch.aza-medwork.empfang.shellwebview" if mode == _MODE_MINICHAT: return "ch.aza-medwork.empfang.minichat" return "ch.aza-medwork.empfang.chatshell" def _effective_profile_mode(mode: str, parent_mode_hint: str = "") -> str: """WebView2-Profil-Mode (Cookies/Login-Session) - MiniChat erbt Parent. Storage darf MiniChat NICHT entkoppeln, sonst geht die Login-Session verloren. Das Fenster selbst hat trotzdem eigenen Titel/AUMID/Pin (logische Trennung). """ m = (mode or "").strip().lower() if m == _MODE_MINICHAT: ph = (parent_mode_hint or "").strip().lower() if ph in (_MODE_DESKTOP_SHELL, _MODE_EMPFANG_CHAT_SHELL): return ph # Default-Parent fuer MiniChat: Chat-Huelle (Standalone-Empfang-Chat-Shell) return _MODE_EMPFANG_CHAT_SHELL return m or _MODE_DESKTOP_SHELL def _empfang_webview_storage_dir(mode: str = _MODE_DESKTOP_SHELL, parent_mode_hint: str = "") -> str: """WebView2-User-Data-Ordner pro Modus, damit Haupt- und Chat-Huelle nicht dasselbe WebView2-Profil sperren. MiniChat ERBT den Storage seines Parent-Modus, damit die Login-Session geteilt wird (Cookies, LocalStorage). - desktop_shell -> %APPDATA%\\AzA\\EmpfangWebView (kompatibel zur Hauptprogramm-Huelle) - empfang_chat_shell -> %APPDATA%\\AzA\\EmpfangChatWebView - minichat (Parent=Desktop) -> EmpfangWebView - minichat (Parent=ChatShell) -> EmpfangChatWebView """ eff = _effective_profile_mode(mode, parent_mode_hint) appdata = (os.environ.get("APPDATA") or "").strip() base = Path(appdata) / "AzA" if appdata else Path.home() / ".aza_empfang_webview" if eff == _MODE_EMPFANG_CHAT_SHELL: return str(base / "EmpfangChatWebView") if appdata else str(base.parent / ".aza_empfang_chat_webview") return str(base / "EmpfangWebView") if appdata else str(base) def _shell_pin_state_path(mode: str = _MODE_DESKTOP_SHELL, parent_mode_hint: str = "") -> Path: """Pin-State pro Fenster-Modus (MiniChat hat eigenes Pin-Verhalten), liegt aber im Profil-Storage des effektiven Parent-Modus. """ storage = _empfang_webview_storage_dir(mode, parent_mode_hint) if (mode or "").strip().lower() == _MODE_MINICHAT: return Path(storage) / "minichat_pin_on_top.json" return Path(storage) / "shell_pin_on_top.json" def _load_shell_pin_on_top(mode: str = _MODE_DESKTOP_SHELL, parent_mode_hint: str = "") -> bool: p = _shell_pin_state_path(mode, parent_mode_hint) try: if p.is_file(): data = json.loads(p.read_text(encoding="utf-8")) return bool(data.get("on_top")) except Exception: pass return False def _save_shell_pin_on_top(value: bool, mode: str = _MODE_DESKTOP_SHELL, parent_mode_hint: str = "") -> None: p = _shell_pin_state_path(mode, parent_mode_hint) try: p.parent.mkdir(parents=True, exist_ok=True) p.write_text( json.dumps({"on_top": bool(value)}, ensure_ascii=False, separators=(",", ":")), encoding="utf-8", ) except Exception: pass class EmpfangWebviewApi: """pywebview JS-API: globaler Patienten-Pinsel (Desktop), Fenster-Pin (always on top).""" def __init__(self, mode: str = _MODE_DESKTOP_SHELL, parent_mode_hint: str = "") -> None: self._window = None self._mode = (mode or _MODE_DESKTOP_SHELL).strip().lower() self._parent_mode_hint = (parent_mode_hint or "").strip().lower() self._on_top = _load_shell_pin_on_top(self._mode, self._parent_mode_hint) self._pin_lock = threading.Lock() def bind_window(self, window) -> None: self._window = window def toggle_on_top(self): """Wechselt always-on-top: persistiert sofort, fuehrt OS-Setter NICHT-BLOCKIEREND aus. Wichtig: gibt sofort den neuen Booleschen Zustand zurueck. Die WinForms-/Win32- Aufrufe (`win.on_top = X`, SetWindowPos) laufen in einem Daemon-Thread, damit der pywebview-API-Thread und damit auch der UI-Thread NICHT durch synchrone WinForms-Marshalls blockiert werden. Kein SetForegroundWindow im Pin-Pfad. """ if not self._pin_lock.acquire(blocking=False): return self._on_top try: self._on_top = not self._on_top new_val = self._on_top finally: self._pin_lock.release() _save_shell_pin_on_top(new_val, self._mode, self._parent_mode_hint) print( f"[EMPFANG_TOPMOST] stage=manual_toggle mode={self._mode} requested={new_val}", flush=True, ) win = self._window def _apply_async() -> None: err_type = "" try: if win is not None: win.on_top = new_val except Exception as exc_a: err_type = type(exc_a).__name__ try: _apply_win32_topmost(new_val) except Exception as exc_b: err_type = err_type or type(exc_b).__name__ if err_type: print( f"[EMPFANG_TOPMOST] stage=apply_error err_type={err_type}", flush=True, ) else: print( f"[EMPFANG_TOPMOST] stage=manual_toggle ok=True value={new_val}", flush=True, ) try: threading.Thread(target=_apply_async, daemon=True).start() except Exception as exc: print( f"[EMPFANG_TOPMOST] stage=manual_toggle_thread_spawn_failed err_type={type(exc).__name__}", flush=True, ) return new_val def get_on_top(self): return self._on_top def apply_saved_on_top(self) -> dict: """Pin-Wiederanwendung beim Start: NICHT-BLOCKIEREND, optional verzoegert. Frueher wurde ``win.on_top = X`` und SetWindowPos(HWND_TOPMOST) hier SYNCHRON im API-/UI-Aufrufpfad gemacht. Das konnte ``AZA_EmpfangShell.exe`` beim WebView2-Coldstart einfrieren ("Keine Rueckmeldung"), bevor die Chat-Oberflaeche geladen war. Verhalten ab jetzt: * Standard: NICHTS tun (Pin wird erst auf manuellen Klick erneut angewendet). * Optional verzoegert NEU anwenden, wenn Umgebungsvariable AZA_EMPFANG_AUTO_TOPMOST_ON_START=1 gesetzt ist. Dann erfolgt der tatsaechliche Aufruf in einem kurzen Daemon-Thread nach ~3 s. * Kein SetForegroundWindow, kein Loop, kein Spam. * Liefert sofort {"ok": True, "on_top": } zurueck. """ saved = bool(self._on_top) auto_env = (os.environ.get("AZA_EMPFANG_AUTO_TOPMOST_ON_START") or "").strip() do_auto = auto_env == "1" and saved if not do_auto: print( f"[EMPFANG_TOPMOST] stage=startup_apply_skipped saved={saved} " f"auto_env_set={int(auto_env == '1')}", flush=True, ) return {"ok": True, "on_top": saved, "applied": False} win = self._window def _delayed_apply() -> None: try: time.sleep(3.0) except Exception: pass err_type = "" try: if win is not None: win.on_top = saved except Exception as exc_a: err_type = type(exc_a).__name__ try: _apply_win32_topmost(saved) except Exception as exc_b: err_type = err_type or type(exc_b).__name__ if err_type: print( f"[EMPFANG_TOPMOST] stage=apply_error err_type={err_type}", flush=True, ) else: print( f"[EMPFANG_TOPMOST] stage=startup_apply_done value={saved}", flush=True, ) try: threading.Thread(target=_delayed_apply, daemon=True).start() print( f"[EMPFANG_TOPMOST] stage=startup_apply_scheduled saved={saved} delay_s=3", flush=True, ) except Exception as exc: print( f"[EMPFANG_TOPMOST] stage=startup_apply_thread_spawn_failed " f"err_type={type(exc).__name__}", flush=True, ) return {"ok": True, "on_top": saved, "applied": False, "scheduled": True} def open_minichat(self, href: str) -> dict: """Zweites schmales Empfang-Fenster (subprocess wie Desktop-Starter). Nicht-blockierend: Popen in Daemon-Thread. Damit hat das MiniChat-Fenster eine eigene pywebview-Instanz inkl. Pin (on_top), ohne window.open-Popup ohne API. Wichtig: Wir reichen den Parent-Modus via ``--shell-mode=`` durch. Damit teilt sich MiniChat das WebView2-Profil seines Parents (Desktop-Huelle ODER Chat-Huelle), und die Login-Session bleibt erhalten. """ url = str(href or "").strip() if not url: return {"ok": False, "reason": "empty"} try: p = urlparse(url) qs = parse_qs(p.query, keep_blank_values=True) qs["minichat"] = ["1"] full = urlunparse(p._replace(query=urlencode(qs, doseq=True))) except Exception: full = url w, h = 600, 820 # MiniChat erbt das effektive Profil seines Parents (nicht "minichat" selbst). parent_mode = _effective_profile_mode(self._mode, self._parent_mode_hint) def _spawn() -> None: try: script = Path(__file__).resolve() mode_arg = f"--shell-mode={parent_mode}" if getattr(sys, "frozen", False): cmd = [sys.executable, mode_arg, full, str(w), str(h)] cwd = str(Path(sys.executable).resolve().parent) else: cmd = [sys.executable, str(script), mode_arg, full, str(w), str(h)] cwd = str(script.parent) subprocess.Popen( cmd, cwd=cwd, close_fds=(sys.platform != "win32"), ) except Exception: pass threading.Thread(target=_spawn, daemon=True).start() return {"ok": True} def open_browser_chat(self, href: str) -> dict: """Volle Empfang-Ansicht ohne minichat: erst anderes Fenster fokussieren, sonst neuer Prozess. Cross-Mode-Fokus wird vermieden: nur Fenster mit dem gleichen Titel-Praefix wie der eigene Modus werden in den Vordergrund geholt. """ url = _strip_minichat_query(href) if not url: return {"ok": False, "reason": "empty"} w_main, h_main = 1180, 820 # Beim "Browser-Chat oeffnen" aus dem MiniChat das volle Empfang-Fenster # im effektiven Parent-Modus oeffnen, nicht erneut als MiniChat. parent_mode = _effective_profile_mode(self._mode, self._parent_mode_hint) prefixes = (_window_title_for_mode(parent_mode),) def _work() -> None: if _focus_other_empfang_host_window(prefixes): return try: script = Path(__file__).resolve() mode_arg = f"--shell-mode={parent_mode}" if getattr(sys, "frozen", False): cmd = [sys.executable, mode_arg, url, str(w_main), str(h_main)] cwd = str(Path(sys.executable).resolve().parent) else: cmd = [sys.executable, str(script), mode_arg, url, str(w_main), str(h_main)] cwd = str(script.parent) subprocess.Popen( cmd, cwd=cwd, close_fds=(sys.platform != "win32"), ) except Exception: pass threading.Thread(target=_work, daemon=True).start() return {"ok": True} @staticmethod def _pinsel_eval_js(win, js_code: str, label: str) -> None: """Lokal nur stdout/stderr; keine Clipboard-/Patient-Inhalte loggen.""" try: win.evaluate_js(js_code) print(f"[pinsel] evaluate_js called ({label})", flush=True) except Exception as exc: print( f"[pinsel] evaluate_js failed ({label}): {type(exc).__name__}: {exc}", flush=True, ) try: win.evaluate_js( "try{if(window.shellPinselDiagSet)window.shellPinselDiagSet('Pinsel: Fehler: evaluate_js_failed');}catch(_e){}" ) except Exception: pass def start_patient_nr_pick(self) -> dict: from aza_empfang_smart_pick import ( compact_and_validate_patient_nr_pick_text, start_global_patient_nr_pick, ) print("[pinsel] api called", flush=True) win = self._window print(f"[pinsel] window bound {win is not None}", flush=True) if win is None: return {"ok": False, "reason": "no_window"} def js_reset() -> None: self._pinsel_eval_js( win, "try{window.shellFinalizeGlobalPickMechanismUi&&window.shellFinalizeGlobalPickMechanismUi();}catch(_e){}", "shellFinalizeGlobalPickMechanismUi", ) def on_clipboard(raw: str) -> None: nn = compact_and_validate_patient_nr_pick_text(raw) if not nn: print("[pinsel] validated_rejected", flush=True) self._pinsel_eval_js( win, "try{window.shellReceivePatientNrPickInvalid&&window.shellReceivePatientNrPickInvalid();}catch(_e){}", "shellReceivePatientNrPickInvalid", ) return print("[pinsel] validated_ok", flush=True) self._pinsel_eval_js( win, "try{window.shellReceivePatientNrFromDesktop(%s);}catch(_e){}" % json.dumps(nn), "shellReceivePatientNrFromDesktop", ) print("[pinsel] starting_global_pick", flush=True) rc = start_global_patient_nr_pick( on_clipboard_pick=on_clipboard, reset_pick_ui=js_reset, owner_tk=None, schedule_ui=None, diagnostics=True, ) print(f"[pinsel] start_global_patient_nr_pick rc={rc}", flush=True) if rc == "cancelled": return {"ok": True, "cancelled": True} if rc == "unavailable": self._pinsel_eval_js( win, "try{window.shellGlobalPinselResetUi&&window.shellGlobalPinselResetUi();}catch(_e){}", "shellGlobalPinselResetUi(unavailable)", ) return {"ok": False, "reason": "no_input_hook"} return {"ok": True} def cancel_patient_nr_pick(self) -> dict: from aza_empfang_smart_pick import stop_global_patient_nr_pick print("[pinsel] cancel_patient_nr_pick called", flush=True) stop_global_patient_nr_pick() return {"ok": True} def start_region_snapshot_capture(self) -> dict: """Startet nativen Bereichs-Screenshot (Tk-Overlay + PIL) fuer die WebView-Huelle. Rueckgabe an JS erfolgt asynchron per evaluate_js -> shellReceiveSnapshotFromDesktop. Keine Bilddaten in Logs. """ from aza_empfang_region_snapshot import start_region_snapshot_capture as _region_snap_start win = self._window if win is None: return {"ok": False, "reason": "no_window"} if sys.platform != "win32": return {"ok": False, "reason": "unsupported_os"} def on_done(data_url: str | None) -> None: if not data_url: self._pinsel_eval_js( win, "try{if(window.shellReceiveSnapshotFromDesktopCancel)" "window.shellReceiveSnapshotFromDesktopCancel();}catch(_e){}", "snapshot_cancel", ) return self._pinsel_eval_js( win, "try{if(window.shellReceiveSnapshotFromDesktop)" "window.shellReceiveSnapshotFromDesktop(%s);}catch(_e){}" % json.dumps(data_url), "shellReceiveSnapshotFromDesktop", ) rc = _region_snap_start(on_done) if rc == "busy": return {"ok": False, "reason": "busy"} return {"ok": True} def play_notification_sound(self, kind: str = "") -> dict: """Spielt einen kurzen Systemton ab (Windows: winsound.MessageBeep). Native, non-blocking, unabhaengig von Browser-Autoplay-Policies und Hintergrundtab-Drosselung. Keine Patientendaten/Chatdaten/Tokens werden geloggt oder verarbeitet. ``kind`` ist nur ein Hinweis (z.B. "incoming"). """ def _do() -> None: if sys.platform != "win32": return try: import winsound # noqa: WPS433 (lazy) # MB_OK ist nicht zu laut, gut hoerbar; SND_ASYNC blockiert nicht. winsound.MessageBeep(winsound.MB_OK) except Exception as exc: print( f"[empfang] play_notification_sound failed: {type(exc).__name__}", flush=True, ) try: threading.Thread(target=_do, daemon=True).start() except Exception: pass return {"ok": True, "kind": str(kind or "")[:32]} def bring_to_front(self) -> dict: """JS-API: Empfang-Huelle in den Vordergrund holen. Nicht-blockierend. Wichtig: Diese Methode wird aus dem JS-Polling/Popup-Pfad aufgerufen und MUSS sofort zurueckkehren. ``win.restore()`` und Win32-Aufrufe werden in einen Daemon-Thread ausgelagert, damit der WebView-Worker nicht haengt ("AzA-Empfang reagiert nicht"). Sucht nur nach Fenstern, deren Titel zum eigenen Modus passt - keine Cross-Mode-Fokussierung (Desktop-Huelle <-> Chat-Huelle). Keine Chat-/Patientendaten werden geloggt oder verarbeitet. """ win = self._window own_title = _window_title_for_mode(self._mode) def _do() -> None: try: if win is not None: try: win.restore() except Exception: pass except Exception: pass if sys.platform != "win32": return try: import ctypes # noqa: WPS433 user32 = ctypes.windll.user32 handles: list[int] = [] my_pid = int(os.getpid()) @ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, ctypes.c_void_p) def _enum(hwnd, _lp): if not user32.IsWindowVisible(hwnd): return True buf = ctypes.create_unicode_buffer(260) user32.GetWindowTextW(hwnd, buf, 260) t = buf.value or "" if not (t == own_title or t.startswith(own_title)): return True pid = ctypes.c_ulong() user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid)) if int(pid.value) == my_pid and int(hwnd): handles.append(int(hwnd)) return True user32.EnumWindows(_enum, 0) hwnd = handles[0] if handles else 0 if not hwnd: return sw_restore = 9 user32.ShowWindow(hwnd, sw_restore) user32.SetForegroundWindow(hwnd) except Exception as exc: print( f"[empfang] bring_to_front failed: {type(exc).__name__}", flush=True, ) threading.Thread(target=_do, daemon=True).start() return {"ok": True, "scheduled": True} def _append_query_marker(url: str, key: str, value: str) -> str: """Setzt/ersetzt einen Query-Parameter, ohne andere Parameter zu zerstoeren.""" try: p = urlparse(url) qs = parse_qs(p.query, keep_blank_values=True) qs[key] = [value] return urlunparse(p._replace(query=urlencode(qs, doseq=True))) except Exception: sep = "&" if "?" in url else "?" return f"{url}{sep}{key}={value}" _EMPFANG_CHAT_SHELL_DEFAULT_BASE = "https://empfang.aza-medwork.ch/empfang/" def _build_default_empfang_chat_shell_url() -> str: """Standard-URL der separaten Empfang-Chat-Huelle (kein Arzt-Desktop). Setzt BEIDE Marker fuer empfang.html (data-empfang-chat-shell + Login-/Handoff-UI): * empfang_chat_shell=1 * shell_source=empfang_chat_shell Optional ueberschreibbar per Umgebungsvariable ``AZA_EMPFANG_CHAT_SHELL_URL`` (z.B. Staging). Es wird absichtlich NICHT aus aza_empfang_app._empfang_url() / backend_url.txt gelesen, damit die Chat-Huelle unabhaengig von einer optionalen Datei neben der EXE arbeitet und keine alte/abweichende Server-URL erbt. """ base = (os.environ.get("AZA_EMPFANG_CHAT_SHELL_URL") or "").strip() if not base: base = _EMPFANG_CHAT_SHELL_DEFAULT_BASE # Sicherstellen, dass /empfang/ Pfad enthalten ist try: p = urlparse(base) if not (p.path or "").rstrip("/").endswith("/empfang"): path = (p.path or "").rstrip("/") new_path = (path + "/empfang/") if not path.endswith("/empfang/") else path base = urlunparse(p._replace(path=new_path or "/empfang/")) except Exception: pass url = _append_query_marker(base, "empfang_chat_shell", "1") url = _append_query_marker(url, "shell_source", "empfang_chat_shell") return url def _normalize_handoff_code(raw: str) -> str: s = (raw or "").strip().upper().replace(" ", "") for ch in ("\u2011", "\u2013", "\u2014", "\u2212", "_"): s = s.replace(ch, "-") return s def _resolve_handoff_code_to_launch_url(base_url: str, code: str) -> str | None: """Loest XXXX-XXXX serverseitig in /empfang/shell/launch?token=... auf. Sicherheit: Code IST das Geheimnis; einmaliger Verbrauch durch Server. """ try: from urllib.parse import quote from urllib.request import urlopen, Request as _Req except Exception: return None try: p = urlparse(base_url) origin = f"{p.scheme}://{p.netloc}" if p.scheme and p.netloc else None if not origin: return None c = _normalize_handoff_code(code) if not c: return None url = f"{origin}/empfang/handoff/lookup?code={quote(c, safe='')}" req = _Req(url, method="GET", headers={ "User-Agent": "AzA-EmpfangShell/handoff", "Accept": "application/json", }) import ssl try: ctx = ssl.create_default_context() with urlopen(req, timeout=10, context=ctx) as r: raw = r.read(8192) except ssl.SSLError: ctx = ssl._create_unverified_context() with urlopen(req, timeout=10, context=ctx) as r: raw = r.read(8192) data = json.loads(raw.decode("utf-8", errors="replace")) if not isinstance(data, dict) or not data.get("success"): return None path = str(data.get("launch_path") or "").strip() if not path or not path.startswith("/"): return None return f"{origin}{path}" except Exception as exc: print( f"[handoff] lookup failed: {type(exc).__name__}", file=sys.stderr, flush=True, ) return None def _split_argv_for_handoff(argv: list[str]) -> tuple[list[str], str, str]: """Filtert --handoff-token=... und --handoff-code=... aus argv heraus. Rueckgabe: (uebriges_argv, handoff_token, handoff_code). """ rest: list[str] = [] tok = "" code = "" for a in argv: s = str(a or "") ls = s.strip() if ls.startswith("--handoff-token="): tok = ls.split("=", 1)[1].strip() continue if ls.startswith("--handoff-code="): code = ls.split("=", 1)[1].strip() continue rest.append(s) return rest, tok, code def main(argv: list[str] | None = None) -> int: argv = list(argv if argv is not None else sys.argv[1:]) argv, handoff_token, handoff_code = _split_argv_for_handoff(argv) argv, parent_mode_hint = _split_argv_for_shell_mode(argv) standalone_default = False if not argv or not str(argv[0]).strip(): # Standalone / Doppelklick: separate Empfang-Chat-Huelle. # Beide Marker (empfang_chat_shell=1 & shell_source=empfang_chat_shell) # werden gesetzt, damit empfang.html die Chat-Shell-UI aktiviert. url = _build_default_empfang_chat_shell_url() argv = [url] standalone_default = True url = str(argv[0]).strip() if handoff_token and "/empfang/shell/launch" not in url: # Direkter Token-Start (z.B. aus AzA-Desktop): bevorzugt verwenden. try: from urllib.parse import quote as _q p = urlparse(url) origin = f"{p.scheme}://{p.netloc}" if p.scheme and p.netloc else "" if origin: url = ( f"{origin}/empfang/shell/launch?token={_q(handoff_token, safe='')}" f"&target=empfang_chat_shell" ) except Exception: pass elif handoff_code: # Verbindungscode (XXXX-XXXX) per Lookup in einen einmaligen Launch-Token tauschen. resolved = _resolve_handoff_code_to_launch_url(url, handoff_code) if resolved: url = resolved else: print( "[handoff] Verbindungscode konnte nicht eingeloest werden — starte Login-Seite.", file=sys.stderr, flush=True, ) # Modus zuerst klassifizieren, damit wir mode-abhaengige Default-Groessen # vergeben koennen (Empfang-Chat-Huelle startet kompakter als Desktop-Huelle). mode = _classify_shell_mode(url, parent_hint=parent_mode_hint) if standalone_default and mode == _MODE_DESKTOP_SHELL: # Default-Doppelklick darf nicht versehentlich als Desktop-Huelle starten. mode = _MODE_EMPFANG_CHAT_SHELL # Default-Groessen pro Modus: # desktop_shell -> 1180 x 820 (unveraendert) # empfang_chat_shell -> 900 x 650 (~70% Flaeche; Chat/Eingabe weiter brauchbar) # minichat -> 600 x 820 (wie bisher; nur ueber subprocess gesetzt) if mode == _MODE_EMPFANG_CHAT_SHELL: default_w, default_h = 900, 650 elif mode == _MODE_MINICHAT: default_w, default_h = 600, 820 else: default_w, default_h = 1180, 820 # Explizit per argv uebergebene Groesse (z.B. open_minichat) hat Vorrang. w = int(argv[1]) if len(argv) > 1 and str(argv[1]).isdigit() else default_w h = int(argv[2]) if len(argv) > 2 and str(argv[2]).isdigit() else default_h # AppUserModelID je Modus (vor webview.create_window setzen, damit Taskleisten- # Gruppierung sauber bleibt). Idempotent: spaeter im __main__ wieder gesetzt. if sys.platform == "win32": try: import ctypes # noqa: WPS433 ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID( _aumid_for_mode(mode), ) except Exception: pass storage_path = _empfang_webview_storage_dir(mode, parent_mode_hint) try: Path(storage_path).mkdir(parents=True, exist_ok=True) except OSError as exc: print( f"WebView-Profil-Ordner konnte nicht angelegt werden ({storage_path}): {exc}", file=sys.stderr, ) return 13 try: import webview # noqa: WPS433 (runtime dependency) except ImportError: print( "pywebview fehlt. Bitte installieren:\n" " pip install pywebview>=5", file=sys.stderr, ) return 11 window_title = _window_title_for_mode(mode) print( f"[EMPFANG_SHELL] start mode={mode} parent_hint={parent_mode_hint or '-'} " f"title={window_title!r} storage_tail={Path(storage_path).name}", flush=True, ) api = EmpfangWebviewApi(mode=mode, parent_mode_hint=parent_mode_hint) start_kw = {"storage_path": storage_path, "private_mode": False} _ico = _empfang_shell_icon_path() if _ico: start_kw["icon"] = _ico try: win = webview.create_window( window_title, url, width=w, height=h, js_api=api, ) api.bind_window(win) webview.start(**start_kw) return 0 except Exception as exc: print(str(exc), file=sys.stderr) return 12 if __name__ == "__main__": # AUMID wird zusaetzlich in main() pro Modus gesetzt. Hier ein konservativer # Default-Wert, damit ein eventueller Crash vor main() korrekt gruppiert wird. if sys.platform == "win32": try: import ctypes ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID( "ch.aza-medwork.empfang.shellwebview", ) except Exception: pass raise SystemExit(main())