开源了: ovh api 抢购代码(go语言)

本来不想发，有点扰乱秩序，不过自从let上那人发了流程说明以后，大家都在写脚本了。那我就为大家省时间吧。
代码在我的博客上，我就不贴过来了，需要的自取。
戳我

感谢15楼的哥们发了gpt改写的python版。有需要的也可以参考。 *经mjj测试，调用api时改用官方库 https://github.com/ovh/python-ovh 方可正常运行，否则会401鉴权错误。同时务必请注意，这个版本它在购买成功后没有停止循环，所以，会不停一直购买！切记修改！

如果是加区帐号，常量定义里，
zone请改成 CA,region改成 ovh-ca。

代码是抢购的ks-le-b ssd版。(25skleb01 bandwidth-300-25skle ram-32g-ecc-2400-25skle softraid-2x450nvme-25skle)

如果要抢5o，参数相应改成
plancode: 24ska01
options: bandwidth-100-24sk ram-64g-noecc-2133-24ska01 softraid-1x480ssd-24ska01

代码是设置的下单完成后即支付，如果你只是想下单晚点再支付，请把checkout的autoPayWithPreferredPaymentMethod自动支付改成false即可。

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

可乐交易中介AI 服务

701days ago edited 701days ago

#15