225 lines
7.4 KiB
Python
225 lines
7.4 KiB
Python
from django.contrib.auth import get_user_model
|
|
from django.utils import timezone
|
|
|
|
from apps.audit.models import OperationAuditLog
|
|
from apps.tickets.models import Ticket, TicketCategory, TicketEvent, TicketPriority
|
|
|
|
|
|
class TicketStatusError(ValueError):
|
|
pass
|
|
|
|
|
|
class TicketDataError(ValueError):
|
|
pass
|
|
|
|
|
|
REQUIRED_TICKET_FIELDS = (
|
|
"title",
|
|
"description",
|
|
"category_code",
|
|
"priority",
|
|
"problem_performance",
|
|
"problem_object",
|
|
"submitter_name",
|
|
"submitter_org",
|
|
"environment",
|
|
)
|
|
|
|
|
|
def validate_ticket_fields(fields: dict) -> tuple[dict, TicketCategory]:
|
|
normalized = {
|
|
"title": str(fields.get("title") or "").strip(),
|
|
"description": str(fields.get("description") or "").strip(),
|
|
"category_code": str(fields.get("category_code") or "").strip(),
|
|
"priority": str(fields.get("priority") or "medium").strip(),
|
|
"problem_performance": str(fields.get("problem_performance") or "").strip(),
|
|
"problem_object": str(fields.get("problem_object") or "").strip(),
|
|
"submitter_name": str(fields.get("submitter_name") or "").strip(),
|
|
"submitter_org": str(fields.get("submitter_org") or "").strip(),
|
|
"impact_scope": str(fields.get("impact_scope") or "").strip(),
|
|
"environment": str(fields.get("environment") or "").strip(),
|
|
"related_system": str(fields.get("related_system") or "").strip(),
|
|
"assignee_id": fields.get("assignee_id"),
|
|
}
|
|
missing = [code for code in REQUIRED_TICKET_FIELDS if not normalized.get(code)]
|
|
if missing:
|
|
raise TicketDataError(f"缺少必填字段: {', '.join(missing)}")
|
|
if normalized["priority"] not in TicketPriority.values:
|
|
raise TicketDataError("工单优先级无效")
|
|
category = TicketCategory.objects.filter(code=normalized["category_code"], enabled=True).first()
|
|
if category is None:
|
|
raise TicketDataError("工单分类无效或已停用")
|
|
if normalized["assignee_id"]:
|
|
assignee = get_user_model().objects.filter(id=normalized["assignee_id"]).first()
|
|
if assignee is None:
|
|
raise TicketDataError("处理人不存在")
|
|
else:
|
|
assignee = None
|
|
normalized["assignee"] = assignee
|
|
return normalized, category
|
|
|
|
|
|
def next_ticket_no() -> str:
|
|
prefix = timezone.localdate().strftime("TK%Y%m%d")
|
|
count = Ticket.objects.filter(ticket_no__startswith=prefix).count() + 1
|
|
return f"{prefix}{count:04d}"
|
|
|
|
|
|
def create_ticket_from_fields(fields: dict, user) -> Ticket:
|
|
normalized, category = validate_ticket_fields(fields)
|
|
ticket = Ticket.objects.create(
|
|
ticket_no=next_ticket_no(),
|
|
title=normalized["title"],
|
|
description=normalized["description"],
|
|
category=category,
|
|
priority=normalized["priority"],
|
|
problem_performance=normalized["problem_performance"],
|
|
problem_object=normalized["problem_object"],
|
|
submitter_name=normalized["submitter_name"],
|
|
submitter_org=normalized["submitter_org"],
|
|
impact_scope=normalized["impact_scope"],
|
|
environment=normalized["environment"],
|
|
related_system=normalized["related_system"],
|
|
assignee=normalized["assignee"],
|
|
requester=user,
|
|
created_by=user,
|
|
)
|
|
TicketEvent.objects.create(
|
|
ticket=ticket,
|
|
event_type="created",
|
|
to_value={"status": ticket.status, "category_code": normalized["category_code"]},
|
|
operator=user,
|
|
)
|
|
return ticket
|
|
|
|
|
|
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 to_status != "closed" and 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
|