517 lines
18 KiB
Python
517 lines
18 KiB
Python
"""任务中心领域服务，负责活动任务去重、SLA 截止时间和自动完成。"""
|
||
|
||
from datetime import timedelta
|
||
|
||
from django.conf import settings
|
||
from django.db import transaction
|
||
from django.utils import timezone
|
||
|
||
from apps.tickets.models import Ticket, TicketDraft, TicketSupportAnalysis
|
||
from apps.workitems.models import WorkItem, WorkItemStatus, WorkItemType
|
||
|
||
|
||
def work_item_due_at(task_type: str):
|
||
hours_by_type = {
|
||
WorkItemType.TICKET_INTAKE_FAILED: settings.WORK_ITEM_INTAKE_FAILURE_SLA_HOURS,
|
||
WorkItemType.SUPPORT_ANALYSIS_FAILED: settings.WORK_ITEM_SUPPORT_FAILURE_SLA_HOURS,
|
||
WorkItemType.TICKET_CUSTOMER_REPLY: settings.WORK_ITEM_CUSTOMER_REPLY_SLA_HOURS,
|
||
WorkItemType.TICKET_CLOSE: settings.WORK_ITEM_TICKET_CLOSE_SLA_HOURS,
|
||
}
|
||
hours = max(int(hours_by_type.get(task_type, settings.WORK_ITEM_DEFAULT_SLA_HOURS)), 1)
|
||
return timezone.now() + timedelta(hours=hours)
|
||
|
||
|
||
@transaction.atomic
|
||
def sync_asset_review_work_item(version, *, actor=None) -> WorkItem | None:
|
||
"""按资产版本状态创建、更新或完成唯一审核任务。"""
|
||
|
||
from apps.knowledge.models import GovernedAssetVersionStatus
|
||
|
||
active = WorkItem.objects.select_for_update().filter(
|
||
task_type=WorkItemType.ASSET_REVIEW,
|
||
object_type="governed_asset_version",
|
||
object_id=str(version.id),
|
||
status=WorkItemStatus.ACTIVE,
|
||
).first()
|
||
if version.status != GovernedAssetVersionStatus.PENDING_REVIEW:
|
||
if active is not None:
|
||
active.status = WorkItemStatus.COMPLETED
|
||
active.completed_by = actor
|
||
active.completed_at = timezone.now()
|
||
active.save(
|
||
update_fields=[
|
||
"status",
|
||
"completed_by",
|
||
"completed_at",
|
||
"updated_at",
|
||
]
|
||
)
|
||
return None
|
||
payload = {
|
||
"asset_public_id": str(version.asset.public_id),
|
||
"asset_type": version.asset.asset_type,
|
||
"object_id": version.asset.object_id,
|
||
"version": version.version,
|
||
"submitted_by_id": version.submitted_by_id,
|
||
}
|
||
if active is None:
|
||
return WorkItem.objects.create(
|
||
task_type=WorkItemType.ASSET_REVIEW,
|
||
object_type="governed_asset_version",
|
||
object_id=str(version.id),
|
||
title=f"审核{version.asset.get_asset_type_display()} v{version.version}",
|
||
description="知识资产已提交审核，发布前需由非提交人核对内容与适用范围。",
|
||
payload=payload,
|
||
created_by=actor,
|
||
due_at=work_item_due_at(WorkItemType.ASSET_REVIEW),
|
||
)
|
||
active.payload = payload
|
||
active.save(update_fields=["payload", "updated_at"])
|
||
return active
|
||
|
||
|
||
@transaction.atomic
|
||
def sync_knowledge_distillation_work_item(
|
||
ticket: Ticket,
|
||
*,
|
||
run=None,
|
||
error_code: str = "",
|
||
error_message: str = "",
|
||
actor=None,
|
||
) -> WorkItem | None:
|
||
"""根据缺失字段同步唯一待补全任务，无缺失时自动完成。"""
|
||
|
||
ticket = Ticket.objects.select_for_update().get(id=ticket.id)
|
||
active = WorkItem.objects.select_for_update().filter(
|
||
task_type=WorkItemType.KNOWLEDGE_DISTILLATION_FAILED,
|
||
object_type="ticket",
|
||
object_id=str(ticket.id),
|
||
status=WorkItemStatus.ACTIVE,
|
||
).first()
|
||
failed = bool(error_code) or bool(run and run.status == "failed")
|
||
if not failed:
|
||
if active is not None:
|
||
active.status = WorkItemStatus.COMPLETED
|
||
active.completed_by = actor or (run.created_by if run else None)
|
||
active.completed_at = timezone.now()
|
||
active.save(
|
||
update_fields=[
|
||
"status",
|
||
"completed_by",
|
||
"completed_at",
|
||
"updated_at",
|
||
]
|
||
)
|
||
return None
|
||
payload = {
|
||
"agent_run_public_id": str(run.public_id) if run is not None else "",
|
||
"error_code": error_code or (run.error_code if run is not None else ""),
|
||
"error_message": error_message
|
||
or (run.error_message if run is not None else ""),
|
||
}
|
||
if active is None:
|
||
return WorkItem.objects.create(
|
||
task_type=WorkItemType.KNOWLEDGE_DISTILLATION_FAILED,
|
||
object_type="ticket",
|
||
object_id=str(ticket.id),
|
||
ticket=ticket,
|
||
title=f"接管工单 {ticket.ticket_no} 知识沉淀",
|
||
description="知识价值评估或草稿生成未完成，需人工判断是否沉淀知识。",
|
||
payload=payload,
|
||
assignee=ticket.assignee,
|
||
created_by=actor or (run.created_by if run is not None else None),
|
||
due_at=work_item_due_at(WorkItemType.KNOWLEDGE_DISTILLATION_FAILED),
|
||
)
|
||
active.payload = payload
|
||
active.assignee = ticket.assignee
|
||
active.save(update_fields=["payload", "assignee", "updated_at"])
|
||
return active
|
||
|
||
|
||
@transaction.atomic
|
||
def sync_tool_confirmation_work_item(call, *, actor=None) -> WorkItem | None:
|
||
active = WorkItem.objects.select_for_update().filter(
|
||
task_type=WorkItemType.TOOL_CONFIRMATION,
|
||
object_type="tool_call",
|
||
object_id=str(call.public_id),
|
||
status=WorkItemStatus.ACTIVE,
|
||
).first()
|
||
if call.status != "pending_confirmation":
|
||
if active is not None:
|
||
active.status = (
|
||
WorkItemStatus.CANCELLED
|
||
if call.status == "rejected"
|
||
else WorkItemStatus.COMPLETED
|
||
)
|
||
active.completed_by = actor
|
||
active.completed_at = timezone.now()
|
||
active.save(
|
||
update_fields=[
|
||
"status",
|
||
"completed_by",
|
||
"completed_at",
|
||
"updated_at",
|
||
]
|
||
)
|
||
return None
|
||
payload = {
|
||
"tool_call_public_id": str(call.public_id),
|
||
"tool_code": call.tool.code,
|
||
"risk_level": call.tool.risk_level,
|
||
"requested_by_id": call.requested_by_id,
|
||
"agent_run_public_id": (
|
||
str(call.agent_run.public_id) if call.agent_run_id else ""
|
||
),
|
||
}
|
||
if active is None:
|
||
return WorkItem.objects.create(
|
||
task_type=WorkItemType.TOOL_CONFIRMATION,
|
||
object_type="tool_call",
|
||
object_id=str(call.public_id),
|
||
title=f"确认高风险工具：{call.tool.name}",
|
||
description="工具调用已暂停，需由非申请人核对输入、风险和业务影响。",
|
||
payload=payload,
|
||
created_by=call.requested_by,
|
||
due_at=work_item_due_at(WorkItemType.TOOL_CONFIRMATION),
|
||
)
|
||
active.payload = payload
|
||
active.save(update_fields=["payload", "updated_at"])
|
||
return active
|
||
|
||
|
||
@transaction.atomic
|
||
def sync_ticket_missing_info_work_item(
|
||
ticket: Ticket,
|
||
missing_fields: list[str],
|
||
*,
|
||
questions: list[str] | None = None,
|
||
draft_public_id: str | None = None,
|
||
actor=None,
|
||
) -> WorkItem | None:
|
||
ticket = Ticket.objects.select_for_update().get(id=ticket.id)
|
||
active = WorkItem.objects.select_for_update().filter(
|
||
task_type=WorkItemType.TICKET_MISSING_INFO,
|
||
object_type="ticket",
|
||
object_id=str(ticket.id),
|
||
status=WorkItemStatus.ACTIVE,
|
||
).first()
|
||
normalized_missing = sorted({str(code) for code in missing_fields if str(code)})
|
||
if ticket.status != "pending_info" or not normalized_missing:
|
||
if active is not None:
|
||
active.status = WorkItemStatus.COMPLETED
|
||
active.completed_by = actor
|
||
active.completed_at = timezone.now()
|
||
active.save(
|
||
update_fields=[
|
||
"status",
|
||
"completed_by",
|
||
"completed_at",
|
||
"updated_at",
|
||
]
|
||
)
|
||
return None
|
||
|
||
previous_payload = active.payload if active is not None else {}
|
||
payload = {
|
||
"missing_fields": normalized_missing,
|
||
"questions": (
|
||
[str(item) for item in questions if str(item).strip()]
|
||
if questions is not None
|
||
else previous_payload.get("questions", [])
|
||
),
|
||
"draft_public_id": (
|
||
draft_public_id
|
||
if draft_public_id is not None
|
||
else previous_payload.get("draft_public_id", "")
|
||
),
|
||
}
|
||
if active is None:
|
||
return WorkItem.objects.create(
|
||
task_type=WorkItemType.TICKET_MISSING_INFO,
|
||
object_type="ticket",
|
||
object_id=str(ticket.id),
|
||
ticket=ticket,
|
||
title=f"补全工单 {ticket.ticket_no} 信息",
|
||
description="该工单由智能受理创建，仍有业务信息需要补充。",
|
||
payload=payload,
|
||
assignee=ticket.assignee,
|
||
created_by=actor,
|
||
due_at=work_item_due_at(WorkItemType.TICKET_MISSING_INFO),
|
||
)
|
||
|
||
active.payload = payload
|
||
active.assignee = ticket.assignee
|
||
active.save(update_fields=["payload", "assignee", "updated_at"])
|
||
return active
|
||
|
||
|
||
def complete_ticket_missing_info_work_item(ticket: Ticket, *, actor=None) -> None:
|
||
sync_ticket_missing_info_work_item(ticket, [], actor=actor)
|
||
|
||
|
||
@transaction.atomic
|
||
def sync_ticket_intake_failure_work_item(
|
||
draft: TicketDraft,
|
||
*,
|
||
run=None,
|
||
actor=None,
|
||
) -> WorkItem | None:
|
||
draft = (
|
||
TicketDraft.objects.select_for_update(of=("self",))
|
||
.select_related("created_by", "latest_agent_run")
|
||
.get(id=draft.id)
|
||
)
|
||
active = WorkItem.objects.select_for_update().filter(
|
||
task_type=WorkItemType.TICKET_INTAKE_FAILED,
|
||
object_type="ticket_draft",
|
||
object_id=str(draft.public_id),
|
||
status=WorkItemStatus.ACTIVE,
|
||
).first()
|
||
if draft.status != "failed":
|
||
if active is not None:
|
||
active.status = WorkItemStatus.COMPLETED
|
||
active.completed_by = actor
|
||
active.completed_at = timezone.now()
|
||
active.save(
|
||
update_fields=[
|
||
"status",
|
||
"completed_by",
|
||
"completed_at",
|
||
"updated_at",
|
||
]
|
||
)
|
||
return None
|
||
|
||
current_run = run or draft.latest_agent_run
|
||
payload = {
|
||
"draft_public_id": str(draft.public_id),
|
||
"agent_run_public_id": (
|
||
str(current_run.public_id) if current_run is not None else ""
|
||
),
|
||
"error_code": draft.error_code,
|
||
"error_message": draft.error_message,
|
||
}
|
||
if active is None:
|
||
return WorkItem.objects.create(
|
||
task_type=WorkItemType.TICKET_INTAKE_FAILED,
|
||
object_type="ticket_draft",
|
||
object_id=str(draft.public_id),
|
||
title="接管失败的智能受理",
|
||
description="自动受理未完成，原始文本、材料和已提取字段仍可恢复。",
|
||
payload=payload,
|
||
assignee=draft.created_by,
|
||
created_by=actor or draft.created_by,
|
||
due_at=work_item_due_at(WorkItemType.TICKET_INTAKE_FAILED),
|
||
)
|
||
|
||
active.payload = payload
|
||
active.assignee = draft.created_by
|
||
active.save(update_fields=["payload", "assignee", "updated_at"])
|
||
return active
|
||
|
||
|
||
@transaction.atomic
|
||
def complete_ticket_intake_failure_work_item(
|
||
draft: TicketDraft,
|
||
*,
|
||
actor=None,
|
||
cancelled: bool = False,
|
||
) -> None:
|
||
active = WorkItem.objects.select_for_update().filter(
|
||
task_type=WorkItemType.TICKET_INTAKE_FAILED,
|
||
object_type="ticket_draft",
|
||
object_id=str(draft.public_id),
|
||
status=WorkItemStatus.ACTIVE,
|
||
).first()
|
||
if active is None:
|
||
return
|
||
active.status = WorkItemStatus.CANCELLED if cancelled else WorkItemStatus.COMPLETED
|
||
active.completed_by = actor
|
||
active.completed_at = timezone.now()
|
||
active.save(
|
||
update_fields=[
|
||
"status",
|
||
"completed_by",
|
||
"completed_at",
|
||
"updated_at",
|
||
]
|
||
)
|
||
|
||
|
||
@transaction.atomic
|
||
def sync_customer_reply_work_item(reply_task, *, actor=None) -> WorkItem | None:
|
||
"""同步唯一客户回复任务，并以回复任务状态作为完成依据。"""
|
||
|
||
active = WorkItem.objects.select_for_update().filter(
|
||
task_type=WorkItemType.TICKET_CUSTOMER_REPLY,
|
||
object_type="ticket",
|
||
object_id=str(reply_task.ticket_id),
|
||
status=WorkItemStatus.ACTIVE,
|
||
).first()
|
||
if reply_task.status not in {"draft", "approved"}:
|
||
if active is not None:
|
||
active.status = (
|
||
WorkItemStatus.COMPLETED
|
||
if reply_task.status == "sent"
|
||
else WorkItemStatus.CANCELLED
|
||
)
|
||
active.completed_by = actor
|
||
active.completed_at = timezone.now()
|
||
active.save(
|
||
update_fields=[
|
||
"status",
|
||
"completed_by",
|
||
"completed_at",
|
||
"updated_at",
|
||
]
|
||
)
|
||
return None
|
||
payload = {
|
||
"reply_task_id": reply_task.id,
|
||
"support_analysis_id": reply_task.support_analysis_id,
|
||
"agent_run_public_id": (
|
||
str(reply_task.support_analysis.run.public_id)
|
||
if reply_task.support_analysis_id
|
||
else ""
|
||
),
|
||
"revision": reply_task.revision,
|
||
"reply_status": reply_task.status,
|
||
}
|
||
if active is None:
|
||
return WorkItem.objects.create(
|
||
task_type=WorkItemType.TICKET_CUSTOMER_REPLY,
|
||
object_type="ticket",
|
||
object_id=str(reply_task.ticket_id),
|
||
ticket=reply_task.ticket,
|
||
title=f"确认并发送工单 {reply_task.ticket.ticket_no} 客户回复",
|
||
description="Agent 已生成回复草稿，需人工核对、编辑并确认已发送。",
|
||
payload=payload,
|
||
assignee=reply_task.ticket.assignee or reply_task.created_by,
|
||
created_by=actor or reply_task.created_by,
|
||
due_at=work_item_due_at(WorkItemType.TICKET_CUSTOMER_REPLY),
|
||
)
|
||
active.payload = payload
|
||
active.assignee = reply_task.ticket.assignee or reply_task.created_by
|
||
active.title = (
|
||
f"等待工单 {reply_task.ticket.ticket_no} 渠道送达回执"
|
||
if reply_task.status == "approved"
|
||
else f"确认并发送工单 {reply_task.ticket.ticket_no} 客户回复"
|
||
)
|
||
active.description = (
|
||
"回复版本已由人工确认并锁定，等待渠道送达；失败时使用原消息重试。"
|
||
if reply_task.status == "approved"
|
||
else "Agent 已生成回复草稿，需人工核对、编辑并确认已发送。"
|
||
)
|
||
active.save(
|
||
update_fields=["payload", "assignee", "title", "description", "updated_at"]
|
||
)
|
||
return active
|
||
|
||
|
||
@transaction.atomic
|
||
def sync_ticket_close_work_item(ticket: Ticket, *, actor=None) -> WorkItem | None:
|
||
"""为已解决但未关闭工单维护唯一待关闭任务。"""
|
||
|
||
ticket = Ticket.objects.select_for_update().get(id=ticket.id)
|
||
active = WorkItem.objects.select_for_update().filter(
|
||
task_type=WorkItemType.TICKET_CLOSE,
|
||
object_type="ticket",
|
||
object_id=str(ticket.id),
|
||
status=WorkItemStatus.ACTIVE,
|
||
).first()
|
||
if ticket.status != "resolved":
|
||
if active is not None:
|
||
active.status = (
|
||
WorkItemStatus.COMPLETED
|
||
if ticket.status == "closed"
|
||
else WorkItemStatus.CANCELLED
|
||
)
|
||
active.completed_by = actor
|
||
active.completed_at = timezone.now()
|
||
active.save(
|
||
update_fields=[
|
||
"status",
|
||
"completed_by",
|
||
"completed_at",
|
||
"updated_at",
|
||
]
|
||
)
|
||
return None
|
||
payload = {
|
||
"resolved_at": ticket.resolved_at.isoformat() if ticket.resolved_at else "",
|
||
}
|
||
if active is None:
|
||
return WorkItem.objects.create(
|
||
task_type=WorkItemType.TICKET_CLOSE,
|
||
object_type="ticket",
|
||
object_id=str(ticket.id),
|
||
ticket=ticket,
|
||
title=f"关闭已解决工单 {ticket.ticket_no}",
|
||
description="工单已有解决结论，需人工确认后关闭。",
|
||
payload=payload,
|
||
assignee=ticket.assignee,
|
||
created_by=actor,
|
||
due_at=work_item_due_at(WorkItemType.TICKET_CLOSE),
|
||
)
|
||
active.payload = payload
|
||
active.assignee = ticket.assignee
|
||
active.save(update_fields=["payload", "assignee", "updated_at"])
|
||
return active
|
||
|
||
|
||
@transaction.atomic
|
||
def sync_support_analysis_work_item(
|
||
run,
|
||
*,
|
||
error_code: str = "",
|
||
error_message: str = "",
|
||
) -> WorkItem | None:
|
||
try:
|
||
analysis = run.ticket_support_analysis
|
||
except TicketSupportAnalysis.DoesNotExist:
|
||
return None
|
||
active = WorkItem.objects.select_for_update().filter(
|
||
task_type=WorkItemType.SUPPORT_ANALYSIS_FAILED,
|
||
object_type="ticket",
|
||
object_id=str(analysis.ticket_id),
|
||
status=WorkItemStatus.ACTIVE,
|
||
).first()
|
||
failed = run.status == "failed" or bool(error_code)
|
||
if not failed:
|
||
if active is not None:
|
||
active.status = WorkItemStatus.COMPLETED
|
||
active.completed_by = run.created_by
|
||
active.completed_at = timezone.now()
|
||
active.save(
|
||
update_fields=[
|
||
"status",
|
||
"completed_by",
|
||
"completed_at",
|
||
"updated_at",
|
||
]
|
||
)
|
||
return None
|
||
payload = {
|
||
"support_analysis_id": analysis.id,
|
||
"agent_run_public_id": str(run.public_id),
|
||
"error_code": error_code or run.error_code,
|
||
"error_message": error_message or run.error_message,
|
||
}
|
||
if active is None:
|
||
return WorkItem.objects.create(
|
||
task_type=WorkItemType.SUPPORT_ANALYSIS_FAILED,
|
||
object_type="ticket",
|
||
object_id=str(analysis.ticket_id),
|
||
ticket=analysis.ticket,
|
||
title=f"接管工单 {analysis.ticket.ticket_no} 支持分析",
|
||
description="支持分析未形成可信结果，需人工排查或重新运行。",
|
||
payload=payload,
|
||
assignee=analysis.ticket.assignee or run.created_by,
|
||
created_by=run.created_by,
|
||
due_at=work_item_due_at(WorkItemType.SUPPORT_ANALYSIS_FAILED),
|
||
)
|
||
active.payload = payload
|
||
active.assignee = analysis.ticket.assignee or run.created_by
|
||
active.save(update_fields=["payload", "assignee", "updated_at"])
|
||
return active
|