2573 lines
94 KiB
Python
2573 lines
94 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 Mapping
|
|
from contextlib import suppress
|
|
from datetime import date
|
|
from os import PathLike
|
|
from pathlib import Path
|
|
from typing import Any, Final, Literal, Protocol
|
|
|
|
from doctor_workstation.core.errors import (
|
|
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",
|
|
}
|
|
"""Canonical permissions used by the routed prescription-library view."""
|
|
|
|
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_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 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 get_prescription(self, prescription_id: int) -> Prescription:
|
|
"""Return one issued prescription."""
|
|
|
|
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 create_diagnosis_appointment(
|
|
self, payload: Mapping[str, Any] | None = None, **fields: Any
|
|
) -> Any:
|
|
"""Create an appointment from the diagnosis 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_assistant_watch_ticket(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Return a receive-only room ticket for the assigned assistant."""
|
|
|
|
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_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,
|
|
) -> dict[str, Any]:
|
|
"""Upload a local video then attach it 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 generate_video_qrcode(
|
|
self, doctor_id: int, patient_id: int, share_user_id: int
|
|
) -> 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) -> CallTicket:
|
|
"""Return short-lived call credentials."""
|
|
|
|
def start_call(self, diagnosis_id: int, patient_id: int, *, call_type: int = 2) -> Any:
|
|
"""Create a call record."""
|
|
|
|
def end_call(self, diagnosis_id: int) -> Any:
|
|
"""End the active diagnosis call."""
|
|
|
|
def bind_call_room(self, diagnosis_id: int, room_id: str) -> Any:
|
|
"""Bind a TRTC room to the active call."""
|
|
|
|
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")
|
|
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 request_filters.get("diag_scope_relax")
|
|
):
|
|
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 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 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_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 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."""
|
|
|
|
payload = self.client.get(
|
|
"tcm.prescription/getByAppointment", {"appointment_id": appointment_id}
|
|
)
|
|
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 forms."""
|
|
|
|
payload = self.client.get("config/dict", {"type": dictionary_type})
|
|
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 create_diagnosis_appointment(
|
|
self, payload: Mapping[str, Any] | None = None, **fields: Any
|
|
) -> Any:
|
|
"""Create an appointment through the canonical diagnosis-list route."""
|
|
|
|
return self.client.post("doctor.appointment/create", _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 get_assistant_watch_ticket(self, diagnosis_id: int) -> dict[str, Any]:
|
|
"""Load the server-authorized receive-only TRTC room parameters."""
|
|
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
return dict(
|
|
_require_mapping(
|
|
self.client.get("tcm.diagnosis/watchCall", {"diagnosis_id": diagnosis_id}),
|
|
"tcm.diagnosis/watchCall",
|
|
)
|
|
)
|
|
|
|
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)
|
|
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 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 = _diagnosis_create_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,
|
|
) -> dict[str, Any]:
|
|
"""Perform the same upload/create/attach sequence as ``CallRecordPanel``."""
|
|
|
|
if diagnosis_id <= 0:
|
|
raise ValueError("diagnosis_id must be positive")
|
|
# Match the admin component exactly: upload first so a failed upload does
|
|
# not leave behind an empty synthetic call record.
|
|
file_url = self.upload_material(path, "video")
|
|
target_id = call_record_id
|
|
if target_id is None:
|
|
target_id = int(self.create_manual_call_record(diagnosis_id).get("id") or 0)
|
|
if target_id <= 0:
|
|
raise ApiProtocolError("tcm.diagnosis/createManualCallRecord returned no id")
|
|
self.attach_local_call_recording(diagnosis_id, file_url, call_record_id=target_id)
|
|
return {
|
|
"diagnosis_id": diagnosis_id,
|
|
"call_record_id": target_id,
|
|
"file_url": file_url,
|
|
}
|
|
|
|
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")
|
|
# The current server validator mistakenly inspects diagnosis_id for this case;
|
|
# mirror the admin DTO until that server contract is fixed.
|
|
body.setdefault("diagnosis_id", body["doctor_id"])
|
|
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 generate_video_qrcode(
|
|
self, doctor_id: int, patient_id: int, share_user_id: int
|
|
) -> dict[str, Any]:
|
|
"""Generate the video QR with the admin's confirmed login-page DTO."""
|
|
|
|
return self.generate_mini_program_qrcode(
|
|
diagnosis_id=doctor_id,
|
|
doctor_id=doctor_id,
|
|
patient_id=patient_id,
|
|
share_user_id=share_user_id,
|
|
mini_program_path="pages/login/login",
|
|
)
|
|
|
|
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) -> 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},
|
|
),
|
|
"tcm.diagnosis/getCallSignature",
|
|
)
|
|
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) -> Any:
|
|
"""Create the server-side call record before ringing participants."""
|
|
|
|
return self.client.post(
|
|
"tcm.diagnosis/startCall",
|
|
{
|
|
"diagnosis_id": diagnosis_id,
|
|
"patient_id": patient_id,
|
|
"call_type": call_type,
|
|
},
|
|
)
|
|
|
|
def end_call(self, diagnosis_id: int) -> Any:
|
|
"""End the active call/recording associated with a diagnosis."""
|
|
|
|
return self.client.post("tcm.diagnosis/endCall", {"diagnosis_id": diagnosis_id})
|
|
|
|
def bind_call_room(self, diagnosis_id: int, room_id: str) -> Any:
|
|
"""Bind the actual TRTC room to the active call record."""
|
|
|
|
if not room_id.strip():
|
|
raise ValueError("room_id is required")
|
|
return self.client.post(
|
|
"tcm.diagnosis/bindCallRoom",
|
|
{"diagnosis_id": diagnosis_id, "room_id": room_id.strip()},
|
|
)
|
|
|
|
# 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 _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 _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 _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 _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
|
|
|
|
|
|
def _diagnosis_create_body(
|
|
payload: Mapping[str, Any] | None,
|
|
fields: Mapping[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""Validate the production ``tcm.diagnosis/add`` identity contract."""
|
|
|
|
body = _body(payload, fields)
|
|
patient_name = str(body.get("patient_name") or "").strip()
|
|
phone = str(body.get("phone") or "").strip()
|
|
diagnosis_type = str(body.get("diagnosis_type") or "").strip()
|
|
local_hospital_name = str(body.get("local_hospital_name") or "").strip()
|
|
if not patient_name:
|
|
raise ValueError("patient_name is required")
|
|
if not re.fullmatch(r"1[3-9]\d{9}", phone):
|
|
raise ValueError("phone must be a valid 11-digit mobile number")
|
|
gender = _to_int(body.get("gender"), -1)
|
|
if gender not in {0, 1}:
|
|
raise ValueError("gender must be 0 or 1")
|
|
age = _to_int(body.get("age"), -1)
|
|
if age < 0 or age > 150:
|
|
raise ValueError("age must be between 0 and 150")
|
|
if not diagnosis_type:
|
|
raise ValueError("diagnosis_type is required")
|
|
if not local_hospital_name:
|
|
raise ValueError("local_hospital_name is required")
|
|
body.update(
|
|
{
|
|
"patient_name": patient_name,
|
|
"phone": phone,
|
|
"gender": gender,
|
|
"age": age,
|
|
"diagnosis_type": diagnosis_type,
|
|
"local_hospital_name": local_hospital_name,
|
|
}
|
|
)
|
|
return body
|
|
|
|
|
|
_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
|