leaudit-platform-backend/fastapi_modules/fastapi_leaudit/services/impl/authServiceImpl.py

"""认证服务实现。

从旧项目 app/routes/auth.py 和 app/auth/auth.py 迁移，业务逻辑完全不变。
仅重组为 Controller → Service(interface+impl) → Model 结构。
"""

from fastapi_common.fastapi_common_logger import logger
from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
from fastapi_common.fastapi_common_security.jwtService import JwtService
from fastapi_common.fastapi_common_web.exception.LeauditException import LeauditException
from fastapi_common.fastapi_common_web.domain.responses import StatusCodeEnum

from fastapi_modules.fastapi_leaudit.domian.vo.auth.loginTokenVo import LoginTokenVO
from fastapi_modules.fastapi_leaudit.services.authService import IAuthService


class AuthServiceImpl(IAuthService):
    """认证服务实现。"""

    async def PasswordLogin(self, Sub: str, Password: str) -> LoginTokenVO:
        """账密登录。

        校验 sso_users 表：sub + password + status=0 + deleted_at IS NULL。
        安全：统一错误提示"账号或密码错误"，防止用户枚举。
        """
        async with GetAsyncSession() as session:
            from sqlalchemy import select, text

            result = await session.execute(
                text("SELECT id, sub, username, nick_name, phone_number, email, "
                     "ou_id, ou_name, is_leader, password, status, deleted_at, "
                     "try_count, try_login_time, area, tenant_name, dep_name, dep_short_name "
                     "FROM sso_users WHERE sub = :sub"),
                {"sub": Sub},
            )
            row = result.fetchone()

            if not row:
                logger.warning(f"登录失败: 用户不存在 - sub={Sub}")
                raise LeauditException(StatusCodeEnum.HTTP_401_UNAUTHORIZED, "账号或密码错误")

            user = dict(row._mapping)

            if user.get("deleted_at") is not None:
                logger.warning(f"登录失败: 账号已删除 - sub={Sub}")
                raise LeauditException(StatusCodeEnum.HTTP_401_UNAUTHORIZED, "账号或密码错误")

            if user.get("status") != 0:
                logger.warning(f"登录失败: 账号已禁用 - sub={Sub}, status={user.get('status')}")
                raise LeauditException(StatusCodeEnum.HTTP_401_UNAUTHORIZED, "账号或密码错误")

            if user.get("password") != Password:
                logger.warning(f"登录失败: 密码错误 - sub={Sub}")
                raise LeauditException(StatusCodeEnum.HTTP_401_UNAUTHORIZED, "账号或密码错误")

            return await self._buildLoginResponse(user, session)

    async def OAuthLogin(self, Sub: str, Username: str | None, Nickname: str | None,
                         Email: str | None, PhoneNumber: str | None,
                         OuId: str | None, OuName: str | None,
                         IsLeader: bool | None, Area: str | None, ExpiresIn: int) -> LoginTokenVO:
        """OAuth 登录。验证 sub 是否存在，不存在则自动创建用户。"""
        async with GetAsyncSession() as session:
            from sqlalchemy import select, text
            from datetime import datetime, timezone

            result = await session.execute(
                text("SELECT id, sub, username, nick_name, phone_number, email, "
                     "ou_id, ou_name, is_leader, status, deleted_at, "
                     "area, tenant_name, dep_name, dep_short_name "
                     "FROM sso_users WHERE sub = :sub"),
                {"sub": Sub},
            )
            row = result.fetchone()

            if row:
                user = dict(row._mapping)
                if user.get("deleted_at") is not None or user.get("status") != 0:
                    raise LeauditException(StatusCodeEnum.HTTP_401_UNAUTHORIZED, "账号已被禁用或删除")
                # 更新最后登录信息
                await session.execute(
                    text("UPDATE sso_users SET username = :username, nick_name = :nick, "
                         "email = :email, phone_number = :phone, ou_id = :ou_id, "
                         "ou_name = :ou_name, is_leader = :is_leader, area = :area, "
                         "updated_at = :now WHERE id = :id"),
                    {"username": Username, "nick": Nickname, "email": Email,
                     "phone": PhoneNumber, "ou_id": OuId, "ou_name": OuName,
                     "is_leader": IsLeader, "area": Area,
                     "now": datetime.now(timezone.utc), "id": user["id"]},
                )
            else:
                # 自动创建用户
                await session.execute(
                    text("INSERT INTO sso_users (sub, username, nick_name, email, "
                         "phone_number, ou_id, ou_name, is_leader, area, status) "
                         "VALUES (:sub, :username, :nick, :email, :phone, :ou_id, "
                         ":ou_name, :is_leader, :area, 0)"),
                    {"sub": Sub, "username": Username, "nick": Nickname, "email": Email,
                     "phone": PhoneNumber, "ou_id": OuId, "ou_name": OuName,
                     "is_leader": IsLeader, "area": Area},
                )
                await session.commit()

                result = await session.execute(
                    text("SELECT id, sub, username, nick_name, phone_number, email, "
                         "ou_id, ou_name, is_leader, area, tenant_name, dep_name, dep_short_name "
                         "FROM sso_users WHERE sub = :sub"),
                    {"sub": Sub},
                )
                row = result.fetchone()
                user = dict(row._mapping) if row else {}

            if not user:
                raise LeauditException(StatusCodeEnum.HTTP_500_INTERNAL_SERVER_ERROR, "用户创建失败")

            return await self._buildLoginResponse(user, session)

    async def _buildLoginResponse(self, user: dict, session) -> LoginTokenVO:
        """组装登录响应：查询角色 → 签发 JWT → 返回 LoginTokenVO。"""
        from sqlalchemy import text

        # 查询用户角色
        roleResult = await session.execute(
            text("SELECT r.role_key FROM user_role ur "
                 "JOIN roles r ON ur.role_id = r.id "
                 "WHERE ur.user_id = :uid LIMIT 1"),
            {"uid": user["id"]},
        )
        roleRow = roleResult.fetchone()
        userRole = roleRow[0] if roleRow else "common"

        # 签发 JWT
        expiresIn = 3600  # 默认 1 小时
        tokens = JwtService.generate(
            userId=user["id"],
            username=user.get("username") or user.get("sub", ""),
            nickName=user.get("nick_name") or "",
            ouId=user.get("ou_id") or "",
            ouName=user.get("ou_name") or "",
            area=user.get("area"),
            userRole=userRole,
        )

        return LoginTokenVO(
            access_token=tokens["access_token"],
            token_type="Bearer",
            expires_in=expiresIn,
            issued_time=tokens.get("issued_time", ""),
            user_info={
                "user_id": user["id"],
                "sub": user.get("sub"),
                "username": user.get("username"),
                "nick_name": user.get("nick_name"),
                "email": user.get("email"),
                "phone_number": user.get("phone_number"),
                "ou_id": user.get("ou_id"),
                "ou_name": user.get("ou_name"),
                "is_leader": user.get("is_leader"),
                "area": user.get("area"),
                "role": userRole,
            },
        )