"""客户工单中期记忆。""" 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"]