92 lines
2.6 KiB
Python
92 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""IPRoyal reseller API wrapper — datacenter/ISP/residential dedicated orders.
|
|
|
|
Base: https://apid.iproyal.com/v1/reseller (auth: X-Access-Token header)
|
|
Products: 3=Datacenter Dedicated, 9=ISP Dedicated, 6=Residential, 11=Web Unblocker
|
|
"""
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
import requests
|
|
|
|
CFG = json.load(open(os.environ.get("PLEIADES_CONFIG", "/opt/pleiades/config.json")))
|
|
IPR = CFG["iproyal"]
|
|
TOKEN = IPR["token"]
|
|
API = IPR["api"]
|
|
|
|
_cache = {"products": None, "ts": 0}
|
|
|
|
|
|
def _hdr():
|
|
return {"X-Access-Token": TOKEN, "Content-Type": "application/json"}
|
|
|
|
|
|
def get_balance():
|
|
r = requests.get(f"{API}/balance", headers=_hdr(), timeout=15)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def get_products(force=False):
|
|
"""Returns the raw product list (cached 1h). Each product: id,name,plans[],locations[]."""
|
|
if _cache["products"] and not force and time.time() - _cache["ts"] < 3600:
|
|
return _cache["products"]
|
|
r = requests.get(f"{API}/products", headers=_hdr(), timeout=25)
|
|
r.raise_for_status()
|
|
data = r.json().get("data", [])
|
|
_cache["products"] = data
|
|
_cache["ts"] = time.time()
|
|
return data
|
|
|
|
|
|
def get_product(product_id):
|
|
for p in get_products():
|
|
if p.get("id") == product_id:
|
|
return p
|
|
return None
|
|
|
|
|
|
def list_locations(product_id):
|
|
"""Flatten country -> city into [(location_id, "Country - City", in_stock)]."""
|
|
prod = get_product(product_id)
|
|
out = []
|
|
for country in prod.get("locations", []):
|
|
cid = country.get("id")
|
|
cname = country.get("name")
|
|
kids = country.get("child_locations") or []
|
|
if kids:
|
|
for k in kids:
|
|
out.append((k.get("id"), f"{cname} - {k.get('name')}", not k.get("out_of_stock")))
|
|
else:
|
|
out.append((cid, cname, not country.get("out_of_stock")))
|
|
return out
|
|
|
|
|
|
def create_order(product_id, plan_id, location_id, quantity=1, auto_extend=False):
|
|
body = {
|
|
"product_id": product_id,
|
|
"product_plan_id": plan_id,
|
|
"product_location_id": location_id,
|
|
"quantity": quantity,
|
|
"auto_extend": auto_extend,
|
|
}
|
|
r = requests.post(f"{API}/orders", headers=_hdr(), json=body, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def list_orders(product_id, status=None):
|
|
params = {"product_id": product_id}
|
|
if status:
|
|
params["status"] = status
|
|
r = requests.get(f"{API}/orders", headers=_hdr(), params=params, timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def get_order(order_id):
|
|
r = requests.get(f"{API}/orders/{order_id}", headers=_hdr(), timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()
|