企业官网建设流程全解析

求职者端--手机号密码登录

在之前项目中，都做的是最基本的账密登录，只进行了账号密码的比对，这显然是不符合实际的。在本次加入了token，更符合现实情况。

首先，将生成token和解析token的方法进行封装，本方案采用 Access/Refresh 长短双 Token 架构，登录时生成两类载荷携带type标识的 JWT 令牌，access 短期有效用于业务鉴权，refresh 长期有效仅负责刷新凭证；封装通用校验方法强制区分令牌类型，避免 Token 越权调用，access 过期后依靠合法 refresh 令牌生成新访问令牌。该机制缩短有效凭证暴露窗口提升安全性，但缺少 Refresh 黑名单与令牌轮换逻辑，登出、改密场景下旧刷新令牌仍可正常使用，生产环境建议搭配 Redis 黑名单优化。

# 1. 生成access_token和refresh_token def create_tokens(user_id: str, username: str) -> Tuple[str, str]: """ 生成访问令牌和刷新令牌 :param user_id: 用户ID :param username: 用户名 :return: (access_token, refresh_token) """ # 构建公共载荷（可根据需求扩展） common_payload = { "user_id": user_id, "username": username, "type": "access" # 标记令牌类型 } # 1. 生成access_token（短期） access_payload = common_payload.copy() access_payload["exp"] = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = jwt.encode(access_payload, SECRET_KEY, algorithm=ALGORITHM) # 2. 生成refresh_token（长期） refresh_payload = common_payload.copy() refresh_payload["type"] = "refresh" # 标记为刷新令牌 refresh_payload["exp"] = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS) refresh_token = jwt.encode(refresh_payload, SECRET_KEY, algorithm=ALGORITHM) return access_token, refresh_token # 2. 通用令牌验证（区分令牌类型） def verify_token(token: str, token_type: str = "access") -> Optional[dict]: """ 验证令牌并解析载荷 :param token: 令牌字符串 :param token_type: 期望的令牌类型（access/refresh） :return: 解析后的载荷（验证失败返回None） """ try: payload = jwt.decode( token, SECRET_KEY, algorithms=[ALGORITHM], options={"verify_exp": True} # 强制校验过期时间 ) # 校验令牌类型 if payload.get("type") != token_type: print(f"验证失败：令牌类型不是{token_type}") return None # 校验必要字段 if not all(key in payload for key in ["user_id", "username"]): print("验证失败：载荷缺少必要字段") return None return payload except JWTError as e: print(f"验证失败：{e}") return None # 3. 使用refresh_token刷新access_token def refresh_access_token(refresh_token: str) -> Optional[str]: """ 通过有效的refresh_token生成新的access_token :param refresh_token: 刷新令牌 :return: 新的access_token（刷新失败返回None） """ # 先验证refresh_token的有效性 payload = verify_token(refresh_token, token_type="refresh") if not payload: return None # 从refresh_token的载荷中提取用户信息，生成新的access_token user_id = payload["user_id"] username = payload["username"] # 构建新的access_token载荷 access_payload = { "user_id": user_id, "username": username, "type": "access", "exp": datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) } new_access_token = jwt.encode(access_payload, SECRET_KEY, algorithm=ALGORITHM) return new_access_token

@staticmethod async def login_by_mobile_password(login_mobile_request: LoginMobileRequest): jobseeker=await JobSeeker.get_or_none(mobile=login_mobile_request.mobile) if jobseeker is None: raise Exception("手机号不存在") if jobseeker.password != login_mobile_request.password: raise Exception("密码错误") access_token,refresh_token = create_tokens(str(jobseeker.id), jobseeker.mobile) return { 'jobseeker': jobseeker, 'access_token': access_token, 'refresh_token': refresh_token }

@job_seeker_router.post("/login_by_mobile_password", summary="手机号密码登录", description="手机号密码登录") async def login_by_mobile_password(login_mobile_request:LoginMobileRequest): res = await JobSeekerService.login_by_mobile_password(login_mobile_request) return { 'code':1, 'message':'登录成功', 'data':res }

