work-statistics/backend/app/services/item_service.py

"""Item service for business logic operations."""
from sqlalchemy.exc import IntegrityError
from app import db
from app.models.item import Item
from app.models.supplier import Supplier
from app.utils.validators import is_valid_name, is_positive_number


class ItemService:
    """Service class for Item CRUD operations."""
    
    @staticmethod
    def create(name, unit_price, supplier_id=None):
        """Create a new item.
        
        Args:
            name: Item's name
            unit_price: Price per unit
            supplier_id: Optional supplier ID
            
        Returns:
            Tuple of (item_dict, error_message)
            On success: (item_dict, None)
            On failure: (None, error_message)
        """
        if not is_valid_name(name):
            return None, "物品名称不能为空"
        
        if not is_positive_number(unit_price):
            return None, "单价必须为正数"
        
        # Check for duplicate name
        existing = Item.query.filter(Item.name == name.strip()).first()
        if existing:
            return None, "物品名称已存在"
        
        # Validate supplier_id if provided
        if supplier_id is not None:
            supplier = db.session.get(Supplier, supplier_id)
            if not supplier:
                return None, f"未找到ID为 {supplier_id} 的供应商"
        
        try:
            item = Item(name=name.strip(), unit_price=float(unit_price), supplier_id=supplier_id)
            db.session.add(item)
            db.session.commit()
            return item.to_dict(), None
        except IntegrityError:
            db.session.rollback()
            return None, "物品名称已存在"
    
    @staticmethod
    def get_all():
        """Get all items.
        
        Returns:
            List of item dictionaries
        """
        items = Item.query.order_by(Item.id).all()
        return [i.to_dict() for i in items]
    
    @staticmethod
    def get_by_id(item_id):
        """Get an item by ID.
        
        Args:
            item_id: Item's ID
            
        Returns:
            Tuple of (item_dict, error_message)
            On success: (item_dict, None)
            On failure: (None, error_message)
        """
        item = db.session.get(Item, item_id)
        if not item:
            return None, f"未找到ID为 {item_id} 的物品"
        
        return item.to_dict(), None
    
    @staticmethod
    def update(item_id, name=None, unit_price=None, supplier_id=None):
        """Update an item's name, unit_price, and/or supplier_id.
        
        Args:
            item_id: Item's ID
            name: New name (optional)
            unit_price: New unit price (optional)
            supplier_id: New supplier ID (optional, use -1 to clear)
            
        Returns:
            Tuple of (item_dict, error_message)
            On success: (item_dict, None)
            On failure: (None, error_message)
        """
        item = db.session.get(Item, item_id)
        if not item:
            return None, f"未找到ID为 {item_id} 的物品"
        
        if name is not None:
            if not is_valid_name(name):
                return None, "物品名称不能为空"
            
            # Check for duplicate name (excluding current item)
            existing = Item.query.filter(
                Item.name == name.strip(),
                Item.id != item_id
            ).first()
            if existing:
                return None, "物品名称已存在"
            
            item.name = name.strip()
        
        if unit_price is not None:
            if not is_positive_number(unit_price):
                return None, "单价必须为正数"
            item.unit_price = float(unit_price)
        
        # Handle supplier_id update
        if supplier_id is not None:
            if supplier_id == -1 or supplier_id == '':
                # Clear supplier
                item.supplier_id = None
            else:
                # Validate supplier exists
                supplier = db.session.get(Supplier, supplier_id)
                if not supplier:
                    return None, f"未找到ID为 {supplier_id} 的供应商"
                item.supplier_id = supplier_id
        
        try:
            db.session.commit()
            return item.to_dict(), None
        except IntegrityError:
            db.session.rollback()
            return None, "物品名称已存在"
    
    @staticmethod
    def delete(item_id):
        """Delete an item.
        
        Args:
            item_id: Item's ID
            
        Returns:
            Tuple of (success, error_message)
            On success: (True, None)
            On failure: (False, error_message)
        """
        item = db.session.get(Item, item_id)
        if not item:
            return False, f"未找到ID为 {item_id} 的物品"
        
        # Check if item has associated work records
        if item.work_records.count() > 0:
            return False, "无法删除已有工作记录的物品"
        
        db.session.delete(item)
        db.session.commit()
        
        return True, None