from __future__ import annotations import json import uuid from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any import requests from xhs_bridge.config import LicenseConfig from xhs_bridge.exceptions import LicenseError, LicenseRequiredError from xhs_bridge.license_codes import ( current_period_label, is_builtin_code_current_month, is_builtin_code_expired, lookup_builtin_code, normalize_activation_code, ) @dataclass class LicenseRecord: code: str device_id: str plan: str expires_at: str # ISO8601 activated_at: str period_label: str = "" # 内置码对应月份，如 2026-03 def is_expired(self) -> bool: exp = datetime.fromisoformat(self.expires_at.replace("Z", "+00:00")) now = datetime.now(timezone.utc) if exp.tzinfo is None: exp = exp.replace(tzinfo=timezone.utc) return now >= exp def days_remaining(self) -> int: exp = datetime.fromisoformat(self.expires_at.replace("Z", "+00:00")) now = datetime.now(timezone.utc) if exp.tzinfo is None: exp = exp.replace(tzinfo=timezone.utc) delta = exp - now return max(0, delta.days) def has_valid_license(manager: "LicenseManager") -> bool: """本地是否存在未过期的授权记录。""" record = manager.load_cached() if record is None: return False if record.is_expired(): manager.clear() return False return True class LicenseManager: """激活码管理：内置 12 个 GUID 月卡码（过期作废），可选 HTTP 对接。""" def __init__(self, config: LicenseConfig) -> None: self.config = config self.cache_path = config.resolved_cache_path() self.cache_path.parent.mkdir(parents=True, exist_ok=True) self._device_id = self._load_or_create_device_id() @property def device_id(self) -> str: return self._device_id def _device_id_path(self) -> Path: return self.cache_path.parent / "device_id" def _load_or_create_device_id(self) -> str: path = self._device_id_path() if path.is_file(): return path.read_text(encoding="utf-8").strip() device_id = str(uuid.uuid4()) path.write_text(device_id, encoding="utf-8") return device_id def load_cached(self) -> LicenseRecord | None: if not self.cache_path.is_file(): return None try: data = json.loads(self.cache_path.read_text(encoding="utf-8")) return LicenseRecord( code=str(data.get("code") or ""), device_id=str(data.get("device_id") or self._device_id), plan=str(data.get("plan") or "monthly"), expires_at=str(data.get("expires_at") or ""), activated_at=str(data.get("activated_at") or ""), period_label=str(data.get("period_label") or ""), ) except Exception: return None def save(self, record: LicenseRecord) -> None: self.cache_path.write_text( json.dumps(asdict(record), ensure_ascii=False, indent=2), encoding="utf-8", ) def clear(self) -> None: if self.cache_path.is_file(): self.cache_path.unlink() def activate(self, code: str) -> LicenseRecord: normalized = normalize_activation_code(code) if self.config.provider == "http": record = self._activate_http(normalized) else: record = self._activate_builtin(normalized) self.save(record) return record def _activate_builtin(self, code: str) -> LicenseRecord: entry = lookup_builtin_code(code) if entry is None: raise LicenseError("激活码无效，请核对后重试。") now = datetime.now(timezone.utc) if is_builtin_code_expired(entry, now=now): raise LicenseError( f"激活码已过期作废（适用月份 {entry.label}，" f"已于 {entry.expires_at[:10]} 失效）。请使用当月有效激活码。" ) if not is_builtin_code_current_month(entry, now=now): current = current_period_label(now=now) if entry.label > current: raise LicenseError( f"该激活码适用于 {entry.label}，尚未到可用时间。" f"请在本月（{current}）使用对应月份的激活码。" ) raise LicenseError( f"该激活码仅适用于 {entry.label}，当前为 {current}。" "请使用当月激活码。" ) return LicenseRecord( code=entry.code, device_id=self._device_id, plan="monthly", expires_at=entry.expires_at, activated_at=now.isoformat(), period_label=entry.label, ) def _activate_http(self, code: str) -> LicenseRecord: url = f"{self.config.api_base_url.rstrip('/')}/api/v1/license/activate" r = requests.post( url, json={"code": code, "device_id": self._device_id}, timeout=30, ) body: dict[str, Any] = r.json() if r.status_code >= 400 or not body.get("ok", body.get("success")): raise LicenseError(body.get("message") or body.get("error") or "激活失败") data = body.get("data", body) return LicenseRecord( code=code, device_id=self._device_id, plan=str(data.get("plan", "monthly")), expires_at=str(data["expires_at"]), activated_at=str(data.get("activated_at", datetime.now(timezone.utc).isoformat())), period_label=str(data.get("period_label") or ""), ) def ensure_valid(self) -> LicenseRecord: record = self.load_cached() if record is None: raise LicenseRequiredError( "未激活。请在「系统设置」中输入有效激活码完成授权。" ) if record.is_expired(): self.clear() period = f"（{record.period_label}）" if record.period_label else "" raise LicenseRequiredError( f"授权已过期作废{period}，到期日 {record.expires_at[:10]}。" "请使用新的月度激活码重新激活。" ) return record def status_text(self) -> str: record = self.load_cached() if record is None: return "授权: 未激活" if record.is_expired(): period = f" {record.period_label}" if record.period_label else "" return f"授权: 已过期作废{period}（{record.expires_at[:10]}）" period = f" | 月份 {record.period_label}" if record.period_label else "" code_line = f"\n激活码: {record.code}" if record.code else "" return ( f"授权: 有效{period} | 剩余 {record.days_remaining()} 天 " f"| 到期 {record.expires_at[:10]} | 设备 {record.device_id[:8]}…" f"{code_line}" )