业务 Service 层调用封装的create_tokens()，使用求职者唯一标识签发 JWT 令牌并响应至前端，后续接口依托令牌完成身份认证，避免反复传递账号凭证，实现无状态鉴权，有效提升系统安全等级。

求职者端--手机号验证码登录

在此处封装了生成验证码的方法，方便之后直接调用，在这里使用的是简单的6位随机数，其实是不太符合实际需求的，如果专业一点，可以直接调用容联云的短信发送。

def get_random_code(length: int = 4): code = ''.join(random.choices('1324567890', k=length)) return code

@staticmethod async def send_sms_code(mobile: str): jobseeker = await JobSeeker.get_or_none(mobile=mobile) if jobseeker is None: raise Exception("手机号不存在") code=get_random_code(length=6) logger.info(f"生成验证码:{code}") key = f"boss-api:login:sms:{mobile}" redis_client.set(key, code, ex=60 * 2) return code

@job_seeker_router.post("/send_sms_code/{mobile}", summary="发送验证码", description="发送验证码") async def send_sms_code(mobile: str): await JobSeekerService.send_sms_code(mobile) return { "code": 1, "message": "发送成功", }

因为验证码是有时效性的，只有在规定时间内才有效，在此处显然缓存十分有效。

@staticmethod async def login_by_mobile_code(login_request: LoginSmsCodeRequest): """ 1:判断手机号是否存在 2:redis中是否存在验证码并且一样,登录成功,生成token,删除redis中的验证码 3:不一样,登录失败 :return: """ jobseeker = await JobSeeker.get_or_none(mobile=login_request.mobile) if jobseeker is None: raise Exception("手机号不存在") key = f"boss-api:login:sms:{login_request.mobile}" redis_code = redis_client.get(key) if redis_code is None: raise Exception("验证码已过期") if redis_code != login_request.code: raise Exception("验证码错误") access_token, refresh_token = create_tokens(str(jobseeker.id), jobseeker.mobile) logger.info(f"生成 access_token:{access_token}") logger.info(f"生成 refresh_token:{refresh_token}") redis_client.delete(key) return { "jobseeker": jobseeker, "access_token": access_token, "refresh_token": refresh_token, }

@job_seeker_router.post("/login_by_mobile_code", summary="手机号验证码登录", description="手机号验证码登录") async def login_by_mobile_code(login_request: LoginSmsCodeRequest): res = await JobSeekerService.login_by_mobile_code(login_request) return { "code": 1, "message": "登录成功", "data": res }

在这里与手机号密码登录同理，在对手机号和验证码的基础判断之后也是调用相同的方法生成token，在生成token之后记得将缓存中对应的验证码删除即可。

求职者端--注册

@staticmethod async def register(register_req: RegisterRequest): mobile = register_req.mobile code = register_req.code password = register_req.password key = f"boss-api:login:sms:{mobile}" redis_code = redis_client.get(key) if redis_code is None: raise Exception("验证码已过期") if redis_code != code: raise Exception("验证码错误") jobseeker = await JobSeeker.get_or_none(mobile=mobile) if jobseeker is not None: raise Exception("手机号已存在,请登录") new_jobseeker = await JobSeeker.create( mobile=mobile, password=password ) await redis_client.delete(key) access_token, refresh_token = create_tokens(str(new_jobseeker.id), new_jobseeker.mobile) return { "jobseeker": new_jobseeker, "access_token": access_token, "refresh_token": refresh_token, }

@job_seeker_router.post("/register", summary="注册", description="注册") async def register(register_req: RegisterRequest): res = await JobSeekerService.register(register_req) return { "code": 1, "message": "注册成功", "data": res }

注册类似于手机号密码登录与手机号验证码登录的结合，首先获取前端传过来的手机号，验证码，密码等信息，在验证码验证的时候去缓存里与验证码进行对比，在进行基础型的判断之后删除对应缓存，创建新用户，并且调用与登录一样的方法生成token。

求职者端--忘记密码/重置密码

