147 lines
4.9 KiB
Python
147 lines
4.9 KiB
Python
"""
|
|
Service domain metadata enrichment.
|
|
Handles Service and ServiceSession entities.
|
|
"""
|
|
from typing import Dict, Any, Set
|
|
|
|
from asgiref.sync import sync_to_async
|
|
|
|
from core.models.enums import EventTypeChoices
|
|
|
|
|
|
SERVICE_EVENTS: Set[str] = {
|
|
EventTypeChoices.SERVICE_CREATED,
|
|
EventTypeChoices.SERVICE_UPDATED,
|
|
EventTypeChoices.SERVICE_DELETED,
|
|
EventTypeChoices.SERVICE_STATUS_CHANGED,
|
|
EventTypeChoices.SERVICE_COMPLETED,
|
|
EventTypeChoices.SERVICE_CANCELLED,
|
|
EventTypeChoices.SERVICE_TEAM_ASSIGNED,
|
|
EventTypeChoices.SERVICE_TEAM_UNASSIGNED,
|
|
EventTypeChoices.SERVICE_DISPATCHED,
|
|
EventTypeChoices.SERVICE_SESSION_OPENED,
|
|
EventTypeChoices.SERVICE_SESSION_CLOSED,
|
|
EventTypeChoices.SERVICE_SESSION_REVERTED,
|
|
EventTypeChoices.SERVICE_TASK_COMPLETED,
|
|
EventTypeChoices.SERVICE_TASK_UNCOMPLETED,
|
|
}
|
|
|
|
|
|
async def enrich_service_domain(entity_type: str, entity_id: str) -> Dict[str, Any]:
|
|
"""Enrich metadata for service-domain events."""
|
|
if entity_type == 'Service':
|
|
return await _load_service_metadata(entity_id)
|
|
elif entity_type == 'ServiceSession':
|
|
return await _load_service_session_metadata(entity_id)
|
|
return {}
|
|
|
|
|
|
async def _load_service_metadata(entity_id: str) -> Dict[str, Any]:
|
|
"""
|
|
Load metadata from a Service entity.
|
|
|
|
NOTE: Services use account_address -> account path.
|
|
The direct service.account FK is deprecated.
|
|
"""
|
|
from core.models.service import Service
|
|
|
|
def _load():
|
|
try:
|
|
service = Service.objects.select_related(
|
|
'account_address',
|
|
'account_address__account',
|
|
'account_address__account__customer',
|
|
).get(pk=entity_id)
|
|
|
|
return _extract_service_metadata(service)
|
|
except Service.DoesNotExist:
|
|
return {}
|
|
|
|
return await sync_to_async(_load)()
|
|
|
|
|
|
async def _load_service_session_metadata(entity_id: str) -> Dict[str, Any]:
|
|
"""
|
|
Load metadata from a ServiceSession entity.
|
|
|
|
NOTE: Uses account_address -> account path, not deprecated direct FKs.
|
|
"""
|
|
from core.models.session import ServiceSession
|
|
|
|
def _load():
|
|
try:
|
|
session = ServiceSession.objects.select_related(
|
|
'service',
|
|
'account_address',
|
|
'account_address__account',
|
|
'account_address__account__customer',
|
|
).get(pk=entity_id)
|
|
|
|
return _extract_service_session_metadata(session)
|
|
except ServiceSession.DoesNotExist:
|
|
return {}
|
|
|
|
return await sync_to_async(_load)()
|
|
|
|
|
|
def _extract_service_metadata(service) -> Dict[str, Any]:
|
|
"""Extract metadata dict from a pre-loaded Service object."""
|
|
metadata = {}
|
|
|
|
# Date
|
|
if service.date:
|
|
metadata['date'] = str(service.date)
|
|
|
|
# Traverse account_address -> account (NOT the deprecated 'service.account')
|
|
if service.account_address_id and service.account_address:
|
|
# Address
|
|
address_parts = []
|
|
if service.account_address.street_address:
|
|
address_parts.append(service.account_address.street_address)
|
|
if service.account_address.city:
|
|
address_parts.append(service.account_address.city)
|
|
if address_parts:
|
|
metadata['address'] = ', '.join(address_parts)
|
|
|
|
# Account name (via account_address)
|
|
if service.account_address.account:
|
|
metadata['account_name'] = service.account_address.account.name or ''
|
|
# Customer name (via account)
|
|
if service.account_address.account.customer:
|
|
metadata['customer_name'] = service.account_address.account.customer.name or ''
|
|
|
|
return metadata
|
|
|
|
|
|
def _extract_service_session_metadata(session) -> Dict[str, Any]:
|
|
"""Extract metadata dict from a preloaded ServiceSession object."""
|
|
metadata = {}
|
|
|
|
# Service ID for reference
|
|
if session.service_id:
|
|
metadata['service_id'] = str(session.service_id)
|
|
|
|
# Date from the service
|
|
if session.service and session.service.date:
|
|
metadata['date'] = str(session.service.date)
|
|
|
|
# Traverse account_address -> account (NOT deprecated session.account)
|
|
if session.account_address_id and session.account_address:
|
|
# Address
|
|
address_parts = []
|
|
if session.account_address.street_address:
|
|
address_parts.append(session.account_address.street_address)
|
|
if session.account_address.city:
|
|
address_parts.append(session.account_address.city)
|
|
if address_parts:
|
|
metadata['address'] = ', '.join(address_parts)
|
|
|
|
# Account name (via account_address)
|
|
if session.account_address.account:
|
|
metadata['account_name'] = session.account_address.account.name or ''
|
|
# Customer name (via account)
|
|
if session.account_address.account.customer:
|
|
metadata['customer_name'] = session.account_address.account.customer.name or ''
|
|
|
|
return metadata
|