zzh
5a26ae0048
- 新增小红书前端模块（启动服务、内容分析、爆款创作、发布记录、技能卡片、拟人设置、AI 助手） - 新增 xhs-agent-desktop 独立桌面 Agent，对接 OMS 任务调度 - 登录页重构，整合微信登录与 OMS JWT 认证 - 订单、渠道、用户中心等页面与 API 增强 Co-authored-by: Cursor <cursoragent@cursor.com>
234 lines
7.2 KiB
Python
234 lines
7.2 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import uuid
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from xhs_bridge.config import load_config
|
||
from xhs_bridge.skill_catalog import (
|
||
PRESET_TEMPLATES,
|
||
SkillRecipe,
|
||
compose_workflow,
|
||
preset_recipe,
|
||
suggest_skill_title,
|
||
)
|
||
|
||
SKILL_TYPE_CREATE = "ai_create"
|
||
SKILL_TYPE_ANALYZE = "ai_analyze"
|
||
|
||
_SKILL_TYPE_LABELS = {
|
||
SKILL_TYPE_CREATE: "AI创作",
|
||
SKILL_TYPE_ANALYZE: "AI分析",
|
||
}
|
||
|
||
|
||
def skill_type_label(skill_type: str) -> str:
|
||
return _SKILL_TYPE_LABELS.get(skill_type, _SKILL_TYPE_LABELS[SKILL_TYPE_CREATE])
|
||
|
||
|
||
def normalize_skill_type(value: str | None) -> str:
|
||
if value == SKILL_TYPE_ANALYZE:
|
||
return SKILL_TYPE_ANALYZE
|
||
return SKILL_TYPE_CREATE
|
||
|
||
|
||
@dataclass
|
||
class Skill:
|
||
"""运营技能：可自主搭配的爆款配方。"""
|
||
|
||
id: str
|
||
title: str
|
||
recipe: SkillRecipe
|
||
created_at: str
|
||
updated_at: str
|
||
preset_id: str = ""
|
||
skill_type: str = SKILL_TYPE_CREATE
|
||
|
||
@property
|
||
def workflow(self) -> str:
|
||
return compose_workflow(self.recipe)
|
||
|
||
@staticmethod
|
||
def new(
|
||
title: str,
|
||
recipe: SkillRecipe,
|
||
*,
|
||
preset_id: str = "",
|
||
skill_type: str = SKILL_TYPE_CREATE,
|
||
) -> Skill:
|
||
now = datetime.now().isoformat(timespec="seconds")
|
||
name = title.strip() or suggest_skill_title(recipe)
|
||
return Skill(
|
||
id=uuid.uuid4().hex[:12],
|
||
title=name,
|
||
recipe=recipe,
|
||
created_at=now,
|
||
updated_at=now,
|
||
preset_id=preset_id,
|
||
skill_type=normalize_skill_type(skill_type),
|
||
)
|
||
|
||
|
||
class SkillsStore:
|
||
"""技能列表持久化（JSON）。"""
|
||
|
||
def __init__(self, path: Path | None = None) -> None:
|
||
if path is None:
|
||
cfg = load_config()
|
||
path = cfg.state_dir / "skills.json"
|
||
self.path = path
|
||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||
self._skills: list[Skill] = []
|
||
self.load()
|
||
|
||
def load(self) -> list[Skill]:
|
||
if not self.path.is_file():
|
||
self._skills = []
|
||
self.seed_official_presets()
|
||
return list(self._skills)
|
||
try:
|
||
with self.path.open("r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
items = data.get("skills") if isinstance(data, dict) else data
|
||
if not isinstance(items, list):
|
||
self._skills = []
|
||
return list(self._skills)
|
||
self._skills = []
|
||
for raw in items:
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
recipe_raw = raw.get("recipe")
|
||
if recipe_raw:
|
||
recipe = SkillRecipe.from_dict(recipe_raw)
|
||
else:
|
||
recipe = SkillRecipe.empty()
|
||
legacy = str(raw.get("workflow") or "").strip()
|
||
if legacy:
|
||
recipe.notes = legacy
|
||
self._skills.append(
|
||
Skill(
|
||
id=str(raw.get("id") or uuid.uuid4().hex[:12]),
|
||
title=str(raw.get("title") or "未命名技能"),
|
||
recipe=recipe,
|
||
created_at=str(raw.get("created_at") or ""),
|
||
updated_at=str(raw.get("updated_at") or ""),
|
||
preset_id=str(raw.get("preset_id") or ""),
|
||
skill_type=normalize_skill_type(raw.get("skill_type")),
|
||
)
|
||
)
|
||
except (json.JSONDecodeError, OSError):
|
||
self._skills = []
|
||
return list(self._skills)
|
||
|
||
def seed_official_presets(self, *, force: bool = False) -> int:
|
||
"""写入超级内容引擎官方示例模板。force=True 时追加缺失项。"""
|
||
if self._skills and not force:
|
||
return 0
|
||
added = 0
|
||
existing_presets = {s.preset_id for s in self._skills if s.preset_id}
|
||
existing_titles = {s.title for s in self._skills}
|
||
for preset_id, name, _raw in PRESET_TEMPLATES:
|
||
if not force and preset_id in existing_presets:
|
||
continue
|
||
if name in existing_titles:
|
||
continue
|
||
recipe = preset_recipe(preset_id)
|
||
if recipe is None:
|
||
continue
|
||
self._skills.append(
|
||
Skill.new(name, recipe, preset_id=preset_id, skill_type=SKILL_TYPE_CREATE)
|
||
)
|
||
added += 1
|
||
if added or not self.path.is_file():
|
||
self.save()
|
||
return added
|
||
|
||
def save(self) -> None:
|
||
payload: dict[str, Any] = {
|
||
"skills": [
|
||
{
|
||
"id": s.id,
|
||
"title": s.title,
|
||
"recipe": s.recipe.to_dict(),
|
||
"preset_id": s.preset_id,
|
||
"skill_type": s.skill_type,
|
||
"created_at": s.created_at,
|
||
"updated_at": s.updated_at,
|
||
}
|
||
for s in self._skills
|
||
],
|
||
"updated_at": datetime.now().isoformat(timespec="seconds"),
|
||
}
|
||
with self.path.open("w", encoding="utf-8") as f:
|
||
json.dump(payload, f, ensure_ascii=False, indent=2)
|
||
|
||
def list_skills(self) -> list[Skill]:
|
||
return list(self._skills)
|
||
|
||
def list_skills_by_type(self, skill_type: str) -> list[Skill]:
|
||
want = normalize_skill_type(skill_type)
|
||
return [s for s in self._skills if s.skill_type == want]
|
||
|
||
def get(self, skill_id: str) -> Skill | None:
|
||
for s in self._skills:
|
||
if s.id == skill_id:
|
||
return s
|
||
return None
|
||
|
||
def add(
|
||
self,
|
||
title: str,
|
||
recipe: SkillRecipe,
|
||
*,
|
||
preset_id: str = "",
|
||
skill_type: str = SKILL_TYPE_CREATE,
|
||
) -> Skill:
|
||
skill = Skill.new(title, recipe, preset_id=preset_id, skill_type=skill_type)
|
||
self._skills.insert(0, skill)
|
||
self.save()
|
||
return skill
|
||
|
||
def update(
|
||
self,
|
||
skill_id: str,
|
||
title: str,
|
||
recipe: SkillRecipe,
|
||
*,
|
||
skill_type: str | None = None,
|
||
) -> Skill | None:
|
||
for s in self._skills:
|
||
if s.id == skill_id:
|
||
s.title = title.strip() or suggest_skill_title(recipe)
|
||
s.recipe = recipe
|
||
if skill_type is not None:
|
||
s.skill_type = normalize_skill_type(skill_type)
|
||
s.updated_at = datetime.now().isoformat(timespec="seconds")
|
||
self.save()
|
||
return s
|
||
return None
|
||
|
||
def duplicate(self, skill_id: str) -> Skill | None:
|
||
src = self.get(skill_id)
|
||
if not src:
|
||
return None
|
||
copy_recipe = SkillRecipe.from_dict(src.recipe.to_dict())
|
||
skill = Skill.new(
|
||
f"{src.title}（副本）",
|
||
copy_recipe,
|
||
skill_type=src.skill_type,
|
||
)
|
||
self._skills.insert(0, skill)
|
||
self.save()
|
||
return skill
|
||
|
||
def delete(self, skill_id: str) -> bool:
|
||
before = len(self._skills)
|
||
self._skills = [s for s in self._skills if s.id != skill_id]
|
||
if len(self._skills) < before:
|
||
self.save()
|
||
return True
|
||
return False
|