djangoadmin/DjangoAdmin_Flask_EleVue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: P4c 最终批次-5 auth 模块（login/index 合并）

Browse Source 
【改造】login(data) 入参数据/返回 dict/raise BizError（验证码/限流/密码校验）；
_record_login 兼容 dict result；logout 修正误导入（auth→log service）
【修复】auth router login+index 合并的 docstring/import 结构混乱
【验收】flake8 零告警；应用启动；登录表单校验运行时正常；12 passed

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
DjangoAdmin Dev
2026-08-07 17:30:44 +08:00
co-authored by Claude
parent 6bd4951f95
commit b9ce9dc510
3 changed files with 96 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/auth/router.py
+53 -55
View File
@@ -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():
"""
src/modules/auth/schema.py
+14 -19
View File
@@ -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，用于从缓存中获取验证码",
)
src/modules/auth/service.py
+29 -47
View File
@@ -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)}")