一、项目概述

1. 需求分析（求职者登录模块）

• 三种登录方式：账号密码登录、手机号 + 短信验证码登录、钉钉第三方登录（未绑定手机号则跳转绑定页）
• 配套能力：用户注册、获取当前登录人信息接口
• 认证方案：业界标准双 Token 机制
  • access_token：短期有效（配置 2 天），用于访问所有需要登录的接口
  • refresh_token：长期有效（配置 7 天），专门用来无感刷新access_token，避免频繁登录
• 前后端交互规范：
  • 前端请求受保护接口，请求头携带Authorization: Bearer {access_token}
  • access_token过期后，前端使用refresh_token调用刷新接口，获取全新双 Token
  • refresh_token失效则清空本地缓存，强制跳转登录页
• 配套资源：阿里云 OSS 存储头像，统一环境配置管理

2. 数据库设计（求职者表 t_job_seeker）

二、核心流程详解

1. 登录流程

1. 用户输入凭证：用户在前端页面输入账号密码、手机号验证码，或点击钉钉授权登录
2. 后端校验：后端接收请求，校验用户信息（密码、验证码或第三方授权码）
3. 生成双 Token：校验通过后，后端调用 JWT 工具类生成access_token（短期）和refresh_token（长期）
4. 返回前端：将双 Token 和用户基本信息返回给前端
5. 前端存储：前端将 Token 存储到本地（如 localStorage 或 Vuex/Pinia），并跳转至个人中心首页

2. 鉴权请求流程

1. 发起请求：前端访问需要登录的接口（如/me获取个人信息）
2. 携带 Token：在请求头中携带Authorization: Bearer {access_token}
3. 后端校验：后端 JWT 工具类校验 Token 的有效性、类型和过期时间
4. 解析用户：校验通过后解析出用户 ID，查询数据库获取用户完整信息
5. 返回数据：将用户信息返回给前端，完成鉴权

3. Token 刷新流程

1. Token 过期：access_token过期后，后端返回 401 鉴权失败状态码
2. 前端检测：前端拦截器检测到 401 状态，从本地读取refresh_token
3. 调用刷新接口：前端使用refresh_token调用专门的 Token 刷新接口
4. 后端校验刷新：后端校验refresh_token的有效性，校验通过后生成全新的access_token
5. 返回新 Token：将新的access_token返回给前端
6. 更新本地存储：前端更新本地存储的access_token
7. 重试请求：前端使用新的access_token重新发起之前失败的请求，用户无感知完成刷新
8. 刷新失败：如果refresh_token也失效，则清空本地缓存，强制跳转登录页

4. 文件存储流程（头像上传）

1. 前端上传：求职者在个人中心选择头像文件并上传
2. 后端直传 OSS：后端接收文件后，直接上传至阿里云 OSS
3. 获取访问链接：OSS 返回文件的访问链接（URL）
4. 更新数据库：后端将头像链接存入t_job_seeker.avatar字段
5. 前端展示：前端个人中心读取该链接并展示用户头像

三、后端核心开发（FastAPI + Tortoise-ORM + JWT）

1. 依赖安装

# 基础 web 框架 pip install fastapi uvicorn JWT 双 token 工具 pip install python-jose passlib bcrypt 数据库 ORM pip install tortoise-orm aiomysql 环境配置管理 pip install pydantic-settings 阿里云 OSS pip install oss2

2. 全局配置 config.py（Pydantic 多环境配置）

from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): # 项目基础 app_env: str = "dev" debug: bool = True # JWT 配置 SECRET_KEY: str = "LTAI5t65xxxx阿里云 OSS 密钥" ALGORITHM: str = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES: int = 2880 # 2 天 REFRESH_TOKEN_EXPIRE_DAYS: int = 7 # 7 天 阿里云 OSS 配置 OSS_ACCESS_KEY_ID: str = "LTAI5t65khxxxx" OSS_ACCESS_KEY_SECRET: str = "l5dKxxxxxxxxx" OSS_ENDPOINT: str = "oss-cn-beijing.aliyuncs.com" OSS_BUCKET: str = "boss-job-file" model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8") settings = Settings()

