"""Skills loader for loading and managing skills from Go backend.""" import logging from dataclasses import dataclass from pathlib import Path from typing import Any import aiohttp logger = logging.getLogger(__name__) @dataclass class Skill: """Skill data model.""" id: str name: str description: str skill_type: str # system/user status: str # active/inactive path: str content: str = "" class SkillsLoader: """Loads skills from Go backend API and local file system.""" def __init__(self, base_url: str): """Initialize skills loader. Args: base_url: Go backend API base URL """ self.base_url = base_url.rstrip("/") self._session = None self._skills_cache: dict[str, Skill] = {} async def _get_session(self) -> aiohttp.ClientSession: """Get or create aiohttp session.""" if self._session is None or self._session.closed: self._session = aiohttp.ClientSession() return self._session async def close(self) -> None: """Close the session.""" if self._session and not self._session.closed: await self._session.close() async def list_skills(self, skill_type: str | None = None) -> list[Skill]: """List all skills from Go backend. Args: skill_type: Optional filter by skill type (system/user) Returns: List of skills """ url = f"{self.base_url}/api/skill/list" params = {} if skill_type: params["type"] = skill_type try: session = await self._get_session() async with session.get(url, params=params) as response: if response.status == 200: result = await response.json() skills_list = result.get("list", []) skills = [] for s in skills_list: skill = Skill( id=s.get("id", ""), name=s.get("skill_name", ""), description=s.get("skill_desc", ""), skill_type=s.get("skill_type", "user"), status=s.get("status", "inactive"), path=s.get("path", ""), ) skills.append(skill) return skills logger.warning(f"Failed to list skills: {response.status}") return [] except Exception as e: logger.error(f"Error listing skills: {e}") return [] async def get_skill(self, skill_id: str) -> Skill | None: """Get a skill by ID. Args: skill_id: Skill ID Returns: Skill object or None if not found """ # Check cache first if skill_id in self._skills_cache: return self._skills_cache[skill_id] url = f"{self.base_url}/api/skill/{skill_id}" try: session = await self._get_session() async with session.get(url) as response: if response.status == 200: result = await response.json() skill_data = result.get("skill", {}) skill = Skill( id=skill_data.get("id", ""), name=skill_data.get("skill_name", ""), description=skill_data.get("skill_desc", ""), skill_type=skill_data.get("skill_type", "user"), status=skill_data.get("status", "inactive"), path=skill_data.get("path", ""), ) self._skills_cache[skill_id] = skill return skill return None except Exception as e: logger.error(f"Error getting skill {skill_id}: {e}") return None async def get_skill_content(self, skill_id: str) -> str | None: """Get skill content (SKILL.md file content). Args: skill_id: Skill ID Returns: Skill content as string, or None if failed """ url = f"{self.base_url}/api/skill/content" params = {"id": skill_id} try: session = await self._get_session() async with session.get(url, params=params) as response: if response.status == 200: content = await response.text() return content logger.warning(f"Failed to get skill content: {response.status}") return None except Exception as e: logger.error(f"Error getting skill content: {e}") return None async def sync_skills(self) -> int: """Manually trigger skills sync from file system. Returns: Number of skills synced """ url = f"{self.base_url}/api/skill/sync" try: session = await self._get_session() async with session.get(url) as response: if response.status == 200: result = await response.json() count = result.get("count", 0) logger.info(f"Synced {count} skills") return count return 0 except Exception as e: logger.error(f"Error syncing skills: {e}") return 0 async def load_skill_with_content(self, skill_id: str) -> Skill | None: """Load skill with its content. Args: skill_id: Skill ID Returns: Skill object with content, or None if failed """ skill = await self.get_skill(skill_id) if skill: content = await self.get_skill_content(skill_id) if content: skill.content = content return skill def load_skill_from_file(self, file_path: str | Path) -> Skill | None: """Load skill from local file system. Args: file_path: Path to SKILL.md file Returns: Skill object or None if failed """ path = Path(file_path) if not path.exists(): logger.warning(f"Skill file not found: {path}") return None try: content = path.read_text(encoding="utf-8") # Parse frontmatter name, description = self._parse_frontmatter(content) return Skill( id="", name=name or path.stem, description=description or "", skill_type="user", status="active", path=str(path), content=content, ) except Exception as e: logger.error(f"Error loading skill from file: {e}") return None def _parse_frontmatter(self, content: str) -> tuple[str | None, str | None]: """Parse YAML frontmatter from skill content. Args: content: Skill markdown content Returns: Tuple of (name, description) """ import re if not content.startswith("---"): return None, None lines = content.split("\n") end_idx = 0 for i in range(1, len(lines)): if lines[i].strip() == "---": end_idx = i break if end_idx == 0: return None, None yaml_content = "\n".join(lines[1:end_idx]) name_match = re.search(r"name:\s*(.+)", yaml_content) name = name_match.group(1).strip() if name_match else None desc_match = re.search(r"description:\s*(.+)", yaml_content) description = desc_match.group(1).strip() if desc_match else None return name, description def clear_cache(self) -> None: """Clear skills cache.""" self._skills_cache.clear()