From 77162e1f0aa56506b2c0486eb9a6d81a4b73f005 Mon Sep 17 00:00:00 2001
From: martsforever <martsforever@aliyun.com>
Date: Fri, 22 Aug 2025 15:17:51 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E7=94=A8=E6=88=B7=E7=99=BB=E5=BD=95?=
 =?UTF-8?q?=E6=B3=A8=E5=86=8C=E7=9B=B8=E5=85=B3=E6=8E=A5=E5=8F=A3=E8=B0=83?=
 =?UTF-8?q?=E6=95=B4=EF=BC=8C=E7=99=BB=E5=BD=95=E6=88=90=E5=8A=9F=E4=B9=8B?=
 =?UTF-8?q?=E5=90=8E=E8=BF=94=E5=9B=9Eaccess=5Ftoken=E4=BB=A5=E5=8F=8Arefr?=
 =?UTF-8?q?esh=5Ftoken?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/controller/add_user_route.py | 30 +++++++++++++++++++---------
 app/utils/CrpyUtils.py           | 34 ++++++++++++++++++++++----------
 2 files changed, 45 insertions(+), 19 deletions(-)

diff --git a/app/controller/add_user_route.py b/app/controller/add_user_route.py
index d70f205..ccae303 100644
--- a/app/controller/add_user_route.py
+++ b/app/controller/add_user_route.py
@@ -84,8 +84,12 @@ def add_user_route(app: FastAPI):
 
     public_user = PublicUser(**user.model_dump())
 
-    active_user_token = CryptUtils.create_access_token(public_user.username, expires_delta=timedelta(days=365 * 3))
-    active_url = f"{env.server_domain}:{env.server_port}/verify?token={active_user_token}"
+    verify_user_token = CryptUtils.create_token(
+      username=public_user.username,
+      type="verify",
+      expires_delta=timedelta(days=365 * 3),
+    )
+    active_url = f"{env.server_domain}:{env.server_port}/verify?token={verify_user_token}"
 
     return {
       "result": public_user,
@@ -95,9 +99,10 @@ def add_user_route(app: FastAPI):
   # 验证用户账号接口
   @app.get("/verify")
   async def _verify(token: str, session: AsyncSessionDep):
-    username = CryptUtils.get_username_from_token(token)
+    token_info = CryptUtils.get_token_info(token)
+    username = token_info.get('username')
 
-    if not username:
+    if not username or token_info.get('type') != 'verify':
       return {"result": None, "error": "token无效或者已经过期"}
 
     query = select(UserModel).where(UserModel.username == username)
@@ -133,14 +138,21 @@ def add_user_route(app: FastAPI):
         headers={"WWW-Authenticate": "Bearer"},
       )
 
-    token = Token(
-      token=CryptUtils.create_access_token(user.username),
-      token_type="Bearer",
+    access_token = CryptUtils.create_token(
+      username=user.username,
+      type="access",
+      expires_delta=timedelta(minutes=env.jwt_access_token_expire_minutes),
+    )
+    refresh_token = CryptUtils.create_token(
+      username=user.username,
+      type="refresh",
+      expires_delta=timedelta(minutes=env.jwt_refresh_token_expire_minutes),
     )
 
     return {
       "result": user,
-      "token": token,
+      "access_token": access_token,
+      "refresh_token": refresh_token,
     }
 
   # 获取用户信息接口
@@ -187,7 +199,7 @@ unauthorized_exception = HTTPException(
 # 获取当前用户信息，通过注入的token来获取当前用户信息，如果token有效则返回用户信息，无效则抛出异常
 async def get_current_user(session: AsyncSessionDep, token: str = Depends(oauth2_scheme)):
   try:
-    username = CryptUtils.get_username_from_token(token)
+    username = CryptUtils.get_token_info(token).get('username')
     if not username:
       raise unauthorized_exception
   except InvalidTokenError:
diff --git a/app/utils/CrpyUtils.py b/app/utils/CrpyUtils.py
index 8703426..61d6c3f 100644
--- a/app/utils/CrpyUtils.py
+++ b/app/utils/CrpyUtils.py
@@ -1,4 +1,5 @@
 from datetime import timedelta, datetime, timezone
+from typing import TypedDict, Literal, TypeAlias
 
 import jwt
 from passlib.context import CryptContext
@@ -7,6 +8,17 @@ from app.config.env import env
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
+# token的类型，access用于接口认证，refresh用于刷新access token，verify用于激活用户账号
+AccessTokenType: TypeAlias = Literal["access", "refresh", "verify"]
+
+
+class TokenInfo(TypedDict):
+  # 用户名信息
+  username: str
+  # token过期时间
+  exp: datetime
+  type: AccessTokenType
+
 
 class CryptUtils:
   @staticmethod
@@ -18,17 +30,19 @@ class CryptUtils:
     return pwd_context.verify(plain_password, hashed_password)
 
   @staticmethod
-  def create_access_token(username: str, expires_delta: timedelta | None = None):
-    data: dict = {"sub": username}
-    if expires_delta:
-      expire = datetime.now(timezone.utc) + expires_delta
-    else:
-      expire = datetime.now(timezone.utc) + timedelta(minutes=env.jwt_access_token_expire_minutes)
-    data.update({'exp': expire})
+  def create_token(
+    username: str,
+    type: AccessTokenType,
+    expires_delta: timedelta
+  ):
+    data: TokenInfo = {
+      "username": username,
+      "type": type,
+      "exp": datetime.now(timezone.utc) + expires_delta
+    }
     return jwt.encode(data, env.jwt_secret_key, env.jwt_algorithm)
 
   @staticmethod
-  def get_username_from_token(token: str):
+  def get_token_info(token: str) -> TokenInfo:
     data = jwt.decode(token, env.jwt_secret_key, algorithms=[env.jwt_algorithm])
-    username = data.get("sub")
-    return username
+    return data