ovh/app.py

"""
OVH API to order a dedicated server
"""
import os
import time
import ovh

APP_KEY = os.environ.get("OVH_APP_KEY")
APP_SECRET = os.environ.get("OVH_APP_SECRET")
CONSUMER_KEY = os.environ.get("OVH_CONSUMER_KEY")
AUTOPAY = os.environ.get("OVH_AUTOPAY", "1") == "1"
DEBUG = os.environ.get("OVH_DEBUG", "0") == "1"
DATACENTERS = os.environ.get("OVH_DATACENTERS", "lon,gra,rbx")
PLAN_CODE = os.environ.get("OVH_PLAN_CODE")
OPTIONS = os.environ.get("OVH_OPTIONS")
QUANTITY = int(os.environ.get("OVH_QUANTITY", "1"))
INTERVAL = int(os.environ.get("OVH_INTERVAL", "5"))

REGION = "ovh-eu"
ZONE = "IE"
PURCHASED = 0

# Create a client
client = ovh.Client(
    application_key=APP_KEY,
    application_secret=APP_SECRET,
    consumer_key=CONSUMER_KEY,
    endpoint="ovh-eu"
)


def get_availability():
    """
    Get availability of the server in the datacenter
    """
    response = client.get("/dedicated/server/datacenter/availabilities")
    for dc in DATACENTERS.split(","):
        for item in response:
            if item.get("planCode") == PLAN_CODE:
                for dc_info in item.get("datacenters", []):
                    if dc_info.get("datacenter") == dc and dc_info.get("availability") != "unavailable":
                        print(f"🔥 Available: {PLAN_CODE} in {dc_info['datacenter']}")
                        return {
                            "fqn": item.get("fqn"),
                            "planCode": item.get("planCode"),
                            "datacenter": dc_info["datacenter"],
                        }
    print("No availability found.")
    return None


def create_cart():
    """
    Create a cart and return the cart ID
    """
    response = client.post("/order/cart", ovhSubsidiary=ZONE)
    cart_id = response.get("cartId")
    print(f"Cart ID: {cart_id}")

    # Assign cart
    assign_url = f"/order/cart/{cart_id}/assign"
    client.post(assign_url)
    return cart_id


def add_item_to_cart(cart_id, plan_code):
    """
    Add item to cart and return item ID
    """
    response = client.post(
        f"/order/cart/{cart_id}/eco",
        planCode=plan_code,
        pricingMode="default",
        duration="P1M",
        quantity=1
    )
    item_id = response.get("itemId")
    print(f"Item ID: {item_id}")
    return item_id


def configure_item(cart_id, item_id, datacenter):
    """
    Configure the item in the cart
    """
    client.post(f"/order/cart/{cart_id}/item/{item_id}/configuration", label="dedicated_datacenter", value=datacenter)
    print("Configured dedicated_datacenter")

    client.post(f"/order/cart/{cart_id}/item/{item_id}/configuration", label="dedicated_os", value="none_64.en")
    print("Configured dedicated_os")

    client.post(f"/order/cart/{cart_id}/item/{item_id}/configuration", label="region", value=REGION)
    print("Configured region")


def add_options(cart_id, item_id):
    """
    Add options to the item in the cart
    Get options plan codes using GET /order/cart/{cartId}/eco/options
    """
    for option in OPTIONS.split(","):
        url = f"/order/cart/{cart_id}/eco/options"
        data = {
            "duration": "P1M",
            "itemId": int(item_id),
            "planCode": option,
            "pricingMode": "default",
            "quantity": 1,
        }
        client.post(
            url,
            **data
        )
        print(f"Added option: {option}")


def checkout_cart(cart_id):
    """
    Checkout the cart
    """
    url = f"/order/cart/{cart_id}/checkout"
    if DEBUG:
        client.get(url)
    else:
        data = {
            "autoPayWithPreferredPaymentMethod": AUTOPAY,
            "waiveRetractationPeriod": True,
        }
        client.post(url, **data)
    print("🎉 Order placed")


def run_task():
    """
    Run the task to order the server
    """
    if not PLAN_CODE:
        print("PLAN_CODE is required")
        return

    if not OPTIONS:
        print("OPTIONS is required")
        return

    if not APP_KEY or not APP_SECRET or not CONSUMER_KEY:
        print("OVH credentials are required")
        return

    try:
        availability = get_availability()
        if not availability:
            return

        plan_code = availability["planCode"]
        datacenter = availability["datacenter"]

        cart_id = create_cart()
        item_id = add_item_to_cart(cart_id, plan_code)
        configure_item(cart_id, item_id, datacenter)
        add_options(cart_id, item_id)
        checkout_cart(cart_id)
        return True

    except Exception as e:
        print(f"An error occurred: {e}")


if __name__ == "__main__":
    while True:
        purchased = run_task()
        if purchased:
            PURCHASED += 1
            print(f"🎉 Purchased: {PURCHASED}")
        if PURCHASED == QUANTITY:
            print("🎉 Complete")
            break
        time.sleep(INTERVAL)