VberAdminPy/VapAdmin/BACKEND/domain/services/admin/sys_user_service.py

from datetime import datetime, timedelta
from typing import Optional

from sqlalchemy import select

from core.enums import StatusTypeEnum, DeleteTypeEnum
from core.exceptions import ServiceWarning
from domain.dtos import SysUserDto, SysUserCreateDto, SysUserUpdateDto
from domain.models import SysUserModel
from domain.services.admin.common_services import CommonService
from domain.services.base_services import CurdServiceBase
from utils import PwdUtil


class SysUserService(
    CurdServiceBase[SysUserModel, SysUserDto, SysUserCreateDto, SysUserUpdateDto]
):
    def __init__(self, db_name: Optional[str] = None):
        super().__init__(
            SysUserModel,
            SysUserDto,
            SysUserCreateDto,
            SysUserUpdateDto,
            db_name=db_name,
        )
        self._common_service = CommonService()

    async def get_user_detail_by_id(self, id: int):
        """
        查询用户详细信息
        :param id:
        :return:
        """
        data = await self._common_service.get_user_detail_by_id(id)
        return data

    async def get_user_by_username(self, username: str) -> SysUserModel | None:
        """
        根据用户名查询用户
        :param username:
        :return:
        """
        try:
            async with await self._get_async_db() as db:
                user = (
                    (
                        await db.execute(
                            select(SysUserModel)
                            .where(
                                SysUserModel.username == username,
                                SysUserModel.status == StatusTypeEnum.NORMAL.key,
                                SysUserModel.is_del == DeleteTypeEnum.NORMAL.key,
                            )
                            .distinct()
                        )
                    )
                    .scalars()
                    .first()
                )
                return user
        except Exception as e:
            raise ServiceWarning(f"查询用户失败：{str(e)}")

    async def get_user_by_phone(self, phone: str) -> SysUserModel | None:
        """
        根据手机号查询用户名
        :param phone:
        :return:
        """
        try:
            async with self._get_async_db() as db:
                user = (
                    (
                        await db.execute(
                            select(SysUserModel)
                            .where(
                                SysUserModel.phonenumber == phone,
                                SysUserModel.status == StatusTypeEnum.NORMAL.key,
                                SysUserModel.is_del == DeleteTypeEnum.NORMAL.key,
                            )
                            .distinct()
                        )
                    )
                    .scalars()
                    .first()
                )
                return user
        except Exception as e:
            raise ServiceWarning(f"查询用户失败：{str(e)}")

    async def get_user_by_email(self, email: str) -> SysUserModel | None:
        """
        根据电子邮箱查询用户
        :param email:
        :return:
        """
        try:
            async with self._get_async_db() as db:
                user = (
                    (
                        await db.execute(
                            select(SysUserModel)
                            .where(
                                SysUserModel.email == email,
                                SysUserModel.status == StatusTypeEnum.NORMAL.key,
                                SysUserModel.is_del == DeleteTypeEnum.NORMAL.key,
                            )
                            .distinct()
                        )
                    )
                    .scalars()
                    .first()
                )
                return user
        except Exception as e:
            raise ServiceWarning(f"查询用户失败：{str(e)}")

    async def create_user(self, create_dto: SysUserCreateDto):
        """
        创建用户
        :param create_dto:
        :return:
        """
        await self.check_user_exists(create_dto)
        create_dto.password = PwdUtil.get_password_hash(create_dto.password)
        try:
            async with self._get_async_db() as db:
                user = SysUserModel(**create_dto.to_dict())
                db.add(user)
                await db.commit()
                await db.refresh(user)
                return user
        except Exception as e:
            raise ServiceWarning(f"创建用户失败：{str(e)}")

    async def update_user(self, update_dto: SysUserUpdateDto):
        """
        更新用户
        :param update_dto:
        :return:
        """
        user = await self.check_user(update_dto.user_id)
        await self.check_user_exists(update_dto, update_dto.user_id)
        if not user:
            raise ServiceWarning("用户不存在")
        try:
            async with self._get_async_db() as db:
                new_user = self._apply_map_update_model(user, update_dto)
                await db.commit()
                await db.refresh(new_user)
                return new_user
        except Exception as e:
            raise ServiceWarning(f"更新用户失败：{str(e)}")

    async def check_user_exists(
        self, dto: SysUserCreateDto | SysUserUpdateDto, user_id: int = None
    ):
        """
        检查用户是否存在
        :param dto:
        :param user_id:
        :return:
        """
        await self.check_user_username_exists(dto.username, user_id)
        await self.check_user_phone_exists(dto.phone, user_id)
        await self.check_user_email_exists(dto.email, user_id)

    async def check_user_username_exists(self, username: str, user_id: int = None):
        """
        检查用户是否存在
        :param username:
        :param user_id:
        :return:
        """
        user = await self.get_user_by_username(username)
        if user and user.id != user_id:
            raise ServiceWarning("用户名已存在")

    async def check_user_phone_exists(self, phone: str, user_id: int = None):
        """
        检查用户是否存在
        :param phone:
        :param user_id:
        :return:
        """
        user = await self.get_user_by_phone(phone)
        if user and user.id != user_id:
            raise ServiceWarning("手机号已存在")

    async def check_user_email_exists(self, email: str, user_id: int = None):
        """
        检查用户是否存在
        :param email:
        :param user_id:
        :return:
        """
        user = await self.get_user_by_email(email)
        if user and user.id != user_id:
            raise ServiceWarning("邮箱已存在")

    async def check_user(self, user_id: int):
        """
        删除用户
        :param user_id:
        :return:
        """
        user = await self.get(user_id)
        if not user:
            raise ServiceWarning("用户不存在")
        if user.status == StatusTypeEnum.DISABLE.key:
            raise ServiceWarning("用户已禁用")
        if user.is_del == DeleteTypeEnum.DELETE.key:
            raise ServiceWarning("用户已删除")
        if user.is_lock == 1:
            raise ServiceWarning("用户已锁定")
        return user

    async def reset_password(self, user_id, new_password):
        """
        重置密码
        :param user_id:
        :param new_password:
        :return:
        """
        try:
            async with self._get_async_db() as db:
                user = await db.get(SysUserModel, user_id)
                if user:
                    user.password = PwdUtil.get_password_hash(new_password)
                    await db.commit()
                    await db.refresh(user)
                    return user
                else:
                    raise ValueError(f"未查询到ID为{user_id}的用户")
        except Exception as e:
            raise ServiceWarning(f"重置密码失败：{str(e)}")

    async def unlock_user(self, user_id):
        """
        解锁用户
        :param user_id:
        :return:
        """
        user = await self.get(user_id)
        try:
            async with self._get_async_db() as db:
                if user:
                    user.is_lock = 0
                    await db.commit()
                    await db.refresh(user)
                    return user
                else:
                    raise ValueError(f"未查询到ID为{user_id}的用户")
        except Exception as e:
            raise ServiceWarning(f"解锁用户失败：{str(e)}")

    async def login(self, username: str, password: str):
        """
        登录
        :param username:
        :param password:
        :return:
        """
        if not isinstance(username, str) or not username.strip():
            raise ServiceWarning("用户名不能为空")
        try:
            user = await self.get_user_by_username(username)
            if not user:
                user = await self.get_user_by_phone(username)
            if not user:
                user = await self.get_user_by_email(username)
            if not user:
                raise ServiceWarning("用户不存在")
            if user.status == StatusTypeEnum.DISABLE.key:
                raise ServiceWarning("用户已禁用")
            if user.is_del == DeleteTypeEnum.DELETE.key:
                raise ServiceWarning("用户已删除")
            if user.is_lock == 1 and datetime.now() - user.lock_at <= timedelta(
                minutes=10
            ):
                raise ServiceWarning("用户已锁定")
            if PwdUtil.verify_password(password, user.password):
                async with self._get_async_db() as db:
                    user.login_fail_count = 0
                    user.is_lock = 0
                    await db.commit()
                    await db.refresh(user)
                return user
            else:
                async with self._get_async_db() as db:
                    if user.login_fail_count >= 5:
                        user.is_lock = 1
                        user.lock_at = datetime.now()
                    else:
                        user.login_fail_count += 1
                    await db.commit()
                    await db.refresh(user)
                raise ServiceWarning("密码错误")

        except Exception as e:
            raise ServiceWarning(f"登录失败：{str(e)}")

    async def change_password(self, user_id, old_password, new_password):
        """
        修改密码
        :param user_id:
        :param old_password:
        :param new_password:
        :return:
        """
        user = await self.check_user(user_id)
        if not PwdUtil.verify_password(old_password, user.password):
            raise ServiceWarning("旧密码错误")
        try:
            async with self._get_async_db() as db:
                user.password = PwdUtil.get_password_hash(new_password)
                await db.commit()
                await db.refresh(user)
                return user
        except Exception as e:
            raise ServiceWarning(f"修改密码失败：{str(e)}")