264 lines
8.7 KiB
Python
264 lines
8.7 KiB
Python
"""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 = ""
|
|
real_entitlement_allowed: bool = False
|
|
|
|
def __post_init__(self):
|
|
if self.state not in _ALLOWED_STATES:
|
|
raise ValueError("未知订阅状态")
|
|
|
|
@property
|
|
def allows_product_workflows(self) -> bool:
|
|
return self.real_entitlement_allowed
|
|
|
|
@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)
|
|
real_entitlement_allowed = data.get("real_entitlement_allowed")
|
|
if type(real_entitlement_allowed) is not bool:
|
|
return SubscriptionStatus(STATUS_UNAVAILABLE)
|
|
if real_entitlement_allowed and normalized not in {
|
|
STATUS_ACTIVE,
|
|
STATUS_GRACE,
|
|
}:
|
|
return SubscriptionStatus(STATUS_UNAVAILABLE)
|
|
if normalized in {STATUS_ACTIVE, STATUS_GRACE} and not real_entitlement_allowed:
|
|
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,
|
|
real_entitlement_allowed=real_entitlement_allowed,
|
|
)
|
|
|
|
|
|
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
|