@job_seeker_router.post("/reset_pwd", summary="忘记密码重置接口") async def reset_password(reset_req: ResetPwdRequest): await JobSeekerService.reset_password(reset_req) return { "code": 1, "message": "重置成功" }

@staticmethod async def reset_password(reset_req: ResetPwdRequest): mobile = reset_req.mobile code = reset_req.code new_pwd=reset_req.new_password key = f"boss-api:login:sms:{mobile}" redis_code = redis_client.get(key) if redis_code is None: raise Exception("验证码已过期") if redis_code != code: raise Exception("验证码错误") jobseeker = await JobSeeker.get_or_none(mobile=mobile) if jobseeker is None: raise Exception("手机号不存在") jobseeker.password = new_pwd await jobseeker.save() await redis_client.delete(key) return 1

忘记密码的本质就是orm的修改，接受前端传过来的手机号，验证码，新密码的信息，对验证码进行判断后，将新密码替换掉旧密码即可，删除对应缓存。

求职者端-我的页面获取个人信息

# 获取用户信息 @staticmethod async def get_job_seeker_info_by_id(job_seeker_id: str): job_seeker=await JobSeeker.get_or_none(id=job_seeker_id) job_seeker.nickname=f'用户{job_seeker.mobile}' if job_seeker.nickname is None else job_seeker.nickname job_seeker.avatar = "https://gips1.baidu.com/it/u=436886321,1020119268&fm=3028&app=3028&f=JPEG&fmt=auto?w=1280&h=960" if job_seeker.avatar is None else job_seeker.avatar return job_seeker

@job_seeker_router.get("/me", summary="获取当前用户信息", description="获取当前用户信息") async def get_me(job_seeker_id=Depends(get_job_seeker_info)): job_seeker = await JobSeekerService.get_job_seeker_info_by_id(job_seeker_id) return { "code": 1, "message": "获取成功", "data": job_seeker }

接口借助依赖注入解析请求头中的 Token 提取求职者 ID，调用 Service 层依据 ID 查询并加工用户数据，完成登录求职者个人信息的拉取。如下时封装的依赖注入方法

def get_job_seeker_info(authorization: str|None = Header(None)) -> str: if not authorization: raise Exception("未登录") token=authorization.replace("Bearer ","") payload=verify_token(token) if payload is None: raise Exception("验证失败") job_seeker_id=payload["user_id"] return job_seeker_id

求职者端-上传/更新头像

async def update_avatar(job_seeker_id: str,file:UploadFile): file_content=await file.read() filename=file.filename oss=AliyunOSSTool() is_success,success_res=oss.upload_single_file(file_content,filename,oss_path="job_seeker_avatar/") if is_success: jobseeker=await JobSeeker.get_or_none(id=job_seeker_id) jobseeker.avatar=success_res["access_url"] await jobseeker.save() return success_res else: raise Exception("上传失败")

@job_seeker_router.post("/update_avatar", summary="上传头像", description="上传头像") async def update_avatar(file: UploadFile = File(...,description="上传文件"), job_seeker_id=Depends(get_job_seeker_info)): await JobSeekerService.update_avatar(job_seeker_id,file) return { "code": 1, "message": "上传头像成功" }

因为用户（不论身份）量是十分庞大的，放置到本地磁盘显然是不现实的，在此处接入了阿里云的对象存储，具体配置可以访问阿里云官方根据SDK进行操作。

上传文件的接口通过依赖注入解析请求头 Token，校验身份并获取当前登录求职者 ID，接收前端提交的头像文件；在 Service 层调用封装好的阿里云 OSS 工具，将图片二进制流上传至云端对象存储。文件上传成功后获取公开访问 URL，查询对应求职者记录并更新 avatar 字段，执行数据持久化；若 OSS 上传流程出现异常，主动抛出业务异常，交由全局异常处理器统一响应错误信息。

总结

求职者模块实现注册、多方式登录、密码重置、个人信息查询与头像上传能力，基于 JWT 长短双 Token 完成无状态身份认证，利用 Redis 维护短信验证码时效，接入阿里云 OSS 处理文件存储，依靠依赖注入统一鉴权，并遵循路由、Service 分层架构规范开发。