55 lines
2.2 KiB
Python
55 lines
2.2 KiB
Python
import time
|
|
from typing import Annotated
|
|
|
|
from redis import Redis
|
|
from redis.asyncio import Redis as AsyncRedis
|
|
from fastapi import Depends, Header
|
|
|
|
from api.core.exception import ResponseException
|
|
from core.character.adventure import get_adv_info
|
|
from database.connect import get_async_redis, get_redis
|
|
from api.core.Auth import AlterState
|
|
import base64
|
|
import json
|
|
from api.core.Redis import get_redis_info
|
|
|
|
RedisDep = Annotated[Redis, Depends(get_redis)]
|
|
AsyncRedisDep = Annotated[AsyncRedis, Depends(get_async_redis)]
|
|
|
|
|
|
def alterToken(redis: RedisDep, alter_token: str | None = Header(None)):
|
|
if alter_token is not None:
|
|
try:
|
|
decoded_bytes = base64.urlsafe_b64decode(alter_token)
|
|
decoded_str = decoded_bytes.decode('utf-8')
|
|
alter = json.loads(decoded_str)
|
|
# 6小时过期
|
|
if (int(time.time() * 1000) - alter['time']) / (60 * 60 * 1000) > 6:
|
|
raise ResponseException('登录过期或无效Token，请刷新后重试')
|
|
alterData = get_redis_info(redis, f'dcalc:adventure:{alter["alter"]}', lambda: get_adv_info(alter['alter']))
|
|
if alterData is None:
|
|
raise ResponseException('登录过期或无效Token，请刷新后重试')
|
|
return AlterState(alterData['class'], alter_token, {}, alter.get('equVersion', '0'), alter['time'])
|
|
except (base64.binascii.Error, json.JSONDecodeError, KeyError, TypeError, UnicodeDecodeError):
|
|
raise ResponseException('登录过期或无效Token，请刷新后重试')
|
|
else:
|
|
raise ResponseException('登录过期或无效Token，请刷新后重试')
|
|
pass
|
|
|
|
|
|
AltersDep = Annotated[AlterState, Depends(alterToken)]
|
|
|
|
# def authorize(redis:RedisDep, access_token: Optional[str] = Header(None)):
|
|
# if access_token is not None:
|
|
# info = readToken(access_token, redis)
|
|
# return info
|
|
# else:
|
|
# raise ResponseException('登录过期或无效Token，请刷新后重试')
|
|
|
|
|
|
# def unauthorize(redis:RedisDep, access_token: Optional[str] = Header(None)):
|
|
# if access_token is not None:
|
|
# return deleteToken(access_token, redis)
|
|
|
|
# TokenDep = Annotated[AlterState, Depends(authorize)]
|