This commit is contained in:
Your Name
2026-10-17 16:51:48 +08:00
parent 56f85f7d89
commit ba2e2e4087
41 changed files with 2851 additions and 96 deletions
+6 -3
View File
@@ -89,7 +89,7 @@ def example_batch() -> dict[str, Any]:
"algorithm_version": "prescription-soft-dice-v1.1.0",
"prompt_version": "manual-prescription-required-candidate-v4",
}
return {"id": 40, "prescription_id": 7556, "patient_id": 1391, "diagnosis_id": 1391,
return {"id": 40, "prescription_id": 7556, "patient_id": 1391, "diagnosis_id": 1391, "agreement_visible": True,
"prescription_revision": 1, "status": "success", "validity": "current",
"comparison_type": "non_independent", "models": models, "coverage_status": "partial",
"created_at": "2026-09-10 15:29:00", "cutoff_at": "2026-09-10 15:29:00",
@@ -143,8 +143,11 @@ class PreviewRepository:
def __init__(self) -> None:
self.batch = example_batch()
def get_prescription_ai_capabilities(self) -> dict[str, Any]:
return {"agreement_visible": True, "can_manage_visibility": False}
def list_prescription_ai_reports(self, **_params: Any) -> dict[str, Any]:
return {"enabled": True, "lists": deepcopy(self.history()), "count": len(self.history())}
return {"agreement_visible": True, "enabled": True, "lists": deepcopy(self.history()), "count": len(self.history())}
def history(self) -> list[dict[str, Any]]:
"""The current batch plus the earlier ones it supersedes, newest first."""
@@ -170,7 +173,7 @@ class PreviewRepository:
"algorithm_version": "prescription-soft-dice-v1.0.0"},
"openai": {"comparison": {"status": "not_comparable"}}}},
]
return [self.batch, *older]
return [{**row, "agreement_visible": True} for row in [self.batch, *older]]
def get_prescription_ai_report(self, _batch_id: int) -> dict[str, Any]:
return deepcopy(self.batch)
@@ -0,0 +1,111 @@
"""Capture agreement authorization states using synthetic data and no network."""
from __future__ import annotations
import os
from copy import deepcopy
from pathlib import Path
from typing import Any
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication
from render_issued_prescription_ai import PreviewRepository
from doctor_workstation.services import DemoDoctorRepository
from doctor_workstation.ui.dialogs import issued_prescription_ai as ai
from doctor_workstation.ui.pages import prescriptions
from doctor_workstation.ui.theme import apply_theme
class VisibilityPreviewRepository(DemoDoctorRepository, PreviewRepository):
def __init__(self) -> None:
DemoDoctorRepository.__init__(self)
PreviewRepository.__init__(self)
self.visible = True
def get_prescription_ai_capabilities(self) -> dict[str, Any]:
return {"agreement_visible": self.visible, "can_manage_visibility": True}
def list_prescription_ai_statuses(self, ids: list[int]) -> dict[str, Any]:
return {"enabled": True, "agreement_visible": self.visible, "items": [
{"prescription_id": value, "agreement_visible": self.visible, "status": "success",
"models": {"qwen": {"status": "success", "score": 68.5},
"openai": {"status": "success", "score": 55.2}}} for value in ids]}
def list_prescription_ai_reports(self, **_params: Any) -> dict[str, Any]:
rows = [{**row, "agreement_visible": self.visible} for row in self.history()]
return {"agreement_visible": self.visible, "enabled": True, "lists": deepcopy(rows), "count": len(rows)}
def get_prescription_ai_report(self, _batch_id: int) -> dict[str, Any]:
return {**deepcopy(self.batch), "agreement_visible": self.visible}
def main() -> None:
application = QApplication.instance() or QApplication([])
apply_theme(application)
output = Path(__file__).resolve().parents[2] / "artifacts" / "prescription-ai-visibility"
output.mkdir(parents=True, exist_ok=True)
def immediate(function: Any, **callbacks: Any) -> None:
try:
result = function()
except Exception as error:
if callbacks.get("on_error"):
callbacks["on_error"](error)
else:
raise
else:
if callbacks.get("on_success"):
callbacks["on_success"](result)
finally:
if callbacks.get("on_finished"):
callbacks["on_finished"]()
ai.run_async = immediate
prescriptions.run_async = immediate
repository = VisibilityPreviewRepository()
def capture(widget: Any, filename: str) -> None:
for _ in range(4):
application.processEvents()
assert widget.grab().save(str(output / filename))
page = prescriptions.PrescriptionsPage(repository, ["*"], {"name": "演示医生", "root": 1})
page.resize(1366, 800)
page.show()
capture(page, "list-authorized.png")
repository.visible = False
page.refresh()
capture(page, "list-restricted.png")
page.close()
repository.visible = True
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=7556,
current_user={"name": "演示医生", "root": 1})
dialog.resize(1280, 860)
dialog.show()
capture(dialog, "report-authorized.png")
repository.visible = False
repository.batch["status"] = "partial"
repository.batch["models"]["openai"].update(
status="failed", report=None, candidate=None, comparison=None,
error_code="upstream_timeout", error_message="演示:模型连接超时,可重试。")
dialog.refresh()
dialog._goto("report")
capture(dialog, "report-restricted.png")
dialog._goto("overview")
capture(dialog, "review-restricted.png")
dialog._goto("per_herb")
capture(dialog, "candidate-restricted.png")
dialog.close()
statistics = ai.PrescriptionAiStatisticsDialog(repository, ["*"])
statistics.resize(1180, 860)
statistics.show()
capture(statistics, "statistics-restricted.png")
statistics.close()
print(output)
if __name__ == "__main__":
main()
@@ -235,6 +235,9 @@ class DoctorRepository(Protocol):
def get_prescription(self, prescription_id: int) -> Prescription:
"""Return one issued prescription."""
def get_prescription_ai_capabilities(self) -> dict[str, Any]:
"""Read the server's current agreement visibility decision."""
def list_prescription_ai_statuses(self, ids: list[int]) -> dict[str, Any]:
"""Read cached dual-model states for up to 100 prescriptions."""
@@ -1611,6 +1614,9 @@ class RemoteDoctorRepository:
payload = _client_request(self.client, "post" if mutation else "get", endpoint, params, timeout=30.0)
return dict(_require_mapping(payload, endpoint))
def get_prescription_ai_capabilities(self) -> dict[str, Any]:
return self._prescription_ai_request("capabilities", {})
def list_prescription_ai_statuses(self, ids: list[int]) -> dict[str, Any]:
if len(ids) > 100 or any(int(value) <= 0 for value in ids):
raise ValueError("statuses requires at most 100 positive prescription IDs")
@@ -3,6 +3,7 @@
from __future__ import annotations
from collections.abc import Mapping
from copy import deepcopy
from html import escape
from math import isfinite
from time import monotonic
@@ -402,6 +403,28 @@ def can_open_issued_ai(permissions: Any) -> bool:
return has_permission(permissions, "tcm.prescriptionAi/reports", default=False) and has_permission(permissions, "tcm.prescriptionAi/detail", default=False)
def agreement_safe_batch(value: Any, visible: bool) -> dict[str, Any]:
"""Discard restricted structured comparisons before retaining any UI cache."""
data = deepcopy(mapping(value))
if visible:
return data
fields = {"comparison", "score", "herb_score", "comparison_status", "comparison_reason_code", "comparison_type",
"algorithm_version", "normalization_version", "denominator", "dictionary_version", "dictionary_versions",
"baseline_eligible", "baseline_exclusion_reasons", "baseline_exclusions_json"}
for item in (data, *mapping(data.get("models")).values()):
if isinstance(item, dict):
for field in fields:
item.pop(field, None)
for model in mapping(data.get("models")).values():
if isinstance(model, dict) and not model.get("error_code"):
# Successful model reasons describe the comparison, while failed
# task reasons explain generation and remain useful with reports.
model.pop("reason", None)
data["agreement_visible"] = False
return data
def _html(value: Any, field: str = "") -> str:
"""Escape all stored/model content, including source URLs, before rendering."""
if isinstance(value, Mapping):
@@ -536,6 +559,8 @@ class IssuedPrescriptionAiDialog(QDialog):
self._progress_frozen_seconds = 0
self._progress_stale = False
self._read_access_denied = False
self._agreement_visible = False
self._history_agreement_visible = False
self.setWindowTitle("诊断与药方 · AI 分析报告")
self.resize(1440, 940)
self.setMinimumSize(940, 640)
@@ -575,7 +600,7 @@ class IssuedPrescriptionAiDialog(QDialog):
identity_block.addWidget(self.breadcrumb)
title_row = QHBoxLayout()
title_row.setSpacing(10)
heading = QLabel("诊断与药方对照", top)
heading = self.heading = QLabel("诊断与药方对照", top)
heading.setObjectName("AiHeroTitle")
title_row.addWidget(heading)
self.chip_row = QHBoxLayout()
@@ -903,7 +928,7 @@ class IssuedPrescriptionAiDialog(QDialog):
footer_text = QVBoxLayout()
footer_text.setSpacing(4)
footer_text.addWidget(self.message)
note = QLabel(DISCLAIMER)
note = self.disclaimer = QLabel(DISCLAIMER)
note.setObjectName("AiFootnote")
note.setWordWrap(True)
footer_text.addWidget(note)
@@ -921,6 +946,7 @@ class IssuedPrescriptionAiDialog(QDialog):
self._progress_timer.timeout.connect(self._render_progress)
self._sync_actions()
self._navigation_changed(self.tabs.currentIndex())
self._apply_agreement_visibility(self._agreement_visible)
REPORT_MODES = (("both", "并排对照"), ("qwen", "仅千问"), ("openai", "仅 OpenAI"),
("differences", "只看不同段落"))
@@ -1063,6 +1089,8 @@ class IssuedPrescriptionAiDialog(QDialog):
return host
def _goto(self, key: str) -> None:
if not self._agreement_visible:
key = "candidate" if key == "per_herb" else "report" if key in {"comparison_raw", "history"} else key
page = self.tab_pages.get(key)
if page is not None:
self.tabs.setCurrentWidget(page)
@@ -1070,7 +1098,10 @@ class IssuedPrescriptionAiDialog(QDialog):
def _navigation_changed(self, index: int) -> None:
widget = self.tabs.widget(index)
current = next((key for key, page in self.tab_pages.items() if page is widget), "")
self.rail.set_current(current)
if not self._agreement_visible and current in {"per_herb", "comparison_raw", "history"}:
self._goto("report")
return
self.rail.set_current("per_herb" if current == "candidate" and not self._agreement_visible else current)
self._current_page = current
self._apply_density()
@@ -1101,6 +1132,36 @@ class IssuedPrescriptionAiDialog(QDialog):
if field in MODELS:
self.model_views[field]["report"].setFocus()
def _apply_agreement_visibility(self, visible: bool) -> None:
"""One server decision controls every comparison surface, without losing review."""
self._agreement_visible = visible
self.agreement.set_agreement_visible(visible)
self.agreement_note.setVisible(visible)
self.heading.setText("诊断与药方对照" if visible else "诊断与药方报告")
self.disclaimer.setText(DISCLAIMER if visible else "候选建议仅供医生复核。")
self.comparison_panel.set_agreement_visible(visible)
self.rail.names["overview"].setText("对比总览" if visible else "报告复核")
self.rail.names["per_herb"].setText("候选与逐味" if visible else "AI 候选处方")
self.nav_buttons["history"].setVisible(visible)
for key in ("differences", "consensus"):
self.rail.focus_values[key].parentWidget().setVisible(visible)
if not visible:
self._batch = agreement_safe_batch(self._batch, False) if self._batch else {}
self._history_scope = {key: agreement_safe_batch(row, False) for key, row in self._history_scope.items()}
self.history_page.set_history([])
self.candidates_page.set_batch({})
for views in self.model_views.values():
views["comparison"].clear()
self.comparison_panel.set_batch(self._batch)
self.rail.set_focus({})
self.rail.badges["overview"].clear()
self.rail.badges["per_herb"].clear()
self.rail.badges["overview"].hide()
self.rail.badges["per_herb"].hide()
if self._current_page in {"per_herb", "comparison_raw", "history"}:
self._goto("report")
@property
def save_review_button(self) -> QPushButton:
"""The review card's own save action for the model currently being annotated."""
@@ -1206,9 +1267,12 @@ class IssuedPrescriptionAiDialog(QDialog):
return
self._read_access_denied = False
data = mapping(result)
self._history_agreement_visible = data.get("agreement_visible") is True
self._apply_agreement_visibility(self._history_agreement_visible)
self._feature_enabled = data.get("enabled", True) is not False
rows = data.get("lists") or []
self.history_page.set_history([mapping(row) for row in rows])
rows = [agreement_safe_batch(row, self._history_agreement_visible and mapping(row).get("agreement_visible") is True)
for row in data.get("lists") or []]
self.history_page.set_history(rows if self._agreement_visible else [])
self.history.blockSignals(True)
self.history.clear()
self._history_scope = {}
@@ -1321,6 +1385,8 @@ class IssuedPrescriptionAiDialog(QDialog):
def _read_failed(self, error: Exception, prefix: str) -> None:
self._timer.stop()
self._history_agreement_visible = False
self._apply_agreement_visibility(False)
text = str(error).lower()
restricted = isinstance(error, (ValueError, AuthenticationExpiredError)) or getattr(error, "status_code", None) in {401, 403, 404} or any(
marker in text for marker in ("forbidden", "unauthorized", "permission", "access_revoked", "not found", "权限", "无权", "登录", "归属", "不存在")
@@ -1354,6 +1420,10 @@ class IssuedPrescriptionAiDialog(QDialog):
if int(batch.get("id") or 0) != batch_id or not identity_matches or (self.prescription_id and int(batch.get("prescription_id") or 0) != self.prescription_id):
self._detail_error(ValueError("报告归属与当前处方或诊单不一致"), generation)
return
visible = (self._history_agreement_visible and batch.get("agreement_visible") is True
and mapping(expected).get("agreement_visible") is True)
self._apply_agreement_visibility(visible)
batch = agreement_safe_batch(batch, visible)
first_load = not self._batch
self._batch = batch
self._read_access_denied = False
@@ -1368,7 +1438,7 @@ class IssuedPrescriptionAiDialog(QDialog):
self.identity.setText(f"处方 {batch.get('prescription_id')} · 诊单 {batch.get('diagnosis_id')} · 版本 {batch.get('prescription_revision', '—')}")
self._set_chips([
(label(batch.get("validity")), "info" if current_batch(batch) else "warn"),
(label(batch.get("comparison_type")), "neutral"),
(label(batch.get("comparison_type")) if self._agreement_visible else "", "neutral"),
])
self.batch_summary.setText(f"{compact_state} {_reason(batch.get('error_message') or batch.get('error_code'))}\n资料截止:{format_record_time(batch.get('cutoff_at'))} · 批次建立:{format_record_time(batch.get('created_at'))}")
self._render_facts(batch)
@@ -1411,9 +1481,11 @@ class IssuedPrescriptionAiDialog(QDialog):
"risk_assessment": report.get("risk_assessment"),
}
_set_html(views["candidate"], _candidate_html(candidate_content))
comparison_html = f"<p><b>药味与剂量一致度:{percentage(score, 1)} 纯药味重合度:{percentage(comparison.get('herb_score'), 1)}</b></p>"
comparison_html += "<p class=sec>逐味贡献</p>" + _html(_comparison_rows(comparison.get("rows")) or "没有可展示的逐味对照。")
comparison_html += _html({field: value for field, value in comparison.items() if field not in {"score", "herb_score", "rows"}})
comparison_html = ""
if self._agreement_visible:
comparison_html = f"<p><b>药味与剂量一致度:{percentage(score, 1)} 纯药味重合度:{percentage(comparison.get('herb_score'), 1)}</b></p>"
comparison_html += "<p class=sec>逐味贡献</p>" + _html(_comparison_rows(comparison.get("rows")) or "没有可展示的逐味对照。")
comparison_html += _html({field: value for field, value in comparison.items() if field not in {"score", "herb_score", "rows"}})
_set_html(views["comparison"], comparison_html)
source_html = _html({"source_summary": batch.get("source_summary"), "missing": batch.get("missing"), "coverage": model.get("coverage"), "baseline_eligible": batch.get("baseline_eligible"), "baseline_exclusion_reasons": batch.get("baseline_exclusion_reasons"), "cutoff_at": format_record_time(batch.get("cutoff_at")), "generated_at": format_record_time(model.get("generated_at"))})
reference_lines = [line for line in report_provenance_lines(model)
@@ -1428,7 +1500,7 @@ class IssuedPrescriptionAiDialog(QDialog):
views["comment"].setPlainText(str(review.get("comment") or ""))
index = views["review_state"].findData(review.get("status"))
views["review_state"].setCurrentIndex(max(0, index))
self.candidates_page.set_batch(batch)
self.candidates_page.set_batch(batch if self._agreement_visible else {})
self.sources_page.set_batch(batch)
self.pipeline_page.set_batch(batch)
self._sync_failures(batch)
@@ -1488,6 +1560,12 @@ class IssuedPrescriptionAiDialog(QDialog):
"restricted": panel.restricted_count(data) if hasattr(panel, "restricted_count") else None,
"consensus": panel.consensus_count() if hasattr(panel, "consensus_count") else None,
})
if not self._agreement_visible:
for key in ("differences", "consensus"):
self.rail.focus_values[key].clear()
for key in ("overview", "per_herb"):
self.rail.badges[key].clear()
self.rail.badges[key].hide()
def _render_facts(self, batch: Any) -> None:
snapshot = mapping(mapping(batch).get("doctor_snapshot"))
@@ -1707,6 +1785,7 @@ class PrescriptionAiStatisticsDialog(QDialog):
super().__init__(parent)
self.repository, self.permissions = repository, permissions
self._generation = 0
self._agreement_visible = False
self.setWindowTitle("医生处方 AI 一致度统计")
self.setStyleSheet(dialog_style() + rail_qss())
self.resize(1180, 860)
@@ -1742,6 +1821,7 @@ class PrescriptionAiStatisticsDialog(QDialog):
self.report.document().setDefaultStyleSheet(_report_css())
self.report.setMinimumHeight(130)
root.addWidget(self.report, 1)
self.panel.hide()
def showEvent(self, event: Any) -> None:
super().showEvent(event)
@@ -1750,9 +1830,19 @@ class PrescriptionAiStatisticsDialog(QDialog):
def hideEvent(self, event: Any) -> None:
self._generation += 1
self._clear_statistics()
super().hideEvent(event)
def _clear_statistics(self) -> None:
self._agreement_visible = False
self.panel.set_statistics({})
self.panel.hide()
self.report.clear()
def refresh(self) -> None:
self._generation += 1
generation = self._generation
self._clear_statistics()
if not has_permission(self.permissions, "tcm.prescriptionAi/statistics", default=False):
self.summary.setText("没有查看医生统计的权限。")
return
@@ -1763,20 +1853,33 @@ class PrescriptionAiStatisticsDialog(QDialog):
if doctor and (not doctor.isdecimal() or int(doctor) <= 0):
self.summary.setText("请填写有效的医生编号。")
return
self._generation += 1
generation = self._generation
self.summary.setText("正在读取统计…")
self.report.clear()
self.summary.setText("正在核验一致度查看权限…")
method = getattr(self.repository, "get_prescription_ai_capabilities", None)
if not callable(method):
self.summary.setText("当前账号无权查看一致度统计。")
return
date_from, date_to = self.date_from.date().toString("yyyy-MM-dd"), self.date_to.date().toString("yyyy-MM-dd")
run_async(lambda: self.repository.prescription_ai_statistics(date_from, date_to, int(doctor) if doctor else None), on_success=lambda result: self._ready(result, generation), on_error=lambda error: self._error(error, generation))
run_async(method, on_success=lambda result: self._capabilities_ready(result, generation, date_from, date_to, int(doctor) if doctor else None),
on_error=lambda error: self._error(error, generation))
def _capabilities_ready(self, result: Any, generation: int, date_from: str, date_to: str, doctor: int | None) -> None:
if generation != self._generation:
return
if mapping(result).get("agreement_visible") is not True or not has_permission(self.permissions, "tcm.prescriptionAi/statistics", default=False):
self._clear_statistics()
self.summary.setText("当前账号无权查看一致度统计。")
return
self._agreement_visible = True
self.summary.setText("正在读取统计…")
run_async(lambda: self.repository.prescription_ai_statistics(date_from, date_to, doctor), on_success=lambda result: self._ready(result, generation), on_error=lambda error: self._error(error, generation))
def _error(self, error: Exception, generation: int) -> None:
if generation == self._generation:
self.summary.setText("统计暂不可用:" + _error_text(error))
self.report.clear()
self._clear_statistics()
def _ready(self, result: Any, generation: int) -> None:
if generation != self._generation:
if generation != self._generation or not self._agreement_visible:
return
data = mapping(result)
if data.get("enabled") is False:
@@ -1786,6 +1889,7 @@ class PrescriptionAiStatisticsDialog(QDialog):
return
self.summary.setText(f"范围内开方事件:{label(data.get('total_count'))} · 患者数:{label(data.get('patient_count'))}。小样本应谨慎解读,未按分数排名。")
self.panel.set_statistics(data)
self.panel.show()
sections = []
for row in data.get("doctors") or []:
doctor = mapping(row)
@@ -190,6 +190,8 @@ class AgreementBar(QFrame):
layout = QHBoxLayout(self)
layout.setContentsMargins(22, 16, 22, 16)
layout.setSpacing(24)
self._agreement_visible = True
self._metric_chrome: list[QWidget] = []
self.blocks: dict[str, dict[str, Any]] = {}
for index, key in enumerate(("qwen", "openai")):
if index:
@@ -197,11 +199,14 @@ class AgreementBar(QFrame):
divider.setObjectName("AiFactDivider")
divider.setFixedWidth(1)
layout.addWidget(divider)
layout.addWidget(self._delta_block())
delta = self._delta_block()
layout.addWidget(delta)
self._metric_chrome.extend((divider, delta))
divider = QFrame(self)
divider.setObjectName("AiFactDivider")
divider.setFixedWidth(1)
layout.addWidget(divider)
self._metric_chrome.append(divider)
# Both heads line up at the top, so a model that failed does not slide its column down.
layout.addWidget(self._model_block(key), 1, Qt.AlignmentFlag.AlignTop)
@@ -294,7 +299,7 @@ class AgreementBar(QFrame):
carrier.setFixedHeight(4)
carrier.setAccessibleName(f"{MODEL_NAMES[key]}与医生方的药味及剂量一致度")
carrier.setVisible(False)
self.blocks[key] = {"score": score, "coverage": coverage, "elapsed": elapsed, "track": track,
self.blocks[key] = {"score": score, "caption": caption, "coverage": coverage, "elapsed": elapsed, "track": track,
"status": status, "failure": failure, "failure_text": failure_text,
"retry_slot": retry_slot, "stage": stage, "carrier": carrier}
return holder
@@ -313,6 +318,30 @@ class AgreementBar(QFrame):
self.blocks[key]["retry_slot"].addWidget(button)
def set_agreement_visible(self, visible: bool) -> None:
"""Keep model states and retries available when the metric is restricted."""
self._agreement_visible = visible
for widget in self._metric_chrome:
widget.setVisible(visible)
for key, block in self.blocks.items():
for field in ("score", "caption", "track"):
block[field].setVisible(visible)
if visible:
block["carrier"].setAccessibleName(f"{MODEL_NAMES[key]}与医生方的药味及剂量一致度")
if not visible:
block["score"].clear()
block["score"].setToolTip("")
block["track"].set_values(None, None, "")
block["track"].setAccessibleDescription("")
block["carrier"].setValue(0)
block["carrier"].setToolTip("")
block["carrier"].setAccessibleName("")
if not visible:
self.delta.clear()
self.delta_note.clear()
self.overlap.clear()
def set_failure(self, key: str, message: str, retryable: bool) -> None:
block = self.blocks[key]
block["failure_text"].setText(message)
@@ -351,7 +380,7 @@ class AgreementBar(QFrame):
elapsed: Mapping[str, str] | None = None) -> None:
data = _mapping(batch)
models = _mapping(data.get("models"))
scores = {key: _score(_mapping(models.get(key))) for key in ("qwen", "openai")}
scores = {key: _score(_mapping(models.get(key))) if self._agreement_visible else None for key in ("qwen", "openai")}
for key in ("qwen", "openai"):
block = self.blocks[key]
value = scores[key]
@@ -364,6 +393,9 @@ class AgreementBar(QFrame):
spent = _mapping(elapsed).get(key, "")
block["elapsed"].setText(spent)
block["elapsed"].setVisible(bool(spent))
if not self._agreement_visible:
self.set_agreement_visible(False)
return
if scores["qwen"] is None or scores["openai"] is None:
self.delta.setText("—")
self.delta_note.setText("两个模型都可比后才给差值")
@@ -537,6 +537,9 @@ class DoseDiffPanel(QFrame):
item = self.body_layout.takeAt(0)
widget = item.widget()
if widget is not None:
widget.hide()
for label in widget.findChildren(QLabel):
label.clear()
widget.setParent(None)
widget.deleteLater()
extent = _axis_extent(deltas)
@@ -620,6 +623,9 @@ class ConclusionsPanel(QFrame):
entry = self.body_layout.takeAt(0)
widget = entry.widget()
if widget is not None:
widget.hide()
for label in widget.findChildren(QLabel):
label.clear()
widget.setParent(None)
widget.deleteLater()
for index, text in enumerate(items, start=1):
@@ -844,6 +850,9 @@ class RiskPanel(QFrame):
entry = self.body_layout.takeAt(0)
widget = entry.widget()
if widget is not None:
widget.hide()
for label in widget.findChildren(QLabel):
label.clear()
widget.setParent(None)
widget.deleteLater()
flagged = []
@@ -1015,6 +1024,9 @@ class ChecklistPanel(QFrame):
entry = self.body_layout.takeAt(0)
widget = entry.widget()
if widget is not None:
widget.hide()
for label in widget.findChildren(QLabel):
label.clear()
widget.setParent(None)
widget.deleteLater()
grouped: dict[str, tuple[str, str, int, str]] = {}
@@ -1132,6 +1144,7 @@ class PrescriptionReviewWorkspace(QWidget):
self._batch: dict[str, Any] = {}
self.rows: list[ReviewDoseRow] = []
self._review_widget: QWidget | None = None
self._agreement_visible = True
self._summary = "选择报告后查看诊断与药方"
self.setStyleSheet(f"""
QWidget#PrescriptionReviewWorkspace {{ background: transparent; }}
@@ -1150,7 +1163,7 @@ class PrescriptionReviewWorkspace(QWidget):
page = QWidget(self.page_scroll)
self.page_scroll.setWidget(page)
outer.addWidget(self.page_scroll)
root = QVBoxLayout(page)
root = self._workspace_layout = QVBoxLayout(page)
root.setContentsMargins(0, 0, 6, 0)
root.setSpacing(14)
@@ -1201,6 +1214,10 @@ class PrescriptionReviewWorkspace(QWidget):
layout.takeAt(0)
for index in range(layout.columnCount()):
layout.setColumnStretch(index, 0)
if not self._agreement_visible:
self.first.addWidget(self.checklist, 0, 0)
self.first.setColumnStretch(0, 1)
return
cards = (self.conclusions, self.attribution, self.checklist)
if columns == 1:
for row, card in enumerate(cards):
@@ -1289,7 +1306,31 @@ class PrescriptionReviewWorkspace(QWidget):
self._batch = deepcopy(data)
self._render()
def set_agreement_visible(self, visible: bool) -> None:
"""Use the existing checklist as a report-only review workspace."""
if visible == self._agreement_visible:
return
self._agreement_visible = visible
for widget in (self.conclusions, self.attribution, self.doses, self.risks, self.links["comparison"]):
widget.setVisible(visible)
self._workspace_layout.setStretch(0, 5 if visible else 1)
self._workspace_layout.setStretch(1, 4 if visible else 0)
self._columns = 0
self._reflow(1 if self.width() < self.STACK_BELOW else 3)
self._render()
def _render(self) -> None:
if not self._agreement_visible:
self.rows = []
self.doses.set_rows([])
self.risks.set_rows([])
self.attribution.apply({})
self.conclusions.set_items([])
self.checklist.set_items(checklist_items(self._batch))
self._summary = "请结合报告与资料缺口记录复核意见。"
self.summary_changed.emit(self._summary)
return
self.rows, _states = review_rows(self._batch)
deltas = dose_deltas(self.rows)
self.doses.set_rows(deltas)
@@ -52,6 +52,7 @@ from PySide6.QtWidgets import (
from .. import icons, motion
from ..dialogs.issued_prescription_ai import (
PrescriptionAiStatisticsDialog,
agreement_safe_batch,
agreement_text,
batch_running,
can_open_issued_ai,
@@ -1007,6 +1008,10 @@ class PrescriptionsPage(QWidget):
self._mutation_pending = False
self._ai_statuses: dict[int, dict[str, Any]] = {}
self._ai_enabled = False
self._agreement_visible = False
self._capability_agreement_visible = False
self._ai_capability_token = 0
self._ai_capability_pending = False
self._ai_request_token = 0
self._ai_pending = False
self._ai_timer = QTimer(self)
@@ -1048,7 +1053,7 @@ class PrescriptionsPage(QWidget):
self.orders_button.clicked.connect(lambda: self._open_orders())
header.add_action(self.orders_button)
self.ai_statistics_button = QPushButton("AI 一致度统计", header)
self.ai_statistics_button.setVisible(has_permission(permissions, "tcm.prescriptionAi/statistics", default=False) and callable(getattr(repository, "prescription_ai_statistics", None)))
self.ai_statistics_button.hide()
self.ai_statistics_button.clicked.connect(self._open_ai_statistics)
header.add_action(self.ai_statistics_button)
self.add_button = QPushButton("新增处方", header)
@@ -1437,6 +1442,7 @@ class PrescriptionsPage(QWidget):
self._refresh_pending = False
self._generation += 1
generation = self._generation
self._load_ai_capabilities()
page_size = self._page_size
start_time, end_time = self._date_filters()
filters: dict[str, Any] = {
@@ -1656,6 +1662,7 @@ class PrescriptionsPage(QWidget):
return
method = getattr(self.repository, "list_prescription_ai_statuses", None)
if not has_permission(self.permissions, "tcm.prescriptionAi/statuses", default=False):
self._set_ai_columns(False)
self.ai_status_notice.setText("AI 分析:当前账号没有查看分析状态的权限,请联系管理员授权。")
return
if not callable(method):
@@ -1687,8 +1694,8 @@ class PrescriptionsPage(QWidget):
def _set_ai_columns(self, enabled: bool) -> None:
changed = enabled != self._ai_enabled
self._ai_enabled = enabled
for column in (11, 12):
self.table.setColumnHidden(column, not enabled)
self.table.setColumnHidden(11, not enabled)
self.table.setColumnHidden(12, not (enabled and self._agreement_visible))
if enabled:
header = self.table.horizontalHeader()
header.moveSection(header.visualIndex(11), header.visualIndex(4) + 1)
@@ -1699,10 +1706,54 @@ class PrescriptionsPage(QWidget):
if changed:
self._decorate_rows()
def _set_agreement_visible(self, visible: bool) -> None:
self._agreement_visible = visible
self.table.setColumnHidden(12, not (self._ai_enabled and visible))
self.ai_statistics_button.setVisible(
visible and has_permission(self.permissions, "tcm.prescriptionAi/statistics", default=False)
and callable(getattr(self.repository, "prescription_ai_statistics", None)))
if not visible:
self._ai_statuses = {key: agreement_safe_batch(row, False) for key, row in self._ai_statuses.items()}
for index in range(self.table.rowCount()):
item = self.table.item(index, 12)
if item is not None:
item.setText("")
item.setToolTip("")
item.setData(Qt.ItemDataRole.AccessibleTextRole, "")
def _load_ai_capabilities(self) -> None:
self._ai_capability_token += 1
token, generation = self._ai_capability_token, self._generation
self._ai_request_token += 1
self._ai_pending = False
self._capability_agreement_visible = False
self._set_agreement_visible(False)
method = getattr(self.repository, "get_prescription_ai_capabilities", None)
self._ai_capability_pending = callable(method)
if not callable(method):
return
run_async(method,
on_success=lambda result: self._ai_capabilities_ready(result, token, generation),
on_error=lambda _error: self._ai_capabilities_ready({}, token, generation))
def _ai_capabilities_ready(self, result: Any, token: int, generation: int) -> None:
if token != self._ai_capability_token or generation != self._generation:
return
self._ai_capability_pending = False
self._capability_agreement_visible = _row_mapping(result).get("agreement_visible") is True
self._set_agreement_visible(self._capability_agreement_visible)
if self._capability_agreement_visible:
# A previous deny discarded scores, including completed cached batches.
# Fetch them again after a fresh grant rather than displaying that cache.
self._ai_statuses.clear()
if self.isVisible() and not self._loading:
self._load_ai_statuses()
def _ai_statuses_ready(self, result: Any, ids: list[int], token: int, generation: int) -> None:
if token != self._ai_request_token or generation != self._generation or not self.isVisible():
return
data = _row_mapping(result)
self._set_agreement_visible(self._capability_agreement_visible and data.get("agreement_visible") is True)
self._set_ai_columns(data.get("enabled") is True)
if not self._ai_enabled:
self._ai_timer.stop()
@@ -1713,15 +1764,17 @@ class PrescriptionsPage(QWidget):
self.ai_status_notice.setToolTip("")
self.ai_report_button.setToolTip("自动分析未启用;仍可查看已保存的历史报告。")
return
self.ai_status_notice.setText("AI 分析已启用:保存手工处方后自动生成两份报告,并显示药味与剂量一致度。")
self.ai_status_notice.setText("AI 分析已启用:保存手工处方后自动生成两份报告。"
+ ("可查看药味与剂量一致度。" if self._agreement_visible else ""))
self.ai_status_notice.setToolTip("")
self.ai_report_button.setToolTip("查看两份 AI 报告、候选处方和逐味对照。")
self.ai_report_button.setToolTip("查看两份 AI 报告和候选处方。"
+ ("可查看逐味对照。" if self._agreement_visible else ""))
for value in ids:
self._ai_statuses[value] = {}
for batch in data.get("items") or []:
value = _int(get_value(batch, "prescription_id"), 0)
if value in ids:
self._ai_statuses[value] = _row_mapping(batch)
self._ai_statuses[value] = agreement_safe_batch(batch, self._agreement_visible and _row_mapping(batch).get("agreement_visible") is True)
self._render_ai_statuses()
if any(batch_running(self._ai_statuses.get(value)) for value in self._visible_ai_ids()):
self._ai_timer.start()
@@ -1741,13 +1794,13 @@ class PrescriptionsPage(QWidget):
if batch is None and not blank:
continue
state = "" if blank else (state_text(batch) if batch else "尚无分析记录")
agreement = "" if blank else agreement_text(batch)
agreement = "" if blank or not self._agreement_visible else agreement_text(batch)
for column, text in ((11, state), (12, agreement)):
item = self.table.item(index, column)
if item.text() != text:
item.setText(text)
changed = True
item.setToolTip("" if blank else status_tooltip(batch))
item.setToolTip("" if blank or column == 12 and not self._agreement_visible else status_tooltip(batch))
item.setData(Qt.ItemDataRole.AccessibleTextRole, text)
if changed:
self.table.resizeRowsToContents()
@@ -1757,6 +1810,7 @@ class PrescriptionsPage(QWidget):
def _ai_statuses_error(self, error: Exception, token: int, generation: int) -> None:
if token == self._ai_request_token and generation == self._generation:
self._ai_timer.stop()
self._set_agreement_visible(False)
self._set_ai_columns(False)
self.ai_status_notice.setText("AI 分析暂不可用:请稍后刷新;持续无法使用时,请联系管理员检查服务。")
self.ai_status_notice.setToolTip(friendly_error(error))
@@ -1781,13 +1835,13 @@ class PrescriptionsPage(QWidget):
if _is_blank_prescription(row):
return
menu = QMenu(self.table)
action = menu.addAction("AI 报告 / 逐味对照")
action = menu.addAction("AI 报告 / 逐味对照" if self._agreement_visible else "AI 报告")
action.triggered.connect(lambda: self._run_row_action(row, self._open_ai_report))
self._ai_context_menu = menu
menu.popup(self.table.viewport().mapToGlobal(position))
def _open_ai_statistics(self) -> None:
if not has_permission(self.permissions, "tcm.prescriptionAi/statistics", default=False):
if not self._agreement_visible or not has_permission(self.permissions, "tcm.prescriptionAi/statistics", default=False):
return
dialog = PrescriptionAiStatisticsDialog(self.repository, self.permissions, self)
dialog.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose, True)
@@ -2207,12 +2261,14 @@ class PrescriptionsPage(QWidget):
if self.table.rowCount() == 0 and not self._loading:
self.refresh()
elif not self._loading:
self._load_ai_statuses()
self._load_ai_capabilities()
def hideEvent(self, event: Any) -> None:
self._ai_timer.stop()
self._ai_scroll_timer.stop()
self._ai_request_token += 1
self._ai_capability_token += 1
self._ai_capability_pending = False
self._ai_pending = False
super().hideEvent(event)
+230 -4
View File
@@ -23,7 +23,7 @@ from doctor_workstation.ui.pages import prescriptions as page_module
def batch(batch_id: int = 40, prescription_id: int = 801, status: str = "running") -> dict[str, Any]:
return {
"id": batch_id, "prescription_id": prescription_id, "prescription_revision": 2,
"id": batch_id, "prescription_id": prescription_id, "prescription_revision": 2, "agreement_visible": True,
"patient_id": 301, "diagnosis_id": 501, "status": status,
"validity": "current", "comparison_type": "independent_baseline",
"source_summary": {"diagnosis_count": 2, "attachment_count": 4},
@@ -48,10 +48,14 @@ class Repository(DemoDoctorRepository):
self.calls: list[tuple[str, Any]] = []
self.batches = [batch()]
self.enabled = True
self.agreement_visible = True
def get_prescription_ai_capabilities(self) -> dict[str, Any]:
return {"agreement_visible": self.agreement_visible, "can_manage_visibility": False}
def list_prescription_ai_reports(self, **params: Any) -> dict[str, Any]:
self.calls.append(("reports", params))
return {"enabled": self.enabled, "lists": deepcopy(self.batches), "count": len(self.batches), "page_no": 1, "page_size": 20}
return {"agreement_visible": self.agreement_visible, "enabled": self.enabled, "lists": deepcopy(self.batches), "count": len(self.batches), "page_no": 1, "page_size": 20}
def get_prescription_ai_report(self, batch_id: int) -> dict[str, Any]:
self.calls.append(("detail", batch_id))
@@ -71,7 +75,7 @@ class Repository(DemoDoctorRepository):
def list_prescription_ai_statuses(self, ids: list[int]) -> dict[str, Any]:
self.calls.append(("statuses", list(ids)))
return {"enabled": self.enabled, "items": [{"prescription_id": value, "batch_id": 40, "status": "running", "validity": "current", "comparison_type": "independent_baseline", "models": {"qwen": {"status": "succeeded", "score": 0, "comparison_status": "comparable"}, "openai": {"status": "running", "score": None}}} for value in ids] if self.enabled else []}
return {"agreement_visible": self.agreement_visible, "enabled": self.enabled, "items": [{"agreement_visible": self.agreement_visible, "prescription_id": value, "batch_id": 40, "status": "running", "validity": "current", "comparison_type": "independent_baseline", "models": {"qwen": {"status": "succeeded", "score": 0, "comparison_status": "comparable"}, "openai": {"status": "running", "score": None}}} for value in ids] if self.enabled else []}
@pytest.fixture(scope="module")
@@ -595,6 +599,9 @@ def test_list_ignores_response_after_filter_generation_changes(application: QApp
def test_statistics_nulls_and_absent_review_samples(application: QApplication, immediate: None) -> None:
class StatisticsRepository:
def get_prescription_ai_capabilities(self) -> dict[str, Any]:
return {"agreement_visible": True}
def prescription_ai_statistics(self, *_args: Any) -> dict[str, Any]:
return {"total_count": 3, "patient_count": 2, "doctors": [{"doctor_name": "张医生", "total_count": 3, "patient_count": 2, "models": {"qwen": {"eligible_count": 0, "coverage_rate": 0, "mean": None, "median": None, "excluded_reasons": {"non_independent": 3}}, "openai": {"eligible_count": 1, "coverage_rate": 33.3, "mean": 0, "median": 0}}, "review": {"evaluated_count": 0, "qualified_count": 0, "qualified_rate": None}}]}
dialog = ai.PrescriptionAiStatisticsDialog(StatisticsRepository(), ["*"])
@@ -635,6 +642,7 @@ def test_repository_exact_read_and_mutation_contracts() -> None:
return {"saved": True}
client = Client()
repository = RemoteDoctorRepository(client)
repository.get_prescription_ai_capabilities()
assert repository.list_prescription_ai_statuses([801, 802, 801]) == {"enabled": False, "items": []}
repository.list_prescription_ai_reports(diagnosis_id=501, page_no=2, page_size=20)
repository.get_prescription_ai_report(40)
@@ -644,6 +652,7 @@ def test_repository_exact_read_and_mutation_contracts() -> None:
repository.review_prescription_ai(40, "qwen", "reviewed", "核对完成")
repository.prescription_ai_statistics("2026-09-01", "2026-09-09", 7)
assert client.calls == [
("GET", "tcm.prescriptionAi/capabilities", {}),
("GET", "tcm.prescriptionAi/statuses", {"ids": "801,802"}),
("GET", "tcm.prescriptionAi/reports", {"diagnosis_id": 501, "page_no": 2, "page_size": 20}),
("GET", "tcm.prescriptionAi/detail", {"batch_id": 40}),
@@ -660,6 +669,217 @@ def test_repository_exact_read_and_mutation_contracts() -> None:
repository.retry_prescription_ai(40, "other")
@pytest.mark.parametrize("flag", [False, None, 1, "true"])
def test_agreement_denied_or_missing_keeps_reports_review_and_retry(
application: QApplication, immediate: None, flag: Any,
) -> None:
repository = Repository()
repository.agreement_visible = flag
value = repository.batches[0]
value["agreement_visible"] = flag
value["models"]["openai"].update(status="failed", error_message="上游超时")
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], diagnosis_id=501,
current_user={"root": 1, "role_ids": [1]})
dialog.show()
application.processEvents()
assert not dialog._agreement_visible
assert "千问分析" in dialog.model_views["qwen"]["report"].toPlainText()
assert dialog.model_views["qwen"]["comment"].isEnabled()
assert dialog.model_views["qwen"]["save"].isVisible()
assert dialog.model_views["openai"]["retry"].isVisible()
assert dialog.model_views["openai"]["retry"].isEnabled()
assert dialog.history.count() == 1
assert dialog.comparison_panel.checklist.isVisible()
assert dialog.comparison_panel.conclusions.isHidden()
assert not dialog.comparison_panel.rows
assert not dialog.comparison_panel.doses.rows
assert dialog.nav_buttons["history"].isHidden()
assert dialog.agreement_note.isHidden()
for key, views in dialog.model_views.items():
assert not views["score"].text() and not views["score"].toolTip()
assert not views["agreement_bar"].toolTip() and not views["agreement_bar"].accessibleName()
assert views["gauge"]._value is None and not views["gauge"].accessibleDescription()
assert not views["comparison"].toPlainText()
assert "comparison" not in dialog._batch["models"][key]
assert not dialog.agreement.delta.text() and not dialog.agreement.overlap.text()
for key in ("differences", "consensus"):
assert not dialog.rail.focus_values[key].text()
assert dialog.rail.focus_values[key].parentWidget().isHidden()
dialog._goto("per_herb")
assert dialog.tabs.currentWidget() is dialog.tab_pages["candidate"]
assert "黄芪" in dialog.model_views["qwen"]["candidate"].toPlainText()
dialog._goto("comparison_raw")
assert dialog.tabs.currentWidget() is dialog.tab_pages["report"]
dialog._goto("history")
assert dialog.tabs.currentWidget() is dialog.tab_pages["report"]
dialog.close()
@pytest.mark.parametrize("missing", ["reports", "history_batch", "detail"])
def test_each_report_visibility_flag_is_required(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch, missing: str) -> None:
repository = Repository()
reports, detail = repository.list_prescription_ai_reports, repository.get_prescription_ai_report
def read_reports(**params: Any) -> dict[str, Any]:
data = reports(**params)
if missing == "reports":
data.pop("agreement_visible")
if missing == "history_batch":
data["lists"][0].pop("agreement_visible")
return data
def read_detail(batch_id: int) -> dict[str, Any]:
data = detail(batch_id)
if missing == "detail":
data.pop("agreement_visible")
return data
monkeypatch.setattr(repository, "list_prescription_ai_reports", read_reports)
monkeypatch.setattr(repository, "get_prescription_ai_report", read_detail)
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
dialog.show()
assert not dialog._agreement_visible
assert not dialog.model_views["qwen"]["score"].text()
assert "千问分析" in dialog.model_views["qwen"]["report"].toPlainText()
dialog.close()
def test_agreement_revocation_and_failed_read_clear_cached_views_and_survive_theme(
application: QApplication, immediate: None,
) -> None:
repository = Repository()
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
dialog.show()
assert dialog.model_views["qwen"]["score"].text() == "0.0%"
old_batch, old_generation = deepcopy(dialog._batch), dialog._generation
dialog.model_views["qwen"]["comment"].setPlainText("保留复核草稿")
repository.agreement_visible = False
dialog.refresh()
dialog._detail_ready(old_batch, old_generation, 40)
assert not dialog._agreement_visible
assert not dialog.model_views["qwen"]["score"].text()
assert dialog.history_page.table.rowCount() == 0
assert not dialog.history_page.chart._points
assert dialog.model_views["qwen"]["comment"].toPlainText() == "保留复核草稿"
dialog._switch_theme()
application.processEvents()
assert not dialog._agreement_visible and not dialog.model_views["qwen"]["score"].text()
assert dialog.model_views["qwen"]["comment"].toPlainText() == "保留复核草稿"
dialog._switch_theme()
repository.agreement_visible = True
dialog.refresh()
assert dialog._agreement_visible
dialog._detail_error(RuntimeError("offline"), dialog._generation)
assert not dialog._agreement_visible
assert not dialog.model_views["qwen"]["score"].text()
assert "comparison" not in dialog._batch["models"]["qwen"]
assert "千问分析" in dialog.model_views["qwen"]["report"].toPlainText()
dialog._detail_ready(old_batch, dialog._generation, 40)
assert not dialog._agreement_visible
dialog.close()
def test_revoked_agreement_cache_discards_flat_comparison_metadata() -> None:
value = batch()
sentinel = "previous-private-comparison"
value["baseline_exclusions_json"] = sentinel
value["models"]["qwen"].update({
"reason": sentinel, "comparison_reason_code": sentinel,
"dictionary_version": sentinel, "error_code": "",
})
value["models"]["openai"].update({
"error_code": "UPSTREAM_TIMEOUT", "reason": "模型请求超时",
})
sanitized = ai.agreement_safe_batch(value, False)
assert sentinel not in str(sanitized)
assert "千问分析" in str(sanitized["models"]["qwen"]["report"])
assert sanitized["models"]["openai"]["reason"] == "模型请求超时"
assert value["models"]["qwen"]["reason"] == sentinel
def test_list_capability_failure_revocation_and_late_status_cannot_restore_scores(
application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch,
) -> None:
repository = Repository()
page = page_module.PrescriptionsPage(repository, ["*"], {"root": 1})
page.resize(1366, 800)
page.show()
application.processEvents()
assert not page.table.isColumnHidden(12)
old_statuses = repository.list_prescription_ai_statuses([801, 802])
old_token = page._ai_request_token
old_capability_token = page._ai_capability_token
monkeypatch.setattr(repository, "get_prescription_ai_capabilities", lambda: (_ for _ in ()).throw(RuntimeError("offline")))
page._load_ai_capabilities()
page._ai_statuses_ready(old_statuses, [801, 802], old_token, page._generation)
page._ai_capabilities_ready({"agreement_visible": True}, old_capability_token, page._generation)
assert page.table.isColumnHidden(12) and not page.table.isColumnHidden(11)
assert page.ai_statistics_button.isHidden()
assert page.ai_report_button.isVisible()
assert page._ai_statuses
for index in range(page.table.rowCount()):
item = page.table.item(index, 12)
assert not item.text() and not item.toolTip() and not item.data(Qt.ItemDataRole.AccessibleTextRole)
assert all("score" not in model for row in page._ai_statuses.values() for model in row.get("models", {}).values())
page.close()
def test_list_rechecks_completed_scores_after_hide_and_show(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
repository = Repository()
original = repository.list_prescription_ai_statuses
def statuses(ids: list[int]) -> dict[str, Any]:
data = original(ids)
for row in data["items"]:
row["status"] = "success"
for model in row["models"].values():
model["status"] = "success"
return data
monkeypatch.setattr(repository, "list_prescription_ai_statuses", statuses)
page = page_module.PrescriptionsPage(repository, ["*"])
page.resize(1366, 800)
page.show()
application.processEvents()
assert "0%" in page.table.item(0, 12).text()
before = len([call for call in repository.calls if call[0] == "statuses"])
page.hide()
page.show()
application.processEvents()
assert len([call for call in repository.calls if call[0] == "statuses"]) > before
assert "0%" in page.table.item(0, 12).text()
page.close()
@pytest.mark.parametrize("flag", [False, None, 1, "true"])
def test_direct_statistics_requires_strict_server_capability(application: QApplication, immediate: None, flag: Any) -> None:
class StatisticsRepository:
calls = 0
def get_prescription_ai_capabilities(self) -> dict[str, Any]:
return {"agreement_visible": flag, "can_manage_visibility": True}
def prescription_ai_statistics(self, *_args: Any) -> dict[str, Any]:
self.calls += 1
return {}
repository = StatisticsRepository()
dialog = ai.PrescriptionAiStatisticsDialog(repository, ["*"])
dialog.show()
assert repository.calls == 0
assert dialog.panel.isHidden() and not dialog.report.toPlainText()
dialog.close()
def test_statistics_failed_capability_clears_old_charts_and_ignores_late_result(
application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch,
) -> None:
repository = ChineseStatisticsRepository()
dialog = ai.PrescriptionAiStatisticsDialog(repository, ["*"])
dialog.show()
assert dialog.panel.isVisible() and dialog.report.toPlainText()
old_generation = dialog._generation
monkeypatch.setattr(repository, "get_prescription_ai_capabilities", lambda: (_ for _ in ()).throw(RuntimeError("offline")))
dialog.refresh()
dialog._ready(repository.prescription_ai_statistics(), old_generation)
assert dialog.panel.isHidden() and not dialog.report.toPlainText()
assert not dialog.panel.distribution.has_data()
assert dialog.panel.doctors.rowCount() == 0
dialog.close()
def test_stale_list_score_is_not_attached_to_current_prescription() -> None:
value = {"validity": "superseded", "models": {"qwen": {"score": 100}, "openai": {"score": 90}}}
assert ai.agreement_text(value) == "千问 —\nOpenAI —"
@@ -773,7 +993,7 @@ def test_actual_history_pagination_contract(application: QApplication, immediate
def reports(**params: Any) -> dict[str, Any]:
repository.calls.append(("reports", params))
page = params["page_no"]
return {"lists": repository.batches[(page - 1) * 20:page * 20], "count": 21, "page_no": page, "page_size": 20}
return {"agreement_visible": True, "lists": repository.batches[(page - 1) * 20:page * 20], "count": 21, "page_no": page, "page_size": 20}
monkeypatch.setattr(repository, "list_prescription_ai_reports", reports)
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
dialog.show()
@@ -795,6 +1015,9 @@ def test_server_empty_batch_statuses_are_not_fake_scores(status: str, expected:
def test_statistics_displays_actual_version_strata_without_filling_null_mean(application: QApplication, immediate: None) -> None:
class StatisticsRepository:
def get_prescription_ai_capabilities(self) -> dict[str, Any]:
return {"agreement_visible": True}
def prescription_ai_statistics(self, *_args: Any) -> dict[str, Any]:
strata = [{"versions": {"model_version": version, "algorithm_version": "v1", "prompt_version": "p1", "dictionary_version": "d1"}, "count": 1, "mean": mean, "median": mean, "distribution": {"[0,20)": 1 if mean == 0 else 0}, "sample_status": "insufficient_sample"} for version, mean in (("model-1", 0), ("model-2", 80))]
return {"total_count": 2, "patient_count": 1, "doctors": [{"doctor_id": 7, "doctor_name": "张医生", "total_count": 2, "patient_count": 1, "paired_count": 0, "models": {"qwen": {"eligible_count": 2, "coverage_rate": 100, "mean": None, "median": None, "excluded_reasons": {}, "strata": strata}}, "review": {"evaluated_count": 0, "qualified_count": 0, "qualified_rate": None}}]}
@@ -965,6 +1188,9 @@ def test_internal_version_labels_keep_revision_numbers(version: str, expected: s
class ChineseStatisticsRepository:
def get_prescription_ai_capabilities(self) -> dict[str, Any]:
return {"agreement_visible": True}
def prescription_ai_statistics(self, *_args: Any) -> dict[str, Any]:
exclusions = {"SOURCE_HISTORY_VERSIONS_UNAVAILABLE": 3, "transcript_not_final": 1,
"ATTACHMENT_TARGET_PLAN_LEAKAGE_UNVERIFIED": 2, "future_reason_a": 4, "future_reason_b": 5}