# -*- coding: utf-8 -*-
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
from aliyunsdkecs.request.v20140526 import StartInstancesRequest, StopInstancesRequest, DescribeInstancesRequest
import json
import sys
import logging
# pip install aliyun-python-sdk-core aliyun-python-sdk-ecs
# ================== 1. 配置日志 ==================
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
stream=sys.stdout
)
logger = logging.getLogger(__name__)
# ================== 2. 配置阿里云凭证和ECS实例信息 ==================
ACCESS_KEY_ID = 'LTAIxxxxxxx' # 您的AccessKey ID
ACCESS_KEY_SECRET = 'xxxxxxxxxx' # 您的AccessKey Secret
REGION_ID = 'cn-hongkong' # 区域ID
ECS_INSTANCE_ID = 'i-xxxxxxxxx' # 您要控制的ECS实例ID
# 流量阈值 (GB)
TRAFFIC_THRESHOLD_GB = 180
# ================== 3. 初始化客户端 ==================
try:
client = AcsClient(ACCESS_KEY_ID, ACCESS_KEY_SECRET, REGION_ID)
logger.info("AcsClient initialized successfully.")
except Exception as e:
logger.error(f"Failed to initialize AcsClient: {e}")
sys.exit(1)
# ================== 4. 查询当前总流量 ==================
def get_total_traffic_gb(client):
request = CommonRequest()
request.set_domain('cdt.aliyuncs.com')
request.set_version('2021-08-13')
request.set_action_name('ListCdtInternetTraffic')
request.set_method('POST')
try:
response = client.do_action_with_exception(request)
response_json = json.loads(response.decode('utf-8'))
total_bytes = sum(d.get('Traffic', 0) for d in response_json.get('TrafficDetails', []))
total_gb = total_bytes / (1024 ** 3)
logger.info(f"当前总互联网流量: {total_gb:.2f} GB")
return total_gb
except Exception as e:
logger.error(f"获取CDT流量失败: {e}")
sys.exit(1)
# ================== 5. 查询ECS实例状态 ==================
def get_ecs_status(client, instance_id):
try:
request = DescribeInstancesRequest.DescribeInstancesRequest()
request.set_InstanceIds([instance_id])
response = client.do_action_with_exception(request)
response_json = json.loads(response.decode('utf-8'))
instances = response_json.get("Instances", {}).get("Instance", [])
if not instances:
logger.error("未找到该ECS实例信息。")
return None
status = instances[0].get("Status")
logger.info(f"ECS实例 {instance_id} 当前状态: {status}")
return status
except Exception as e:
logger.error(f"获取ECS实例状态失败: {e}")
return None
# ================== 6. 启动ECS实例 ==================
def ecs_start(client, instance_id):
status = get_ecs_status(client, instance_id)
if status == "Running":
logger.info(f"ECS实例 {instance_id} 已经是运行状态，无需启动。")
return
try:
request = StartInstancesRequest.StartInstancesRequest()
request.set_InstanceIds([instance_id])
request.set_accept_format('json')
response = client.do_action_with_exception(request)
logger.info(f"ECS启动响应: {response.decode('utf-8')}")
except Exception as e:
logger.error(f"启动ECS实例失败: {e}")
# ================== 7. 停止ECS实例 ==================
def ecs_stop(client, instance_id):
status = get_ecs_status(client, instance_id)
if status == "Stopped":
logger.info(f"ECS实例 {instance_id} 已经是停止状态，无需再次停止。")
return
try:
request = StopInstancesRequest.StopInstancesRequest()
request.set_InstanceIds([instance_id])
request.set_ForceStop(False)
request.set_accept_format('json')
response = client.do_action_with_exception(request)
logger.info(f"ECS停止响应: {response.decode('utf-8')}")
except Exception as e:
logger.error(f"停止ECS实例失败: {e}")
# ================== 8. 主流程 ==================
def main():
total_gb = get_total_traffic_gb(client)
if total_gb < TRAFFIC_THRESHOLD_GB:
logger.info(f"流量 {total_gb:.2f} GB < 阈值 {TRAFFIC_THRESHOLD_GB} GB，尝试启动 ECS")
ecs_start(client, ECS_INSTANCE_ID)
else:
logger.info(f"流量 {total_gb:.2f} GB ≥ 阈值 {TRAFFIC_THRESHOLD_GB} GB，尝试停止 ECS")
ecs_stop(client, ECS_INSTANCE_ID)
logger.info("脚本执行完毕。")
if __name__ == "__main__":
main()