600 lines
20 KiB
Python
600 lines
20 KiB
Python
"""
|
|
Observations services - Business logic for observation management.
|
|
|
|
This module provides services for:
|
|
- Creating and managing observations
|
|
- Converting observations to PX Actions
|
|
- Sending notifications
|
|
- Status management with audit logging
|
|
"""
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.db import transaction
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
from apps.notifications.services import NotificationService
|
|
from apps.organizations.models import Department, Hospital
|
|
|
|
from .models import (
|
|
Observation,
|
|
ObservationAttachment,
|
|
ObservationNote,
|
|
ObservationStatus,
|
|
ObservationStatusLog,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
User = get_user_model()
|
|
|
|
|
|
class ObservationService:
|
|
"""
|
|
Service class for observation management.
|
|
"""
|
|
|
|
@staticmethod
|
|
@transaction.atomic
|
|
def create_observation(
|
|
description: str,
|
|
severity: str = 'medium',
|
|
category=None,
|
|
title: str = '',
|
|
location_text: str = '',
|
|
incident_datetime=None,
|
|
reporter_staff_id: str = '',
|
|
reporter_name: str = '',
|
|
reporter_phone: str = '',
|
|
reporter_email: str = '',
|
|
client_ip: str = None,
|
|
user_agent: str = '',
|
|
attachments: list = None,
|
|
) -> Observation:
|
|
"""
|
|
Create a new observation.
|
|
|
|
Args:
|
|
description: Detailed description of the observation
|
|
severity: Severity level (low, medium, high, critical)
|
|
category: ObservationCategory instance (optional)
|
|
title: Short title (optional)
|
|
location_text: Location description (optional)
|
|
incident_datetime: When the incident occurred (optional, defaults to now)
|
|
reporter_staff_id: Staff ID of reporter (optional)
|
|
reporter_name: Name of reporter (optional)
|
|
reporter_phone: Phone of reporter (optional)
|
|
reporter_email: Email of reporter (optional)
|
|
client_ip: IP address of submitter (optional)
|
|
user_agent: Browser user agent (optional)
|
|
attachments: List of uploaded files (optional)
|
|
|
|
Returns:
|
|
Created Observation instance
|
|
"""
|
|
observation = Observation.objects.create(
|
|
description=description,
|
|
severity=severity,
|
|
category=category,
|
|
title=title,
|
|
location_text=location_text,
|
|
incident_datetime=incident_datetime or timezone.now(),
|
|
reporter_staff_id=reporter_staff_id,
|
|
reporter_name=reporter_name,
|
|
reporter_phone=reporter_phone,
|
|
reporter_email=reporter_email,
|
|
client_ip=client_ip,
|
|
user_agent=user_agent,
|
|
)
|
|
|
|
# Create initial status log
|
|
ObservationStatusLog.objects.create(
|
|
observation=observation,
|
|
from_status='',
|
|
to_status=ObservationStatus.NEW,
|
|
comment='Observation submitted'
|
|
)
|
|
|
|
# Handle attachments
|
|
if attachments:
|
|
for file in attachments:
|
|
ObservationAttachment.objects.create(
|
|
observation=observation,
|
|
file=file,
|
|
)
|
|
|
|
# Send notification to PX360 triage team
|
|
ObservationService.notify_new_observation(observation)
|
|
|
|
logger.info(f"Created observation {observation.tracking_code}")
|
|
return observation
|
|
|
|
@staticmethod
|
|
@transaction.atomic
|
|
def change_status(
|
|
observation: Observation,
|
|
new_status: str,
|
|
changed_by: User = None,
|
|
comment: str = '',
|
|
) -> ObservationStatusLog:
|
|
"""
|
|
Change observation status with audit logging.
|
|
|
|
Args:
|
|
observation: Observation instance
|
|
new_status: New status value
|
|
changed_by: User making the change (optional)
|
|
comment: Comment about the change (optional)
|
|
|
|
Returns:
|
|
Created ObservationStatusLog instance
|
|
"""
|
|
old_status = observation.status
|
|
|
|
# Update observation
|
|
observation.status = new_status
|
|
|
|
# Handle status-specific updates
|
|
if new_status == ObservationStatus.TRIAGED:
|
|
observation.triaged_at = timezone.now()
|
|
observation.triaged_by = changed_by
|
|
elif new_status == ObservationStatus.RESOLVED:
|
|
observation.resolved_at = timezone.now()
|
|
observation.resolved_by = changed_by
|
|
elif new_status == ObservationStatus.CLOSED:
|
|
observation.closed_at = timezone.now()
|
|
observation.closed_by = changed_by
|
|
|
|
observation.save()
|
|
|
|
# Create status log
|
|
status_log = ObservationStatusLog.objects.create(
|
|
observation=observation,
|
|
from_status=old_status,
|
|
to_status=new_status,
|
|
changed_by=changed_by,
|
|
comment=comment,
|
|
)
|
|
|
|
# Send notifications based on status change
|
|
if new_status == ObservationStatus.ASSIGNED and observation.assigned_to:
|
|
ObservationService.notify_assignment(observation)
|
|
elif new_status in [ObservationStatus.RESOLVED, ObservationStatus.CLOSED]:
|
|
ObservationService.notify_resolution(observation)
|
|
|
|
logger.info(
|
|
f"Observation {observation.tracking_code} status changed: "
|
|
f"{old_status} -> {new_status} by {changed_by}"
|
|
)
|
|
|
|
return status_log
|
|
|
|
@staticmethod
|
|
@transaction.atomic
|
|
def triage_observation(
|
|
observation: Observation,
|
|
triaged_by: User,
|
|
assigned_department: Department = None,
|
|
assigned_to: User = None,
|
|
new_status: str = None,
|
|
note: str = '',
|
|
) -> Observation:
|
|
"""
|
|
Triage an observation - assign department/owner and update status.
|
|
|
|
Args:
|
|
observation: Observation instance
|
|
triaged_by: User performing the triage
|
|
assigned_department: Department to assign (optional)
|
|
assigned_to: User to assign (optional)
|
|
new_status: New status (optional, defaults to TRIAGED or ASSIGNED)
|
|
note: Triage note (optional)
|
|
|
|
Returns:
|
|
Updated Observation instance
|
|
"""
|
|
# Update assignment
|
|
if assigned_department:
|
|
observation.assigned_department = assigned_department
|
|
if assigned_to:
|
|
observation.assigned_to = assigned_to
|
|
|
|
# Determine new status
|
|
if not new_status:
|
|
if assigned_to:
|
|
new_status = ObservationStatus.ASSIGNED
|
|
else:
|
|
new_status = ObservationStatus.TRIAGED
|
|
|
|
observation.save()
|
|
|
|
# Change status
|
|
ObservationService.change_status(
|
|
observation=observation,
|
|
new_status=new_status,
|
|
changed_by=triaged_by,
|
|
comment=note or f"Triaged by {triaged_by.get_full_name()}"
|
|
)
|
|
|
|
# Add note if provided
|
|
if note:
|
|
ObservationNote.objects.create(
|
|
observation=observation,
|
|
note=note,
|
|
created_by=triaged_by,
|
|
is_internal=True,
|
|
)
|
|
|
|
return observation
|
|
|
|
@staticmethod
|
|
@transaction.atomic
|
|
def convert_to_action(
|
|
observation: Observation,
|
|
created_by: User,
|
|
title: str = None,
|
|
description: str = None,
|
|
category: str = 'other',
|
|
priority: str = None,
|
|
assigned_department: Department = None,
|
|
assigned_to: User = None,
|
|
):
|
|
"""
|
|
Convert an observation to a PX Action.
|
|
|
|
Args:
|
|
observation: Observation instance
|
|
created_by: User creating the action
|
|
title: Action title (optional, defaults to observation title)
|
|
description: Action description (optional)
|
|
category: Action category
|
|
priority: Action priority (optional, derived from severity)
|
|
assigned_department: Department to assign (optional)
|
|
assigned_to: User to assign (optional)
|
|
|
|
Returns:
|
|
Created PXAction instance
|
|
"""
|
|
from apps.px_action_center.models import ActionSource, PXAction
|
|
|
|
# Get hospital from department or use first hospital
|
|
hospital = None
|
|
if assigned_department:
|
|
hospital = assigned_department.hospital
|
|
elif observation.assigned_department:
|
|
hospital = observation.assigned_department.hospital
|
|
else:
|
|
hospital = Hospital.objects.first()
|
|
|
|
if not hospital:
|
|
raise ValueError("No hospital found for creating action")
|
|
|
|
# Prepare title and description
|
|
if not title:
|
|
title = f"Observation {observation.tracking_code}"
|
|
if observation.title:
|
|
title += f" - {observation.title}"
|
|
elif observation.category:
|
|
title += f" - {observation.category.name_en}"
|
|
|
|
if not description:
|
|
description = f"""
|
|
Observation Details:
|
|
- Tracking Code: {observation.tracking_code}
|
|
- Category: {observation.category.name_en if observation.category else 'N/A'}
|
|
- Severity: {observation.get_severity_display()}
|
|
- Location: {observation.location_text or 'N/A'}
|
|
- Incident Date: {observation.incident_datetime.strftime('%Y-%m-%d %H:%M')}
|
|
- Reporter: {observation.reporter_display}
|
|
|
|
Description:
|
|
{observation.description}
|
|
|
|
View observation: /observations/{observation.id}/
|
|
""".strip()
|
|
|
|
# Map severity to priority
|
|
if not priority:
|
|
severity_to_priority = {
|
|
'low': 'low',
|
|
'medium': 'medium',
|
|
'high': 'high',
|
|
'critical': 'critical',
|
|
}
|
|
priority = severity_to_priority.get(observation.severity, 'medium')
|
|
|
|
# Create PX Action
|
|
action = PXAction.objects.create(
|
|
source_type=ActionSource.MANUAL,
|
|
content_type=ContentType.objects.get_for_model(Observation),
|
|
object_id=observation.id,
|
|
title=title,
|
|
description=description,
|
|
hospital=hospital,
|
|
department=assigned_department or observation.assigned_department,
|
|
category=category,
|
|
priority=priority,
|
|
severity=observation.severity,
|
|
assigned_to=assigned_to or observation.assigned_to,
|
|
metadata={
|
|
'observation_tracking_code': observation.tracking_code,
|
|
'observation_id': str(observation.id),
|
|
}
|
|
)
|
|
|
|
# Update observation with action link
|
|
observation.action_id = action.id
|
|
observation.save(update_fields=['action_id'])
|
|
|
|
# Add note to observation
|
|
ObservationNote.objects.create(
|
|
observation=observation,
|
|
note=f"Converted to PX Action: {action.id}",
|
|
created_by=created_by,
|
|
is_internal=True,
|
|
)
|
|
|
|
logger.info(
|
|
f"Observation {observation.tracking_code} converted to action {action.id}"
|
|
)
|
|
|
|
return action
|
|
|
|
@staticmethod
|
|
def add_note(
|
|
observation: Observation,
|
|
note: str,
|
|
created_by: User,
|
|
is_internal: bool = True,
|
|
) -> ObservationNote:
|
|
"""
|
|
Add a note to an observation.
|
|
|
|
Args:
|
|
observation: Observation instance
|
|
note: Note text
|
|
created_by: User creating the note
|
|
is_internal: Whether the note is internal-only
|
|
|
|
Returns:
|
|
Created ObservationNote instance
|
|
"""
|
|
return ObservationNote.objects.create(
|
|
observation=observation,
|
|
note=note,
|
|
created_by=created_by,
|
|
is_internal=is_internal,
|
|
)
|
|
|
|
@staticmethod
|
|
def add_attachment(
|
|
observation: Observation,
|
|
file,
|
|
description: str = '',
|
|
) -> ObservationAttachment:
|
|
"""
|
|
Add an attachment to an observation.
|
|
|
|
Args:
|
|
observation: Observation instance
|
|
file: Uploaded file
|
|
description: File description (optional)
|
|
|
|
Returns:
|
|
Created ObservationAttachment instance
|
|
"""
|
|
return ObservationAttachment.objects.create(
|
|
observation=observation,
|
|
file=file,
|
|
description=description,
|
|
)
|
|
|
|
@staticmethod
|
|
def notify_new_observation(observation: Observation):
|
|
"""
|
|
Send notification for new observation to PX360 triage team.
|
|
|
|
Args:
|
|
observation: Observation instance
|
|
"""
|
|
try:
|
|
# Get PX Admin users to notify
|
|
px_admins = User.objects.filter(
|
|
is_active=True,
|
|
groups__name='PX Admin'
|
|
)
|
|
|
|
subject = f"New Observation: {observation.tracking_code}"
|
|
message = f"""
|
|
A new observation has been submitted and requires triage.
|
|
|
|
Tracking Code: {observation.tracking_code}
|
|
Category: {observation.category.name_en if observation.category else 'N/A'}
|
|
Severity: {observation.get_severity_display()}
|
|
Location: {observation.location_text or 'N/A'}
|
|
Reporter: {observation.reporter_display}
|
|
|
|
Description:
|
|
{observation.description[:500]}{'...' if len(observation.description) > 500 else ''}
|
|
|
|
Please review and triage this observation.
|
|
""".strip()
|
|
|
|
for admin in px_admins:
|
|
if admin.email:
|
|
NotificationService.send_email(
|
|
email=admin.email,
|
|
subject=subject,
|
|
message=message,
|
|
related_object=observation,
|
|
metadata={
|
|
'observation_id': str(observation.id),
|
|
'tracking_code': observation.tracking_code,
|
|
'notification_type': 'new_observation',
|
|
}
|
|
)
|
|
|
|
logger.info(f"Sent new observation notification for {observation.tracking_code}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to send new observation notification: {e}")
|
|
|
|
@staticmethod
|
|
def notify_assignment(observation: Observation):
|
|
"""
|
|
Send notification when observation is assigned.
|
|
|
|
Args:
|
|
observation: Observation instance
|
|
"""
|
|
try:
|
|
if not observation.assigned_to:
|
|
return
|
|
|
|
subject = f"Observation Assigned: {observation.tracking_code}"
|
|
message = f"""
|
|
An observation has been assigned to you.
|
|
|
|
Tracking Code: {observation.tracking_code}
|
|
Category: {observation.category.name_en if observation.category else 'N/A'}
|
|
Severity: {observation.get_severity_display()}
|
|
Location: {observation.location_text or 'N/A'}
|
|
|
|
Description:
|
|
{observation.description[:500]}{'...' if len(observation.description) > 500 else ''}
|
|
|
|
Please review and take appropriate action.
|
|
""".strip()
|
|
|
|
if observation.assigned_to.email:
|
|
NotificationService.send_email(
|
|
email=observation.assigned_to.email,
|
|
subject=subject,
|
|
message=message,
|
|
related_object=observation,
|
|
metadata={
|
|
'observation_id': str(observation.id),
|
|
'tracking_code': observation.tracking_code,
|
|
'notification_type': 'observation_assigned',
|
|
}
|
|
)
|
|
|
|
logger.info(
|
|
f"Sent assignment notification for {observation.tracking_code} "
|
|
f"to {observation.assigned_to.email}"
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to send assignment notification: {e}")
|
|
|
|
@staticmethod
|
|
def notify_resolution(observation: Observation):
|
|
"""
|
|
Send internal notification when observation is resolved/closed.
|
|
|
|
Args:
|
|
observation: Observation instance
|
|
"""
|
|
try:
|
|
# Notify assigned user and department manager
|
|
recipients = []
|
|
|
|
if observation.assigned_to and observation.assigned_to.email:
|
|
recipients.append(observation.assigned_to.email)
|
|
|
|
if observation.assigned_department and observation.assigned_department.manager:
|
|
if observation.assigned_department.manager.email:
|
|
recipients.append(observation.assigned_department.manager.email)
|
|
|
|
if not recipients:
|
|
return
|
|
|
|
subject = f"Observation {observation.get_status_display()}: {observation.tracking_code}"
|
|
message = f"""
|
|
An observation has been {observation.get_status_display().lower()}.
|
|
|
|
Tracking Code: {observation.tracking_code}
|
|
Category: {observation.category.name_en if observation.category else 'N/A'}
|
|
Status: {observation.get_status_display()}
|
|
|
|
Resolution Notes:
|
|
{observation.resolution_notes or 'No resolution notes provided.'}
|
|
""".strip()
|
|
|
|
for email in set(recipients):
|
|
NotificationService.send_email(
|
|
email=email,
|
|
subject=subject,
|
|
message=message,
|
|
related_object=observation,
|
|
metadata={
|
|
'observation_id': str(observation.id),
|
|
'tracking_code': observation.tracking_code,
|
|
'notification_type': 'observation_resolved',
|
|
}
|
|
)
|
|
|
|
logger.info(f"Sent resolution notification for {observation.tracking_code}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to send resolution notification: {e}")
|
|
|
|
@staticmethod
|
|
def get_statistics(hospital=None, department=None, date_from=None, date_to=None):
|
|
"""
|
|
Get observation statistics.
|
|
|
|
Args:
|
|
hospital: Filter by hospital (optional)
|
|
department: Filter by department (optional)
|
|
date_from: Start date (optional)
|
|
date_to: End date (optional)
|
|
|
|
Returns:
|
|
Dictionary with statistics
|
|
"""
|
|
from django.db.models import Count, Q
|
|
|
|
queryset = Observation.objects.all()
|
|
|
|
if hospital:
|
|
queryset = queryset.filter(assigned_department__hospital=hospital)
|
|
if department:
|
|
queryset = queryset.filter(assigned_department=department)
|
|
if date_from:
|
|
queryset = queryset.filter(created_at__gte=date_from)
|
|
if date_to:
|
|
queryset = queryset.filter(created_at__lte=date_to)
|
|
|
|
# Status counts
|
|
status_counts = queryset.values('status').annotate(count=Count('id'))
|
|
status_dict = {item['status']: item['count'] for item in status_counts}
|
|
|
|
# Severity counts
|
|
severity_counts = queryset.values('severity').annotate(count=Count('id'))
|
|
severity_dict = {item['severity']: item['count'] for item in severity_counts}
|
|
|
|
# Category counts
|
|
category_counts = queryset.values(
|
|
'category__name_en'
|
|
).annotate(count=Count('id')).order_by('-count')[:10]
|
|
|
|
return {
|
|
'total': queryset.count(),
|
|
'new': status_dict.get('new', 0),
|
|
'triaged': status_dict.get('triaged', 0),
|
|
'assigned': status_dict.get('assigned', 0),
|
|
'in_progress': status_dict.get('in_progress', 0),
|
|
'resolved': status_dict.get('resolved', 0),
|
|
'closed': status_dict.get('closed', 0),
|
|
'rejected': status_dict.get('rejected', 0),
|
|
'duplicate': status_dict.get('duplicate', 0),
|
|
'anonymous_count': queryset.filter(
|
|
Q(reporter_staff_id='') & Q(reporter_name='')
|
|
).count(),
|
|
'severity': severity_dict,
|
|
'top_categories': list(category_counts),
|
|
}
|