167 lines
4.7 KiB
Python
167 lines
4.7 KiB
Python
"""
|
||
资产状态机服务
|
||
定义资产状态的转换规则和验证
|
||
"""
|
||
from typing import Dict, List, Optional
|
||
from enum import Enum
|
||
|
||
|
||
class AssetStatus(str, Enum):
|
||
"""资产状态枚举"""
|
||
PENDING = "pending" # 待入库
|
||
IN_STOCK = "in_stock" # 库存中
|
||
IN_USE = "in_use" # 使用中
|
||
TRANSFERRING = "transferring" # 调拨中
|
||
MAINTENANCE = "maintenance" # 维修中
|
||
PENDING_SCRAP = "pending_scrap" # 待报废
|
||
SCRAPPED = "scrapped" # 已报废
|
||
LOST = "lost" # 已丢失
|
||
|
||
|
||
class StateMachineService:
|
||
"""状态机服务类"""
|
||
|
||
# 状态转换规则
|
||
TRANSITIONS: Dict[str, List[str]] = {
|
||
AssetStatus.PENDING: [
|
||
AssetStatus.IN_STOCK,
|
||
AssetStatus.PENDING_SCRAP,
|
||
],
|
||
AssetStatus.IN_STOCK: [
|
||
AssetStatus.IN_USE,
|
||
AssetStatus.TRANSFERRING,
|
||
AssetStatus.MAINTENANCE,
|
||
AssetStatus.PENDING_SCRAP,
|
||
AssetStatus.LOST,
|
||
],
|
||
AssetStatus.IN_USE: [
|
||
AssetStatus.IN_STOCK,
|
||
AssetStatus.TRANSFERRING,
|
||
AssetStatus.MAINTENANCE,
|
||
AssetStatus.PENDING_SCRAP,
|
||
AssetStatus.LOST,
|
||
],
|
||
AssetStatus.TRANSFERRING: [
|
||
AssetStatus.IN_STOCK,
|
||
AssetStatus.IN_USE,
|
||
],
|
||
AssetStatus.MAINTENANCE: [
|
||
AssetStatus.IN_STOCK,
|
||
AssetStatus.IN_USE,
|
||
AssetStatus.PENDING_SCRAP,
|
||
],
|
||
AssetStatus.PENDING_SCRAP: [
|
||
AssetStatus.SCRAPPED,
|
||
AssetStatus.IN_STOCK, # 取消报废
|
||
],
|
||
AssetStatus.SCRAPPED: [], # 终态，不可转换
|
||
AssetStatus.LOST: [], # 终态，不可转换
|
||
}
|
||
|
||
# 状态显示名称
|
||
STATUS_NAMES: Dict[str, str] = {
|
||
AssetStatus.PENDING: "待入库",
|
||
AssetStatus.IN_STOCK: "库存中",
|
||
AssetStatus.IN_USE: "使用中",
|
||
AssetStatus.TRANSFERRING: "调拨中",
|
||
AssetStatus.MAINTENANCE: "维修中",
|
||
AssetStatus.PENDING_SCRAP: "待报废",
|
||
AssetStatus.SCRAPPED: "已报废",
|
||
AssetStatus.LOST: "已丢失",
|
||
}
|
||
|
||
def can_transition(self, current_status: str, target_status: str) -> bool:
|
||
"""
|
||
检查状态是否可以转换
|
||
|
||
Args:
|
||
current_status: 当前状态
|
||
target_status: 目标状态
|
||
|
||
Returns:
|
||
是否可以转换
|
||
"""
|
||
allowed_transitions = self.TRANSITIONS.get(current_status, [])
|
||
return target_status in allowed_transitions
|
||
|
||
def validate_transition(
|
||
self,
|
||
current_status: str,
|
||
target_status: str
|
||
) -> Optional[str]:
|
||
"""
|
||
验证状态转换并返回错误信息
|
||
|
||
Args:
|
||
current_status: 当前状态
|
||
target_status: 目标状态
|
||
|
||
Returns:
|
||
错误信息，如果转换有效则返回None
|
||
"""
|
||
if current_status == target_status:
|
||
return "当前状态与目标状态相同"
|
||
|
||
if current_status not in self.TRANSITIONS:
|
||
return f"无效的当前状态: {current_status}"
|
||
|
||
if target_status not in self.TRANSITIONS:
|
||
return f"无效的目标状态: {target_status}"
|
||
|
||
if not self.can_transition(current_status, target_status):
|
||
return f"无法从状态 '{self.get_status_name(current_status)}' 转换到 '{self.get_status_name(target_status)}'"
|
||
|
||
return None
|
||
|
||
def get_status_name(self, status: str) -> str:
|
||
"""
|
||
获取状态的显示名称
|
||
|
||
Args:
|
||
status: 状态值
|
||
|
||
Returns:
|
||
状态显示名称
|
||
"""
|
||
return self.STATUS_NAMES.get(status, status)
|
||
|
||
def get_allowed_transitions(self, current_status: str) -> List[str]:
|
||
"""
|
||
获取允许的转换状态列表
|
||
|
||
Args:
|
||
current_status: 当前状态
|
||
|
||
Returns:
|
||
允许转换到的状态列表
|
||
"""
|
||
return self.TRANSITIONS.get(current_status, [])
|
||
|
||
def is_terminal_state(self, status: str) -> bool:
|
||
"""
|
||
判断是否为终态
|
||
|
||
Args:
|
||
status: 状态值
|
||
|
||
Returns:
|
||
是否为终态
|
||
"""
|
||
return len(self.TRANSITIONS.get(status, [])) == 0
|
||
|
||
def get_available_statuses(self) -> List[Dict[str, str]]:
|
||
"""
|
||
获取所有可用状态列表
|
||
|
||
Returns:
|
||
状态列表，每个状态包含value和name
|
||
"""
|
||
return [
|
||
{"value": status, "name": name}
|
||
for status, name in self.STATUS_NAMES.items()
|
||
]
|
||
|
||
|
||
# 创建全局实例
|
||
state_machine_service = StateMachineService()
|