zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
99 lines
2.9 KiB
Python
99 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from xhs_bridge.config import load_config
|
|
|
|
|
|
@dataclass
|
|
class CustomCommentItem:
|
|
id: str
|
|
text: str
|
|
|
|
@staticmethod
|
|
def from_dict(raw: dict[str, Any]) -> CustomCommentItem:
|
|
return CustomCommentItem(
|
|
id=str(raw.get("id") or uuid.uuid4().hex[:12]),
|
|
text=str(raw.get("text") or "").strip(),
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, str]:
|
|
return {"id": self.id, "text": self.text}
|
|
|
|
|
|
class CustomCommentsStore:
|
|
"""持久化自定义评论模板库。"""
|
|
|
|
_instance: CustomCommentsStore | None = None
|
|
|
|
def __init__(self, path: Path | None = None) -> None:
|
|
if path is None:
|
|
path = load_config().state_dir / "custom_comments.json"
|
|
self._path = path
|
|
self._items: list[CustomCommentItem] = []
|
|
self._load()
|
|
|
|
@classmethod
|
|
def instance(cls) -> CustomCommentsStore:
|
|
if cls._instance is None:
|
|
cls._instance = cls()
|
|
return cls._instance
|
|
|
|
def _load(self) -> None:
|
|
if not self._path.is_file():
|
|
self._items = []
|
|
return
|
|
try:
|
|
data = json.loads(self._path.read_text(encoding="utf-8"))
|
|
raw_list = data.get("items") if isinstance(data, dict) else data
|
|
if not isinstance(raw_list, list):
|
|
self._items = []
|
|
return
|
|
self._items = [
|
|
CustomCommentItem.from_dict(x)
|
|
for x in raw_list
|
|
if isinstance(x, dict) and str(x.get("text") or "").strip()
|
|
]
|
|
except (OSError, json.JSONDecodeError):
|
|
self._items = []
|
|
|
|
def _save(self) -> None:
|
|
self._path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = {"items": [it.to_dict() for it in self._items]}
|
|
self._path.write_text(
|
|
json.dumps(payload, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def list_items(self) -> list[CustomCommentItem]:
|
|
return list(self._items)
|
|
|
|
def texts(self) -> list[str]:
|
|
return [it.text for it in self._items if it.text]
|
|
|
|
def replace_all(self, texts: list[str]) -> None:
|
|
seen: list[CustomCommentItem] = []
|
|
for text in texts:
|
|
t = str(text).strip()
|
|
if t:
|
|
seen.append(CustomCommentItem(id=uuid.uuid4().hex[:12], text=t))
|
|
self._items = seen
|
|
self._save()
|
|
|
|
def add(self, text: str) -> CustomCommentItem:
|
|
t = str(text).strip()
|
|
if not t:
|
|
raise ValueError("评论内容不能为空")
|
|
item = CustomCommentItem(id=uuid.uuid4().hex[:12], text=t)
|
|
self._items.append(item)
|
|
self._save()
|
|
return item
|
|
|
|
def remove(self, item_id: str) -> None:
|
|
self._items = [it for it in self._items if it.id != item_id]
|
|
self._save()
|