shanxia/bx/bx-backend/app/services/user.py

from sqlalchemy import distinct, func, or_
from sqlalchemy.orm import Session

from app.core.security import verify_password, hash_password
from app.models.user import RoleUser, User
from app.models.role import Role
from app.services.base import CrudService


class UserService:

    @staticmethod
    def get_by_id(db: Session, id: int) -> User | None:
        return CrudService.get_by_id(User, db, id)

    @staticmethod
    def get_paged(db: Session, page: int, page_size: int, filters: list | None = None):
        return CrudService.get_paged(User, db, page, page_size, filters)

    @staticmethod
    def create(db: Session, obj_in: dict) -> User:
        if "password" in obj_in and obj_in["password"]:
            obj_in["password"] = hash_password(obj_in["password"])
        return CrudService.create(User, db, obj_in)

    @staticmethod
    def update(db: Session, id: int, obj_in: dict) -> User | None:
        if "password" in obj_in and obj_in["password"]:
            obj_in["password"] = hash_password(obj_in["password"])
        return CrudService.update(User, db, id, obj_in)

    @staticmethod
    def soft_delete(db: Session, id: int) -> bool:
        return CrudService.soft_delete(User, db, id)

    @staticmethod
    def authenticate(db: Session, username: str, password: str) -> User | None:
        user = (
            db.query(User)
            .filter(User.username == username, User.del_flag == 0, User.status == 1)
            .first()
        )
        if not user:
            return None
        if not verify_password(password, user.password):
            return None
        return user

    @staticmethod
    def get_user_roles(db: Session, user_id: int) -> list[Role]:
        user = db.query(User).filter(User.id == user_id, User.del_flag == 0).first()
        if not user:
            return []
        return list(user.roles)

    @staticmethod
    def summary_stats(db: Session) -> dict:
        """用户管理顶部卡片：全量聚合（与列表分页无关）。"""
        base = User.del_flag == 0
        total = db.query(func.count()).select_from(User).filter(base).scalar() or 0
        enabled = db.query(func.count()).select_from(User).filter(base, User.status == 1).scalar() or 0
        disabled = db.query(func.count()).select_from(User).filter(base, User.status == 0).scalar() or 0
        admin = (
            db.query(func.count(distinct(User.id)))
            .select_from(User)
            .join(RoleUser, RoleUser.user_id == User.id)
            .join(Role, Role.id == RoleUser.role_id)
            .filter(base, or_(Role.role_code == "ADMIN", Role.role_name == "管理员"))
            .scalar()
            or 0
        )
        return {
            "total_count": int(total),
            "enabled_count": int(enabled),
            "disabled_count": int(disabled),
            "admin_count": int(admin),
        }