41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
"""客户工单中期记忆。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from repositories.work_order import WorkOrderRepo
|
|
|
|
|
|
class WorkOrderMemory:
|
|
"""以 MySQL 为事实来源读取客户工单状态。"""
|
|
|
|
def __init__(self, *, repository_factory=WorkOrderRepo):
|
|
self.repository_factory = repository_factory
|
|
|
|
async def list(self, db, customer_id: int, *, active_only: bool = True) -> list[dict[str, Any]]:
|
|
"""按客户 ID 查询工单并转换为客服上下文结构。"""
|
|
orders = await self.repository_factory(db).list_by_customer(
|
|
customer_id, active_only=active_only
|
|
)
|
|
return [
|
|
{
|
|
"id": order.id,
|
|
"work_order_no": order.work_order_no,
|
|
"order_type": order.order_type,
|
|
"sub_type": order.sub_type,
|
|
"customer_id": order.customer_id,
|
|
"handler_id": order.handler_id,
|
|
"current_node": order.current_node,
|
|
"priority": order.priority,
|
|
"status": order.status,
|
|
"biz_content": order.biz_content,
|
|
"create_time": order.create_time,
|
|
"update_time": order.update_time,
|
|
}
|
|
for order in orders
|
|
]
|
|
|
|
|
|
__all__ = ["WorkOrderMemory"]
|