cmshoppe/app/subscription.py

"""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