Python 脚本：自动和手动 MySQL 主库切换

你好！以下是实现你所需功能的 Python 脚本：

import os

def read_file(file_path):
    with open(file_path, 'r') as file:
        content = file.read()
    return content.strip().split()

def check_container(ip):
    command = f"docker ps --filter name=mha*{ip}*4001 --format '{{.Names}}'"  
    result = os.popen(command).read().strip()
    return result

def start_container(script_path):
    os.system(f"sh {script_path}/aaa.sh")

def stop_container(ip, container_name):
    os.system(f"docker stop {container_name}")

def check_switch_result(ip, container_name):
    command = f"docker exec -i {container_name} mysql -uroot -ppassword -e 'SHOW SLAVE STATUS\G'"  
    result = os.popen(command).read().strip()
    if "Slave_IO_Running: Yes" in result and "Slave_SQL_Running: Yes" in result:
        new_master_ip = result.split("Master_Host: ")[1].split("\n")[0]
        return True, new_master_ip
    else:
        return False, container_name

def execute_command(ip, new_master_ip):
    command1 = f"mysql -uroot -ppassword -h{ip} -e 'CHANGE MASTER TO MASTER_HOST='{new_master_ip}', MASTER_PORT=4001, MASTER_AUTO_POSITION=1, MASTER_USER='repl', MASTER_PASSWORD='xxxxxxxxx';'"  
    command2 = f"mysql -uroot -ppassword -h{ip} -e 'start slave'"  
    os.system(command1)
    os.system(command2)

def switch(ip, container_name):
    command = f"docker exec -i mha4mysql-manager-0.58-4001 sh -c 'masterha_master_switch --master_state=alive --conf=/etc/mha/masterha_default.cnf --new_master_host={ip} --new_master_port=4001 --orig_master_is_new_slave --global_conf=/etc/mha/masterha_default.cnf'"  
    os.system(command)

def main():
    file_path = 'a.txt'
    script_path = '/path/to/aaa.sh'
    ip = '10.111.111.111'

    content = read_file(file_path)
    container_info = content[0].split()
    db_ip = container_info[0]
    db_name = container_info[2]

    if db_ip == ip and db_name.startswith('mha') and db_name.endswith('4001') and not check_container(ip):
        start_container(script_path)

    stop_container(db_ip, db_name)

    success, new_master_ip = check_switch_result(ip, db_name)
    if not success:
        print(f"{db_name} 自动切换失败")
        return

    execute_command(db_ip, new_master_ip)

    switch(db_ip, db_name)

    success, _ = check_switch_result(ip, db_name)
    if success:
        print("手动切换成功")
    else:
        print(f"{db_name} 手动切换失败")

if __name__ == '__main__':
    main()

请替换文件路径file_path和脚本路径script_path为实际的路径。