zkxw/backend/app/services/issue_user_tokens.py

"""为已通过校验的 SysUser 签发与 web 登录一致的 JWT 会话包。"""
from __future__ import annotations

from typing import List, Optional, Tuple

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.crud.user import list_post_ids_for_user
from app.models.models import SysDept, SysPost, 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 resolve_user_org_labels(db: Session, user: SysUser) -> Tuple[Optional[str], List[str]]:
    """解析用户所属部门名称与职位名称列表（保序）。"""
    tenant_no = (getattr(user, "tenant_no", None) or "").strip()
    dept_name: Optional[str] = None
    dept_id = (getattr(user, "dept_id", None) or "").strip()
    if dept_id:
        dept = (
            db.query(SysDept)
            .filter(
                SysDept.id == dept_id,
                SysDept.tenant_no == tenant_no,
                SysDept.del_flag == 0,
            )
            .first()
        )
        if dept and (dept.dept_name or "").strip():
            dept_name = str(dept.dept_name).strip()

    post_names: List[str] = []
    post_ids = list_post_ids_for_user(
        db, user.id, tenant_no=tenant_no
    )
    if post_ids:
        posts = (
            db.query(SysPost)
            .filter(
                SysPost.id.in_(post_ids),
                SysPost.tenant_no == tenant_no,
                SysPost.status == "1",
                SysPost.del_flag == 0,
            )
            .all()
        )
        name_map = {
            p.id: str(p.post_name).strip()
            for p in posts
            if p.id and (p.post_name or "").strip()
        }
        post_names = [name_map[pid] for pid in post_ids if pid in name_map]
    return dept_name, post_names


def build_user_info_payload(
    user: SysUser,
    web_role: str,
    role_names: list = None,
    *,
    db: Session = None,
) -> dict:
    phone = (getattr(user, "phone_no", None) or "").strip() or None
    dept_name: Optional[str] = None
    post_names: List[str] = []
    if db is not None:
        dept_name, post_names = resolve_user_org_labels(db, user)
    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": dept_name,
        "deptName": dept_name,
        "postNames": post_names,
        "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("账号已被停用，请联系管理员")
    tenant_no = (getattr(user, "tenant_no", None) or "").strip() or None
    if not user_has_active_roles(db, user.id, tenant_no=tenant_no):
        raise ValueError("账号未分配角色，请联系管理员")

    web_role = resolve_web_role_for_user(
        db, user.id, tenant_no=tenant_no
    )
    role_names = get_user_role_names(db, user.id, tenant_no=tenant_no)

    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, db=db),
    }