102 lines
3.6 KiB
Python
102 lines
3.6 KiB
Python
"""DigitalOcean 适配器（REST API v2，Bearer Token）
|
||
|
||
API 文档：https://docs.digitalocean.com/reference/api/
|
||
所需配置：{"api_key": "..."}
|
||
"""
|
||
|
||
import httpx
|
||
|
||
from app.adapters.base import (
|
||
AccountInfo,
|
||
BaseAdapter,
|
||
NormalizedDomain,
|
||
NormalizedVPS,
|
||
)
|
||
from app.adapters.registry import register
|
||
|
||
|
||
@register("do-api")
|
||
class DigitalOceanAdapter(BaseAdapter):
|
||
required_config = ["api_key"]
|
||
BASE = "https://api.digitalocean.com/v2"
|
||
|
||
def _headers(self) -> dict:
|
||
return {"Authorization": f"Bearer {self.config.get('api_key', '')}"}
|
||
|
||
def _get(self, path: str) -> dict:
|
||
with httpx.Client(timeout=30) as client:
|
||
resp = client.get(self.BASE + path, headers=self._headers())
|
||
resp.raise_for_status()
|
||
return resp.json()
|
||
|
||
def test_connection(self) -> dict:
|
||
try:
|
||
data = self._get("/account")
|
||
balance = data.get("account", {}).get("balance")
|
||
return {"ok": True, "message": f"连接成功，账户余额 {balance} USD"}
|
||
except httpx.HTTPStatusError as e:
|
||
return {"ok": False, "message": f"HTTP {e.response.status_code}：Token 无效或权限不足"}
|
||
except Exception as e: # noqa: BLE001
|
||
return {"ok": False, "message": str(e)}
|
||
|
||
def get_account(self) -> AccountInfo:
|
||
acc = self._get("/account").get("account", {})
|
||
return AccountInfo(balance=acc.get("balance"), currency="USD", raw=acc)
|
||
|
||
@staticmethod
|
||
def _main_ip(droplet: dict) -> str:
|
||
try:
|
||
v4 = droplet.get("networks", {}).get("v4", [])
|
||
return v4[0].get("ip_address") if v4 else None
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
|
||
def list_vps(self) -> list:
|
||
result = []
|
||
page = 1
|
||
while True:
|
||
data = self._get(f"/droplets?per_page=100&page={page}")
|
||
for d in data.get("droplets", []):
|
||
image = d.get("image", {})
|
||
os_name = f"{image.get('distribution', '')} {image.get('name', '')}".strip()
|
||
mem_mb = d.get("memory") or 0
|
||
region = d.get("region", {})
|
||
result.append(
|
||
NormalizedVPS(
|
||
external_id=str(d.get("id")),
|
||
name=d.get("name"),
|
||
ip_address=self._main_ip(d),
|
||
region=region.get("slug") or region.get("name"),
|
||
os=os_name or None,
|
||
cpu_cores=d.get("vcpus"),
|
||
memory_gb=round(mem_mb / 1024, 1) if mem_mb else None,
|
||
disk_gb=d.get("disk"),
|
||
status="active" if d.get("status") == "active" else (d.get("status") or "unknown"),
|
||
currency="USD",
|
||
raw=d,
|
||
)
|
||
)
|
||
if not data.get("links", {}).get("pages", {}).get("next"):
|
||
break
|
||
page += 1
|
||
return result
|
||
|
||
def list_domains(self) -> list:
|
||
result = []
|
||
page = 1
|
||
while True:
|
||
data = self._get(f"/domains?per_page=100&page={page}")
|
||
for d in data.get("domains", []):
|
||
result.append(
|
||
NormalizedDomain(
|
||
external_id=d.get("name"),
|
||
domain_name=d.get("name"),
|
||
registrar="digitalocean",
|
||
raw=d,
|
||
)
|
||
)
|
||
if not data.get("links", {}).get("pages", {}).get("next"):
|
||
break
|
||
page += 1
|
||
return result
|