shanxia2.0/backend/app/core/db_auth.py

"""地磅模块认证适配：门户 JWT 用户 → db 业务上下文。"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Dict, Optional

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

from app.core.auth import get_current_user_by_token
from app.core.database import get_db
from app.core.menu_guard import ADMIN_ROLES, require_admin_or_super_admin

optional_oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="/api/public/login",
    auto_error=False,
)


@dataclass
class DbUserContext:
    """db Router/Service 使用的当前用户上下文。"""

    id: str
    name: str
    dept_id: str
    role: str
    tenant_no: str

    def to_tenant_user_dict(self) -> Dict[str, Any]:
        return {
            "user_id": self.id,
            "role": self.role,
            "tenant_no": self.tenant_no,
        }

    def to_user_info(self) -> Dict[str, Any]:
        """与门户 JWT / apply_data_scope 所需的 user_info 字段对齐。"""
        return {
            "user_id": self.id,
            "role": self.role,
            "tenant_no": self.tenant_no,
            "dept_id": self.dept_id,
            "region_id": self.dept_id,
        }


def get_db_current_user(
    user: dict = Depends(require_admin_or_super_admin),
) -> DbUserContext:
    return DbUserContext(
        id=(user.get("user_id") or "").strip(),
        name=(user.get("real_name") or user.get("username") or "").strip(),
        dept_id=(user.get("region_id") or "").strip(),
        role=(user.get("role") or "").strip(),
        tenant_no=(user.get("tenant_no") or "").strip(),
    )


def get_db_current_user_optional(
    token: Optional[str] = Depends(optional_oauth2_scheme),
    db: Session = Depends(get_db),
) -> DbUserContext | None:
    """有 Bearer 且为 admin/super_admin 时返回上下文，否则 None（供三方 H5 无登录调用）。"""
    if not token:
        return None
    try:
        user = get_current_user_by_token(token, db)
    except HTTPException:
        return None
    role = user.get("role")
    if role not in ADMIN_ROLES:
        return None
    return DbUserContext(
        id=(user.get("user_id") or "").strip(),
        name=(user.get("real_name") or user.get("username") or "").strip(),
        dept_id=(user.get("region_id") or "").strip(),
        role=(role or "").strip(),
        tenant_no=(user.get("tenant_no") or "").strip(),
    )