3. JWT 工具类 app/utils/jwt_util.py（双 Token 生成、校验、刷新）

from datetime import datetime, timedelta from typing import Optional, Tuple from jose import jwt, JWTError from app.config import settings def create_tokens(user_id: str, username: str) -> Tuple[str, str]: """ 同时生成 access_token、refresh_token :param user_id: 用户 ID :param username: 手机号（登录标识） :return: (access_token, refresh_token) """ # 公共载荷 common_payload = { "user_id": user_id, "username": username } # 1. 短期访问 token access_payload = common_payload.copy() access_payload["type"] = "access" access_expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) access_payload["exp"] = access_expire access_token = jwt.encode(access_payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM) 2. 长期刷新 token refresh_payload = common_payload.copy() refresh_payload["type"] = "refresh" refresh_expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS) refresh_payload["exp"] = refresh_expire refresh_token = jwt.encode(refresh_payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return access_token, refresh_token def verify_token(token: str, token_type: str = "access") -> Optional[dict]: """ 校验 token 合法性、类型、有效期 :param token: jwt 字符串 :param token_type: access / refresh :return: 载荷字典，失败返回 None """ try: payload = jwt.decode( token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM], options={"verify_exp": True} ) 校验 token 类型 if payload.get("type") != token_type: return None 校验必填字段 if not all(k in payload for k in ["user_id", "username"]): return None return payload except JWTError: return None def refresh_access_token(refresh_token: str) -> Optional[str]: """ 使用 refresh_token 刷新全新 access_token """ payload = verify_token(refresh_token, token_type="refresh") if not payload: return None 生成新 access_token user_id = payload["user_id"] username = payload["username"] access_token, _ = create_tokens(user_id, username) return access_token

4. 阿里云 OSS 工具类 app/utils/oss_util.py（头像上传、删除、权限控制）

