python读取配置文件scp文件或文件夹

config.json 文件

{"common_params": {"deploy_location": "/data","source_file": "/home/test/aitest"},"devices": [{"name": "Device1","ip_address": "172.28.9.50","username": "test","password": "test","upgrade": false,"deploy_test_program": true},{"name": "Device2","ip_address": "192.168.1.101","username": "user2","password": "password2","upgrade": false,"deploy_test_program": false},{"name": "Device3","ip_address": "192.168.1.102","username": "user3","password": "password3","upgrade": false,"deploy_test_program": false}]
}

python读取配置文件scp

#!/usr/bin/python3
# encoding: utf-8import os
import json
import paramikodef upgrade_device(upgrade_option):if upgrade_option:# 实现升级被测设备的逻辑print("开始升级被测设备...")else:print("不进行升级被测设备。")def deploy_program(device_info, common_params):name = device_info['name']ip_address = device_info['ip_address']username = device_info['username']password = device_info['password']dest_path = common_params['deploy_location']source_file = common_params['source_file']upgrade_option = device_info['upgrade']deploy_test_program = device_info['deploy_test_program']# 填写被测设备IPprint(f"填写{name}的IP地址: {ip_address}")# 程序部署位置，默认为 /dataprint(f"{name}的程序部署位置: {dest_path}")print(f"{name}的源程序: {source_file}")print(f"{name}的用户名: {username}")print(f"{name}的密码: {password}")# 获取本地文件名filename = os.path.basename(source_file)# 在远程目录下拼接文件名remote_path = os.path.join(dest_path, filename)# 判断是否进行升级upgrade_device(upgrade_option)# 判断是否部署测试程序if deploy_test_program:SCP_CMD_BASE = r"""expect -c "set timeout 10 ;spawn scp -P {port} -r {localsource} {username}@{host}:{remotedest} ;expect *assword* {{{{ send {password}\r }}}} ;expect *\r ;expect \r ;expect eof"""".format(username=username, password=password, host=ip_address, localsource=source_file, remotedest=dest_path, port=22)SCP_CMD = SCP_CMD_BASE.format(localsource=source_file)print("execute SCP_CMD: ", SCP_CMD)
#        p = subprocess.Popen(SCP_CMD, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
#        p.communicate()os.system(SCP_CMD)print(f"{name}的程序部署完成。")else:print(f"{name}不需要部署测试程序。")if __name__ == "__main__":with open('config.json', 'r') as f:config = json.load(f)common_params = config['common_params']devices = config['devices']for device_info in devices:deploy_program(device_info, common_params)