89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""HydraProxy management API wrapper.
|
|
|
|
Base: https://app.hydraproxy.com/api/ (auth: Authorization: Token <key>)
|
|
NOTE: HydraProxy's API is MANAGEMENT-ONLY — it CANNOT place/buy proxy orders.
|
|
Orders are purchased manually from app.hydraproxy.com; this wrapper lists and
|
|
manages them (details, rotation, location, whitelist, usage history).
|
|
|
|
Gotchas (from their docs):
|
|
- All digit POST params must be sent as STRINGS (not ints).
|
|
- POST URLs must end with a trailing slash.
|
|
- API only accepts calls from whitelisted IPs (up to 3, manual, 30-min lockdown).
|
|
- First item of every response is `status` ("OK" or "ERROR").
|
|
"""
|
|
import json
|
|
import os
|
|
|
|
import requests
|
|
|
|
CFG = json.load(open(os.environ.get("PLEIADES_CONFIG", "/opt/pleiades/config.json")))
|
|
HP = CFG["hydraproxy"]
|
|
TOKEN = HP["token"]
|
|
API = HP["api"].rstrip("/")
|
|
|
|
|
|
def _hdr():
|
|
return {"Authorization": f"Token {TOKEN}", "Accept": "application/json"}
|
|
|
|
|
|
def get_account_info():
|
|
"""{status, username, email, balance_usd}"""
|
|
r = requests.get(f"{API}/get-account-info/", headers=_hdr(), timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def get_locations():
|
|
"""List of available geo locations."""
|
|
r = requests.get(f"{API}/get-locations-list/", headers=_hdr(), timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def proxy_details(order_id):
|
|
"""Details of one order (network, proxy_info, proxy host/port)."""
|
|
r = requests.post(f"{API}/proxy-details/", headers=_hdr(),
|
|
data={"order_id": str(order_id)}, timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def force_rotation(order_id):
|
|
r = requests.post(f"{API}/force-rotation/", headers=_hdr(),
|
|
data={"order_id": str(order_id)}, timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def update_location(order_id, location):
|
|
r = requests.post(f"{API}/update-location/", headers=_hdr(),
|
|
data={"order_id": str(order_id), "location": str(location)}, timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def update_rotation(order_id, auto=True):
|
|
r = requests.post(f"{API}/update-rotation/", headers=_hdr(),
|
|
data={"order_id": str(order_id), "auto": "1" if auto else "0"}, timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def update_whitelist_ip(order_id, ip):
|
|
r = requests.post(f"{API}/update-whitelist-ip/", headers=_hdr(),
|
|
data={"order_id": str(order_id), "whitelist_ip": str(ip)}, timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def view_usage_history(order_id, date_from="", date_to=""):
|
|
data = {"order_id": str(order_id)}
|
|
if date_from:
|
|
data["date_from"] = date_from
|
|
if date_to:
|
|
data["date_to"] = date_to
|
|
r = requests.post(f"{API}/view-usage-history/", headers=_hdr(), data=data, timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()
|