import oss2 import uuid from datetime import datetime, timedelta from typing import Optional, Tuple from fastapi import HTTPException, UploadFile from app.config import settings class OSSUtil: """阿里云 OSS 文件上传、删除、访问权限控制工具类""" def __init__(self): """初始化 OSS 客户端""" auth = oss2.Auth( settings.OSS_ACCESS_KEY_ID, settings.OSS_ACCESS_KEY_SECRET ) self.bucket = oss2.Bucket( auth, settings.OSS_ENDPOINT, settings.OSS_BUCKET ) def generate_object_name(self, user_id: int, original_filename: str) -> str: """ 生成唯一的 OSS 对象名称 :param user_id: 用户 ID :param original_filename: 原始文件名 :return: 对象名称，格式：avatar/{user_id}/{timestamp_uuid}.{ext} """ # 提取文件扩展名 ext = original_filename.split('.')[-1] if '.' in original_filename else 'jpg' # 生成唯一文件名：时间戳 + UUID timestamp = datetime.now().strftime("%Y%m%d%H%M%S") unique_id = str(uuid.uuid4())[:8] # 构建对象路径 object_name = f"avatar/{user_id}/{timestamp}_{unique_id}.{ext}" return object_name async def upload_avatar(self, user_id: int, file: UploadFile) -> Tuple[str, str]: """ 上传用户头像到 OSS :param user_id: 用户 ID :param file: FastAPI UploadFile 对象 :return: (对象名称, 访问URL) """ try: # 1. 验证文件类型 allowed_types = ['image/jpeg', 'image/png', 'image/gif', 'image/webp'] if file.content_type not in allowed_types: raise HTTPException( status_code=400, detail=f"不支持的文件类型，仅支持：{', '.join(allowed_types)}" ) # 2. 验证文件大小（限制 5MB） max_size = 5 * 1024 * 1024 # 5MB file_content = await file.read() if len(file_content) > max_size: raise HTTPException( status_code=400, detail="文件大小超过 5MB 限制" ) # 3. 生成唯一对象名称 object_name = self.generate_object_name(user_id, file.filename) # 4. 上传到 OSS result = self.bucket.put_object(object_name, file_content) if result.status != 200: raise HTTPException( status_code=500, detail="文件上传到 OSS 失败" ) # 5. 生成访问 URL（带签名，30分钟有效） access_url = self.generate_signed_url(object_name, expires=1800) return object_name, access_url except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException( status_code=500, detail=f"上传过程中发生错误：{str(e)}" ) def generate_signed_url(self, object_name: str, expires: int = 1800) -> str: """ 生成带签名的临时访问 URL :param object_name: OSS 对象名称 :param expires: 过期时间（秒），默认30分钟 :return: 带签名的访问 URL """ try: # 生成带签名的 URL signed_url = self.bucket.sign_url( 'GET', object_name, expires, slash_safe=True ) return signed_url except Exception as e: raise HTTPException( status_code=500, detail=f"生成签名 URL 失败：{str(e)}" ) def delete_avatar(self, object_name: str) -> bool: """ 删除 OSS 上的头像文件 :param object_name: OSS 对象名称 :return: 是否删除成功 """ try: # 检查文件是否存在 if not self.bucket.object_exists(object_name): return True # 文件不存在，视为删除成功 # 删除文件 result = self.bucket.delete_object(object_name) return result.status == 204 except Exception as e: raise HTTPException( status_code=500, detail=f"删除文件失败：{str(e)}" ) def set_object_acl(self, object_name: str, acl: str = 'private') -> bool: """ 设置 OSS 对象的访问权限 :param object_name: OSS 对象名称 :param acl: 权限类型，可选值：private（私有）、public-read（公共读）、public-read-write（公共读写） :return: 是否设置成功 """ try: # 验证权限类型 valid_acls = ['private', 'public-read', 'public-read-write'] if acl not in valid_acls: raise ValueError(f"无效的 ACL 类型，可选值：{', '.join(valid_acls)}") # 设置对象 ACL self.bucket.put_object_acl(object_name, acl) return True except Exception as e: raise HTTPException( status_code=500, detail=f"设置文件权限失败：{str(e)}" ) def get_object_info(self, object_name: str) -> Optional[dict]: """ 获取 OSS 对象信息 :param object_name: OSS 对象名称 :return: 对象信息字典，包含大小、类型、最后修改时间等 """ try: # 获取对象元数据 meta = self.bucket.get_object_meta(object_name) return { 'size': meta.content_length, 'type': meta.content_type, 'last_modified': meta.last_modified, 'etag': meta.etag, 'object_name': object_name } except oss2.exceptions.NoSuchKey: return None except Exception as e: raise HTTPException( status_code=500, detail=f"获取文件信息失败：{str(e)}" ) 全局 OSS 工具实例 oss_util = OSSUtil()

5. 头像管理 Service（整合 OSS 与数据库）

