"""地磅模块认证适配:门户 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(), )