MySQL 巡检用户创建脚本（Python 版）

脚本依赖

需先安装 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 密码，脚本会自动完成版本适配、用户检查、权限授予。