mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-11 12:17:45 +02:00
119 lines
4.2 KiB
Python
119 lines
4.2 KiB
Python
import json
|
|
from backend.apps.skills.SkillStore.Skill import Skill
|
|
from typeguard import typechecked
|
|
from pydantic import BaseModel
|
|
import os
|
|
from typing import Dict, Optional
|
|
|
|
|
|
class SkillStore(BaseModel):
|
|
"""File-backed persistence layer for local skills."""
|
|
skills_dir: str
|
|
index_path: str
|
|
|
|
@typechecked
|
|
def __init__(self, skills_dir: str, index_filename: str = ".skills_index.json"):
|
|
os.makedirs(skills_dir, exist_ok=True)
|
|
super().__init__(
|
|
skills_dir=skills_dir,
|
|
index_path=os.path.join(skills_dir, index_filename),
|
|
)
|
|
|
|
@typechecked
|
|
def p_load_index(self) -> Dict[str, dict]:
|
|
if os.path.exists(self.index_path):
|
|
with open(self.index_path, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
return {}
|
|
|
|
@typechecked
|
|
def p_save_index(self, index: Dict[str, dict]) -> None:
|
|
with open(self.index_path, "w", encoding="utf-8") as f:
|
|
json.dump(index, f, indent=2)
|
|
|
|
@staticmethod
|
|
@typechecked
|
|
def slug(name: str) -> str:
|
|
return name.lower().replace(" ", "-")
|
|
|
|
@typechecked
|
|
def p_skill_path(self, skill_id: str) -> str:
|
|
return os.path.join(self.skills_dir, f"{skill_id}.md")
|
|
|
|
@typechecked
|
|
def list_all(self) -> list[Skill]:
|
|
index = self.p_load_index()
|
|
result: list[Skill] = []
|
|
if not os.path.exists(self.skills_dir):
|
|
return result
|
|
for fname in os.listdir(self.skills_dir):
|
|
if not fname.endswith(".md"):
|
|
continue
|
|
fpath = os.path.join(self.skills_dir, fname)
|
|
with open(fpath, encoding="utf-8") as f:
|
|
content = f.read()
|
|
skill_id = fname.removesuffix(".md")
|
|
meta = index.get(skill_id, {})
|
|
result.append(Skill(
|
|
id=skill_id,
|
|
name=meta.get("name", skill_id.replace("-", " ").replace("_", " ").title()),
|
|
description=meta.get("description", ""),
|
|
content=content,
|
|
file_path=fpath,
|
|
command=meta.get("command", skill_id),
|
|
))
|
|
return result
|
|
|
|
@typechecked
|
|
def get(self, skill_id: str) -> Optional[Skill]:
|
|
for s in self.list_all():
|
|
if s.id == skill_id:
|
|
return s
|
|
return None
|
|
|
|
@typechecked
|
|
def create(self, name: str, description: str, content: str, command: str = "") -> Skill:
|
|
slug = self.slug(name)
|
|
fpath = self.p_skill_path(slug)
|
|
with open(fpath, "w", encoding="utf-8") as f:
|
|
f.write(content)
|
|
index = self.p_load_index()
|
|
index[slug] = {"name": name, "description": description, "command": command or slug}
|
|
self.p_save_index(index)
|
|
return Skill(id=slug, name=name, description=description,
|
|
content=content, file_path=fpath, command=command or slug)
|
|
|
|
@typechecked
|
|
def update(self, skill_id: str, *, name: Optional[str] = None,
|
|
description: Optional[str] = None, content: Optional[str] = None,
|
|
command: Optional[str] = None) -> Skill:
|
|
fpath = self.p_skill_path(skill_id)
|
|
if not os.path.exists(fpath):
|
|
raise FileNotFoundError(skill_id)
|
|
if content is not None:
|
|
with open(fpath, "w", encoding="utf-8") as f:
|
|
f.write(content)
|
|
index = self.p_load_index()
|
|
meta = index.get(skill_id, {})
|
|
if name is not None:
|
|
meta["name"] = name
|
|
if description is not None:
|
|
meta["description"] = description
|
|
if command is not None:
|
|
meta["command"] = command
|
|
index[skill_id] = meta
|
|
self.p_save_index(index)
|
|
with open(fpath, encoding="utf-8") as f:
|
|
content = f.read()
|
|
return Skill(id=skill_id, name=meta.get("name", skill_id),
|
|
description=meta.get("description", ""),
|
|
content=content, file_path=fpath, command=meta.get("command", skill_id))
|
|
|
|
@typechecked
|
|
def delete(self, skill_id: str) -> None:
|
|
fpath = self.p_skill_path(skill_id)
|
|
if os.path.exists(fpath):
|
|
os.remove(fpath)
|
|
index = self.p_load_index()
|
|
index.pop(skill_id, None)
|
|
self.p_save_index(index) |