53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
from django.utils import timezone
|
|
|
|
from apps.audit.models import OperationAuditLog
|
|
from apps.tickets.models import Ticket, TicketEvent
|
|
|
|
|
|
class TicketStatusError(ValueError):
|
|
pass
|
|
|
|
|
|
ALLOWED_TRANSITIONS: dict[str, set[str]] = {
|
|
"pending_info": {"pending", "closed"},
|
|
"pending": {"in_progress", "closed"},
|
|
"in_progress": {"pending_confirm", "resolved", "closed"},
|
|
"pending_confirm": {"resolved", "in_progress", "closed"},
|
|
"resolved": {"closed"},
|
|
"closed": set(),
|
|
}
|
|
|
|
|
|
def validate_status_transition(from_status: str, to_status: str) -> None:
|
|
if to_status not in ALLOWED_TRANSITIONS.get(from_status, set()):
|
|
raise TicketStatusError(f"Invalid ticket status transition: {from_status} -> {to_status}")
|
|
|
|
|
|
def transition_ticket(ticket: Ticket, to_status: str, remark: str = "", operator=None) -> Ticket:
|
|
from_status = ticket.status
|
|
validate_status_transition(from_status, to_status)
|
|
ticket.status = to_status
|
|
if to_status == "resolved":
|
|
ticket.resolved_at = timezone.now()
|
|
if to_status == "closed":
|
|
ticket.closed_at = timezone.now()
|
|
ticket.save(update_fields=["status", "resolved_at", "closed_at", "updated_at"])
|
|
TicketEvent.objects.create(
|
|
ticket=ticket,
|
|
event_type="status_change",
|
|
from_value={"status": from_status},
|
|
to_value={"status": to_status},
|
|
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
|