import requests import json import time import logging # 常量定义 APP_KEY = "ovh appkey" # OVH的应用key APP_SECRET = "ovh appsecret" # OVH的应用secret CONSUMER_KEY = "ovh consumerkey" # OVH的消费者key REGION = "ovh-eu" # 区域设置为 TG_TOKEN = "telegrambot token" # 你的Telegram Bot Token TG_CHAT_ID = "telegram chatid" # 你希望发送消息的Telegram Chat ID IAM = "go-ovh-ie" # 你的标识 ZONE = "IE" # OVH子公司区域设置 # 设置日志输出 logging.basicConfig(level=logging.INFO) def send_telegram_msg(bot_token, chat_id, message): """向指定的 Telegram 发送消息""" url = f"https://api.telegram.org/bot{bot_token}/sendMessage" payload = { "chat_id": chat_id, "text": message } headers = {"Content-Type": "application/json"} try: # 发送POST请求 response = requests.post(url, json=payload, headers=headers) response.raise_for_status() # 检查请求是否成功 if response.status_code != 200: logging.error(f"发送消息失败: {response.status_code}, {response.text}") return False return True except requests.exceptions.RequestException as e: logging.error(f"发送Telegram消息时发生错误: {e}") return False def run_task(): """执行任务，检查OVH的可用性并处理购买流程""" # 创建OVH API客户端 try: # 获取数据中心的可用性信息 url = f"https://api.ovh.com/1.0/dedicated/server/datacenter/availabilities" headers = { "X-Ovh-Application": APP_KEY, "X-Ovh-Consumer": CONSUMER_KEY } response = requests.get(url, headers=headers) response.raise_for_status() result = response.json() # 解析返回的JSON数据 except requests.exceptions.RequestException as e: logging.error(f"获取OVH数据中心可用性时失败: {e}") return found_available = False fqn, plan_code, datacenter = None, None, None for item in result: if item["planCode"] == "25skleb01": # 如果找到了符合条件的计划 fqn = item["fqn"] plan_code = item["planCode"] datacenters = item["datacenters"] for dc_info in datacenters: availability = dc_info["availability"] datacenter = dc_info["datacenter"] logging.info(f"FQN: {fqn}") logging.info(f"Availability: {availability}") logging.info(f"Datacenter: {datacenter}") logging.info("------------------------") if availability != "unavailable": # 如果该数据中心是可用的 found_available = True break if found_available: logging.info(f"正在继续操作，FQN: {fqn} Datacenter: {datacenter}") break if not found_available: logging.info("没有可购买的记录") return msg = f"{IAM}: 在 {datacenter} 找到 ks-le-b 可用" if not send_telegram_msg(TG_TOKEN, TG_CHAT_ID, msg): logging.error("发送Telegram消息失败") return # 创建购物车 try: logging.info("创建购物车") url = f"https://api.ovh.com/1.0/order/cart" payload = {"ovhSubsidiary": ZONE} response = requests.post(url, json=payload, headers=headers) response.raise_for_status() cart_result = response.json() except requests.exceptions.RequestException as e: logging.error(f"创建购物车失败: {e}") return cart_id = cart_result["cartId"] logging.info(f"购物车ID: {cart_id}") # 绑定购物车 try: logging.info("绑定购物车") url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/assign" response = requests.post(url, headers=headers) response.raise_for_status() except requests.exceptions.RequestException as e: logging.error(f"绑定购物车失败: {e}") return # 将商品添加到购物车 try: logging.info("将商品添加到购物车") url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/eco" payload = { "planCode": plan_code, "pricingMode": "default", "duration": "P1M", # 持续时间为1个月 "quantity": 1 } response = requests.post(url, json=payload, headers=headers) response.raise_for_status() item_result = response.json() except requests.exceptions.RequestException as e: logging.error(f"将商品添加到购物车失败: {e}") return item_id = item_result["itemId"] logging.info(f"商品ID: {item_id}") # 获取必需的配置 try: logging.info("检查必需的配置") url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/item/{item_id}/requiredConfiguration" response = requests.get(url, headers=headers) response.raise_for_status() required_config = response.json() except requests.exceptions.RequestException as e: logging.error(f"获取必需配置失败: {e}") return dedicated_os = "none_64.en" region_value = None for config in required_config: if config["label"] == "region": allowed_values = config["allowedValues"] if allowed_values: region_value = allowed_values[0] configurations = [ {"label": "dedicated_datacenter", "value": datacenter}, {"label": "dedicated_os", "value": dedicated_os}, {"label": "region", "value": region_value}, ] # 配置购物车中的商品 for config in configurations: try: logging.info(f"配置 {config['label']}") url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/item/{item_id}/configuration" payload = {"label": config["label"], "value": config["value"]} response = requests.post(url, json=payload, headers=headers) response.raise_for_status() except requests.exceptions.RequestException as e: logging.error(f"配置 {config['label']} 失败: {e}") return # 添加附加选项 options = [ "bandwidth-300-25skle", "ram-32g-ecc-2400-25skle", "softraid-2x450nvme-25skle" ] for option in options: try: logging.info(f"添加选项 {option}") url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/eco/options" payload = { "duration": "P1M", # 选项持续时间为1个月 "itemId": int(item_id), "planCode": option, "pricingMode": "default", "quantity": 1 } response = requests.post(url, json=payload, headers=headers) response.raise_for_status() except requests.exceptions.RequestException as e: logging.error(f"添加选项 {option} 失败: {e}") return # 进行结账 try: logging.info("结账") url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/checkout" response = requests.get(url, headers=headers) response.raise_for_status() checkout_result = response.json() # 提交结账请求 url = f"https://api.ovh.com/1.0/order/cart/{cart_id}/checkout" payload = { "autoPayWithPreferredPaymentMethod": True, "waiveRetractationPeriod": True # 自动支付并跳过撤销期 } response = requests.post(url, json=payload, headers=headers) response.raise_for_status() except requests.exceptions.RequestException as e: logging.error(f"结账失败: {e}") return logging.info("完成购买！") def main(): """主函数：循环执行任务""" while True: run_task() # 执行任务 time.sleep(20) # 每20秒执行一次任务 if __name__ == "__main__": main() GPT转python
@若如初见 #10 开不开机是ovh的事情，只要api能下单然后进入等待开机状态就算结束了，后面听天命。
@jayolim #9 这个其实不算脚本，这是官方api，只要不滥用问题不大。 我看到一个毛胚版的，它就疯狂下单，这个肯定很容易风控，至少得检查一下有没有货及机房啊。
@陌生人 #12
鸡腿奉上，谢谢你的分享。大家一起蹲ks-a
@jayolim #13 加油，祝你好运。
GPT转python
刚琢磨完python版的
给你鸡脚
@陌生人 #0
老哥 你的ks-a开机了吗
感谢分享！
马克