3509 lines
133 KiB
Python
3509 lines
133 KiB
Python
"""Remote repository adapting the confirmed admin endpoints to domain models."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import mimetypes
|
|
import re
|
|
import time
|
|
from collections.abc import Callable, Iterator, Mapping, Sequence
|
|
from contextlib import suppress
|
|
from datetime import date
|
|
from inspect import Parameter, signature
|
|
from io import BytesIO
|
|
from os import PathLike
|
|
from pathlib import Path
|
|
from typing import Any, Final, Literal, Protocol
|
|
from uuid import uuid4
|
|
|
|
from doctor_workstation.core.errors import (
|
|
ApiBusinessError,
|
|
ApiHttpError,
|
|
ApiProtocolError,
|
|
ApiTransportError,
|
|
AuthenticationExpiredError,
|
|
)
|
|
from doctor_workstation.core.models import (
|
|
Appointment,
|
|
CallTicket,
|
|
Consultation,
|
|
PageResult,
|
|
Patient,
|
|
Prescription,
|
|
PrescriptionTemplate,
|
|
UserProfile,
|
|
)
|
|
from doctor_workstation.core.permissions import PermissionSet
|
|
from doctor_workstation.core.session import Session
|
|
|
|
from .api_client import ApiClient
|
|
from .token_store import TokenStore
|
|
|
|
AuditAction = Literal["approve", "reject"]
|
|
|
|
PRESCRIPTION_LIBRARY_PERMISSIONS: Final[dict[str, str]] = {
|
|
"create": "wcf.prescription/add",
|
|
"read": "wcf.prescription/read",
|
|
"update": "wcf.prescription/edit",
|
|
"delete": "wcf.prescription/delete",
|
|
"ai_reports": "tcm.prescriptionLibrary/aiReports",
|
|
"generate_ai_reports": "tcm.prescriptionLibrary/generateAiReports",
|
|
"edit_ai_report": "tcm.prescriptionLibrary/editAiReport",
|
|
}
|
|
"""Canonical permissions used by the routed prescription-library view."""
|
|
|
|
DIAGNOSIS_AI_PERMISSIONS: Final[dict[str, str]] = {
|
|
"ai_reports": "tcm.diagnosis/aiReports",
|
|
"generate_ai_reports": "tcm.diagnosis/generateAiReports",
|
|
"edit_ai_report": "tcm.diagnosis/editAiReport",
|
|
"analysis": "tcm.diagnosis/aiAnalysis",
|
|
"assistant": "tcm.diagnosis/aiAssistant",
|
|
}
|
|
"""Canonical permissions used by reception and patient-profile AI reports."""
|
|
|
|
PRESCRIPTION_PERMISSIONS: Final[dict[str, str]] = {
|
|
"create": "cf.prescription/add",
|
|
"read": "cf.prescription/read",
|
|
"update": "cf.prescription/edit",
|
|
"audit": "cf.prescription/audit",
|
|
"delete": "cf.prescription/del",
|
|
"patch_patient": "tcm.prescription/patchPatient",
|
|
"create_order": "tcm.prescriptionOrder/create",
|
|
"list_orders": "tcm.prescriptionOrder/lists",
|
|
"set_ship_mode": "tcm.prescriptionOrder/setShipMode",
|
|
"edit_extra_remark": "tcm.prescriptionOrder/editRemarkExtra",
|
|
}
|
|
"""Canonical permissions used by the routed issued-prescription view."""
|
|
|
|
|
|
class DoctorRepository(Protocol):
|
|
"""UI-facing contract shared by remote and fully in-memory repositories."""
|
|
|
|
def login(
|
|
self,
|
|
account: str,
|
|
password: str,
|
|
*,
|
|
remember_account: bool = False,
|
|
) -> Session:
|
|
"""Authenticate a doctor and return a validated, complete session."""
|
|
|
|
def list_appointments(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[Appointment]:
|
|
"""Return a filtered appointment page."""
|
|
|
|
def list_patients(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[Patient]:
|
|
"""Return a doctor-scoped patient page."""
|
|
|
|
def list_consultations(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[Consultation]:
|
|
"""Return a diagnosis/consultation page."""
|
|
|
|
def list_ai_patient_options(
|
|
self, *, page_no: int = 1, page_size: int = 20, keyword: str = ""
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Return privacy-safe diagnosis choices for patient AI workflows."""
|
|
|
|
def list_prescription_templates(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[PrescriptionTemplate]:
|
|
"""Return a prescription-library page."""
|
|
|
|
def list_prescriptions(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[Prescription]:
|
|
"""Return an issued-prescription page."""
|
|
|
|
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
|
"""Return the aggregate reception record for one appointment."""
|
|
|
|
def get_doctor_notes(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
"""Return doctor notes and media for one diagnosis."""
|
|
|
|
def upload_material(
|
|
self,
|
|
path: str | PathLike[str],
|
|
material_type: Literal["image", "video", "file", "tongue_images", "report_files"],
|
|
cid: int = 0,
|
|
) -> str:
|
|
"""Upload one local note material and return its server URI."""
|
|
|
|
def upload_material_bytes(
|
|
self,
|
|
content: bytes,
|
|
filename: str,
|
|
material_type: Literal["image", "video", "file"],
|
|
cid: int = 0,
|
|
) -> str:
|
|
"""Upload an in-memory capture and return its server URI."""
|
|
|
|
def get_prescription_template(self, template_id: int) -> PrescriptionTemplate:
|
|
"""Return one prescription-library record."""
|
|
|
|
def create_prescription_template(
|
|
self,
|
|
template: PrescriptionTemplate | Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> PrescriptionTemplate:
|
|
"""Create a prescription-library record."""
|
|
|
|
def update_prescription_template(
|
|
self,
|
|
template: int | PrescriptionTemplate | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> PrescriptionTemplate:
|
|
"""Update a prescription-library record."""
|
|
|
|
def delete_prescription_template(self, template_id: int) -> Any:
|
|
"""Delete a prescription-library record."""
|
|
|
|
def list_prescription_template_ai_reports(self, template_id: int) -> dict[str, Any]:
|
|
"""Return saved AI interpretation reports for one library template."""
|
|
|
|
def generate_prescription_template_ai_reports(self, template_id: int) -> dict[str, Any]:
|
|
"""Regenerate every model report for one library template."""
|
|
|
|
def edit_prescription_template_ai_report(
|
|
self,
|
|
template_id: int,
|
|
*,
|
|
report_id: int,
|
|
content: str,
|
|
) -> dict[str, Any]:
|
|
"""Save a manually edited AI report for one library template."""
|
|
|
|
def list_diagnosis_ai_reports(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Return saved AI reports for one diagnosis / patient profile."""
|
|
|
|
def generate_diagnosis_ai_reports(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Regenerate every model report for one diagnosis."""
|
|
|
|
def edit_diagnosis_ai_report(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
report_id: int,
|
|
content: str,
|
|
) -> dict[str, Any]:
|
|
"""Save a manually edited AI report for one diagnosis."""
|
|
|
|
def analyze_diagnosis_ai(
|
|
self,
|
|
diagnosis_id: int,
|
|
prompt: str,
|
|
*,
|
|
task: str = "custom",
|
|
) -> dict[str, Any]:
|
|
"""Ask the first-party diagnosis assistant; the server selects the model."""
|
|
|
|
def stream_diagnosis_ai(
|
|
self,
|
|
diagnosis_id: int,
|
|
prompt: str,
|
|
*,
|
|
task: str = "custom",
|
|
cancelled: Callable[[], bool] | None = None,
|
|
) -> Iterator[dict[str, Any]]:
|
|
"""Yield normalized ``start``/``delta``/``done`` assistant events."""
|
|
|
|
def generate_ai_prescription(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Generate a server-validated prescription draft for doctor review."""
|
|
|
|
def get_diagnosis_ai_analysis(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
model: Literal["qwen", "openai"] = "qwen",
|
|
) -> dict[str, Any]:
|
|
"""Generate one model's structured reception analysis for a diagnosis."""
|
|
|
|
def list_patient_ai_reports(self, patient_id: int) -> dict[str, Any]:
|
|
"""Return every persisted AI diagnosis snapshot for one patient."""
|
|
|
|
def generate_patient_ai_report(
|
|
self,
|
|
patient_id: int,
|
|
*,
|
|
model: Literal["qwen", "openai"],
|
|
) -> dict[str, Any]:
|
|
"""Append one model-specific AI diagnosis snapshot for one patient."""
|
|
|
|
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."""
|
|
|
|
def list_prescription_ai_reports(self, *, prescription_id: int = 0, diagnosis_id: int = 0, page_no: int = 1, page_size: int = 20) -> dict[str, Any]:
|
|
"""Read immutable analysis batch history in the authorized scope."""
|
|
|
|
def get_prescription_ai_report(self, batch_id: int) -> dict[str, Any]:
|
|
"""Read one analysis batch without starting model work."""
|
|
|
|
def regenerate_prescription_ai(self, prescription_id: int, reason: str) -> dict[str, Any]:
|
|
"""Explicitly request a new analysis batch with a reason."""
|
|
|
|
def retry_prescription_ai(self, batch_id: int, model_key: str) -> dict[str, Any]:
|
|
"""Retry only the requested failed model."""
|
|
|
|
def review_prescription_ai(self, batch_id: int, model_key: str, status: str, comment: str) -> dict[str, Any]:
|
|
"""Save a doctor's review separately from immutable AI output."""
|
|
|
|
def prescription_ai_statistics(self, date_from: str, date_to: str, doctor_id: int | None = None) -> dict[str, Any]:
|
|
"""Read baseline agreement statistics, never diagnostic accuracy."""
|
|
|
|
def create_prescription(
|
|
self,
|
|
prescription: Prescription | Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Prescription:
|
|
"""Create an issued prescription."""
|
|
|
|
def update_prescription(
|
|
self,
|
|
prescription: int | Prescription | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Prescription:
|
|
"""Update an issued prescription."""
|
|
|
|
def delete_prescription(self, prescription_id: int) -> Any:
|
|
"""Delete an issued prescription."""
|
|
|
|
def patch_prescription_patient(
|
|
self,
|
|
prescription_id: int,
|
|
*,
|
|
patient_name: str,
|
|
phone: str,
|
|
gender: int,
|
|
) -> Any:
|
|
"""Correct the patient identity printed on a prescription."""
|
|
|
|
def audit_prescription(
|
|
self,
|
|
prescription_id: int,
|
|
*,
|
|
action: AuditAction,
|
|
remark: str = "",
|
|
) -> Any:
|
|
"""Approve or reject an issued prescription."""
|
|
|
|
def list_medicines(
|
|
self,
|
|
*,
|
|
name: str = "",
|
|
page_no: int = 1,
|
|
page_size: int = 100,
|
|
status: int = 1,
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Return selectable active medicines."""
|
|
|
|
def patient_orders(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Return the doctor-scoped patient-order workspace."""
|
|
|
|
def patient_progress(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Return the doctor-scoped interview-progress workspace."""
|
|
|
|
def patient_detail(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Return a patient's permission-aware readonly diagnosis detail."""
|
|
|
|
def appointment_history(
|
|
self, diagnosis_id: int, *, page_no: int = 1, page_size: int = 500
|
|
) -> PageResult[Appointment]:
|
|
"""Return all appointments associated with a diagnosis."""
|
|
|
|
def assign_history(
|
|
self, diagnosis_id: int, *, page_no: int = 1, page_size: int = 20
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Return medical-assistant assignment history."""
|
|
|
|
def list_patient_assistants(self) -> list[dict[str, Any]]:
|
|
"""Return assistants selectable from the patient workspace."""
|
|
|
|
def assign_patient(
|
|
self,
|
|
diagnosis_id: int,
|
|
assistant_id: int,
|
|
*,
|
|
is_inherit: int | None = None,
|
|
) -> Any:
|
|
"""Assign or reassign a patient diagnosis to an assistant."""
|
|
|
|
def fill_patient_id_card(self, diagnosis_id: int, id_card: str) -> Any:
|
|
"""Complete a patient's identity card through the scoped endpoint."""
|
|
|
|
def book_patient_appointment(
|
|
self, payload: Mapping[str, Any] | None = None, **fields: Any
|
|
) -> Any:
|
|
"""Create an appointment from the patient workspace."""
|
|
|
|
def cancel_patient_appointment(self, appointment_id: int) -> Any:
|
|
"""Cancel an appointment from the patient workspace."""
|
|
|
|
def cancel_diagnosis_appointment(self, appointment_id: int) -> Any:
|
|
"""Cancel a diagnosis-list appointment through the doctor route."""
|
|
|
|
def get_appointment_detail(self, appointment_id: int) -> dict[str, Any]:
|
|
"""Return one appointment detail from ``doctor.appointment/detail``."""
|
|
|
|
def list_departments(self, *, apply_data_scope: bool = False) -> list[dict[str, Any]]:
|
|
"""Return a department tree, optionally scoped to the account's data permissions."""
|
|
|
|
def get_diagnosis_detail(self, diagnosis_id: int, *, readonly: bool = False) -> dict[str, Any]:
|
|
"""Return an editable or permission-aware readonly diagnosis detail."""
|
|
|
|
def update_diagnosis(
|
|
self,
|
|
diagnosis: int | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> dict[str, Any]:
|
|
"""Update a diagnosis using its full form DTO."""
|
|
|
|
def restore_session(self, token: str | None = None) -> Session | None:
|
|
"""Restore and validate a previously issued access token."""
|
|
|
|
def get_session(self) -> Session:
|
|
"""Return the authoritative current session."""
|
|
|
|
def get_current_user(self) -> UserProfile:
|
|
"""Return the current authenticated profile."""
|
|
|
|
def logout(self, *, forget_account: bool = False) -> None:
|
|
"""Clear authentication and optionally remembered account metadata."""
|
|
|
|
def list_reception_queue(
|
|
self,
|
|
*,
|
|
status: int,
|
|
keyword: str = "",
|
|
page_no: int = 1,
|
|
page_size: int = 15,
|
|
on_date: date | str | None = None,
|
|
) -> PageResult[Appointment]:
|
|
"""Return one same-day reception queue."""
|
|
|
|
def list_appointment_rosters(
|
|
self,
|
|
*,
|
|
doctor_id: int,
|
|
start_date: str,
|
|
end_date: str,
|
|
status: int = 1,
|
|
page_no: int = 1,
|
|
page_size: int = 100,
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Return one doctor's dated appointment rosters."""
|
|
|
|
def get_available_appointment_slots(
|
|
self,
|
|
*,
|
|
doctor_id: int,
|
|
appointment_date: str,
|
|
period: str = "all",
|
|
) -> dict[str, Any]:
|
|
"""Return server-authoritative available slots for one doctor/date."""
|
|
|
|
def notify_assistant(self, appointment_id: int) -> Any:
|
|
"""Notify the assistant assigned to an appointment."""
|
|
|
|
def add_doctor_note(
|
|
self,
|
|
diagnosis_id: int,
|
|
content: str = "",
|
|
*,
|
|
tongue_images: list[str] | tuple[str, ...] | None = None,
|
|
report_files: list[str] | tuple[str, ...] | None = None,
|
|
) -> Any:
|
|
"""Append text and media to a diagnosis note."""
|
|
|
|
def delete_doctor_note_image(
|
|
self,
|
|
note_id: int,
|
|
image_type: Literal["tongue_images", "report_files"],
|
|
image_path: str,
|
|
) -> Any:
|
|
"""Delete one image or report from a note."""
|
|
|
|
def complete_appointment(self, appointment_id: int) -> Any:
|
|
"""Mark one appointment complete."""
|
|
|
|
def void_prescription(self, prescription_id: int) -> Any:
|
|
"""Void a diagnosis-context prescription."""
|
|
|
|
def list_prescriptions_by_diagnosis(self, diagnosis_id: int) -> list[Prescription]:
|
|
"""Return prescriptions linked to one diagnosis."""
|
|
|
|
def get_prescription_by_appointment(self, appointment_id: int) -> Prescription | None:
|
|
"""Return the prescription linked to an appointment, if present."""
|
|
|
|
def list_prescription_orders(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Return prescription fulfilment orders."""
|
|
|
|
def get_prescription_order(self, order_id: int) -> dict[str, Any]:
|
|
"""Return one prescription fulfilment order."""
|
|
|
|
def create_prescription_order(
|
|
self, payload: Mapping[str, Any] | None = None, **fields: Any
|
|
) -> dict[str, Any]:
|
|
"""Create a prescription fulfilment order."""
|
|
|
|
def list_paid_prescription_orders(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
prescription_order_id: int | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Return paid orders eligible for association."""
|
|
|
|
def search_diagnosis_patients(
|
|
self, keyword: str, *, page_no: int = 1, page_size: int = 10
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Search diagnoses for an order patient selector."""
|
|
|
|
def get_dictionary(self, dictionary_type: str) -> list[dict[str, Any]]:
|
|
"""Return one configuration dictionary."""
|
|
|
|
def get_dictionaries(
|
|
self, dictionary_types: Sequence[str]
|
|
) -> dict[str, list[dict[str, Any]]]:
|
|
"""Return several configuration dictionaries in one call."""
|
|
|
|
def list_chat_notifications(self) -> list[dict[str, Any]]:
|
|
"""Consume this account's queued chat notifications."""
|
|
|
|
def get_patient_order(self, order_id: int) -> dict[str, Any]:
|
|
"""Return a patient-scoped order detail."""
|
|
|
|
def edit_patient_order(
|
|
self,
|
|
order: int | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Any:
|
|
"""Edit a patient-scoped order."""
|
|
|
|
def audit_patient_order_prescription(
|
|
self, order_id: int, action: AuditAction, remark: str = ""
|
|
) -> Any:
|
|
"""Audit a patient order's prescription."""
|
|
|
|
def revoke_patient_order_prescription_audit(self, order_id: int) -> Any:
|
|
"""Revoke a patient order's prescription audit."""
|
|
|
|
def audit_patient_order_payment(
|
|
self, order_id: int, action: AuditAction, remark: str = ""
|
|
) -> Any:
|
|
"""Audit a patient order's payment."""
|
|
|
|
def revoke_patient_order_payment_audit(self, order_id: int) -> Any:
|
|
"""Revoke a patient order's payment audit."""
|
|
|
|
def update_patient_order_shipping(
|
|
self, order_id: int, express_company: str, tracking_number: str
|
|
) -> Any:
|
|
"""Update a patient order's courier fields."""
|
|
|
|
def ship_patient_order(
|
|
self,
|
|
order_id: int,
|
|
express_company: str,
|
|
tracking_number: str,
|
|
*,
|
|
ship_mode: Literal["gancao", "direct"] | None = None,
|
|
) -> Any:
|
|
"""Advance a patient order to shipped."""
|
|
|
|
def add_patient_order_payment(
|
|
self,
|
|
order_id: int,
|
|
order_type: int,
|
|
pay_amount: float,
|
|
*,
|
|
pay_remark: str = "",
|
|
completion_request: int | None = None,
|
|
pay_create_type: Literal["fubei", "express_cod"] | None = None,
|
|
) -> Any:
|
|
"""Add a payment to a shipped patient order."""
|
|
|
|
def complete_patient_order(self, order_id: int, fulfillment_status: int) -> Any:
|
|
"""Complete a patient order."""
|
|
|
|
def refund_patient_order(
|
|
self, order_id: int, reason: str, refund_amount: float | None = None
|
|
) -> Any:
|
|
"""Refund a patient order."""
|
|
|
|
def withdraw_patient_order(self, order_id: int) -> Any:
|
|
"""Withdraw a patient order."""
|
|
|
|
def upload_patient_order_to_pharmacy(self, order_id: int) -> Any:
|
|
"""Submit a patient order to a pharmacy."""
|
|
|
|
def diagnosis_readonly_detail(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Return the readonly diagnosis aggregate."""
|
|
|
|
def create_diagnosis(
|
|
self, diagnosis: Mapping[str, Any] | None = None, **fields: Any
|
|
) -> dict[str, Any]:
|
|
"""Create a diagnosis."""
|
|
|
|
def delete_diagnosis(self, diagnosis_id: int) -> Any:
|
|
"""Delete a diagnosis."""
|
|
|
|
def set_revisit_slot_start_offset(self, diagnosis_id: int, offset: int) -> Any:
|
|
"""Set a diagnosis revisit-statistics offset."""
|
|
|
|
def assign_diagnosis(
|
|
self,
|
|
diagnosis_id: int,
|
|
assistant_id: int,
|
|
*,
|
|
is_inherit: int | None = None,
|
|
) -> Any:
|
|
"""Assign a diagnosis to an assistant."""
|
|
|
|
def list_diagnosis_assistants(self) -> list[dict[str, Any]]:
|
|
"""Return assistants selectable in diagnosis forms."""
|
|
|
|
def list_diagnosis_doctors(self) -> list[dict[str, Any]]:
|
|
"""Return doctors selectable in diagnosis forms."""
|
|
|
|
def get_tracking_window(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
start_date: str = "",
|
|
end_date: str = "",
|
|
) -> dict[str, Any]:
|
|
"""Return a diagnosis tracking date window."""
|
|
|
|
def list_tracking_notes(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
"""Return diagnosis tracking notes."""
|
|
|
|
def add_tracking_note(self, diagnosis_id: int, content: str) -> Any:
|
|
"""Append a diagnosis tracking note."""
|
|
|
|
def add_blood_record(self, payload: Mapping[str, Any] | None = None, **fields: Any) -> Any:
|
|
"""Create one doctor-entered blood glucose / pressure record."""
|
|
|
|
def update_blood_record(
|
|
self,
|
|
record: int | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Any:
|
|
"""Update one blood glucose / pressure record."""
|
|
|
|
def add_diet_record(self, payload: Mapping[str, Any] | None = None, **fields: Any) -> Any:
|
|
"""Create one daily diet record."""
|
|
|
|
def update_diet_record(
|
|
self,
|
|
record: int | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Any:
|
|
"""Update one daily diet record."""
|
|
|
|
def add_exercise_record(self, payload: Mapping[str, Any] | None = None, **fields: Any) -> Any:
|
|
"""Create one daily exercise record."""
|
|
|
|
def update_exercise_record(
|
|
self,
|
|
record: int | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Any:
|
|
"""Update one daily exercise record."""
|
|
|
|
def check_diagnosis_phone(self, payload: Mapping[str, Any]) -> Any:
|
|
"""Check diagnosis phone uniqueness."""
|
|
|
|
def check_diagnosis_id_card(self, payload: Mapping[str, Any]) -> Any:
|
|
"""Check diagnosis identity-card uniqueness."""
|
|
|
|
def fill_diagnosis_id_card(self, diagnosis_id: int, id_card: str) -> Any:
|
|
"""Fill a diagnosis identity card."""
|
|
|
|
def list_diagnosis_todos(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
page_no: int = 1,
|
|
page_size: int = 20,
|
|
status: int | None = None,
|
|
creator_id: int | None = None,
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Return diagnosis follow-up todos."""
|
|
|
|
def add_diagnosis_todo(self, diagnosis_id: int, content: str, remind_time: int) -> Any:
|
|
"""Create a diagnosis follow-up todo."""
|
|
|
|
def cancel_diagnosis_todo(self, todo_id: int) -> Any:
|
|
"""Cancel a diagnosis follow-up todo."""
|
|
|
|
def list_call_records(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
"""Return persisted audio/video call records and replay URLs."""
|
|
|
|
def create_manual_call_record(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Create the server record used to hold a manually uploaded replay."""
|
|
|
|
def attach_local_call_recording(
|
|
self,
|
|
diagnosis_id: int,
|
|
file_url: str,
|
|
*,
|
|
call_record_id: int | None = None,
|
|
) -> Any:
|
|
"""Attach one uploaded replay URI to a call record."""
|
|
|
|
def upload_call_recording(
|
|
self,
|
|
path: str | PathLike[str],
|
|
diagnosis_id: int,
|
|
*,
|
|
call_record_id: int | None = None,
|
|
mime_type: str = "audio/webm",
|
|
) -> dict[str, Any]:
|
|
"""Upload a locally captured call recording to a real call record."""
|
|
|
|
def list_im_chat_messages(
|
|
self, diagnosis_id: int, *, only_archived: bool = True
|
|
) -> dict[str, Any]:
|
|
"""Return diagnosis IM messages, defaulting to the fast archive-only path."""
|
|
|
|
def sync_im_chat_messages(self, diagnosis_id: int) -> Any:
|
|
"""Queue the server-side IM archive synchronisation job."""
|
|
|
|
def list_diagnosis_appointment_logs(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
"""Return per-diagnosis registration / cancellation audit rows."""
|
|
|
|
def generate_mini_program_qrcode(
|
|
self, payload: Mapping[str, Any] | None = None, **fields: Any
|
|
) -> dict[str, Any]:
|
|
"""Generate the diagnosis or video-login mini-program QR code."""
|
|
|
|
def get_mini_program_config(self) -> dict[str, Any]:
|
|
"""Return the mini-program configuration required before QR generation."""
|
|
|
|
def download_public_image(self, url: str) -> bytes:
|
|
"""Download a server-provided public image with bounded memory use."""
|
|
|
|
def generate_video_qrcode(
|
|
self,
|
|
doctor_id: int,
|
|
patient_id: int,
|
|
share_user_id: int,
|
|
*,
|
|
diagnosis_id: int | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Generate the video-login QR variant."""
|
|
|
|
def generate_diagnosis_qrcode(
|
|
self, diagnosis_id: int, doctor_id: int, patient_id: int, share_user_id: int
|
|
) -> dict[str, Any]:
|
|
"""Generate the diagnosis-confirmation QR variant."""
|
|
|
|
def list_appointment_logs(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
"""Compatibility name for diagnosis registration logs."""
|
|
|
|
def create_diagnosis_order(
|
|
self,
|
|
patient_id: int,
|
|
order_type: int,
|
|
amount: float,
|
|
*,
|
|
remark: str = "",
|
|
) -> dict[str, Any]:
|
|
"""Create the diagnosis-list generic payment order (not a prescription order)."""
|
|
|
|
def generate_order_qrcode(self, order_no: str) -> dict[str, Any]:
|
|
"""Generate the payment mini-program QR code for a generic order."""
|
|
|
|
def get_call_ticket(self, patient_id: int, diagnosis_id: int, *, appointment_id: int = 0) -> CallTicket:
|
|
"""Return short-lived call credentials."""
|
|
|
|
def start_call(self, diagnosis_id: int, patient_id: int, *, call_type: int = 2, appointment_id: int = 0) -> Any:
|
|
"""Create a call record."""
|
|
|
|
def end_call(
|
|
self, diagnosis_id: int, *, call_record_id: int | str | None = None
|
|
) -> Any:
|
|
"""End the exact active call when its server identity is available."""
|
|
|
|
def bind_call_room(
|
|
self,
|
|
diagnosis_id: int,
|
|
room_id: str,
|
|
*,
|
|
call_record_id: int | str | None = None,
|
|
) -> Any:
|
|
"""Bind a TRTC room to the exact active call and start cloud recording."""
|
|
|
|
def start_call_transcription(
|
|
self,
|
|
diagnosis_id: int,
|
|
call_record_id: int | str,
|
|
transcription_session_id: str,
|
|
*,
|
|
language: str = "zh-CN",
|
|
) -> Any:
|
|
"""Start a transcript stored on one exact call record."""
|
|
|
|
def upsert_call_transcript_segments(
|
|
self,
|
|
diagnosis_id: int,
|
|
call_record_id: int | str,
|
|
transcription_session_id: str,
|
|
segments: Sequence[Mapping[str, Any]],
|
|
) -> Any:
|
|
"""Idempotently persist completed transcript segments."""
|
|
|
|
def finish_call_transcription(
|
|
self,
|
|
diagnosis_id: int,
|
|
call_record_id: int | str,
|
|
transcription_session_id: str,
|
|
expected_segment_count: int,
|
|
*,
|
|
status: str = "completed",
|
|
) -> Any:
|
|
"""Finalize and materialize the transcript text on a call record."""
|
|
|
|
def my_self(self) -> Session:
|
|
"""Compatibility alias for :meth:`get_session`."""
|
|
|
|
def reception(self, appointment_id: int) -> dict[str, Any]:
|
|
"""Compatibility alias for :meth:`get_reception`."""
|
|
|
|
def add_prescription_template(
|
|
self,
|
|
template: PrescriptionTemplate | Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> PrescriptionTemplate:
|
|
"""Compatibility alias for prescription-template creation."""
|
|
|
|
def edit_prescription_template(
|
|
self,
|
|
template: int | PrescriptionTemplate | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> PrescriptionTemplate:
|
|
"""Compatibility alias for prescription-template editing."""
|
|
|
|
|
|
class RemoteDoctorRepository:
|
|
"""Translate doctor-workstation operations into confirmed admin endpoints."""
|
|
|
|
def __init__(self, client: ApiClient, token_store: TokenStore | None = None) -> None:
|
|
"""Create a repository around an API client and optional token store."""
|
|
|
|
self.client = client
|
|
self.token_store = token_store
|
|
self._login_metadata: dict[str, Any] = {}
|
|
self._session: Session | None = None
|
|
|
|
def login(
|
|
self,
|
|
account: str,
|
|
password: str,
|
|
*,
|
|
remember_account: bool = False,
|
|
) -> Session:
|
|
"""Authenticate, validate ``mySelf``, then persist the issued token.
|
|
|
|
Token persistence is the final step of the transaction. A failed
|
|
profile/session validation clears both the candidate in memory and any
|
|
previously persisted token, so a partial login cannot survive restart.
|
|
"""
|
|
|
|
clean_account = account.strip()
|
|
if not clean_account or not password:
|
|
raise ValueError("account and password are required")
|
|
try:
|
|
result = _require_mapping(
|
|
self.client.post(
|
|
"login/account",
|
|
{"account": clean_account, "password": password, "terminal": 1},
|
|
),
|
|
"login/account",
|
|
)
|
|
token = str(result.get("token") or "").strip()
|
|
if not token:
|
|
raise ApiProtocolError(
|
|
"login/account returned no token",
|
|
data=dict(result),
|
|
)
|
|
self.client.set_token(token)
|
|
self._login_metadata = dict(result)
|
|
session = self.get_session()
|
|
if not isinstance(session, Session) or not session.authenticated:
|
|
raise ApiProtocolError(
|
|
"auth.admin/mySelf returned an incomplete session",
|
|
data=dict(result),
|
|
)
|
|
if self.token_store is not None:
|
|
self.token_store.save_token(
|
|
token,
|
|
account=clean_account if remember_account else "",
|
|
scope=self.client.base_url,
|
|
)
|
|
except Exception:
|
|
self._clear_authentication(clear_persisted=True)
|
|
raise
|
|
self._session = session
|
|
return session
|
|
|
|
def restore_session(self, token: str | None = None) -> Session | None:
|
|
"""Restore a persisted token and validate it by loading ``mySelf``."""
|
|
|
|
candidate = token
|
|
persisted_candidate = candidate is None
|
|
if candidate is None and self.token_store is not None:
|
|
candidate = self.token_store.load_token(scope=self.client.base_url)
|
|
if not candidate or not candidate.strip():
|
|
self._clear_authentication(clear_persisted=False)
|
|
return None
|
|
clean_token = candidate.strip()
|
|
self.client.set_token(clean_token)
|
|
self._login_metadata.clear()
|
|
try:
|
|
session = self.get_session()
|
|
if not isinstance(session, Session) or not session.authenticated:
|
|
raise ApiProtocolError("persisted token returned an incomplete session")
|
|
except AuthenticationExpiredError:
|
|
self._clear_authentication(clear_persisted=persisted_candidate)
|
|
raise
|
|
except ApiTransportError:
|
|
# A transient transport failure must not destroy an otherwise valid
|
|
# persisted credential; it remains available on the next startup.
|
|
self._clear_authentication(clear_persisted=False)
|
|
raise
|
|
except Exception:
|
|
# Protocol and business/control-flow failures are deterministic for
|
|
# this token and must not be replayed on every application launch.
|
|
self._clear_authentication(clear_persisted=persisted_candidate)
|
|
raise
|
|
self._session = session
|
|
return session
|
|
|
|
def get_session(self) -> Session:
|
|
"""Load the current user, permissions and dynamic menu from ``mySelf``."""
|
|
|
|
result = _require_mapping(self.client.get("auth.admin/mySelf"), "auth.admin/mySelf")
|
|
user = UserProfile.from_dict(result)
|
|
permission_values = _strings(result.get("permissions")) or user.permissions
|
|
user.permissions = permission_values
|
|
menu = _safe_menu(result.get("menu"))
|
|
is_paw = _to_int(user.raw.get("is_paw", self._login_metadata.get("is_paw", 1)), 1)
|
|
need_bind = _to_bool(
|
|
self._login_metadata.get(
|
|
"need_bind_work_wechat", result.get("need_bind_work_wechat", False)
|
|
)
|
|
)
|
|
session = Session(
|
|
token=self.client.token,
|
|
user=user,
|
|
permissions=PermissionSet(permission_values),
|
|
menu=menu,
|
|
is_paw=is_paw,
|
|
need_bind_work_wechat=need_bind,
|
|
metadata={
|
|
key: value
|
|
for key, value in self._login_metadata.items()
|
|
if key not in {"token", "password"}
|
|
},
|
|
)
|
|
self._session = session
|
|
return session
|
|
|
|
def get_current_user(self) -> UserProfile:
|
|
"""Return the current authenticated user profile."""
|
|
|
|
return self._session.user if self._session is not None else self.get_session().user
|
|
|
|
def logout(self, *, forget_account: bool = False) -> None:
|
|
"""Clear local authentication without implicitly retrying a remote write."""
|
|
|
|
self.client.clear_token()
|
|
self._login_metadata.clear()
|
|
self._session = None
|
|
if self.token_store is not None:
|
|
if forget_account:
|
|
self.token_store.clear()
|
|
else:
|
|
self.token_store.clear_token()
|
|
|
|
def _clear_authentication(self, *, clear_persisted: bool) -> None:
|
|
"""Reset partial authentication state without masking its root error."""
|
|
|
|
self.client.clear_token()
|
|
self._login_metadata.clear()
|
|
self._session = None
|
|
if clear_persisted and self.token_store is not None:
|
|
with suppress(Exception):
|
|
self.token_store.clear_token()
|
|
|
|
def list_appointments(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[Appointment]:
|
|
"""List appointments using ``doctor.appointment/lists``.
|
|
|
|
Reception callers historically use this generic method with status 1
|
|
or 4. Those two queue states are always scoped to the local calendar
|
|
day unless the caller explicitly supplies a date range. History
|
|
callers use :meth:`appointment_history` and are never narrowed here.
|
|
"""
|
|
|
|
request_filters = dict(filters)
|
|
if request_filters.get("keyword") and not request_filters.get("patient_name"):
|
|
request_filters["patient_name"] = request_filters.pop("keyword")
|
|
# Internal client flag — never forward to the HTTP API.
|
|
relax_scope = bool(request_filters.pop("diag_scope_relax", False))
|
|
if (
|
|
str(request_filters.get("status") or "") in {"1", "4"}
|
|
and not request_filters.get("start_date")
|
|
and not request_filters.get("end_date")
|
|
and not relax_scope
|
|
):
|
|
today = date.today().isoformat()
|
|
request_filters.update({"start_date": today, "end_date": today})
|
|
payload = self.client.get(
|
|
"doctor.appointment/lists",
|
|
_page_params(page_no, page_size, request_filters),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload, Appointment.from_dict, page_no=page_no, page_size=page_size
|
|
)
|
|
|
|
def get_appointment_detail(self, appointment_id: int) -> dict[str, Any]:
|
|
"""Load one appointment through ``doctor.appointment/detail``."""
|
|
|
|
if appointment_id <= 0:
|
|
raise ValueError("appointment_id must be positive")
|
|
payload = self.client.get("doctor.appointment/detail", {"id": appointment_id})
|
|
return dict(_require_mapping(payload, "doctor.appointment/detail"))
|
|
|
|
def list_departments(self, *, apply_data_scope: bool = False) -> list[dict[str, Any]]:
|
|
"""Load ``dept.dept/all`` with optional server-enforced role data scope."""
|
|
|
|
payload = self.client.get(
|
|
"dept.dept/all", {"apply_data_scope": 1} if apply_data_scope else None
|
|
)
|
|
for _depth in range(5):
|
|
if isinstance(payload, list):
|
|
return [dict(row) for row in payload if isinstance(row, Mapping)]
|
|
if isinstance(payload, Mapping):
|
|
payload = payload.get("lists", payload.get("data", payload.get("tree")))
|
|
else:
|
|
break
|
|
raise ApiProtocolError("部门数据格式异常，请重试")
|
|
|
|
def list_reception_queue(
|
|
self,
|
|
*,
|
|
status: int,
|
|
keyword: str = "",
|
|
page_no: int = 1,
|
|
page_size: int = 15,
|
|
on_date: date | str | None = None,
|
|
) -> PageResult[Appointment]:
|
|
"""Return one of the two reception queues, forcibly scoped to one day."""
|
|
|
|
if status not in {1, 4}:
|
|
raise ValueError("reception status must be 1 or 4")
|
|
day = on_date.isoformat() if isinstance(on_date, date) else str(on_date or "")
|
|
day = day.strip() or date.today().isoformat()
|
|
return self.list_appointments(
|
|
status=status,
|
|
keyword=keyword,
|
|
start_date=day,
|
|
end_date=day,
|
|
page_no=page_no,
|
|
page_size=page_size,
|
|
)
|
|
|
|
def list_appointment_rosters(
|
|
self,
|
|
*,
|
|
doctor_id: int,
|
|
start_date: str,
|
|
end_date: str,
|
|
status: int = 1,
|
|
page_no: int = 1,
|
|
page_size: int = 100,
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Load one doctor's active rosters through ``doctor.roster/lists``."""
|
|
|
|
if doctor_id <= 0:
|
|
raise ValueError("doctor_id must be positive")
|
|
clean_start = start_date.strip()
|
|
clean_end = end_date.strip()
|
|
if not clean_start or not clean_end:
|
|
raise ValueError("start_date and end_date are required")
|
|
if clean_start > clean_end:
|
|
raise ValueError("start_date cannot be after end_date")
|
|
payload = self.client.get(
|
|
"doctor.roster/lists",
|
|
_page_params(
|
|
page_no,
|
|
page_size,
|
|
{
|
|
"doctor_id": doctor_id,
|
|
"start_date": clean_start,
|
|
"end_date": clean_end,
|
|
"status": status,
|
|
},
|
|
),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload,
|
|
lambda row: dict(row),
|
|
page_no=page_no,
|
|
page_size=page_size,
|
|
)
|
|
|
|
def get_available_appointment_slots(
|
|
self,
|
|
*,
|
|
doctor_id: int,
|
|
appointment_date: str,
|
|
period: str = "all",
|
|
) -> dict[str, Any]:
|
|
"""Load server-authoritative slots through ``availableSlots``."""
|
|
|
|
if doctor_id <= 0:
|
|
raise ValueError("doctor_id must be positive")
|
|
clean_date = appointment_date.strip()
|
|
if not clean_date:
|
|
raise ValueError("appointment_date is required")
|
|
payload = self.client.get(
|
|
"doctor.appointment/availableSlots",
|
|
{
|
|
"doctor_id": doctor_id,
|
|
"appointment_date": clean_date,
|
|
"period": period.strip() or "all",
|
|
},
|
|
)
|
|
return dict(_require_mapping(payload, "doctor.appointment/availableSlots"))
|
|
|
|
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
|
"""Load aggregated appointment, diagnosis and doctor-note details."""
|
|
|
|
return dict(
|
|
_require_mapping(
|
|
self.client.get("doctor.appointment/reception", {"id": appointment_id}),
|
|
"doctor.appointment/reception",
|
|
)
|
|
)
|
|
|
|
def upload_material(
|
|
self,
|
|
path: str | PathLike[str],
|
|
material_type: Literal["image", "video", "file", "tongue_images", "report_files"],
|
|
cid: int = 0,
|
|
) -> str:
|
|
"""Upload a local note material and return only its server reference."""
|
|
|
|
if cid < 0:
|
|
raise ValueError("cid must be non-negative")
|
|
kind = _material_kind(material_type)
|
|
source = Path(path)
|
|
if not source.is_file():
|
|
raise FileNotFoundError(f"material file does not exist: {source}")
|
|
mime_type = mimetypes.guess_type(source.name)[0] or "application/octet-stream"
|
|
endpoint = f"upload/{kind}"
|
|
with source.open("rb") as stream:
|
|
payload = self.client.post_multipart(
|
|
endpoint,
|
|
files={"file": (source.name, stream, mime_type)},
|
|
data={"cid": str(cid)},
|
|
)
|
|
return _normalise_material_reference(payload, endpoint)
|
|
|
|
def upload_material_bytes(
|
|
self,
|
|
content: bytes,
|
|
filename: str,
|
|
material_type: Literal["image", "video", "file"],
|
|
cid: int = 0,
|
|
) -> str:
|
|
"""Upload a trusted in-memory capture without a local plaintext file."""
|
|
|
|
if cid < 0:
|
|
raise ValueError("cid must be non-negative")
|
|
if not content:
|
|
raise ValueError("material content must not be empty")
|
|
if len(content) > 10 * 1024 * 1024:
|
|
raise ValueError("material content exceeds 10 MB")
|
|
safe_name = Path(filename).name.strip()
|
|
if not safe_name or safe_name in {".", ".."}:
|
|
raise ValueError("filename is required")
|
|
kind = _material_kind(material_type)
|
|
mime_type = mimetypes.guess_type(safe_name)[0] or "application/octet-stream"
|
|
endpoint = f"upload/{kind}"
|
|
payload = self.client.post_multipart(
|
|
endpoint,
|
|
files={"file": (safe_name, BytesIO(content), mime_type)},
|
|
data={"cid": str(cid)},
|
|
)
|
|
return _normalise_material_reference(payload, endpoint)
|
|
|
|
def notify_assistant(self, appointment_id: int) -> Any:
|
|
"""Ask the server to notify the assigned medical assistant."""
|
|
|
|
return self.client.post("doctor.appointment/notifyAssistant", {"id": appointment_id})
|
|
|
|
def add_doctor_note(
|
|
self,
|
|
diagnosis_id: int,
|
|
content: str = "",
|
|
*,
|
|
tongue_images: list[str] | tuple[str, ...] | None = None,
|
|
report_files: list[str] | tuple[str, ...] | None = None,
|
|
) -> Any:
|
|
"""Append a doctor's text, tongue images and report files to a diagnosis."""
|
|
|
|
if len(content) > 500:
|
|
raise ValueError("doctor note content cannot exceed 500 characters")
|
|
if len(tongue_images or ()) > 99 or len(report_files or ()) > 99:
|
|
raise ValueError("doctor note media cannot exceed 99 items per type")
|
|
if not content.strip() and not tongue_images and not report_files:
|
|
raise ValueError("a note must contain text, an image or a report")
|
|
clean_tongue = _server_materials(tongue_images, "tongue_images")
|
|
clean_reports = _server_materials(report_files, "report_files")
|
|
body: dict[str, Any] = {"diagnosis_id": diagnosis_id, "content": content.strip()}
|
|
if tongue_images is not None:
|
|
body["tongue_images"] = clean_tongue
|
|
if report_files is not None:
|
|
body["report_files"] = clean_reports
|
|
return self.client.post("doctor.appointment/addDoctorNote", body)
|
|
|
|
def get_doctor_notes(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
"""Load all text and media notes for one diagnosis."""
|
|
|
|
payload = self.client.get("doctor.appointment/doctorNotes", {"diagnosis_id": diagnosis_id})
|
|
return _mapping_rows(payload)
|
|
|
|
def delete_doctor_note_image(
|
|
self,
|
|
note_id: int,
|
|
image_type: Literal["tongue_images", "report_files"],
|
|
image_path: str,
|
|
) -> Any:
|
|
"""Delete one tongue image or report attachment from a doctor note."""
|
|
|
|
if image_type not in {"tongue_images", "report_files"}:
|
|
raise ValueError("image_type must be tongue_images or report_files")
|
|
if not image_path.strip():
|
|
raise ValueError("image_path is required")
|
|
return self.client.post(
|
|
"doctor.appointment/deleteDoctorNoteImage",
|
|
{
|
|
"note_id": note_id,
|
|
"image_type": image_type,
|
|
"image_path": image_path.strip(),
|
|
},
|
|
)
|
|
|
|
def complete_appointment(self, appointment_id: int) -> Any:
|
|
"""Mark an appointment complete using the server-authoritative action."""
|
|
|
|
return self.client.post("doctor.appointment/complete", {"id": appointment_id})
|
|
|
|
def list_prescription_templates(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[PrescriptionTemplate]:
|
|
"""List reusable formulas from ``tcm.prescriptionLibrary/lists``."""
|
|
|
|
request_filters = dict(filters)
|
|
if request_filters.get("keyword") and not request_filters.get("prescription_name"):
|
|
request_filters["prescription_name"] = request_filters.pop("keyword")
|
|
if request_filters.get("formula_type") not in (None, ""):
|
|
request_filters["formula_type"] = _formula_type_for_api(request_filters["formula_type"])
|
|
payload = self.client.get(
|
|
"tcm.prescriptionLibrary/lists",
|
|
_page_params(page_no, page_size, request_filters),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload,
|
|
PrescriptionTemplate.from_dict,
|
|
page_no=page_no,
|
|
page_size=page_size,
|
|
)
|
|
|
|
def get_prescription_template(self, template_id: int) -> PrescriptionTemplate:
|
|
"""Load one prescription-library formula by identifier."""
|
|
|
|
result = _require_mapping(
|
|
self.client.get("tcm.prescriptionLibrary/detail", {"id": template_id}),
|
|
"tcm.prescriptionLibrary/detail",
|
|
)
|
|
return PrescriptionTemplate.from_dict(result)
|
|
|
|
def create_prescription_template(
|
|
self,
|
|
template: PrescriptionTemplate | Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> PrescriptionTemplate:
|
|
"""Create a reusable formula and return its normalised representation."""
|
|
|
|
body = _template_payload(template, fields, include_id=False)
|
|
result = self.client.post("tcm.prescriptionLibrary/add", body)
|
|
merged = dict(body)
|
|
if isinstance(result, Mapping):
|
|
merged.update(result)
|
|
elif isinstance(result, (int, str)):
|
|
merged["id"] = result
|
|
return PrescriptionTemplate.from_dict(merged)
|
|
|
|
def update_prescription_template(
|
|
self,
|
|
template: int | PrescriptionTemplate | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> PrescriptionTemplate:
|
|
"""Update an existing formula and return the submitted server state."""
|
|
|
|
extra = dict(changes or {})
|
|
extra.update(fields)
|
|
if isinstance(template, int):
|
|
extra["id"] = template
|
|
source: PrescriptionTemplate | Mapping[str, Any] | None = None
|
|
else:
|
|
source = template
|
|
body = _template_payload(source, extra, include_id=True)
|
|
if not _to_int(body.get("id"), 0):
|
|
raise ValueError("template id is required")
|
|
result = self.client.post("tcm.prescriptionLibrary/edit", body)
|
|
merged = dict(body)
|
|
if isinstance(result, Mapping):
|
|
merged.update(result)
|
|
return PrescriptionTemplate.from_dict(merged)
|
|
|
|
def delete_prescription_template(self, template_id: int) -> Any:
|
|
"""Delete a reusable formula by identifier."""
|
|
|
|
return self.client.post("tcm.prescriptionLibrary/delete", {"id": template_id})
|
|
|
|
def list_prescription_template_ai_reports(self, template_id: int) -> dict[str, Any]:
|
|
"""Read persisted AI reports without triggering model generation."""
|
|
|
|
payload = _client_request(
|
|
self.client,
|
|
"get",
|
|
"tcm.prescriptionLibrary/aiReports",
|
|
{"id": template_id},
|
|
timeout=30.0,
|
|
)
|
|
return dict(_require_mapping(payload, "tcm.prescriptionLibrary/aiReports"))
|
|
|
|
def generate_prescription_template_ai_reports(self, template_id: int) -> dict[str, Any]:
|
|
"""Regenerate the multi-model diagnosis report; not retried by the client."""
|
|
|
|
payload = _client_request(
|
|
self.client,
|
|
"post",
|
|
"tcm.prescriptionLibrary/generateAiReports",
|
|
{"id": template_id},
|
|
timeout=210.0,
|
|
)
|
|
return dict(_require_mapping(payload, "tcm.prescriptionLibrary/generateAiReports"))
|
|
|
|
def edit_prescription_template_ai_report(
|
|
self,
|
|
template_id: int,
|
|
*,
|
|
report_id: int,
|
|
content: str,
|
|
) -> dict[str, Any]:
|
|
"""Persist a manual edit of one saved model report."""
|
|
|
|
payload = _client_request(
|
|
self.client,
|
|
"post",
|
|
"tcm.prescriptionLibrary/editAiReport",
|
|
{"id": template_id, "report_id": report_id, "content": content},
|
|
timeout=30.0,
|
|
)
|
|
return dict(_require_mapping(payload, "tcm.prescriptionLibrary/editAiReport"))
|
|
|
|
def list_diagnosis_ai_reports(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Read persisted diagnosis AI reports without triggering model generation."""
|
|
|
|
payload = _client_request(
|
|
self.client,
|
|
"get",
|
|
"tcm.diagnosis/aiReports",
|
|
{"id": diagnosis_id},
|
|
timeout=30.0,
|
|
)
|
|
return dict(_require_mapping(payload, "tcm.diagnosis/aiReports"))
|
|
|
|
def generate_diagnosis_ai_reports(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Regenerate the multi-model diagnosis report; not retried by the client."""
|
|
|
|
payload = _client_request(
|
|
self.client,
|
|
"post",
|
|
"tcm.diagnosis/generateAiReports",
|
|
{"id": diagnosis_id},
|
|
timeout=210.0,
|
|
)
|
|
return dict(_require_mapping(payload, "tcm.diagnosis/generateAiReports"))
|
|
|
|
def edit_diagnosis_ai_report(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
report_id: int,
|
|
content: str,
|
|
) -> dict[str, Any]:
|
|
"""Persist a manual edit of one saved diagnosis report."""
|
|
|
|
payload = _client_request(
|
|
self.client,
|
|
"post",
|
|
"tcm.diagnosis/editAiReport",
|
|
{"id": diagnosis_id, "report_id": report_id, "content": content},
|
|
timeout=30.0,
|
|
)
|
|
return dict(_require_mapping(payload, "tcm.diagnosis/editAiReport"))
|
|
|
|
def analyze_diagnosis_ai(
|
|
self,
|
|
diagnosis_id: int,
|
|
prompt: str,
|
|
*,
|
|
task: str = "custom",
|
|
) -> dict[str, Any]:
|
|
"""Submit a diagnosis question to the first-party server assistant."""
|
|
|
|
clean_prompt, clean_task = _diagnosis_ai_request(diagnosis_id, prompt, task)
|
|
payload = _client_request(
|
|
self.client,
|
|
"post",
|
|
"tcm.diagnosis/aiAssistant",
|
|
{"id": diagnosis_id, "prompt": clean_prompt, "task": clean_task},
|
|
# The upstream is allowed 90 seconds by server configuration. Keep a
|
|
# small transport buffer so the desktop can receive the server's own
|
|
# timeout response instead of racing it.
|
|
timeout=105.0,
|
|
)
|
|
return dict(_require_mapping(payload, "tcm.diagnosis/aiAssistant"))
|
|
|
|
def stream_diagnosis_ai(
|
|
self,
|
|
diagnosis_id: int,
|
|
prompt: str,
|
|
*,
|
|
task: str = "custom",
|
|
cancelled: Callable[[], bool] | None = None,
|
|
) -> Iterator[dict[str, Any]]:
|
|
"""Stream a diagnosis answer; use the legacy route only when streaming is absent."""
|
|
|
|
clean_prompt, clean_task = _diagnosis_ai_request(diagnosis_id, prompt, task)
|
|
body = {"id": diagnosis_id, "prompt": clean_prompt, "task": clean_task}
|
|
received_event = False
|
|
received_done = False
|
|
try:
|
|
for raw_event in self.client.post_event_stream(
|
|
"tcm.diagnosis/aiAssistantStream",
|
|
body,
|
|
timeout=105.0,
|
|
cancelled=cancelled,
|
|
):
|
|
if cancelled is not None and cancelled():
|
|
return
|
|
received_event = True
|
|
event = _normalise_diagnosis_ai_event(raw_event)
|
|
if event is None:
|
|
continue
|
|
kind = event["event"]
|
|
if kind == "done":
|
|
received_done = True
|
|
yield event
|
|
if kind == "done":
|
|
return
|
|
if cancelled is not None and cancelled():
|
|
return
|
|
if not received_done:
|
|
raise ApiProtocolError("AI assistant stream ended before done")
|
|
except ApiHttpError as error:
|
|
if (
|
|
error.status_code not in {404, 405}
|
|
or received_event
|
|
or (cancelled is not None and cancelled())
|
|
):
|
|
raise
|
|
|
|
# Only a missing stream route on an older deployment justifies another
|
|
# full request. Timeouts, transport failures and malformed streams do not.
|
|
result = self.analyze_diagnosis_ai(
|
|
diagnosis_id,
|
|
clean_prompt,
|
|
task=clean_task,
|
|
)
|
|
if cancelled is not None and cancelled():
|
|
return
|
|
yield {"event": "start", "fallback": True}
|
|
answer = str(result.get("answer") or result.get("content") or "")
|
|
if answer:
|
|
yield {"event": "delta", "text": answer, "fallback": True}
|
|
yield {**result, "event": "done", "fallback": True}
|
|
|
|
def generate_ai_prescription(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Generate a structured draft using the server's full patient context."""
|
|
|
|
return self.analyze_diagnosis_ai(
|
|
diagnosis_id,
|
|
"请依据患者全部纵向资料生成一份可复核的中医处方草稿。",
|
|
task="prescription_generate",
|
|
)
|
|
|
|
def get_diagnosis_ai_analysis(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
model: Literal["qwen", "openai"] = "qwen",
|
|
) -> dict[str, Any]:
|
|
"""Generate one model's structured analysis via the first-party API."""
|
|
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
clean_model = str(model).strip().lower()
|
|
if clean_model not in {"qwen", "openai"}:
|
|
raise ValueError("model must be qwen or openai")
|
|
payload = _client_request(
|
|
self.client,
|
|
"post",
|
|
"tcm.diagnosis/aiAnalysis",
|
|
{"id": diagnosis_id, "model": clean_model},
|
|
timeout=105.0,
|
|
)
|
|
return dict(_require_mapping(payload, "tcm.diagnosis/aiAnalysis"))
|
|
|
|
def list_patient_ai_reports(self, patient_id: int) -> dict[str, Any]:
|
|
"""Read patient-level snapshots without triggering model generation."""
|
|
|
|
if patient_id <= 0:
|
|
raise ValueError("patient_id must be positive")
|
|
payload = _client_request(
|
|
self.client,
|
|
"get",
|
|
"tcm.diagnosis/patientAiReports",
|
|
{"patient_id": patient_id},
|
|
timeout=30.0,
|
|
)
|
|
return dict(_require_mapping(payload, "tcm.diagnosis/patientAiReports"))
|
|
|
|
def generate_patient_ai_report(
|
|
self,
|
|
patient_id: int,
|
|
*,
|
|
model: Literal["qwen", "openai"],
|
|
) -> dict[str, Any]:
|
|
"""Append one patient snapshot; the desktop never sends provider secrets."""
|
|
|
|
if patient_id <= 0:
|
|
raise ValueError("patient_id must be positive")
|
|
clean_model = str(model).strip().lower()
|
|
if clean_model not in {"qwen", "openai"}:
|
|
raise ValueError("model must be qwen or openai")
|
|
payload = _client_request(
|
|
self.client,
|
|
"post",
|
|
"tcm.diagnosis/generatePatientAiReport",
|
|
{"patient_id": patient_id, "model": clean_model},
|
|
timeout=105.0,
|
|
)
|
|
return dict(
|
|
_require_mapping(payload, "tcm.diagnosis/generatePatientAiReport")
|
|
)
|
|
|
|
def list_medicines(
|
|
self,
|
|
*,
|
|
name: str = "",
|
|
page_no: int = 1,
|
|
page_size: int = 100,
|
|
status: int = 1,
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""List selectable medicines using the exact medicine-picker contract."""
|
|
|
|
payload = self.client.get(
|
|
"doctor.medicine/lists",
|
|
_page_params(
|
|
page_no,
|
|
page_size,
|
|
{"name": name.strip(), "status": status},
|
|
),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload,
|
|
lambda row: dict(row),
|
|
page_no=page_no,
|
|
page_size=page_size,
|
|
)
|
|
|
|
def list_prescriptions(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[Prescription]:
|
|
"""List issued prescriptions using ``tcm.prescription/lists``."""
|
|
|
|
request_filters = dict(filters)
|
|
if str(request_filters.get("audit_filter") or "").lower() == "all":
|
|
request_filters["audit_filter"] = ""
|
|
if str(request_filters.get("source_filter") or "").lower() == "all":
|
|
request_filters["source_filter"] = ""
|
|
keyword = str(request_filters.pop("keyword", "") or "").strip()
|
|
if keyword:
|
|
key = (
|
|
"sn"
|
|
if keyword.upper().startswith(("RX", "CF")) or keyword.isdigit()
|
|
else "patient_name"
|
|
)
|
|
request_filters.setdefault(key, keyword)
|
|
status = request_filters.pop("status", None)
|
|
legacy_status = request_filters.pop("audit_status", None)
|
|
if status in (None, ""):
|
|
status = legacy_status
|
|
if status not in (None, ""):
|
|
status_text = str(status).strip().lower()
|
|
audit_filter = (
|
|
status_text
|
|
if status_text in {"pending", "passed", "not_passed", "rejected"}
|
|
else {0: "pending", 1: "passed", 2: "rejected"}.get(_to_int(status, -99))
|
|
)
|
|
if audit_filter:
|
|
request_filters["audit_filter"] = audit_filter
|
|
payload = self.client.get(
|
|
"tcm.prescription/lists",
|
|
_page_params(page_no, page_size, request_filters),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload, Prescription.from_dict, page_no=page_no, page_size=page_size
|
|
)
|
|
|
|
def get_prescription(self, prescription_id: int) -> Prescription:
|
|
"""Load one issued prescription using ``tcm.prescription/detail``."""
|
|
|
|
result = _require_mapping(
|
|
self.client.get("tcm.prescription/detail", {"id": prescription_id}),
|
|
"tcm.prescription/detail",
|
|
)
|
|
return Prescription.from_dict(result)
|
|
|
|
def _prescription_ai_request(self, action: str, params: dict[str, Any], *, mutation: bool = False) -> dict[str, Any]:
|
|
endpoint = f"tcm.prescriptionAi/{action}"
|
|
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")
|
|
return self._prescription_ai_request("statuses", {"ids": ",".join(str(int(value)) for value in dict.fromkeys(ids))})
|
|
|
|
def list_prescription_ai_reports(self, *, prescription_id: int = 0, diagnosis_id: int = 0, page_no: int = 1, page_size: int = 20) -> dict[str, Any]:
|
|
if bool(prescription_id) == bool(diagnosis_id):
|
|
raise ValueError("Exactly one prescription_id or diagnosis_id is required")
|
|
params = {"prescription_id": prescription_id} if prescription_id else {"diagnosis_id": diagnosis_id}
|
|
params.update(page_no=page_no, page_size=page_size)
|
|
return self._prescription_ai_request("reports", params)
|
|
|
|
def get_prescription_ai_report(self, batch_id: int) -> dict[str, Any]:
|
|
return self._prescription_ai_request("detail", {"batch_id": batch_id})
|
|
|
|
def regenerate_prescription_ai(self, prescription_id: int, reason: str) -> dict[str, Any]:
|
|
if not reason.strip():
|
|
raise ValueError("重新分析需要填写原因")
|
|
return self._prescription_ai_request("regenerate", {"prescription_id": prescription_id, "reason": reason.strip()}, mutation=True)
|
|
|
|
def retry_prescription_ai(self, batch_id: int, model_key: str) -> dict[str, Any]:
|
|
if model_key not in {"qwen", "openai"}:
|
|
raise ValueError("Unknown model_key")
|
|
return self._prescription_ai_request("retry", {"batch_id": batch_id, "model_key": model_key}, mutation=True)
|
|
|
|
def review_prescription_ai(self, batch_id: int, model_key: str, status: str, comment: str) -> dict[str, Any]:
|
|
if model_key not in {"qwen", "openai"} or status not in {"viewed", "needs_information", "not_adopted", "reviewed"}:
|
|
raise ValueError("Invalid review state")
|
|
return self._prescription_ai_request("review", {"batch_id": batch_id, "model_key": model_key, "status": status, "comment": comment}, mutation=True)
|
|
|
|
def prescription_ai_statistics(self, date_from: str, date_to: str, doctor_id: int | None = None) -> dict[str, Any]:
|
|
params: dict[str, Any] = {"date_from": date_from, "date_to": date_to}
|
|
if doctor_id is not None:
|
|
params["doctor_id"] = doctor_id
|
|
return self._prescription_ai_request("statistics", params)
|
|
|
|
def create_prescription(
|
|
self,
|
|
prescription: Prescription | Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Prescription:
|
|
"""Create an issued prescription with the complete admin form DTO."""
|
|
|
|
body = _prescription_payload(prescription, fields, include_id=False)
|
|
result = self.client.post("tcm.prescription/add", body)
|
|
return Prescription.from_dict(_merge_result(body, result))
|
|
|
|
def update_prescription(
|
|
self,
|
|
prescription: int | Prescription | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Prescription:
|
|
"""Edit a prescription; the server resets audit state authoritatively."""
|
|
|
|
extra = dict(changes or {})
|
|
extra.update(fields)
|
|
if isinstance(prescription, int):
|
|
extra["id"] = prescription
|
|
source: Prescription | Mapping[str, Any] | None = None
|
|
else:
|
|
source = prescription
|
|
body = _prescription_payload(source, extra, include_id=True)
|
|
if not _to_int(body.get("id"), 0):
|
|
raise ValueError("prescription id is required")
|
|
result = self.client.post("tcm.prescription/edit", body)
|
|
return Prescription.from_dict(_merge_result(body, result))
|
|
|
|
def delete_prescription(self, prescription_id: int) -> Any:
|
|
"""Delete an editable, non-approved prescription."""
|
|
|
|
return self.client.post("tcm.prescription/delete", {"id": prescription_id})
|
|
|
|
def patch_prescription_patient(
|
|
self,
|
|
prescription_id: int,
|
|
*,
|
|
patient_name: str,
|
|
phone: str,
|
|
gender: int,
|
|
) -> Any:
|
|
"""Correct name, phone and gender without changing audit state."""
|
|
|
|
clean_name = patient_name.strip()
|
|
if not clean_name:
|
|
raise ValueError("patient_name is required")
|
|
return self.client.post(
|
|
"tcm.prescription/patchPatient",
|
|
{
|
|
"id": prescription_id,
|
|
"patient_name": clean_name,
|
|
"phone": phone.strip(),
|
|
"gender": gender,
|
|
},
|
|
)
|
|
|
|
def audit_prescription(
|
|
self,
|
|
prescription_id: int,
|
|
*,
|
|
action: AuditAction,
|
|
remark: str = "",
|
|
) -> Any:
|
|
"""Approve or reject; rejection requires the admin view's remark."""
|
|
|
|
action = _audit_action(action, remark)
|
|
return self.client.post(
|
|
"tcm.prescription/audit",
|
|
{"id": prescription_id, "action": action, "remark": remark.strip()},
|
|
)
|
|
|
|
def void_prescription(self, prescription_id: int) -> Any:
|
|
"""Void a diagnosis-context prescription when no business order blocks it."""
|
|
|
|
return self.client.post("tcm.prescription/void", {"id": prescription_id})
|
|
|
|
def list_prescriptions_by_diagnosis(self, diagnosis_id: int) -> list[Prescription]:
|
|
"""Return all prescriptions attached to one diagnosis."""
|
|
|
|
payload = self.client.get(
|
|
"tcm.prescription/listByDiagnosis", {"diagnosis_id": diagnosis_id}
|
|
)
|
|
return PageResult.from_payload(payload, Prescription.from_dict).items
|
|
|
|
def get_prescription_by_appointment(self, appointment_id: int) -> Prescription | None:
|
|
"""Return the prescription attached to an appointment, if one exists."""
|
|
|
|
try:
|
|
payload = self.client.get(
|
|
"tcm.prescription/getByAppointment", {"appointment_id": appointment_id}
|
|
)
|
|
except ApiProtocolError as exc:
|
|
# Older servers returned a top-level JSON empty string for "not found".
|
|
# Keep this compatibility endpoint-scoped; every other API stays strict.
|
|
if exc.data == "":
|
|
return None
|
|
raise
|
|
if payload in (None, "", [], {}):
|
|
return None
|
|
if isinstance(payload, Mapping):
|
|
for key in ("prescription", "detail"):
|
|
if isinstance(payload.get(key), Mapping):
|
|
return Prescription.from_dict(payload[key])
|
|
return Prescription.from_dict(payload)
|
|
rows = PageResult.from_payload(payload, Prescription.from_dict).items
|
|
return rows[0] if rows else None
|
|
|
|
def list_prescription_orders(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""List prescription fulfilment orders without changing their data scope."""
|
|
|
|
payload = self.client.get(
|
|
"tcm.prescriptionOrder/lists",
|
|
_page_params(page_no, page_size, filters),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload, lambda row: dict(row), page_no=page_no, page_size=page_size
|
|
)
|
|
|
|
def get_prescription_order(self, order_id: int) -> dict[str, Any]:
|
|
"""Return one prescription fulfilment order."""
|
|
|
|
return dict(
|
|
_require_mapping(
|
|
self.client.get("tcm.prescriptionOrder/detail", {"id": order_id}),
|
|
"tcm.prescriptionOrder/detail",
|
|
)
|
|
)
|
|
|
|
def create_prescription_order(
|
|
self, payload: Mapping[str, Any] | None = None, **fields: Any
|
|
) -> dict[str, Any]:
|
|
"""Create a fulfilment order from the admin three-step DTO."""
|
|
|
|
body = _body(payload, fields)
|
|
for key in ("prescription_id", "diagnosis_id", "recipient_name", "recipient_phone"):
|
|
if body.get(key) in (None, ""):
|
|
raise ValueError(f"{key} is required")
|
|
result = self.client.post("tcm.prescriptionOrder/create", body)
|
|
return _merge_result(body, result)
|
|
|
|
def list_paid_prescription_orders(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
prescription_order_id: int | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Return selectable paid orders and the server deposit threshold."""
|
|
|
|
params: dict[str, Any] = {"diagnosis_id": diagnosis_id}
|
|
if prescription_order_id is not None:
|
|
params["prescription_order_id"] = prescription_order_id
|
|
payload = self.client.get("tcm.prescriptionOrder/paidPayOrders", params)
|
|
if isinstance(payload, Mapping):
|
|
return dict(payload)
|
|
return {"lists": _mapping_rows(payload), "deposit_min_amount": None}
|
|
|
|
def search_diagnosis_patients(
|
|
self, keyword: str, *, page_no: int = 1, page_size: int = 10
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Search diagnosis patients for prescription-order creation."""
|
|
|
|
payload = self.client.get(
|
|
"tcm.diagnosis/searchPatient",
|
|
_page_params(page_no, page_size, {"keyword": keyword.strip()}),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload, lambda row: dict(row), page_no=page_no, page_size=page_size
|
|
)
|
|
|
|
def get_dictionary(self, dictionary_type: str) -> list[dict[str, Any]]:
|
|
"""Return configuration dictionary options used by order/diagnosis forms.
|
|
|
|
Admin ``config/dict`` returns ``{type_value: [rows...]}`` (see
|
|
``ConfigLogic::getDictByType``). Fall back to page-list envelopes for
|
|
older callers that already unwrap into arrays.
|
|
"""
|
|
|
|
clean = str(dictionary_type or "").strip()
|
|
if not clean:
|
|
raise ValueError("dictionary_type is required")
|
|
payload = self.client.get("config/dict", {"type": clean})
|
|
return _dictionary_rows(payload, clean)
|
|
|
|
def get_dictionaries(
|
|
self, dictionary_types: Sequence[str]
|
|
) -> dict[str, list[dict[str, Any]]]:
|
|
"""Load several dictionaries with one request.
|
|
|
|
``ConfigLogic::getDictByType`` splits ``type`` on commas and answers
|
|
``{type_value: [rows...]}``, so readonly screens can translate every
|
|
dictionary field without one round trip per type.
|
|
"""
|
|
|
|
wanted = [
|
|
str(dictionary_type or "").strip()
|
|
for dictionary_type in dictionary_types
|
|
if str(dictionary_type or "").strip()
|
|
]
|
|
ordered = list(dict.fromkeys(wanted))
|
|
if not ordered:
|
|
return {}
|
|
payload = self.client.get("config/dict", {"type": ",".join(ordered)})
|
|
return {
|
|
dictionary_type: _dictionary_rows(payload, dictionary_type)
|
|
for dictionary_type in ordered
|
|
}
|
|
|
|
def list_chat_notifications(self) -> list[dict[str, Any]]:
|
|
"""Consume the queued chat notifications for the signed-in account.
|
|
|
|
与后台一致：``ChatController::notifications`` 取一次即消费（服务端删除缓存），
|
|
所以每条通知只会被送到一个已登录的客户端一次，不能重复轮询后再补偿。
|
|
"""
|
|
|
|
payload = self.client.get("chat/notifications")
|
|
return _mapping_rows(payload)
|
|
|
|
def list_patients(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[Patient]:
|
|
"""List patients within the server-authoritative first-visit data scope."""
|
|
|
|
request_filters = dict(filters)
|
|
if "status" in request_filters and "status_filter" not in request_filters:
|
|
request_filters["status_filter"] = request_filters.pop("status")
|
|
payload = self.client.get(
|
|
"firstvisit.myPatient/lists",
|
|
_page_params(page_no, page_size, request_filters),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload, Patient.from_dict, page_no=page_no, page_size=page_size
|
|
)
|
|
|
|
def patient_orders(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""List orders in the server-scoped patient order workspace."""
|
|
|
|
payload = self.client.get(
|
|
"firstvisit.myPatient/orders",
|
|
_page_params(page_no, page_size, filters),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload, lambda row: dict(row), page_no=page_no, page_size=page_size
|
|
)
|
|
|
|
def patient_progress(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""List interview progress and preserve all schedule/scope extension data."""
|
|
|
|
payload = self.client.get(
|
|
"firstvisit.myPatient/progress",
|
|
_page_params(page_no, page_size, filters),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload, lambda row: dict(row), page_no=page_no, page_size=page_size
|
|
)
|
|
|
|
def get_patient_order(self, order_id: int) -> dict[str, Any]:
|
|
"""Return an order through the patient-data-scope endpoint."""
|
|
|
|
return dict(
|
|
_require_mapping(
|
|
self.client.get("firstvisit.myPatient/orderDetail", {"id": order_id}),
|
|
"firstvisit.myPatient/orderDetail",
|
|
)
|
|
)
|
|
|
|
def edit_patient_order(
|
|
self,
|
|
order: int | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Any:
|
|
"""Edit an order through the patient-scoped endpoint."""
|
|
|
|
body = _identified_body(order, changes, fields)
|
|
return self.client.post("firstvisit.myPatient/orderEdit", body)
|
|
|
|
def audit_patient_order_prescription(
|
|
self,
|
|
order_id: int,
|
|
action: AuditAction,
|
|
remark: str = "",
|
|
) -> Any:
|
|
"""Audit the prescription side of a scoped patient order."""
|
|
|
|
action = _audit_action(action, remark)
|
|
return self.client.post(
|
|
"firstvisit.myPatient/orderAuditPrescription",
|
|
{"id": order_id, "action": action, "remark": remark.strip()},
|
|
)
|
|
|
|
def revoke_patient_order_prescription_audit(self, order_id: int) -> Any:
|
|
"""Revoke a scoped order's prescription audit."""
|
|
|
|
return self.client.post("firstvisit.myPatient/orderRevokeRxAudit", {"id": order_id})
|
|
|
|
def audit_patient_order_payment(
|
|
self,
|
|
order_id: int,
|
|
action: AuditAction,
|
|
remark: str = "",
|
|
) -> Any:
|
|
"""Audit the payment side of a scoped patient order."""
|
|
|
|
action = _audit_action(action, remark)
|
|
return self.client.post(
|
|
"firstvisit.myPatient/orderAuditPayment",
|
|
{"id": order_id, "action": action, "remark": remark.strip()},
|
|
)
|
|
|
|
def revoke_patient_order_payment_audit(self, order_id: int) -> Any:
|
|
"""Revoke a scoped order's payment audit."""
|
|
|
|
return self.client.post("firstvisit.myPatient/orderRevokePayAudit", {"id": order_id})
|
|
|
|
def update_patient_order_shipping(
|
|
self,
|
|
order_id: int,
|
|
express_company: str,
|
|
tracking_number: str,
|
|
) -> Any:
|
|
"""Update courier metadata regardless of fulfilment state."""
|
|
|
|
return self.client.post(
|
|
"firstvisit.myPatient/orderDdcode",
|
|
{
|
|
"id": order_id,
|
|
"express_company": express_company.strip(),
|
|
"tracking_number": tracking_number.strip(),
|
|
},
|
|
)
|
|
|
|
def ship_patient_order(
|
|
self,
|
|
order_id: int,
|
|
express_company: str,
|
|
tracking_number: str,
|
|
*,
|
|
ship_mode: Literal["gancao", "direct"] | None = None,
|
|
) -> Any:
|
|
"""Advance a scoped order to shipped and store courier metadata."""
|
|
|
|
body: dict[str, Any] = {
|
|
"id": order_id,
|
|
"express_company": express_company.strip(),
|
|
"tracking_number": tracking_number.strip(),
|
|
}
|
|
if ship_mode is not None:
|
|
body["ship_mode"] = ship_mode
|
|
return self.client.post("firstvisit.myPatient/orderShip", body)
|
|
|
|
def add_patient_order_payment(
|
|
self,
|
|
order_id: int,
|
|
order_type: int,
|
|
pay_amount: float,
|
|
*,
|
|
pay_remark: str = "",
|
|
completion_request: int | None = None,
|
|
pay_create_type: Literal["fubei", "express_cod"] | None = None,
|
|
) -> Any:
|
|
"""Add a payment to a shipped scoped order."""
|
|
|
|
body: dict[str, Any] = {
|
|
"id": order_id,
|
|
"order_type": order_type,
|
|
"pay_amount": pay_amount,
|
|
}
|
|
if pay_remark:
|
|
body["pay_remark"] = pay_remark
|
|
if completion_request is not None:
|
|
body["completion_request"] = completion_request
|
|
if pay_create_type is not None:
|
|
body["pay_create_type"] = pay_create_type
|
|
return self.client.post("firstvisit.myPatient/orderAddPayOrder", body)
|
|
|
|
def complete_patient_order(self, order_id: int, fulfillment_status: int) -> Any:
|
|
"""Complete a scoped order with the selected terminal business state."""
|
|
|
|
return self.client.post(
|
|
"firstvisit.myPatient/orderComplete",
|
|
{"id": order_id, "fulfillment_status": fulfillment_status},
|
|
)
|
|
|
|
def refund_patient_order(
|
|
self,
|
|
order_id: int,
|
|
reason: str,
|
|
refund_amount: float | None = None,
|
|
) -> Any:
|
|
"""Refund a scoped patient order; a reason is mandatory."""
|
|
|
|
if not reason.strip():
|
|
raise ValueError("refund reason is required")
|
|
body: dict[str, Any] = {"id": order_id, "reason": reason.strip()}
|
|
if refund_amount is not None:
|
|
body["refund_amount"] = refund_amount
|
|
return self.client.post("firstvisit.myPatient/orderRefund", body)
|
|
|
|
def withdraw_patient_order(self, order_id: int) -> Any:
|
|
"""Withdraw an eligible scoped patient order."""
|
|
|
|
return self.client.post("firstvisit.myPatient/orderWithdraw", {"id": order_id})
|
|
|
|
def upload_patient_order_to_pharmacy(self, order_id: int) -> Any:
|
|
"""Submit an eligible scoped order to its selected pharmacy."""
|
|
|
|
return self.client.post("firstvisit.myPatient/orderUploadToPharmacy", {"id": order_id})
|
|
|
|
def list_patient_assistants(self) -> list[dict[str, Any]]:
|
|
"""List medical assistants available within the current patient scope."""
|
|
|
|
return _mapping_rows(self.client.get("firstvisit.myPatient/assistants"))
|
|
|
|
def assign_patient(
|
|
self,
|
|
diagnosis_id: int,
|
|
assistant_id: int,
|
|
*,
|
|
is_inherit: int | None = None,
|
|
) -> Any:
|
|
"""Assign a patient and optionally inherit the assignment downstream."""
|
|
|
|
body: dict[str, Any] = {"id": diagnosis_id, "assistant_id": assistant_id}
|
|
if is_inherit is not None:
|
|
body["is_inherit"] = is_inherit
|
|
return self.client.post("firstvisit.myPatient/assign", body)
|
|
|
|
def fill_patient_id_card(self, diagnosis_id: int, id_card: str) -> Any:
|
|
"""Fill a patient's identity card using the patient-scope mutation."""
|
|
|
|
if not id_card.strip():
|
|
raise ValueError("id_card is required")
|
|
return self.client.post(
|
|
"firstvisit.myPatient/fillIdCard",
|
|
{"id": diagnosis_id, "id_card": id_card.strip()},
|
|
)
|
|
|
|
def book_patient_appointment(
|
|
self, payload: Mapping[str, Any] | None = None, **fields: Any
|
|
) -> Any:
|
|
"""Create an appointment while preserving the full routed form DTO."""
|
|
|
|
return self.client.post("firstvisit.myPatient/createAppointment", _body(payload, fields))
|
|
|
|
def cancel_patient_appointment(self, appointment_id: int) -> Any:
|
|
"""Cancel an appointment through the patient-scoped endpoint."""
|
|
|
|
return self.client.post("firstvisit.myPatient/cancelAppointment", {"id": appointment_id})
|
|
|
|
def cancel_diagnosis_appointment(self, appointment_id: int) -> Any:
|
|
"""Cancel through the exact route used by the admin diagnosis list."""
|
|
|
|
if appointment_id <= 0:
|
|
raise ValueError("appointment_id must be positive")
|
|
return self.client.post("doctor.appointment/cancel", {"id": appointment_id})
|
|
|
|
def patient_detail(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Compatibility name for permission-aware readonly diagnosis details."""
|
|
|
|
return self.get_diagnosis_detail(diagnosis_id, readonly=True)
|
|
|
|
def list_consultations(
|
|
self, *, page_no: int = 1, page_size: int = 20, **filters: Any
|
|
) -> PageResult[Consultation]:
|
|
"""List diagnosis records using ``tcm.diagnosis/lists``."""
|
|
|
|
request_filters = dict(filters)
|
|
department_id = request_filters.pop("department_id", None)
|
|
if department_id not in (None, ""):
|
|
request_filters.setdefault("assistant_dept_id", department_id)
|
|
start_date = str(request_filters.pop("start_date", "") or "").strip()
|
|
end_date = str(request_filters.pop("end_date", "") or "").strip()
|
|
if start_date and start_date == end_date:
|
|
request_filters.setdefault("appointment_date", start_date)
|
|
else:
|
|
if start_date:
|
|
request_filters.setdefault("latest_appointment_start_date", start_date)
|
|
if end_date:
|
|
request_filters.setdefault("latest_appointment_end_date", end_date)
|
|
if request_filters.get("status") not in (None, ""):
|
|
status = _to_int(request_filters.pop("status"), 0)
|
|
if status == 3:
|
|
request_filters.setdefault("completed_appointment", 1)
|
|
else:
|
|
request_filters.setdefault("appointment_status", status)
|
|
payload = self.client.get(
|
|
"tcm.diagnosis/lists",
|
|
_page_params(page_no, page_size, request_filters),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload, Consultation.from_dict, page_no=page_no, page_size=page_size
|
|
)
|
|
|
|
def list_ai_patient_options(
|
|
self, *, page_no: int = 1, page_size: int = 20, keyword: str = ""
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""List privacy-safe diagnosis choices through the dedicated AI endpoint."""
|
|
|
|
clean_keyword = str(keyword or "").strip()
|
|
payload = self.client.get(
|
|
"tcm.diagnosis/aiPatientOptions",
|
|
_page_params(page_no, page_size, {"keyword": clean_keyword}),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload,
|
|
_ai_patient_option,
|
|
page_no=page_no,
|
|
page_size=page_size,
|
|
)
|
|
|
|
def get_diagnosis_detail(self, diagnosis_id: int, *, readonly: bool = False) -> dict[str, Any]:
|
|
"""Load a full diagnosis using the appropriate routed endpoint."""
|
|
|
|
endpoint = "tcm.diagnosis/readonlyDetail" if readonly else "tcm.diagnosis/detail"
|
|
return dict(
|
|
_require_mapping(
|
|
self.client.get(endpoint, {"id": diagnosis_id}),
|
|
endpoint,
|
|
)
|
|
)
|
|
|
|
def diagnosis_readonly_detail(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Alias for the hidden permission-aware readonly diagnosis route."""
|
|
|
|
return self.get_diagnosis_detail(diagnosis_id, readonly=True)
|
|
|
|
def create_diagnosis(
|
|
self, diagnosis: Mapping[str, Any] | None = None, **fields: Any
|
|
) -> dict[str, Any]:
|
|
"""Create a diagnosis using the complete edit-form mapping."""
|
|
|
|
body = _body(diagnosis, fields)
|
|
result = self.client.post("tcm.diagnosis/add", body)
|
|
return _merge_result(body, result)
|
|
|
|
def update_diagnosis(
|
|
self,
|
|
diagnosis: int | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> dict[str, Any]:
|
|
"""Edit a diagnosis without discarding forward-compatible fields."""
|
|
|
|
body = _identified_body(diagnosis, changes, fields)
|
|
result = self.client.post("tcm.diagnosis/edit", body)
|
|
return _merge_result(body, result)
|
|
|
|
def delete_diagnosis(self, diagnosis_id: int) -> Any:
|
|
"""Delete a diagnosis through its canonical endpoint."""
|
|
|
|
return self.client.post("tcm.diagnosis/delete", {"id": diagnosis_id})
|
|
|
|
def set_revisit_slot_start_offset(self, diagnosis_id: int, offset: int) -> Any:
|
|
"""Set the diagnosis revisit-statistics starting offset."""
|
|
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
if offset not in range(0, 21):
|
|
raise ValueError("revisit_slot_start_offset must be between 0 and 20")
|
|
|
|
return self.client.post(
|
|
"tcm.diagnosis/setRevisitSlotStartOffset",
|
|
{"id": diagnosis_id, "revisit_slot_start_offset": offset},
|
|
)
|
|
|
|
def appointment_history(
|
|
self, diagnosis_id: int, *, page_no: int = 1, page_size: int = 500
|
|
) -> PageResult[Appointment]:
|
|
"""Load appointment history with the audited relaxed diagnosis scope."""
|
|
|
|
payload = self.client.get(
|
|
"doctor.appointment/lists",
|
|
_page_params(
|
|
page_no,
|
|
page_size,
|
|
{"diagnosis_id": diagnosis_id, "diag_scope_relax": 1},
|
|
),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload, Appointment.from_dict, page_no=page_no, page_size=page_size
|
|
)
|
|
|
|
def assign_history(
|
|
self, diagnosis_id: int, *, page_no: int = 1, page_size: int = 20
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Load diagnosis assignment history."""
|
|
|
|
payload = self.client.get("tcm.diagnosis/assignLogList", {"id": diagnosis_id})
|
|
return PageResult.from_payload(
|
|
payload, lambda row: dict(row), page_no=page_no, page_size=page_size
|
|
)
|
|
|
|
def assign_diagnosis(
|
|
self,
|
|
diagnosis_id: int,
|
|
assistant_id: int,
|
|
*,
|
|
is_inherit: int | None = None,
|
|
) -> Any:
|
|
"""Assign a diagnosis through the general diagnosis endpoint."""
|
|
|
|
body: dict[str, Any] = {"id": diagnosis_id, "assistant_id": assistant_id}
|
|
if is_inherit is not None:
|
|
body["is_inherit"] = is_inherit
|
|
return self.client.post("tcm.diagnosis/assign", body)
|
|
|
|
def list_diagnosis_assistants(self) -> list[dict[str, Any]]:
|
|
"""Return assistants available to diagnosis assignment."""
|
|
|
|
return _mapping_rows(self.client.get("tcm.diagnosis/getAssistants"))
|
|
|
|
def list_diagnosis_doctors(self) -> list[dict[str, Any]]:
|
|
"""Return doctors available to diagnosis forms and appointments."""
|
|
|
|
return _mapping_rows(self.client.get("tcm.diagnosis/getDoctors"))
|
|
|
|
def check_diagnosis_phone(self, payload: Mapping[str, Any]) -> Any:
|
|
"""Check diagnosis phone uniqueness with the unmodified form contract."""
|
|
|
|
return self.client.post("tcm.diagnosis/checkPhone", dict(payload))
|
|
|
|
def check_diagnosis_id_card(self, payload: Mapping[str, Any]) -> Any:
|
|
"""Check diagnosis identity-card uniqueness."""
|
|
|
|
return self.client.post("tcm.diagnosis/checkIdCard", dict(payload))
|
|
|
|
def fill_diagnosis_id_card(self, diagnosis_id: int, id_card: str) -> Any:
|
|
"""Fill an identity card through the general diagnosis endpoint."""
|
|
|
|
return self.client.post(
|
|
"tcm.diagnosis/fillIdCard",
|
|
{"id": diagnosis_id, "id_card": id_card.strip()},
|
|
)
|
|
|
|
def get_tracking_window(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
start_date: str = "",
|
|
end_date: str = "",
|
|
) -> dict[str, Any]:
|
|
"""Load lazy blood, diet and exercise tracking data for a date window."""
|
|
|
|
return dict(
|
|
_require_mapping(
|
|
self.client.get(
|
|
"tcm.diagnosis/trackingWindow",
|
|
{
|
|
"id": diagnosis_id,
|
|
"start_date": start_date,
|
|
"end_date": end_date,
|
|
},
|
|
),
|
|
"tcm.diagnosis/trackingWindow",
|
|
)
|
|
)
|
|
|
|
def list_tracking_notes(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
"""Return date-grouped tracking notes newest first."""
|
|
|
|
return _mapping_rows(
|
|
self.client.get("tcm.diagnosis/trackingNotes", {"diagnosis_id": diagnosis_id})
|
|
)
|
|
|
|
def add_tracking_note(self, diagnosis_id: int, content: str) -> Any:
|
|
"""Append a textual daily tracking note."""
|
|
|
|
clean_content = content.strip()
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
if not clean_content or len(clean_content) > 1000:
|
|
raise ValueError("tracking_content must contain 1 to 1000 characters")
|
|
return self.client.post(
|
|
"tcm.diagnosis/addTrackingNote",
|
|
{"diagnosis_id": diagnosis_id, "tracking_content": clean_content},
|
|
)
|
|
|
|
def add_blood_record(self, payload: Mapping[str, Any] | None = None, **fields: Any) -> Any:
|
|
"""Create a blood record using the fields consumed by ``DailyMatrix``."""
|
|
|
|
return self.client.post("tcm.bloodRecord/add", _daily_record_body("blood", payload, fields))
|
|
|
|
def update_blood_record(
|
|
self,
|
|
record: int | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Any:
|
|
"""Update a blood record without weakening add-time validation."""
|
|
|
|
body = _identified_body(record, changes, fields)
|
|
return self.client.post("tcm.bloodRecord/edit", _daily_record_body("blood", body, {}))
|
|
|
|
def add_diet_record(self, payload: Mapping[str, Any] | None = None, **fields: Any) -> Any:
|
|
"""Create a breakfast/lunch/dinner daily record."""
|
|
|
|
return self.client.post("tcm.dietRecord/add", _daily_record_body("diet", payload, fields))
|
|
|
|
def update_diet_record(
|
|
self,
|
|
record: int | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Any:
|
|
"""Update a diet record."""
|
|
|
|
body = _identified_body(record, changes, fields)
|
|
return self.client.post("tcm.dietRecord/edit", _daily_record_body("diet", body, {}))
|
|
|
|
def add_exercise_record(self, payload: Mapping[str, Any] | None = None, **fields: Any) -> Any:
|
|
"""Create a validated exercise record."""
|
|
|
|
return self.client.post(
|
|
"tcm.exerciseRecord/add", _daily_record_body("exercise", payload, fields)
|
|
)
|
|
|
|
def update_exercise_record(
|
|
self,
|
|
record: int | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> Any:
|
|
"""Update an exercise record."""
|
|
|
|
body = _identified_body(record, changes, fields)
|
|
return self.client.post("tcm.exerciseRecord/edit", _daily_record_body("exercise", body, {}))
|
|
|
|
def list_diagnosis_todos(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
page_no: int = 1,
|
|
page_size: int = 20,
|
|
status: int | None = None,
|
|
creator_id: int | None = None,
|
|
) -> PageResult[dict[str, Any]]:
|
|
"""Return diagnosis follow-up todos."""
|
|
|
|
payload = self.client.get(
|
|
"tcm.diagnosisTodo/lists",
|
|
_page_params(
|
|
page_no,
|
|
page_size,
|
|
{
|
|
"diagnosis_id": diagnosis_id,
|
|
"status": status,
|
|
"creator_id": creator_id,
|
|
},
|
|
),
|
|
)
|
|
return PageResult.from_payload(
|
|
payload, lambda row: dict(row), page_no=page_no, page_size=page_size
|
|
)
|
|
|
|
def add_diagnosis_todo(self, diagnosis_id: int, content: str, remind_time: int) -> Any:
|
|
"""Create a follow-up todo using a Unix-seconds reminder."""
|
|
|
|
clean_content = content.strip()
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
if not clean_content or len(clean_content) > 500:
|
|
raise ValueError("todo content must contain 1 to 500 characters")
|
|
if remind_time <= int(time.time()) + 30:
|
|
raise ValueError("remind_time must be at least 30 seconds in the future")
|
|
|
|
return self.client.post(
|
|
"tcm.diagnosisTodo/add",
|
|
{
|
|
"diagnosis_id": diagnosis_id,
|
|
"content": clean_content,
|
|
"remind_time": remind_time,
|
|
},
|
|
)
|
|
|
|
def cancel_diagnosis_todo(self, todo_id: int) -> Any:
|
|
"""Cancel an outstanding diagnosis todo."""
|
|
|
|
if todo_id <= 0:
|
|
raise ValueError("todo_id must be positive")
|
|
|
|
return self.client.post("tcm.diagnosisTodo/cancel", {"id": todo_id})
|
|
|
|
def list_call_records(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
"""Load the confirmed replay list endpoint."""
|
|
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
return _mapping_rows(
|
|
self.client.get("tcm.diagnosis/getCallRecords", {"diagnosis_id": diagnosis_id})
|
|
)
|
|
|
|
def create_manual_call_record(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Create a synthetic call record before a toolbar video upload."""
|
|
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
result = self.client.post(
|
|
"tcm.diagnosis/createManualCallRecord", {"diagnosis_id": diagnosis_id}
|
|
)
|
|
return dict(_require_mapping(result, "tcm.diagnosis/createManualCallRecord"))
|
|
|
|
def attach_local_call_recording(
|
|
self,
|
|
diagnosis_id: int,
|
|
file_url: str,
|
|
*,
|
|
call_record_id: int | None = None,
|
|
) -> Any:
|
|
"""Attach an uploaded replay URI to the selected call record."""
|
|
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
clean_url = file_url.strip()
|
|
if not clean_url:
|
|
raise ValueError("file_url is required")
|
|
body: dict[str, Any] = {"diagnosis_id": diagnosis_id, "file_url": clean_url}
|
|
if call_record_id is not None:
|
|
if call_record_id <= 0:
|
|
raise ValueError("call_record_id must be positive")
|
|
body["call_record_id"] = call_record_id
|
|
return self.client.post("tcm.diagnosis/attachLocalCallRecording", body)
|
|
|
|
def upload_call_recording(
|
|
self,
|
|
path: str | PathLike[str],
|
|
diagnosis_id: int,
|
|
*,
|
|
call_record_id: int | None = None,
|
|
mime_type: str = "audio/webm",
|
|
) -> dict[str, Any]:
|
|
"""Chunk-upload one local audio recording and attach it atomically."""
|
|
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
if call_record_id is not None and call_record_id <= 0:
|
|
raise ValueError("call_record_id must be positive")
|
|
recording_path = Path(path).expanduser().resolve()
|
|
if not recording_path.is_file():
|
|
raise ValueError(f"recording file does not exist: {recording_path}")
|
|
file_size = recording_path.stat().st_size
|
|
if file_size <= 0:
|
|
raise ValueError("recording file must not be empty")
|
|
|
|
chunk_size = 4 * 1024 * 1024
|
|
chunk_total = (file_size + chunk_size - 1) // chunk_size
|
|
upload_id = f"local_audio_{uuid4().hex}"
|
|
file_name = recording_path.name
|
|
clean_mime = str(mime_type or "audio/webm").strip()[:120] or "audio/webm"
|
|
final_result: Mapping[str, Any] | None = None
|
|
bound_call_record_id = call_record_id
|
|
with recording_path.open("rb") as stream:
|
|
for chunk_index in range(chunk_total):
|
|
content = stream.read(chunk_size)
|
|
if not content:
|
|
raise ApiProtocolError("local recording ended before all chunks were read")
|
|
result = self.client.post_multipart(
|
|
"tcm.diagnosis/uploadCallRecording",
|
|
files={"file": (file_name, content, clean_mime)},
|
|
data={
|
|
"diagnosis_id": diagnosis_id,
|
|
"call_record_id": bound_call_record_id or 0,
|
|
"upload_id": upload_id,
|
|
"file_name": file_name,
|
|
"file_size": file_size,
|
|
"mime_type": clean_mime,
|
|
"chunk_index": chunk_index,
|
|
"chunk_total": chunk_total,
|
|
},
|
|
)
|
|
final_result = _require_mapping(
|
|
result,
|
|
"tcm.diagnosis/uploadCallRecording",
|
|
)
|
|
returned_id = int(final_result.get("call_record_id") or 0)
|
|
if returned_id <= 0:
|
|
raise ApiProtocolError(
|
|
"uploadCallRecording returned no call_record_id"
|
|
)
|
|
if bound_call_record_id is None:
|
|
bound_call_record_id = returned_id
|
|
elif returned_id != bound_call_record_id:
|
|
raise ApiProtocolError(
|
|
"uploadCallRecording returned a different call_record_id"
|
|
)
|
|
|
|
if final_result is None or not bool(final_result.get("completed")):
|
|
raise ApiProtocolError("uploadCallRecording did not complete the local audio upload")
|
|
file_url = str(final_result.get("file_url") or "").strip()
|
|
if not file_url:
|
|
raise ApiProtocolError("uploadCallRecording returned no COS file URL")
|
|
return dict(final_result)
|
|
|
|
def list_im_chat_messages(
|
|
self, diagnosis_id: int, *, only_archived: bool = True
|
|
) -> dict[str, Any]:
|
|
"""Load archive-only messages unless the caller explicitly requests live merging."""
|
|
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
payload = self.client.get(
|
|
"tcm.diagnosis/getImChatMessages",
|
|
{"diagnosis_id": diagnosis_id, "only_archived": int(only_archived)},
|
|
)
|
|
return dict(_require_mapping(payload, "tcm.diagnosis/getImChatMessages"))
|
|
|
|
def sync_im_chat_messages(self, diagnosis_id: int) -> Any:
|
|
"""Queue the confirmed shutdown-function archive sync endpoint."""
|
|
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
return self.client.post("tcm.diagnosis/triggerImChatSync", {"diagnosis_id": diagnosis_id})
|
|
|
|
def list_diagnosis_appointment_logs(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
"""Return registration/cancellation logs from ``guahaoLogList``."""
|
|
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
return _mapping_rows(self.client.get("tcm.diagnosis/guahaoLogList", {"id": diagnosis_id}))
|
|
|
|
def generate_mini_program_qrcode(
|
|
self, payload: Mapping[str, Any] | None = None, **fields: Any
|
|
) -> dict[str, Any]:
|
|
"""Generate either QR variant using the server's shared DTO."""
|
|
|
|
body = _body(payload, fields)
|
|
for key in ("patient_id", "share_user_id"):
|
|
if int(body.get(key) or 0) <= 0:
|
|
raise ValueError(f"{key} must be positive")
|
|
page = str(body.get("mini_program_path") or "").strip()
|
|
if page == "pages/login/login":
|
|
if int(body.get("doctor_id") or 0) <= 0:
|
|
raise ValueError("doctor_id must be positive for a video QR code")
|
|
elif int(body.get("diagnosis_id") or 0) <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
payload_result = self.client.post("tcm.diagnosis/generateMiniProgramQrcode", body)
|
|
result = dict(_require_mapping(payload_result, "tcm.diagnosis/generateMiniProgramQrcode"))
|
|
if not str(result.get("qrcode_url") or "").strip():
|
|
raise ApiProtocolError(
|
|
"tcm.diagnosis/generateMiniProgramQrcode returned no qrcode_url",
|
|
data=result,
|
|
)
|
|
return result
|
|
|
|
def get_mini_program_config(self) -> dict[str, Any]:
|
|
"""Read the same mini-program readiness check used by the admin page."""
|
|
|
|
payload = self.client.get("channel.mnp_settings/getConfig")
|
|
return dict(_require_mapping(payload, "channel.mnp_settings/getConfig"))
|
|
|
|
def download_public_image(self, url: str) -> bytes:
|
|
"""Download a generated QR image without expecting an API envelope."""
|
|
|
|
return self.client.get_bytes(url)
|
|
|
|
def generate_video_qrcode(
|
|
self,
|
|
doctor_id: int,
|
|
patient_id: int,
|
|
share_user_id: int,
|
|
*,
|
|
diagnosis_id: int | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Generate the video QR with the admin's confirmed login-page DTO."""
|
|
|
|
fields: dict[str, Any] = {
|
|
"doctor_id": doctor_id,
|
|
"patient_id": patient_id,
|
|
"share_user_id": share_user_id,
|
|
"mini_program_path": "pages/login/login",
|
|
}
|
|
if diagnosis_id is not None and diagnosis_id > 0:
|
|
fields["diagnosis_id"] = diagnosis_id
|
|
return self.generate_mini_program_qrcode(
|
|
**fields,
|
|
)
|
|
|
|
def generate_diagnosis_qrcode(
|
|
self, diagnosis_id: int, doctor_id: int, patient_id: int, share_user_id: int
|
|
) -> dict[str, Any]:
|
|
"""Generate the diagnosis-confirmation QR."""
|
|
|
|
return self.generate_mini_program_qrcode(
|
|
diagnosis_id=diagnosis_id,
|
|
doctor_id=doctor_id,
|
|
patient_id=patient_id,
|
|
share_user_id=share_user_id,
|
|
)
|
|
|
|
def list_appointment_logs(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
"""Compatibility name used by the diagnosis-list menu."""
|
|
|
|
return self.list_diagnosis_appointment_logs(diagnosis_id)
|
|
|
|
def create_diagnosis_order(
|
|
self,
|
|
patient_id: int,
|
|
order_type: int,
|
|
amount: float,
|
|
*,
|
|
remark: str = "",
|
|
) -> dict[str, Any]:
|
|
"""Create the generic order used by the diagnosis-list action."""
|
|
|
|
if patient_id <= 0:
|
|
raise ValueError("patient_id must be positive")
|
|
if order_type not in range(1, 9):
|
|
raise ValueError("order_type must be between 1 and 8")
|
|
if amount <= 0:
|
|
raise ValueError("amount must be positive")
|
|
payload = self.client.post(
|
|
"order.order/create",
|
|
{
|
|
"patient_id": patient_id,
|
|
"order_type": order_type,
|
|
"amount": round(float(amount), 2),
|
|
"remark": remark.strip(),
|
|
},
|
|
)
|
|
return dict(_require_mapping(payload, "order.order/create"))
|
|
|
|
def generate_order_qrcode(self, order_no: str) -> dict[str, Any]:
|
|
"""Generate the QR shown immediately after a generic order is created."""
|
|
|
|
clean_order_no = order_no.strip()
|
|
if not clean_order_no:
|
|
raise ValueError("order_no is required")
|
|
payload = self.client.post(
|
|
"tcm.diagnosis/generateOrderQrcode", {"order_no": clean_order_no}
|
|
)
|
|
result = dict(_require_mapping(payload, "tcm.diagnosis/generateOrderQrcode"))
|
|
if not str(result.get("qrcode_url") or "").strip():
|
|
raise ApiProtocolError(
|
|
"tcm.diagnosis/generateOrderQrcode returned no qrcode_url", data=result
|
|
)
|
|
return result
|
|
|
|
def get_call_ticket(self, patient_id: int, diagnosis_id: int, *, appointment_id: int = 0) -> CallTicket:
|
|
"""Obtain short-lived Tencent credentials for a consultation call."""
|
|
|
|
result = _require_mapping(
|
|
self.client.post(
|
|
"tcm.diagnosis/getCallSignature",
|
|
{"patient_id": patient_id, "diagnosis_id": diagnosis_id, "appointment_id": appointment_id},
|
|
),
|
|
"tcm.diagnosis/getCallSignature",
|
|
)
|
|
policy_fields = {
|
|
"appointment_id", "appointmentId", "appointment_type", "appointment_type_desc",
|
|
"can_video_call", "can_audio_call", "call_disabled_reason",
|
|
}
|
|
if appointment_id > 0 and not policy_fields.intersection(result):
|
|
# Older signature endpoints authenticate the patient/diagnosis but
|
|
# do not echo the appointment. Keep the requested chat context;
|
|
# never treat it as server authorization for an audio/video call.
|
|
if (
|
|
str(result.get("diagnosis_id", "")).strip() != str(diagnosis_id)
|
|
or str(result.get("patient_id", "")).strip() != str(patient_id)
|
|
or result.get("patientUserId") != f"patient_{patient_id}"
|
|
):
|
|
raise ValueError("聊天凭证与当前患者或诊单不匹配，请重新打开聊天窗口。")
|
|
result = dict(result)
|
|
result.update(
|
|
appointment_id=appointment_id,
|
|
can_video_call=False,
|
|
can_audio_call=False,
|
|
call_disabled_reason="暂未获取到本次挂号的通话权限，仍可图文聊天",
|
|
)
|
|
if appointment_id > 0:
|
|
returned_id = result.get("appointment_id")
|
|
if (
|
|
isinstance(returned_id, bool)
|
|
or not isinstance(returned_id, (int, str))
|
|
or str(returned_id).strip() != str(appointment_id)
|
|
):
|
|
raise ValueError("通话凭证与本次挂号不匹配，请重新打开聊天窗口。")
|
|
ticket = CallTicket.from_dict(result)
|
|
if ticket.diagnosis_id is None:
|
|
ticket.diagnosis_id = diagnosis_id
|
|
return ticket
|
|
|
|
def start_call(self, diagnosis_id: int, patient_id: int, *, call_type: int = 2, appointment_id: int = 0) -> Any:
|
|
"""Create the server-side call record before ringing participants."""
|
|
|
|
payload = self.client.post(
|
|
"tcm.diagnosis/startCall",
|
|
{
|
|
"diagnosis_id": diagnosis_id,
|
|
"patient_id": patient_id,
|
|
"call_type": call_type,
|
|
"appointment_id": appointment_id,
|
|
},
|
|
)
|
|
if not isinstance(payload, Mapping):
|
|
raise ApiProtocolError(
|
|
"tcm.diagnosis/startCall returned no valid call_record_id object", data=payload
|
|
)
|
|
raw_id = next(
|
|
(
|
|
payload[key]
|
|
for key in ("call_record_id", "callRecordId", "id")
|
|
if key in payload
|
|
),
|
|
None,
|
|
)
|
|
try:
|
|
if isinstance(raw_id, bool):
|
|
raise ValueError
|
|
call_record_id = int(raw_id)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ApiProtocolError(
|
|
"tcm.diagnosis/startCall returned no valid call_record_id",
|
|
data=dict(payload),
|
|
) from exc
|
|
if call_record_id <= 0:
|
|
raise ApiProtocolError(
|
|
"tcm.diagnosis/startCall returned no valid call_record_id",
|
|
data=dict(payload),
|
|
)
|
|
return {"call_record_id": call_record_id}
|
|
|
|
@staticmethod
|
|
def _optional_call_record_id(call_record_id: int | str | None) -> int | str | None:
|
|
if call_record_id is None:
|
|
return None
|
|
if isinstance(call_record_id, bool) or not str(call_record_id).strip():
|
|
raise ValueError("call_record_id must be a positive identifier")
|
|
try:
|
|
normalized = int(str(call_record_id).strip())
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError("call_record_id must be a positive identifier") from exc
|
|
if normalized <= 0:
|
|
raise ValueError("call_record_id must be a positive identifier")
|
|
return normalized
|
|
|
|
def end_call(
|
|
self, diagnosis_id: int, *, call_record_id: int | str | None = None
|
|
) -> Any:
|
|
"""Stop COS cloud recording and end the exact call record."""
|
|
|
|
body: dict[str, Any] = {"diagnosis_id": diagnosis_id}
|
|
normalized_id = self._optional_call_record_id(call_record_id)
|
|
if normalized_id is not None:
|
|
body["call_record_id"] = normalized_id
|
|
return self.client.post("tcm.diagnosis/endCall", body)
|
|
|
|
def bind_call_room(
|
|
self,
|
|
diagnosis_id: int,
|
|
room_id: str,
|
|
*,
|
|
call_record_id: int | str | None = None,
|
|
) -> Any:
|
|
"""Bind the actual TRTC room and auto-start COS cloud recording."""
|
|
|
|
if not room_id.strip():
|
|
raise ValueError("room_id is required")
|
|
body: dict[str, Any] = {
|
|
"diagnosis_id": diagnosis_id,
|
|
"room_id": room_id.strip(),
|
|
}
|
|
normalized_id = self._optional_call_record_id(call_record_id)
|
|
if normalized_id is not None:
|
|
body["call_record_id"] = normalized_id
|
|
return self.client.post(
|
|
"tcm.diagnosis/bindCallRoom",
|
|
body,
|
|
)
|
|
|
|
@staticmethod
|
|
def _transcription_identity(
|
|
diagnosis_id: int,
|
|
call_record_id: int | str,
|
|
transcription_session_id: str,
|
|
) -> dict[str, Any]:
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
if isinstance(call_record_id, bool) or not str(call_record_id).strip():
|
|
raise ValueError("call_record_id is required")
|
|
clean_session = transcription_session_id.strip()
|
|
if not clean_session or len(clean_session) > 128:
|
|
raise ValueError("transcription_session_id must contain 1 to 128 characters")
|
|
return {
|
|
"diagnosis_id": diagnosis_id,
|
|
"call_record_id": call_record_id,
|
|
"transcription_session_id": clean_session,
|
|
}
|
|
|
|
def start_call_transcription(
|
|
self,
|
|
diagnosis_id: int,
|
|
call_record_id: int | str,
|
|
transcription_session_id: str,
|
|
*,
|
|
language: str = "zh-CN",
|
|
) -> Any:
|
|
"""Create the server transcript session for one call record."""
|
|
|
|
body = self._transcription_identity(
|
|
diagnosis_id, call_record_id, transcription_session_id
|
|
)
|
|
body["language"] = language.strip()[:32] or "zh-CN"
|
|
return self.client.post("tcm.diagnosis/startCallTranscription", body)
|
|
|
|
def upsert_call_transcript_segments(
|
|
self,
|
|
diagnosis_id: int,
|
|
call_record_id: int | str,
|
|
transcription_session_id: str,
|
|
segments: Sequence[Mapping[str, Any]],
|
|
) -> Any:
|
|
"""Upsert a bounded batch using each segment_id as the idempotency key."""
|
|
|
|
body = self._transcription_identity(
|
|
diagnosis_id, call_record_id, transcription_session_id
|
|
)
|
|
normalized: list[dict[str, Any]] = []
|
|
if len(segments) > 50:
|
|
raise ValueError("at most 50 transcript segments may be submitted at once")
|
|
for segment in segments:
|
|
segment_id = str(segment.get("segment_id") or "").strip()
|
|
text = str(segment.get("text") or "").strip()
|
|
if not segment_id or len(segment_id) > 160:
|
|
raise ValueError("transcript segment_id is invalid")
|
|
if not text or len(text) > 4_000:
|
|
raise ValueError("transcript text is invalid")
|
|
normalized.append(
|
|
{
|
|
"segment_id": segment_id,
|
|
"speaker_user_id": str(segment.get("speaker_user_id") or "")[:160],
|
|
"speaker_role": str(segment.get("speaker_role") or "unknown")[:20],
|
|
"timestamp": max(int(segment.get("timestamp") or 0), 0),
|
|
"text": text,
|
|
}
|
|
)
|
|
if not normalized:
|
|
raise ValueError("at least one transcript segment is required")
|
|
body["segments"] = normalized
|
|
return self.client.post("tcm.diagnosis/upsertCallTranscriptSegments", body)
|
|
|
|
def finish_call_transcription(
|
|
self,
|
|
diagnosis_id: int,
|
|
call_record_id: int | str,
|
|
transcription_session_id: str,
|
|
expected_segment_count: int,
|
|
*,
|
|
status: str = "completed",
|
|
) -> Any:
|
|
"""Finalize a transcript; repeated requests use the same session identity."""
|
|
|
|
body = self._transcription_identity(
|
|
diagnosis_id, call_record_id, transcription_session_id
|
|
)
|
|
clean_status = status.strip().lower()
|
|
if clean_status not in {"completed", "partial", "failed"}:
|
|
raise ValueError("transcription status is invalid")
|
|
if expected_segment_count < 0:
|
|
raise ValueError("expected_segment_count must not be negative")
|
|
body.update(
|
|
{
|
|
"expected_segment_count": expected_segment_count,
|
|
"status": clean_status,
|
|
}
|
|
)
|
|
return self.client.post("tcm.diagnosis/finishCallTranscription", body)
|
|
|
|
# Compatibility aliases keep UI naming independent from endpoint history.
|
|
def my_self(self) -> Session:
|
|
"""Compatibility alias for :meth:`get_session`."""
|
|
|
|
return self.get_session()
|
|
|
|
def reception(self, appointment_id: int) -> dict[str, Any]:
|
|
"""Compatibility alias for :meth:`get_reception`."""
|
|
|
|
return self.get_reception(appointment_id)
|
|
|
|
def add_prescription_template(
|
|
self,
|
|
template: PrescriptionTemplate | Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> PrescriptionTemplate:
|
|
"""Compatibility alias for :meth:`create_prescription_template`."""
|
|
|
|
return self.create_prescription_template(template, **fields)
|
|
|
|
def edit_prescription_template(
|
|
self,
|
|
template: int | PrescriptionTemplate | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> PrescriptionTemplate:
|
|
"""Compatibility alias for :meth:`update_prescription_template`."""
|
|
|
|
return self.update_prescription_template(template, changes, **fields)
|
|
|
|
|
|
def _diagnosis_ai_request(diagnosis_id: int, prompt: str, task: str) -> tuple[str, str]:
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
clean_prompt = prompt.strip()
|
|
if not clean_prompt:
|
|
raise ValueError("prompt is required")
|
|
if len(clean_prompt) > 500:
|
|
raise ValueError("prompt must not exceed 500 characters")
|
|
clean_task = task.strip().lower() or "custom"
|
|
if clean_task not in {
|
|
"summary",
|
|
"tcm_pattern",
|
|
"prescription_review",
|
|
"prescription_generate",
|
|
"medication_review",
|
|
"exam_review",
|
|
"complication_risk",
|
|
"guideline_review",
|
|
"custom",
|
|
}:
|
|
raise ValueError("task is not supported")
|
|
return clean_prompt, clean_task
|
|
|
|
|
|
def _normalise_diagnosis_ai_event(raw: Mapping[str, Any]) -> dict[str, Any] | None:
|
|
kind = str(raw.get("event") or "message").strip().lower()
|
|
data = raw.get("data")
|
|
payload = dict(data) if isinstance(data, Mapping) else {}
|
|
if kind == "message":
|
|
kind = str(payload.get("event") or payload.get("type") or "message").lower()
|
|
if kind == "start":
|
|
payload.pop("event", None)
|
|
payload.pop("type", None)
|
|
return {**payload, "event": "start"}
|
|
if kind == "delta":
|
|
text = (
|
|
data
|
|
if isinstance(data, str)
|
|
else payload.get("delta")
|
|
or payload.get("content")
|
|
or payload.get("text")
|
|
or ""
|
|
)
|
|
if not isinstance(text, str):
|
|
raise ApiProtocolError("AI assistant delta content must be text", data=data)
|
|
return {"event": "delta", "text": text}
|
|
if kind == "done":
|
|
payload.pop("event", None)
|
|
payload.pop("type", None)
|
|
return {**payload, "event": "done"}
|
|
if kind == "error":
|
|
message = (
|
|
data
|
|
if isinstance(data, str)
|
|
else payload.get("message") or payload.get("msg") or payload.get("error")
|
|
)
|
|
raise ApiBusinessError(str(message or "AI assistant stream failed"), data=data)
|
|
return None
|
|
|
|
|
|
def _page_params(page_no: int, page_size: int, filters: Mapping[str, Any]) -> dict[str, Any]:
|
|
if page_no < 1 or page_size < 1:
|
|
raise ValueError("page_no and page_size must be positive")
|
|
result = {
|
|
key: value
|
|
for key, value in filters.items()
|
|
if value is not None and not (isinstance(value, (list, tuple, set)) and not value)
|
|
}
|
|
result.update({"page_no": page_no, "page_size": page_size})
|
|
return result
|
|
|
|
|
|
def _ai_patient_option(row: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""Project one AI patient option without exposing a plaintext phone number."""
|
|
|
|
diagnosis_id = _to_int(row.get("diagnosis_id", row.get("id")), 0)
|
|
source_patient_id = _to_int(row.get("source_patient_id", row.get("patient_id")), 0)
|
|
phone = str(row.get("phone_masked") or "").strip()
|
|
digits = re.sub(r"\D", "", phone)
|
|
if len(digits) >= 7 and "*" not in phone:
|
|
phone = f"{digits[:3]}****{digits[-4:]}"
|
|
option: dict[str, Any] = {
|
|
"diagnosis_id": diagnosis_id,
|
|
"source_patient_id": source_patient_id,
|
|
"patient_name": str(row.get("patient_name", row.get("name")) or "").strip(),
|
|
"phone_masked": phone,
|
|
}
|
|
for key in (
|
|
"gender",
|
|
"gender_desc",
|
|
"age",
|
|
"diagnosis_date",
|
|
"diagnosis_summary",
|
|
"syndrome_type",
|
|
"clinical_diagnosis",
|
|
"last_visit_at",
|
|
"next_appointment_at",
|
|
):
|
|
if key in row:
|
|
option[key] = row[key]
|
|
return option
|
|
|
|
|
|
def _require_mapping(value: object, endpoint: str) -> Mapping[str, Any]:
|
|
if not isinstance(value, Mapping):
|
|
raise ApiProtocolError(f"{endpoint} returned an object of the wrong shape", data=value)
|
|
return value
|
|
|
|
|
|
def _client_request(
|
|
client: Any,
|
|
method: str,
|
|
endpoint: str,
|
|
payload: Mapping[str, Any] | None = None,
|
|
*,
|
|
timeout: float | None = None,
|
|
) -> Any:
|
|
"""Call get/post once, omitting timeout for legacy test doubles.
|
|
|
|
Capability is determined before invocation. Catching ``TypeError`` from
|
|
the call itself is unsafe for POST because the transport may already have
|
|
submitted the write before raising, and a fallback call would duplicate it.
|
|
"""
|
|
|
|
fn = getattr(client, method)
|
|
params = dict(payload or {})
|
|
if timeout is not None:
|
|
try:
|
|
parameters = signature(fn).parameters.values()
|
|
except (TypeError, ValueError):
|
|
parameters = ()
|
|
accepts_timeout = any(
|
|
parameter.kind is Parameter.VAR_KEYWORD
|
|
or (
|
|
parameter.name == "timeout"
|
|
and parameter.kind
|
|
in {Parameter.POSITIONAL_OR_KEYWORD, Parameter.KEYWORD_ONLY}
|
|
)
|
|
for parameter in parameters
|
|
)
|
|
if accepts_timeout:
|
|
return fn(endpoint, params, timeout=timeout)
|
|
return fn(endpoint, params)
|
|
|
|
|
|
def _material_kind(
|
|
material_type: str,
|
|
) -> Literal["image", "video", "file"]:
|
|
"""Map workstation concepts to the three audited upload endpoints."""
|
|
|
|
value = material_type.strip().lower()
|
|
if value in {"image", "tongue_images"}:
|
|
return "image"
|
|
if value == "video":
|
|
return "video"
|
|
if value in {"file", "report_files"}:
|
|
return "file"
|
|
raise ValueError("material_type must be image/video/file or tongue_images/report_files")
|
|
|
|
|
|
def _is_local_material_reference(value: str) -> bool:
|
|
"""Return whether a would-be server URI is a local/unsafe path."""
|
|
|
|
text = value.strip()
|
|
lower = text.lower()
|
|
return (
|
|
lower.startswith("file:")
|
|
or text.startswith(("\\\\", "//"))
|
|
or (len(text) >= 3 and text[0].isalpha() and text[1] == ":" and text[2] in "\\/")
|
|
or "\\" in text
|
|
)
|
|
|
|
|
|
def _normalise_material_reference(value: object, endpoint: str) -> str:
|
|
"""Extract a safe server ``uri``/``url`` from an upload response."""
|
|
|
|
candidate: object = value
|
|
if isinstance(value, Mapping):
|
|
candidate = value.get("uri") or value.get("url")
|
|
if candidate in (None, "") and isinstance(value.get("data"), Mapping):
|
|
nested = value["data"]
|
|
candidate = nested.get("uri") or nested.get("url")
|
|
reference = str(candidate or "").strip()
|
|
if not reference:
|
|
raise ApiProtocolError(
|
|
f"{endpoint} returned no material uri/url",
|
|
data=value,
|
|
)
|
|
if _is_local_material_reference(reference):
|
|
raise ApiProtocolError(
|
|
f"{endpoint} returned an unsafe local material path",
|
|
data=value,
|
|
)
|
|
return reference
|
|
|
|
|
|
def _server_materials(
|
|
values: list[str] | tuple[str, ...] | None,
|
|
field: str,
|
|
) -> list[str]:
|
|
"""Validate that a note JSON contains server references, never local paths."""
|
|
|
|
result: list[str] = []
|
|
for raw_value in values or ():
|
|
value = str(raw_value).strip()
|
|
if not value:
|
|
raise ValueError(f"{field} contains an empty material reference")
|
|
if _is_local_material_reference(value):
|
|
raise ValueError(f"{field} must contain server uri/url values, not local paths")
|
|
result.append(value)
|
|
return result
|
|
|
|
|
|
def _strings(value: object) -> tuple[str, ...]:
|
|
if isinstance(value, str):
|
|
candidates: object = value.split(",")
|
|
else:
|
|
candidates = value
|
|
if not isinstance(candidates, (list, tuple, set, frozenset)):
|
|
return ()
|
|
return tuple(item for item in (str(value).strip() for value in candidates) if item)
|
|
|
|
|
|
def _to_int(value: object, default: int = 0) -> int:
|
|
try:
|
|
return int(value) # type: ignore[arg-type]
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _to_bool(value: object) -> bool:
|
|
if isinstance(value, str):
|
|
return value.strip().lower() not in {"", "0", "false", "no", "off"}
|
|
return bool(value)
|
|
|
|
|
|
def _template_payload(
|
|
template: PrescriptionTemplate | Mapping[str, Any] | None,
|
|
fields: Mapping[str, Any],
|
|
*,
|
|
include_id: bool,
|
|
) -> dict[str, Any]:
|
|
if isinstance(template, PrescriptionTemplate):
|
|
result = template.to_api_dict(include_id=include_id)
|
|
elif isinstance(template, Mapping):
|
|
result = dict(template)
|
|
elif template is None:
|
|
result = {}
|
|
else:
|
|
raise TypeError("template must be a PrescriptionTemplate or mapping")
|
|
result.update(fields)
|
|
if "name" in result and "prescription_name" not in result:
|
|
result["prescription_name"] = result.pop("name")
|
|
if "formula_type" in result:
|
|
result["formula_type"] = _formula_type_for_api(result["formula_type"])
|
|
if "is_public" in result:
|
|
result["is_public"] = int(_to_bool(result["is_public"]))
|
|
if "disable_edit" in result:
|
|
result["disable_edit"] = int(_to_bool(result["disable_edit"]))
|
|
if "herbs" in result and isinstance(result["herbs"], (list, tuple)):
|
|
result["herbs"] = [dict(item) for item in result["herbs"] if isinstance(item, Mapping)]
|
|
if not str(result.get("prescription_name") or "").strip():
|
|
raise ValueError("prescription template name is required")
|
|
result.pop("raw", None)
|
|
if not include_id:
|
|
result.pop("id", None)
|
|
return result
|
|
|
|
|
|
def _formula_type_for_api(value: object) -> str:
|
|
text = str(value or "").strip().lower()
|
|
return "辅方" if text in {"2", "aux", "auxiliary", "secondary", "辅方"} else "主方"
|
|
|
|
|
|
def _dictionary_rows(value: object, dictionary_type: str) -> list[dict[str, Any]]:
|
|
"""Unwrap admin ``config/dict`` payloads into plain option rows."""
|
|
|
|
clean = str(dictionary_type or "").strip()
|
|
|
|
def as_rows(candidate: object) -> list[dict[str, Any]] | None:
|
|
if isinstance(candidate, list):
|
|
return [dict(row) for row in candidate if isinstance(row, Mapping)]
|
|
return None
|
|
|
|
if isinstance(value, Mapping):
|
|
typed = as_rows(value.get(clean))
|
|
if typed is not None:
|
|
return typed
|
|
data = value.get("data")
|
|
if isinstance(data, Mapping):
|
|
typed = as_rows(data.get(clean))
|
|
if typed is not None:
|
|
return typed
|
|
typed = as_rows(data.get("lists"))
|
|
if typed is not None:
|
|
return typed
|
|
typed = as_rows(value.get("lists"))
|
|
if typed is not None:
|
|
return typed
|
|
typed = as_rows(value.get("items"))
|
|
if typed is not None:
|
|
return typed
|
|
typed = as_rows(value)
|
|
if typed is not None:
|
|
return typed
|
|
return _mapping_rows(value)
|
|
|
|
|
|
def _mapping_rows(value: object) -> list[dict[str, Any]]:
|
|
"""Extract mapping rows from direct arrays or tolerant page envelopes."""
|
|
|
|
return PageResult.from_payload(value, lambda row: dict(row)).items
|
|
|
|
|
|
def _body(
|
|
payload: Mapping[str, Any] | None,
|
|
fields: Mapping[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""Merge a forward-compatible mutation body without retaining ``raw``."""
|
|
|
|
result = dict(payload or {})
|
|
result.update(fields)
|
|
result.pop("raw", None)
|
|
return result
|
|
|
|
|
|
_RECORD_DATE_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$")
|
|
_RECORD_TIME_PATTERN = re.compile(r"^(?:[01]\d|2[0-3]):[0-5]\d$")
|
|
|
|
|
|
def _daily_record_body(
|
|
kind: Literal["blood", "diet", "exercise"],
|
|
payload: Mapping[str, Any] | None,
|
|
fields: Mapping[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""Validate the exact DTOs used by the admin ``DailyMatrix`` dialogs."""
|
|
|
|
body = _body(payload, fields)
|
|
for key in ("diagnosis_id", "patient_id"):
|
|
value = _to_int(body.get(key), 0)
|
|
if value <= 0:
|
|
raise ValueError(f"{key} must be positive")
|
|
body[key] = value
|
|
record_date = str(body.get("record_date") or "").strip()
|
|
if not _RECORD_DATE_PATTERN.fullmatch(record_date):
|
|
raise ValueError("record_date must use YYYY-MM-DD")
|
|
parsed_date = date.fromisoformat(record_date)
|
|
if parsed_date > date.today():
|
|
raise ValueError("record_date cannot be in the future")
|
|
body["record_date"] = record_date
|
|
|
|
if kind == "blood":
|
|
record_time = str(body.get("record_time") or "").strip()
|
|
if record_time and not _RECORD_TIME_PATTERN.fullmatch(record_time):
|
|
raise ValueError("record_time must use HH:MM")
|
|
body["record_time"] = record_time
|
|
limits = {
|
|
"fasting_blood_sugar": 50.0,
|
|
"postprandial_blood_sugar": 50.0,
|
|
"other_blood_sugar": 50.0,
|
|
"systolic_pressure": 300.0,
|
|
"diastolic_pressure": 200.0,
|
|
}
|
|
has_content = False
|
|
for key, maximum in limits.items():
|
|
raw = body.get(key)
|
|
if raw in (None, ""):
|
|
continue
|
|
try:
|
|
number = float(raw)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(f"{key} must be numeric") from exc
|
|
if number < 0 or number > maximum:
|
|
raise ValueError(f"{key} must be between 0 and {maximum:g}")
|
|
body[key] = number
|
|
has_content = has_content or number > 0
|
|
for key in ("western_medicine", "insulin", "remark"):
|
|
clean = str(body.get(key) or "").strip()
|
|
body[key] = clean
|
|
has_content = has_content or bool(clean)
|
|
if not has_content:
|
|
raise ValueError("a blood record must contain at least one measurement or note")
|
|
return body
|
|
|
|
if kind == "diet":
|
|
has_content = False
|
|
for meal in ("breakfast", "lunch", "dinner"):
|
|
foods_key = f"{meal}_foods"
|
|
foods = str(body.get(foods_key) or body.get(meal) or "").strip()
|
|
body[foods_key] = foods
|
|
has_content = has_content or bool(foods)
|
|
images_key = f"{meal}_images"
|
|
raw_images = body.get(images_key) or []
|
|
if not isinstance(raw_images, (list, tuple)):
|
|
raise ValueError(f"{images_key} must be an array")
|
|
if len(raw_images) > 3:
|
|
raise ValueError(f"{images_key} cannot contain more than 3 images")
|
|
body[images_key] = _server_materials(raw_images, images_key)
|
|
has_content = has_content or bool(body[images_key])
|
|
body["note"] = str(body.get("note") or "").strip()
|
|
has_content = has_content or bool(body["note"])
|
|
if not has_content:
|
|
raise ValueError("a diet record must contain food, an image or a note")
|
|
return body
|
|
|
|
exercise_type = str(body.get("exercise_type") or "").strip()
|
|
if not exercise_type:
|
|
raise ValueError("exercise_type is required")
|
|
duration = _to_int(body.get("duration"), 0)
|
|
if duration < 1 or duration > 300:
|
|
raise ValueError("duration must be between 1 and 300 minutes")
|
|
intensity = _to_int(body.get("intensity"), 0)
|
|
if intensity not in {1, 2, 3}:
|
|
raise ValueError("intensity must be 1, 2 or 3")
|
|
raw_images = body.get("images") or []
|
|
if not isinstance(raw_images, (list, tuple)):
|
|
raise ValueError("images must be an array")
|
|
if len(raw_images) > 3:
|
|
raise ValueError("images cannot contain more than 3 items")
|
|
body.update(
|
|
{
|
|
"exercise_type": exercise_type,
|
|
"duration": duration,
|
|
"intensity": intensity,
|
|
"images": _server_materials(raw_images, "images"),
|
|
"note": str(body.get("note") or "").strip(),
|
|
}
|
|
)
|
|
return body
|
|
|
|
|
|
def _identified_body(
|
|
value: int | Mapping[str, Any],
|
|
changes: Mapping[str, Any] | None,
|
|
fields: Mapping[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""Build an edit payload from either an identifier or a current mapping."""
|
|
|
|
result = {"id": value} if isinstance(value, int) else dict(value)
|
|
result.update(changes or {})
|
|
result.update(fields)
|
|
result.pop("raw", None)
|
|
if not _to_int(result.get("id"), 0):
|
|
raise ValueError("id is required")
|
|
return result
|
|
|
|
|
|
def _prescription_payload(
|
|
prescription: Prescription | Mapping[str, Any] | None,
|
|
fields: Mapping[str, Any],
|
|
*,
|
|
include_id: bool,
|
|
) -> dict[str, Any]:
|
|
"""Build the complete prescription add/edit DTO while preserving aliases."""
|
|
|
|
if isinstance(prescription, Prescription):
|
|
result = prescription.to_api_dict(include_id=include_id)
|
|
elif isinstance(prescription, Mapping):
|
|
result = dict(prescription)
|
|
elif prescription is None:
|
|
result = {}
|
|
else:
|
|
raise TypeError("prescription must be a Prescription or mapping")
|
|
result.update(fields)
|
|
result.pop("raw", None)
|
|
if "herbs" in result and isinstance(result["herbs"], (list, tuple)):
|
|
result["herbs"] = [dict(item) for item in result["herbs"] if isinstance(item, Mapping)]
|
|
if "visible_role_ids" in result and isinstance(
|
|
result["visible_role_ids"], (tuple, set, frozenset)
|
|
):
|
|
result["visible_role_ids"] = list(result["visible_role_ids"])
|
|
if "need_decoction" in result:
|
|
result["need_decoction"] = int(_to_bool(result["need_decoction"]))
|
|
if "is_shared" in result:
|
|
result["is_shared"] = int(_to_bool(result["is_shared"]))
|
|
if not include_id:
|
|
result.pop("id", None)
|
|
return result
|
|
|
|
|
|
def _merge_result(body: Mapping[str, Any], result: object) -> dict[str, Any]:
|
|
"""Merge common mutation response shapes over the submitted body."""
|
|
|
|
merged = dict(body)
|
|
if isinstance(result, Mapping):
|
|
nested = result.get("data")
|
|
merged.update(nested if isinstance(nested, Mapping) else result)
|
|
elif isinstance(result, (int, str)):
|
|
merged["id"] = result
|
|
return merged
|
|
|
|
|
|
def _audit_action(action: str, remark: str) -> AuditAction:
|
|
"""Validate the two audit actions and the routed rejection boundary."""
|
|
|
|
normalised = action.strip().lower()
|
|
if normalised not in {"approve", "reject"}:
|
|
raise ValueError("action must be approve or reject")
|
|
if normalised == "reject" and not remark.strip():
|
|
raise ValueError("remark is required when rejecting")
|
|
return normalised # type: ignore[return-value]
|
|
|
|
|
|
def _safe_menu(value: object) -> list[dict[str, Any]]:
|
|
"""Copy dynamic menu JSON without evaluating or coercing untrusted objects.
|
|
|
|
Unknown JSON fields are intentionally preserved so the composition root can
|
|
honour future backend menu metadata. Non-string keys and non-JSON runtime
|
|
objects are dropped, which prevents callables or framework objects from
|
|
leaking into navigation construction.
|
|
"""
|
|
|
|
if not isinstance(value, (list, tuple)):
|
|
return []
|
|
result: list[dict[str, Any]] = []
|
|
for item in value:
|
|
safe = _safe_json_mapping(item)
|
|
if safe is not None:
|
|
result.append(safe)
|
|
return result
|
|
|
|
|
|
def _safe_json_mapping(value: object) -> dict[str, Any] | None:
|
|
if not isinstance(value, Mapping):
|
|
return None
|
|
result: dict[str, Any] = {}
|
|
for key, candidate in value.items():
|
|
if not isinstance(key, str):
|
|
continue
|
|
safe = _safe_json_value(candidate)
|
|
if safe is not _UNSAFE:
|
|
result[key] = safe
|
|
return result
|
|
|
|
|
|
_UNSAFE: Final[object] = object()
|
|
|
|
|
|
def _safe_json_value(value: object) -> Any:
|
|
if value is None or isinstance(value, (str, int, float, bool)):
|
|
return value
|
|
if isinstance(value, Mapping):
|
|
return _safe_json_mapping(value)
|
|
if isinstance(value, (list, tuple)):
|
|
return [safe for item in value if (safe := _safe_json_value(item)) is not _UNSAFE]
|
|
return _UNSAFE
|