Backend: AutoClose Rule Initilaized

Author: drona-gyawali

management/core/admin.py

@@ -19,11 +19,10 @@
 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,6 +11,12 @@
 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
@@ -21,13 +27,6 @@
 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/auto_close.py

@@ -0,0 +1,66 @@
+"""
+The AutoClose class, which is responsible for automatically closing
+inactive tickets after a specified period of inactivity. It extends the BaseRule class
+and provides methods to check if the rule should be applied and to apply the rule.
+
+The AutoClose class ensures that tickets in a "WAITING" status for a defined number of
+days are closed automatically, helping to manage ticket lifecycles efficiently.
+
+Copyright (c) Supportix. All rights reserved.
+Written in 2025 by Dorna Raj Gyawali <dronarajgyawali@gmail.com>
+"""
+
+from datetime import timedelta
+
+from core.automation.base_rule import BaseRule
+from core.constants import Status
+from core.models import AutoEscalate, Ticket
+from django.utils import timezone
+
+
+class AutoClose(BaseRule):
+    def __init__(self, ticket_id, **kwargs):
+        super().__init__(ticket_id, **kwargs)
+        self.inactive_days = kwargs.get("inactive_days", 90)
+
+    def should_apply(self):
+        try:
+            ticket = Ticket.objects.get(ticket_id=self.ticket_id)
+        except Ticket.DoesNotExist:
+            return False
+
+        return (
+            ticket.status == Status.WAITING
+            and ticket.created_at < timezone.now() - timedelta(days=self.inactive_days)
+        )
+
+    def apply(self):
+        try:
+            ticket = Ticket.objects.get(ticket_id=self.ticket_id)
+            if not self.should_apply():
+                return {
+                    "operation": "apply",
+                    "status": "failed",
+                    "reason": "condition doesnot meet",
+                }
+            kwargs = {
+                "ticket_id": ticket,
+                "new_status": Status.CLOSED,
+            }
+            if ticket.agent:
+                kwargs["new_agent"] = ticket.agent
+
+            result = AutoEscalate.escalate_changes(**kwargs)
+            return {
+                "operation": "apply",
+                "status": "success" if result["success"] else "failed",
+                "details": result["message"],
+            }
+        except Ticket.DoesNotExist:
+            return {
+                "operation": "apply",
+                "status": "failed",
+                "reason": "Ticket not found",
+            }
+        except Exception as e:
+            return {"operation": "apply", "status": "failed", "reason": str(e)}

management/core/automation/base_rule.py

@@ -0,0 +1,51 @@
+"""
+BaseRule is a base class for rules used for automation.
+
+This abstract base class defines the structure and contract for automation rules
+that can be applied to tickets. Subclasses must implement the abstract methods
+to provide specific rule logic.
+
+Copyright (c) Supportix. All rights reserved.
+Written in 2025 by Dorna Raj Gyawali <dronarajgyawali@gmail.com>
+"""
+
+from abc import ABC, abstractmethod
+
+
+class BaseRule(ABC):
+    """
+    BaseRule is an abstract base class that defines the structure for automation rules.
+
+    Attributes:
+        ticket_id (Any): The identifier of the ticket to which the rule applies.
+        params (dict): Additional parameters for the rule.
+
+    Methods:
+        should_apply():
+            Abstract method to determine if the rule should be applied.
+            Must be implemented by subclasses.
+
+        apply():
+            Abstract method to apply the rule logic.
+            Must be implemented by subclasses.
+    """
+
+    def __init__(self, ticket, **kwargs):
+        self.ticket_id = ticket
+        self.params = kwargs
+
+    @abstractmethod
+    def should_apply(self):
+        """
+        Abstract method to determine if the rule should be applied.
+        Must be implemented by subclasses.
+        """
+        pass
+
+    @abstractmethod
+    def apply(self):
+        """
+        Abstract method to determine if the rule should be applied.
+        Must be implemented by subclasses.
+        """
+        pass

management/core/automation/rule_runner.py

