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