X-Financial/server/src/app/api/deps.py

from collections.abc import Generator
from dataclasses import dataclass
from typing import Annotated

from fastapi import Depends, Header, HTTPException, status
from sqlalchemy.orm import Session

from app.db.session import get_session_factory
from app.services.auth import AuthService
from app.services.auth_sessions import AuthSessionService


def get_db() -> Generator[Session, None, None]:
    db = get_session_factory()()
    try:
        yield db
    finally:
        db.close()


@dataclass(slots=True)
class CurrentUserContext:
    username: str
    name: str
    role_codes: list[str]
    is_admin: bool
    tenant_id: str = "default"
    department_name: str = ""
    cost_center: str = ""
    position: str = ""
    grade: str = ""
    employee_no: str = ""
    manager_name: str = ""
    employee_id: str = ""
    auth_session_id: str = ""


def get_current_user(
    db: Annotated[Session, Depends(get_db)],
    authorization: Annotated[
        str | None,
        Header(description="登录接口签发的 `Authorization: Bearer <token>`。"),
    ] = None,
) -> CurrentUserContext:
    return _authenticate_bearer_user(db, authorization)


def get_optional_current_user(
    db: Annotated[Session, Depends(get_db)],
    authorization: Annotated[
        str | None,
        Header(description="系统已初始化时用于管理员校验的 Bearer Token。"),
    ] = None,
) -> CurrentUserContext | None:
    if not str(authorization or "").strip():
        return None
    return _authenticate_bearer_user(db, authorization)


def _authenticate_bearer_user(db: Session, authorization: str | None) -> CurrentUserContext:
    access_token = _extract_bearer_token(authorization)
    auth_session = AuthSessionService(db).authenticate(access_token)
    if auth_session is None:
        raise _unauthorized("登录会话不存在、已过期或已退出，请重新登录。")

    user = AuthService(db).get_session_user(auth_session)
    if user is None:
        raise _unauthorized("当前登录用户不存在、已停用或权限已失效。")

    return CurrentUserContext(
        username=user.username,
        name=user.name,
        role_codes=[_normalize_role_code(item) for item in user.role_codes],
        is_admin=user.is_admin,
        tenant_id=user.tenant_id,
        department_name=user.department,
        cost_center=user.cost_center,
        position=user.position,
        grade=user.grade,
        employee_no=user.employee_no,
        manager_name=user.manager_name,
        employee_id=user.employee_id or "",
        auth_session_id=auth_session.id,
    )


def _extract_bearer_token(authorization: str | None) -> str:
    normalized = str(authorization or "").strip()
    scheme, separator, token = normalized.partition(" ")
    if not separator or scheme.casefold() != "bearer" or not token.strip():
        raise _unauthorized("请先登录后再访问。")
    return token.strip()


def _unauthorized(detail: str) -> HTTPException:
    return HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail=detail,
        headers={"WWW-Authenticate": "Bearer"},
    )


def _normalize_role_code(value: str | None) -> str:
    role_code = str(value or "").strip().lower()
    if role_code == "auditor":
        return "budget_monitor"
    return role_code


def _current_user_role_codes(current_user: CurrentUserContext) -> set[str]:
    return {
        _normalize_role_code(item)
        for item in current_user.role_codes
        if _normalize_role_code(item)
    }


def require_admin_user(
    current_user: Annotated[CurrentUserContext, Depends(get_current_user)],
) -> CurrentUserContext:
    if current_user.is_admin or "manager" in _current_user_role_codes(current_user):
        return current_user

    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="只有管理员可以上传、删除或修改知识库文件。",
    )


def require_platform_admin_user(
    current_user: Annotated[CurrentUserContext, Depends(get_current_user)],
) -> CurrentUserContext:
    if current_user.is_admin:
        return current_user

    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="只有 admin 管理员可以执行该操作。",
    )


def require_rule_editor_user(
    current_user: Annotated[CurrentUserContext, Depends(get_current_user)],
) -> CurrentUserContext:
    role_codes = _current_user_role_codes(current_user)
    if current_user.is_admin or "manager" in role_codes or "finance" in role_codes:
        return current_user

    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="只有财务人员或高级财务人员可以编辑规则草稿。",
    )


def require_rule_reviewer_user(
    current_user: Annotated[CurrentUserContext, Depends(get_current_user)],
) -> CurrentUserContext:
    role_codes = _current_user_role_codes(current_user)
    if current_user.is_admin or "manager" in role_codes:
        return current_user

    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="只有高级财务人员或 admin 管理员可以执行该操作。",
    )


def require_budget_viewer_user(
    current_user: Annotated[CurrentUserContext, Depends(get_current_user)],
) -> CurrentUserContext:
    role_codes = _current_user_role_codes(current_user)
    if current_user.is_admin or role_codes & {"budget_monitor", "executive"}:
        return current_user

    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="只有预算监控员或高级财务人员可以查看预算中心。",
    )


def require_budget_editor_user(
    current_user: Annotated[CurrentUserContext, Depends(get_current_user)],
) -> CurrentUserContext:
    role_codes = _current_user_role_codes(current_user)
    if current_user.is_admin or "executive" in role_codes:
        return current_user

    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="只有 admin 管理员或高级财务人员可以维护预算额度。",
    )


def is_budget_scope_limited_user(current_user: CurrentUserContext) -> bool:
    if current_user.is_admin:
        return False

    role_codes = _current_user_role_codes(current_user)
    return "budget_monitor" in role_codes and "executive" not in role_codes