ajyy/backend/app/services/auth_service.py

"""认证业务：登录、当前用户、菜单树、操作日志"""
import uuid
from typing import Any, Dict, List, Optional

from sqlalchemy.orm import Session

from app.core.security import create_access_token, verify_password
from app.crud import menu as menu_crud
from app.crud import user as user_crud
from app.models.system import OperLog, SysMenu, SysUser


class AuthFailure(Exception):
    """登录失败等业务错误。"""

    def __init__(self, message: str = "用户名或密码错误"):
        self.message = message
        super().__init__(message)


def _uuid() -> str:
    return str(uuid.uuid4())


def write_oper_log(
    db: Session,
    *,
    module: str,
    action: str,
    user_id: Optional[str] = None,
    username: Optional[str] = None,
    detail: Optional[str] = None,
    ip: Optional[str] = None,
) -> None:
    db.add(
        OperLog(
            id=_uuid(),
            user_id=user_id,
            username=username,
            module=module,
            action=action,
            detail=detail,
            ip=ip,
        )
    )
    db.commit()


def authenticate_user(db: Session, username: str, password: str) -> SysUser:
    user = user_crud.get_user_by_username(db, username)
    if not user or user.status != "0":
        raise AuthFailure()
    if not verify_password(password, user.password_hash):
        raise AuthFailure()
    return user


def login(
    db: Session,
    username: str,
    password: str,
    ip: Optional[str] = None,
) -> Dict[str, str]:
    try:
        user = authenticate_user(db, username, password)
    except AuthFailure:
        write_oper_log(
            db,
            module="auth",
            action="login_failed",
            username=username,
            detail="invalid credentials",
            ip=ip,
        )
        raise

    token = create_access_token(sub=user.id, extra={"username": user.username})
    write_oper_log(
        db,
        module="auth",
        action="login",
        user_id=user.id,
        username=user.username,
        detail="login success",
        ip=ip,
    )
    return {"token": token, "tokenType": "Bearer"}


def current_user_payload(db: Session, user_id: str) -> Dict[str, Any]:
    user, role = user_crud.get_user_with_role(db, user_id)
    if not user or not role:
        raise AuthFailure("用户不存在或已禁用")
    return {
        "id": user.id,
        "username": user.username,
        "displayName": user.display_name,
        "roleCode": role.role_code,
        "roleName": role.role_name,
    }


def _menu_to_node(menu: SysMenu) -> Dict[str, Any]:
    return {
        "id": menu.id,
        "parentId": menu.parent_id,
        "menuCode": menu.menu_code,
        "menuName": menu.menu_name,
        "path": menu.path or "",
        "menuType": menu.menu_type,
        "sort": menu.sort,
        "children": [],
    }


def build_menu_tree(menus: List[SysMenu]) -> List[Dict[str, Any]]:
    node_list = [_menu_to_node(m) for m in menus]
    by_id = {n["id"]: n for n in node_list}
    roots: List[Dict[str, Any]] = []
    for n in node_list:
        parent_id = n["parentId"]
        if parent_id and parent_id != "0" and parent_id in by_id:
            by_id[parent_id]["children"].append(n)
        else:
            roots.append(n)
    return roots


def menus_payload(db: Session, role_id: str) -> Dict[str, Any]:
    page_menus = menu_crud.list_page_menus_for_role(db, role_id)
    perm_codes = menu_crud.list_menu_codes_for_role(db, role_id)
    return {
        "tree": build_menu_tree(page_menus),
        "permCodes": perm_codes,
    }