vps-manager/app/adapters/base.py

"""云厂商 SDK 适配层 — 抽象基类与标准化数据结构

各云厂商适配器继承 BaseAdapter，实现统一接口，将平台特定的 API 返回
转换为标准化的资产结构（NormalizedVPS / NormalizedDomain / AccountInfo），
供同步服务写入数据库。这样上层逻辑无需关心各平台 API 差异。
"""

from abc import ABC, abstractmethod
from dataclasses import asdict, dataclass, field
from datetime import date, datetime
from typing import Any, Optional


def parse_iso_date(value: Any) -> Optional[date]:
    """把到期日归一化为 date 对象（接受 date / datetime / ISO 字符串，解析失败返回 None）

    各厂商返回形态不一（'2027-03-15'、'2027-03-15T00:00:00Z'、'2027-03-15T00:00:00+00:00'）。
    必须统一成 date 再交给上层：expiry_date 是 SQLAlchemy Date 列，
    直接塞字符串会在写库时抛 TypeError，导致整批同步在循环末尾的 commit 上全量失败。
    """
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.date()
    if isinstance(value, date):
        return value
    text = str(value).strip()
    if not text:
        return None
    try:
        return date.fromisoformat(text[:10])
    except ValueError:
        return None


@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[date] = None  # 到期日 date；厂商原始值请过 parse_iso_date
    status: str = "active"
    raw: dict = field(default_factory=dict)

    def to_dict(self) -> dict:
        d = asdict(self)
        # date 不可直接 JSON 序列化，转回 ISO 字符串（对外形态与旧契约一致）
        if d.get("expiry_date") is not None:
            d["expiry_date"] = d["expiry_date"].isoformat()
        return d


@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"),
        }