diff --git a/VERSION b/VERSION index 1d0ba9e..8f0916f 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.4.0 +0.5.0 diff --git a/app/adapters/registry.py b/app/adapters/registry.py index d33fefc..8faf0a9 100644 --- a/app/adapters/registry.py +++ b/app/adapters/registry.py @@ -35,6 +35,7 @@ def load_all() -> None: namesilo, tencent, vultr, + whmcs, ) diff --git a/app/adapters/whmcs.py b/app/adapters/whmcs.py new file mode 100644 index 0000000..39b9291 --- /dev/null +++ b/app/adapters/whmcs.py @@ -0,0 +1,248 @@ +"""WHMCS 账单面板适配器 + +适用于 TotHost 等使用 WHMCS 客户区、但没有开放 API 的平台。 +通过账号关联的登录凭据进入客户区，读取「我的产品/服务」列表： +服务 ID、主机名、下次到期日、状态。 + +只做只读核对，不执行开通/续费/停机等写操作。 +""" + +import html as html_lib +import re +from datetime import date +from typing import Optional +from urllib.parse import urlparse + +import httpx + +from app.adapters.base import BaseAdapter, NormalizedVPS, cloud_error_text +from app.adapters.registry import register + + +class WHMCSError(Exception): + """面板登录 / 页面解析失败（消息面向使用者，不含密码）""" + + +#: WHMCS 客户区状态 CSS 类后缀 -> 本项目资产状态 +_STATUS_MAP = { + "active": "active", + "suspended": "stopped", + "inactive": "stopped", + "terminated": "cancelled", + "cancelled": "cancelled", + "fraud": "cancelled", + "pending": "unknown", +} + +_IPV4_RE = re.compile(r"^(?:\d{1,3}\.){3}\d{1,3}$") + +#: 登录成功标志：客户区一定会渲染登出入口，但主题会按语言输出不同文案 +_LOGGED_IN_RE = re.compile( + r"logout|>\s*(?:退出登录|退出|登出|登出帳號|注销|登録解除|ログアウト)\s*<", re.I +) + + +@register("whmcs-panel") +class WHMCSAdapter(BaseAdapter): + """WHMCS 客户区只读适配器 + + 配置由账号凭据注入： + - base_url: 平台 console_url（如 https://billing.example.com/clientarea.php） + - username / password: accounts.credential_id 关联的登录凭据 + """ + + #: 凭据从 credentials 表注入，不在平台 API 配置里重复填写 + required_config: list = [] + + def __init__(self, config: Optional[dict] = None): + super().__init__(config) + raw = str(self.config.get("base_url") or "").strip() + if not raw: + raise WHMCSError("缺少面板地址（base_url），请先在平台资料中填写管理面板 URL") + if "://" not in raw: + raw = "https://" + raw + parsed = urlparse(raw) + if not parsed.netloc: + raise WHMCSError(f"面板地址无效：{raw}") + self.origin = f"{parsed.scheme}://{parsed.netloc}" + + # ------------------------------------------------------------------ # + # 面板交互 + # ------------------------------------------------------------------ # + def _client(self) -> httpx.Client: + return httpx.Client( + base_url=self.origin, + follow_redirects=True, + timeout=25, + headers={"User-Agent": "Mozilla/5.0 (compatible; vps-manager)"}, + ) + + def _login(self, client: httpx.Client) -> None: + username = str(self.config.get("username") or "").strip() + password = str(self.config.get("password") or "") + if not username or not password: + raise WHMCSError("缺少面板登录凭据（请在账号上关联带用户名和密码的登录凭据）") + + page = client.get("/login") + if page.status_code >= 400: + raise WHMCSError(f"打开登录页失败：HTTP {page.status_code}") + token = self._extract_token(page.text) + if not token: + raise WHMCSError("登录页缺少 CSRF token，面板结构可能已变化") + + resp = client.post( + "/login", + data={"username": username, "password": password, "token": token}, + ) + body = resp.text or "" + if not _LOGGED_IN_RE.search(body): + if re.search(r"two[- ]?factor|\b2fa\b|第二因素|兩步|双重验证|雙重驗證", body, re.I): + raise WHMCSError("面板启用了二次验证（2FA），无法自动登录") + raise WHMCSError("面板登录失败：账号或密码错误，或需要验证码") + + @staticmethod + def _extract_token(page: str) -> Optional[str]: + """提取登录表单的 CSRF token + + 优先在 action/id 含 login 的表单内查找——页面上可能同时存在工单、 + 改密等多个带 token 的表单，取错会导致提交被拒。 + """ + patterns = ( + r']*name=["\']token["\'][^>]*value=["\']([^"\']+)["\']', + r']*value=["\']([^"\']+)["\'][^>]*name=["\']token["\']', + ) + forms = re.findall(r"]*>.*?", page, re.I | re.S) + login_forms = [ + form for form in forms + if re.search(r'(?:action|id|class)=["\'][^"\']*(?:login|dologin)', form, re.I) + ] + for form in login_forms or forms: + for pattern in patterns: + found = re.search(pattern, form, re.I | re.S) + if found: + return found.group(1) + for pattern in patterns: + found = re.search(pattern, page, re.I | re.S) + if found: + return found.group(1) + return None + + def fetch_services(self, include_terminated: bool = True) -> list: + """登录面板并返回服务列表（默认包含已删除，供续费核对判断服务是否还在）""" + with self._client() as client: + self._login(client) + resp = client.get("/clientarea.php", params={"action": "products"}) + if resp.status_code >= 400: + raise WHMCSError(f"打开服务列表失败：HTTP {resp.status_code}") + services = self.parse_services(resp.text) + if not include_terminated: + services = [s for s in services if s["status"] != "cancelled"] + return services + + # ------------------------------------------------------------------ # + # 页面解析 + # ------------------------------------------------------------------ # + @classmethod + def parse_services(cls, page: str) -> list: + """从 WHMCS 服务列表页解析服务行 + + 依赖客户区主题输出： + - 行属性 data-element-id / data-domain + - 到期日 span.w-hidden 中的 ISO 日期 + - 状态 span 的 status-{class} 类名（与界面语言无关） + """ + start = re.search(r']*id=["\']tableServicesList["\']', page, re.I) + if not start: + raise WHMCSError("未在面板页面找到服务列表（页面结构可能已变化）") + end = page.find("", start.end()) + table = page[start.start(): end + 8 if end != -1 else len(page)] + + services = [] + for row in re.findall(r"]*>(.*?)", table, re.S | re.I): + service_id = re.search(r'data-element-id=["\'](\d+)["\']', row) + if not service_id: + continue + status = re.search( + r'class=["\'][^"\']*status\s+status-([a-z]+)[^"\']*["\'][^>]*>(.*?)', + row, + re.S | re.I, + ) + if not status: + continue + + domain_match = re.search(r'data-domain=["\']([^"\']*)["\']', row) + product_match = re.search(r"(.*?)", row, re.S | re.I) + due_match = re.search( + r']*class=["\'][^"\']*w-hidden[^"\']*["\'][^>]*>(\d{4}-\d{2}-\d{2})', + row, + re.I, + ) + if not due_match: + due_match = re.search(r"(\d{4}-\d{2}-\d{2})", row) + + domain = html_lib.unescape(domain_match.group(1)).strip() if domain_match else "" + product = cls._text(product_match.group(1)) if product_match else "" + status_class = status.group(1).lower() + status_text = cls._text(status.group(2)) + expiry = None + if due_match: + try: + expiry = date.fromisoformat(due_match.group(1)) + except ValueError: + expiry = None + + amount = re.search(r'data-order=["\']([\d.]+)["\']', row) + cycle = re.search(r']*class=["\'][^"\']*text-muted[^"\']*["\'][^>]*>(.*?)', row, re.I) + + services.append({ + "id": service_id.group(1), + "product": product, + "domain": domain, + "label": f"{product} {domain}".strip(), + "expiry_date": expiry, + "status": _STATUS_MAP.get(status_class, "unknown"), + "status_class": status_class, + "status_text": status_text, + "amount": float(amount.group(1)) if amount else None, + "cycle": cls._text(cycle.group(1)) if cycle else "", + }) + return services + + @staticmethod + def _text(fragment: str) -> str: + return re.sub(r"\s+", " ", html_lib.unescape(re.sub(r"<[^>]+>", " ", fragment))).strip() + + # ------------------------------------------------------------------ # + # BaseAdapter 接口 + # ------------------------------------------------------------------ # + def test_connection(self) -> dict: + try: + services = self.fetch_services(include_terminated=True) + return { + "ok": True, + "message": f"面板登录成功，读取到 {len(services)} 个服务", + } + except Exception as e: # noqa: BLE001 + return {"ok": False, "message": f"面板核对失败：{cloud_error_text(e)}"} + + def list_vps(self) -> list: + result = [] + for service in self.fetch_services(include_terminated=False): + domain = service["domain"] + result.append( + NormalizedVPS( + external_id=str(service["id"]), + name=service["label"] or str(service["id"]), + ip_address=domain if _IPV4_RE.match(domain) else "", + status=service["status"], + expiry_date=service["expiry_date"], + raw={ + "hostname": domain, + "product": service["product"], + "status_text": service["status_text"], + "amount": service["amount"], + "cycle": service["cycle"], + }, + ) + ) + return result diff --git a/app/services/renewal_service.py b/app/services/renewal_service.py index 832dd31..5c8bd1a 100644 --- a/app/services/renewal_service.py +++ b/app/services/renewal_service.py @@ -1,25 +1,28 @@ """续费检查服务 检查即将到期的资产，生成提醒消息并通过通知渠道发送。 +发送前会先尝试从平台侧核对真实状态（见 verification_service）， +核对到的到期日/状态会回写本地记录，避免按过期的手工数据误报。 """ +import logging from datetime import date, timedelta from sqlmodel import Session, select from app.core.config import settings from app.models.asset import Asset, AssetStatus -from app.services import notification_service +from app.services import notification_service, verification_service -# 已注销的资产不参与续费提醒 -_SKIP_STATUS = {AssetStatus.CANCELLED} +logger = logging.getLogger("vps-manager.renewal") def get_expiring_assets(session: Session, threshold_days: int) -> list: - """返回 threshold_days 天内到期（含已过期未处理）的资产列表 [(asset, days), ...]，按剩余天数升序 + """返回 threshold_days 天内到期（含已过期未处理）的资产列表 [(asset, days), ...] 到期日不设下界：过了续费日仍未注销/续费的资产，提醒不能就此消失。 - 性能优化：日期范围与状态过滤均下推到 SQL 层。 + 已注销资产直接排除；已停止资产仅在开启自动续费时提醒 + （停止实例不等于停止扣费，自动续费仍可能产生费用）。 """ today = date.today() deadline = today + timedelta(days=threshold_days) @@ -27,9 +30,14 @@ def get_expiring_assets(session: Session, threshold_days: int) -> list: select(Asset).where( Asset.expiry_date.is_not(None), Asset.expiry_date <= deadline, - Asset.status.notin_(_SKIP_STATUS), + Asset.status != AssetStatus.CANCELLED, ) ).all() + assets = [ + asset + for asset in assets + if not (asset.status == AssetStatus.STOPPED and not asset.auto_renew) + ] expiring = [(asset, (asset.expiry_date - today).days) for asset in assets] expiring.sort(key=lambda x: x[1]) return expiring @@ -48,7 +56,7 @@ def day_text(days: int) -> str: return f"{days} 天后" -def build_renewal_message(expiring: list, threshold_days: int) -> str: +def build_renewal_message(expiring: list, threshold_days: int, verification: dict = None) -> str: """生成续费提醒消息（纯文本，兼容 Telegram/邮件）""" if not expiring: return "" @@ -60,20 +68,52 @@ def build_renewal_message(expiring: list, threshold_days: int) -> str: lines = [head, ""] for asset, days in expiring: lines.append(f"• {asset.name} [{asset.provider}] （{day_text(days)}） 到期 {asset.expiry_date}") + result = (verification or {}).get(getattr(asset, "id", None)) + if result is not None: + if result.verified: + lines.append(f" ↳ 已核对：{result.detail}") + else: + lines.append(f" ↳ 未核对：{result.detail}，按手工登记日期提醒") + + if verification: + verified = sum( + 1 for asset, _ in expiring + if getattr(verification.get(getattr(asset, "id", None)), "verified", False) + ) + lines.append("") + lines.append(f"核对：已核对 {verified} 项，未核对 {len(expiring) - verified} 项") return "\n".join(lines) -def check_renewals(session: Session, threshold_days: int = None, send: bool = True) -> dict: +def check_renewals( + session: Session, + threshold_days: int = None, + send: bool = True, + verify: bool = True, +) -> dict: """检查续费并（可选）发送提醒 threshold_days: 提醒阈值天数，默认取配置 RENEWAL_THRESHOLD_DAYS send: 是否实际发送通知（False 时仅返回预览） + verify: 发送前是否向平台核对真实状态（测试/排查时可关闭） """ if threshold_days is None: threshold_days = settings.RENEWAL_THRESHOLD_DAYS expiring = get_expiring_assets(session, threshold_days) - message = build_renewal_message(expiring, threshold_days) + verification = {} + if verify and expiring: + try: + verification = verification_service.verify_assets( + session, [asset for asset, _ in expiring] + ) + except Exception: # noqa: BLE001 + # 核对是增强项：核对服务整体异常时仍按本地数据提醒，不能因此不发消息 + logger.exception("续费核对异常，退回本地登记数据") + # 核对可能已把到期日/状态改成最新值，重新筛选 + expiring = get_expiring_assets(session, threshold_days) + + message = build_renewal_message(expiring, threshold_days, verification) result = { "threshold_days": threshold_days, @@ -84,11 +124,25 @@ def check_renewals(session: Session, threshold_days: int = None, send: bool = Tr "provider": asset.provider, "days": days, "expiry_date": str(asset.expiry_date), + "verified": getattr(verification.get(asset.id), "verified", None), + "verify_source": getattr(verification.get(asset.id), "source", ""), + "verify_detail": getattr(verification.get(asset.id), "detail", ""), } for asset, days in expiring ], "message": message, } + if verification: + verified_count = sum( + 1 for asset, _ in expiring + if getattr(verification.get(asset.id), "verified", False) + ) + result["verification"] = { + "verified": verified_count, + "unverified": len(expiring) - verified_count, + } + else: + result["verification"] = None if send and message: result["notifications"] = notification_service.notify(message) diff --git a/app/services/sync_service.py b/app/services/sync_service.py index a3192cd..dff9912 100644 --- a/app/services/sync_service.py +++ b/app/services/sync_service.py @@ -31,6 +31,7 @@ from app.models.asset import ( DomainDetail, VPSDetail, ) +from app.models.credential import Credential from app.models.provider import Provider _VALID_STATUS = {s.value for s in AssetStatus} @@ -47,6 +48,19 @@ def _load_account_config(account: Account) -> dict: return {} +def _credential_login_config(session: Session, account: Account) -> dict: + """从凭据表取面板登录所需的用户名/密码（凭据表是唯一事实源）""" + if not account.credential_id: + return {} + cred = session.get(Credential, account.credential_id) + if not cred or not cred.password_encrypted: + return {} + password = crypto.decrypt(cred.password_encrypted) + if not password: + return {} + return {"username": cred.username or account.name, "password": password} + + def _find_provider_by_platform(session: Session, platform: str) -> Provider: """按账号的 platform（slug 或名称）定位平台""" provider = session.exec(select(Provider).where(Provider.slug == platform)).first() @@ -96,6 +110,16 @@ def _build_adapter_for_account(session: Session, account: Account): ) provider = _find_provider_by_platform(session, account.platform) config = _load_account_config(account) + if provider.sdk_type == "whmcs-panel": + # 面板登录类适配器不重复存密码：用户名/密码取自关联凭据， + # base_url 取自平台的管理面板地址。 + config = {**_credential_login_config(session, account), **config} + config.setdefault("base_url", provider.console_url or provider.website or "") + if not config.get("password"): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="该平台通过账单面板登录核对，请为账号关联带密码的登录凭据", + ) adapter = _build_adapter(provider, config) return provider, adapter @@ -121,14 +145,29 @@ def test_provider(session: Session, provider_id: int) -> dict: return test_account(session, account.id) +def _account_has_credentials(session: Session, account: Account, provider: Provider) -> bool: + """账号是否具备该平台可用的同步凭证 + + 普通云厂商看 api_config_encrypted；WHMCS 面板类平台看关联登录凭据。 + """ + if account.api_config_encrypted: + return True + if provider.sdk_type == "whmcs-panel": + return bool(_credential_login_config(session, account)) + return False + + def _first_account_with_config(session: Session, provider: Provider): - """找该平台下第一个配置了 API 凭证的账号（兼容旧入口）""" - return session.exec( + """找该平台下第一个具备可用凭证的账号（兼容旧入口）""" + accounts = session.exec( select(Account) .where(Account.platform == provider.slug) - .where(Account.api_config_encrypted.is_not(None)) # type: ignore[union-attr] .order_by(Account.id.asc()) - ).first() + ).all() + for account in accounts: + if _account_has_credentials(session, account, provider): + return account + return None def test_account(session: Session, account_id: int) -> dict: @@ -158,6 +197,32 @@ def _find_asset(session: Session, provider_id: int, external_id: str, asset_type ).first() +def _find_unlinked_vps(session: Session, provider: Provider, vps) -> Optional[Asset]: + """在手工登记、尚无 external_id 的同平台 VPS 中按 IP/名称找对应项 + + 手工登记资产没有云端实例 ID，直接同步会重复建一条； + 能唯一对上时补写 external_id，把手工台账接进自动同步。 + """ + candidates = session.exec( + select(Asset).where( + Asset.provider_id == provider.id, + Asset.asset_type == AssetType.VPS, + Asset.external_id.is_(None), + ) + ).all() + vps_ip = (vps.ip_address or "").strip() + vps_name = (vps.name or "").strip().lower() + for asset in candidates: + detail = session.exec( + select(VPSDetail).where(VPSDetail.asset_id == asset.id) + ).first() + if vps_ip and detail and (detail.ip_address or "").strip() == vps_ip: + return asset + if vps_name and (asset.name or "").strip().lower() == vps_name: + return asset + return None + + def _sync_vps(session: Session, provider: Provider, adapter: BaseAdapter, account_id: Optional[int] = None) -> dict: created = updated = 0 for vps in adapter.list_vps(): @@ -165,6 +230,10 @@ def _sync_vps(session: Session, provider: Provider, adapter: BaseAdapter, accoun # 漏了这一步，云端同步进来的 VPS 永远进不了续费提醒——而那是台账的核心用途。 expiry = parse_iso_date(vps.expiry_date) existing = _find_asset(session, provider.id, vps.external_id, AssetType.VPS) + if not existing: + existing = _find_unlinked_vps(session, provider, vps) + if existing: + existing.external_id = vps.external_id if existing: existing.name = vps.name or existing.name existing.status = _norm_status(vps.status) @@ -460,7 +529,8 @@ def sync_all_assets(session: Session, max_workers: int = 5) -> dict: Session（SQLite Session 不能跨线程共享，WAL 模式下多连接读写安全）。 未配凭证或平台无适配器的账号计入 skipped 而非 failed——否则每次定时同步 - 都会把手工登记的账号报成一堆错误。 + 都会把手工登记的账号报成一堆错误。WHMCS 面板类账号也跳过：续费检查会 + 按需登录核对，避免全量同步重命名手写资产或为历史服务新建台账。 """ from concurrent.futures import ThreadPoolExecutor @@ -470,11 +540,9 @@ def sync_all_assets(session: Session, max_workers: int = 5) -> dict: targets = [] skipped_no_config = [] skipped_no_adapter = [] + skipped_panel = [] for acc in accounts: label = f"{acc.platform}/{acc.name}" - if not acc.api_config_encrypted: - skipped_no_config.append(label) - continue try: provider = _find_provider_by_platform(session, acc.platform) except HTTPException: @@ -483,6 +551,14 @@ def sync_all_assets(session: Session, max_workers: int = 5) -> dict: if not provider.sdk_type or not registry.is_supported(provider.sdk_type): skipped_no_adapter.append(f"{label}（{provider.sdk_type or '未配 sdk_type'}）") continue + if provider.sdk_type == "whmcs-panel": + # 面板类平台由续费检查按需核对：定时全量同步会把手写资产名改成面板名、 + # 并可能为面板里的历史服务新建台账，副作用大于收益。 + skipped_panel.append(label) + continue + if not _account_has_credentials(session, acc, provider): + skipped_no_config.append(label) + continue targets.append((acc.id, label)) def _sync_one(account_id: int) -> dict: @@ -531,6 +607,7 @@ def sync_all_assets(session: Session, max_workers: int = 5) -> dict: "updated": updated, "skipped_no_config": len(skipped_no_config), "skipped_no_adapter": len(skipped_no_adapter), + "skipped_panel": len(skipped_panel), "errors": errors, "synced_at": utcnow().isoformat(), } diff --git a/app/services/verification_service.py b/app/services/verification_service.py new file mode 100644 index 0000000..0fd99d8 --- /dev/null +++ b/app/services/verification_service.py @@ -0,0 +1,359 @@ +"""资产真实状态核对服务 + +续费提醒发送前，尽可能从平台侧确认资产是否还在、是否已续费： +- whmcs-panel：登录账单面板读取服务列表（TotHost 等） +- 云厂商 API：复用 app/adapters 的只读清单能力 +- 无可用通道：标记未核对，保留手工登记值并继续提醒（避免漏续费） + +核对结果会回写 assets 的 expiry_date / status / external_id， +随后续费检查按更正后的数据重新筛选。 +""" + +from collections import defaultdict +from dataclasses import dataclass +from datetime import date +from typing import Optional + +from fastapi import HTTPException +from sqlmodel import Session, select + +from app.adapters import registry +from app.adapters.base import cloud_error_text, parse_iso_date +from app.models.asset import ( + Account, + Asset, + AssetStatus, + AssetType, + DomainDetail, + VPSDetail, +) +from app.models.provider import Provider +from app.services import sync_service + +_WHMCS = "whmcs-panel" + +_STATUS_MAP = { + "active": AssetStatus.ACTIVE, + "expired": AssetStatus.EXPIRED, + "stopped": AssetStatus.STOPPED, + "cancelled": AssetStatus.CANCELLED, + "unknown": AssetStatus.UNKNOWN, +} + + +@dataclass +class VerificationResult: + """单个资产的核对结果""" + + asset_id: int + verified: bool + source: str = "" + status: Optional[AssetStatus] = None + expiry_date: Optional[date] = None + external_id: Optional[str] = None + detail: str = "" + + +def _error_text(e: Exception) -> str: + if isinstance(e, HTTPException): + return str(e.detail) + return cloud_error_text(e) + + +def _unverified(asset_id: int, detail: str) -> VerificationResult: + return VerificationResult(asset_id=asset_id, verified=False, detail=detail) + + +def _status(raw: Optional[str]) -> Optional[AssetStatus]: + return _STATUS_MAP.get((raw or "").strip().lower()) + + +def verify_assets(session: Session, assets: list) -> dict: + """按账号分组核对资产，返回 {asset_id: VerificationResult}""" + results: dict = {} + if not assets: + return results + + groups = defaultdict(list) + for asset in assets: + groups[(asset.provider_id, asset.account_id)].append(asset) + + cache: dict = {} + for (provider_id, account_id), group in groups.items(): + provider = session.get(Provider, provider_id) if provider_id else None + account = session.get(Account, account_id) if account_id else None + if not provider: + for asset in group: + results[asset.id] = _unverified(asset.id, "资产未关联平台记录") + continue + + sdk = (provider.sdk_type or "").strip() + if sdk == _WHMCS: + ctx = _whmcs_context(session, provider, account, cache) + for asset in group: + results[asset.id] = _verify_whmcs_asset(session, provider, asset, group, ctx) + elif sdk and registry.is_supported(sdk): + if all(asset.asset_type == AssetType.AI_AGENT for asset in group): + # AI 适配器只提供余额，没有统一的订阅到期查询，别白跑一次 API + for asset in group: + results[asset.id] = _unverified( + asset.id, "AI 平台没有统一的订阅到期查询接口" + ) + continue + ctx = _api_context(session, provider, account, cache) + for asset in group: + results[asset.id] = _verify_api_asset(session, asset, ctx) + else: + for asset in group: + results[asset.id] = _unverified( + asset.id, f"平台「{provider.name}」未接入自动核对" + ) + + _apply(session, assets, results) + return results + + +# --------------------------------------------------------------------------- # +# WHMCS 面板通道 +# --------------------------------------------------------------------------- # +def _whmcs_context(session: Session, provider: Provider, account: Optional[Account], cache: dict) -> dict: + key = ("whmcs", provider.id, account.id if account else None) + if key in cache: + return cache[key] + + if not account: + ctx = {"error": "资产未关联账号，无法确定用哪个面板账号核对"} + else: + try: + _, adapter = sync_service._build_adapter_for_account(session, account) + fetch = getattr(adapter, "fetch_services", None) + if not fetch: + raise RuntimeError("该平台适配器不支持面板服务列表") + services = fetch(include_terminated=True) + ctx = { + "services": {str(s["id"]): s for s in services}, + "active": [s for s in services if s["status"] == "active"], + "source": f"{provider.name} 面板", + } + except Exception as e: # noqa: BLE001 + ctx = {"error": f"面板核对失败：{_error_text(e)}"} + + cache[key] = ctx + return ctx + + +def _verify_whmcs_asset( + session: Session, provider: Provider, asset: Asset, group: list, ctx: dict +) -> VerificationResult: + if ctx.get("error"): + return _unverified(asset.id, ctx["error"]) + + services = ctx["services"] + service_id = (asset.external_id or "").strip() + if service_id: + service = services.get(service_id) + if not service: + return VerificationResult( + asset_id=asset.id, + verified=True, + source=ctx["source"], + status=AssetStatus.CANCELLED, + detail="面板中已无该服务（可能已删除）", + ) + return _from_whmcs_service(asset, service, ctx["source"]) + + service = _match_whmcs_service(session, asset, services) + if not service: + # 账号下只有一个未关联资产、且面板只有一个在用服务时，可以安全地唯一对应 + unlinked = [a for a in group if not (a.external_id or "").strip()] + if len(unlinked) == 1 and len(ctx["active"]) == 1: + service = ctx["active"][0] + if not service: + return _unverified( + asset.id, "无法与面板服务对应：请设置平台实例 ID 或关联正确的账号" + ) + return _from_whmcs_service(asset, service, ctx["source"], external_id=str(service["id"])) + + +def _match_whmcs_service(session: Session, asset: Asset, services: dict) -> Optional[dict]: + detail = session.exec( + select(VPSDetail).where(VPSDetail.asset_id == asset.id) + ).first() + ip = (detail.ip_address or "").strip().lower() if detail else "" + name = (asset.name or "").strip().lower() + + for service in services.values(): + domain = (service.get("domain") or "").strip().lower() + label = (service.get("label") or "").strip().lower() + if ip and domain and domain == ip: + return service + if name and (domain == name or label == name): + return service + if name and domain and len(domain) >= 4 and (domain in name or name in domain): + return service + return None + + +def _from_whmcs_service( + asset: Asset, service: dict, source: str, external_id: Optional[str] = None +) -> VerificationResult: + detail = f"服务 {service['label']}：{service.get('status_text') or service.get('status_class')}" + if service.get("expiry_date"): + detail += f"，下次到期 {service['expiry_date']}" + return VerificationResult( + asset_id=asset.id, + verified=True, + source=source, + status=_status(service.get("status")), + expiry_date=service.get("expiry_date"), + external_id=external_id, + detail=detail, + ) + + +# --------------------------------------------------------------------------- # +# 云厂商 API 通道 +# --------------------------------------------------------------------------- # +def _api_context(session: Session, provider: Provider, account: Optional[Account], cache: dict) -> dict: + key = ("api", provider.id, account.id if account else None) + if key in cache: + return cache[key] + + if not account: + ctx = {"error": "资产未关联账号，无法使用云端 API 核对"} + else: + try: + _, adapter = sync_service._build_adapter_for_account(session, account) + caps = adapter.capabilities() + ctx = {"source": f"{provider.name} API", "vps": {}, "domains": {}} + if caps.get("list_vps"): + ctx["vps"] = {str(v.external_id): v for v in adapter.list_vps()} + if caps.get("list_domains"): + ctx["domains"] = {str(d.external_id): d for d in adapter.list_domains()} + if not caps.get("list_vps") and not caps.get("list_domains"): + ctx = {"error": f"{provider.name} 适配器不支持到期信息查询"} + except Exception as e: # noqa: BLE001 + ctx = {"error": f"云端 API 核对失败：{_error_text(e)}"} + + cache[key] = ctx + return ctx + + +def _verify_api_asset(session: Session, asset: Asset, ctx: dict) -> VerificationResult: + if ctx.get("error"): + return _unverified(asset.id, ctx["error"]) + + if asset.asset_type == AssetType.VPS: + items = ctx.get("vps", {}) + item = items.get(str(asset.external_id)) if asset.external_id else None + if item is None and not asset.external_id: + item = _match_api_vps(session, asset, items.values()) + if item is None: + if asset.external_id: + return VerificationResult( + asset_id=asset.id, + verified=True, + source=ctx["source"], + status=AssetStatus.CANCELLED, + detail="云端列表中已无该实例（可能已删除）", + ) + return _unverified(asset.id, "无法与云端实例对应：请设置平台实例 ID") + expiry = parse_iso_date(getattr(item, "expiry_date", None)) + detail = f"实例 {getattr(item, 'name', asset.name)} 状态 {getattr(item, 'status', 'unknown')}" + if expiry: + detail += f"，到期 {expiry}" + return VerificationResult( + asset_id=asset.id, + verified=True, + source=ctx["source"], + status=sync_service._norm_status(getattr(item, "status", None)), + expiry_date=expiry, + external_id=str(getattr(item, "external_id", "")) or None, + detail=detail, + ) + + if asset.asset_type == AssetType.DOMAIN: + items = ctx.get("domains", {}) + item = items.get(str(asset.external_id)) if asset.external_id else None + if item is None and not asset.external_id: + item = _match_api_domain(session, asset, items.values()) + if item is None: + if asset.external_id: + return VerificationResult( + asset_id=asset.id, + verified=True, + source=ctx["source"], + status=AssetStatus.CANCELLED, + detail="云端列表中已无该域名（可能已删除或转出）", + ) + return _unverified(asset.id, "无法与云端域名对应：请设置平台实例 ID") + expiry = parse_iso_date(getattr(item, "expiry_date", None)) + detail = f"域名 {getattr(item, 'domain_name', asset.name)} 状态 {getattr(item, 'status', 'unknown')}" + if expiry: + detail += f"，到期 {expiry}" + return VerificationResult( + asset_id=asset.id, + verified=True, + source=ctx["source"], + status=sync_service._norm_status(getattr(item, "status", None)), + expiry_date=expiry, + external_id=str(getattr(item, "external_id", "")) or None, + detail=detail, + ) + + return _unverified(asset.id, "该资产类型没有云端到期查询") + + +def _match_api_vps(session: Session, asset: Asset, items) -> Optional[object]: + detail = session.exec( + select(VPSDetail).where(VPSDetail.asset_id == asset.id) + ).first() + ip = (detail.ip_address or "").strip() if detail else "" + name = (asset.name or "").strip().lower() + for item in items: + if ip and (getattr(item, "ip_address", "") or "").strip() == ip: + return item + if name and (getattr(item, "name", "") or "").strip().lower() == name: + return item + return None + + +def _match_api_domain(session: Session, asset: Asset, items) -> Optional[object]: + detail = session.exec( + select(DomainDetail).where(DomainDetail.asset_id == asset.id) + ).first() + domain = (detail.domain_name or "").strip().lower() if detail else "" + name = (asset.name or "").strip().lower() + for item in items: + candidate = (getattr(item, "domain_name", "") or "").strip().lower() + if domain and candidate == domain: + return item + if name and candidate == name: + return item + return None + + +# --------------------------------------------------------------------------- # +# 回写 +# --------------------------------------------------------------------------- # +def _apply(session: Session, assets: list, results: dict) -> None: + changed = False + for asset in assets: + result = results.get(asset.id) + if not result or not result.verified: + continue + touched = False + if result.expiry_date is not None and asset.expiry_date != result.expiry_date: + asset.expiry_date = result.expiry_date + touched = True + if result.status is not None and asset.status != result.status: + asset.status = result.status + touched = True + if result.external_id and not asset.external_id: + asset.external_id = result.external_id + touched = True + if touched: + session.add(asset) + changed = True + if changed: + session.commit() diff --git a/scripts/check_renewals.py b/scripts/check_renewals.py index 5492ee4..fa5c142 100644 --- a/scripts/check_renewals.py +++ b/scripts/check_renewals.py @@ -1,9 +1,13 @@ """定时检查续费并发送提醒（供 systemd timer 调用） -用法：python scripts/check_renewals.py [threshold_days] -依赖 .env 中的通知渠道配置（TELEGRAM_*/SMTP_*）。 +用法： + python scripts/check_renewals.py [threshold_days] [--dry-run] [--no-verify] + +默认发送通知；--dry-run 只打印、不发送，便于核对平台真实状态。 +依赖 .env 中的通知渠道配置与账号登录凭据（TELEGRAM_*/SMTP_*、credentials）。 """ +import argparse import sys from pathlib import Path @@ -17,21 +21,45 @@ from app.services import renewal_service def main() -> int: - threshold = int(sys.argv[1]) if len(sys.argv) > 1 else None + parser = argparse.ArgumentParser(description="检查资产续费并发送提醒") + parser.add_argument("threshold_days", nargs="?", type=int, default=None) + parser.add_argument("--dry-run", action="store_true", help="只打印，不发送通知") + parser.add_argument("--no-verify", action="store_true", help="跳过平台真实状态核对") + args = parser.parse_args() + init_db() with Session(assets_engine) as session: - result = renewal_service.check_renewals(session, threshold_days=threshold) + result = renewal_service.check_renewals( + session, + threshold_days=args.threshold_days, + send=not args.dry_run, + verify=not args.no_verify, + ) + # days 可为负（已过期），汇总行拆开统计，单项文案复用 day_text（与提醒消息一致） overdue = sum(1 for item in result["expiring"] if item["days"] < 0) - print( + summary = ( f"[续费检查] 阈值 {result['threshold_days']} 天 | 已过期 {overdue} 项 | " f"即将到期 {result['expiring_count'] - overdue} 项" ) + verification = result.get("verification") + if verification: + summary += ( + f" | 已核对 {verification['verified']} 项" + f" | 未核对 {verification['unverified']} 项" + ) + print(summary) + for item in result["expiring"]: print( f" - {item['name']} [{item['provider']}] " f"{renewal_service.day_text(item['days'])}（到期 {item['expiry_date']}）" ) + if item.get("verified") is True: + print(f" 已核对[{item['verify_source']}]: {item['verify_detail']}") + elif item.get("verified") is False: + print(f" 未核对: {item['verify_detail']}") + for n in result.get("notifications", []): status = "成功" if n["ok"] else f"失败：{n['message']}" print(f" 通知[{n['channel']}]: {status}") diff --git a/scripts/sync_all_assets.py b/scripts/sync_all_assets.py index d78c141..5290e26 100644 --- a/scripts/sync_all_assets.py +++ b/scripts/sync_all_assets.py @@ -26,7 +26,8 @@ def main() -> int: print( f"[云资产同步] 账号 {result['total']} | 成功 {result['success']} | 失败 {result['failed']} " f"| 新增 {result['created']} | 更新 {result['updated']} " - f"| 跳过(无凭证) {result['skipped_no_config']} | 跳过(无适配器) {result['skipped_no_adapter']}" + f"| 跳过(无凭证) {result['skipped_no_config']} | 跳过(无适配器) {result['skipped_no_adapter']} " + f"| 跳过(面板核对) {result.get('skipped_panel', 0)}" ) for err in result["errors"]: print(f" - {err}") diff --git a/tests/test_renewal_service.py b/tests/test_renewal_service.py index e23c1df..cd24fc2 100644 --- a/tests/test_renewal_service.py +++ b/tests/test_renewal_service.py @@ -24,6 +24,7 @@ import app.models.provider # noqa: F401 import app.models.ssl # noqa: F401 from app.models.asset import Asset, AssetStatus, AssetType from app.services import renewal_service +from app.services.verification_service import VerificationResult @pytest.fixture() @@ -36,11 +37,11 @@ def engine(tmp_path): eng.dispose() -def _seed(engine, name, *, expiry, status=AssetStatus.ACTIVE, provider="vultr"): +def _seed(engine, name, *, expiry, status=AssetStatus.ACTIVE, provider="vultr", auto_renew=False): with Session(engine) as s: s.add(Asset( name=name, asset_type=AssetType.VPS, provider=provider, - expiry_date=expiry, status=status, + expiry_date=expiry, status=status, auto_renew=auto_renew, )) s.commit() @@ -91,6 +92,27 @@ def test_get_expiring_assets_excludes_cancelled(engine): assert names == ["active"] +def test_get_expiring_assets_stopped_without_auto_renew_excluded(engine): + """已停止且未开自动续费的资产不再提醒""" + today = date.today() + _seed(engine, "stopped", expiry=today + timedelta(days=5), status=AssetStatus.STOPPED) + with Session(engine) as s: + result = renewal_service.get_expiring_assets(s, threshold_days=30) + assert result == [] + + +def test_get_expiring_assets_stopped_with_auto_renew_included(engine): + """已停止但自动续费仍开着，可能继续扣费，必须提醒""" + today = date.today() + _seed( + engine, "stopped-autorenew", expiry=today + timedelta(days=5), + status=AssetStatus.STOPPED, auto_renew=True, + ) + with Session(engine) as s: + result = renewal_service.get_expiring_assets(s, threshold_days=30) + assert [a.name for a, _ in result] == ["stopped-autorenew"] + + def test_get_expiring_assets_no_expiry_date_skipped(engine): """未填到期日的资产不出现在到期列表""" _seed(engine, "no-expiry", expiry=None) @@ -132,6 +154,26 @@ def test_build_message_no_overdue(): assert "共 1 项" in msg +def test_build_message_labels_verification_source(): + """消息中区分「已核对」与「未核对」，并给出核对汇总""" + today = date.today() + a1 = Asset(id=1, name="vps-1", provider="tothost", expiry_date=today - timedelta(days=1)) + a2 = Asset(id=2, name="vps-2", provider="agnes", expiry_date=today + timedelta(days=5)) + verification = { + 1: VerificationResult( + asset_id=1, verified=True, source="TotHost 面板", + detail="服务 TOT K11 vpn.host：可用，下次到期 2026-11-25", + ), + 2: VerificationResult(asset_id=2, verified=False, detail="平台「Agnes」未接入自动核对"), + } + msg = renewal_service.build_renewal_message( + [(a1, -1), (a2, 5)], threshold_days=30, verification=verification + ) + assert "已核对：服务 TOT K11 vpn.host" in msg + assert "未核对：平台「Agnes」未接入自动核对" in msg + assert "核对：已核对 1 项，未核对 1 项" in msg + + # ----------------------------- check_renewals ----------------------------- # def test_check_renewals_no_send(engine): """send=False 时不触发通知，仅返回预览""" @@ -172,3 +214,48 @@ def test_check_renewals_default_threshold_from_config(engine): with Session(engine) as s: result = renewal_service.check_renewals(s, send=False) assert result["threshold_days"] == renewal_service.settings.RENEWAL_THRESHOLD_DAYS + + +def test_check_renewals_verification_suppresses_false_alarm(engine): + """核对发现已续费到 90 天后时，本轮不再提醒""" + today = date.today() + _seed(engine, "renewed", expiry=today - timedelta(days=1)) + + def fake_verify(session, assets): + results = {} + for asset in assets: + asset.expiry_date = today + timedelta(days=90) + session.add(asset) + results[asset.id] = VerificationResult( + asset_id=asset.id, verified=True, source="TotHost 面板", + expiry_date=asset.expiry_date, detail="已续费", + ) + session.commit() + return results + + with Session(engine) as s: + with patch.object( + renewal_service.verification_service, "verify_assets", side_effect=fake_verify + ) as mock_verify: + result = renewal_service.check_renewals(s, threshold_days=30, send=False) + mock_verify.assert_called_once() + assert result["expiring_count"] == 0 + assert result["message"] == "" + assert result["notifications"] == [] + + +def test_check_renewals_can_disable_verification(engine): + """verify=False 时不调用核对服务""" + today = date.today() + _seed(engine, "soon", expiry=today + timedelta(days=5)) + with Session(engine) as s: + with patch.object( + renewal_service.verification_service, "verify_assets", + side_effect=AssertionError("不应调用核对"), + ) as mock_verify: + result = renewal_service.check_renewals( + s, threshold_days=30, send=False, verify=False + ) + mock_verify.assert_not_called() + assert result["expiring_count"] == 1 + assert result["verification"] is None diff --git a/tests/test_sync_chain.py b/tests/test_sync_chain.py index 1bb429a..c0e9cf4 100644 --- a/tests/test_sync_chain.py +++ b/tests/test_sync_chain.py @@ -93,7 +93,7 @@ def test_script_summary_prints_every_key_the_service_returns(script, capsys): assert rc == 0 assert "[云资产同步]" in out for fragment in ("账号 0", "成功 0", "失败 0", "新增 0", "更新 0", - "跳过(无凭证) 0", "跳过(无适配器) 0"): + "跳过(无凭证) 0", "跳过(无适配器) 0", "跳过(面板核对) 0"): assert fragment in out, f"摘要缺少 {fragment}：{out!r}" diff --git a/tests/test_verification_service.py b/tests/test_verification_service.py new file mode 100644 index 0000000..82fdda3 --- /dev/null +++ b/tests/test_verification_service.py @@ -0,0 +1,207 @@ +"""verification_service 测试：发送前平台状态核对 + +覆盖： +- WHMCS 面板：按 external_id 更新到期日/状态 +- WHMCS 面板：服务已从面板消失 -> 标记已注销 +- WHMCS 面板：账号下唯一未关联资产 + 唯一在用服务 -> 自动补 external_id +- 云厂商 API：按 external_id 更新到期日/状态 +- 平台无适配器 -> 未核对 + +网络适配器全部用 fake 替换，不发起真实请求。 +""" + +from datetime import date + +import pytest +from sqlmodel import Session, SQLModel, create_engine + +import app.models.asset # noqa: F401 +import app.models.credential # noqa: F401 +import app.models.monitor # noqa: F401 +import app.models.provider # noqa: F401 +import app.models.ssl # noqa: F401 +from app.adapters.base import NormalizedVPS +from app.models.asset import Account, Asset, AssetStatus, AssetType, VPSDetail +from app.models.provider import Provider, ProviderCategory +from app.services import sync_service, verification_service + + +@pytest.fixture() +def engine(tmp_path): + eng = create_engine( + f"sqlite:///{tmp_path / 'assets.db'}", connect_args={"check_same_thread": False} + ) + SQLModel.metadata.create_all(eng) + yield eng + eng.dispose() + + +def _seed_provider(session, sdk_type="whmcs-panel"): + provider = Provider( + slug="tothost", name="TotHost", category=ProviderCategory.VPS, + services="vps", sdk_type=sdk_type, + console_url="https://billing.example.com/clientarea.php", + ) + session.add(provider) + session.commit() + session.refresh(provider) + return provider + + +def _seed_account(session, provider): + account = Account(name="user@example.com", platform=provider.slug) + session.add(account) + session.commit() + session.refresh(account) + return account + + +def _seed_asset(session, provider, account, *, name="vn-200", ip="103.70.115.57", + expiry=date(2026, 10, 6), external_id=None): + asset = Asset( + name=name, asset_type=AssetType.VPS, provider=provider.slug, + provider_id=provider.id, account_id=account.id if account else None, + expiry_date=expiry, status=AssetStatus.ACTIVE, external_id=external_id, + ) + session.add(asset) + session.commit() + session.refresh(asset) + session.add(VPSDetail(asset_id=asset.id, ip_address=ip)) + session.commit() + return asset + + +class FakeWHMCS: + def __init__(self, services): + self.services = services + + def fetch_services(self, include_terminated=True): + if include_terminated: + return self.services + return [s for s in self.services if s["status"] != "cancelled"] + + +def _service(sid="36979", domain="remote-dev", expiry=date(2027, 1, 7), status="active"): + return { + "id": sid, "product": "TOT K11", "domain": domain, + "label": f"TOT K11 {domain}", "expiry_date": expiry, "status": status, + "status_class": status, "status_text": "可用", "amount": None, "cycle": "季繳", + } + + +def test_whmcs_updates_expiry_and_links_external_id(engine, monkeypatch): + with Session(engine) as s: + provider = _seed_provider(s) + account = _seed_account(s, provider) + asset = _seed_asset(s, provider, account) + + fake = FakeWHMCS([_service()]) + monkeypatch.setattr( + sync_service, "_build_adapter_for_account", lambda session, acct: (provider, fake) + ) + + results = verification_service.verify_assets(s, [asset]) + s.refresh(asset) + snapshot = (asset.expiry_date, asset.status, asset.external_id) + + result = results[asset.id] + assert result.verified is True + assert result.source == "TotHost 面板" + assert snapshot == (date(2027, 1, 7), AssetStatus.ACTIVE, "36979") + + +def test_whmcs_missing_service_marks_cancelled(engine, monkeypatch): + with Session(engine) as s: + provider = _seed_provider(s) + account = _seed_account(s, provider) + asset = _seed_asset(s, provider, account, external_id="99999") + + monkeypatch.setattr( + sync_service, "_build_adapter_for_account", + lambda session, acct: (provider, FakeWHMCS([_service()])), + ) + + results = verification_service.verify_assets(s, [asset]) + s.refresh(asset) + + assert results[asset.id].verified is True + assert results[asset.id].status == AssetStatus.CANCELLED + assert asset.status == AssetStatus.CANCELLED + assert "已无该服务" in results[asset.id].detail + + +def test_whmcs_without_account_is_unverified(engine, monkeypatch): + with Session(engine) as s: + provider = _seed_provider(s) + asset = _seed_asset(s, provider, None) + + results = verification_service.verify_assets(s, [asset]) + + assert results[asset.id].verified is False + assert "未关联账号" in results[asset.id].detail + + +def test_whmcs_ambiguous_unlinked_asset_is_unverified(engine, monkeypatch): + """同账号有两个未关联资产、面板只有一个在用服务时不做猜测""" + with Session(engine) as s: + provider = _seed_provider(s) + account = _seed_account(s, provider) + a1 = _seed_asset(s, provider, account, name="vps-a", ip="1.1.1.1") + a2 = _seed_asset(s, provider, account, name="vps-b", ip="2.2.2.2") + + monkeypatch.setattr( + sync_service, "_build_adapter_for_account", + lambda session, acct: (provider, FakeWHMCS([_service()])), + ) + + results = verification_service.verify_assets(s, [a1, a2]) + + assert results[a1.id].verified is False + assert results[a2.id].verified is False + assert "无法与面板服务对应" in results[a1.id].detail + + +class FakeAPI: + def __init__(self, vps): + self._vps = vps + + def capabilities(self): + return {"list_vps": True, "list_domains": False} + + def list_vps(self): + return self._vps + + +def test_api_updates_expiry_and_status(engine, monkeypatch): + with Session(engine) as s: + provider = _seed_provider(s, sdk_type="vultr-api") + account = _seed_account(s, provider) + asset = _seed_asset(s, provider, account, external_id="abc") + + fake = FakeAPI([NormalizedVPS( + external_id="abc", name="vps", ip_address="103.70.115.57", + status="active", expiry_date=date(2027, 5, 1), + )]) + monkeypatch.setattr( + sync_service, "_build_adapter_for_account", lambda session, acct: (provider, fake) + ) + + results = verification_service.verify_assets(s, [asset]) + s.refresh(asset) + snapshot = (asset.expiry_date, asset.status) + + assert results[asset.id].verified is True + assert results[asset.id].source == "TotHost API" + assert snapshot == (date(2027, 5, 1), AssetStatus.ACTIVE) + + +def test_unsupported_provider_is_unverified(engine): + with Session(engine) as s: + provider = _seed_provider(s, sdk_type=None) + account = _seed_account(s, provider) + asset = _seed_asset(s, provider, account) + + results = verification_service.verify_assets(s, [asset]) + + assert results[asset.id].verified is False + assert "未接入自动核对" in results[asset.id].detail diff --git a/tests/test_whmcs_adapter.py b/tests/test_whmcs_adapter.py new file mode 100644 index 0000000..310f62f --- /dev/null +++ b/tests/test_whmcs_adapter.py @@ -0,0 +1,140 @@ +"""WHMCS 面板适配器测试：页面解析与状态映射 + +使用真实客户区结构的 HTML 片段，不发起网络请求。 +""" + +from datetime import date + +import pytest + +from app.adapters.whmcs import WHMCSAdapter, WHMCSError + +SERVICES_HTML = """ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
產品 / 服務價格下次到期日狀態
TOT K11
remote-dev		226,950VND 季繳2027-01-07星期四, 一月 7第, 2027可用
TOT K11
vpn.host		$3.03 月繳2026-11-25星期三, 十一月 25第, 2026可用
TOT D12
123.31.29.104		275,000VND 月繳2026-06-11星期四, 六月 11第, 2026已刪除
TOT K11
paused.host		$3.03 月繳2026-12-01星期二, 十二月 1第, 2026已暫停
+""" + + +def _adapter(): + return WHMCSAdapter({ + "base_url": "https://billing.example.com/clientarea.php", + "username": "user@example.com", + "password": "secret", + }) + + +def test_parse_services_extracts_all_fields(): + services = WHMCSAdapter.parse_services(SERVICES_HTML) + assert [s["id"] for s in services] == ["36979", "28911", "36980", "36981"] + + first = services[0] + assert first["product"] == "TOT K11" + assert first["domain"] == "remote-dev" + assert first["label"] == "TOT K11 remote-dev" + assert first["expiry_date"] == date(2027, 1, 7) + assert first["status"] == "active" + assert first["status_text"] == "可用" + assert first["amount"] == 226950.0 + assert first["cycle"] == "季繳" + + +def test_parse_services_maps_whmcs_status_classes(): + services = {s["id"]: s for s in WHMCSAdapter.parse_services(SERVICES_HTML)} + assert services["36979"]["status"] == "active" + assert services["36980"]["status"] == "cancelled" + assert services["36981"]["status"] == "stopped" + + +def test_parse_services_requires_service_table(): + with pytest.raises(WHMCSError): + WHMCSAdapter.parse_services("no table") + + +def test_list_vps_filters_terminated_and_keeps_ip(monkeypatch): + adapter = _adapter() + services = WHMCSAdapter.parse_services(SERVICES_HTML) + monkeypatch.setattr(adapter, "fetch_services", lambda include_terminated=True: [ + s for s in services if include_terminated or s["status"] != "cancelled" + ]) + + vps = adapter.list_vps() + assert [v.external_id for v in vps] == ["36979", "28911", "36981"] + + by_id = {v.external_id: v for v in vps} + assert by_id["36979"].name == "TOT K11 remote-dev" + assert by_id["36979"].ip_address == "" # 主机名不是 IP + assert by_id["36979"].expiry_date == date(2027, 1, 7) + assert "36980" not in by_id # 已删除项不出现在 list_vps + + +def test_extract_token_supports_attribute_orders(): + assert WHMCSAdapter._extract_token('') == "abc" + assert WHMCSAdapter._extract_token('') == "xyz" + + +def test_extract_token_prefers_login_form(): + """页面上有多个带 token 的表单时，必须取登录表单里的那个""" + page = """ +
+ +
+
+ +
+ """ + assert WHMCSAdapter._extract_token(page) == "right" + + +@pytest.mark.parametrize("body", [ + 'Logout', + '退出登录', + '登出', + 'ログアウト', +]) +def test_logged_in_markers_cover_localized_themes(body): + """主题本地化后不再输出英文 Logout，误判会让核对永远失败""" + from app.adapters.whmcs import _LOGGED_IN_RE + + assert _LOGGED_IN_RE.search(body) + + +def test_logged_in_marker_ignores_plain_text(): + from app.adapters.whmcs import _LOGGED_IN_RE + + assert not _LOGGED_IN_RE.search("

请勿退出页面

") + + + +def test_adapter_requires_base_url(): + with pytest.raises(WHMCSError): + WHMCSAdapter({"username": "u", "password": "p"})