diff --git a/src/modules/auth/router.py b/src/modules/auth/router.py index e95d418..4161af3 100644 --- a/src/modules/auth/router.py +++ b/src/modules/auth/router.py @@ -43,30 +43,33 @@ 登录认证控制器模块 ============================================================= -提供系统登录相关的API接口，包括： -- 用户登录认证 -- 验证码生成 - -所属包： application.views -依赖模块： application.services.logins - 登录业务逻辑层 - utils.captcha - 验证码生成工具 +提供登录、验证码、退出登录以及用户自助信息接口。 """ - from flask import Blueprint - +from sqlalchemy import and_ +from application.middleware.check_demo import check_demo from src.modules.auth import service as login - -login_blue = Blueprint('login', __name__, url_prefix='/') +from src.modules.log import service as login_log +from src.modules.system.user.models import User +from src.modules.system.menu import service as menu +from src.modules.system.user import service as user +from src.common.utils.request import parse_request_body from src.core import response as R from src.common.utils.captcha import generate_captcha from src.core.jwt import add_token_to_blacklist, get_access_token +from src.common.utils.rich_text import get_file_url +from src.core.security import get_user_id + + +login_blue = Blueprint("login", __name__, url_prefix="/") # ============================================================ # 用户登录 # ============================================================ -@login_blue.route('/login', methods=["POST"]) + +@login_blue.route("/login", methods=["POST"]) def user_login(): """ 用户登录认证 @@ -77,15 +80,18 @@ def user_login(): Returns: JSON: 登录结果，成功时返回 token，失败时返回错误信息 """ - result = login.login() - return result + json_data, err = parse_request_body() + if err: + return R.failed(msg=err) + return R.ok(msg="登录成功", data=login.login(json_data)) # ============================================================ # 验证码获取 # ============================================================ -@login_blue.route('/captcha', methods=["GET"]) + +@login_blue.route("/captcha", methods=["GET"]) def captcha(): """ 获取图形验证码 @@ -100,8 +106,8 @@ def captcha(): """ result = generate_captcha() - id_key = result['idKey'] # 验证码唯一标识 - data = result['data'] # 验证码图片 Base64 数据 + id_key = result["idKey"] # 验证码唯一标识 + data = result["data"] # 验证码图片 Base64 数据 # 拼接完整的图片 DataURL 格式 img_str = "data:image/png;base64," + data @@ -113,7 +119,8 @@ def captcha(): # 退出登录 # ============================================================ -@login_blue.route('/logout', methods=["GET"]) + +@login_blue.route("/logout", methods=["GET"]) def logout(): """ 用户退出登录 @@ -134,18 +141,17 @@ def logout(): import time from flask import g, request from src.modules.log.models import LoginLog - from src.modules.auth import service as login_log from src.core.security import get_username from src.common.utils.ip import get_client_ip from src.common.utils.ip2region import get_ip_location from src.common.utils.user_agent import parse_user_agent - username = get_username() or '未知' + username = get_username() or "未知" client_ip = get_client_ip() ua_info = parse_user_agent() # 请求耗时 - start_time = getattr(g, 'request_start_time', None) + start_time = getattr(g, "request_start_time", None) consume_time = int((time.time() - start_time) * 1000) if start_time else 0 login_log.record_login_log( @@ -154,25 +160,25 @@ def logout(): status=LoginLog.Status.NORMAL, log_type=LoginLog.LogType.LOGOUT, method=request.method, - request_method=request.endpoint or '', + request_method=request.endpoint or "", url=request.path, - param='', - result='退出成功', + param="", + result="退出成功", ip=client_ip, location=get_ip_location(client_ip), - os=ua_info.get('os', '未知'), - browser=ua_info.get('browser', '未知'), + os=ua_info.get("os", "未知"), + browser=ua_info.get("browser", "未知"), consume_time=consume_time, - error='', + error="", ) except Exception as e: import logging + logging.getLogger(__name__).warning(f"记录退出日志失败: {str(e)}") return R.ok(msg="注销成功") - """ 首页控制器模块 ============================================================= @@ -189,26 +195,17 @@ def logout(): application.services.user - 用户业务逻辑层 application.models.user - 用户数据模型 """ -from sqlalchemy import and_ -from flask import Blueprint -from application.middleware.check_demo import check_demo -from src.modules.system.user.models import User -from src.modules.system.menu import service as menu -from src.modules.system.user import service as user - -index_blue = Blueprint('index', __name__, url_prefix='/index') -from src.core import response as R -from src.common.utils.rich_text import get_file_url -from src.core.security import get_user_id +index_blue = Blueprint("index", __name__, url_prefix="/index") # ============================================================ # 系统菜单获取 # ============================================================ -@index_blue.route('/getMenu', methods=["GET"]) + +@index_blue.route("/getMenu", methods=["GET"]) def get_menus(): """ 获取系统菜单列表 @@ -227,7 +224,8 @@ def get_menus(): # 用户信息获取 # ============================================================ -@index_blue.route('/getUser', methods=["GET"]) + +@index_blue.route("/getUser", methods=["GET"]) def get_user(): """ 获取当前用户信息 @@ -241,9 +239,7 @@ def get_user(): JSON: 用户详细信息 """ # 查询用户信息 - current_user = User.query.filter( - and_(User.id == get_user_id(), User.is_delete == 0) - ).first() + current_user = User.query.filter(and_(User.id == get_user_id(), User.is_delete == 0)).first() if not current_user: return R.failed("用户不存在") @@ -253,15 +249,15 @@ def get_user(): # 构建用户信息 info = { - 'realname': current_user.realname, - 'avatar': get_file_url(current_user.avatar), - 'gender': current_user.gender, - 'mobile': current_user.mobile, - 'email': current_user.email, - 'intro': current_user.intro, - 'address': current_user.address, - 'roles': [], - 'permissions': permissions, + "realname": current_user.realname, + "avatar": get_file_url(current_user.avatar), + "gender": current_user.gender, + "mobile": current_user.mobile, + "email": current_user.email, + "intro": current_user.intro, + "address": current_user.address, + "roles": [], + "permissions": permissions, } return R.ok(data=info) @@ -271,7 +267,8 @@ def get_user(): # 个人中心信息更新 # ============================================================ -@index_blue.route('/userInfo', methods=["PUT"]) + +@index_blue.route("/userInfo", methods=["PUT"]) @check_demo def update_user_info(): """ @@ -291,7 +288,8 @@ def update_user_info(): # 密码更新 # ============================================================ -@index_blue.route('/updatePwd', methods=["PUT"]) + +@index_blue.route("/updatePwd", methods=["PUT"]) @check_demo def update_pwd(): """ diff --git a/src/modules/auth/schema.py b/src/modules/auth/schema.py index 7144dda..ad131b0 100644 --- a/src/modules/auth/schema.py +++ b/src/modules/auth/schema.py @@ -57,6 +57,7 @@ from src.common.forms.base_form import BaseForm # 登录表单验证 # ============================================================ + class LoginForm(BaseForm): """ 登录表单验证类 @@ -74,45 +75,39 @@ class LoginForm(BaseForm): # 登录账号 # ============================================================ username = StringField( - label='登录账号', - validators=[ - DataRequired(message='登录账号不能为空'), - Length(max=20, message='登录账号长度不得超过20个字符') - ], - description='用户登录账号，支持用户名/手机号/邮箱' + label="登录账号", + validators=[DataRequired(message="登录账号不能为空"), Length(max=20, message="登录账号长度不得超过20个字符")], + description="用户登录账号，支持用户名/手机号/邮箱", ) # ============================================================ # 登录密码 # ============================================================ password = StringField( - label='登录密码', + label="登录密码", validators=[ - DataRequired(message='登录密码不能为空'), - Length(min=6, max=12, message='登录密码长度在6~12个字符之间') + DataRequired(message="登录密码不能为空"), + Length(min=6, max=12, message="登录密码长度在6~12个字符之间"), ], - description='用户登录密码，长度6-12个字符' + description="用户登录密码，长度6-12个字符", ) # ============================================================ # 验证码 # ============================================================ code = StringField( - label='验证码', - validators=[ - DataRequired(message='验证码不能为空'), - Length(min=6, max=6, message='验证码长度为6个字符') - ], - description='图片验证码，不区分大小写' + label="验证码", + validators=[DataRequired(message="验证码不能为空"), Length(min=6, max=6, message="验证码长度为6个字符")], + description="图片验证码，不区分大小写", ) # ============================================================ # KEY值 # ============================================================ key = StringField( - label='KEY值', + label="KEY值", validators=[ - DataRequired(message='KEY值不能为空'), + DataRequired(message="KEY值不能为空"), ], - description='验证码唯一标识KEY，用于从缓存中获取验证码' + description="验证码唯一标识KEY，用于从缓存中获取验证码", ) diff --git a/src/modules/auth/service.py b/src/modules/auth/service.py index 16fef53..0563e4a 100644 --- a/src/modules/auth/service.py +++ b/src/modules/auth/service.py @@ -48,9 +48,8 @@ from src.modules.system.user.models import User from src.modules.log import service as login_log from config import CAPTCHA_BYPASS_CODE, FLASK_DEMO, LOGIN_FAIL_MAX, LOGIN_LOCK_SECONDS from src.common.utils import regular -from src.core import response as R +from src.core.exceptions import BizError from src.core.password import verify_password -from src.common.utils.request import parse_request_body from src.core.jwt import create_token from src.core.redis_client import redis from src.common.utils.ip import get_client_ip @@ -59,7 +58,7 @@ from src.common.utils.user_agent import parse_user_agent # 系统登录 -def login(): +def login(data: dict) -> dict: """ 用户登录接口 @@ -70,16 +69,11 @@ def login(): - 成功: {"code": 200, "msg": "登录成功", "data": {"access_token": "token值"}} - 失败: {"code": 非200, "msg": "错误信息"} """ - # 获取参数 - json_data, error = parse_request_body() - if error: - return R.failed(msg=error) # 表单验证 - form = LoginForm(data=json_data) + form = LoginForm(data=data) if not form.validate(): # 获取表单验证失败的错误信息 - err_msg = regular.get_err(form) - return R.failed(msg=err_msg) + raise BizError(regular.get_err(form)) # 提取表单数据 username = form.username.data # 登录账号 @@ -94,27 +88,22 @@ def login(): else: # 验证码不区分大小写比对 if not image_code or image_code.lower() != code.lower(): - return R.failed("验证码不正确") + raise BizError("验证码不正确") # 验证码使用后立即删除，防止重复使用 redis.delete(key) # 查询用户信息（只查询未删除的用户） - user = User.query.filter( - and_( - User.username == username, - User.is_delete == 0 - ) - ).first() + user = User.query.filter(and_(User.username == username, User.is_delete == 0)).first() # 校验用户是否存在 if not user: _record_login(username, LoginLog.Status.ERROR, error="用户不存在") - return R.failed("用户不存在") + raise BizError("用户不存在") # 校验用户状态（可选，根据业务需求） - if hasattr(user, 'status') and user.status != 1: + if hasattr(user, "status") and user.status != 1: _record_login(username, LoginLog.Status.ERROR, error="账号已被禁用") - return R.failed("账号已被禁用，请联系管理员") + raise BizError("账号已被禁用，请联系管理员") # 检查登录失败次数（在密码验证之前，避免锁定后仍执行高成本的bcrypt计算） client_ip = request.remote_addr @@ -123,7 +112,7 @@ def login(): if int(attempts) >= LOGIN_FAIL_MAX: _record_login(username, LoginLog.Status.ERROR, error="登录失败次数过多，账号被锁定") lock_minutes = LOGIN_LOCK_SECONDS // 60 - return R.failed(msg=f"登录失败次数过多，请{lock_minutes}分钟后再试") + raise BizError(f"登录失败次数过多，请{lock_minutes}分钟后再试") # 密码验证 if not verify_password(password, user.salt, user.password): @@ -133,30 +122,21 @@ def login(): else: redis.incr(login_attempts_key) _record_login(username, LoginLog.Status.ERROR, error="密码不正确") - return R.failed(msg="密码不正确") + raise BizError("密码不正确") # 登录成功后清除失败计数 redis.delete(login_attempts_key) # 生成访问令牌 - access_token = create_token({ - "userId": user.id, - "username": user.username, - "realname": user.realname - }) + access_token = create_token({"userId": user.id, "username": user.username, "realname": user.realname}) - # 登录成功，返回令牌 - result = R.ok( - msg="登录成功", - data={ - "access_token": access_token - } - ) + # 登录成功，返回令牌（纯数据） + result = {"code": 0, "data": {"access_token": access_token}, "msg": "登录成功", "ok": True} - # 记录登录成功日志（传入result用于记录返回结果） + # 记录登录成功日志 _record_login(username, LoginLog.Status.NORMAL, result_data=result) - return result + return {"access_token": access_token} def _record_login(username, status, error="", result_data=None): @@ -180,35 +160,36 @@ def _record_login(username, status, error="", result_data=None): ua_info = parse_user_agent() # 请求参数（密码脱敏） - param = '' + param = "" try: json_data = request.get_json(silent=True) if json_data: safe_data = dict(json_data) - if 'password' in safe_data: - safe_data['password'] = '******' + if "password" in safe_data: + safe_data["password"] = "******" param = json.dumps(safe_data, ensure_ascii=False) except Exception: pass # 返回结果（只取响应JSON中的data字段） - result = '' + result = "" if result_data is not None: try: - data = result_data.get_json(silent=True) + # 兼容：新范式传 dict，旧范式传 R 响应 + data = result_data.get_json(silent=True) if hasattr(result_data, "get_json") else result_data if data and isinstance(data, dict): - result_data_value = data.get('data') + result_data_value = data.get("data") if result_data_value is not None: result = json.dumps(result_data_value, ensure_ascii=False) else: - result = data.get('msg', '') + result = data.get("msg", "") except Exception: result = str(result_data) elif error: result = error # 请求耗时 - start_time = getattr(g, 'request_start_time', None) + start_time = getattr(g, "request_start_time", None) consume_time = int((time.time() - start_time) * 1000) if start_time else 0 login_log.record_login_log( @@ -217,18 +198,19 @@ def _record_login(username, status, error="", result_data=None): status=status, log_type=LoginLog.LogType.LOGIN, method=request.method, - request_method=request.endpoint or '', + request_method=request.endpoint or "", url=request.path, param=param, result=result, ip=client_ip, location=get_ip_location(client_ip), - os=ua_info.get('os', '未知'), - browser=ua_info.get('browser', '未知'), + os=ua_info.get("os", "未知"), + browser=ua_info.get("browser", "未知"), consume_time=consume_time, error=error, ) except Exception as e: # 日志记录失败不影响主流程 import logging + logging.getLogger(__name__).warning(f"记录登录日志失败: {str(e)}")