You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
110 lines
4.0 KiB
110 lines
4.0 KiB
import bcrypt
|
|
from typing import Dict, Any, Optional
|
|
from util.mysql_utils import mysql_client
|
|
|
|
class UserService:
|
|
"""用户服务类，处理用户相关的业务逻辑"""
|
|
|
|
def __init__(self):
|
|
self.mysql = mysql_client
|
|
|
|
def get_user_by_username(self, username: str) -> Optional[Dict[str, Any]]:
|
|
"""根据用户名获取用户信息"""
|
|
try:
|
|
sql = "SELECT * FROM users WHERE username = %s"
|
|
users = self.mysql.execute_query(sql, (username,))
|
|
return users[0] if users else None
|
|
except Exception as e:
|
|
print(f"查询用户失败: {e}")
|
|
return None
|
|
|
|
def verify_user(self, username: str, password: str) -> Optional[Dict[str, Any]]:
|
|
"""验证用户登录信息"""
|
|
try:
|
|
user = self.get_user_by_username(username)
|
|
if not user:
|
|
return None
|
|
|
|
stored_password = user.get("password")
|
|
|
|
# 验证密码 - 使用bcrypt验证
|
|
if bcrypt.checkpw(password.encode('utf-8'), stored_password.encode('utf-8')):
|
|
return {
|
|
"id": user.get("id"),
|
|
"username": user.get("username"),
|
|
"avatar": user.get("avatar")
|
|
}
|
|
|
|
return None
|
|
except Exception as e:
|
|
print(f"验证用户失败: {e}")
|
|
return None
|
|
|
|
def update_user_avatar(self, username: str, avatar_path: str) -> bool:
|
|
"""更新用户头像"""
|
|
try:
|
|
# 检查数据库连接状态
|
|
if not self.mysql.is_connected():
|
|
print("数据库未连接，尝试重新连接...")
|
|
if not self.mysql.connect():
|
|
print("数据库连接失败")
|
|
return False
|
|
|
|
sql = "UPDATE users SET avatar = %s WHERE username = %s"
|
|
print(f"执行SQL: {sql}")
|
|
print(f"参数: {avatar_path}, {username}")
|
|
|
|
result = self.mysql.execute_update(sql, (avatar_path, username))
|
|
print(f"更新结果: {result}")
|
|
return result > 0
|
|
except Exception as e:
|
|
print(f"更新用户头像失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
def update_user_password(self, username: str, new_password: str) -> bool:
|
|
"""更新用户密码"""
|
|
try:
|
|
# 检查数据库连接状态
|
|
if not self.mysql.is_connected():
|
|
print("数据库未连接，尝试重新连接...")
|
|
if not self.mysql.connect():
|
|
print("数据库连接失败")
|
|
return False
|
|
|
|
# 使用bcrypt加密新密码
|
|
hashed_password = bcrypt.hashpw(new_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
|
|
sql = "UPDATE users SET password = %s WHERE username = %s"
|
|
print(f"执行SQL: {sql}")
|
|
print(f"参数: {hashed_password}, {username}")
|
|
|
|
result = self.mysql.execute_update(sql, (hashed_password, username))
|
|
print(f"更新结果: {result}")
|
|
return result > 0
|
|
except Exception as e:
|
|
print(f"更新用户密码失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
def verify_password(self, password: str, hashed_password: str) -> bool:
|
|
"""验证密码"""
|
|
try:
|
|
return bcrypt.checkpw(password.encode('utf-8'), hashed_password.encode('utf-8'))
|
|
except Exception as e:
|
|
print(f"验证密码失败: {e}")
|
|
return False
|
|
|
|
# 创建全局用户服务实例
|
|
user_service = UserService()
|
|
|
|
# 初始化MySQL连接
|
|
def init_mysql_connection():
|
|
"""初始化MySQL连接"""
|
|
try:
|
|
return user_service.mysql.connect()
|
|
except Exception as e:
|
|
print(f"初始化MySQL连接失败: {e}")
|
|
return False
|