zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
200 lines
7.1 KiB
Python
200 lines
7.1 KiB
Python
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}"
|
||
)
|