KGPython/service/UserService.py

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')):
                avatar_path = user.get("avatar")
                # 如果头像为空，设置默认头像
                if not avatar_path:
                    avatar_path = "/resource/avatar/4.png"
                
                return {
                    "id": user.get("id"),
                    "username": user.get("username"),
                    "avatar": avatar_path
                }
            
            return None
        except Exception as e:
            print(f"验证用户失败: {e}")
            return None
    
    def get_user_info(self, username: str) -> Optional[Dict[str, Any]]:
        """获取用户信息（不验证密码）"""
        try:
            user = self.get_user_by_username(username)
            if not user:
                return None
            
            avatar_path = user.get("avatar")
            # 如果头像为空，设置默认头像
            if not avatar_path:
                avatar_path = "/resource/avatar/4.png"
            
            return {
                "id": user.get("id"),
                "username": user.get("username"),
                "avatar": avatar_path
            }
        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
    
    def create_user(self, username: str, password: str, avatar_path: str = None) -> int:
        """创建新用户"""
        try:
            # 检查数据库连接状态
            if not self.mysql.is_connected():
                print("数据库未连接，尝试重新连接...")
                if not self.mysql.connect():
                    print("数据库连接失败")
                    return 0
            
            # 使用bcrypt加密密码
            hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
            
            # 设置默认头像
            if not avatar_path:
                avatar_path = "/resource/avatar/4.png"
            
            # 插入新用户
            sql = "INSERT INTO users (username, password, avatar) VALUES (%s, %s, %s)"
            print(f"执行SQL: {sql}")
            print(f"参数: {username}, {hashed_password}, {avatar_path}")
            
            # 执行插入
            result = self.mysql.execute_update(sql, (username, hashed_password, avatar_path))
            
            if result > 0:
                # 获取插入的用户ID
                id_sql = "SELECT LAST_INSERT_ID() as user_id"
                id_result = self.mysql.execute_query(id_sql)
                if id_result:
                    user_id = id_result[0].get("user_id", 0)
                    print(f"插入结果，用户ID: {user_id}")
                    return user_id
            
            return 0
        except Exception as e:
            print(f"创建用户失败: {e}")
            import traceback
            traceback.print_exc()
            return 0

# 创建全局用户服务实例
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