From 1aacf51bea1f9661e6bdc50f6bd59dd4948c1c11 Mon Sep 17 00:00:00 2001 From: gouki Date: Sun, 2 Aug 2026 15:56:48 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BA=91=E5=8E=82=E5=95=86SDK=E9=80=82?= =?UTF-8?q?=E9=85=8D=E5=B1=82=EF=BC=88Vultr/DO/Cloudflare/=E9=98=BF?= =?UTF-8?q?=E9=87=8C=E4=BA=91/=E8=85=BE=E8=AE=AF=E4=BA=91=EF=BC=89+=20?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E6=9C=8D=E5=8A=A1=20+=20=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E8=BF=9E=E6=8E=A5/=E5=90=8C=E6=AD=A5=E6=8E=A5=E5=8F=A3?= =?UTF-8?q?=E4=B8=8E=E5=89=8D=E7=AB=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/adapters/__init__.py | 5 + app/adapters/aliyun.py | 103 ++++++++++++++++ app/adapters/base.py | 101 ++++++++++++++++ app/adapters/cloudflare.py | 57 +++++++++ app/adapters/digitalocean.py | 89 ++++++++++++++ app/adapters/registry.py | 72 ++++++++++++ app/adapters/tencent.py | 117 +++++++++++++++++++ app/adapters/vultr.py | 84 +++++++++++++ app/database.py | 13 +++ app/models/asset.py | 3 + app/routers/providers.py | 29 ++++- app/services/sync_service.py | 220 +++++++++++++++++++++++++++++++++++ static/js/app.js | 40 ++++++- 13 files changed, 930 insertions(+), 3 deletions(-) create mode 100644 app/adapters/__init__.py create mode 100644 app/adapters/aliyun.py create mode 100644 app/adapters/base.py create mode 100644 app/adapters/cloudflare.py create mode 100644 app/adapters/digitalocean.py create mode 100644 app/adapters/registry.py create mode 100644 app/adapters/tencent.py create mode 100644 app/adapters/vultr.py create mode 100644 app/services/sync_service.py diff --git a/app/adapters/__init__.py b/app/adapters/__init__.py new file mode 100644 index 0000000..21961a2 --- /dev/null +++ b/app/adapters/__init__.py @@ -0,0 +1,5 @@ +"""云厂商 SDK 适配层 + +通过 registry.get_adapter(sdk_type, config) 获取对应平台适配器， +调用 test_connection / list_vps / list_domains / get_account 等统一接口。 +""" diff --git a/app/adapters/aliyun.py b/app/adapters/aliyun.py new file mode 100644 index 0000000..88015d7 --- /dev/null +++ b/app/adapters/aliyun.py @@ -0,0 +1,103 @@ +"""阿里云适配器（含国际版，OpenAPI RPC 风格 + HMAC-SHA1 签名） + +API 文档：https://help.aliyun.com/document_detail/25484.html +所需配置：{"access_key_id": "...", "access_key_secret": "...", "region": "cn-hangzhou"} +国际版（alibabacloud）只需将 region 设为海外区域（如 ap-southeast-1），端点自动按区域构造。 +说明：此处实现 RPC V1 签名（HMAC-SHA1），可直接调用 ECS OpenAPI； + 生产环境也可替换为官方 alibabacloud-ecs SDK，接口契约保持不变。 +""" + +import base64 +import hashlib +import hmac +import time +import uuid +from urllib.parse import quote + +import httpx + +from app.adapters.base import BaseAdapter, NormalizedVPS +from app.adapters.registry import register + + +@register("aliyun-sdk", "alibabacloud-sdk") +class AliyunAdapter(BaseAdapter): + required_config = ["access_key_id", "access_key_secret"] + API_VERSION = "2014-05-26" + + def _region(self) -> str: + return self.config.get("region") or "cn-hangzhou" + + def _endpoint(self) -> str: + return f"https://ecs.{self._region()}.aliyuncs.com" + + @staticmethod + def _percent_encode(value) -> str: + return quote(str(value), safe="~") + + def _sign(self, params: dict, method: str = "GET") -> str: + canonical = "&".join( + f"{self._percent_encode(k)}={self._percent_encode(v)}" + for k, v in sorted(params.items()) + ) + string_to_sign = f"{method}&{self._percent_encode('/')}&{self._percent_encode(canonical)}" + key = (self.config.get("access_key_secret", "") + "&").encode() + digest = hmac.new(key, string_to_sign.encode(), hashlib.sha1).digest() + return base64.b64encode(digest).decode() + + def _call(self, action: str, biz_params: dict = None) -> dict: + params = { + "Format": "JSON", + "Version": self.API_VERSION, + "AccessKeyId": self.config.get("access_key_id", ""), + "SignatureMethod": "HMAC-SHA1", + "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "SignatureVersion": "1.0", + "SignatureNonce": str(uuid.uuid4()), + "Action": action, + "RegionId": self._region(), + } + if biz_params: + params.update(biz_params) + params["Signature"] = self._sign(params) + with httpx.Client(timeout=30) as client: + resp = client.get(self._endpoint(), params=params) + resp.raise_for_status() + data = resp.json() + if data.get("Code"): + raise RuntimeError(f"{data.get('Code')}: {data.get('Message')}") + return data + + def test_connection(self) -> dict: + try: + data = self._call("DescribeRegions") + count = len(data.get("Regions", {}).get("Region", [])) + return {"ok": True, "message": f"连接成功，可用区域 {count} 个"} + 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": str(e)} + + def list_vps(self) -> list: + data = self._call("DescribeInstances", {"PageSize": "100"}) + result = [] + for inst in data.get("Instances", {}).get("Instance", []): + public_ips = inst.get("PublicIpAddress", {}).get("IpAddress", []) + eip = inst.get("EipAddress", {}).get("IpAddress") + mem_mb = inst.get("Memory") or 0 + status_map = {"Running": "active", "Stopped": "stopped"} + result.append( + NormalizedVPS( + external_id=inst.get("InstanceId"), + name=inst.get("InstanceName") or inst.get("InstanceId"), + ip_address=eip or (public_ips[0] if public_ips else None), + region=inst.get("RegionId"), + os=inst.get("OSName"), + cpu_cores=inst.get("Cpu"), + memory_gb=round(mem_mb / 1024, 1) if mem_mb else None, + status=status_map.get(inst.get("Status"), inst.get("Status") or "unknown"), + currency="CNY", + raw=inst, + ) + ) + return result diff --git a/app/adapters/base.py b/app/adapters/base.py new file mode 100644 index 0000000..6aa75c0 --- /dev/null +++ b/app/adapters/base.py @@ -0,0 +1,101 @@ +"""云厂商 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"), + } diff --git a/app/adapters/cloudflare.py b/app/adapters/cloudflare.py new file mode 100644 index 0000000..e0b9b83 --- /dev/null +++ b/app/adapters/cloudflare.py @@ -0,0 +1,57 @@ +"""Cloudflare 适配器（REST API v4，Bearer Token） + +API 文档：https://developers.cloudflare.com/api/ +所需配置：{"api_token": "..."}（建议用 API Token，权限含 Zone:Read / Account:Read） +Cloudflare 无传统 VPS，主要同步托管域名（zones）；Workers/R2/Tunnel 子资产留待后续阶段。 +""" + +import httpx + +from app.adapters.base import BaseAdapter, NormalizedDomain +from app.adapters.registry import register + + +@register("cloudflare-api") +class CloudflareAdapter(BaseAdapter): + required_config = ["api_token"] + BASE = "https://api.cloudflare.com/client/v4" + + def _headers(self) -> dict: + return {"Authorization": f"Bearer {self.config.get('api_token', '')}"} + + 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() + data = resp.json() + if not data.get("success", True): + errors = data.get("errors", []) + raise RuntimeError(errors[0].get("message") if errors else "Cloudflare API 返回失败") + return data + + def test_connection(self) -> dict: + try: + data = self._get("/user/tokens/verify") + status = data.get("result", {}).get("status") + if status == "active": + return {"ok": True, "message": "API Token 有效"} + return {"ok": False, "message": f"Token 状态异常：{status}"} + 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 list_domains(self) -> list: + data = self._get("/zones?per_page=50") + result = [] + for z in data.get("result", []): + result.append( + NormalizedDomain( + external_id=z.get("id"), + domain_name=z.get("name"), + registrar="cloudflare", + status="active" if z.get("status") == "active" else (z.get("status") or "unknown"), + raw=z, + ) + ) + return result diff --git a/app/adapters/digitalocean.py b/app/adapters/digitalocean.py new file mode 100644 index 0000000..bdc21a2 --- /dev/null +++ b/app/adapters/digitalocean.py @@ -0,0 +1,89 @@ +"""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: + data = self._get("/droplets") + result = [] + 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, + ) + ) + return result + + def list_domains(self) -> list: + data = self._get("/domains") + return [ + NormalizedDomain( + external_id=d.get("name"), + domain_name=d.get("name"), + registrar="digitalocean", + raw=d, + ) + for d in data.get("domains", []) + ] diff --git a/app/adapters/registry.py b/app/adapters/registry.py new file mode 100644 index 0000000..a3669b2 --- /dev/null +++ b/app/adapters/registry.py @@ -0,0 +1,72 @@ +"""适配器注册表与工厂 + +通过 @register("sdk-type") 装饰器将适配器类登记到注册表， +上层按 Provider.sdk_type 取得对应适配器实例。 +""" + +from typing import Optional + +from app.adapters.base import BaseAdapter + +_REGISTRY: dict = {} + + +def register(*sdk_types: str): + """装饰器：将适配器类注册到一个或多个 sdk_type""" + + def decorator(cls): + for t in sdk_types: + _REGISTRY[t] = cls + cls.sdk_types = list(sdk_types) + return cls + + return decorator + + +def load_all() -> None: + """导入所有适配器模块以触发注册（幂等）""" + from app.adapters import ( # noqa: F401 + aliyun, + cloudflare, + digitalocean, + tencent, + vultr, + ) + + +def get_adapter(sdk_type: str, config: Optional[dict] = None) -> BaseAdapter: + """按 sdk_type 创建适配器实例""" + load_all() + cls = _REGISTRY.get(sdk_type) + if not cls: + raise ValueError(f"未支持的 SDK 类型：{sdk_type}") + return cls(config) + + +def is_supported(sdk_type: Optional[str]) -> bool: + load_all() + return sdk_type in _REGISTRY + + +def supported_types() -> list: + load_all() + return sorted(_REGISTRY.keys()) + + +def adapter_meta() -> list: + """返回所有已注册适配器的元信息（供前端展示所需凭证字段与能力）""" + load_all() + seen = set() + result = [] + for sdk_type, cls in sorted(_REGISTRY.items()): + if cls.__name__ in seen: + continue + seen.add(cls.__name__) + result.append( + { + "class": cls.__name__, + "sdk_types": getattr(cls, "sdk_types", [sdk_type]), + "required_config": getattr(cls, "required_config", []), + } + ) + return result diff --git a/app/adapters/tencent.py b/app/adapters/tencent.py new file mode 100644 index 0000000..ea5e052 --- /dev/null +++ b/app/adapters/tencent.py @@ -0,0 +1,117 @@ +"""腾讯云适配器（含国际版，CVM OpenAPI + TC3-HMAC-SHA256 签名） + +API 文档：https://cloud.tencent.com/document/api/213/15753 +所需配置：{"secret_id": "...", "secret_key": "...", "region": "ap-guangzhou"} +国际版（intl）将 region 设为海外区域即可，端点统一为 cvm.tencentcloudapi.com。 +说明：此处实现 TC3-HMAC-SHA256 签名，可直接调用 CVM OpenAPI； + 生产环境也可替换为官方 tencentcloud-sdk-python，接口契约保持不变。 +""" + +import hashlib +import hmac +import json +import time + +import httpx + +from app.adapters.base import BaseAdapter, NormalizedVPS +from app.adapters.registry import register + + +@register("tencent-sdk", "tencent-intl-sdk") +class TencentAdapter(BaseAdapter): + required_config = ["secret_id", "secret_key"] + HOST = "cvm.tencentcloudapi.com" + SERVICE = "cvm" + VERSION = "2017-03-12" + ALGORITHM = "TC3-HMAC-SHA256" + + def _region(self) -> str: + return self.config.get("region") or "ap-guangzhou" + + @staticmethod + def _hmac_sha256(key: bytes, msg: str) -> bytes: + return hmac.new(key, msg.encode(), hashlib.sha256).digest() + + def _build_authorization(self, action: str, payload_json: str, timestamp: int, date: str): + secret_id = self.config.get("secret_id", "") + secret_key = self.config.get("secret_key", "") + content_type = "application/json; charset=utf-8" + canonical_headers = ( + f"content-type:{content_type}\nhost:{self.HOST}\nx-tc-action:{action.lower()}\n" + ) + signed_headers = "content-type;host;x-tc-action" + hashed_payload = hashlib.sha256(payload_json.encode()).hexdigest() + canonical_request = ( + f"POST\n/\n\n{canonical_headers}\n{signed_headers}\n{hashed_payload}" + ) + credential_scope = f"{date}/{self.SERVICE}/tc3_request" + hashed_canonical = hashlib.sha256(canonical_request.encode()).hexdigest() + string_to_sign = ( + f"{self.ALGORITHM}\n{timestamp}\n{credential_scope}\n{hashed_canonical}" + ) + secret_date = self._hmac_sha256(("TC3" + secret_key).encode(), date) + secret_service = self._hmac_sha256(secret_date, self.SERVICE) + secret_signing = self._hmac_sha256(secret_service, "tc3_request") + signature = hmac.new(secret_signing, string_to_sign.encode(), hashlib.sha256).hexdigest() + return ( + f"{self.ALGORITHM} Credential={secret_id}/{credential_scope}, " + f"SignedHeaders={signed_headers}, Signature={signature}" + ) + + def _call(self, action: str, payload: dict = None) -> dict: + payload_json = json.dumps(payload or {}) + timestamp = int(time.time()) + date = time.strftime("%Y-%m-%d", time.gmtime(timestamp)) + authorization = self._build_authorization(action, payload_json, timestamp, date) + headers = { + "Authorization": authorization, + "Content-Type": "application/json; charset=utf-8", + "Host": self.HOST, + "X-TC-Action": action, + "X-TC-Timestamp": str(timestamp), + "X-TC-Version": self.VERSION, + "X-TC-Region": self._region(), + } + with httpx.Client(timeout=30) as client: + resp = client.post(f"https://{self.HOST}", content=payload_json, headers=headers) + resp.raise_for_status() + data = resp.json() + response = data.get("Response", {}) + if response.get("Error"): + err = response["Error"] + raise RuntimeError(f"{err.get('Code')}: {err.get('Message')}") + return response + + def test_connection(self) -> dict: + try: + data = self._call("DescribeRegions") + count = len(data.get("RegionSet", [])) + return {"ok": True, "message": f"连接成功，可用区域 {count} 个"} + 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": str(e)} + + def list_vps(self) -> list: + data = self._call("DescribeInstances", {"Limit": 100}) + result = [] + status_map = {"RUNNING": "active", "STOPPED": "stopped"} + for inst in data.get("InstanceSet", []): + public_ips = inst.get("PublicIpAddresses", []) + result.append( + NormalizedVPS( + external_id=inst.get("InstanceId"), + name=inst.get("InstanceName") or inst.get("InstanceId"), + ip_address=public_ips[0] if public_ips else None, + region=inst.get("Placement", {}).get("Zone"), + os=inst.get("OsName"), + cpu_cores=inst.get("CPU"), + memory_gb=inst.get("Memory"), + disk_gb=inst.get("SystemDisk", {}).get("DiskSize"), + status=status_map.get(inst.get("InstanceState"), inst.get("InstanceState") or "unknown"), + currency="CNY", + raw=inst, + ) + ) + return result diff --git a/app/adapters/vultr.py b/app/adapters/vultr.py new file mode 100644 index 0000000..435f066 --- /dev/null +++ b/app/adapters/vultr.py @@ -0,0 +1,84 @@ +"""Vultr 适配器（REST API v2，Bearer Token） + +API 文档：https://www.vultr.com/api/ +所需配置：{"api_key": "..."} +""" + +import httpx + +from app.adapters.base import ( + AccountInfo, + BaseAdapter, + NormalizedDomain, + NormalizedVPS, +) +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 _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}：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: + data = self._get("/instances") + result = [] + 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", + raw=inst, + ) + ) + return result + + def list_domains(self) -> list: + data = self._get("/domains") + return [ + NormalizedDomain( + external_id=d.get("domain"), + domain_name=d.get("domain"), + registrar="vultr", + raw=d, + ) + for d in data.get("domains", []) + ] diff --git a/app/database.py b/app/database.py index 86fbab3..57d82e6 100644 --- a/app/database.py +++ b/app/database.py @@ -47,6 +47,19 @@ def init_db() -> None: SQLModel.metadata.create_all( metrics_engine, tables=[m.__table__ for m in metric_models] ) + _migrate_assets_db() + + +def _migrate_assets_db() -> None: + """轻量迁移：为已有 assets 表补充新增列（SQLite create_all 不会修改已有表结构）""" + import sqlalchemy as sa + + with assets_engine.begin() as conn: + if not sa.inspect(conn).has_table("assets"): + return + cols = {c["name"] for c in sa.inspect(conn).get_columns("assets")} + if "external_id" not in cols: + conn.execute(sa.text("ALTER TABLE assets ADD COLUMN external_id VARCHAR")) def get_session() -> Generator[Session, None, None]: diff --git a/app/models/asset.py b/app/models/asset.py index dbd71b8..9d19639 100644 --- a/app/models/asset.py +++ b/app/models/asset.py @@ -46,6 +46,9 @@ class Asset(SQLModel, table=True): provider_id: Optional[int] = Field( default=None, foreign_key="providers.id", index=True, description="关联平台 ID" ) + external_id: Optional[str] = Field( + default=None, index=True, description="平台内实例 ID（用于 SDK 同步去重）" + ) renewal_cycle: Optional[str] = Field( default=None, description="续费周期：monthly/quarterly/yearly" ) diff --git a/app/routers/providers.py b/app/routers/providers.py index 5c97478..185e10c 100644 --- a/app/routers/providers.py +++ b/app/routers/providers.py @@ -10,7 +10,8 @@ from app.core.seed import seed_providers from app.database import get_session from app.models.provider import ProviderCategory from app.schemas.provider import ProviderCreate, ProviderRead, ProviderUpdate -from app.services import provider_service +from app.adapters import registry +from app.services import provider_service, sync_service router = APIRouter(prefix="/api/providers", tags=["providers"]) @@ -25,6 +26,14 @@ def seed(session: Session = Depends(get_session)) -> dict: return {"added": added} +@router.get("/adapters", summary="已支持的 SDK 适配器元信息") +def adapters() -> dict: + return { + "adapters": registry.adapter_meta(), + "supported_types": registry.supported_types(), + } + + @router.get("", response_model=List[ProviderRead], summary="平台列表") def list_providers( category: Optional[ProviderCategory] = Query(default=None), @@ -70,3 +79,21 @@ def update_provider( ) def delete_provider(provider_id: int, session: Session = Depends(get_session)): provider_service.delete_provider(session, provider_id) + + +@router.post( + "/{provider_id}/test", + summary="测试平台连接 / 凭证有效性", + dependencies=[Depends(require_api_key)], +) +def test_connection(provider_id: int, session: Session = Depends(get_session)) -> dict: + return sync_service.test_provider(session, provider_id) + + +@router.post( + "/{provider_id}/sync", + summary="同步平台资产到本地库", + dependencies=[Depends(require_api_key)], +) +def sync_assets(provider_id: int, session: Session = Depends(get_session)) -> dict: + return sync_service.sync_provider(session, provider_id) diff --git a/app/services/sync_service.py b/app/services/sync_service.py new file mode 100644 index 0000000..3342595 --- /dev/null +++ b/app/services/sync_service.py @@ -0,0 +1,220 @@ +"""云厂商同步服务 + +将适配器返回的标准化资产（NormalizedVPS / NormalizedDomain / AccountInfo） +写入或更新到资产库。以 (provider_id, external_id) 作为去重键， +已存在则更新状态/详情，不存在则新建资产。 +""" + +import json +from typing import Optional + +from fastapi import HTTPException, status +from sqlmodel import Session, select + +from app.adapters import registry +from app.adapters.base import BaseAdapter +from app.core import crypto +from app.models.asset import ( + Asset, + AssetStatus, + AssetType, + DomainDetail, + VPSDetail, +) +from app.models.provider import Provider + +_VALID_STATUS = {s.value for s in AssetStatus} + + +def _load_config(provider: Provider) -> dict: + """解密平台的 API 配置 JSON""" + plain = crypto.decrypt(provider.api_config_encrypted) + if not plain: + return {} + try: + return json.loads(plain) + except (json.JSONDecodeError, TypeError): + return {} + + +def _get_provider(session: Session, provider_id: int) -> Provider: + provider = session.get(Provider, provider_id) + if not provider: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="平台不存在") + return provider + + +def _build_adapter(provider: Provider) -> BaseAdapter: + if not provider.sdk_type: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, detail="该平台未配置 SDK 类型（sdk_type）" + ) + if not registry.is_supported(provider.sdk_type): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"暂不支持的 SDK 类型：{provider.sdk_type}", + ) + return registry.get_adapter(provider.sdk_type, _load_config(provider)) + + +def _norm_status(raw: Optional[str]) -> AssetStatus: + return AssetStatus(raw) if raw in _VALID_STATUS else AssetStatus.UNKNOWN + + +def _missing_config(adapter: BaseAdapter) -> list: + """检查适配器所需凭证是否已配置""" + return [f for f in adapter.required_config if not adapter.config.get(f)] + + +def test_provider(session: Session, provider_id: int) -> dict: + """测试平台连接 / 凭证有效性""" + provider = _get_provider(session, provider_id) + adapter = _build_adapter(provider) + base = {"capabilities": adapter.capabilities(), "sdk_type": provider.sdk_type} + missing = _missing_config(adapter) + if missing: + return { + "ok": False, + "message": f"缺少凭证配置：{', '.join(missing)}（请在平台编辑里填写 API 配置）", + **base, + } + result = adapter.test_connection() + result.update(base) + return result + + +def _find_asset(session: Session, provider_id: int, external_id: str, asset_type: AssetType): + return session.exec( + select(Asset).where( + Asset.provider_id == provider_id, + Asset.external_id == external_id, + Asset.asset_type == asset_type, + ) + ).first() + + +def _sync_vps(session: Session, provider: Provider, adapter: BaseAdapter) -> dict: + created = updated = 0 + for vps in adapter.list_vps(): + existing = _find_asset(session, provider.id, vps.external_id, AssetType.VPS) + if existing: + existing.name = vps.name or existing.name + existing.status = _norm_status(vps.status) + if vps.monthly_cost is not None: + existing.cost = vps.monthly_cost + existing.currency = vps.currency + session.add(existing) + detail = session.exec( + select(VPSDetail).where(VPSDetail.asset_id == existing.id) + ).first() + if detail: + detail.ip_address = vps.ip_address or detail.ip_address + detail.region = vps.region or detail.region + detail.os = vps.os or detail.os + detail.cpu_cores = vps.cpu_cores or detail.cpu_cores + detail.memory_gb = vps.memory_gb or detail.memory_gb + detail.disk_gb = vps.disk_gb or detail.disk_gb + session.add(detail) + updated += 1 + else: + asset = Asset( + name=vps.name or vps.external_id, + asset_type=AssetType.VPS, + provider=provider.slug, + provider_id=provider.id, + external_id=vps.external_id, + status=_norm_status(vps.status), + cost=vps.monthly_cost or 0, + currency=vps.currency, + ) + session.add(asset) + session.commit() + session.refresh(asset) + session.add( + VPSDetail( + asset_id=asset.id, + ip_address=vps.ip_address or "", + region=vps.region, + os=vps.os, + cpu_cores=vps.cpu_cores or 1, + memory_gb=vps.memory_gb or 1, + disk_gb=vps.disk_gb or 20, + ) + ) + created += 1 + session.commit() + return {"created": created, "updated": updated} + + +def _sync_domains(session: Session, provider: Provider, adapter: BaseAdapter) -> dict: + created = updated = 0 + for dom in adapter.list_domains(): + existing = _find_asset(session, provider.id, dom.external_id, AssetType.DOMAIN) + if existing: + existing.status = _norm_status(dom.status) + if dom.expiry_date: + existing.expiry_date = dom.expiry_date + session.add(existing) + detail = session.exec( + select(DomainDetail).where(DomainDetail.asset_id == existing.id) + ).first() + if detail: + detail.domain_name = dom.domain_name or detail.domain_name + detail.registrar = dom.registrar or detail.registrar + session.add(detail) + updated += 1 + else: + asset = Asset( + name=dom.domain_name, + asset_type=AssetType.DOMAIN, + provider=provider.slug, + provider_id=provider.id, + external_id=dom.external_id, + status=_norm_status(dom.status), + expiry_date=dom.expiry_date, + ) + session.add(asset) + session.commit() + session.refresh(asset) + session.add( + DomainDetail( + asset_id=asset.id, + domain_name=dom.domain_name, + registrar=dom.registrar or provider.slug, + ) + ) + created += 1 + session.commit() + return {"created": created, "updated": updated} + + +def sync_provider(session: Session, provider_id: int) -> dict: + """同步平台资产到本地库""" + provider = _get_provider(session, provider_id) + adapter = _build_adapter(provider) + missing = _missing_config(adapter) + if missing: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"缺少凭证配置：{', '.join(missing)}（请先在平台编辑里填写 API 配置）", + ) + caps = adapter.capabilities() + result = {"provider": provider.slug, "sdk_type": provider.sdk_type} + + if caps["list_vps"]: + try: + result["vps"] = _sync_vps(session, provider, adapter) + except Exception as e: # noqa: BLE001 + result["vps_error"] = str(e) + if caps["list_domains"]: + try: + result["domains"] = _sync_domains(session, provider, adapter) + except Exception as e: # noqa: BLE001 + result["domains_error"] = str(e) + if caps["get_account"]: + try: + result["account"] = adapter.get_account().to_dict() + except Exception as e: # noqa: BLE001 + result["account_error"] = str(e) + + return result diff --git a/static/js/app.js b/static/js/app.js index a1d4623..16e3d07 100644 --- a/static/js/app.js +++ b/static/js/app.js @@ -316,14 +316,50 @@ const ProvidersView = { {{ p.enabled ? '启用' : '停用' }} -
+
+ +
+
{{ results[p.id].text }}
`, - setup() { return { store, Fmt, seed: seedProviders, create: openProviderCreate, edit: openProviderEdit, del: deleteProvider }; }, + setup() { + const testing = reactive({}); + const syncing = reactive({}); + const results = reactive({}); + const supported = ref([]); + onMounted(async () => { + try { const r = await Api.get('/providers/adapters'); supported.value = r.supported_types || []; } catch (e) { /* ignore */ } + }); + function isSupported(t) { return !!t && supported.value.includes(t); } + async function test(p) { + testing[p.id] = true; delete results[p.id]; + try { + const r = await Api.post('/providers/' + p.id + '/test', {}); + results[p.id] = { ok: r.ok, text: (r.ok ? '✓ ' : '✗ ') + r.message }; + } catch (e) { results[p.id] = { ok: false, text: '✗ ' + e.message }; } + finally { testing[p.id] = false; } + } + async function sync(p) { + syncing[p.id] = true; delete results[p.id]; + try { + const r = await Api.post('/providers/' + p.id + '/sync', {}); + const parts = []; + if (r.vps) parts.push('VPS 新增' + r.vps.created + '/更新' + r.vps.updated); + if (r.domains) parts.push('域名 新增' + r.domains.created + '/更新' + r.domains.updated); + if (r.vps_error) parts.push('VPS错误:' + r.vps_error); + if (r.domains_error) parts.push('域名错误:' + r.domains_error); + if (r.account && r.account.balance !== null && r.account.balance !== undefined) parts.push('余额 ' + r.account.balance); + results[p.id] = { ok: true, text: '✓ 同步完成：' + (parts.join('，') || '无变化') }; + await loadAll(); + } catch (e) { results[p.id] = { ok: false, text: '✗ ' + e.message }; } + finally { syncing[p.id] = false; } + } + return { store, Fmt, testing, syncing, results, isSupported, test, sync, seed: seedProviders, create: openProviderCreate, edit: openProviderEdit, del: deleteProvider }; + }, }; const ServersView = {