"""工单领域服务。 所有工单写入都应经过本模块，以统一执行字段校验、人工字段保护、状态机、证据记录和审计。 """ from django.contrib.auth import get_user_model from django.db import transaction from django.utils import timezone from django.utils.dateparse import parse_datetime from apps.audit.models import OperationAuditLog from apps.tickets.models import ( Ticket, TicketActorSource, TicketCategory, TicketDraft, TicketEvent, TicketFieldEvidence, TicketPriority, TicketStatus, ) class TicketStatusError(ValueError): pass class TicketDataError(ValueError): pass class TicketFieldProtectedError(TicketDataError): pass def schedule_ticket_search_index(ticket_id: int) -> None: from apps.knowledge.tasks import dispatch_ticket_search_index transaction.on_commit(lambda: dispatch_ticket_search_index(ticket_id)) REQUIRED_TICKET_FIELDS = ( "title", "description", "category_code", "priority", "problem_performance", "problem_object", "submitter_name", "submitter_org", "environment", ) EDITABLE_TICKET_FIELDS = { "title", "description", "category_code", "priority", "problem_performance", "problem_object", "submitter_name", "submitter_org", "impact_scope", "environment", "related_system", "product_name", "module_name", "tags", "attempted_actions", "error_information", "customer_request", "current_result", "assignee_id", "problem_started_at", "customer_last_message_at", "support_first_response_at", "support_last_reply_at", } BUSINESS_TIME_FIELDS = { "problem_started_at", "customer_last_message_at", "support_first_response_at", "support_last_reply_at", } def normalize_tags(value) -> list[str]: if not isinstance(value, list): return [] tags = [] for item in value: tag = str(item).strip() if tag and tag not in tags: tags.append(tag[:50]) if len(tags) >= 20: break return tags def normalize_business_datetime(value): if value in (None, ""): return None parsed = parse_datetime(value.strip()) if isinstance(value, str) else value if parsed is None or not hasattr(parsed, "utcoffset") or timezone.is_naive(parsed): return None return parsed def _json_value(value): if hasattr(value, "isoformat"): return value.isoformat() return value def _field_value(ticket: Ticket, field_name: str): if field_name == "category_code": return ticket.category.code if ticket.category_id else None if field_name == "assignee_id": return ticket.assignee_id return getattr(ticket, field_name) def get_missing_ticket_fields(ticket: Ticket) -> list[str]: missing = [] for field_name in REQUIRED_TICKET_FIELDS: if field_name == "priority": continue value = _field_value(ticket, field_name) if not value: missing.append(field_name) continue if field_name == "title": evidence = ticket.field_evidence.filter(field_name="title").order_by( "-created_at" ).first() if evidence and evidence.evidence.get("generated_from") == "description": missing.append(field_name) return missing def get_outstanding_intake_fields(ticket: Ticket) -> list[str]: missing = set(get_missing_ticket_fields(ticket)) try: draft = ticket.source_draft except TicketDraft.DoesNotExist: return sorted(missing) draft_missing = { code for code in draft.missing_fields if code in EDITABLE_TICKET_FIELDS and not _field_value(ticket, code) } normalized = sorted(missing | draft_missing) if normalized != sorted(draft.missing_fields): draft.missing_fields = normalized draft.save(update_fields=["missing_fields", "updated_at"]) return normalized def record_field_evidence( ticket: Ticket, field_name: str, value, *, source: str, operator=None, confidence=None, evidence: dict | None = None, agent_run_id: int | None = None, is_protected: bool = False, ) -> TicketFieldEvidence: return TicketFieldEvidence.objects.create( ticket=ticket, field_name=field_name, value=_json_value(value), source=source, confidence=confidence, evidence=evidence or {}, agent_run_id=agent_run_id, is_protected=is_protected, created_by=operator, ) def has_protected_field(ticket: Ticket, field_name: str) -> bool: latest = ( TicketFieldEvidence.objects.filter(ticket=ticket, field_name=field_name) .order_by("-created_at", "-id") .first() ) return bool(latest and latest.is_protected) 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(), "product_name": str(fields.get("product_name") or "").strip(), "module_name": str(fields.get("module_name") or "").strip(), "tags": normalize_tags(fields.get("tags")), "attempted_actions": str(fields.get("attempted_actions") or "").strip(), "error_information": str(fields.get("error_information") or "").strip(), "customer_request": str(fields.get("customer_request") or "").strip(), "current_result": str(fields.get("current_result") or "").strip(), "assignee_id": fields.get("assignee_id"), "problem_started_at": normalize_business_datetime(fields.get("problem_started_at")), "customer_last_message_at": normalize_business_datetime( fields.get("customer_last_message_at") ), "support_first_response_at": normalize_business_datetime( fields.get("support_first_response_at") ), "support_last_reply_at": normalize_business_datetime( fields.get("support_last_reply_at") ), } 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}" @transaction.atomic def create_ticket_from_fields( fields: dict, user, *, source: str = TicketActorSource.HUMAN, agent_run_id: int | None = None, evidence_by_field: dict | None = None, confidence_by_field: dict | None = None, ) -> 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"], product_name=normalized["product_name"], module_name=normalized["module_name"], tags=normalized["tags"], attempted_actions=normalized["attempted_actions"], error_information=normalized["error_information"], customer_request=normalized["customer_request"], current_result=normalized["current_result"], assignee=normalized["assignee"], problem_started_at=normalized["problem_started_at"], customer_last_message_at=normalized["customer_last_message_at"], support_first_response_at=normalized["support_first_response_at"], support_last_reply_at=normalized["support_last_reply_at"], 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, source=source, agent_run_id=agent_run_id, ) evidence_by_field = evidence_by_field or {} confidence_by_field = confidence_by_field or {} for field_name in EDITABLE_TICKET_FIELDS: value = _field_value(ticket, field_name) if value not in (None, ""): record_field_evidence( ticket, field_name, value, source=source, operator=user, confidence=confidence_by_field.get(field_name), evidence=evidence_by_field.get(field_name), agent_run_id=agent_run_id, is_protected=source == TicketActorSource.HUMAN, ) from apps.tickets.similarity import schedule_ticket_similarity_refresh schedule_ticket_similarity_refresh(ticket) return ticket @transaction.atomic def create_ticket_from_intake( fields: dict, user, *, agent_run_id: int, evidence_by_field: dict | None = None, confidence_by_field: dict | None = None, human_field_names: set[str] | None = None, reported_missing_fields: list[str] | None = None, ) -> tuple[Ticket, list[str]]: """在问题描述有效后尽早建单，不为缺失业务事实编造默认值。 返回正式工单及仍待补全的字段；Agent 字段保留证据但默认不受人工字段保护。 """ 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(), "product_name": str(fields.get("product_name") or "").strip(), "module_name": str(fields.get("module_name") or "").strip(), "tags": normalize_tags(fields.get("tags")), "attempted_actions": str(fields.get("attempted_actions") or "").strip(), "error_information": str(fields.get("error_information") or "").strip(), "customer_request": str(fields.get("customer_request") or "").strip(), "current_result": str(fields.get("current_result") or "").strip(), "assignee_id": fields.get("assignee_id"), "problem_started_at": normalize_business_datetime(fields.get("problem_started_at")), "customer_last_message_at": normalize_business_datetime( fields.get("customer_last_message_at") ), "support_first_response_at": normalize_business_datetime( fields.get("support_first_response_at") ), "support_last_reply_at": normalize_business_datetime( fields.get("support_last_reply_at") ), } if not normalized["description"]: raise TicketDataError("尚未形成有效问题描述") category = TicketCategory.objects.filter( code=normalized["category_code"], enabled=True, ).first() if category is None: normalized["category_code"] = "" if normalized["priority"] not in TicketPriority.values: normalized["priority"] = TicketPriority.MEDIUM missing_fields = [ code for code in REQUIRED_TICKET_FIELDS if code != "priority" and ( not normalized.get(code) or (code == "category_code" and category is None) ) ] missing_fields = sorted( set(missing_fields) | { code for code in (reported_missing_fields or []) if code in EDITABLE_TICKET_FIELDS and not normalized.get(code) } ) if not normalized["title"]: normalized["title"] = f"待补全问题：{normalized['description'][:170]}".strip() assignee = None if normalized["assignee_id"]: assignee = get_user_model().objects.filter(id=normalized["assignee_id"]).first() ticket = Ticket.objects.create( ticket_no=next_ticket_no(), title=normalized["title"][:200], description=normalized["description"], category=category, status=TicketStatus.PENDING_INFO if missing_fields else TicketStatus.PENDING, 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"], product_name=normalized["product_name"], module_name=normalized["module_name"], tags=normalized["tags"], attempted_actions=normalized["attempted_actions"], error_information=normalized["error_information"], customer_request=normalized["customer_request"], current_result=normalized["current_result"], assignee=assignee, problem_started_at=normalized["problem_started_at"], customer_last_message_at=normalized["customer_last_message_at"], support_first_response_at=normalized["support_first_response_at"], support_last_reply_at=normalized["support_last_reply_at"], requester=user, created_by=user, ) TicketEvent.objects.create( ticket=ticket, event_type="created", to_value={ "status": ticket.status, "category_code": normalized["category_code"] or None, }, operator=user, source=TicketActorSource.AGENT, agent_run_id=agent_run_id, metadata={"missing_fields": missing_fields, "creation_mode": "auto_intake"}, ) evidence_by_field = evidence_by_field or {} confidence_by_field = confidence_by_field or {} human_field_names = human_field_names or set() for field_name in EDITABLE_TICKET_FIELDS: value = _field_value(ticket, field_name) if value in (None, ""): continue evidence = evidence_by_field.get(field_name) or {} if field_name == "title" and "title" in missing_fields: evidence = {**evidence, "generated_from": "description"} field_source = ( TicketActorSource.HUMAN if field_name in human_field_names else TicketActorSource.AGENT ) record_field_evidence( ticket, field_name, value, source=field_source, operator=user, confidence=( None if field_source == TicketActorSource.HUMAN else confidence_by_field.get(field_name) ), evidence=evidence, agent_run_id=agent_run_id, is_protected=field_source == TicketActorSource.HUMAN, ) from apps.tickets.similarity import schedule_ticket_similarity_refresh schedule_ticket_similarity_refresh(ticket) return ticket, missing_fields @transaction.atomic def update_ticket_fields( ticket: Ticket, changes: dict, operator=None, *, source: str = TicketActorSource.HUMAN, agent_run_id: int | None = None, evidence_by_field: dict | None = None, confidence_by_field: dict | None = None, ) -> tuple[Ticket, dict[str, str]]: """通过统一入口更新工单字段，并返回被人工保护规则拦截的字段及原因。""" unknown_fields = set(changes) - EDITABLE_TICKET_FIELDS if unknown_fields: raise TicketDataError(f"不允许修改字段: {', '.join(sorted(unknown_fields))}") ticket = Ticket.objects.select_for_update().select_related("category").get(id=ticket.id) evidence_by_field = evidence_by_field or {} confidence_by_field = confidence_by_field or {} before: dict = {} applied: dict = {} skipped: dict[str, str] = {} update_fields: set[str] = set() for field_name, value in changes.items(): if source == TicketActorSource.AGENT and has_protected_field(ticket, field_name): skipped[field_name] = "protected_by_human" continue before[field_name] = _json_value(_field_value(ticket, field_name)) if field_name == "tags": ticket.tags = normalize_tags(value) update_fields.add("tags") elif field_name in BUSINESS_TIME_FIELDS: normalized_datetime = normalize_business_datetime(value) if value not in (None, "") and normalized_datetime is None: raise TicketDataError(f"{field_name} 必须包含明确时区") setattr(ticket, field_name, normalized_datetime) update_fields.add(field_name) elif field_name == "assignee_id": assignee = get_user_model().objects.filter(id=value).first() if value else None if value and assignee is None: raise TicketDataError("处理人不存在") ticket.assignee = assignee update_fields.add("assignee") elif field_name == "category_code": category = ( TicketCategory.objects.filter(code=value, enabled=True).first() if value else None ) if value and category is None: raise TicketDataError("工单分类无效或已停用") ticket.category = category update_fields.add("category") else: setattr(ticket, field_name, value) update_fields.add(field_name) applied[field_name] = _json_value(_field_value(ticket, field_name)) if update_fields: ticket.save(update_fields=[*sorted(update_fields), "updated_at"]) for field_name, value in applied.items(): record_field_evidence( ticket, field_name, value, source=source, operator=operator, confidence=confidence_by_field.get(field_name), evidence=evidence_by_field.get(field_name), agent_run_id=agent_run_id, is_protected=source == TicketActorSource.HUMAN, ) TicketEvent.objects.create( ticket=ticket, event_type="updated", from_value=before, to_value=applied, operator=operator, source=source, agent_run_id=agent_run_id, metadata={"skipped_fields": skipped}, ) OperationAuditLog.objects.create( user=operator, module="tickets", action="update", target_type="ticket", target_id=str(ticket.id), result="success", request_snapshot={ "fields": sorted(applied), "skipped_fields": skipped, "source": source, "agent_run_id": agent_run_id, }, ) from apps.workitems.services import sync_ticket_missing_info_work_item outstanding_fields = get_outstanding_intake_fields(ticket) if ticket.status == TicketStatus.PENDING_INFO and not outstanding_fields: previous_status = ticket.status ticket.status = TicketStatus.PENDING ticket.save(update_fields=["status", "updated_at"]) TicketEvent.objects.create( ticket=ticket, event_type="status_change", from_value={"status": previous_status}, to_value={"status": ticket.status}, operator=operator, source=TicketActorSource.SYSTEM, agent_run_id=agent_run_id, remark="待补全信息已完成，工单进入待处理", ) sync_ticket_missing_info_work_item( ticket, outstanding_fields, actor=operator, ) if { "title", "description", "problem_performance", "error_information", "product_name", "module_name", "related_system", "attempted_actions", "customer_request", "current_result", "tags", } & set(applied): from apps.tickets.similarity import schedule_ticket_similarity_refresh schedule_ticket_similarity_refresh(ticket) if ticket.status in {TicketStatus.RESOLVED, TicketStatus.CLOSED}: schedule_ticket_search_index(ticket.id) return ticket, skipped def get_ticket_resolution(ticket: Ticket): from apps.solutions.models import TicketResolution return TicketResolution.objects.filter(ticket=ticket).select_related("linked_solution").first() def record_ticket_resolution( ticket: Ticket, solution_text: str, operator=None, *, resolution_type: str = "resolved", root_cause: str = "", linked_solution=None, source: str = TicketActorSource.HUMAN, agent_run_id: int | None = None, ): from apps.solutions.models import TicketResolution normalized_text = solution_text.strip() if not normalized_text: raise TicketDataError("解决方案不能为空") resolution, _ = TicketResolution.objects.update_or_create( ticket=ticket, defaults={ "resolution_type": resolution_type, "source": source, "root_cause": root_cause, "solution_text": normalized_text, "linked_solution": linked_solution, "agent_run_id": agent_run_id, "confirmed_by": operator if source == TicketActorSource.HUMAN else None, }, ) return resolution 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"} @transaction.atomic def transition_ticket( ticket: Ticket, to_status: str, remark: str = "", operator=None, *, resolution: str = "", replied_to_customer: bool | None = None, resolution_type: str = "resolved", root_cause: str = "", linked_solution=None, source: str = TicketActorSource.HUMAN, agent_run_id: int | None = None, ) -> Ticket: """校验状态机和必需业务结果后完成状态流转、事件记录及后续任务同步。""" from_status = ticket.status category_code = get_ticket_category_code(ticket) validate_status_transition(from_status, to_status, category_code) resolution_record = get_ticket_resolution(ticket) if resolution: resolution_record = record_ticket_resolution( ticket, resolution, operator, resolution_type=resolution_type, root_cause=root_cause, linked_solution=linked_solution, source=source, agent_run_id=agent_run_id, ) if to_status == "resolved" and resolution_record is None: raise TicketDataError("标记已解决前必须提供解决方案") ticket.status = to_status if to_status == "resolved": ticket.resolved_at = ticket.resolved_at or timezone.now() if to_status == "closed": ticket.closed_at = timezone.now() if replied_to_customer is not None: ticket.replied_to_customer = replied_to_customer ticket.save( update_fields=[ "status", "resolved_at", "closed_at", "replied_to_customer", "updated_at", ] ) if from_status == "pending_info" and to_status != "pending_info": from apps.workitems.services import complete_ticket_missing_info_work_item complete_ticket_missing_info_work_item(ticket, actor=operator) from apps.workitems.services import sync_ticket_close_work_item sync_ticket_close_work_item(ticket, actor=operator) TicketEvent.objects.create( ticket=ticket, event_type="status_change", from_value={"status": from_status}, to_value={ "status": to_status, "resolution_id": resolution_record.id if resolution_record else None, "resolution": resolution_record.solution_text if resolution_record else "", "is_resolved": resolution_record is not None, "replied_to_customer": ticket.replied_to_customer, }, operator=operator, source=source, agent_run_id=agent_run_id, 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, "source": source, "agent_run_id": agent_run_id, }, ) if to_status == "resolved": from apps.knowledge.tasks import dispatch_ticket_knowledge_distillation actor_id = operator.id if operator is not None else None transaction.on_commit( lambda: dispatch_ticket_knowledge_distillation(ticket.id, actor_id) ) schedule_ticket_search_index(ticket.id) return ticket @transaction.atomic def adopt_solution_for_ticket( ticket: Ticket, solution, operator=None, *, remark: str = "采纳解决方案并标记工单已解决", replied_to_customer: bool = False, source: str = TicketActorSource.HUMAN, agent_run_id: int | None = None, ): from apps.solutions.models import Solution, TicketSolutionLink locked_solution = Solution.objects.select_for_update().get(id=solution.id) if locked_solution.status != "published": raise TicketDataError("只能采纳已发布的解决方案") locked_ticket = ( Ticket.objects.select_for_update() .select_related("category", "resolution") .get(id=ticket.id) ) from_status = locked_ticket.status link, _ = TicketSolutionLink.objects.get_or_create( ticket=locked_ticket, solution=locked_solution, link_type="adopted", defaults={ "agent_run_id": agent_run_id, "created_by": operator, }, ) transition_ticket( locked_ticket, TicketStatus.RESOLVED, remark, operator, resolution=locked_solution.solution, replied_to_customer=replied_to_customer, root_cause=locked_solution.root_cause, linked_solution=locked_solution, source=source, agent_run_id=agent_run_id, ) locked_solution.usage_count = locked_solution.ticket_links.filter( link_type="adopted" ).count() locked_solution.save(update_fields=["usage_count", "updated_at"]) locked_ticket.events.create( event_type="solution_adopted", from_value={"status": from_status}, to_value={ "status": TicketStatus.RESOLVED, "solution_id": locked_solution.id, "solution_title": locked_solution.title, }, operator=operator, source=source, agent_run_id=agent_run_id, remark=remark, ) OperationAuditLog.objects.create( user=operator, module="tickets", action="solution_adopted", target_type="ticket", target_id=str(locked_ticket.id), request_snapshot={ "solution_id": locked_solution.id, "source": source, "agent_run_id": agent_run_id, }, result="success", ) return locked_ticket, link @transaction.atomic def set_ticket_archived(ticket: Ticket, archived: bool, operator=None) -> Ticket: ticket = Ticket.objects.select_for_update().get(id=ticket.id) before = ticket.archived_at ticket.archived_at = timezone.now() if archived else None ticket.archived_by = operator if archived else None ticket.save(update_fields=["archived_at", "archived_by", "updated_at"]) TicketEvent.objects.create( ticket=ticket, event_type="archived" if archived else "restored", from_value={"archived_at": _json_value(before)}, to_value={"archived_at": _json_value(ticket.archived_at)}, operator=operator, ) OperationAuditLog.objects.create( user=operator, module="tickets", action="archive" if archived else "restore", target_type="ticket", target_id=str(ticket.id), result="success", request_snapshot={}, ) schedule_ticket_search_index(ticket.id) return ticket