python-fastapi-mm-zhcs-yj-api/routers/prod_api/sso.py

# -*- coding: utf-8 -*-

from fastapi import APIRouter, Depends, Request, Header, Form, Body
from fastapi.responses import FileResponse, StreamingResponse
from sqlalchemy.orm import Session
from fastapi.responses import JSONResponse
from database import get_db
from utils import *
from utils.vcode import *
from utils.redis_util import *
import base64
from common.const import *
from io import BytesIO
from utils.StripTagsHTMLParser import *
from common import security
from datetime import timedelta
from common.security import valid_access_token
from common.auth_user import *
from common import YzyApi
from models import *
from urllib.parse import quote
import requests
import jwt
import traceback
from common.enc import mpfun

router = APIRouter()


# 提供给数科使用的单点登录token
@router.get('/token/create')
def sso_token(request: Request,
    user_id: int = Depends(valid_access_token),
    db: Session = Depends(get_db)):

    sso_token_expires = timedelta(seconds = 3600 * 24)
    sso_token = security.create_access_token(
        data={"sub": user_id}, expires_delta = sso_token_expires
    )

    print('sso_token:', sso_token)

    return {
        "code": 200,
        "msg": "操作成功",
        "data": {
            "sso_token": sso_token
        }
    }

# token校验
@router.get('/token/valid')
def sso_token(request: Request,
    sso_token: str,
    db: Session = Depends(get_db)):

    try:
        payload = jwt.decode(sso_token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        # print(payload,payload.get("sub"))
        user_id: str = payload.get("sub")

        row = db.query(SysUser).filter(SysUser.user_id == int(user_id)).first()
        if row is None:
            return {
                "code": 500,
                "msg": "token异常"
            }
        
        # 角色信息
        roles = []
        role_ids = db.query(SysUserRole).filter(SysUserRole.user_id == int(user_id)).all()
        for role in role_ids:
            role_info = db.query(SysRole).filter(SysRole.role_id == role.role_id).first()
            roles.append(
                {
                    "roleId": role_info.role_id,
                    "roleName": role_info.role_name,
                    "roleKey": role_info.role_key
                }
            )

        role_keys = [
            n['roleKey']
            for n in roles
        ]
        
        data = {
            "userId": row.user_id,
            "userName": mpfun.dec_data(row.user_name),
            "nickName": row.nick_name,
            "roles": role_keys
        }
    
        return {
            "code": 200,
            "msg": "操作成功",
            "data": data
        }
        
    except Exception:
        traceback.print_exc()
        return {
            "code": 500,
            "msg": "token异常"
        }