340 lines
12 KiB
Python
340 lines
12 KiB
Python
"""ERPNext Healthcare-specific integrations.
|
|
|
|
This module provides high-level operations for the ERPNext Healthcare module,
|
|
abstracting the Frappe REST API into clinic-specific workflows.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
import structlog
|
|
|
|
from src.infrastructure.erpnext.client import ERPNextClient, get_erpnext_client
|
|
from src.core.exceptions import ERPNextError
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ERPNext Healthcare Doctypes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
DOCTYPE_PATIENT = "Patient"
|
|
DOCTYPE_PRACTITIONER = "Healthcare Practitioner"
|
|
DOCTYPE_APPOINTMENT = "Patient Appointment"
|
|
DOCTYPE_SERVICE_UNIT = "Healthcare Service Unit"
|
|
DOCTYPE_CLINICAL_PROCEDURE = "Clinical Procedure Template"
|
|
|
|
|
|
class ERPNextHealthcare:
|
|
"""High-level ERPNext Healthcare operations."""
|
|
|
|
def __init__(self, client: ERPNextClient | None = None) -> None:
|
|
self.client = client
|
|
|
|
async def _get_client(self) -> ERPNextClient:
|
|
if self.client is None:
|
|
self.client = await get_erpnext_client()
|
|
return self.client
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Patients
|
|
# -----------------------------------------------------------------------
|
|
|
|
async def find_patient_by_phone(self, phone: str) -> dict[str, Any] | None:
|
|
"""Find a patient by mobile number."""
|
|
client = await self._get_client()
|
|
patients = await client.get_list(
|
|
DOCTYPE_PATIENT,
|
|
filters=[["mobile", "=", phone]],
|
|
fields=[
|
|
"name", "patient_name", "mobile", "phone", "sex",
|
|
"dob", "blood_group", "allergies", "medical_history",
|
|
],
|
|
limit=1,
|
|
)
|
|
return patients[0] if patients else None
|
|
|
|
async def create_patient(
|
|
self,
|
|
first_name: str,
|
|
mobile: str,
|
|
sex: str = "Female",
|
|
dob: str | None = None,
|
|
email: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a new patient record."""
|
|
client = await self._get_client()
|
|
data = {
|
|
"doctype": DOCTYPE_PATIENT,
|
|
"first_name": first_name,
|
|
"mobile": mobile,
|
|
"sex": sex,
|
|
}
|
|
if dob:
|
|
data["dob"] = dob
|
|
if email:
|
|
data["email"] = email
|
|
|
|
result = await client.create_document(DOCTYPE_PATIENT, data)
|
|
logger.info("patient_created", patient_id=result.get("name"), name=first_name)
|
|
return result
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Practitioners (Doctors)
|
|
# -----------------------------------------------------------------------
|
|
|
|
async def get_practitioners(
|
|
self,
|
|
department: str | None = None,
|
|
is_active: bool = True,
|
|
) -> list[dict[str, Any]]:
|
|
"""List healthcare practitioners (doctors)."""
|
|
client = await self._get_client()
|
|
filters: list[list[Any]] = []
|
|
if is_active:
|
|
filters.append(["status", "=", "Active"])
|
|
if department:
|
|
filters.append(["department", "=", department])
|
|
|
|
return await client.get_list(
|
|
DOCTYPE_PRACTITIONER,
|
|
filters=filters if filters else None,
|
|
fields=["name", "practitioner_name", "department", "status", "mobile_phone"],
|
|
limit=50,
|
|
)
|
|
|
|
async def get_practitioner_schedule(
|
|
self,
|
|
practitioner: str,
|
|
date: str,
|
|
) -> dict[str, Any]:
|
|
"""Get availability schedule for a practitioner on a specific date.
|
|
|
|
Uses the Frappe whitelisted method from ERPNext Healthcare.
|
|
"""
|
|
client = await self._get_client()
|
|
try:
|
|
result = await client.call_method(
|
|
"healthcare.healthcare.doctype.patient_appointment.patient_appointment.get_availability_data",
|
|
{
|
|
"practitioner": practitioner,
|
|
"date": date,
|
|
},
|
|
)
|
|
return result.get("message", {})
|
|
except ERPNextError:
|
|
# Fallback: query existing appointments and return inverse
|
|
return await self._fallback_availability(practitioner, date)
|
|
|
|
async def _fallback_availability(
|
|
self,
|
|
practitioner: str,
|
|
date: str,
|
|
) -> dict[str, Any]:
|
|
"""Fallback availability check by querying existing appointments."""
|
|
client = await self._get_client()
|
|
existing = await client.get_list(
|
|
DOCTYPE_APPOINTMENT,
|
|
filters=[
|
|
["practitioner", "=", practitioner],
|
|
["appointment_date", "=", date],
|
|
["status", "in", ["Scheduled", "Open"]],
|
|
],
|
|
fields=["appointment_time", "duration"],
|
|
limit=100,
|
|
)
|
|
|
|
# Standard clinic hours: 09:00 - 18:00, 30-min slots
|
|
slots = []
|
|
for hour in range(9, 18):
|
|
for minute in (0, 30):
|
|
time_str = f"{hour:02d}:{minute:02d}"
|
|
# Check if slot is taken
|
|
taken = any(
|
|
appt["appointment_time"] == time_str
|
|
for appt in existing
|
|
)
|
|
if not taken:
|
|
slots.append({
|
|
"from_time": time_str,
|
|
"available": True,
|
|
})
|
|
|
|
return {
|
|
"practitioner": practitioner,
|
|
"date": date,
|
|
"available_slots": slots,
|
|
"appointment_list": existing,
|
|
}
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Appointments
|
|
# -----------------------------------------------------------------------
|
|
|
|
async def get_appointments(
|
|
self,
|
|
patient: str | None = None,
|
|
practitioner: str | None = None,
|
|
date: str | None = None,
|
|
status: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""Query patient appointments."""
|
|
client = await self._get_client()
|
|
filters: list[list[Any]] = []
|
|
if patient:
|
|
filters.append(["patient", "=", patient])
|
|
if practitioner:
|
|
filters.append(["practitioner", "=", practitioner])
|
|
if date:
|
|
filters.append(["appointment_date", "=", date])
|
|
if status:
|
|
filters.append(["status", "=", status])
|
|
|
|
return await client.get_list(
|
|
DOCTYPE_APPOINTMENT,
|
|
filters=filters if filters else None,
|
|
fields=[
|
|
"name", "patient", "patient_name", "practitioner",
|
|
"appointment_date", "appointment_time", "duration",
|
|
"status", "department", "notes",
|
|
],
|
|
limit=50,
|
|
order_by="appointment_date desc, appointment_time desc",
|
|
)
|
|
|
|
async def create_appointment(
|
|
self,
|
|
patient: str,
|
|
practitioner: str,
|
|
appointment_date: str,
|
|
appointment_time: str,
|
|
duration: int = 30,
|
|
department: str = "Dermatology",
|
|
appointment_type: str = "Consulta General",
|
|
appointment_for: str = "Practitioner",
|
|
notes: str = "",
|
|
service_unit: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a patient appointment in ERPNext Healthcare.
|
|
|
|
Args:
|
|
patient: Patient ID (name field in ERPNext).
|
|
practitioner: Practitioner ID.
|
|
appointment_date: Date in YYYY-MM-DD format.
|
|
appointment_time: Time in HH:MM format.
|
|
duration: Duration in minutes.
|
|
department: Medical department.
|
|
notes: Additional notes.
|
|
service_unit: Healthcare Service Unit (consultation room).
|
|
|
|
Returns:
|
|
Created appointment document.
|
|
"""
|
|
client = await self._get_client()
|
|
|
|
# Validate patient exists
|
|
patient_doc = await client.get_document(DOCTYPE_PATIENT, patient)
|
|
if not patient_doc:
|
|
raise ERPNextError(f"Patient {patient} not found", status_code=404)
|
|
|
|
# Validate practitioner exists
|
|
practitioner_doc = await client.get_document(DOCTYPE_PRACTITIONER, practitioner)
|
|
if not practitioner_doc:
|
|
raise ERPNextError(f"Practitioner {practitioner} not found", status_code=404)
|
|
|
|
# Check for conflicts
|
|
conflicts = await client.get_list(
|
|
DOCTYPE_APPOINTMENT,
|
|
filters=[
|
|
["practitioner", "=", practitioner],
|
|
["appointment_date", "=", appointment_date],
|
|
["appointment_time", "=", appointment_time],
|
|
["status", "in", ["Scheduled", "Open"]],
|
|
],
|
|
fields=["name"],
|
|
limit=1,
|
|
)
|
|
if conflicts:
|
|
raise ERPNextError(
|
|
f"Time slot conflict: {practitioner} is not available at {appointment_time}",
|
|
status_code=409,
|
|
)
|
|
|
|
data = {
|
|
"doctype": DOCTYPE_APPOINTMENT,
|
|
"patient": patient,
|
|
"practitioner": practitioner,
|
|
"appointment_date": appointment_date,
|
|
"appointment_time": appointment_time,
|
|
"duration": duration,
|
|
"department": department,
|
|
"appointment_type": appointment_type,
|
|
"appointment_for": appointment_for,
|
|
"notes": notes,
|
|
"status": "Scheduled",
|
|
}
|
|
if service_unit:
|
|
data["service_unit"] = service_unit
|
|
|
|
result = await client.create_document(DOCTYPE_APPOINTMENT, data)
|
|
logger.info(
|
|
"appointment_created",
|
|
appointment_id=result.get("name"),
|
|
patient=patient,
|
|
practitioner=practitioner,
|
|
date=appointment_date,
|
|
time=appointment_time,
|
|
)
|
|
return result
|
|
|
|
async def cancel_appointment(self, appointment_id: str, reason: str = "") -> dict[str, Any]:
|
|
"""Cancel an existing appointment."""
|
|
client = await self._get_client()
|
|
result = await client.update_document(
|
|
DOCTYPE_APPOINTMENT,
|
|
appointment_id,
|
|
{"status": "Cancelled", "notes": f"Cancelled via WhatsApp. {reason}"},
|
|
)
|
|
logger.info("appointment_cancelled", appointment_id=appointment_id, reason=reason)
|
|
return result
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Services / Procedures
|
|
# -----------------------------------------------------------------------
|
|
|
|
async def get_clinical_procedures(
|
|
self,
|
|
is_active: bool = True,
|
|
) -> list[dict[str, Any]]:
|
|
"""List available clinical procedures / services."""
|
|
client = await self._get_client()
|
|
filters: list[list[Any]] = []
|
|
if is_active:
|
|
filters.append(["is_active", "=", 1])
|
|
|
|
return await client.get_list(
|
|
DOCTYPE_CLINICAL_PROCEDURE,
|
|
filters=filters if filters else None,
|
|
fields=["name", "template", "item_code", "rate", "medical_department"],
|
|
limit=100,
|
|
)
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Wallet / Custom fields (if implemented in ERPNext)
|
|
# -----------------------------------------------------------------------
|
|
|
|
async def get_patient_wallet(self, patient: str) -> dict[str, Any]:
|
|
"""Get patient wallet balance if custom doctype exists."""
|
|
client = await self._get_client()
|
|
try:
|
|
wallets = await client.get_list(
|
|
"Patient Wallet",
|
|
filters=[["patient", "=", patient]],
|
|
fields=["name", "balance", "points"],
|
|
limit=1,
|
|
)
|
|
if wallets:
|
|
return wallets[0]
|
|
return {"balance": 0.0, "points": 0, "note": "No wallet record"}
|
|
except ERPNextError:
|
|
return {"balance": 0.0, "points": 0, "note": "Wallet module not configured"}
|