Autoescalation model

Signed-off-by: drona-gyawali <dronarajgyawali@gmail.com>

Author: drona-gyawali

management/core/admin.py

@@ -19,10 +19,11 @@
 Written in 2025 by Dorna Raj Gyawali <dronarajgyawali@gmail.com>
 """
 
-from core.models import Agent, Customer, Department, Ticket, User
 from django.contrib import admin
 from django.contrib.auth.admin import UserAdmin as BaseUserAdmin
 
+from core.models import Agent, Customer, Department, Ticket, User
+
 
 @admin.register(User)
 class UserAdmin(BaseUserAdmin):

management/core/api/viewset.py

@@ -11,12 +11,6 @@
 import logging
 from datetime import datetime
 
-from core import dumps, validators
-from core.constants import Status
-from core.models import Agent, Customer, Ticket
-from core.permissions import CanEditOwnOrAdmin
-from core.serializer import RegisterSerializer, TicketCreateSerializer
-from core.tasks import process_ticket_queue
 from django.db import transaction
 from django.shortcuts import get_object_or_404
 from django.utils import timezone
@@ -27,6 +21,13 @@
 from rest_framework.response import Response
 from rest_framework_simplejwt.tokens import RefreshToken
 
+from core import dumps, validators
+from core.constants import Status
+from core.models import Agent, Customer, Ticket
+from core.permissions import CanEditOwnOrAdmin
+from core.serializer import RegisterSerializer, TicketCreateSerializer
+from core.tasks import process_ticket_queue
+
 logger = logging.getLogger(__name__)
 
 

management/core/automation/state_machine.py

@@ -8,8 +8,8 @@
 
 
 class TicketStateMachine:
-    def __init__(self, ticket_id):
-        self.ticket_id = ticket_id
+    def __init__(self, ticket):
+        self.ticket_id = ticket.id
 
     def can_state_change(self, new_status):
         transition = {

management/core/migrations/0004_alter_ticket_status_statuschange_autoescalate.py

@@ -0,0 +1,107 @@
+# Generated by Django 5.2.1 on 2025-05-17 16:51
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0003_ticket_queued_at"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="ticket",
+            name="status",
+            field=models.CharField(
+                choices=[
+                    ("waiting", "Waiting"),
+                    ("assigned", "Assigned"),
+                    ("progress", "Progress"),
+                    ("completed", "Completed"),
+                    ("closed", "Closed"),
+                ],
+                default="waiting",
+                max_length=20,
+            ),
+        ),
+        migrations.CreateModel(
+            name="StatusChange",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "new_status",
+                    models.CharField(
+                        choices=[
+                            ("waiting", "Waiting"),
+                            ("assigned", "Assigned"),
+                            ("progress", "Progress"),
+                            ("completed", "Completed"),
+                            ("closed", "Closed"),
+                        ],
+                        default="waiting",
+                        max_length=20,
+                    ),
+                ),
+                ("updated_at", models.DateTimeField(auto_now_add=True)),
+                ("new_queued_at", models.DateTimeField(blank=True, null=True)),
+                (
+                    "new_agent",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="status_changes",
+                        to="core.agent",
+                    ),
+                ),
+                (
+                    "ticket",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="status_changes",
+                        to="core.ticket",
+                    ),
+                ),
+            ],
+        ),
+        migrations.CreateModel(
+            name="AutoEscalate",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                (
+                    "ticket",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="auto_escalate",
+                        to="core.ticket",
+                    ),
+                ),
+                (
+                    "status_change",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="auto_escalate",
+                        to="core.statuschange",
+                    ),
+                ),
+            ],
+        ),
+    ]

management/core/models.py

@@ -18,13 +18,16 @@
 Written in 2025 by Dorna Raj Gyawali <dronarajgyawali@gmail.com>
 """
 
-from core.constants import Role, Status
-from core.dumps import CONTEXT_403
+from datetime import timezone
+
 from django.conf import settings
 from django.contrib.auth.models import AbstractUser
 from django.db import models
 from django.forms import model_to_dict
 
