"""云厂商 SDK 适配层 — 抽象基类与标准化数据结构 各云厂商适配器继承 BaseAdapter,实现统一接口,将平台特定的 API 返回 转换为标准化的资产结构(NormalizedVPS / NormalizedDomain / AccountInfo), 供同步服务写入数据库。这样上层逻辑无需关心各平台 API 差异。 """ from abc import ABC, abstractmethod from dataclasses import asdict, dataclass, field from typing import Optional @dataclass class NormalizedVPS: """标准化的 VPS 实例""" external_id: str name: str ip_address: Optional[str] = None region: Optional[str] = None os: Optional[str] = None cpu_cores: Optional[int] = None memory_gb: Optional[float] = None disk_gb: Optional[int] = None status: str = "active" monthly_cost: Optional[float] = None currency: str = "USD" raw: dict = field(default_factory=dict) def to_dict(self) -> dict: return asdict(self) @dataclass class NormalizedDomain: """标准化的域名""" external_id: str domain_name: str registrar: Optional[str] = None expiry_date: Optional[str] = None # ISO 日期字符串 YYYY-MM-DD status: str = "active" raw: dict = field(default_factory=dict) def to_dict(self) -> dict: return asdict(self) @dataclass class AccountInfo: """标准化的账户信息""" balance: Optional[float] = None currency: str = "USD" pending_charges: Optional[float] = None raw: dict = field(default_factory=dict) def to_dict(self) -> dict: return asdict(self) def _overridden(instance: "BaseAdapter", method_name: str) -> bool: """判断子类是否重写了某方法(用于能力探测)""" return type(instance).__dict__.get(method_name) is not None class BaseAdapter(ABC): """云厂商适配器抽象基类 子类需: - 通过 @register("xxx-sdk") 注册到注册表 - 设置 required_config(所需凭证字段,供前端表单提示) - 实现 test_connection;按需重写 list_vps / list_domains / get_account """ #: 适配器所需的配置字段名(如 ["api_key"] 或 ["access_key_id", "access_key_secret"]) required_config: list = [] def __init__(self, config: Optional[dict] = None): self.config = config or {} @abstractmethod def test_connection(self) -> dict: """测试连接 / 凭证有效性,返回 {"ok": bool, "message": str, ...}""" def list_vps(self) -> list: raise NotImplementedError(f"{type(self).__name__} 未实现 list_vps") def list_domains(self) -> list: raise NotImplementedError(f"{type(self).__name__} 未实现 list_domains") def get_account(self) -> AccountInfo: raise NotImplementedError(f"{type(self).__name__} 未实现 get_account") def capabilities(self) -> dict: """返回适配器实际支持的能力(是否重写了相应方法)""" return { "list_vps": _overridden(self, "list_vps"), "list_domains": _overridden(self, "list_domains"), "get_account": _overridden(self, "get_account"), }