@@ -0,0 +1,32 @@
+"""
+RuleEngine is a class responsible for managing and executing a set of rules on a given ticket.
+Attributes:
+    ticket_id (int): The unique identifier of the ticket to which the rules will be applied.
+    rules (list): A list of rule objects that define the logic to be executed.
+Methods:
+    __init__(ticket_id):
+        Initializes the RuleEngine with a ticket ID and a predefined set of rules.
+    run():
+        Executes all the rules in the `rules` list. For each rule, it checks if the rule should be applied
+        using the `should_apply` method. If applicable, it applies the rule using the `apply` method and
+        collects the results in a context list. Returns the context containing details of applied rules.
+
+Copyright (c) Supportix. All rights reserved.
+Written in 2025 by Dorna Raj Gyawali <dronarajgyawali@gmail.com>
+"""
+
+from core.automation.auto_close import AutoClose
+
+
+class RuleEngine:
+    def __init__(self, ticket_id):
+        self.ticket_id = ticket_id
+        self.rules = [AutoClose(ticket_id, inactive_days=1)]
+
+    def run(self):
+        context = []
+        for rule in self.rules:
+            if rule.should_apply():
+                result = rule.apply()
+                context.append({"rule": rule.__class__.__name__, "details": result})
+        return context

management/core/models.py

@@ -18,15 +18,13 @@
 Written in 2025 by Dorna Raj Gyawali <dronarajgyawali@gmail.com>
 """
 
-from datetime import timezone
-
+from core.constants import Role, Status
+from core.dumps import CONTEXT_403
 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
+from django.utils import timezone
 
 
 class User(AbstractUser):
@@ -200,11 +198,12 @@ 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(
+                ticket=ticket,
                 new_status=new_status,
                 new_agent=new_agent,
                 new_queued_at=timezone.now(),
             )
-            cls.objects.create(ticket=ticket, status_changes=status_change)
+            cls.objects.create(ticket=ticket, status_change=status_change)
             ticket.status = new_status
             if new_agent:
                 ticket.agent = new_agent

management/core/serializer.py

@@ -1,8 +1,7 @@
+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,12 @@
 from datetime import timedelta
 
 from celery import shared_task
-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
+from django.db import transaction
+from django.db.models import Q
+from django.utils import timezone
 
 
 @shared_task(bind=True)

management/core/tests/test_api.py

@@ -9,16 +9,15 @@
 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/tests/test_auto_close.py

@@ -0,0 +1,67 @@
+from datetime import timedelta
+
+from core.automation.auto_close import AutoClose
+from core.automation.rule_runner import RuleEngine
+from core.models import Agent, Customer, Department, Status, Ticket, User
+from django.test import TestCase
+from django.utils import timezone
+
+
+class AutoCloseRuleTestCase(TestCase):
+    def setUp(self):
+        self.department = Department.objects.create(name="Support")
+
+        # Create user with agent role
+        self.agent_user = User.objects.create_user(
+            username="testagent",
+            email="testagent@example.com",
+            password="testpassword123",
+            role="agent",
+        )
+        self.agent = Agent.objects.create(
+            user=self.agent_user,
+            department=self.department,
+            is_available=True,
+            max_customers=5,
+            current_customers=0,
+        )
+
+        # Create user with customer role
+        self.customer_user = User.objects.create_user(
+            username="testcustomer",
+            email="testcustomer@example.com",
+            password="testpassword123",
+            role="customer",
+        )
+        self.customer = Customer.objects.create(
+            user=self.customer_user,
+            is_paid=True,
+        )
+
+        # Create ticket normally
+        self.ticket = Ticket.objects.create(
+            ticket_id="TID-TEST123",
+            customer=self.customer,
+            agent=self.agent,
+            issue_title="Test issue for autoclose",
+            issue_desc="Description",
+            status=Status.WAITING,
+        )
+        Ticket.objects.filter(pk=self.ticket.pk).update(
+            created_at=timezone.now() - timedelta(days=2)
+        )
+        self.ticket.refresh_from_db()
+
+    def test_autoclose_rule_escalates_ticket(self):
+        rule = AutoClose(ticket_id=self.ticket.ticket_id, inactive_days=1)
+
+        self.assertTrue(rule.should_apply())
+
+        result = rule.apply()
+
+        self.ticket.refresh_from_db()
+
+        self.assertEqual(self.ticket.status, Status.CLOSED)
+
+        self.assertEqual(result["status"], "success")
+        self.assertIn("escalated", result["details"].lower())