+from core.constants import Role, Status
+from core.dumps import CONTEXT_403
+
 
 class User(AbstractUser):
     role = models.CharField(
@@ -164,3 +167,53 @@ def save(self, *args, **kwargs):
 
     def __str__(self):
         return self.ticket_id or f"TID-{self.pk}"
+
+
+class StatusChange(models.Model):
+    ticket = models.ForeignKey(
+        Ticket, on_delete=models.CASCADE, related_name="status_changes"
+    )
+    new_status = models.CharField(
+        max_length=20, choices=Status.choices, default=Status.WAITING
+    )
+    updated_at = models.DateTimeField(auto_now_add=True)
+    new_queued_at = models.DateTimeField(null=True, blank=True)
+    new_agent = models.ForeignKey(
+        Agent,
+        on_delete=models.SET_NULL,
+        related_name="status_changes",
+        null=True,
+        blank=True,
+    )
+
+
+class AutoEscalate(models.Model):
+    ticket = models.ForeignKey(
+        Ticket, on_delete=models.CASCADE, related_name="auto_escalate"
+    )
+    status_change = models.ForeignKey(
+        StatusChange, on_delete=models.CASCADE, related_name="auto_escalate"
+    )
+
+    @classmethod
+    def escalate_changes(cls, ticket_id, new_status, new_agent=None):
+        try:
+            ticket = Ticket.objects.get(ticket_id=ticket_id)
+            status_change = StatusChange.objects.create(
+                new_status=new_status,
+                new_agent=new_agent,
+                new_queued_at=timezone.now(),
+            )
+            cls.objects.create(ticket=ticket, status_changes=status_change)
+            ticket.status = new_status
+            if new_agent:
+                ticket.agent = new_agent
+            ticket.save()
+            return {"success": True, "message": f"Ticket {ticket_id} escalated."}
+        except Ticket.DoesNotExist:
+            return {
+                "success": False,
+                "message": f"Ticket series {ticket_id} not found.",
+            }
+        except Exception as e:
+            return {"success": False, "message": f"Error during escalation: {str(e)}"}

management/core/serializer.py

@@ -1,7 +1,8 @@
-from core.models import Agent, Customer, Department, Role, Ticket
 from django.contrib.auth import get_user_model
 from rest_framework import serializers
 
+from core.models import Agent, Customer, Department, Role, Ticket
+
 User = get_user_model()
 
 

management/core/tasks.py

@@ -1,13 +1,14 @@
 from datetime import timedelta
 
 from celery import shared_task
-from core.automation.state_machine import TicketStateMachine
-from core.constants import Status
-from core.models import Agent, Ticket
 from django.db import transaction
 from django.db.models import Q
 from django.utils import timezone
 
+from core.automation.state_machine import TicketStateMachine
+from core.constants import Status
+from core.models import Agent, Ticket
+
 
 @shared_task(bind=True)
 def process_ticket_queue(self):
@@ -55,8 +56,8 @@ def delete_completed_tickets(self):
 
 
 @shared_task(bind=True)
-def process_state_changed(ticket_id, new_staus):
+def process_state_changed(ticket_id, new_status):
     with transaction.atomic:
         ticket_id = Ticket.objects.select_for_update.get(ticket_id=ticket_id)
         state_machine = TicketStateMachine(ticket_id)
-        return state_machine.transition_to(new_staus)
+        return state_machine.transition_to(new_status)

management/core/tests/test_api.py

@@ -9,15 +9,16 @@
 from datetime import datetime, timedelta
 from unittest import mock
 
-from core.constants import Status
-from core.models import Agent, Customer, Department, Ticket, User
 from django.db.models import F
 from django.test import TestCase
 from django.urls import reverse
 from django.utils.timezone import make_aware
 from rest_framework import status
 from rest_framework.test import APIClient
 
+from core.constants import Status
+from core.models import Agent, Customer, Department, Ticket, User
+
 
 class ApiViewsTest(TestCase):
     """Test cases for all API endpoints in a single class."""

management/core/urls.py

@@ -1,6 +1,7 @@
-from core.api import viewset
 from django.urls import path
 
+from core.api import viewset
+
 urlpatterns = [
     path("customer/<int:pk>/detail/", viewset.customer_detail, name="customer_detail"),
     path("agent/<int:pk>/detail/", viewset.agent_detail, name="agent_detail"),

management/main/asgi.py

@@ -12,9 +12,10 @@
 from channels.auth import AuthMiddlewareStack
 from channels.routing import ProtocolTypeRouter, URLRouter
 from channels.security.websocket import AllowedHostsOriginValidator
-from chat import routing
 from django.core.asgi import get_asgi_application
 
+from chat import routing
+
 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "main.settings")
 
 django_asgi_app = get_asgi_application()