# ============================================================================= # 企微IT智能服务台 — Portal 统一入口 API # ============================================================================= # 说明：统一入口（Portal）相关接口 # 包含： # 1. 获取当前用户角色信息 # 2. 切换当前角色 # 3. 获取角色对应的入口 URL # 所有接口需要有效的 Bearer Token # ============================================================================= import json import logging from typing import Optional from fastapi import APIRouter, Depends from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.dependencies import get_current_user, UserInfo from app.config import settings from app.database import get_db from app.models.role import Role from app.models.user_role import UserRole from app.schemas.role import ( PortalUserInfo, RoleResponse, SwitchRoleRequest, SwitchRoleResponse, ) from app.services.token_service import TokenService from app.utils.response import AppException, success_response logger = logging.getLogger(__name__) # HTTP Bearer 认证方案 security = HTTPBearer() # 创建路由器 router = APIRouter(prefix="/portal") # -------------------------------------------------------------------------- # 获取当前用户角色信息 # -------------------------------------------------------------------------- @router.get("/roles") async def get_user_roles( current_user: UserInfo = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """获取当前用户的角色信息。 返回用户的基本信息和角色列表，用于路由选择页展示。 Args: current_user: 当前用户（通过认证依赖注入） db: 数据库会话 Returns: Dict: 统一响应格式，包含用户信息和角色列表 """ # 查询用户拥有的角色 stmt = ( select(Role, UserRole) .join(UserRole, Role.id == UserRole.role_id) .where(UserRole.employee_id == current_user.employee_id) .where( # 过滤已过期的角色 (UserRole.expires_at.is_(None)) | (UserRole.expires_at > func.now()) ) ) result = await db.execute(stmt) role_rows = result.all() # 构建角色列表 roles = [] for role, user_role in role_rows: roles.append( RoleResponse( id=role.id, name=role.name, display_name=role.display_name, description=role.description, permissions=role.permissions or [], is_default=role.is_default, created_at=role.created_at, updated_at=role.updated_at, ) ) # 如果用户没有任何角色，添加默认的 user 角色 if not roles: # 查询 user 角色 user_role_stmt = select(Role).where(Role.name == "user") user_role_result = await db.execute(user_role_stmt) user_role = user_role_result.scalars().first() if user_role: roles.append( RoleResponse( id=user_role.id, name=user_role.name, display_name=user_role.display_name, description=user_role.description, permissions=user_role.permissions or [], is_default=user_role.is_default, created_at=user_role.created_at, updated_at=user_role.updated_at, ) ) # 构建响应 user_info = PortalUserInfo( employee_id=current_user.employee_id, name=current_user.name, department=current_user.department, avatar=current_user.avatar, roles=roles, current_role=current_user.current_role, ) return success_response(data=user_info.model_dump()) # -------------------------------------------------------------------------- # 切换当前角色 # -------------------------------------------------------------------------- @router.post("/switch-role") async def switch_role( body: SwitchRoleRequest, current_user: UserInfo = Depends(get_current_user), db: AsyncSession = Depends(get_db), credentials: HTTPAuthorizationCredentials = Depends(security), ): """切换当前角色。 更新 Redis Token 中的 current_role 字段，返回目标角色的入口 URL。 Args: body: 切换角色请求 current_user: 当前用户（通过认证依赖注入） db: 数据库会话 Returns: Dict: 统一响应格式，包含切换后的角色和重定向 URL """ # 验证用户是否有目标角色 stmt = ( select(Role) .join(UserRole, Role.id == UserRole.role_id) .where(UserRole.employee_id == current_user.employee_id) .where(Role.name == body.new_role) ) result = await db.execute(stmt) target_role = result.scalars().first() if not target_role: raise AppException(4003, f"没有 {body.new_role} 角色权限") # 更新 Redis Token 中的 current_role from app.dependencies import get_redis redis_client = await get_redis() token_service = TokenService(redis_client) # 从请求头获取 token token = credentials.credentials switch_success = await token_service.switch_role(token, body.new_role) if not switch_success: raise AppException(4003, "角色切换失败") # 获取目标角色的入口 URL redirect_url = _get_role_url(body.new_role) logger.info(f"用户 {current_user.employee_id} 切换角色到 {body.new_role}") return success_response( data=SwitchRoleResponse( current_role=body.new_role, redirect_url=redirect_url, ).model_dump() ) # -------------------------------------------------------------------------- # 获取角色对应的入口 URL # -------------------------------------------------------------------------- @router.get("/entry/{role_name}") async def get_role_entry( role_name: str, current_user: UserInfo = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """获取角色对应的入口 URL。 Args: role_name: 角色标识 current_user: 当前用户（通过认证依赖注入） db: 数据库会话 Returns: Dict: 统一响应格式，包含角色信息和入口 URL """ # 验证用户是否有目标角色 stmt = ( select(Role) .join(UserRole, Role.id == UserRole.role_id) .where(UserRole.employee_id == current_user.employee_id) .where(Role.name == role_name) ) result = await db.execute(stmt) target_role = result.scalars().first() if not target_role: raise AppException(4003, f"没有 {role_name} 角色权限") # 获取入口 URL redirect_url = _get_role_url(role_name) return success_response( data={ "role": role_name, "url": redirect_url, "display_name": target_role.display_name, } ) # -------------------------------------------------------------------------- # 辅助函数：获取角色对应的 URL # -------------------------------------------------------------------------- def _get_role_url(role_name: str) -> str: """获取角色对应的前端 URL。 Args: role_name: 角色标识 Returns: str: 前端 URL """ role_urls = { "user": "/itdesk/", "agent": "/itagent/", "admin": "/itadmin/", } return role_urls.get(role_name, "/itdesk/")