"""Vultr 适配器（REST API v2，Bearer Token） API 文档：https://www.vultr.com/api/ 所需配置：{"api_key": "..."} 支持功能：实例列表、订阅账单、区域查询、实例启停/重启、DNS 记录管理 DNS 的 zone_id 即域名本身（Vultr 用域名字符串定位，非 uuid）。 Vultr 为按小时计费的订阅制，无「续费」概念，故不实现 renew_instance。 """ import httpx from app.adapters.base import ( AccountInfo, BaseAdapter, NormalizedDNSRecord, NormalizedDomain, NormalizedVPS, parse_iso_date, ) from app.adapters.registry import register @register("vultr-api") class VultrAdapter(BaseAdapter): required_config = ["api_key"] BASE = "https://api.vultr.com/v2" def _headers(self) -> dict: return {"Authorization": f"Bearer {self.config.get('api_key', '')}"} def _request(self, method: str, path: str, json_body: dict = None) -> dict: with httpx.Client(timeout=30) as client: resp = client.request(method, self.BASE + path, headers=self._headers(), json=json_body) resp.raise_for_status() # Vultr 的动作类接口成功时返回 204 无 body return resp.json() if resp.content else {} def _get(self, path: str) -> dict: return self._request("GET", path) 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}：API Key 无效或权限不足"} 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", pending_charges=acc.get("pending_charges"), raw=acc, ) def list_vps(self) -> list: result = [] cursor = "" while True: path = "/instances?per_page=100" if cursor: path += f"&cursor={cursor}" data = self._get(path) for inst in data.get("instances", []): ram_mb = inst.get("ram") or 0 result.append( NormalizedVPS( external_id=inst.get("id"), name=inst.get("label") or inst.get("id"), ip_address=inst.get("main_ip"), region=inst.get("region"), os=inst.get("os"), cpu_cores=inst.get("vcpu_count"), memory_gb=round(ram_mb / 1024, 1) if ram_mb else None, disk_gb=inst.get("disk"), status="active" if inst.get("status") == "active" else (inst.get("status") or "unknown"), monthly_cost=inst.get("monthly_cost"), currency="USD", # Vultr 按小时计费、无到期概念，标 postpaid 让前端不去催续费 charge_type="postpaid", raw=inst, ) ) cursor = data.get("meta", {}).get("links", {}).get("next") or "" if not cursor: break return result def list_domains(self) -> list: result = [] cursor = "" while True: path = "/domains?per_page=100" if cursor: path += f"&cursor={cursor}" data = self._get(path) for d in data.get("domains", []): result.append( NormalizedDomain( external_id=d.get("domain"), domain_name=d.get("domain"), registrar="vultr", # 非 Vultr 注册的域名 expiry_date 为空，解析失败统一归为 None expiry_date=parse_iso_date(d.get("expiry_date")), raw=d, ) ) cursor = data.get("meta", {}).get("links", {}).get("next") or "" if not cursor: break return result # ---------------- 实例管理动作 ---------------- def _instance_action(self, external_id: str, verb: str, label: str) -> dict: try: self._request("POST", f"/instances/{external_id}/{verb}") return {"ok": True, "message": f"{label}指令已下发"} except httpx.HTTPStatusError as e: return {"ok": False, "message": f"HTTP {e.response.status_code}：{label}失败"} except Exception as e: # noqa: BLE001 return {"ok": False, "message": f"{label}失败：{e}"} def start_instance(self, external_id: str) -> dict: return self._instance_action(external_id, "start", "开机") def stop_instance(self, external_id: str) -> dict: return self._instance_action(external_id, "halt", "关机") def reboot_instance(self, external_id: str) -> dict: return self._instance_action(external_id, "reboot", "重启") # ---------------- DNS 记录管理 ---------------- def list_dns_records(self, zone_id: str) -> list: result = [] cursor = "" while True: path = f"/domains/{zone_id}/records?per_page=100" if cursor: path += f"&cursor={cursor}" data = self._get(path) for r in data.get("records", []): result.append( NormalizedDNSRecord( record_id=str(r.get("id")), zone_id=zone_id, name=r.get("name") or zone_id, type=r.get("type"), content=r.get("data"), ttl=r.get("ttl"), priority=r.get("priority"), raw=r, ) ) cursor = data.get("meta", {}).get("links", {}).get("next") or "" if not cursor: break return result @staticmethod def _record_body(record: dict) -> dict: body = { "name": record.get("name", ""), "type": record.get("type", "A"), "data": record.get("content", ""), "ttl": record.get("ttl") or 300, } if record.get("priority") is not None: body["priority"] = record["priority"] return body def create_dns_record(self, zone_id: str, record: dict) -> dict: try: data = self._request( "POST", f"/domains/{zone_id}/records", self._record_body(record) ) return { "ok": True, "message": "记录已创建", "record_id": str(data.get("record", {}).get("id", "")), "raw": data, } except httpx.HTTPStatusError as e: return {"ok": False, "message": f"HTTP {e.response.status_code}：创建失败"} except Exception as e: # noqa: BLE001 return {"ok": False, "message": f"创建失败：{e}"} def update_dns_record(self, zone_id: str, record_id: str, record: dict) -> dict: try: self._request( "PATCH", f"/domains/{zone_id}/records/{record_id}", self._record_body(record) ) return {"ok": True, "message": "记录已更新"} except httpx.HTTPStatusError as e: return {"ok": False, "message": f"HTTP {e.response.status_code}：更新失败"} except Exception as e: # noqa: BLE001 return {"ok": False, "message": f"更新失败：{e}"} def delete_dns_record(self, zone_id: str, record_id: str) -> dict: try: self._request("DELETE", f"/domains/{zone_id}/records/{record_id}") return {"ok": True, "message": "记录已删除"} except httpx.HTTPStatusError as e: return {"ok": False, "message": f"HTTP {e.response.status_code}：删除失败"} except Exception as e: # noqa: BLE001 return {"ok": False, "message": f"删除失败：{e}"}