阿里云CDT价格

@hongdashao #10 发布于10/5/2026, 3:45:12 PM
抢占式香港的 用的好好的；刚用了一个月多点；一个月四五块钱；而且把我 甲骨文新加坡中转了，落地新加坡 ai也解决了，这…… 还要啥自行车啊；

666 枪战士 不是后面就没了
我的ip 还被收了费用

@qqab #11 有保活脚本啊；没了 不删，会保留；等资源有了 就开机了；

@hongdashao #10 大佬，这个提醒项目地址发我一下，谢谢

@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()

@TK神话王思聪 #1 老哥 看我主页