from fastapi import HTTPException, UploadFile from app.models.job_seeker import JobSeeker from app.utils.oss_util import oss_util import asyncio class AvatarService: """头像管理服务，整合 OSS 操作与数据库更新""" @staticmethod async def upload_avatar(user_id: int, file: UploadFile) -&gt; dict: """ 上传头像并更新数据库 :param user_id: 用户 ID :param file: 头像文件 :return: 上传结果信息 """ try: # 1. 查询用户信息 seeker = await JobSeeker.get_or_none(id=user_id) if not seeker: raise HTTPException(status_code=404, detail="用户不存在") # 2. 如果用户已有头像，先删除旧头像 old_avatar = seeker.avatar if old_avatar: # 从 avatar URL 中提取 object_name # 假设 avatar 字段存储的是完整的 OSS 对象名称或 URL old_object_name = old_avatar.split('/')[-1] if '/' in old_avatar else old_avatar try: oss_util.delete_avatar(f"avatar/{user_id}/{old_object_name}") except: # 删除旧头像失败不影响新头像上传，记录日志即可 pass # 3. 上传新头像到 OSS object_name, access_url = await oss_util.upload_avatar(user_id, file) # 4. 设置文件权限为私有（默认） oss_util.set_object_acl(object_name, 'private') # 5. 更新数据库中的头像链接 # 存储完整的带签名 URL（30分钟有效） seeker.avatar = access_url await seeker.save() # 6. 返回上传结果 return { "success": True, "message": "头像上传成功", "avatar_url": access_url, "object_name": object_name, "expires_in": 1800 # 30分钟 } except HTTPException: raise except Exception as e: raise HTTPException( status_code=500, detail=f"头像上传失败：{str(e)}" ) @staticmethod async def get_avatar_url(user_id: int) -&gt; Optional[str]: """ 获取用户头像访问 URL（带签名） :param user_id: 用户 ID :return: 带签名的头像 URL，如果用户没有头像则返回 None """ try: # 查询用户信息 seeker = await JobSeeker.get_or_none(id=user_id) if not seeker or not seeker.avatar: return None # 从数据库中的 avatar 字段提取 object_name # 这里假设 avatar 字段存储的是完整的 URL，需要提取 object_name avatar_url = seeker.avatar if 'avatar/' in avatar_url: # 从 URL 中提取 object_name parts = avatar_url.split('avatar/') if len(parts) &amp;gt; 1: object_name = f"avatar/{parts[1].split('?')[0]}" else: object_name = seeker.avatar else: object_name = seeker.avatar # 生成新的带签名 URL（30分钟有效） signed_url = oss_util.generate_signed_url(object_name, expires=1800) # 更新数据库中的 URL（可选，可以定期清理过期 URL） seeker.avatar = signed_url await seeker.save() return signed_url except Exception as e: # 如果生成签名 URL 失败，返回原始 URL return seeker.avatar if seeker and seeker.avatar else None @staticmethod async def delete_avatar(user_id: int) -&gt; dict: """ 删除用户头像 :param user_id: 用户 ID :return: 删除结果信息 """ try: # 查询用户信息 seeker = await JobSeeker.get_or_none(id=user_id) if not seeker: raise HTTPException(status_code=404, detail="用户不存在") if not seeker.avatar: return { "success": True, "message": "用户没有头像，无需删除" } # 从 avatar URL 中提取 object_name avatar_url = seeker.avatar if 'avatar/' in avatar_url: parts = avatar_url.split('avatar/') if len(parts) &amp;gt; 1: object_name = f"avatar/{parts[1].split('?')[0]}" else: object_name = seeker.avatar else: object_name = seeker.avatar # 删除 OSS 上的文件 delete_result = oss_util.delete_avatar(object_name) if delete_result: # 清空数据库中的头像链接 seeker.avatar = None await seeker.save() return { "success": True, "message": "头像删除成功" } else: return { "success": False, "message": "头像删除失败" } except HTTPException: raise except Exception as e: raise HTTPException( status_code=500, detail=f"头像删除失败：{str(e)}" ) @staticmethod async def refresh_avatar_url(user_id: int) -&gt; Optional[str]: """ 刷新用户头像 URL 签名（防止 URL 过期） :param user_id: 用户 ID :return: 新的带签名 URL """ return await AvatarService.get_avatar_url(user_id)</code></pre>

4. 求职者业务 Service（登录、获取个人信息）

from fastapi import HTTPException from app.models.job_seeker import JobSeeker from app.utils.jwt_util import create_tokens from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") class JobSeekerService: @staticmethod async def login_by_mobile_pwd(mobile: str, password: str): """手机号+密码登录，返回用户信息+双 token""" # 查询用户 seeker = await JobSeeker.get_or_none(mobile=mobile) if not seeker or not pwd_context.verify(password, seeker.password): raise HTTPException(status_code=401, detail="手机号或密码错误") # 生成双 token access_token, refresh_token = create_tokens(str(seeker.id), seeker.mobile) return { "jobseeker": seeker, "access_token": access_token, "refresh_token": refresh_token } @staticmethod async def get_current_seeker(authorization: str): """解析 header token，获取当前登录求职者""" if not authorization or not authorization.startswith("Bearer "): raise HTTPException(status_code=401, detail="认证信息无效") token = authorization.split(" ")[1] payload = verify_token(token, token_type="access") if not payload: raise HTTPException(status_code=401, detail="Token 无效或已过期") seeker = await JobSeeker.get_or_none(id=int(payload["user_id"])) if not seeker: raise HTTPException(status_code=401, detail="用户不存在") return seeker</code></pre>