liuziyi

腳本依賴

需先安裝 MySQL 官方 Python 連接器，執行以下命令安裝：

pip install mysql-connector-python

完整 Python 腳本

import os
import getpass
import mysql.connector
from mysql.connector import Error

# 終端顏色配置（兼容大多數終端）
class Color:
    BLUE = '\033[34m'    # 藍色（開始/完成提示）
    RED = '\033[31m'     # 紅色（錯誤/警告）
    GREEN = '\033[32m'   # 綠色（成功提示）
    RESET = '\033[0m'    # 重置顏色

def print_separator():
    """打印分隔線，統一輸出格式"""
    print("-" * 31)

def connect_mysql(root_user, root_pwd, host="localhost", port="3306"):
    """
    連接 MySQL 數據庫
    返回：數據庫連接對象（成功）/ None（失敗）
    """
    try:
        conn = mysql.connector.connect(
            user=root_user,
            password=root_pwd,
            host=host,
            port=port,
            auth_plugin='mysql_native_password'  # 兼容 MySQL 5.x/8.x 認證
        )
        if conn.is_connected():
            return conn
    except Error as e:
        print(f"\n{Color.RED}MySQL 連接失敗：{str(e)}{Color.RESET}")
        # 常見錯誤提示
        if "Access denied" in str(e):
            print(f"{Color.RED}可能原因：root 密碼錯誤 / 該 IP 無訪問權限{Color.RESET}")
        elif "Can't connect to MySQL server" in str(e):
            print(f"{Color.RED}可能原因：MySQL 服務未啟動 / 端口被占用{Color.RESET}")
    return None

def get_mysql_version(conn):
    """獲取 MySQL 主版本號（5 或 8）"""
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT version()")
        version_str = cursor.fetchone()[0]  # 示例：5.7.36 / 8.0.32
        main_version = int(version_str.split('.')[0])
        cursor.close()
        print(f"\n{Color.GREEN}數據庫版本：{version_str}{Color.RESET}")
        return main_version
    except Error as e:
        print(f"\n{Color.RED}獲取版本失敗：{str(e)}{Color.RESET}")
        return None

def check_zyjc_user(conn):
    """檢查 zyjc@% 用戶是否已存在"""
    try:
        cursor = conn.cursor()
        # 精確查詢 zyjc@%（修復原腳本主機匹配問題）
        cursor.execute("SELECT 1 FROM mysql.user WHERE user = 'zyjc' AND host = '%'")
        result = cursor.fetchone()
        cursor.close()
        return result is not None  # 存在返回 True，不存在返回 False
    except Error as e:
        print(f"\n{Color.RED}檢查用戶失敗：{str(e)}{Color.RESET}")
        return False

def execute_sqls(conn, sql_list):
    """批量執行 SQL 語句"""
    try:
        cursor = conn.cursor()
        for sql in sql_list:
            # 打印當前執行的 SQL 描述（非原始 SQL，避免密碼泄露）
            if "CREATE USER" in sql:
                print(f"\n{Color.BLUE}(2) 創建 zyjc@% 用戶{Color.RESET}")
            elif "GRANT SELECT" in sql:
                print(f"{Color.BLUE}(3) 授予普通表查詢權限{Color.RESET}")
            elif "GRANT SHOW DATABASES" in sql:
                print(f"{Color.BLUE}(3) 授予 SHOW DATABASES 權限{Color.RESET}")
            elif "GRANT PROCESS" in sql:
                print(f"{Color.BLUE}(3) 授予線程查詢權限{Color.RESET}")
            elif "GRANT REPLICATION CLIENT" in sql:
                print(f"{Color.BLUE}(3) 授予主從復制查詢權限{Color.RESET}")
            elif "GRANT SERVICE_CONNECTION_ADMIN" in sql:
                print(f"{Color.BLUE}(3) 授予 MySQL 8.x 專屬連接權限{Color.RESET}")
            elif "FLUSH PRIVILEGES" in sql:
                print(f"\n{Color.BLUE}(4) 刷新權限緩存{Color.RESET}")
            
            # 執行 SQL
            cursor.execute(sql)
        conn.commit()
        cursor.close()
        print(f"\n{Color.GREEN}用戶創建與授權完成！{Color.RESET}")
        print(f"{Color.GREEN}用戶名：zyjc@% | 初始密碼：Gs5tgb%TGB6yhn^YHN{Color.RESET}")
        return True
    except Error as e:
        conn.rollback()
        print(f"\n{Color.RED}SQL 執行失敗：{str(e)}{Color.RESET}")
        return False

def main():
    # 1. 初始界面
    print_separator()
    print(f"|     {Color.BLUE}BEGIN!!!{Color.RESET}           |")
    print_separator()

    # 2. 輸入 root 密碼（隱藏輸入，避免明文泄露）
    print("\n(1) 請輸入 MySQL root 用戶密碼")
    root_pwd = getpass.getpass(prompt="root 密碼：")
    if not root_pwd:
        print(f"\n{Color.RED}密碼不能為空！{Color.RESET}")
        return

    # 3. 連接 MySQL
    conn = connect_mysql(root_user="root", root_pwd=root_pwd)
    if not conn:
        return  # 連接失敗直接退出

    try:
        # 4. 獲取 MySQL 主版本
        main_version = get_mysql_version(conn)
        if main_version not in [5, 8]:
            print(f"\n{Color.RED}不支持的 MySQL 版本（僅支持 5.x/8.x）{Color.RESET}")
            return

        # 5. 檢查 zyjc@% 用戶是否存在
        user_exist = check_zyjc_user(conn)
        if user_exist:
            print(f"\n{Color.RED}用戶 zyjc@% 已存在！無需重復創建！{Color.RESET}")
            return

        # 6. 生成對應版本的 SQL 列表（統一用戶主機為 %，修復原腳本 Bug）
        zyjc_password = "Gs5tgb%TGB6yhn^YHN"  # 初始密碼，可根據需求修改
        sql_list = [
            # 創建用戶（指定 % 主機）
            f"CREATE USER 'zyjc'@'%' IDENTIFIED BY '{zyjc_password}';",
            # 基礎權限
            "GRANT SELECT ON *.* TO 'zyjc'@'%';",
            "GRANT SHOW DATABASES ON *.* TO 'zyjc'@'%';",
            "GRANT PROCESS ON *.* TO 'zyjc'@'%';",
            "GRANT REPLICATION CLIENT ON *.* TO 'zyjc'@'%';",
            # 刷新權限
            "FLUSH PRIVILEGES;"
        ]

        # MySQL 8.x 新增必要權限（修復監控連接問題）
        if main_version == 8:
            sql_list.insert(-1, "GRANT SERVICE_CONNECTION_ADMIN ON *.* TO 'zyjc'@'%';")

        # 7. 執行 SQL 并創建用戶
        execute_sqls(conn, sql_list)

    finally:
        # 8. 關閉數據庫連接
        if conn.is_connected():
            conn.close()
        # 9. 結束界面
        print("\n" * 2)
        print_separator()
        print(f"|    {Color.BLUE}COMPLETION!!!{Color.RESET}       |")
        print_separator()

if __name__ == "__main__":
    main()

腳本使用方法

1. 將腳本保存為 create_zyjc_user.py。
2. 執行腳本：

   python create_zyjc_user.py
   

3. 按照提示輸入 MySQL root 密碼，腳本會自動完成版本適配、用戶檢查、權限授予。