140 lines
4.2 KiB
Python
140 lines
4.2 KiB
Python
from django.utils import timezone
|
|
|
|
from apps.audit.models import OperationAuditLog
|
|
from apps.tickets.models import Ticket, TicketEvent
|
|
|
|
|
|
class TicketStatusError(ValueError):
|
|
pass
|
|
|
|
|
|
STANDARD_TRANSITIONS: dict[str, set[str]] = {
|
|
"pending_info": {"pending", "closed"},
|
|
"pending": {"in_progress", "closed"},
|
|
"in_progress": {"pending_confirm"},
|
|
"pending_confirm": {"resolved", "in_progress", "closed"},
|
|
"resolved": {"closed"},
|
|
"closed": set(),
|
|
}
|
|
|
|
SHORT_TRANSITIONS: dict[str, set[str]] = {
|
|
"pending_info": {"pending", "resolved", "closed"},
|
|
"pending": {"resolved", "closed"},
|
|
"in_progress": {"resolved", "closed"},
|
|
"pending_confirm": {"resolved", "closed"},
|
|
"resolved": {"closed"},
|
|
"closed": set(),
|
|
}
|
|
|
|
MEDIUM_TRANSITIONS: dict[str, set[str]] = {
|
|
"pending_info": {"pending", "closed"},
|
|
"pending": {"in_progress", "resolved", "closed"},
|
|
"in_progress": {"resolved", "pending_confirm", "closed"},
|
|
"pending_confirm": {"resolved", "in_progress", "closed"},
|
|
"resolved": {"closed"},
|
|
"closed": set(),
|
|
}
|
|
|
|
CATEGORY_TRANSITION_POLICIES = {
|
|
"business_issue": SHORT_TRANSITIONS,
|
|
"operation_issue": SHORT_TRANSITIONS,
|
|
"data_issue": MEDIUM_TRANSITIONS,
|
|
"technical_issue": STANDARD_TRANSITIONS,
|
|
}
|
|
|
|
# Backward compatible export used by older tests and internal imports.
|
|
ALLOWED_TRANSITIONS = STANDARD_TRANSITIONS
|
|
|
|
|
|
def get_ticket_category_code(ticket: Ticket) -> str | None:
|
|
if ticket.category_id and ticket.category:
|
|
return ticket.category.code
|
|
return None
|
|
|
|
|
|
def get_allowed_transitions(from_status: str, category_code: str | None = None) -> set[str]:
|
|
transitions = CATEGORY_TRANSITION_POLICIES.get(category_code or "", STANDARD_TRANSITIONS)
|
|
return transitions.get(from_status, set())
|
|
|
|
|
|
def validate_status_transition(
|
|
from_status: str,
|
|
to_status: str,
|
|
category_code: str | None = None,
|
|
) -> None:
|
|
if to_status not in get_allowed_transitions(from_status, category_code):
|
|
raise TicketStatusError(f"Invalid ticket status transition: {from_status} -> {to_status}")
|
|
|
|
|
|
def transition_requires_remark(
|
|
from_status: str,
|
|
to_status: str,
|
|
category_code: str | None = None,
|
|
) -> bool:
|
|
if to_status == "resolved":
|
|
return False
|
|
if category_code in {"business_issue", "operation_issue"} and to_status == "closed":
|
|
return False
|
|
return to_status in {"in_progress", "pending_confirm", "closed"}
|
|
|
|
|
|
def transition_ticket(
|
|
ticket: Ticket,
|
|
to_status: str,
|
|
remark: str = "",
|
|
operator=None,
|
|
*,
|
|
resolution: str = "",
|
|
is_resolved: bool | None = None,
|
|
replied_to_customer: bool | None = None,
|
|
) -> Ticket:
|
|
from_status = ticket.status
|
|
category_code = get_ticket_category_code(ticket)
|
|
validate_status_transition(from_status, to_status, category_code)
|
|
ticket.status = to_status
|
|
if to_status == "resolved":
|
|
ticket.resolved_at = timezone.now()
|
|
ticket.is_resolved = True if is_resolved is None else is_resolved
|
|
elif is_resolved is not None:
|
|
ticket.is_resolved = is_resolved
|
|
if to_status == "closed":
|
|
ticket.closed_at = timezone.now()
|
|
if resolution:
|
|
ticket.resolution_summary = resolution
|
|
if replied_to_customer is not None:
|
|
ticket.replied_to_customer = replied_to_customer
|
|
ticket.save(
|
|
update_fields=[
|
|
"status",
|
|
"resolved_at",
|
|
"closed_at",
|
|
"resolution_summary",
|
|
"is_resolved",
|
|
"replied_to_customer",
|
|
"updated_at",
|
|
]
|
|
)
|
|
TicketEvent.objects.create(
|
|
ticket=ticket,
|
|
event_type="status_change",
|
|
from_value={"status": from_status},
|
|
to_value={
|
|
"status": to_status,
|
|
"resolution": ticket.resolution_summary,
|
|
"is_resolved": ticket.is_resolved,
|
|
"replied_to_customer": ticket.replied_to_customer,
|
|
},
|
|
operator=operator,
|
|
remark=remark,
|
|
)
|
|
OperationAuditLog.objects.create(
|
|
user=operator,
|
|
module="tickets",
|
|
action="status_transition",
|
|
target_type="ticket",
|
|
target_id=str(ticket.id),
|
|
result="success",
|
|
request_snapshot={"from": from_status, "to": to_status, "remark": remark},
|
|
)
|
|
return ticket
|