evyzacq
109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
# Skill 管理服务：扫描、验证、列出所有可用 Skill
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from server.config import settings
|
|
|
|
logger = logging.getLogger("testflow")
|
|
|
|
REQUIRED_SKILL_FILES = ["skill.yaml", "ir_schema.json", "extract_ir_prompt.j2", "gen_cases_prompt.j2"]
|
|
|
|
|
|
class SkillManager:
|
|
"""Discover and validate skill packages from .zeekerwatchmen/skills/."""
|
|
|
|
def __init__(self):
|
|
self._skills_dir = settings.SKILLS_DIR
|
|
self._cache: dict[str, dict] = {}
|
|
self._scan()
|
|
|
|
def _scan(self):
|
|
"""Scan the skills directory and cache valid skills."""
|
|
self._cache = {}
|
|
if not self._skills_dir.exists():
|
|
logger.warning("[SKILL] s directory not found: %s", self._skills_dir)
|
|
return
|
|
|
|
for skill_dir in sorted(self._skills_dir.iterdir()):
|
|
if not skill_dir.is_dir():
|
|
continue
|
|
|
|
skill_name = skill_dir.name
|
|
skill = self._load_skill(skill_name)
|
|
if skill:
|
|
self._cache[skill_name] = skill
|
|
logger.info("[SKILL] 加载: %s (v%s)", skill_name, skill.get("version", "?"))
|
|
|
|
logger.info("[SKILL] 发现 %d skills: %s", len(self._cache), list(self._cache.keys()))
|
|
|
|
def _load_skill(self, skill_name: str) -> dict | None:
|
|
"""Load and validate a single skill package."""
|
|
skill_dir = self._skills_dir / skill_name
|
|
|
|
# Check required files
|
|
missing = []
|
|
for filename in REQUIRED_SKILL_FILES:
|
|
if not (skill_dir / filename).exists():
|
|
missing.append(filename)
|
|
|
|
if missing:
|
|
logger.warning("[SKILL] '%s' missing files: %s", skill_name, missing)
|
|
return None
|
|
|
|
# Load metadata
|
|
try:
|
|
with open(skill_dir / "skill.yaml", "r", encoding="utf-8") as f:
|
|
meta = yaml.safe_load(f) or {}
|
|
except yaml.YAMLError as e:
|
|
logger.error("[SKILL] '%s': invalid skill.yaml: %s", skill_name, e)
|
|
return None
|
|
|
|
# Load schema
|
|
try:
|
|
with open(skill_dir / "ir_schema.json", "r", encoding="utf-8") as f:
|
|
schema = json.load(f)
|
|
except (json.JSONDecodeError, IOError) as e:
|
|
logger.error("[SKILL] '%s': invalid ir_schema.json: %s", skill_name, e)
|
|
return None
|
|
|
|
return {
|
|
"name": skill_name,
|
|
"display_name": meta.get("display_name", skill_name),
|
|
"description": meta.get("description", ""),
|
|
"version": meta.get("version", "0.0.0"),
|
|
"domain": meta.get("domain", "general"),
|
|
"author": meta.get("author", ""),
|
|
"auto_match": meta.get("auto_match", {}),
|
|
"schema_fields": list(schema.get("properties", {}).get("features", {}).get("items", {}).get("properties", {}).keys()),
|
|
}
|
|
|
|
def list_skills(self) -> list[dict]:
|
|
"""Return metadata for all discovered skills."""
|
|
return [
|
|
{
|
|
"name": s["name"],
|
|
"display_name": s["display_name"],
|
|
"description": s["description"],
|
|
"version": s["version"],
|
|
"domain": s["domain"],
|
|
}
|
|
for s in self._cache.values()
|
|
]
|
|
|
|
def get_skill(self, name: str) -> dict | None:
|
|
"""Get a specific skill's full metadata."""
|
|
return self._cache.get(name)
|
|
|
|
def reload(self):
|
|
"""Re-scan the skills directory (for hot-reload)."""
|
|
logger.info("[SKILL] 热重载 skills...")
|
|
self._cache = {}
|
|
self._scan()
|
|
|
|
|
|
skill_manager = SkillManager()
|