ismail
7369d08012
All checks were successful
Build and Push Docker Image / build (push) Successful in 4m14s
Reference numbers (unified scheme PREFIX-YYYYMM-HOSP-NNNN, e.g. CMP-202606-HHN-0001): - new ReferenceSequence model + generate_reference() helper (apps/core) - Complaint/Inquiry/Observation/Appreciation/Suggestion emit unified refs via save() - prefix-based auto-routing in public track API (CMP/INQ/OBS trackable; APR/SGT internal-only) - removed legacy CMP-/INQ- generators in ui_views, integrations, px_sources - migrations: core.0003_referencesequence, appreciation.0006, feedback.0008, observations.0012 - unit tests (format, sanitization, monthly reset, 40-thread concurrency) QA audit: - isolated E2E hospital sandbox mirroring HH-N + 10 role users (create_e2e_isolated_env) - feedback-modules-audit.spec.ts + audit helper (headed, run-to-completion) - reports/feedback-modules-qa-report.md Also bundles accumulated in-progress work across complaints, observations, organizations, templates, and other modules.
374 lines
12 KiB
Python
374 lines
12 KiB
Python
"""
|
|
Services for Staff management
|
|
"""
|
|
import secrets
|
|
import string
|
|
from django.contrib.auth import get_user_model
|
|
from django.template.loader import render_to_string
|
|
from django.conf import settings
|
|
from django.utils import timezone
|
|
|
|
from apps.core.services import AuditService
|
|
from apps.notifications.services import NotificationService
|
|
from apps.accounts.services import PasswordResetTokenService
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class StaffService:
|
|
"""Service for managing staff user accounts"""
|
|
|
|
@staticmethod
|
|
def generate_username(staff):
|
|
"""
|
|
Generate a unique username from staff name.
|
|
Format: first.last (lowercase)
|
|
If duplicate exists, append number.
|
|
"""
|
|
base_username = f"{staff.first_name.lower()}.{staff.last_name.lower()}"
|
|
username = base_username
|
|
counter = 1
|
|
|
|
# Ensure uniqueness
|
|
while User.objects.filter(username=username).exists():
|
|
username = f"{base_username}{counter}"
|
|
counter += 1
|
|
|
|
return username
|
|
|
|
@staticmethod
|
|
def generate_password(length=12):
|
|
"""
|
|
Generate a secure random password.
|
|
"""
|
|
alphabet = string.ascii_letters + string.digits + string.punctuation
|
|
password = ''.join(secrets.choice(alphabet) for _ in range(length))
|
|
return password
|
|
|
|
@staticmethod
|
|
def create_user_for_staff(staff, role='staff', request=None):
|
|
"""
|
|
Create a User account for a Staff member.
|
|
If a user with the same email already exists, link it to the staff member instead.
|
|
|
|
Args:
|
|
staff: Staff instance
|
|
role: Role name to assign (default: 'staff')
|
|
request: HTTP request for audit logging
|
|
|
|
Returns:
|
|
tuple: (User instance, was_created: bool, password: None)
|
|
- was_created is True if a new user was created
|
|
- password is always None because users set passwords through a secure email link
|
|
|
|
Raises:
|
|
ValueError: If staff already has a user account or has no email
|
|
"""
|
|
if staff.user:
|
|
raise ValueError("Staff member already has a user account")
|
|
|
|
# Generate email (required for authentication)
|
|
if not staff.email:
|
|
raise ValueError("Staff member must have an email address")
|
|
|
|
# Check if user with this email already exists
|
|
existing_user = User.objects.filter(email=staff.email).first()
|
|
|
|
if existing_user:
|
|
# Link existing user to staff
|
|
staff.user = existing_user
|
|
staff.save(update_fields=['user'])
|
|
|
|
# Update user's organization data if not set
|
|
if not existing_user.hospital:
|
|
existing_user.hospital = staff.hospital
|
|
if not existing_user.department:
|
|
existing_user.department = staff.department
|
|
if not existing_user.employee_id:
|
|
existing_user.employee_id = staff.employee_id
|
|
existing_user.save(update_fields=['hospital', 'department', 'employee_id'])
|
|
|
|
# Assign role if not already assigned
|
|
from apps.accounts.models import Role as RoleModel
|
|
try:
|
|
role_obj = RoleModel.objects.get(name=role)
|
|
if not existing_user.groups.filter(id=role_obj.group.id).exists():
|
|
existing_user.groups.add(role_obj.group)
|
|
except RoleModel.DoesNotExist:
|
|
pass
|
|
|
|
# Log the action
|
|
if request:
|
|
AuditService.log_from_request(
|
|
event_type='other',
|
|
description=f"Existing user account linked to staff member {staff.get_full_name()}",
|
|
request=request,
|
|
content_object=existing_user,
|
|
metadata={
|
|
'staff_id': str(staff.id),
|
|
'staff_name': staff.get_full_name(),
|
|
'user_id': str(existing_user.id),
|
|
'action': 'linked_existing_user'
|
|
}
|
|
)
|
|
|
|
return existing_user, False, None # Existing user was linked, no password
|
|
|
|
# Create new user account
|
|
# Generate username (optional, for backward compatibility)
|
|
username = StaffService.generate_username(staff)
|
|
|
|
# Create user - email is now the username field
|
|
user = User.objects.create_user(
|
|
email=staff.email,
|
|
password=None,
|
|
first_name=staff.first_name,
|
|
last_name=staff.last_name,
|
|
username=username, # Optional field
|
|
employee_id=staff.employee_id,
|
|
hospital=staff.hospital,
|
|
department=staff.department,
|
|
is_active=True,
|
|
is_provisional=False
|
|
)
|
|
|
|
# Assign role
|
|
from apps.accounts.models import Role as RoleModel
|
|
try:
|
|
role_obj = RoleModel.objects.get(name=role)
|
|
user.groups.add(role_obj.group)
|
|
except RoleModel.DoesNotExist:
|
|
pass
|
|
|
|
# Link to staff
|
|
staff.user = user
|
|
staff.save(update_fields=['user'])
|
|
|
|
# Log the action
|
|
if request:
|
|
AuditService.log_from_request(
|
|
event_type='user_creation',
|
|
description=f"User account created for staff member {staff.get_full_name()}",
|
|
request=request,
|
|
content_object=user,
|
|
metadata={
|
|
'staff_id': str(staff.id),
|
|
'staff_name': staff.get_full_name(),
|
|
'role': role,
|
|
'action': 'created_new_user'
|
|
}
|
|
)
|
|
|
|
return user, True, None # New user was created with no emailed password
|
|
|
|
@staticmethod
|
|
def link_user_to_staff(staff, user_id, request=None):
|
|
"""
|
|
Link an existing User account to a Staff member.
|
|
|
|
Args:
|
|
staff: Staff instance
|
|
user_id: UUID of the user to link
|
|
request: HTTP request for audit logging
|
|
|
|
Returns:
|
|
Staff: Updated staff instance
|
|
|
|
Raises:
|
|
ValueError: If staff already has a user account or user not found
|
|
"""
|
|
if staff.user:
|
|
raise ValueError("Staff member already has a user account")
|
|
|
|
try:
|
|
user = User.objects.get(id=user_id)
|
|
except User.DoesNotExist:
|
|
raise ValueError("User not found")
|
|
|
|
# Link to staff
|
|
staff.user = user
|
|
staff.save(update_fields=['user'])
|
|
|
|
# Update user's organization data
|
|
if not user.hospital:
|
|
user.hospital = staff.hospital
|
|
if not user.department:
|
|
user.department = staff.department
|
|
if not user.employee_id:
|
|
user.employee_id = staff.employee_id
|
|
user.save(update_fields=['hospital', 'department', 'employee_id'])
|
|
|
|
# Log the action
|
|
if request:
|
|
AuditService.log_from_request(
|
|
event_type='other',
|
|
description=f"User {user.email} linked to staff member {staff.get_full_name()}",
|
|
request=request,
|
|
content_object=staff,
|
|
metadata={'user_id': str(user.id)}
|
|
)
|
|
|
|
return staff
|
|
|
|
@staticmethod
|
|
def unlink_user_from_staff(staff, request=None):
|
|
"""
|
|
Remove User account association from a Staff member.
|
|
|
|
Args:
|
|
staff: Staff instance
|
|
request: HTTP request for audit logging
|
|
|
|
Returns:
|
|
Staff: Updated staff instance
|
|
|
|
Raises:
|
|
ValueError: If staff has no user account
|
|
"""
|
|
if not staff.user:
|
|
raise ValueError("Staff member has no user account")
|
|
|
|
user = staff.user
|
|
staff.user = None
|
|
staff.save(update_fields=['user'])
|
|
|
|
# Log the action
|
|
if request:
|
|
AuditService.log_from_request(
|
|
event_type='other',
|
|
description=f"User {user.email} unlinked from staff member {staff.get_full_name()}",
|
|
request=request,
|
|
content_object=staff,
|
|
metadata={'user_id': str(user.id)}
|
|
)
|
|
|
|
return staff
|
|
|
|
@staticmethod
|
|
def send_password_reset_email(staff, request=None):
|
|
"""
|
|
Send a one-time password reset link to a staff member.
|
|
"""
|
|
if not staff.email:
|
|
raise ValueError("Staff member has no email address")
|
|
|
|
user = staff.user
|
|
if not user:
|
|
raise ValueError("Staff member has no user account")
|
|
|
|
reset_token = PasswordResetTokenService.create_reset_token(user)
|
|
if request:
|
|
base_url = request.build_absolute_uri("/")
|
|
else:
|
|
base_url = settings.SITE_URL if hasattr(settings, "SITE_URL") else "http://localhost:8000"
|
|
reset_url = PasswordResetTokenService.build_reset_url(base_url, reset_token)
|
|
|
|
context = {
|
|
'staff': staff,
|
|
'user': user,
|
|
'reset_url': reset_url,
|
|
}
|
|
|
|
subject = "Set Your PX360 Password"
|
|
html_message = render_to_string('organizations/emails/staff_credentials.html', context)
|
|
|
|
plain_message = f"""Welcome to PX360!
|
|
|
|
Dear {staff.get_full_name()},
|
|
|
|
Your PX360 account has been created successfully. For your security, no password is sent by email.
|
|
|
|
Username: {user.username}
|
|
Email: {staff.email}
|
|
|
|
Set your password here: {reset_url}
|
|
|
|
This link expires in 24 hours. If you did not request this, please contact your system administrator.
|
|
|
|
Best regards,
|
|
The PX360 Team
|
|
"""
|
|
|
|
notification_log = NotificationService.send_email(
|
|
email=staff.email,
|
|
subject=subject,
|
|
message=plain_message,
|
|
html_message=html_message,
|
|
related_object=staff,
|
|
metadata={
|
|
'notification_type': 'staff_password_reset',
|
|
'staff_id': str(staff.id),
|
|
'user_id': str(user.id),
|
|
'username': user.username
|
|
}
|
|
)
|
|
|
|
if request:
|
|
AuditService.log_from_request(
|
|
event_type='password_reset',
|
|
description=f"Password reset link sent to {staff.email} for staff member {staff.get_full_name()}",
|
|
request=request,
|
|
content_object=staff,
|
|
metadata={
|
|
'notification_log_id': str(notification_log.id) if notification_log else None
|
|
}
|
|
)
|
|
|
|
return notification_log
|
|
|
|
@staticmethod
|
|
def send_credentials_email(staff, password=None, request=None):
|
|
return StaffService.send_password_reset_email(staff, request)
|
|
|
|
@staticmethod
|
|
def reset_password_and_resend_credentials(staff, request=None):
|
|
"""
|
|
Reset user password and resend password reset link.
|
|
|
|
Args:
|
|
staff: Staff instance
|
|
request: HTTP request for building absolute URLs and audit logging
|
|
|
|
Returns:
|
|
tuple: (None, notification_log: NotificationLog)
|
|
|
|
Raises:
|
|
ValueError: If staff has no user account or no email
|
|
"""
|
|
if not staff.user:
|
|
raise ValueError("Staff member has no user account")
|
|
|
|
if not staff.email:
|
|
raise ValueError("Staff member has no email address")
|
|
|
|
notification_log = StaffService.send_password_reset_email(staff, request)
|
|
|
|
# Log the action
|
|
if request:
|
|
AuditService.log_from_request(
|
|
event_type='password_reset',
|
|
description=f"Password reset link resent to {staff.email} for staff member {staff.get_full_name()}",
|
|
request=request,
|
|
content_object=staff,
|
|
metadata={
|
|
'user_id': str(staff.user.id),
|
|
'notification_log_id': str(notification_log.id) if notification_log else None
|
|
}
|
|
)
|
|
|
|
return None, notification_log
|
|
|
|
@staticmethod
|
|
def get_staff_type_role(staff_type):
|
|
"""
|
|
Map staff_type to role name.
|
|
Currently all staff get the 'staff' role.
|
|
"""
|
|
role_mapping = {
|
|
'physician': 'staff',
|
|
'nurse': 'staff',
|
|
'admin': 'staff',
|
|
'other': 'staff'
|
|
}
|
|
return role_mapping.get(staff_type, 'staff')
|