222 lines
7.2 KiB
Python
222 lines
7.2 KiB
Python
"""
|
||
认证相关 API
|
||
登录、退出、修改密码
|
||
"""
|
||
import hashlib
|
||
import secrets
|
||
from datetime import datetime, timedelta
|
||
from flask import Blueprint, request, jsonify, g
|
||
from lib.db import execute_query, execute_update
|
||
from lib.logger import log_info, log_error, log_warning
|
||
|
||
auth_bp = Blueprint('auth', __name__, url_prefix='/api/auth')
|
||
|
||
|
||
def md5_password(password):
|
||
"""MD5加密密码"""
|
||
return hashlib.md5(password.encode('utf-8')).hexdigest()
|
||
|
||
|
||
def generate_token():
|
||
"""生成登录令牌"""
|
||
return secrets.token_hex(32)
|
||
|
||
|
||
def get_user_by_token(token):
|
||
"""根据token获取用户信息"""
|
||
if not token:
|
||
return None
|
||
|
||
try:
|
||
user = execute_query(
|
||
"SELECT id, username, real_name, token_expire FROM t_daily_report_admin WHERE token = %s",
|
||
(token,)
|
||
)
|
||
if user and user[0]:
|
||
user_info = user[0]
|
||
if user_info.get('token_expire') and user_info['token_expire'] > datetime.now():
|
||
return user_info
|
||
except Exception as e:
|
||
log_warning(f"[认证] 获取用户信息失败: {e}", 'auth')
|
||
|
||
return None
|
||
|
||
|
||
def get_current_user():
|
||
"""获取当前登录用户（从token）"""
|
||
token = request.headers.get('Authorization', '')
|
||
if token.startswith('Bearer '):
|
||
token = token[7:]
|
||
|
||
if not token:
|
||
return None
|
||
|
||
try:
|
||
user = execute_query(
|
||
"SELECT id, username, real_name, token_expire FROM t_daily_report_admin WHERE token = %s",
|
||
(token,)
|
||
)
|
||
if user and user[0]:
|
||
user_info = user[0]
|
||
if user_info.get('token_expire') and user_info['token_expire'] > datetime.now():
|
||
return user_info
|
||
except Exception as e:
|
||
log_warning(f"[认证] 获取用户信息失败: {e}", 'auth')
|
||
|
||
return None
|
||
|
||
|
||
def login_required(f):
|
||
"""登录验证装饰器"""
|
||
from functools import wraps
|
||
@wraps(f)
|
||
def decorated_function(*args, **kwargs):
|
||
user = get_current_user()
|
||
if not user:
|
||
return jsonify({'success': False, 'message': '未登录或登录已过期', 'code': 401}), 401
|
||
g.current_user = user
|
||
return f(*args, **kwargs)
|
||
return decorated_function
|
||
|
||
|
||
@auth_bp.route('/login', methods=['POST'])
|
||
def login():
|
||
"""登录"""
|
||
try:
|
||
data = request.get_json()
|
||
username = data.get('username', '').strip()
|
||
password = data.get('password', '').strip()
|
||
|
||
if not username or not password:
|
||
return jsonify({'success': False, 'message': '用户名和密码不能为空'})
|
||
|
||
# 查询用户
|
||
users = execute_query(
|
||
"SELECT id, username, password, real_name FROM t_daily_report_admin WHERE username = %s",
|
||
(username,)
|
||
)
|
||
|
||
if not users or not users[0]:
|
||
log_warning(f"[登录] 用户不存在: {username}", 'auth')
|
||
return jsonify({'success': False, 'message': '用户名或密码错误'})
|
||
|
||
user = users[0]
|
||
|
||
# 验证密码
|
||
password_md5 = md5_password(password)
|
||
if user['password'] != password_md5:
|
||
log_warning(f"[登录] 密码错误: {username}", 'auth')
|
||
return jsonify({'success': False, 'message': '用户名或密码错误'})
|
||
|
||
# 生成token，有效期7天
|
||
token = generate_token()
|
||
token_expire = datetime.now() + timedelta(days=7)
|
||
|
||
# 获取客户端IP
|
||
client_ip = request.remote_addr
|
||
|
||
# 更新用户token和登录信息
|
||
execute_update(
|
||
"""UPDATE t_daily_report_admin
|
||
SET token = %s, token_expire = %s, last_login_time = NOW(), last_login_ip = %s
|
||
WHERE id = %s""",
|
||
(token, token_expire, client_ip, user['id'])
|
||
)
|
||
|
||
log_info(f"[登录] 登录成功: {username}", 'auth')
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'message': '登录成功',
|
||
'data': {
|
||
'token': token,
|
||
'username': user['username'],
|
||
'real_name': user.get('real_name', ''),
|
||
'expires_in': 7 * 24 * 3600
|
||
}
|
||
})
|
||
except Exception as e:
|
||
log_error(f"[登录] 异常: {e}", 'auth', exc_info=True)
|
||
return jsonify({'success': False, 'message': '登录失败: ' + str(e)})
|
||
|
||
|
||
@auth_bp.route('/logout', methods=['POST'])
|
||
@login_required
|
||
def logout():
|
||
"""退出登录"""
|
||
try:
|
||
user = g.current_user
|
||
# 清除token
|
||
execute_update(
|
||
"UPDATE t_daily_report_admin SET token = NULL, token_expire = NULL WHERE id = %s",
|
||
(user['id'],)
|
||
)
|
||
log_info(f"[退出] 用户退出登录: {user['username']}", 'auth')
|
||
return jsonify({'success': True, 'message': '退出成功'})
|
||
except Exception as e:
|
||
log_error(f"[退出] 异常: {e}", 'auth', exc_info=True)
|
||
return jsonify({'success': False, 'message': '退出失败: ' + str(e)})
|
||
|
||
|
||
@auth_bp.route('/info', methods=['GET'])
|
||
@login_required
|
||
def get_user_info():
|
||
"""获取当前用户信息"""
|
||
try:
|
||
user = g.current_user
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'id': user['id'],
|
||
'username': user['username'],
|
||
'real_name': user.get('real_name', '')
|
||
}
|
||
})
|
||
except Exception as e:
|
||
return jsonify({'success': False, 'message': str(e)})
|
||
|
||
|
||
@auth_bp.route('/change-password', methods=['POST'])
|
||
@login_required
|
||
def change_password():
|
||
"""修改密码"""
|
||
try:
|
||
data = request.get_json()
|
||
old_password = data.get('old_password', '').strip()
|
||
new_password = data.get('new_password', '').strip()
|
||
|
||
if not old_password or not new_password:
|
||
return jsonify({'success': False, 'message': '旧密码和新密码不能为空'})
|
||
|
||
if len(new_password) < 6:
|
||
return jsonify({'success': False, 'message': '新密码长度不能少于6位'})
|
||
|
||
user = g.current_user
|
||
|
||
# 验证旧密码
|
||
users = execute_query(
|
||
"SELECT password FROM t_daily_report_admin WHERE id = %s",
|
||
(user['id'],)
|
||
)
|
||
|
||
if not users or not users[0]:
|
||
return jsonify({'success': False, 'message': '用户不存在'})
|
||
|
||
old_password_md5 = md5_password(old_password)
|
||
if users[0]['password'] != old_password_md5:
|
||
return jsonify({'success': False, 'message': '旧密码错误'})
|
||
|
||
# 更新密码
|
||
new_password_md5 = md5_password(new_password)
|
||
execute_update(
|
||
"UPDATE t_daily_report_admin SET password = %s, token = NULL, token_expire = NULL, update_time = NOW() WHERE id = %s",
|
||
(new_password_md5, user['id'])
|
||
)
|
||
|
||
log_info(f"[修改密码] 用户 {user['username']} 修改密码成功", 'auth')
|
||
|
||
return jsonify({'success': True, 'message': '密码修改成功，请重新登录'})
|
||
except Exception as e:
|
||
log_error(f"[修改密码] 异常: {e}", 'auth', exc_info=True)
|
||
return jsonify({'success': False, 'message': '修改失败: ' + str(e)})
|