"""cmshopee account subscription status helpers. The remote service remains the authority for subscription enforcement. This module only normalizes the status response for desktop UI and preflight use; it deliberately never persists or exposes the API key it uses. """ from __future__ import annotations from dataclasses import dataclass from datetime import datetime from urllib.parse import urlsplit from . import ai, appconfig PRODUCT_CODE = "cmshopee" STATUS_NOT_CONFIGURED = "not_configured" STATUS_ACTIVE = "active" STATUS_GRACE = "grace" STATUS_REQUIRED = "required" STATUS_EXPIRED = "expired" STATUS_REVOKED = "revoked" STATUS_ACCOUNT_DISABLED = "account_disabled" STATUS_KEY_INVALID = "key_invalid" STATUS_UNAVAILABLE = "unavailable" STATUS_LEGACY = "legacy" _ALLOWED_STATES = { STATUS_NOT_CONFIGURED, STATUS_ACTIVE, STATUS_GRACE, STATUS_REQUIRED, STATUS_EXPIRED, STATUS_REVOKED, STATUS_ACCOUNT_DISABLED, STATUS_KEY_INVALID, STATUS_UNAVAILABLE, STATUS_LEGACY, } @dataclass(frozen=True) class SubscriptionStatus: """A redacted, UI-safe view of one cmshopee subscription lookup.""" state: str account_name: str = "" plan_name: str = "" expires_at: str = "" grace_expires_at: str = "" manage_url: str = "" notice_id: str = "" def __post_init__(self): if self.state not in _ALLOWED_STATES: raise ValueError("未知订阅状态") @property def allows_product_workflows(self) -> bool: return self.state in {STATUS_ACTIVE, STATUS_GRACE, STATUS_LEGACY} @property def credentials_accepted(self) -> bool: return self.state not in { STATUS_NOT_CONFIGURED, STATUS_KEY_INVALID, STATUS_UNAVAILABLE, } @property def interface_available(self) -> bool: return self.state != STATUS_LEGACY @property def user_message(self) -> str: return { STATUS_NOT_CONFIGURED: "请先在设置配置 cmhub API Key", STATUS_ACTIVE: "会员有效", STATUS_GRACE: "会员处于宽限期", STATUS_REQUIRED: "当前账号尚未开通蝦皮圈会员", STATUS_EXPIRED: "当前账号的蝦皮圈会员已到期", STATUS_REVOKED: "当前账号的蝦皮圈会员已撤销", STATUS_ACCOUNT_DISABLED: "cmhub 账号当前不可用", STATUS_KEY_INVALID: "cmhub API Key 无效，请在设置重新填写", STATUS_UNAVAILABLE: "暂时无法确认会员状态", STATUS_LEGACY: "会员服务尚未启用，当前按原有方式运行", }[self.state] def check_status( config=None, *, cmhub_config_path=None, base_url=None, api_key=None, request_json=None, ) -> SubscriptionStatus: """Look up the current API-key account subscription without leaking secrets. A service that has not yet deployed the endpoint returns ``404``. That is an explicit rollout compatibility state rather than an invalid membership. """ cfg = appconfig.load_config() if config is None else config try: cmhub = appconfig.cmhub_config(cfg) configured_base_url = ( cmhub.get("base_url") if base_url is None else base_url ) resolved_base_url = appconfig.normalize_cmhub_base_url(configured_base_url) key_path = ( cmhub_config_path or cfg.get("cmhub_config_path") or appconfig.cmhub_config_file_path(cfg) ) resolved_api_key = ( appconfig.get_cmhub_api_key(path=key_path) if api_key is None else str(api_key or "").strip() ) except Exception: return SubscriptionStatus(STATUS_NOT_CONFIGURED) if not resolved_base_url or not resolved_api_key: return SubscriptionStatus(STATUS_NOT_CONFIGURED) requester = request_json or ai.request_cmhub_json try: data = requester( "GET", resolved_base_url, "/api/v1/cmshopee/subscription/status", resolved_api_key, connect_timeout=cmhub.get( "connect_timeout", appconfig.CMHUB_CONNECT_TIMEOUT_DEFAULT, ), read_timeout=30, use_system_proxy=bool(appconfig.ai_config(cfg).get("use_system_proxy")), ) except ai.CMHubError as exc: return _status_from_cmhub_error(exc) except Exception: return SubscriptionStatus(STATUS_UNAVAILABLE) return _parse_status_response(data, resolved_base_url) def format_expiry(value: str) -> str: """Return a compact local date for the title-area membership label.""" text = str(value or "").strip() if not text: return "" try: parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) except ValueError: return "" return parsed.date().isoformat() def _status_from_cmhub_error(exc: ai.CMHubError) -> SubscriptionStatus: code = str(getattr(exc, "code", "") or "").strip().lower() if code == "not_found": return SubscriptionStatus(STATUS_LEGACY) if code == "unauthorized": return SubscriptionStatus(STATUS_KEY_INVALID) if code == "account_disabled": return SubscriptionStatus(STATUS_ACCOUNT_DISABLED) if code in {"subscription_required", "subscription_missing"}: return SubscriptionStatus(STATUS_REQUIRED) if code in {"subscription_expired", "license_expired"}: return SubscriptionStatus(STATUS_EXPIRED) if code in {"subscription_revoked", "subscription_cancelled"}: return SubscriptionStatus(STATUS_REVOKED) return SubscriptionStatus(STATUS_UNAVAILABLE) def _parse_status_response(data, base_url: str) -> SubscriptionStatus: if not isinstance(data, dict): return SubscriptionStatus(STATUS_UNAVAILABLE) if str(data.get("product_code") or "").strip().lower() != PRODUCT_CODE: return SubscriptionStatus(STATUS_UNAVAILABLE) normalized = _normalize_remote_state(data.get("status")) if not normalized: return SubscriptionStatus(STATUS_UNAVAILABLE) account = data.get("account") if isinstance(data.get("account"), dict) else {} plan = data.get("plan") if isinstance(data.get("plan"), dict) else {} account_name = str(account.get("display_name") or "").strip() plan_name = str(plan.get("display_name") or "").strip() expires_at = str(data.get("expires_at") or "").strip() grace_expires_at = str(data.get("grace_expires_at") or "").strip() manage_url = safe_manage_url(data.get("manage_url"), base_url) notice_id = str(data.get("notice_id") or "").strip() if normalized in {STATUS_ACTIVE, STATUS_GRACE}: if not account_name or not plan_name or not format_expiry(expires_at): return SubscriptionStatus(STATUS_UNAVAILABLE) if ( normalized == STATUS_GRACE and grace_expires_at and not format_expiry(grace_expires_at) ): return SubscriptionStatus(STATUS_UNAVAILABLE) return SubscriptionStatus( normalized, account_name=account_name, plan_name=plan_name, expires_at=expires_at, grace_expires_at=grace_expires_at, manage_url=manage_url, notice_id=notice_id, ) def _normalize_remote_state(value) -> str: normalized = str(value or "").strip().lower().replace("-", "_") return { "active": STATUS_ACTIVE, "grace": STATUS_GRACE, "required": STATUS_REQUIRED, "subscription_required": STATUS_REQUIRED, "expired": STATUS_EXPIRED, "subscription_expired": STATUS_EXPIRED, "revoked": STATUS_REVOKED, "cancelled": STATUS_REVOKED, "subscription_revoked": STATUS_REVOKED, "account_disabled": STATUS_ACCOUNT_DISABLED, }.get(normalized, "") def safe_manage_url(value, base_url: str) -> str: text = str(value or "").strip() if not text: return "" parsed = urlsplit(text) base = urlsplit(str(base_url or "")) if parsed.scheme != "https" or not parsed.hostname: return "" if ( parsed.username or parsed.password or not base.hostname or parsed.hostname.lower() != base.hostname.lower() ): return "" return text _safe_manage_url = safe_manage_url