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

"""Person service for business logic operations."""
from sqlalchemy.exc import IntegrityError
from app import db
from app.models.person import Person
from app.utils.validators import is_valid_name


class PersonService:
    """Service class for Person CRUD operations."""
    
    @staticmethod
    def create(name):
        """Create a new person.
        
        Args:
            name: Person's name
            
        Returns:
            Tuple of (person_dict, error_message)
            On success: (person_dict, None)
            On failure: (None, error_message)
        """
        if not is_valid_name(name):
            return None, "人员名称不能为空"
        
        # Check for duplicate name
        existing = Person.query.filter(Person.name == name.strip()).first()
        if existing:
            return None, "人员名称已存在"
        
        try:
            person = Person(name=name.strip())
            db.session.add(person)
            db.session.commit()
            return person.to_dict(), None
        except IntegrityError:
            db.session.rollback()
            return None, "人员名称已存在"
    
    @staticmethod
    def get_all():
        """Get all persons.
        
        Returns:
            List of person dictionaries
        """
        persons = Person.query.order_by(Person.id).all()
        return [p.to_dict() for p in persons]
    
    @staticmethod
    def get_by_id(person_id):
        """Get a person by ID.
        
        Args:
            person_id: Person's ID
            
        Returns:
            Tuple of (person_dict, error_message)
            On success: (person_dict, None)
            On failure: (None, error_message)
        """
        person = db.session.get(Person, person_id)
        if not person:
            return None, f"未找到ID为 {person_id} 的人员"
        
        return person.to_dict(), None
    
    @staticmethod
    def update(person_id, name):
        """Update a person's name.
        
        Args:
            person_id: Person's ID
            name: New name
            
        Returns:
            Tuple of (person_dict, error_message)
            On success: (person_dict, None)
            On failure: (None, error_message)
        """
        if not is_valid_name(name):
            return None, "人员名称不能为空"
        
        person = db.session.get(Person, person_id)
        if not person:
            return None, f"未找到ID为 {person_id} 的人员"
        
        # Check for duplicate name (excluding current person)
        existing = Person.query.filter(
            Person.name == name.strip(),
            Person.id != person_id
        ).first()
        if existing:
            return None, "人员名称已存在"
        
        try:
            person.name = name.strip()
            db.session.commit()
            return person.to_dict(), None
        except IntegrityError:
            db.session.rollback()
            return None, "人员名称已存在"
    
    @staticmethod
    def delete(person_id):
        """Delete a person.
        
        Args:
            person_id: Person's ID
            
        Returns:
            Tuple of (success, error_message)
            On success: (True, None)
            On failure: (False, error_message)
        """
        person = db.session.get(Person, person_id)
        if not person:
            return False, f"未找到ID为 {person_id} 的人员"
        
        # Check if person has associated work records
        if person.work_records.count() > 0:
            return False, "无法删除已有工作记录的人员"
        
        db.session.delete(person)
        db.session.commit()
        
        return True, None