1103 lines
40 KiB
Python
1103 lines
40 KiB
Python
"""Shared widgets and safe asynchronous helpers for the UI layer."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import inspect
|
||
import re
|
||
import traceback
|
||
from collections.abc import Callable, Iterable, Mapping, Sequence
|
||
from dataclasses import dataclass
|
||
from datetime import date, datetime
|
||
from typing import Any
|
||
|
||
from PySide6.QtCore import QObject, QRunnable, Qt, QThreadPool, QTimer, Signal, Slot
|
||
from PySide6.QtGui import QColor, QPainter, QPaintEvent, QResizeEvent
|
||
from PySide6.QtWidgets import (
|
||
QAbstractItemView,
|
||
QFrame,
|
||
QHBoxLayout,
|
||
QLabel,
|
||
QMenu,
|
||
QPushButton,
|
||
QSizePolicy,
|
||
QTableWidget,
|
||
QTableWidgetItem,
|
||
QToolButton,
|
||
QVBoxLayout,
|
||
QWidget,
|
||
)
|
||
|
||
from doctor_workstation.core.errors import (
|
||
ApiHttpError,
|
||
ApiProtocolError,
|
||
ApiTimeoutError,
|
||
ApiTransportError,
|
||
AuthenticationExpiredError,
|
||
)
|
||
|
||
AuthenticationExpiredHandler = Callable[[AuthenticationExpiredError], bool]
|
||
_AUTHENTICATION_EXPIRED_HANDLER: AuthenticationExpiredHandler | None = None
|
||
|
||
|
||
def set_authentication_expired_handler(
|
||
handler: AuthenticationExpiredHandler | None,
|
||
) -> None:
|
||
"""Install the application-level expired-session callback.
|
||
|
||
The callback returns ``True`` when it consumed the error. Returning
|
||
``False`` preserves the originating operation's local error handling, as is
|
||
required when authentication itself fails on the login screen.
|
||
"""
|
||
|
||
global _AUTHENTICATION_EXPIRED_HANDLER
|
||
_AUTHENTICATION_EXPIRED_HANDLER = handler
|
||
|
||
|
||
def get_value(value: Any, key: str, default: Any = None) -> Any:
|
||
"""Read a dotted key from mappings, dataclasses, or ordinary objects."""
|
||
|
||
current = value
|
||
for part in key.split("."):
|
||
if current is None:
|
||
return default
|
||
if isinstance(current, Mapping):
|
||
current = current.get(part, default)
|
||
else:
|
||
marker = object()
|
||
candidate = getattr(current, part, marker)
|
||
if candidate is marker:
|
||
raw = getattr(current, "raw", None)
|
||
candidate = raw.get(part, marker) if isinstance(raw, Mapping) else marker
|
||
current = default if candidate is marker else candidate
|
||
if current is default:
|
||
return default
|
||
return current
|
||
|
||
|
||
def first_value(value: Any, *keys: str, default: Any = None) -> Any:
|
||
"""Return the first present, non-empty value from ``keys``."""
|
||
|
||
for key in keys:
|
||
candidate = get_value(value, key, None)
|
||
if candidate is not None and candidate != "":
|
||
return candidate
|
||
return default
|
||
|
||
|
||
def display_text(value: Any, default: str = "—") -> str:
|
||
if value is None or value == "":
|
||
return default
|
||
if isinstance(value, bool):
|
||
return "是" if value else "否"
|
||
if isinstance(value, (datetime, date)):
|
||
return value.strftime("%Y-%m-%d %H:%M" if isinstance(value, datetime) else "%Y-%m-%d")
|
||
return str(value)
|
||
|
||
|
||
_UNIX_TIMESTAMP_RE = re.compile(r"\d{10,13}(?:\.\d+)?")
|
||
_RECORD_TIME_FORMAT = "%Y-%m-%d %H:%M"
|
||
|
||
|
||
def format_record_time(value: Any, default: str = "—") -> str:
|
||
"""Format a record timestamp that may arrive as Unix seconds, milliseconds, or ISO text.
|
||
|
||
The desktop pages used to render raw ``str(int)`` for fields like
|
||
``create_time`` whenever the server returned a numeric epoch. This
|
||
helper keeps the legacy strings (already formatted dates) untouched
|
||
while normalising the numeric and ISO variants to ``YYYY-MM-DD HH:MM``
|
||
so the column reads as a real creation date.
|
||
"""
|
||
|
||
if value is None or value == "":
|
||
return default
|
||
if isinstance(value, datetime):
|
||
return value.strftime(_RECORD_TIME_FORMAT)
|
||
if isinstance(value, date):
|
||
return value.strftime("%Y-%m-%d")
|
||
raw = str(value).strip()
|
||
if not raw:
|
||
return default
|
||
if _UNIX_TIMESTAMP_RE.fullmatch(raw):
|
||
stamp = float(raw)
|
||
if stamp >= 10_000_000_000:
|
||
stamp /= 1000.0
|
||
try:
|
||
return datetime.fromtimestamp(stamp).strftime(_RECORD_TIME_FORMAT)
|
||
except (OSError, OverflowError, ValueError):
|
||
return raw
|
||
normalized = raw.replace("T", " ").replace("Z", "")
|
||
if len(normalized) >= 16:
|
||
return normalized[:16]
|
||
return raw
|
||
|
||
|
||
def gender_text(value: Any, default: str = "—") -> str:
|
||
"""Format the legacy gender codes without exposing numeric API values."""
|
||
|
||
normalized = str(value).strip().lower() if value not in (None, "") else ""
|
||
labels = {
|
||
"0": "未知",
|
||
"1": "男",
|
||
"2": "女",
|
||
"m": "男",
|
||
"male": "男",
|
||
"f": "女",
|
||
"female": "女",
|
||
"unknown": "未知",
|
||
}
|
||
return labels.get(normalized, display_text(value, default))
|
||
|
||
|
||
def page_items(result: Any) -> list[Any]:
|
||
"""Extract rows from common PageResult/dict response shapes."""
|
||
|
||
if result is None:
|
||
return []
|
||
if isinstance(result, (list, tuple)):
|
||
return list(result)
|
||
for key in ("items", "lists", "results", "rows", "data"):
|
||
items = get_value(result, key, None)
|
||
if isinstance(items, (list, tuple)):
|
||
return list(items)
|
||
if key == "data" and items is not None and items is not result:
|
||
nested = page_items(items)
|
||
if nested:
|
||
return nested
|
||
return []
|
||
|
||
|
||
def page_total(result: Any, fallback: int = 0) -> int:
|
||
for key in ("total", "count", "total_count", "data.total", "data.count"):
|
||
value = get_value(result, key, None)
|
||
if value is not None:
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return fallback
|
||
|
||
|
||
def has_permission(permissions: Any, codes: str | Sequence[str], default: bool = True) -> bool:
|
||
"""Check canonical permissions with exact and resource-wildcard semantics.
|
||
|
||
A sequence uses OR semantics, matching the admin client's route guards.
|
||
Permission names are opaque: ``resource/action`` never aliases
|
||
``resource.action``.
|
||
"""
|
||
|
||
if permissions is None:
|
||
return default
|
||
requested = tuple(
|
||
code.strip()
|
||
for code in ((codes,) if isinstance(codes, str) else tuple(codes))
|
||
if code and code.strip()
|
||
)
|
||
if not requested:
|
||
return True
|
||
|
||
for method_name in ("allows", "has", "can", "contains", "has_permission"):
|
||
method = getattr(permissions, method_name, None)
|
||
if callable(method):
|
||
for code in requested:
|
||
try:
|
||
if bool(method(code)):
|
||
return True
|
||
except (TypeError, ValueError):
|
||
continue
|
||
|
||
raw = permissions
|
||
for attr in ("codes", "permissions", "values"):
|
||
candidate = getattr(permissions, attr, None)
|
||
if candidate is not None and not callable(candidate):
|
||
raw = candidate
|
||
break
|
||
if isinstance(raw, Mapping):
|
||
available = {str(key).strip() for key, enabled in raw.items() if enabled}
|
||
elif isinstance(raw, str):
|
||
available = {raw}
|
||
else:
|
||
try:
|
||
available = {str(item).strip() for item in raw}
|
||
except TypeError:
|
||
return default
|
||
return any(
|
||
"*" in available
|
||
or code in available
|
||
or any(grant.endswith("/*") and code.startswith(grant[:-1]) for grant in available)
|
||
for code in requested
|
||
)
|
||
|
||
|
||
def invoke(repository: Any, method_name: str, /, **kwargs: Any) -> Any:
|
||
"""Invoke a repository method with keyword filtering for contract tolerance."""
|
||
|
||
method = getattr(repository, method_name, None)
|
||
if method is None and method_name == "save_prescription_template":
|
||
template = kwargs.get("template") or {}
|
||
template_id = kwargs.get("template_id", kwargs.get("id"))
|
||
if template_id is None:
|
||
creator = (
|
||
getattr(repository, "create_prescription_template", None)
|
||
or repository.add_prescription_template
|
||
)
|
||
return creator(template=template)
|
||
updater = (
|
||
getattr(repository, "update_prescription_template", None)
|
||
or repository.edit_prescription_template
|
||
)
|
||
return updater(template_id, changes=template)
|
||
|
||
aliases = {
|
||
"reception_queue": "list_appointments",
|
||
"reception_detail": "get_reception",
|
||
"prescription_library": "list_prescription_templates",
|
||
"prescriptions": "list_prescriptions",
|
||
"prescription_detail": "get_prescription",
|
||
"patients": "list_patients",
|
||
"consultations": "list_consultations",
|
||
}
|
||
resolved_name = method_name
|
||
if method is None:
|
||
resolved_name = aliases.get(method_name, method_name)
|
||
method = getattr(repository, resolved_name, None)
|
||
if method is None and method_name == "reception_detail":
|
||
resolved_name = "reception"
|
||
method = getattr(repository, resolved_name)
|
||
if method is None:
|
||
raise AttributeError(f"repository has no method {method_name!r}")
|
||
|
||
call_kwargs = dict(kwargs)
|
||
if resolved_name.startswith("list_") and "page" in call_kwargs and "page_no" not in call_kwargs:
|
||
call_kwargs["page_no"] = call_kwargs.pop("page")
|
||
if (
|
||
resolved_name == "get_prescription"
|
||
and "id" in call_kwargs
|
||
and "prescription_id" not in call_kwargs
|
||
):
|
||
call_kwargs["prescription_id"] = call_kwargs.pop("id")
|
||
try:
|
||
signature = inspect.signature(method)
|
||
except (TypeError, ValueError):
|
||
return method(**call_kwargs)
|
||
parameters = signature.parameters
|
||
if any(p.kind == inspect.Parameter.VAR_KEYWORD for p in parameters.values()):
|
||
return method(**call_kwargs)
|
||
accepted = {
|
||
name: value
|
||
for name, value in call_kwargs.items()
|
||
if name in parameters
|
||
and parameters[name].kind
|
||
in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
|
||
}
|
||
if len(parameters) == 1 and not accepted:
|
||
only = next(iter(parameters.values()))
|
||
if only.name in {"payload", "data", "query", "filters", "params"}:
|
||
return method(call_kwargs)
|
||
return method(**accepted)
|
||
|
||
|
||
class WorkerSignals(QObject):
|
||
result = Signal(object)
|
||
error = Signal(object, str)
|
||
finished = Signal()
|
||
|
||
|
||
class Worker(QRunnable):
|
||
"""A small QRunnable that marshals results back through Qt signals."""
|
||
|
||
def __init__(self, function: Callable[..., Any], *args: Any, **kwargs: Any) -> None:
|
||
super().__init__()
|
||
self.function = function
|
||
self.args = args
|
||
self.kwargs = kwargs
|
||
self.signals = WorkerSignals()
|
||
|
||
@Slot()
|
||
def run(self) -> None:
|
||
try:
|
||
result = self.function(*self.args, **self.kwargs)
|
||
except Exception as exc: # UI boundary: report domain and transport errors alike.
|
||
self.signals.error.emit(exc, traceback.format_exc())
|
||
else:
|
||
self.signals.result.emit(result)
|
||
finally:
|
||
self.signals.finished.emit()
|
||
|
||
|
||
_RUNNING_WORKERS: set[Worker] = set()
|
||
|
||
|
||
def _dispatch_async_error(
|
||
error: Exception,
|
||
local_handler: Callable[[Exception], None] | None,
|
||
) -> None:
|
||
"""Route session expiry globally before falling back to a page handler."""
|
||
|
||
handled = False
|
||
if isinstance(error, AuthenticationExpiredError):
|
||
handler = _AUTHENTICATION_EXPIRED_HANDLER
|
||
if handler is not None:
|
||
try:
|
||
handled = bool(handler(error))
|
||
except Exception:
|
||
traceback.print_exc()
|
||
if not handled and local_handler is not None:
|
||
local_handler(error)
|
||
|
||
|
||
def run_async(
|
||
function: Callable[..., Any],
|
||
*args: Any,
|
||
on_success: Callable[[Any], None] | None = None,
|
||
on_error: Callable[[Exception], None] | None = None,
|
||
on_finished: Callable[[], None] | None = None,
|
||
pool: QThreadPool | None = None,
|
||
priority: int = 0,
|
||
**kwargs: Any,
|
||
) -> Worker:
|
||
"""Run ``function`` off the GUI thread and return its Worker handle.
|
||
|
||
Higher-priority queued work starts first when a pool thread becomes free.
|
||
Running work is never interrupted.
|
||
"""
|
||
|
||
worker = Worker(function, *args, **kwargs)
|
||
_RUNNING_WORKERS.add(worker)
|
||
if on_success is not None:
|
||
worker.signals.result.connect(on_success)
|
||
worker.signals.error.connect(lambda exc, _tb: _dispatch_async_error(exc, on_error))
|
||
if on_finished is not None:
|
||
worker.signals.finished.connect(on_finished)
|
||
worker.signals.finished.connect(lambda: _RUNNING_WORKERS.discard(worker))
|
||
(pool or QThreadPool.globalInstance()).start(worker, priority)
|
||
return worker
|
||
|
||
|
||
def friendly_error(error: Any) -> str:
|
||
text = str(error).strip()
|
||
lowered = text.lower()
|
||
if "certificate_verify_failed" in lowered or "self-signed certificate" in lowered:
|
||
return (
|
||
"服务器证书不受系统信任。若这是可信内网的自签名服务器，请展开“服务器设置”，"
|
||
"勾选“信任自签名证书（仅内网调试）”后重新登录，设置会自动应用。"
|
||
)
|
||
if isinstance(error, AuthenticationExpiredError):
|
||
return "登录状态已失效，请重新登录。"
|
||
if isinstance(error, ApiTimeoutError) or "timed out" in lowered or "timeout" in lowered:
|
||
return "连接服务器超时，请检查网络后重试。"
|
||
if isinstance(error, ApiProtocolError) or any(
|
||
marker in lowered
|
||
for marker in (
|
||
"api response envelope",
|
||
"api response is not valid json",
|
||
"invalid json",
|
||
)
|
||
):
|
||
return "服务器返回的数据格式不正确，请联系管理员检查接口。"
|
||
if isinstance(error, ApiHttpError):
|
||
status_code = getattr(error, "status_code", None)
|
||
suffix = f"（状态码 {status_code}）" if status_code else ""
|
||
return f"服务器请求失败{suffix}，请稍后重试。"
|
||
if isinstance(error, ApiTransportError) or any(
|
||
marker in lowered
|
||
for marker in (
|
||
"connection refused",
|
||
"connecterror",
|
||
"connection error",
|
||
"failed to connect",
|
||
"getaddrinfo failed",
|
||
"name or service not known",
|
||
"network is unreachable",
|
||
)
|
||
):
|
||
return "无法连接服务器，请检查服务器地址与网络。"
|
||
if any(marker in lowered for marker in ("unauthorized", "forbidden", "permission denied")):
|
||
return "当前账号无权执行此操作。"
|
||
if "not found" in lowered:
|
||
return "未找到所需数据。"
|
||
# 友好的 Dify / AI 上游错误映射。原文来自服务端 DifyChatService 的 error_code 分支，
|
||
# 当上游拒绝、超时或 quota 受限时，原文对医生不友好，按场景给出可执行引导。
|
||
upstream_hint = _ai_upstream_hint(text)
|
||
if upstream_hint is not None:
|
||
return upstream_hint
|
||
if any("\u4e00" <= character <= "\u9fff" for character in text):
|
||
return text
|
||
if isinstance(error, TypeError):
|
||
return "程序执行失败，请重试；若问题持续出现，请联系管理员。"
|
||
return "操作未完成，请稍后重试。"
|
||
|
||
|
||
_AI_UPSTREAM_HINTS: tuple[tuple[str, str], ...] = (
|
||
("模型未能处理本次请求", "AI 助手暂时无法处理本次请求，请稍后重试；若多次出现请联系管理员检查 AI 服务配置。"),
|
||
("模型响应超时", "AI 助手响应超时，请稍后重试。"),
|
||
("暂时无法连接 AI 服务", "无法连接 AI 助手服务，请检查网络或稍后重试。"),
|
||
("模型服务繁忙", "AI 助手服务繁忙，请稍后重试。"),
|
||
("AI 助手未返回内容", "AI 助手未返回内容，请稍后重试。"),
|
||
("AI 助手暂时不可用", "AI 助手暂时不可用，请稍后重试。"),
|
||
("病例数据编码失败", "病例数据无法发送到 AI 助手，请联系管理员。"),
|
||
("无法初始化 AI 请求", "无法初始化 AI 助手请求，请联系管理员。"),
|
||
("AI 服务凭据无效或无权限", "AI 助手服务凭据无效或无权限，请联系管理员。"),
|
||
("AI 服务配置无效", "AI 助手服务配置无效，请联系管理员。"),
|
||
("该模型服务尚未完整配置", "AI 助手服务尚未完整配置，请联系管理员。"),
|
||
("AI 报告功能未启用", "AI 助手功能尚未启用，请联系管理员。"),
|
||
("不支持的 AI 模型", "AI 助手模型不受支持，请联系管理员。"),
|
||
)
|
||
|
||
|
||
_AI_UPSTREAM_CODE = re.compile(r"[（(]([A-Z][A-Z0-9_]{2,39})[)）]")
|
||
|
||
|
||
def _ai_upstream_hint(text: str) -> str | None:
|
||
for marker, hint in _AI_UPSTREAM_HINTS:
|
||
if marker in text:
|
||
# 服务端会在消息尾部附带上游错误码。保留它，医生截图反馈时管理员
|
||
# 能直接区分是配置问题、资料体积超限还是上游拒绝。
|
||
match = _AI_UPSTREAM_CODE.search(text)
|
||
if match is not None:
|
||
return f"{hint}（{match.group(1)}）"
|
||
return hint
|
||
return None
|
||
|
||
|
||
class PageHeader(QWidget):
|
||
"""Reference-design page heading with breadcrumb, copy and actions."""
|
||
|
||
def __init__(
|
||
self,
|
||
title: str,
|
||
subtitle: str = "",
|
||
parent: QWidget | None = None,
|
||
) -> None:
|
||
super().__init__(parent)
|
||
self.setObjectName("PageHeader")
|
||
layout = QVBoxLayout(self)
|
||
layout.setContentsMargins(0, 0, 0, 0)
|
||
layout.setSpacing(8)
|
||
|
||
breadcrumb = QHBoxLayout()
|
||
breadcrumb.setContentsMargins(0, 0, 0, 0)
|
||
breadcrumb.setSpacing(7)
|
||
home_label = QLabel("首页", self)
|
||
home_label.setProperty("role", "breadcrumb")
|
||
breadcrumb.addWidget(home_label)
|
||
separator = QLabel("›", self)
|
||
separator.setProperty("role", "breadcrumbSeparator")
|
||
breadcrumb.addWidget(separator)
|
||
self.breadcrumb_label = QLabel(title, self)
|
||
self.breadcrumb_label.setProperty("role", "breadcrumbCurrent")
|
||
breadcrumb.addWidget(self.breadcrumb_label)
|
||
breadcrumb.addStretch(1)
|
||
layout.addLayout(breadcrumb)
|
||
|
||
heading = QHBoxLayout()
|
||
heading.setContentsMargins(0, 0, 0, 0)
|
||
heading.setSpacing(16)
|
||
text_layout = QVBoxLayout()
|
||
text_layout.setSpacing(3)
|
||
self.title_label = QLabel(title, self)
|
||
self.title_label.setProperty("role", "pageTitle")
|
||
text_layout.addWidget(self.title_label)
|
||
self.subtitle_label = QLabel(subtitle, self)
|
||
self.subtitle_label.setProperty("role", "muted")
|
||
self.subtitle_label.setWordWrap(True)
|
||
self.subtitle_label.setVisible(bool(subtitle))
|
||
text_layout.addWidget(self.subtitle_label)
|
||
heading.addLayout(text_layout, 1)
|
||
self.actions = QHBoxLayout()
|
||
self.actions.setSpacing(8)
|
||
heading.addLayout(self.actions)
|
||
layout.addLayout(heading)
|
||
|
||
def add_action(self, widget: QWidget) -> QWidget:
|
||
self.actions.addWidget(widget)
|
||
return widget
|
||
|
||
def set_subtitle(self, text: str) -> None:
|
||
self.subtitle_label.setText(text)
|
||
self.subtitle_label.setVisible(bool(text))
|
||
|
||
|
||
class MetricCard(QFrame):
|
||
"""Compact dashboard metric used above dense business tables."""
|
||
|
||
def __init__(
|
||
self,
|
||
title: str,
|
||
value: str = "0",
|
||
*,
|
||
hint: str = "",
|
||
kind: str = "accent",
|
||
glyph: str = "",
|
||
parent: QWidget | None = None,
|
||
) -> None:
|
||
super().__init__(parent)
|
||
self.setObjectName("MetricCard")
|
||
self.setProperty("kind", kind)
|
||
self.setMinimumHeight(82)
|
||
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
||
layout = QHBoxLayout(self)
|
||
layout.setContentsMargins(16, 12, 14, 12)
|
||
layout.setSpacing(12)
|
||
copy = QVBoxLayout()
|
||
copy.setSpacing(3)
|
||
title_label = QLabel(title, self)
|
||
title_label.setProperty("role", "metricTitle")
|
||
copy.addWidget(title_label)
|
||
self.value_label = QLabel(str(value), self)
|
||
self.value_label.setProperty("role", "metricValue")
|
||
copy.addWidget(self.value_label)
|
||
self.hint_label = QLabel(hint, self)
|
||
self.hint_label.setProperty("role", "metricHint")
|
||
self.hint_label.setVisible(bool(hint))
|
||
copy.addWidget(self.hint_label)
|
||
layout.addLayout(copy, 1)
|
||
self.glyph_label = QLabel(glyph, self)
|
||
self.glyph_label.setObjectName("MetricGlyph")
|
||
self.glyph_label.setProperty("kind", kind)
|
||
self.glyph_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
self.glyph_label.setFixedSize(38, 38)
|
||
self.glyph_label.setVisible(bool(glyph))
|
||
layout.addWidget(self.glyph_label)
|
||
|
||
def set_value(self, value: Any) -> None:
|
||
self.value_label.setText(display_text(value, "0"))
|
||
|
||
def set_hint(self, text: str) -> None:
|
||
self.hint_label.setText(text)
|
||
self.hint_label.setVisible(bool(text))
|
||
|
||
|
||
class StatusBadge(QLabel):
|
||
def __init__(
|
||
self, text: str = "", kind: str = "neutral", parent: QWidget | None = None
|
||
) -> None:
|
||
super().__init__(text, parent)
|
||
self.setObjectName("StatusBadge")
|
||
self.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
self.setSizePolicy(QSizePolicy.Policy.Maximum, QSizePolicy.Policy.Fixed)
|
||
# QSS padding alone is not a reliable minimum on every Windows DPI
|
||
# scale. Keep enough physical row height so badge text is never
|
||
# squeezed into the thin coloured strip seen in list item widgets.
|
||
self.setMinimumHeight(24)
|
||
self.set_kind(kind)
|
||
|
||
def set_kind(self, kind: str) -> None:
|
||
self.setProperty("kind", kind)
|
||
self.style().unpolish(self)
|
||
self.style().polish(self)
|
||
|
||
def set_status(self, text: str, kind: str = "neutral") -> None:
|
||
self.setText(text)
|
||
self.set_kind(kind)
|
||
|
||
|
||
class EmptyState(QWidget):
|
||
action_requested = Signal()
|
||
|
||
def __init__(
|
||
self,
|
||
title: str = "暂无数据",
|
||
description: str = "调整筛选条件后再试试。",
|
||
action_text: str = "",
|
||
parent: QWidget | None = None,
|
||
) -> None:
|
||
super().__init__(parent)
|
||
self.setObjectName("EmptyState")
|
||
layout = QVBoxLayout(self)
|
||
layout.setContentsMargins(24, 44, 24, 44)
|
||
layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
layout.setSpacing(8)
|
||
glyph = QLabel("○", self)
|
||
glyph.setObjectName("EmptyStateGlyph")
|
||
glyph.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
glyph.setFixedSize(44, 44)
|
||
layout.addWidget(glyph)
|
||
title_label = QLabel(title, self)
|
||
title_label.setProperty("role", "sectionTitle")
|
||
title_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
layout.addWidget(title_label)
|
||
description_label = QLabel(description, self)
|
||
description_label.setProperty("role", "muted")
|
||
description_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
description_label.setWordWrap(True)
|
||
layout.addWidget(description_label)
|
||
self.action_button = QPushButton(action_text, self)
|
||
self.action_button.setProperty("variant", "secondary")
|
||
self.action_button.setVisible(bool(action_text))
|
||
self.action_button.clicked.connect(self.action_requested)
|
||
layout.addWidget(self.action_button, 0, Qt.AlignmentFlag.AlignCenter)
|
||
|
||
|
||
class MessageBanner(QFrame):
|
||
def __init__(self, text: str = "", kind: str = "info", parent: QWidget | None = None) -> None:
|
||
super().__init__(parent)
|
||
self.setObjectName("MessageBanner")
|
||
self.setProperty("kind", kind)
|
||
layout = QHBoxLayout(self)
|
||
layout.setContentsMargins(12, 9, 12, 9)
|
||
layout.setSpacing(9)
|
||
self.icon = QLabel("i", self)
|
||
self.icon.setObjectName("MessageBannerIcon")
|
||
self.icon.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
self.icon.setFixedSize(20, 20)
|
||
self.label = QLabel(text, self)
|
||
self.label.setObjectName("MessageBannerText")
|
||
self.label.setWordWrap(True)
|
||
layout.addWidget(self.icon)
|
||
layout.addWidget(self.label, 1)
|
||
self.setVisible(bool(text))
|
||
|
||
def show_message(self, text: str, kind: str = "info") -> None:
|
||
glyphs = {"info": "i", "success": "✓", "warning": "!", "danger": "!"}
|
||
self.label.setText(text)
|
||
self.icon.setText(glyphs.get(kind, "i"))
|
||
self.setProperty("kind", kind)
|
||
self.style().unpolish(self)
|
||
self.style().polish(self)
|
||
self.setVisible(bool(text))
|
||
|
||
def clear(self) -> None:
|
||
self.setVisible(False)
|
||
self.label.clear()
|
||
|
||
|
||
class Toast(QLabel):
|
||
def __init__(self, parent: QWidget) -> None:
|
||
super().__init__(parent)
|
||
self.setObjectName("Toast")
|
||
self.setWordWrap(True)
|
||
self.setMaximumWidth(420)
|
||
self.hide()
|
||
self._timer = QTimer(self)
|
||
self._timer.setSingleShot(True)
|
||
self._timer.timeout.connect(self.hide)
|
||
|
||
def show_message(self, text: str, kind: str = "info", duration: int = 2800) -> None:
|
||
self.setText(text)
|
||
self.setProperty("kind", kind)
|
||
self.style().unpolish(self)
|
||
self.style().polish(self)
|
||
self.adjustSize()
|
||
parent = self.parentWidget()
|
||
if parent is not None:
|
||
self.move(max(16, parent.width() - self.width() - 24), 20)
|
||
self.raise_()
|
||
self.show()
|
||
self._timer.start(duration)
|
||
|
||
|
||
def show_toast(parent: QWidget, text: str, kind: str = "info", duration: int = 2800) -> None:
|
||
window = parent.window()
|
||
toast = getattr(window, "_doctor_workstation_toast", None)
|
||
if not isinstance(toast, Toast):
|
||
toast = Toast(window)
|
||
window._doctor_workstation_toast = toast
|
||
toast.show_message(text, kind, duration)
|
||
|
||
|
||
class _BusyTrack(QWidget):
|
||
"""Indeterminate bar painted in-process, without Windows QProgressBar HWNDs.
|
||
|
||
``QProgressBar.setRange(0, 0)`` uses the native Vista/Win11 animation, which
|
||
creates extra untitled windows titled with the application name.
|
||
"""
|
||
|
||
def __init__(self, parent: QWidget) -> None:
|
||
super().__init__(parent)
|
||
self.setObjectName("BusyOverlayProgress")
|
||
self.setFixedSize(140, 6)
|
||
self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True)
|
||
self._phase = 0.0
|
||
self._timer = QTimer(self)
|
||
self._timer.setInterval(32)
|
||
self._timer.timeout.connect(self._tick)
|
||
|
||
def showEvent(self, event: Any) -> None:
|
||
self._timer.start()
|
||
super().showEvent(event)
|
||
|
||
def hideEvent(self, event: Any) -> None:
|
||
self._timer.stop()
|
||
super().hideEvent(event)
|
||
|
||
def _tick(self) -> None:
|
||
self._phase = (self._phase + 0.035) % 1.0
|
||
self.update()
|
||
|
||
def paintEvent(self, event: QPaintEvent) -> None: # noqa: N802 - Qt API
|
||
del event
|
||
painter = QPainter(self)
|
||
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
||
painter.setPen(Qt.PenStyle.NoPen)
|
||
rect = self.rect()
|
||
painter.setBrush(QColor("#D8DEEA"))
|
||
painter.drawRoundedRect(rect, 3, 3)
|
||
chunk_width = max(36, int(rect.width() * 0.32))
|
||
span = rect.width() + chunk_width
|
||
x = int(self._phase * span) - chunk_width
|
||
painter.setBrush(QColor("#4F63D9"))
|
||
painter.drawRoundedRect(x, 0, chunk_width, rect.height(), 3, 3)
|
||
|
||
|
||
class BusyOverlay(QFrame):
|
||
"""Non-blocking visual guard for a card or page while a worker is active."""
|
||
|
||
def __init__(self, parent: QWidget, text: str = "正在加载…") -> None:
|
||
super().__init__(parent)
|
||
self.setObjectName("BusyOverlay")
|
||
self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, True)
|
||
layout = QVBoxLayout(self)
|
||
layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
layout.setSpacing(10)
|
||
self.label = QLabel(text, self)
|
||
self.label.setProperty("role", "muted")
|
||
self.progress = _BusyTrack(self)
|
||
layout.addWidget(self.label, 0, Qt.AlignmentFlag.AlignCenter)
|
||
layout.addWidget(self.progress, 0, Qt.AlignmentFlag.AlignCenter)
|
||
self.hide()
|
||
|
||
def set_message(self, text: str) -> None:
|
||
self.label.setText(text)
|
||
|
||
def showEvent(self, event: Any) -> None:
|
||
parent = self.parentWidget()
|
||
if parent is not None:
|
||
self.setGeometry(parent.rect())
|
||
self.raise_()
|
||
super().showEvent(event)
|
||
|
||
|
||
class OverlayHost(QWidget):
|
||
"""Widget base that automatically sizes a BusyOverlay child."""
|
||
|
||
def resizeEvent(self, event: QResizeEvent) -> None:
|
||
overlay = getattr(self, "busy_overlay", None)
|
||
if isinstance(overlay, BusyOverlay):
|
||
overlay.setGeometry(self.rect())
|
||
super().resizeEvent(event)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TableColumn:
|
||
key: str
|
||
title: str
|
||
width: int = 0
|
||
formatter: Callable[[Any, Any], str] | None = None
|
||
alignment: Qt.AlignmentFlag = Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter
|
||
|
||
|
||
class SortableTable(QTableWidget):
|
||
"""A QTableWidget that safely retains the source object after sorting."""
|
||
|
||
def __init__(self, columns: Sequence[TableColumn], parent: QWidget | None = None) -> None:
|
||
super().__init__(parent)
|
||
self.columns = list(columns)
|
||
self.setColumnCount(len(self.columns))
|
||
self.setHorizontalHeaderLabels([column.title for column in self.columns])
|
||
self.setAlternatingRowColors(True)
|
||
self.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
|
||
self.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
|
||
self.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
|
||
self.setHorizontalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel)
|
||
self.setVerticalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel)
|
||
self.setMinimumHeight(0)
|
||
self.setSortingEnabled(True)
|
||
self.verticalHeader().setVisible(False)
|
||
self.horizontalHeader().setStretchLastSection(True)
|
||
for index, column in enumerate(self.columns):
|
||
if column.width:
|
||
self.setColumnWidth(index, column.width)
|
||
|
||
def set_rows(self, rows: Iterable[Any]) -> None:
|
||
selected_id = first_value(self.current_data(), "id", "appointment_id", default=None)
|
||
self.setSortingEnabled(False)
|
||
self.clearContents()
|
||
materialized = list(rows)
|
||
self.setRowCount(len(materialized))
|
||
row_to_select = -1
|
||
for row_index, row in enumerate(materialized):
|
||
row_id = first_value(row, "id", "appointment_id", default=None)
|
||
if selected_id is not None and row_id == selected_id:
|
||
row_to_select = row_index
|
||
for column_index, column in enumerate(self.columns):
|
||
raw = get_value(row, column.key, None)
|
||
text = column.formatter(raw, row) if column.formatter else display_text(raw)
|
||
item = QTableWidgetItem(text)
|
||
item.setTextAlignment(column.alignment)
|
||
item.setData(Qt.ItemDataRole.UserRole, row)
|
||
item.setToolTip(text if "\n" in text or len(text) > 18 else "")
|
||
self.setItem(row_index, column_index, item)
|
||
self.setSortingEnabled(True)
|
||
if row_to_select >= 0:
|
||
self.selectRow(row_to_select)
|
||
|
||
def current_data(self) -> Any:
|
||
row = self.currentRow()
|
||
if row < 0:
|
||
return None
|
||
item = self.item(row, 0)
|
||
return item.data(Qt.ItemDataRole.UserRole) if item is not None else None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RowAction:
|
||
"""One entry in a table row's action cell."""
|
||
|
||
label: str
|
||
callback: Callable[[], None]
|
||
danger: bool = False
|
||
#: Keep this action inline even when it would otherwise overflow.
|
||
pinned: bool = False
|
||
|
||
|
||
class RowActions(QWidget):
|
||
"""Row action cell that shows a couple of buttons and hides the rest.
|
||
|
||
List rows used to render every permitted action as its own small button —
|
||
up to seven on the diagnosis list. That widened the action column, pushed
|
||
the data columns into ellipsis, and left each row a different width, which
|
||
is what made the tables feel heavy and misaligned. Only the first
|
||
``max_visible`` non-destructive actions stay inline now; everything else,
|
||
destructive actions included, moves into a single 更多 menu.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
actions: Sequence[RowAction],
|
||
parent: QWidget | None = None,
|
||
*,
|
||
max_visible: int = 2,
|
||
) -> None:
|
||
super().__init__(parent)
|
||
self.setObjectName("RowActions")
|
||
layout = QHBoxLayout(self)
|
||
layout.setContentsMargins(4, 2, 4, 2)
|
||
layout.setSpacing(2)
|
||
|
||
inline: list[RowAction] = []
|
||
overflow: list[RowAction] = []
|
||
for action in actions:
|
||
room_left = len(inline) < max_visible
|
||
if action.pinned or (room_left and not action.danger):
|
||
inline.append(action)
|
||
else:
|
||
overflow.append(action)
|
||
|
||
self.buttons: list[QPushButton] = []
|
||
for action in inline:
|
||
button = QPushButton(action.label, self)
|
||
button.setProperty("rowAction", True)
|
||
if action.danger:
|
||
button.setProperty("variant", "dangerGhost")
|
||
button.setCursor(Qt.CursorShape.PointingHandCursor)
|
||
button.clicked.connect(lambda _checked=False, run=action.callback: run())
|
||
layout.addWidget(button)
|
||
self.buttons.append(button)
|
||
|
||
self.more_button: QToolButton | None = None
|
||
if overflow:
|
||
more = QToolButton(self)
|
||
more.setObjectName("RowActionsMore")
|
||
more.setText("更多")
|
||
more.setPopupMode(QToolButton.ToolButtonPopupMode.InstantPopup)
|
||
more.setCursor(Qt.CursorShape.PointingHandCursor)
|
||
menu = QMenu(more)
|
||
for action in overflow:
|
||
entry = menu.addAction(action.label)
|
||
if action.danger:
|
||
entry.setProperty("danger", True)
|
||
entry.triggered.connect(lambda _checked=False, run=action.callback: run())
|
||
more.setMenu(menu)
|
||
layout.addWidget(more)
|
||
self.more_button = more
|
||
self.menu = menu
|
||
|
||
layout.addStretch(1)
|
||
|
||
@property
|
||
def visible_labels(self) -> list[str]:
|
||
return [button.text() for button in self.buttons]
|
||
|
||
@property
|
||
def overflow_labels(self) -> list[str]:
|
||
if self.more_button is None:
|
||
return []
|
||
return [entry.text() for entry in self.more_button.menu().actions()]
|
||
|
||
|
||
class BusinessPager(QWidget):
|
||
"""Compact numbered pager shared by dense business-list pages."""
|
||
|
||
page_changed = Signal(int)
|
||
|
||
def __init__(self, page_size: int = 15, parent: QWidget | None = None) -> None:
|
||
super().__init__(parent)
|
||
self.setObjectName("BusinessPager")
|
||
self.setFixedHeight(42)
|
||
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
||
self.page = 1
|
||
self.page_size = page_size
|
||
self.total = 0
|
||
|
||
layout = QHBoxLayout(self)
|
||
layout.setContentsMargins(16, 4, 16, 4)
|
||
layout.setSpacing(7)
|
||
self.summary = QLabel("共 0 条", self)
|
||
self.summary.setProperty("role", "muted")
|
||
layout.addWidget(self.summary)
|
||
layout.addStretch(1)
|
||
|
||
self.previous = QPushButton("‹", self)
|
||
self.previous.setProperty("variant", "ghost")
|
||
self.previous.setAccessibleName("上一页")
|
||
self.previous.clicked.connect(lambda: self._request(self.page - 1))
|
||
layout.addWidget(self.previous)
|
||
|
||
self.pages_host = QWidget(self)
|
||
self.pages_layout = QHBoxLayout(self.pages_host)
|
||
self.pages_layout.setContentsMargins(0, 0, 0, 0)
|
||
self.pages_layout.setSpacing(5)
|
||
layout.addWidget(self.pages_host)
|
||
|
||
self.next = QPushButton("›", self)
|
||
self.next.setProperty("variant", "ghost")
|
||
self.next.setAccessibleName("下一页")
|
||
self.next.clicked.connect(lambda: self._request(self.page + 1))
|
||
layout.addWidget(self.next)
|
||
|
||
# The business contract fixes this list to one page size. A label is
|
||
# intentionally used instead of a one-option combo box so the control
|
||
# does not advertise an interaction that cannot change anything.
|
||
self.page_size_label = QLabel(f"{page_size} 条/页", self)
|
||
self.page_size_label.setProperty("pagerSize", True)
|
||
self.page_size_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
layout.addWidget(self.page_size_label)
|
||
|
||
self.page_label: QPushButton | None = None
|
||
self.update_state(1, 0)
|
||
|
||
@property
|
||
def page_count(self) -> int:
|
||
return max(1, (self.total + self.page_size - 1) // self.page_size)
|
||
|
||
def update_state(self, page: int, total: int) -> None:
|
||
self.total = max(0, total)
|
||
self.page = min(max(1, page), self.page_count)
|
||
self.summary.setText(f"共 {self.total} 条")
|
||
while self.pages_layout.count():
|
||
item = self.pages_layout.takeAt(0)
|
||
widget = item.widget()
|
||
if widget is not None:
|
||
widget.deleteLater()
|
||
|
||
count = self.page_count
|
||
if count <= 4:
|
||
pages: list[int | None] = list(range(1, count + 1))
|
||
elif self.page <= 3:
|
||
pages = [1, 2, 3, None, count]
|
||
elif self.page >= count - 2:
|
||
pages = [1, None, count - 2, count - 1, count]
|
||
else:
|
||
pages = [1, None, self.page, None, count]
|
||
|
||
self.page_label = None
|
||
for number in pages:
|
||
if number is None:
|
||
ellipsis = QLabel("…", self.pages_host)
|
||
ellipsis.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
ellipsis.setFixedWidth(24)
|
||
self.pages_layout.addWidget(ellipsis)
|
||
continue
|
||
button = QPushButton(str(number), self.pages_host)
|
||
button.setProperty("pagerPage", True)
|
||
button.setProperty("active", number == self.page)
|
||
button.setCursor(Qt.CursorShape.PointingHandCursor)
|
||
button.clicked.connect(lambda _checked=False, value=number: self._request(value))
|
||
self.pages_layout.addWidget(button)
|
||
if number == self.page:
|
||
self.page_label = button
|
||
self.previous.setEnabled(self.page > 1)
|
||
self.next.setEnabled(self.page < count)
|
||
|
||
def _request(self, page: int) -> None:
|
||
if 1 <= page <= self.page_count and page != self.page:
|
||
self.page_changed.emit(page)
|
||
|
||
|
||
class Pager(BusinessPager):
|
||
"""Numbered pager for the appointment/patient lists.
|
||
|
||
This used to be a separate ``上一页 / 1 / 1 / 下一页`` control, so the
|
||
workstation shipped three different pagination footers depending on which
|
||
list you opened. It now reuses :class:`BusinessPager` verbatim and only
|
||
keeps its own object name and default page size, giving every list the same
|
||
footer while leaving the ``page_changed`` / ``update_state`` API unchanged.
|
||
"""
|
||
|
||
def __init__(self, page_size: int = 20, parent: QWidget | None = None) -> None:
|
||
super().__init__(page_size, parent)
|
||
self.setObjectName("Pager")
|
||
|
||
|
||
def card_layout(card: QFrame, margins: int = 18, spacing: int = 12) -> QVBoxLayout:
|
||
layout = QVBoxLayout(card)
|
||
layout.setContentsMargins(margins, margins, margins, margins)
|
||
layout.setSpacing(spacing)
|
||
return layout
|
||
|
||
|
||
def section_title(text: str, trailing: QWidget | None = None) -> QWidget:
|
||
container = QWidget()
|
||
layout = QHBoxLayout(container)
|
||
layout.setContentsMargins(0, 0, 0, 0)
|
||
title = QLabel(text)
|
||
title.setProperty("role", "sectionTitle")
|
||
layout.addWidget(title)
|
||
layout.addStretch(1)
|
||
if trailing is not None:
|
||
layout.addWidget(trailing)
|
||
return container
|
||
|
||
|
||
def clear_layout(layout: QVBoxLayout | QHBoxLayout) -> None:
|
||
while layout.count():
|
||
item = layout.takeAt(0)
|
||
widget = item.widget()
|
||
child_layout = item.layout()
|
||
if widget is not None:
|
||
widget.deleteLater()
|
||
elif child_layout is not None:
|
||
clear_layout(child_layout) # type: ignore[arg-type]
|
||
|
||
|
||
__all__ = [
|
||
"BusinessPager",
|
||
"BusyOverlay",
|
||
"EmptyState",
|
||
"MessageBanner",
|
||
"MetricCard",
|
||
"OverlayHost",
|
||
"PageHeader",
|
||
"Pager",
|
||
"SortableTable",
|
||
"StatusBadge",
|
||
"TableColumn",
|
||
"Toast",
|
||
"Worker",
|
||
"card_layout",
|
||
"clear_layout",
|
||
"display_text",
|
||
"first_value",
|
||
"format_record_time",
|
||
"friendly_error",
|
||
"get_value",
|
||
"has_permission",
|
||
"invoke",
|
||
"page_items",
|
||
"page_total",
|
||
"run_async",
|
||
"section_title",
|
||
"set_authentication_expired_handler",
|
||
"show_toast",
|
||
]
|