62 lines
2.4 KiB
Python
62 lines
2.4 KiB
Python
"""认证业务：登录、建号、改密。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import settings
|
|
from app.core.exceptions import AuthError, BusinessError, ConflictError
|
|
from app.core.security import create_access_token, hash_password, verify_password
|
|
from app.dao.account_dao import AccountDao
|
|
from app.model import Account, Role
|
|
from app.schema.auth_schema import AccountCreate, PasswordChange
|
|
|
|
|
|
class AuthService:
|
|
@classmethod
|
|
def login(cls, db: Session, username: str, password: str) -> tuple[Account, str, int]:
|
|
account = AccountDao.get_by_username(db, username)
|
|
if account is None or not verify_password(password, account.hashed_password):
|
|
# 不区分"用户不存在"和"密码错误"，避免账号枚举
|
|
raise AuthError("用户名或密码不正确")
|
|
if not account.is_active:
|
|
raise AuthError("该账号已被停用，请联系管理员")
|
|
|
|
account.last_login_at = datetime.now()
|
|
db.flush()
|
|
|
|
expires_seconds = settings.JWT_EXPIRE_MINUTES * 60
|
|
token = create_access_token(
|
|
{"sub": str(account.id), "username": account.username, "role": account.role}
|
|
)
|
|
return account, token, expires_seconds
|
|
|
|
@classmethod
|
|
def create_account(cls, db: Session, payload: AccountCreate) -> Account:
|
|
if AccountDao.get_by_username(db, payload.username, with_deleted=True):
|
|
raise ConflictError(f"登录名「{payload.username}」已被占用")
|
|
if payload.role not in {r.value for r in Role}:
|
|
raise BusinessError(f"角色只能是 {'/'.join(r.value for r in Role)}")
|
|
|
|
account = Account(
|
|
username=payload.username,
|
|
hashed_password=hash_password(payload.password),
|
|
real_name=payload.real_name,
|
|
role=payload.role,
|
|
advisor_id=payload.advisor_id,
|
|
)
|
|
db.add(account)
|
|
db.flush()
|
|
return account
|
|
|
|
@classmethod
|
|
def change_password(cls, db: Session, account: Account, payload: PasswordChange) -> None:
|
|
if not verify_password(payload.old_password, account.hashed_password):
|
|
raise BusinessError("原密码不正确")
|
|
if payload.old_password == payload.new_password:
|
|
raise BusinessError("新密码不能与原密码相同")
|
|
account.hashed_password = hash_password(payload.new_password)
|
|
db.flush()
|