@Christopher #13 这个我是使用 ai工具 自己手搓的;很简单 就一个 tg机器人token 然后 阿里云 里面创建 用户 给他相应的权限;如下:我脱敏了,改成自己配置:python定时任务执行就行了;格式自己转了,反正下面的就是全部的脚本代码:
-- 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
import requests
from datetime import datetime
pip install aliyun-python-sdk-core aliyun-python-sdk-ecs requests
================== 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 = '' # 您的AccessKey ID
ACCESS_KEY_SECRET = '' # 您的AccessKey Secret
REGION_ID = 'cn-hongkong' # 区域ID
ECS_INSTANCE_ID = '' # 您要控制的ECS实例ID
流量阈值 (GB)
TRAFFIC_THRESHOLD_GB = 200
余额预警阈值(低于这个数会用⚠️提醒)
BALANCE_WARNING_THRESHOLD = 10
================== Telegram 配置 ==================
TG_BOT_TOKEN = ""
TG_CHAT_ID = ""
================== 查询账户余额 ==================
def get_account_balance(client):
request = CommonRequest()
request.set_domain('business.aliyuncs.com')
request.set_version('2017-12-14')
request.set_action_name('QueryAccountBalance')
request.set_method('POST')
try:
response = client.do_action_with_exception(request)
response_json = json.loads(response.decode('utf-8'))
balance_info = response_json.get('Data', {})
available = balance_info.get('AvailableAmount', '未知')
currency = balance_info.get('Currency', '')
logger.info(f"账户可用余额: {available} {currency}")
return available, currency
except Exception as e:
logger.error(f"获取账户余额失败: {e}")
return None, None
def send_tg_summary(bot_token, chat_id, total_gb, threshold_gb, status_before, action, status_after, balance=None, currency=""):
icon = "🟢" if action == "启动" else "🔴" if action == "停止" else "⚪"
balance_line = ""
if balance is not None:
try:
balance_num = float(str(balance).replace(',', ''))
balance_icon = "⚠️" if balance_num < BALANCE_WARNING_THRESHOLD else "💰"
except (ValueError, TypeError):
balance_icon = "💰"
balance_line = f"{balance_icon} 账户余额:<b>{balance} {currency}</b>\n"
text = (
f"<b>📊 阿里云 CDT 流量监控报告</b>\n"
f"━━━━━━━━━━━━━━━━━\n"
f"🕒 <code>{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</code>\n\n"
f"📈 当前流量:<b>{total_gb:.2f} GB</b> / 阈值 {threshold_gb} GB\n"
f"🖥 实例状态:<code>{status_before}</code> → <code>{status_after}</code>\n"
f"{icon} 执行动作:<b>{action}</b>\n"
f"{balance_line}"
)
try:
requests.post(
f"https://api.telegram.org/bot{bot_token}/sendMessage",
data={"chat_id": chat_id, "text": text, "parse_mode": "HTML"},
timeout=5
)
except Exception as e:
logger.error(f"Telegram汇总消息发送失败: {e}")
================== 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)
status_before = get_ecs_status(client, ECS_INSTANCE_ID)
balance, currency = get_account_balance(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)
action = "启动"
else:
logger.info(f"流量 {total_gb:.2f} GB ≥ 阈值 {TRAFFIC_THRESHOLD_GB} GB,尝试停止 ECS")
ecs_stop(client, ECS_INSTANCE_ID)
action = "停止"
status_after = get_ecs_status(client, ECS_INSTANCE_ID)
send_tg_summary(
bot_token=TG_BOT_TOKEN,
chat_id=TG_CHAT_ID,
total_gb=total_gb,
threshold_gb=TRAFFIC_THRESHOLD_GB,
status_before=status_before,
action=action,
status_after=status_after,
balance=balance,
currency=currency
)
logger.info("脚本执行完毕。")
if name == "main":
main()