# -*- coding: utf-8 -*- """ AzA Empfang Web-Huelle: eigener Desktop-Prozess (pywebview). Wird vom Desktop per subprocess gestartet, damit keine GUI-Kollision mit Tkinter entsteht. Argument: erste Start-URL (z.B. GET /empfang/shell/launch?token=...). WebView2-Profil: pywebview nutzt standardmaessig ``private_mode=True`` (ephemeral) und speichert keine Site-Permissions zwischen Sessions. Für die Huelle setzen wir einen festen ``storage_path`` unter %APPDATA%\\AzA\\EmpfangWebView und ``private_mode=False``, damit z. B. Mikrofon-Erlaubnis für die App-URL persistiert (wie im normalen Edge-Profil), ohne den Systembrowser zu verwenden. """ 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() -> bool: """Bringt ein anderes AzA-Empfang-Fenster (anderer Prozess) in den Vordergrund.""" 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] = [] @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) if not (buf.value == "AzA-Empfang" or buf.value.startswith("AzA-Empfang")): 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.""" p = Path(__file__).resolve().parent / "logo.ico" return str(p) if p.is_file() else None def _empfang_webview_storage_dir() -> str: """Eigener User-Data-Ordner, isoliert vom normalen Browser.""" appdata = (os.environ.get("APPDATA") or "").strip() if appdata: return str(Path(appdata) / "AzA" / "EmpfangWebView") return str(Path.home() / ".aza_empfang_webview") def _shell_pin_state_path() -> Path: return Path(_empfang_webview_storage_dir()) / "shell_pin_on_top.json" def _load_shell_pin_on_top() -> bool: p = _shell_pin_state_path() 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) -> None: p = _shell_pin_state_path() 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) -> None: self._window = None self._on_top = _load_shell_pin_on_top() 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) print( f"[EMPFANG_TOPMOST] stage=manual_toggle 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. """ 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 def _spawn() -> None: try: script = Path(__file__).resolve() if getattr(sys, "frozen", False): cmd = [sys.executable, full, str(w), str(h)] cwd = str(Path(sys.executable).resolve().parent) else: cmd = [sys.executable, str(script), 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.""" url = _strip_minichat_query(href) if not url: return {"ok": False, "reason": "empty"} w_main, h_main = 1180, 820 def _work() -> None: if _focus_other_empfang_host_window(): return try: script = Path(__file__).resolve() if getattr(sys, "frozen", False): cmd = [sys.executable, url, str(w_main), str(h_main)] cwd = str(Path(sys.executable).resolve().parent) else: cmd = [sys.executable, str(script), 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 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"). Keine Chat-/Patientendaten werden geloggt oder verarbeitet. """ win = self._window 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 == "AzA-Empfang" or t.startswith("AzA-Empfang")): 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}" def _build_default_empfang_chat_shell_url() -> str: """Standard-URL der separaten Empfang-Chat-Huelle (kein Arzt-Desktop): nutzt aza_empfang_app._empfang_url() bzw. den oeffentlichen Fallback und setzt den Marker ?empfang_chat_shell=1 fuer empfang.html (Login-/Handoff-UI). """ try: from aza_empfang_app import _empfang_url # noqa: WPS433 (lazy) base = _empfang_url() except Exception: base = "https://empfang.aza-medwork.ch/empfang/" return _append_query_marker(base, "empfang_chat_shell", "1") 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) if not argv or not str(argv[0]).strip(): # Standalone / Doppelklick: separate Empfang-Chat-Huelle. Marker fuer empfang.html setzen. url = _build_default_empfang_chat_shell_url() argv = [url] 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, ) w = int(argv[1]) if len(argv) > 1 and str(argv[1]).isdigit() else 1180 h = int(argv[2]) if len(argv) > 2 and str(argv[2]).isdigit() else 820 storage_path = _empfang_webview_storage_dir() 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 def _window_title_for_start_url(start_url: str) -> str: """Titel nach Start-URL: Arzt-Desktop-Shell vs. separate Chat-Huelle vs. manueller Test.""" u = (start_url or "") low = u.lower() if "/empfang/shell/launch" in u and "token=" in low: if "target=empfang_chat_shell" in low or "target%3Dempfang_chat_shell" in low: return "AzA-Empfang \u00b7 Chat-H\u00fclle" return "AzA-Empfang \u00b7 Desktop" if "shell_source=aza_desktop" in u or "shell_source%3Daza_desktop" in low: return "AzA-Empfang \u00b7 Desktop" return "AzA-Empfang" api = EmpfangWebviewApi() 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_for_start_url(url), 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__": if sys.platform == "win32": try: import ctypes ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID( "ch.aza-medwork.empfang.shellwebview", ) except Exception: pass raise SystemExit(main())