X-Financial/server/src/app/repositories/employee.py

from __future__ import annotations

from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session, selectinload

from app.models.employee import Employee
from app.models.organization import OrganizationUnit
from app.models.role import Role


class EmployeeRepository:
    def __init__(self, db: Session) -> None:
        self.db = db

    def list(self, status: str | None = None, keyword: str | None = None) -> list[Employee]:
        stmt = (
            select(Employee)
            .options(
                selectinload(Employee.organization_unit),
                selectinload(Employee.manager),
                selectinload(Employee.roles),
                selectinload(Employee.change_logs),
            )
            .order_by(Employee.updated_at.desc(), Employee.name.asc())
        )

        if status and status != "全部员工":
            stmt = stmt.where(Employee.employment_status == status)

        if keyword:
            pattern = f"%{keyword.strip()}%"
            stmt = stmt.where(
                or_(
                    Employee.name.ilike(pattern),
                    Employee.employee_no.ilike(pattern),
                    Employee.email.ilike(pattern),
                    Employee.position.ilike(pattern),
                )
            )

        return list(self.db.execute(stmt).scalars().unique().all())

    def get(self, employee_id: str) -> Employee | None:
        stmt = (
            select(Employee)
            .options(
                selectinload(Employee.organization_unit),
                selectinload(Employee.manager),
                selectinload(Employee.roles),
                selectinload(Employee.change_logs),
            )
            .where(Employee.id == employee_id)
        )
        return self.db.execute(stmt).scalars().unique().first()

    def get_by_employee_no(self, employee_no: str) -> Employee | None:
        stmt = select(Employee).where(Employee.employee_no == employee_no)
        return self.db.execute(stmt).scalars().first()

    def get_by_email(self, email: str) -> Employee | None:
        stmt = select(Employee).where(Employee.email == email)
        return self.db.execute(stmt).scalars().first()

    def list_roles(self) -> list[Role]:
        stmt = select(Role)
        return list(self.db.execute(stmt).scalars().all())

    def get_role_by_code(self, role_code: str) -> Role | None:
        stmt = select(Role).where(Role.role_code == role_code)
        return self.db.execute(stmt).scalars().first()

    def list_organization_units(self) -> list[OrganizationUnit]:
        stmt = select(OrganizationUnit)
        return list(self.db.execute(stmt).scalars().all())

    def get_organization_by_code(self, unit_code: str) -> OrganizationUnit | None:
        stmt = select(OrganizationUnit).where(OrganizationUnit.unit_code == unit_code)
        return self.db.execute(stmt).scalars().first()

    def count_employees(self) -> int:
        stmt = select(func.count()).select_from(Employee)
        return int(self.db.execute(stmt).scalar_one())

    def count_roles(self) -> int:
        stmt = select(func.count()).select_from(Role)
        return int(self.db.execute(stmt).scalar_one())

    def count_organization_units(self) -> int:
        stmt = select(func.count()).select_from(OrganizationUnit)
        return int(self.db.execute(stmt).scalar_one())

    def create(self, employee: Employee) -> Employee:
        self.db.add(employee)
        self.db.commit()
        self.db.refresh(employee)
        return employee

    def save(self, employee: Employee) -> Employee:
        self.db.add(employee)
        self.db.commit()
        self.db.refresh(employee)
        return employee