103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
"""为已通过校验的 SysUser 签发与 web 登录一致的 JWT 会话包。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.auth import (
|
|
create_access_token,
|
|
create_refresh_token,
|
|
store_refresh_token,
|
|
store_session,
|
|
)
|
|
from app.core.config import settings
|
|
from app.core.utils import generate_uuid
|
|
from app.models.models import SysUser
|
|
from app.services.web_auth_roles import (
|
|
get_user_role_names,
|
|
resolve_web_role_for_user,
|
|
user_has_active_roles,
|
|
)
|
|
|
|
|
|
def _avatar_url_field(user: SysUser) -> Optional[str]:
|
|
raw = getattr(user, "avatar", None)
|
|
if not raw:
|
|
return None
|
|
s = str(raw).strip()
|
|
return s if s else None
|
|
|
|
|
|
def build_user_info_payload(user: SysUser, web_role: str, role_names: list = None) -> dict:
|
|
phone = (getattr(user, "phone_no", None) or "").strip() or None
|
|
return {
|
|
"id": user.id,
|
|
"username": user.user_no,
|
|
"realName": user.nick_name,
|
|
"phone": phone,
|
|
"email": (getattr(user, "email", None) or "").strip() or None,
|
|
"hireDate": user.hire_date.strftime("%Y-%m-%d") if getattr(user, "hire_date", None) else None,
|
|
"probationMonths": getattr(user, "probation_months", None),
|
|
"lastWorkDay": user.last_work_day.strftime("%Y-%m-%d")
|
|
if getattr(user, "last_work_day", None)
|
|
else None,
|
|
"sex": getattr(user, "sex", None) or "0",
|
|
"remark": (getattr(user, "remark", None) or "").strip() or None,
|
|
"role": web_role,
|
|
"roleNames": role_names or [],
|
|
"regionId": user.dept_id,
|
|
"regionName": None,
|
|
"avatarUrl": _avatar_url_field(user),
|
|
}
|
|
|
|
|
|
def _jwt_user_payload(user: SysUser, web_role: str) -> dict:
|
|
return {
|
|
"sub": user.id,
|
|
"username": user.user_no,
|
|
"role": web_role,
|
|
"region_id": user.dept_id,
|
|
"real_name": user.nick_name,
|
|
}
|
|
|
|
|
|
async def issue_tokens_for_user(db: Session, user: SysUser) -> dict:
|
|
"""校验停用/角色后签发 token。失败抛 ValueError（中文 message）。"""
|
|
if user.status != "1":
|
|
raise ValueError("账号已被停用，请联系管理员")
|
|
if not user_has_active_roles(db, user.id):
|
|
raise ValueError("账号未分配角色，请联系管理员")
|
|
|
|
web_role = resolve_web_role_for_user(db, user.id)
|
|
role_names = get_user_role_names(db, user.id)
|
|
|
|
session_id = generate_uuid()
|
|
access_jti = generate_uuid()
|
|
refresh_jti = generate_uuid()
|
|
|
|
access_token = create_access_token(
|
|
data=_jwt_user_payload(user, web_role),
|
|
session_id=session_id,
|
|
token_jti=access_jti,
|
|
)
|
|
refresh_token = create_refresh_token(
|
|
data={
|
|
"sub": user.id,
|
|
"username": user.user_no,
|
|
"role": web_role,
|
|
},
|
|
session_id=session_id,
|
|
refresh_jti=refresh_jti,
|
|
)
|
|
|
|
await store_session(session_id=session_id, user_id=user.id)
|
|
await store_refresh_token(refresh_jti=refresh_jti, user_id=user.id)
|
|
|
|
return {
|
|
"token": access_token,
|
|
"refreshToken": refresh_token,
|
|
"expiresIn": settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
|
|
"userInfo": build_user_info_payload(user, web_role, role_names),
|
|
}
|