financial_system/financial_system/repositories/employee_repository.py

from __future__ import annotations

from sqlalchemy import or_
from sqlalchemy.orm import Session

from ..database.orm import EmployeeORM


class EmployeeRepository:
    """员工仓库：封装 employees 表的基础维护。"""

    def __init__(self, session: Session):
        self.session = session

    def create_employee(
        self,
        *,
        employee_no: str,
        name: str,
        dingtalk_user_id: str = "",
        department: str = "",
        position: str = "",
        phone: str = "",
        employment_status: str = "active",
        remark: str = "",
    ) -> EmployeeORM:
        employee = EmployeeORM(
            employee_no=employee_no,
            name=name,
            dingtalk_user_id=dingtalk_user_id,
            department=department,
            position=position,
            phone=phone,
            employment_status=employment_status,
            remark=remark,
        )
        self.session.add(employee)
        self.session.commit()
        self.session.refresh(employee)
        return employee

    def update_employee(
        self,
        employee_id: int,
        *,
        employee_no: str,
        name: str,
        dingtalk_user_id: str,
        department: str,
        position: str,
        phone: str,
        employment_status: str,
        remark: str,
    ) -> EmployeeORM:
        employee = self.require_by_id(employee_id)
        employee.employee_no = employee_no
        employee.name = name
        employee.dingtalk_user_id = dingtalk_user_id
        employee.department = department
        employee.position = position
        employee.phone = phone
        employee.employment_status = employment_status
        employee.remark = remark
        self.session.commit()
        self.session.refresh(employee)
        return employee

    def get_by_id(self, employee_id: int) -> EmployeeORM | None:
        return self.session.query(EmployeeORM).filter(EmployeeORM.id == employee_id).one_or_none()

    def require_by_id(self, employee_id: int) -> EmployeeORM:
        employee = self.get_by_id(employee_id)
        if employee is None:
            raise ValueError("员工不存在")
        return employee

    def get_by_employee_no(self, employee_no: str) -> EmployeeORM | None:
        return self.session.query(EmployeeORM).filter(EmployeeORM.employee_no == employee_no).one_or_none()

    def list_employee_numbers_by_prefix(self, prefix: str) -> list[str]:
        """查询指定前缀的员工编号，供服务层生成下一个编号。"""
        rows = (
            self.session.query(EmployeeORM.employee_no)
            .filter(EmployeeORM.employee_no.like(f"{prefix}%"))
            .all()
        )
        return [row[0] for row in rows]

    def list_employees(
        self,
        *,
        keyword: str | None = None,
        employee_keyword: str | None = None,
        organization_keyword: str | None = None,
        employment_status: str | None = None,
    ) -> list[EmployeeORM]:
        query = self._filtered_query(
            keyword=keyword,
            employee_keyword=employee_keyword,
            organization_keyword=organization_keyword,
            employment_status=employment_status,
        )
        return query.order_by(EmployeeORM.id.desc()).all()

    def list_employees_page(
        self,
        *,
        page: int,
        page_size: int,
        keyword: str | None = None,
        employee_keyword: str | None = None,
        organization_keyword: str | None = None,
        employment_status: str | None = None,
    ) -> tuple[list[EmployeeORM], int]:
        query = self._filtered_query(
            keyword=keyword,
            employee_keyword=employee_keyword,
            organization_keyword=organization_keyword,
            employment_status=employment_status,
        )
        total = query.count()
        items = (
            query.order_by(EmployeeORM.id.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
            .all()
        )
        return items, total

    def _filtered_query(
        self,
        *,
        keyword: str | None,
        employee_keyword: str | None,
        organization_keyword: str | None,
        employment_status: str | None,
    ):
        query = self.session.query(EmployeeORM)
        if employment_status:
            query = query.filter(EmployeeORM.employment_status == employment_status)
        if keyword:
            like_keyword = f"%{keyword}%"
            query = query.filter(
                or_(
                    EmployeeORM.employee_no.like(like_keyword),
                    EmployeeORM.name.like(like_keyword),
                    EmployeeORM.dingtalk_user_id.like(like_keyword),
                    EmployeeORM.department.like(like_keyword),
                    EmployeeORM.position.like(like_keyword),
                    EmployeeORM.phone.like(like_keyword),
                )
            )
        if employee_keyword:
            like_employee_keyword = f"%{employee_keyword}%"
            query = query.filter(
                or_(
                    EmployeeORM.employee_no.like(like_employee_keyword),
                    EmployeeORM.name.like(like_employee_keyword),
                    EmployeeORM.dingtalk_user_id.like(like_employee_keyword),
                    EmployeeORM.phone.like(like_employee_keyword),
                )
            )
        if organization_keyword:
            like_organization_keyword = f"%{organization_keyword}%"
            query = query.filter(
                or_(
                    EmployeeORM.department.like(like_organization_keyword),
                    EmployeeORM.position.like(like_organization_keyword),
                )
            )
        return query