"""Application composition root for the doctor workstation.""" from __future__ import annotations import logging import os import sys import time from contextlib import suppress from typing import Any from PySide6.QtCore import QLibraryInfo, QLocale, QObject, Qt, QTimer, QTranslator from PySide6.QtGui import QGuiApplication, QIcon from PySide6.QtWidgets import ( QApplication, QDialog, QFrame, QHBoxLayout, QLabel, QPushButton, QVBoxLayout, QWidget, ) from doctor_workstation.config import AppConfig from doctor_workstation.core import Session from doctor_workstation.core.errors import AuthenticationExpiredError from doctor_workstation.logging_setup import configure_logging from doctor_workstation.resources import resource_path, video_dist_path from doctor_workstation.services import ( DemoDoctorRepository, RemoteDoctorRepository, TokenStore, build_repository, ) from doctor_workstation.ui import LoginWindow, ShellWindow, apply_theme from doctor_workstation.ui.widgets import ( friendly_error, run_async, set_authentication_expired_handler, show_toast, ) from doctor_workstation.video import BackendMode, launch_video_call from doctor_workstation.video.window import WEBENGINE_AVAILABLE LOGGER = logging.getLogger(__name__) class _ChineseQtTranslator(QTranslator): """Guarantee Chinese labels for common Qt standard buttons. Qt's packaged ``qtbase_zh_CN`` catalog remains the primary source for framework text. This small fallback also keeps release builds localized when a packager omits the optional ``.qm`` files. """ _BUTTON_TEXT = { "OK": "确定", "Open": "打开", "Save": "保存", "Save All": "全部保存", "Cancel": "取消", "Close": "关闭", "Yes": "是", "Yes to All": "全部确认", "No": "否", "No to All": "全部否定", "Abort": "中止", "Retry": "重试", "Ignore": "忽略", "Discard": "放弃", "Help": "帮助", "Apply": "应用", "Reset": "重置", "Restore Defaults": "恢复默认设置", "Don't Save": "不保存", } def translate( self, context: str, source_text: str, disambiguation: str | None = None, n: int = -1, ) -> str | None: del context, disambiguation, n # Returning an empty string tells Qt that an unknown source string has # a valid, deliberately empty translation. That also erased internal # values such as QPageSize's "A4", producing zero-sized PDF pages. # ``None`` delegates unknown text to Qt's installed catalog/source. return self._BUTTON_TEXT.get(source_text.replace("&", "")) def _install_chinese_translations(application: QApplication) -> None: """Install Simplified Chinese Qt catalogs once for the whole process.""" if getattr(application, "_doctor_workstation_chinese_translators", None): return QLocale.setDefault(QLocale("zh_CN")) translators: list[QTranslator] = [] translations_path = QLibraryInfo.path(QLibraryInfo.LibraryPath.TranslationsPath) for catalog in ("qt_zh_CN", "qtbase_zh_CN"): translator = QTranslator(application) if translator.load(catalog, translations_path): application.installTranslator(translator) translators.append(translator) fallback = _ChineseQtTranslator(application) application.installTranslator(fallback) translators.append(fallback) application._doctor_workstation_chinese_translators = translators # type: ignore[attr-defined] class _UnconfiguredRepository: """Login boundary used until an administrator supplies a backend URL.""" def login( self, account: str, password: str, *, remember_account: bool = False, ) -> Session: del account, password, remember_account raise ValueError("请先展开“服务器设置”，填写管理员提供的 HTTPS 接口地址。") def get_current_user(self) -> None: return None class DemoVideoDialog(QDialog): """Non-network video-room preview used only by the explicit demo mode.""" def __init__(self, patient_name: str, parent: QWidget | None = None) -> None: super().__init__(parent) self._seconds = 0 self.setWindowTitle("视频面诊 · 演示模式") self.setMinimumSize(760, 520) self.resize(980, 660) self.setModal(False) self.setStyleSheet( "QDialog{background:#F7F9FE;color:#111F46;}" "QLabel{color:#111F46;}" "QFrame#RemoteStage{background:#0E1421;border:1px solid #29334F;border-radius:16px;}" "QFrame#RemoteStage QLabel{color:#F7F9FE;}" "QFrame#LocalStage{background:#151D31;border:1px solid #3F4E75;border-radius:12px;}" "QPushButton{min-width:96px;min-height:40px;padding:0 16px;border-radius:9px;" "background:#FFFFFF;color:#3F4E75;border:1px solid #E6EAF5;font-weight:600;}" "QPushButton:hover{color:#4451E2;background:#F0F2FF;border-color:#5761F4;}" "QPushButton:checked{color:#FFFFFF;background:#5761F4;border-color:#5761F4;}" "QPushButton#Hangup{color:#FFFFFF;background:#F15B67;border-color:#F15B67;}" "QPushButton#Hangup:hover{background:#D94857;border-color:#D94857;}" ) root = QVBoxLayout(self) root.setContentsMargins(22, 18, 22, 22) root.setSpacing(14) header = QHBoxLayout() title = QLabel(f"与 {patient_name or '患者'} 的视频面诊") title.setStyleSheet("font-size:18px;font-weight:700;") header.addWidget(title) header.addStretch(1) demo = QLabel("● 演示模式 · 未连接腾讯云") demo.setStyleSheet("color:#7886AA;font-size:12px;") header.addWidget(demo) self.duration_label = QLabel("00:00") self.duration_label.setStyleSheet("font-weight:700;") header.addWidget(self.duration_label) root.addLayout(header) stage = QFrame() stage.setObjectName("RemoteStage") stage_layout = QVBoxLayout(stage) stage_layout.setContentsMargins(22, 22, 22, 22) stage_layout.addStretch(1) avatar = QLabel((patient_name or "患")[:1]) avatar.setAlignment(Qt.AlignmentFlag.AlignCenter) avatar.setFixedSize(104, 104) avatar.setStyleSheet( "background:#F0F2FF;color:#5761F4;border-radius:52px;font-size:42px;font-weight:700;" ) stage_layout.addWidget(avatar, 0, Qt.AlignmentFlag.AlignHCenter) waiting = QLabel("等待患者接听…") waiting.setAlignment(Qt.AlignmentFlag.AlignCenter) waiting.setStyleSheet("font-size:17px;font-weight:600;") stage_layout.addWidget(waiting) hint = QLabel("生产模式将通过后端短时 UserSig 初始化腾讯 TUICallKit") hint.setAlignment(Qt.AlignmentFlag.AlignCenter) hint.setStyleSheet("color:#A4ADC3;font-size:12px;") stage_layout.addWidget(hint) stage_layout.addStretch(1) local = QFrame(stage) local.setObjectName("LocalStage") local.setGeometry(24, 24, 178, 112) local_layout = QVBoxLayout(local) local_label = QLabel("医生画面") local_label.setAlignment(Qt.AlignmentFlag.AlignCenter) local_label.setStyleSheet("color:#D9E0F2;font-weight:600;") local_layout.addWidget(local_label) root.addWidget(stage, 1) controls = QHBoxLayout() controls.addStretch(1) self.mic_button = QPushButton("麦克风 开") self.mic_button.setCheckable(True) self.mic_button.toggled.connect( lambda muted: self.mic_button.setText("麦克风 关" if muted else "麦克风 开") ) controls.addWidget(self.mic_button) self.camera_button = QPushButton("摄像头 开") self.camera_button.setCheckable(True) self.camera_button.toggled.connect( lambda off: self.camera_button.setText("摄像头 关" if off else "摄像头 开") ) controls.addWidget(self.camera_button) hangup = QPushButton("结束面诊") hangup.setObjectName("Hangup") hangup.clicked.connect(self.close) controls.addWidget(hangup) controls.addStretch(1) root.addLayout(controls) self._timer = QTimer(self) self._timer.timeout.connect(self._tick) self._timer.start(1000) def _tick(self) -> None: self._seconds += 1 minutes, seconds = divmod(self._seconds, 60) self.duration_label.setText(f"{minutes:02d}:{seconds:02d}") class ApplicationController(QObject): """Own windows, repositories and the authenticated application lifecycle.""" def __init__(self, application: QApplication, config: AppConfig) -> None: super().__init__() self.application = application self.config = config self.token_store = TokenStore(config.config_dir / "credentials.json") self.demo_repository = DemoDoctorRepository() self.remote_repository: RemoteDoctorRepository | None = None self.login_window: LoginWindow | None = None self.shell_window: ShellWindow | None = None self.current_repository: Any = None self.current_demo_mode = config.demo_mode self.video_calls: dict[str, Any] = {} self.video_pending: dict[str, object] = {} self.demo_video_dialogs: dict[str, DemoVideoDialog] = {} self._restore_generation = 0 self._restore_in_progress = False self._restore_worker: Any = None self._shutting_down = False self._authentication_expiry_in_progress = False self._rebuild_remote_repository() set_authentication_expired_handler(self._on_authentication_expired) application.aboutToQuit.connect(self.shutdown) def start(self) -> None: """Show login immediately, then validate a production token off-thread.""" self._show_login() self._begin_session_restore() def _base_repository(self) -> Any: return self.remote_repository or _UnconfiguredRepository() def _show_login(self) -> None: if self.login_window is None: self.login_window = LoginWindow( self._base_repository(), self.config, self.demo_repository, credential_store=self.token_store, ) self.login_window.login_succeeded.connect(self._on_login_succeeded) self.login_window.config_changed.connect(self._on_config_changed) self.login_window.demo_mode_changed.connect(self._on_demo_mode_changed) self._apply_window_icon(self.login_window) else: self.login_window.repository = self._base_repository() self.login_window.config = self.config if not self.login_window.demo_check.isChecked(): self.login_window.active_repository = self._base_repository() self.login_window.restore_remembered_credentials() self.login_window.show() self.login_window.raise_() self.login_window.activateWindow() def _on_config_changed(self, payload: Any) -> None: previous_connection = ( self.config.api_base_url, self.config.request_timeout, self.config.verify_ssl, ) if isinstance(payload, AppConfig): updated = payload elif isinstance(payload, dict): aliases = { "base_url": "api_base_url", "read_timeout": "request_timeout", } changes = {aliases.get(key, key): value for key, value in payload.items()} allowed = { "api_base_url", "demo_mode", "video_mode", "video_web_url", "verify_ssl", "request_timeout", "log_level", "remembered_account", } updated = self.config.with_updates( **{key: value for key, value in changes.items() if key in allowed} ) else: return self.config = updated with suppress(OSError): self.config.save_preferences() next_connection = ( self.config.api_base_url, self.config.request_timeout, self.config.verify_ssl, ) if next_connection != previous_connection: self._cancel_session_restore() self._rebuild_remote_repository() if self.login_window is not None: self.login_window.repository = self._base_repository() if not self.login_window.demo_check.isChecked(): self.login_window.active_repository = self._base_repository() if self.login_window is not None: self.login_window.config = self.config def _on_demo_mode_changed(self, enabled: bool) -> None: self.current_demo_mode = enabled if enabled: self._cancel_session_restore() def _rebuild_remote_repository(self) -> None: old = self.remote_repository self.remote_repository = None if self.config.api_base_url: repository = build_repository( demo=False, base_url=self.config.api_base_url, token_store=self.token_store, timeout=self.config.request_timeout, verify=self.config.verify_ssl, ) if isinstance(repository, RemoteDoctorRepository): self.remote_repository = repository if old is not None and old is not self.current_repository: with suppress(Exception): old.client.close() def _begin_session_restore(self) -> None: """Validate a scoped persisted token without racing manual login.""" repository = self.remote_repository if ( self._shutting_down or self.config.demo_mode or repository is None or self.current_repository is not None ): return self._restore_generation += 1 generation = self._restore_generation self._restore_in_progress = True if self.login_window is not None: self.login_window.set_session_restore_pending(True) self._restore_worker = run_async( repository.restore_session, on_success=lambda session: self._on_restore_success( session, repository, generation, ), on_error=lambda error: self._on_restore_error( error, repository, generation, ), on_finished=lambda: self._on_restore_finished(repository, generation), ) def _restore_is_current( self, repository: RemoteDoctorRepository, generation: int, ) -> bool: """Return whether a restore callback still owns the login surface.""" return ( not self._shutting_down and self._restore_in_progress and generation == self._restore_generation and repository is self.remote_repository ) def _finish_session_restore( self, repository: RemoteDoctorRepository, generation: int, ) -> bool: """Release login controls only for the currently active restore.""" if not self._restore_is_current(repository, generation): return False self._restore_in_progress = False self._restore_worker = None if self.login_window is not None: self.login_window.set_session_restore_pending(False) return True def _on_restore_success( self, session: object, repository: RemoteDoctorRepository, generation: int, ) -> None: """Enter the shell only for a current, fully validated session.""" if not self._finish_session_restore(repository, generation): return if session is None: return if not isinstance(session, Session) or not session.authenticated: with suppress(Exception): repository.logout() self._login_guard_error("已保存的登录状态无效，请重新登录。") return self._on_login_succeeded( { "session": session, "user": session.user, "repository": repository, "demo_mode": False, "restored_session": True, } ) def _on_restore_error( self, error: Exception, repository: RemoteDoctorRepository, generation: int, ) -> None: """Return control to login after a current restore attempt fails.""" if not self._finish_session_restore(repository, generation): return if isinstance(error, AuthenticationExpiredError): message = "登录状态已失效，请重新登录。" else: message = f"自动恢复登录失败：{friendly_error(error)}" self._login_guard_error(message) def _on_restore_finished( self, repository: RemoteDoctorRepository, generation: int, ) -> None: """Release controls when a worker finishes without a result callback.""" self._finish_session_restore(repository, generation) def _cancel_session_restore(self) -> None: """Invalidate late callbacks and re-enable manual login controls.""" self._restore_generation += 1 if not self._restore_in_progress: return self._restore_in_progress = False self._restore_worker = None if self.login_window is not None: self.login_window.set_session_restore_pending(False) def _on_login_succeeded(self, payload: dict[str, Any]) -> None: self._cancel_session_restore() session = payload.get("session") repository = payload.get("repository") if not isinstance(session, Session) or repository is None: self._login_guard_error("登录响应不完整，请重试。") return if session.password_change_required: with suppress(Exception): repository.logout() self._login_guard_error("该账号需要先修改初始密码，请在管理后台完成后重新登录。") return if session.work_wechat_binding_required: with suppress(Exception): repository.logout() self._login_guard_error("该账号需要先绑定企业微信，请在管理后台完成绑定后重新登录。") return demo_mode = bool(payload.get("demo_mode")) if not demo_mode and not session.menu: with suppress(Exception): repository.logout() self._login_guard_error("当前账号没有可用医生端菜单，请联系管理员授权。") return self.current_repository = repository self._authentication_expiry_in_progress = False self.current_demo_mode = demo_mode self.shell_window = ShellWindow(repository, payload, session.permissions) self.shell_window.logout_requested.connect(self._logout) self.shell_window.video_requested.connect(self._request_video) self._apply_window_icon(self.shell_window) if self.login_window is not None: self.login_window.hide() self.shell_window.show() self.shell_window.raise_() self.shell_window.activateWindow() def _login_guard_error(self, message: str) -> None: if self.login_window is not None: self.login_window.error_banner.show_message(message, "warning") self.login_window.show() def _on_authentication_expired(self, error: AuthenticationExpiredError) -> bool: """Consume an active-shell expiry and atomically return to login.""" if self.shell_window is None or self.current_repository is None: return False if self._authentication_expiry_in_progress: return True self._authentication_expiry_in_progress = True LOGGER.info( "authenticated session expired", extra={"api_code": error.code, "request_id": error.request_id}, ) self._logout(message="登录状态已失效，请重新登录。") return True def _logout(self, *, message: str = "") -> None: """Clear authenticated resources and return to the login window.""" calls = tuple(self.video_calls.values()) for call in calls: with suppress(Exception): call.close() self._wait_for_video_lifecycle(calls, timeout=1.25) self.video_calls.clear() self.video_pending.clear() for dialog in self.demo_video_dialogs.values(): dialog.close() self.demo_video_dialogs.clear() if self.current_repository is not None: with suppress(Exception): self.current_repository.logout() self.current_repository = None if self.shell_window is not None: self.shell_window.close() self.shell_window.deleteLater() self.shell_window = None self._show_login() if message and self.login_window is not None: self.login_window.error_banner.show_message(message, "warning") def _request_video(self, payload: dict[str, Any]) -> None: parent = self.shell_window if parent is None or self.current_repository is None: return patient_id = payload.get("patient_id") diagnosis_id = payload.get("diagnosis_id") patient_name = str(payload.get("patient_name") or "患者") open_im = str(payload.get("mode") or "video").lower() == "im" if patient_id in (None, "") or diagnosis_id in (None, ""): show_toast(parent, "患者或诊单信息不完整，无法发起视频。", "danger", 4200) return call_key = str(diagnosis_id) existing_call = self.video_calls.get(call_key) if open_im and existing_call is not None and getattr(existing_call, "open_im", False): qt_window = getattr(existing_call, "qt_window", None) if qt_window is not None: qt_window.show() qt_window.raise_() qt_window.activateWindow() show_toast(parent, "该患者的 IM 会话已经打开。", "info", 3200) return if ( call_key in self.video_pending or existing_call is not None or call_key in self.demo_video_dialogs ): show_toast(parent, "该问诊的视频正在准备或通话中。", "info", 3600) return closed_previous_im = False if open_im: for key, call in tuple(self.video_calls.items()): if key == call_key or not getattr(call, "open_im", False): continue closed_previous_im = True self.video_calls.pop(key, None) with suppress(Exception): call.close() if self.current_demo_mode: dialog = DemoVideoDialog(patient_name, parent) dialog.finished.connect( lambda _result, key=call_key, item=dialog: self._forget_demo_dialog( key, item, ) ) self.demo_video_dialogs[call_key] = dialog dialog.show() return show_toast( parent, "正在打开患者 IM 会话…" if open_im else "正在获取安全通话凭证…", "info", ) repository = self.current_repository marker = object() self.video_pending[call_key] = marker def get_ticket() -> Any: return repository.get_call_ticket( patient_id=int(patient_id), diagnosis_id=int(diagnosis_id), ) def request_ticket() -> None: if self.video_pending.get(call_key) is not marker: return run_async( get_ticket, on_success=lambda ticket: self._launch_video( ticket, diagnosis_id=diagnosis_id, patient_id=patient_id, repository=repository, call_key=call_key, marker=marker, open_im=open_im, patient_name=patient_name, ), on_error=lambda error: self._video_ticket_error( call_key, marker, parent, error, ), ) # Tencent IM may take a brief moment to release the previous browser # connection. The admin version also has only one ChatDialog instance. if closed_previous_im: QTimer.singleShot(400, request_ticket) else: request_ticket() def _video_ticket_error( self, call_key: str, marker: object, parent: QWidget, error: Exception, ) -> None: if self.video_pending.get(call_key) is not marker: return self.video_pending.pop(call_key, None) if self.shell_window is parent: show_toast( parent, f"视频准备失败：{friendly_error(error)}", "danger", 5200, ) def _launch_video( self, ticket: Any, *, diagnosis_id: Any, patient_id: Any, repository: Any, call_key: str, marker: object, open_im: bool = False, patient_name: str = "患者", ) -> None: if self.video_pending.get(call_key) is not marker: return self.video_pending.pop(call_key, None) if ( self.shell_window is None or self.current_repository is not repository or call_key in self.video_calls ): return try: mode = BackendMode.parse(self.config.video_mode) if mode is BackendMode.BROWSER: raise ValueError("当前后端未提供一次性通话交接票据，浏览器视频模式已安全停用。") if not WEBENGINE_AVAILABLE: raise ValueError("当前安装缺少 QtWebEngine，无法打开受信任的视频窗口。") call = launch_video_call( ticket, repository=repository, diagnosis_id=diagnosis_id, patient_id=patient_id, backend_mode=mode, local_dist=video_dist_path(), remote_url=self.config.video_web_url or None, logger=logging.getLogger("doctor_workstation.video"), open_im=open_im, patient_name=patient_name, ) except Exception as error: LOGGER.exception("video call could not be launched") show_toast( self.shell_window, f"视频启动失败：{friendly_error(error)}", "danger", 5600, ) return self.video_calls[call_key] = call qt_window = getattr(call, "qt_window", None) if qt_window is not None: qt_window.destroyed.connect( lambda _obj=None, key=call_key, expected=call: self._release_video_call( key, expected, ) ) def _release_video_call(self, call_key: str, call: Any) -> None: if self.video_calls.get(call_key) is call: self.video_calls.pop(call_key, None) def _forget_demo_dialog(self, call_key: str, dialog: DemoVideoDialog) -> None: if self.demo_video_dialogs.get(call_key) is dialog: self.demo_video_dialogs.pop(call_key, None) dialog.deleteLater() @staticmethod def _wait_for_video_lifecycle(calls: tuple[Any, ...], *, timeout: float) -> bool: """Give queued endCall writes one short shared deadline after media closes.""" deadline = time.monotonic() + max(0.0, timeout) complete = True for call in calls: remaining = deadline - time.monotonic() if remaining <= 0: complete = False break wait = getattr(call, "wait_for_lifecycle", None) if callable(wait) and not wait(remaining): complete = False if not complete: LOGGER.warning("video lifecycle cleanup exceeded its bounded deadline") return complete @staticmethod def _apply_window_icon(window: QWidget) -> None: icon_file = resource_path("icon.svg") if icon_file.exists(): window.setWindowIcon(QIcon(str(icon_file))) def shutdown(self) -> None: """Invalidate asynchronous restoration and release owned resources.""" if self._shutting_down: return self._shutting_down = True self._cancel_session_restore() set_authentication_expired_handler(None) calls = tuple(self.video_calls.values()) for call in calls: with suppress(Exception): call.close() self._wait_for_video_lifecycle(calls, timeout=1.25) if self.remote_repository is not None: with suppress(Exception): self.remote_repository.client.close() def _create_application(argv: list[str]) -> QApplication: with suppress(AttributeError): QGuiApplication.setHighDpiScaleFactorRoundingPolicy( Qt.HighDpiScaleFactorRoundingPolicy.PassThrough ) with suppress(AttributeError): QApplication.setAttribute(Qt.ApplicationAttribute.AA_DontCreateNativeWidgetSiblings, True) application = QApplication(argv) _install_chinese_translations(application) application.setApplicationName("甄养堂医生工作站") application.setApplicationDisplayName("甄养堂医生工作站") application.setOrganizationName("ZhenYangTang") application.setOrganizationDomain("zhenyangtang.com") application.setQuitOnLastWindowClosed(True) icon_file = resource_path("icon.svg") if icon_file.exists(): application.setWindowIcon(QIcon(str(icon_file))) apply_theme(application) return application def main(argv: list[str] | None = None) -> int: """Start the GUI application and return its process exit code.""" config = AppConfig.load() configure_logging(config.log_dir, config.log_level) LOGGER.info("doctor workstation starting", extra={"demo_mode": config.demo_mode}) raw_argv = list(sys.argv if argv is None else argv) smoke_test = "--smoke-test" in raw_argv application = _create_application( [argument for argument in raw_argv if argument != "--smoke-test"] ) controller = ApplicationController(application, config) controller.start() if smoke_test or os.getenv("DOCTOR_SMOKE_TEST") == "1": QTimer.singleShot(1200, application.quit) return application.exec() __all__ = ["ApplicationController", "DemoVideoDialog", "main"]