内存工单存储(生产环境应替换为数据库)
| 67 | |
| 68 | |
| 69 | class TicketStore: |
| 70 | """内存工单存储(生产环境应替换为数据库)""" |
| 71 | |
| 72 | def __init__(self): |
| 73 | self._tickets: dict[str, dict] = {} |
| 74 | |
| 75 | def create(self, ticket_type: str, priority: str, summary: str, details: str, user_id: str) -> dict: |
| 76 | ticket_id = f"TK-{datetime.now().strftime('%Y%m%d')}-{uuid.uuid4().hex[:6].upper()}" |
| 77 | ticket = { |
| 78 | "ticket_id": ticket_id, |
| 79 | "type": ticket_type, |
| 80 | "priority": priority, |
| 81 | "status": TicketStatus.CREATED.value, |
| 82 | "summary": summary, |
| 83 | "details": details, |
| 84 | "user_id": user_id, |
| 85 | "created_at": datetime.now().isoformat(), |
| 86 | "updated_at": datetime.now().isoformat(), |
| 87 | } |
| 88 | self._tickets[ticket_id] = ticket |
| 89 | return ticket |
| 90 | |
| 91 | def query(self, ticket_id: str) -> dict | None: |
| 92 | return self._tickets.get(ticket_id) |
| 93 | |
| 94 | def query_by_user(self, user_id: str) -> list[dict]: |
| 95 | return [t for t in self._tickets.values() if t["user_id"] == user_id] |
| 96 | |
| 97 | def update_status(self, ticket_id: str, status: str) -> dict | None: |
| 98 | ticket = self._tickets.get(ticket_id) |
| 99 | if ticket: |
| 100 | ticket["status"] = status |
| 101 | ticket["updated_at"] = datetime.now().isoformat() |
| 102 | return ticket |
| 103 | |
| 104 | |
| 105 | class TicketHandlerAgent: |