playwrite/db/bip_users_dao.py

"""
BIPUsers DAO - Data access object for user authentication and management
"""
from typing import Optional, Dict, Any, List
from db.base_dao import BaseDAO
from db.connection import get_connection
from config.schema import DatabaseType


class BIPUsersDAO(BaseDAO):
    """Data access object for BIPUsers table"""

    def authenticate(self, username: str, password: str) -> Optional[Dict[str, Any]]:
        """
        Authenticate a user with username and password

        Args:
            username: The username to authenticate
            password: The password to verify

        Returns:
            Dict with user info if authentication successful, None otherwise
            Returns: {id, username, user_type}
        """
        table_name = self._convert_sql('[dbo].[BIPUsers]')
        placeholder = self._get_placeholder()

        # 根据数据库类型选择列名格式
        if self._db_type == DatabaseType.MYSQL:
            sql = f"""
            SELECT ID, UserName, UserType
            FROM {table_name}
            WHERE UserName = {placeholder} AND Password = {placeholder}
            """
        else:
            sql = f"""
            SELECT [ID], [UserName], [UserType]
            FROM {table_name}
            WHERE [UserName] = {placeholder} AND [Password] = {placeholder}
            """

        with get_connection() as db:
            results = db.execute_query(sql, (username, password))
            if results:
                return {
                    'id': results[0]['ID'],
                    'username': results[0]['UserName'],
                    'user_type': results[0]['UserType']
                }
        return None

    def authenticate_by_computer_name(self, computer_name: str) -> Optional[Dict[str, Any]]:
        """
        Authenticate a user using computer name (silent login)

        Args:
            computer_name: The computer name to authenticate

        Returns:
            Dict with user info if authentication successful, None otherwise
            Returns: {id, username, user_type}
        """
        table_name = self._convert_sql('[dbo].[BIPUsers]')
        placeholder = self._get_placeholder()

        # Note: Column name is 'ComputerNmae' (typo in database schema)
        if self._db_type == DatabaseType.MYSQL:
            sql = f"""
            SELECT ID, UserName, UserType
            FROM {table_name}
            WHERE ComputerNmae = {placeholder}
            """
        else:
            sql = f"""
            SELECT [ID], [UserName], [UserType]
            FROM {table_name}
            WHERE [ComputerNmae] = {placeholder}
            """

        with get_connection() as db:
            results = db.execute_query(sql, (computer_name,))
            if results:
                return {
                    'id': results[0]['ID'],
                    'username': results[0]['UserName'],
                    'user_type': results[0]['UserType']
                }
        return None

    def get_all_users(self) -> List[Dict[str, Any]]:
        """
        Get all users from the database

        Returns:
            List of user dictionaries: [{id, username, user_type, create_time}]
        """
        table_name = self._convert_sql('[dbo].[BIPUsers]')

        # 根据数据库类型选择列名格式
        if self._db_type == DatabaseType.MYSQL:
            sql = f"""
            SELECT ID, UserName, UserType, CreateTime
            FROM {table_name}
            ORDER BY UserName
            """
        else:
            sql = f"""
            SELECT [ID], [UserName], [UserType], [CreateTime]
            FROM {table_name}
            ORDER BY [UserName]
            """

        with get_connection() as db:
            results = db.execute_query(sql)
            return [
                {
                    'id': row['ID'],
                    'username': row['UserName'],
                    'user_type': row['UserType'],
                    'create_time': row['CreateTime']
                }
                for row in results
            ]

    def create_user(self, username: str, password: str, user_type: str, computer_name: str = '') -> bool:
        """
        Create a new user with optional computer name for silent login

        Args:
            username: The username (must be unique)
            password: The password (plain text for internal tool)
            user_type: User type ('Admin', 'User', or 'Guest')
            computer_name: Optional computer name for silent login (default: '')

        Returns:
            True if successful, False otherwise
        """
        table_name = self._convert_sql('[dbo].[BIPUsers]')
        placeholder = self._get_placeholder()

        # 根据数据库类型选择列名格式
        if self._db_type == DatabaseType.MYSQL:
            if computer_name:
                sql = f"""
                INSERT INTO {table_name} (UserName, Password, UserType, ComputerNmae)
                VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder})
                """
                params = (username, password, user_type, computer_name)
            else:
                sql = f"""
                INSERT INTO {table_name} (UserName, Password, UserType)
                VALUES ({placeholder}, {placeholder}, {placeholder})
                """
                params = (username, password, user_type)
        else:
            if computer_name:
                sql = f"""
                INSERT INTO {table_name} ([UserName], [Password], [UserType], [ComputerNmae])
                VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder})
                """
                params = (username, password, user_type, computer_name)
            else:
                sql = f"""
                INSERT INTO {table_name} ([UserName], [Password], [UserType])
                VALUES ({placeholder}, {placeholder}, {placeholder})
                """
                params = (username, password, user_type)

        try:
            with get_connection() as db:
                db.execute_update(sql, params)
                return True
        except Exception as e:
            print(f"Error creating user: {e}")
            return False

    def update_user_type(self, username: str, user_type: str) -> bool:
        """
        Update a user's type

        Args:
            username: The username to update
            user_type: New user type ('Admin', 'User', or 'Guest')

        Returns:
            True if successful, False otherwise
        """
        table_name = self._convert_sql('[dbo].[BIPUsers]')
        placeholder = self._get_placeholder()

        # 根据数据库类型选择列名格式
        if self._db_type == DatabaseType.MYSQL:
            sql = f"""
            UPDATE {table_name}
            SET UserType = {placeholder}
            WHERE UserName = {placeholder}
            """
        else:
            sql = f"""
            UPDATE {table_name}
            SET [UserType] = {placeholder}
            WHERE [UserName] = {placeholder}
            """

        try:
            with get_connection() as db:
                db.execute_update(sql, (user_type, username))
                return True
        except Exception as e:
            print(f"Error updating user type: {e}")
            return False

    def update_password(self, username: str, new_password: str) -> bool:
        """
        Update a user's password

        Args:
            username: The username to update
            new_password: The new password (plain text for internal tool)

        Returns:
            True if successful, False otherwise
        """
        table_name = self._convert_sql('[dbo].[BIPUsers]')
        placeholder = self._get_placeholder()

        # 根据数据库类型选择列名格式
        if self._db_type == DatabaseType.MYSQL:
            sql = f"""
            UPDATE {table_name}
            SET Password = {placeholder}
            WHERE UserName = {placeholder}
            """
        else:
            sql = f"""
            UPDATE {table_name}
            SET [Password] = {placeholder}
            WHERE [UserName] = {placeholder}
            """

        try:
            with get_connection() as db:
                db.execute_update(sql, (new_password, username))
                return True
        except Exception as e:
            print(f"Error updating password: {e}")
            return False

    def delete_user(self, username: str) -> bool:
        """
        Delete a user

        Args:
            username: The username to delete

        Returns:
            True if successful, False otherwise
        """
        table_name = self._convert_sql('[dbo].[BIPUsers]')
        placeholder = self._get_placeholder()

        # 根据数据库类型选择列名格式
        if self._db_type == DatabaseType.MYSQL:
            sql = f"""
            DELETE FROM {table_name}
            WHERE UserName = {placeholder}
            """
        else:
            sql = f"""
            DELETE FROM {table_name}
            WHERE [UserName] = {placeholder}
            """

        try:
            with get_connection() as db:
                db.execute_update(sql, (username,))
                return True
        except Exception as e:
            print(f"Error deleting user: {e}")
            return False

    def user_exists(self, username: str) -> bool:
        """
        Check if a username already exists

        Args:
            username: The username to check

        Returns:
            True if username exists, False otherwise
        """
        table_name = self._convert_sql('[dbo].[BIPUsers]')
        placeholder = self._get_placeholder()

        # 根据数据库类型选择列名格式
        if self._db_type == DatabaseType.MYSQL:
            sql = f"""
            SELECT COUNT(*) as count FROM {table_name}
            WHERE UserName = {placeholder}
            """
        else:
            sql = f"""
            SELECT COUNT(*) as count FROM {table_name}
            WHERE [UserName] = {placeholder}
            """

        with get_connection() as db:
            results = db.execute_query(sql, (username,))
            return results[0]['count'] > 0 if results else False