zcglxt/app/services/state_machine_service.py

"""
资产状态机服务
定义资产状态的转换规则和验证
"""
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()