"""Helpdesk AI service layer. Orchestrates AI suggestion lifecycle and audit logging for the helpdesk module. Delegates to the platform AI app for actual model interaction; this bounded context handles only helpdesk-specific semantics. """ from __future__ import annotations import structlog from django.db import transaction from django.utils import timezone _log = structlog.get_logger("simorgh.helpdesk.ai") class AISuggestionService: """Manage AI-generated ticket suggestions.""" @staticmethod @transaction.atomic def create_suggestion( *, ticket_id: int, tenant_id: int, suggestion_type: str, payload: dict, provider: str = "default", confidence_score: float | None = None, rationale: str = "", model_version: str = "", ) -> "AITicketSuggestion": # noqa: F821 from simorgh.apps.events.bus import dispatch from simorgh.apps.helpdesk.ai.models import AITicketSuggestion from simorgh.apps.helpdesk.models import Ticket ticket = Ticket.objects.get(pk=ticket_id) suggestion = AITicketSuggestion.objects.create( ticket=ticket, tenant_id=tenant_id, organization_node=ticket.organization_node, provider=provider, suggestion_type=suggestion_type, payload=payload, confidence_score=confidence_score, rationale=rationale, model_version=model_version, ) dispatch("helpdesk.ai.suggestion.created", { "suggestion_id": suggestion.pk, "ticket_id": ticket_id, "tenant_id": tenant_id, "suggestion_type": suggestion_type, }) _log.info( "helpdesk.ai.suggestion.created", suggestion_id=suggestion.pk, ticket_id=ticket_id, suggestion_type=suggestion_type, ) return suggestion @staticmethod @transaction.atomic def decide_suggestion( *, suggestion_id: int, approved: bool, reviewed_by_id: int, review_comment: str = "", ) -> "AITicketSuggestion": # noqa: F821 from simorgh.apps.events.bus import dispatch from simorgh.apps.helpdesk.ai.models import ( AIActionLog, AISuggestionStatus, AITicketSuggestion, ) from simorgh.apps.helpdesk.models import Ticket suggestion = AITicketSuggestion.objects.select_related("ticket").get( pk=suggestion_id, ) now = timezone.now() suggestion.status = ( AISuggestionStatus.APPROVED if approved else AISuggestionStatus.REJECTED ) suggestion.reviewed_by_id = reviewed_by_id suggestion.reviewed_at = now suggestion.review_comment = review_comment suggestion.save() if approved: _apply_suggestion(suggestion, executed_by_id=reviewed_by_id) event = "helpdesk.ai.suggestion.approved" if approved else "helpdesk.ai.suggestion.rejected" dispatch(event, { "suggestion_id": suggestion.pk, "ticket_id": suggestion.ticket_id, "tenant_id": suggestion.tenant_id, "reviewed_by_id": reviewed_by_id, }) _log.info( event, suggestion_id=suggestion.pk, ticket_id=suggestion.ticket_id, reviewed_by_id=reviewed_by_id, ) return suggestion class AIAuditService: """Read-side queries for AI audit logs.""" @staticmethod def get_logs_for_ticket(*, ticket_id: int, tenant_id: int) -> list: from simorgh.apps.helpdesk.ai.models import AIActionLog return list( AIActionLog.objects.filter( ticket_id=ticket_id, tenant_id=tenant_id, ).order_by("-created_at") ) @staticmethod def get_pending_suggestions( *, tenant_id: int, ticket_id: int | None = None ) -> list: from simorgh.apps.helpdesk.ai.models import AISuggestionStatus, AITicketSuggestion qs = AITicketSuggestion.objects.filter( tenant_id=tenant_id, status=AISuggestionStatus.PENDING, ) if ticket_id is not None: qs = qs.filter(ticket_id=ticket_id) return list(qs.order_by("-created_at")) # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _apply_suggestion(suggestion, *, executed_by_id: int) -> None: """Execute the AI suggestion payload against the ticket and log the action.""" from simorgh.apps.helpdesk.ai.models import AIActionLog, AISuggestionType from simorgh.apps.helpdesk.models import Ticket ticket = Ticket.objects.get(pk=suggestion.ticket_id) payload = suggestion.payload or {} action_type = _suggestion_type_to_action(suggestion.suggestion_type) before_snapshot = {} try: if suggestion.suggestion_type == AISuggestionType.PRIORITY: new_priority = payload.get("priority") if new_priority: before_snapshot = {"priority": ticket.priority} ticket.priority = new_priority ticket.save(update_fields=["priority", "updated_at"]) elif suggestion.suggestion_type == AISuggestionType.CATEGORY: category_id = payload.get("category_id") if category_id: before_snapshot = {"category_id": ticket.category_id} ticket.category_id = category_id ticket.save(update_fields=["category_id", "updated_at"]) elif suggestion.suggestion_type == AISuggestionType.TAGS: tag_ids = [t.get("id") for t in payload.get("tags", []) if t.get("id")] if tag_ids: from simorgh.apps.helpdesk.models import TicketTag before_snapshot = { "tag_ids": list( TicketTag.objects.filter(ticket=ticket).values_list( "tag_id", flat=True ) ) } TicketTag.objects.filter(ticket=ticket).delete() TicketTag.objects.bulk_create([ TicketTag(ticket=ticket, tag_id=tid, added_by_id=executed_by_id) for tid in tag_ids ], ignore_conflicts=True) elif suggestion.suggestion_type == AISuggestionType.ROUTING: updates = {} if "assigned_to_id" in payload: before_snapshot["assigned_to_id"] = ticket.assigned_to_id updates["assigned_to_id"] = payload["assigned_to_id"] updates["assigned_at"] = timezone.now() if "queue_id" in payload: before_snapshot["queue_id"] = ticket.queue_id updates["queue_id"] = payload["queue_id"] if updates: for k, v in updates.items(): setattr(ticket, k, v) ticket.save(update_fields=[*list(updates.keys()), "updated_at"]) AIActionLog.objects.create( ticket=ticket, suggestion=suggestion, action_type=action_type, provider=suggestion.provider, executed_by_id=executed_by_id, payload_before=before_snapshot, payload_after=payload, tenant_id=suggestion.tenant_id, organization_node=ticket.organization_node, ) except Exception: _log.exception( "helpdesk.ai.apply_suggestion.error", suggestion_id=suggestion.pk, ticket_id=ticket.pk, ) AIActionLog.objects.create( ticket=ticket, suggestion=suggestion, action_type=action_type, provider=suggestion.provider, executed_by_id=executed_by_id, payload_before=before_snapshot, payload_after={"error": True}, error_message="Failed to apply suggestion.", tenant_id=suggestion.tenant_id, organization_node=ticket.organization_node, ) def _suggestion_type_to_action(suggestion_type: str) -> str: from simorgh.apps.helpdesk.ai.models import AISuggestionType mapping = { AISuggestionType.PRIORITY: "change_priority", AISuggestionType.CATEGORY: "change_category", AISuggestionType.TAGS: "add_tag", AISuggestionType.ROUTING: "assign", AISuggestionType.REPLY_DRAFT: "send_reply", } return mapping.get(suggestion_type, suggestion_type)