scripts/utils.py

import logging
from typing import List
from alibabacloud_ecs20140526.client import Client as Ecs20140526Client
from alibabacloud_ecs20140526 import models as ecs_20140526_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
import config


# 发送文件到ECS实例
def send_file_to_ecs(client: Ecs20140526Client, instance_id_list: List[str], target_dir: str, name: str, content: str) -> None:
    for instance_id in instance_id_list:  # 遍历实例ID列表
        describe_instance_attribute_request = ecs_20140526_models.DescribeInstanceAttributeRequest(  # 创建查询实例属性的请求
            instance_id=instance_id  # 实例ID
        )
        runtime = util_models.RuntimeOptions()  # 创建运行时选项
        response = client.describe_instance_attribute_with_options(describe_instance_attribute_request, runtime)  # 调用ECS客户端的查询实例属性方法
        vpc_attributes = response.body.vpc_attributes  # 获取VPC属性
        vpc_id = vpc_attributes.vpc_id  # 获取VPC ID
        inner_ip_address = response.body.inner_ip_address  # 获取内网IP地址
        ssh_client = UtilClient(vpc_id, inner_ip_address, 'root', config.ALIBABA_CLOUD_ACCESS_KEY_ID, config.ALIBABA_CLOUD_ACCESS_KEY_SECRET)  # 创建SSH客户端
        ssh_client.upload_file(target_dir, name, content)  # 上传文件到实例
        logging.info(f"File {name} sent to instance {instance_id}")  # 记录日志

# 在ECS实例上执行命令
def run_command(client: Ecs20140526Client, instance_ids: List[str], command: str) -> None:
    for instance_id in instance_ids:  # 遍历实例ID列表
        describe_instance_attribute_request = ecs_20140526_models.DescribeInstanceAttributeRequest(  # 创建查询实例属性的请求
            instance_id=instance_id  # 实例ID
        )
        runtime = util_models.RuntimeOptions()  # 创建运行时选项
        response = client.describe_instance_attribute_with_options(describe_instance_attribute_request, runtime)  # 调用ECS客户端的查询实例属性方法
        vpc_attributes = response.body.vpc_attributes  # 获取VPC属性
        vpc_id = vpc_attributes.vpc_id  # 获取VPC ID
        inner_ip_address = response.body.inner_ip_address  # 获取内网IP地址
        ssh_client = UtilClient(vpc_id, inner_ip_address, 'root', config.ALIBABA_CLOUD_ACCESS_KEY_ID, config.ALIBABA_CLOUD_ACCESS_KEY_SECRET)  # 创建SSH客户端
        ssh_client.run_command(command)  # 在实例上执行命令
        logging.info(f"Command '{command}' executed on instance {instance_id}")  # 记录日志