436 lines
13 KiB
Python
436 lines
13 KiB
Python
"""
|
|
Integrations models for the Tenhal Multidisciplinary Healthcare Platform.
|
|
|
|
This module handles external integrations:
|
|
- Lab and Radiology orders
|
|
- NPHIES (Insurance e-Claims via FHIR)
|
|
- ZATCA (E-Invoicing)
|
|
"""
|
|
|
|
from django.db import models
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from core.models import (
|
|
UUIDPrimaryKeyMixin,
|
|
TimeStampedMixin,
|
|
TenantOwnedMixin,
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# Lab & Radiology Integration
|
|
# ============================================================================
|
|
|
|
class ExternalOrder(UUIDPrimaryKeyMixin, TimeStampedMixin, TenantOwnedMixin):
|
|
"""
|
|
Lab and radiology orders to external providers.
|
|
Placeholder for integration with lab/radiology systems.
|
|
"""
|
|
|
|
class OrderType(models.TextChoices):
|
|
LAB = 'LAB', _('Laboratory')
|
|
RADIOLOGY = 'RADIOLOGY', _('Radiology')
|
|
|
|
class Status(models.TextChoices):
|
|
ORDERED = 'ORDERED', _('Ordered')
|
|
IN_PROGRESS = 'IN_PROGRESS', _('In Progress')
|
|
COMPLETED = 'COMPLETED', _('Completed')
|
|
CANCELLED = 'CANCELLED', _('Cancelled')
|
|
|
|
patient = models.ForeignKey(
|
|
'core.Patient',
|
|
on_delete=models.CASCADE,
|
|
related_name='external_orders',
|
|
verbose_name=_("Patient")
|
|
)
|
|
order_type = models.CharField(
|
|
max_length=20,
|
|
choices=OrderType.choices,
|
|
verbose_name=_("Order Type")
|
|
)
|
|
order_details = models.JSONField(
|
|
default=dict,
|
|
help_text=_("Order details (tests, studies, etc.)"),
|
|
verbose_name=_("Order Details")
|
|
)
|
|
status = models.CharField(
|
|
max_length=20,
|
|
choices=Status.choices,
|
|
default=Status.ORDERED,
|
|
verbose_name=_("Status")
|
|
)
|
|
result_url = models.URLField(
|
|
blank=True,
|
|
verbose_name=_("Result URL")
|
|
)
|
|
result_data = models.JSONField(
|
|
default=dict,
|
|
blank=True,
|
|
verbose_name=_("Result Data")
|
|
)
|
|
ordered_by = models.ForeignKey(
|
|
'core.User',
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
related_name='external_orders_made',
|
|
verbose_name=_("Ordered By")
|
|
)
|
|
ordered_at = models.DateTimeField(
|
|
auto_now_add=True,
|
|
verbose_name=_("Ordered At")
|
|
)
|
|
completed_at = models.DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Completed At")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("External Order")
|
|
verbose_name_plural = _("External Orders")
|
|
ordering = ['-ordered_at']
|
|
indexes = [
|
|
models.Index(fields=['patient', 'order_type']),
|
|
models.Index(fields=['status', 'ordered_at']),
|
|
models.Index(fields=['tenant', 'ordered_at']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.get_order_type_display()} Order - {self.patient} - {self.ordered_at.date()}"
|
|
|
|
|
|
# ============================================================================
|
|
# NPHIES (Insurance e-Claims) Integration
|
|
# ============================================================================
|
|
|
|
class NphiesMessage(UUIDPrimaryKeyMixin, TimeStampedMixin, TenantOwnedMixin):
|
|
"""
|
|
NPHIES FHIR messages for insurance e-claims.
|
|
Tracks eligibility, prior authorization, claims, and payment reconciliation.
|
|
"""
|
|
|
|
class Direction(models.TextChoices):
|
|
OUTBOUND = 'OUTBOUND', _('Outbound')
|
|
INBOUND = 'INBOUND', _('Inbound')
|
|
|
|
class ResourceType(models.TextChoices):
|
|
ELIGIBILITY = 'ELIGIBILITY', _('Coverage Eligibility')
|
|
PRIOR_AUTH = 'PRIOR_AUTH', _('Prior Authorization')
|
|
CLAIM = 'CLAIM', _('Claim')
|
|
PAYMENT_NOTICE = 'PAYMENT_NOTICE', _('Payment Notice')
|
|
PAYMENT_RECONCILIATION = 'PAYMENT_RECONCILIATION', _('Payment Reconciliation')
|
|
|
|
class Status(models.TextChoices):
|
|
QUEUED = 'QUEUED', _('Queued')
|
|
SENT = 'SENT', _('Sent')
|
|
ACK = 'ACK', _('Acknowledged')
|
|
ERROR = 'ERROR', _('Error')
|
|
|
|
direction = models.CharField(
|
|
max_length=20,
|
|
choices=Direction.choices,
|
|
verbose_name=_("Direction")
|
|
)
|
|
resource_type = models.CharField(
|
|
max_length=30,
|
|
choices=ResourceType.choices,
|
|
verbose_name=_("Resource Type")
|
|
)
|
|
fhir_json = models.JSONField(
|
|
default=dict,
|
|
help_text=_("FHIR resource JSON"),
|
|
verbose_name=_("FHIR JSON")
|
|
)
|
|
status = models.CharField(
|
|
max_length=20,
|
|
choices=Status.choices,
|
|
default=Status.QUEUED,
|
|
verbose_name=_("Status")
|
|
)
|
|
correlation_id = models.CharField(
|
|
max_length=100,
|
|
blank=True,
|
|
help_text=_("Correlation ID for request/response matching"),
|
|
verbose_name=_("Correlation ID")
|
|
)
|
|
response_http_status = models.PositiveIntegerField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Response HTTP Status")
|
|
)
|
|
error_code = models.CharField(
|
|
max_length=100,
|
|
blank=True,
|
|
verbose_name=_("Error Code")
|
|
)
|
|
error_message = models.TextField(
|
|
blank=True,
|
|
verbose_name=_("Error Message")
|
|
)
|
|
sent_at = models.DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Sent At")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("NPHIES Message")
|
|
verbose_name_plural = _("NPHIES Messages")
|
|
ordering = ['-created_at']
|
|
indexes = [
|
|
models.Index(fields=['direction', 'resource_type']),
|
|
models.Index(fields=['status', 'created_at']),
|
|
models.Index(fields=['correlation_id']),
|
|
models.Index(fields=['tenant', 'created_at']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.get_direction_display()} {self.get_resource_type_display()} - {self.get_status_display()}"
|
|
|
|
|
|
class NphiesEncounterLink(UUIDPrimaryKeyMixin, TenantOwnedMixin):
|
|
"""
|
|
Links appointments to NPHIES encounters and claims.
|
|
Maintains relationship between internal and NPHIES identifiers.
|
|
"""
|
|
|
|
patient = models.ForeignKey(
|
|
'core.Patient',
|
|
on_delete=models.CASCADE,
|
|
related_name='nphies_encounter_links',
|
|
verbose_name=_("Patient")
|
|
)
|
|
appointment = models.ForeignKey(
|
|
'appointments.Appointment',
|
|
on_delete=models.CASCADE,
|
|
related_name='nphies_encounter_links',
|
|
verbose_name=_("Appointment")
|
|
)
|
|
encounter_id = models.CharField(
|
|
max_length=100,
|
|
help_text=_("NPHIES Encounter ID"),
|
|
verbose_name=_("Encounter ID")
|
|
)
|
|
claim_id = models.CharField(
|
|
max_length=100,
|
|
blank=True,
|
|
help_text=_("NPHIES Claim ID"),
|
|
verbose_name=_("Claim ID")
|
|
)
|
|
claim_response_id = models.CharField(
|
|
max_length=100,
|
|
blank=True,
|
|
help_text=_("NPHIES Claim Response ID"),
|
|
verbose_name=_("Claim Response ID")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("NPHIES Encounter Link")
|
|
verbose_name_plural = _("NPHIES Encounter Links")
|
|
unique_together = [['tenant', 'encounter_id']]
|
|
|
|
def __str__(self):
|
|
return f"NPHIES Link - {self.patient} - Encounter: {self.encounter_id}"
|
|
|
|
|
|
class PayerContract(UUIDPrimaryKeyMixin, TimeStampedMixin, TenantOwnedMixin):
|
|
"""
|
|
Insurance payer configurations for NPHIES integration.
|
|
Stores credentials and endpoints per payer.
|
|
"""
|
|
|
|
payer_code = models.CharField(
|
|
max_length=50,
|
|
help_text=_("Official payer code"),
|
|
verbose_name=_("Payer Code")
|
|
)
|
|
payer_name = models.CharField(
|
|
max_length=200,
|
|
verbose_name=_("Payer Name")
|
|
)
|
|
credentials = models.JSONField(
|
|
default=dict,
|
|
help_text=_("Encrypted credentials (OAuth2, mTLS, etc.)"),
|
|
verbose_name=_("Credentials")
|
|
)
|
|
endpoints = models.JSONField(
|
|
default=dict,
|
|
help_text=_("API endpoints for different operations"),
|
|
verbose_name=_("Endpoints")
|
|
)
|
|
supports_eligibility = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Supports Eligibility Check")
|
|
)
|
|
supports_prior_auth = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Supports Prior Authorization")
|
|
)
|
|
supports_claims = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Supports Claims")
|
|
)
|
|
is_active = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Is Active")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Payer Contract")
|
|
verbose_name_plural = _("Payer Contracts")
|
|
ordering = ['payer_name']
|
|
unique_together = [['tenant', 'payer_code']]
|
|
|
|
def __str__(self):
|
|
return f"{self.payer_name} ({self.payer_code})"
|
|
|
|
|
|
# ============================================================================
|
|
# ZATCA (E-Invoicing) Integration
|
|
# ============================================================================
|
|
|
|
class EInvoice(UUIDPrimaryKeyMixin, TimeStampedMixin, TenantOwnedMixin):
|
|
"""
|
|
ZATCA e-invoices (FATOORA Phase 2).
|
|
Tracks e-invoice generation, submission, and clearance.
|
|
"""
|
|
|
|
class ClearanceStatus(models.TextChoices):
|
|
PENDING = 'PENDING', _('Pending')
|
|
CLEARED = 'CLEARED', _('Cleared')
|
|
REJECTED = 'REJECTED', _('Rejected')
|
|
REPORTED = 'REPORTED', _('Reported')
|
|
|
|
class SubmissionMode(models.TextChoices):
|
|
CLEARANCE = 'CLEARANCE', _('Clearance (B2B)')
|
|
REPORTING = 'REPORTING', _('Reporting (B2C)')
|
|
|
|
invoice = models.ForeignKey(
|
|
'finance.Invoice',
|
|
on_delete=models.CASCADE,
|
|
related_name='e_invoices',
|
|
verbose_name=_("Invoice")
|
|
)
|
|
uuid = models.UUIDField(
|
|
unique=True,
|
|
help_text=_("Unique invoice UUID for ZATCA"),
|
|
verbose_name=_("UUID")
|
|
)
|
|
xml_payload = models.TextField(
|
|
blank=True,
|
|
help_text=_("Signed XML invoice"),
|
|
verbose_name=_("XML Payload")
|
|
)
|
|
qr_base64 = models.TextField(
|
|
blank=True,
|
|
help_text=_("Base64-encoded QR code (TLV format)"),
|
|
verbose_name=_("QR Code (Base64)")
|
|
)
|
|
clearance_status = models.CharField(
|
|
max_length=20,
|
|
choices=ClearanceStatus.choices,
|
|
default=ClearanceStatus.PENDING,
|
|
verbose_name=_("Clearance Status")
|
|
)
|
|
zatca_document_type = models.CharField(
|
|
max_length=50,
|
|
blank=True,
|
|
help_text=_("ZATCA document type code"),
|
|
verbose_name=_("Document Type")
|
|
)
|
|
submission_mode = models.CharField(
|
|
max_length=20,
|
|
choices=SubmissionMode.choices,
|
|
verbose_name=_("Submission Mode")
|
|
)
|
|
response_payload = models.JSONField(
|
|
default=dict,
|
|
blank=True,
|
|
help_text=_("ZATCA API response"),
|
|
verbose_name=_("Response Payload")
|
|
)
|
|
error_code = models.CharField(
|
|
max_length=100,
|
|
blank=True,
|
|
verbose_name=_("Error Code")
|
|
)
|
|
error_message = models.TextField(
|
|
blank=True,
|
|
verbose_name=_("Error Message")
|
|
)
|
|
submitted_at = models.DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Submitted At")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("E-Invoice")
|
|
verbose_name_plural = _("E-Invoices")
|
|
ordering = ['-created_at']
|
|
indexes = [
|
|
models.Index(fields=['invoice']),
|
|
models.Index(fields=['clearance_status', 'created_at']),
|
|
models.Index(fields=['uuid']),
|
|
models.Index(fields=['tenant', 'created_at']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"E-Invoice {self.uuid} - {self.invoice.invoice_number} - {self.get_clearance_status_display()}"
|
|
|
|
|
|
class ZatcaCredential(UUIDPrimaryKeyMixin, TimeStampedMixin, TenantOwnedMixin):
|
|
"""
|
|
ZATCA credentials (CSIDs) for different environments.
|
|
Manages compliance and production certificates.
|
|
"""
|
|
|
|
class Environment(models.TextChoices):
|
|
SIMULATION = 'SIMULATION', _('Simulation')
|
|
COMPLIANCE = 'COMPLIANCE', _('Compliance')
|
|
PRODUCTION = 'PRODUCTION', _('Production')
|
|
|
|
environment = models.CharField(
|
|
max_length=20,
|
|
choices=Environment.choices,
|
|
verbose_name=_("Environment")
|
|
)
|
|
csid = models.TextField(
|
|
help_text=_("Cryptographic Stamp Identifier"),
|
|
verbose_name=_("CSID")
|
|
)
|
|
certificate = models.TextField(
|
|
help_text=_("X.509 certificate (PEM format)"),
|
|
verbose_name=_("Certificate")
|
|
)
|
|
private_key = models.TextField(
|
|
help_text=_("Private key (encrypted, PEM format)"),
|
|
verbose_name=_("Private Key")
|
|
)
|
|
is_active = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Is Active")
|
|
)
|
|
expires_at = models.DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Expires At")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("ZATCA Credential")
|
|
verbose_name_plural = _("ZATCA Credentials")
|
|
ordering = ['-created_at']
|
|
unique_together = [['tenant', 'environment', 'is_active']]
|
|
|
|
def __str__(self):
|
|
return f"ZATCA {self.get_environment_display()} Credential - {self.tenant}"
|
|
|
|
@property
|
|
def is_expired(self):
|
|
"""Check if credential has expired."""
|
|
if self.expires_at:
|
|
from django.utils import timezone
|
|
return timezone.now() > self.expires_at
|
|
return False
|