自动管理腾讯云CVM按量付费实例，可以自动启动并修改/etc/hosts为新的公网ip，查询实例信息以及关闭实例

main.py

import os
import json
import time
import logging
import argparse
import yaml
from pprint import pprint
from tencentcloud.common import credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.common.exception.tencent_cloud_sdk_exception import (
    TencentCloudSDKException,
)
from tencentcloud.cvm.v20170312 import cvm_client, models


def setup_logger(logname=None, verbose=False):
    FORMAT = "[%(asctime)s] p%(process)d {%(pathname)s:%(lineno)d} %(levelname)s - %(message)s"
    logFormatter = logging.Formatter(FORMAT, datefmt="%m-%d %H:%M:%S")
    rootLogger = logging.getLogger()
    if verbose:
        rootLogger.setLevel(logging.DEBUG)
    else:
        rootLogger.setLevel(logging.INFO)

    if logname is not None:
        fileHandler = logging.FileHandler(logname)
        fileHandler.setFormatter(logFormatter)
        rootLogger.addHandler(fileHandler)

    consoleHandler = logging.StreamHandler()
    consoleHandler.setFormatter(logFormatter)
    rootLogger.addHandler(consoleHandler)


class Manager(object):
    def __init__(self, client: cvm_client.CvmClient, instance_id: str) -> None:
        self._client = client
        self._instance_id = instance_id

    def stop_ecs(self):
        req = models.StopInstancesRequest()
        params = {
            "InstanceIds": [self._instance_id],
            "StopType": "SOFT_FIRST",
            "StoppedMode": "STOP_CHARGING",
        }
        req.from_json_string(json.dumps(params))

        resp = self._client.StopInstances(req)
        # 输出json格式的字符串回包
        return json.loads(resp.to_json_string())

    def start_ecs(self):
        req = models.StartInstancesRequest()
        params = {"InstanceIds": [self._instance_id]}
        req.from_json_string(json.dumps(params))

        resp = self._client.StartInstances(req)
        # 输出json格式的字符串回包
        return json.loads(resp.to_json_string())

    def ensure_start(self, interval: int = 1):
        result = self.get_info()
        state = result["InstanceSet"][0]["InstanceState"]
        while state != "RUNNING":
            time.sleep(interval)
            try:
                result = self.start_ecs()
                logging.info("try to start ecs again...")
                result = self.get_info()
                state = result["InstanceSet"][0]["InstanceState"]
            except TencentCloudSDKException as e:
                logging.info(f"exc code: {e.code}")
                if (
                    e.code == "UnsupportedOperation.InstanceStateStarting"
                    or e.code == "OperationDenied.InstanceOperationInProgress"
                ):
                    time.sleep(20)
                    continue
                elif e.code == "UnsupportedOperation.InstanceStateRunning":
                    result = self.get_info()
                    state = result["InstanceSet"][0]["InstanceState"]
                else:
                    raise e
            except Exception as e:
                logging.exception(f"Failed to start ecs, err: {e}")

    def get_public_ip(self, interval: int = 1):
        result = self.get_info()
        public_ip_list = result["InstanceSet"][0]["PublicIpAddresses"]
        while public_ip_list is None:
            result = self.get_info()
            logging.info(f"ecs info in get_public_ip: {json.dumps(result, indent=4)}")
            public_ip_list = result["InstanceSet"][0]["PublicIpAddresses"]
            time.sleep(interval)
        return public_ip_list[0]

    def get_info(self):
        req = models.DescribeInstancesRequest()
        params = {"InstanceIds": [self._instance_id]}
        req.from_json_string(json.dumps(params))

        resp = self._client.DescribeInstances(req)
        return json.loads(resp.to_json_string())

    def is_running(self):
        result = self.get_info()
        logging.debug(f"ecs info from is_running: {json.dumps(result, indent=4)}")
        return result["InstanceSet"][0]["InstanceState"] == "RUNNING"


def edit_hosts_w_ip(public_ip):
    os.system(f"sudo sed -i '' 's/^.*{args.host}$/{public_ip} {args.host}/' /etc/hosts")


def main(action: str, config: dict, edit_hosts: bool = False):
    cred = credential.Credential(config["secret_id"], config["secret_key"])
    httpProfile = HttpProfile()
    httpProfile.endpoint = config["endpoint"]

    clientProfile = ClientProfile()
    clientProfile.httpProfile = httpProfile
    client = cvm_client.CvmClient(cred, config["ap"], clientProfile)

    manager = Manager(client, config["instance_id"])

    try:
        if action == "start":
            if manager.is_running():  # RUNNING, STOPPED
                public_ip = manager.get_public_ip()
                logging.info(f"ecs has already started, with public_ip: {public_ip}")
                if edit_hosts:
                    edit_hosts_w_ip(public_ip)
                    logging.info("/etc/hosts is modified")
                return

            manager.ensure_start()
            public_ip = manager.get_public_ip()
            logging.info(f"public ip: {public_ip}")
            if edit_hosts:
                edit_hosts_w_ip(public_ip)
                logging.info("/etc/hosts is modified")
        elif action == "stop":
            result = manager.stop_ecs()
            logging.info(f"stop ecs result: {json.dumps(result, indent=4)}")
        elif action == "info":
            result = manager.get_info()
            logging.info(f"get info: {json.dumps(result, indent=4)}")
            state = result["InstanceSet"][0]["InstanceState"]
            logging.info(f"instance state: {state}")
        else:
            raise NotImplementedError(f"unsupported action: {action}")
    except TencentCloudSDKException as e:
        logging.exception(f"Failed to {action}, err: {e}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-c", "--config", type=str, default="./config.yaml", help="path to config file"
    )
    parser.add_argument(
        "-e",
        "--edit-hosts",
        dest="edit_hosts",
        action="store_true",
        help="edit /etc/hosts directly",
    )
    parser.add_argument("action", type=str, choices=["start", "stop", "info"])
    parser.add_argument("--endpoint", type=str, default=None)
    parser.add_argument("--ap", type=str, default=None)
    parser.add_argument("--instance-id", dest="instance_id", type=str, default=None)
    parser.add_argument("--secret-id", dest="secret_id", type=str, default=None)
    parser.add_argument("--secret-key", dest="secret_key", type=str, default=None)
    parser.add_argument(
        "--host", type=str, default="tecs_gpu", help="item name in /etc/hosts"
    )
    args = parser.parse_args()

    setup_logger()

    if os.path.exists(args.config):
        with open(args.config, "r") as f:
            config = yaml.safe_load(f)
    else:
        config = {}

    config_from_args = {
        "endpoint": args.endpoint,
        "ap": args.ap,
        "instance_id": args.instance_id,
        "secret_id": args.secret_id,
        "secret_key": args.secret_key,
    }

    config_from_args = {k: v for k, v in config_from_args.items() if v is not None}

    config.update(config_from_args)

    main(args.action, config, args.edit_hosts)