40 lines
1.5 KiB
Python
40 lines
1.5 KiB
Python
from django.contrib.auth import get_user_model
|
|
from django.core.management.base import BaseCommand, CommandError
|
|
|
|
from apps.tickets.draft_services import (
|
|
create_agent_run_for_draft,
|
|
execute_draft_analysis,
|
|
)
|
|
from apps.tickets.models import TicketDraft
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Run the fixed real-text ticket intake acceptance workflow."
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument("--username", default="admin")
|
|
|
|
def handle(self, *args, **options):
|
|
user = get_user_model().objects.filter(username=options["username"]).first()
|
|
if user is None:
|
|
raise CommandError("验收用户不存在")
|
|
draft = TicketDraft.objects.create(
|
|
created_by=user,
|
|
source_text=(
|
|
"生产环境中，客户 A 的采集任务 JOB-1024 从上午 10 点起没有更新，"
|
|
"影响单个客户，请技术支持排查。"
|
|
),
|
|
)
|
|
run = create_agent_run_for_draft(draft)
|
|
execute_draft_analysis(draft.id, run.id)
|
|
draft.refresh_from_db()
|
|
if draft.status != "ticket_created" or draft.confirmed_ticket_id is None:
|
|
raise CommandError(f"真实文本分析失败: {draft.error_code} {draft.error_message}")
|
|
self.stdout.write(
|
|
self.style.SUCCESS(
|
|
"Ticket intake acceptance passed: "
|
|
f"draft={draft.public_id} run={run.public_id} "
|
|
f"ticket={draft.confirmed_ticket.ticket_no} decision={draft.decision}"
|
|
)
|
|
)
|