302 lines
10 KiB
Python
302 lines
10 KiB
Python
# -*- 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 sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
|
|
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()
|
|
self._pin_initial_applied = False
|
|
|
|
def bind_window(self, window) -> None:
|
|
self._window = window
|
|
self._schedule_initial_pin_apply()
|
|
|
|
def _schedule_initial_pin_apply(self) -> None:
|
|
"""Persistierten Pin-Zustand EINMAL anwenden, ohne UI-Thread zu blockieren.
|
|
|
|
Wir haengen uns NICHT an ``window.events.loaded`` (das wuerde bei jedem
|
|
Reload/Navigation erneut feuern und kann unter pywebview/Windows den
|
|
UI-Thread blockieren -> "Keine Rueckmeldung"). Statt dessen einmaliges,
|
|
kurz verzoegertes Setzen aus einem Daemon-Thread.
|
|
"""
|
|
|
|
def _apply_once() -> None:
|
|
if self._pin_initial_applied:
|
|
return
|
|
self._pin_initial_applied = True
|
|
time.sleep(0.6)
|
|
try:
|
|
if self._window is not None:
|
|
self._window.on_top = self._on_top
|
|
except Exception:
|
|
pass
|
|
|
|
threading.Thread(target=_apply_once, daemon=True).start()
|
|
|
|
def toggle_on_top(self):
|
|
"""Wechselt always-on-top (nur dieses Empfang-Fenster). Wie aza_empfang_app._Api."""
|
|
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)
|
|
|
|
def _apply() -> None:
|
|
try:
|
|
win = self._window
|
|
if win is not None:
|
|
win.on_top = new_val
|
|
except Exception:
|
|
pass
|
|
|
|
threading.Thread(target=_apply, daemon=True).start()
|
|
return new_val
|
|
|
|
def get_on_top(self):
|
|
return self._on_top
|
|
|
|
@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 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
|
|
hwnd = user32.FindWindowW(None, "AzA-Empfang")
|
|
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 main(argv: list[str] | None = None) -> int:
|
|
argv = argv if argv is not None else sys.argv[1:]
|
|
if not argv or not argv[0].strip():
|
|
print(
|
|
"Usage:\n"
|
|
' python aza_empfang_webview.py "https://host/empfang/shell/launch?token=..."',
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
url = argv[0].strip()
|
|
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
|
|
|
|
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("AzA-Empfang", 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__":
|
|
raise SystemExit(main())
|