skill_processor.py

# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
# SPDX-License-Identifier: Apache-2.0
"""
Skill Processor for OpenViking.

Handles skill parsing, LLM generation, and storage operations.
"""

import tempfile
import zipfile
from pathlib import Path
from typing import Any, Dict, List, Optional

from openviking.core.context import Context, ContextType, Vectorize
from openviking.core.mcp_converter import is_mcp_format, mcp_to_skill
from openviking.core.skill_loader import SkillLoader
from openviking.server.identity import RequestContext
from openviking.storage import VikingDBManager
from openviking.storage.queuefs.embedding_msg_converter import EmbeddingMsgConverter
from openviking.storage.viking_fs import VikingFS
from openviking_cli.utils import get_logger
from openviking_cli.utils.config import get_openviking_config

logger = get_logger(__name__)


class SkillProcessor:
    """
    Handles skill processing and storage.

    Workflow:
    1. Parse skill data (directory, file, string, or dict)
    2. Generate L1 overview using VLM
    3. Write skill content to VikingFS
    4. Write auxiliary files
    5. Index to vector store
    """

    def __init__(self, vikingdb: VikingDBManager):
        """Initialize skill processor."""
        self.vikingdb = vikingdb

    async def process_skill(
        self,
        data: Any,
        viking_fs: VikingFS,
        ctx: RequestContext,
    ) -> Dict[str, Any]:
        """
        Process and store a skill.

        Args:
            data: Skill data (directory path, file path, string, or dict)
            viking_fs: VikingFS instance for storage
            user: Username for context

        Returns:
            Processing result with status and metadata
        """

        config = get_openviking_config()

        skill_dict, auxiliary_files, base_path = self._parse_skill(data)

        context = Context(
            uri=f"viking://agent/skills/{skill_dict['name']}",
            parent_uri="viking://agent/skills",
            is_leaf=False,
            abstract=skill_dict.get("description", ""),
            context_type=ContextType.SKILL.value,
            user=ctx.user,
            account_id=ctx.account_id,
            owner_space=ctx.user.agent_space_name(),
            meta={
                "name": skill_dict["name"],
                "description": skill_dict.get("description", ""),
                "allowed_tools": skill_dict.get("allowed_tools", []),
                "tags": skill_dict.get("tags", []),
                "source_path": skill_dict.get("source_path", ""),
            },
        )
        context.set_vectorize(Vectorize(text=context.abstract))

        overview = await self._generate_overview(skill_dict, config)

        skill_dir_uri = f"viking://agent/skills/{context.meta['name']}"

        await self._write_skill_content(
            viking_fs=viking_fs,
            skill_dict=skill_dict,
            skill_dir_uri=skill_dir_uri,
            overview=overview,
            ctx=ctx,
        )

        await self._write_auxiliary_files(
            viking_fs=viking_fs,
            auxiliary_files=auxiliary_files,
            base_path=base_path,
            skill_dir_uri=skill_dir_uri,
            ctx=ctx,
        )

        await self._index_skill(
            context=context,
            skill_dir_uri=skill_dir_uri,
        )

        return {
            "status": "success",
            "uri": skill_dir_uri,
            "name": skill_dict["name"],
            "auxiliary_files": len(auxiliary_files),
        }

    def _parse_skill(self, data: Any) -> tuple[Dict[str, Any], List[Path], Optional[Path]]:
        """Parse skill data from various formats."""
        auxiliary_files = []
        base_path = None

        if isinstance(data, str):
            path_obj = Path(data)
            if path_obj.exists():
                if zipfile.is_zipfile(path_obj):
                    temp_dir = Path(tempfile.mkdtemp())
                    with zipfile.ZipFile(path_obj, "r") as zipf:
                        zipf.extractall(temp_dir)
                    data = temp_dir
                else:
                    data = path_obj

        if isinstance(data, Path):
            if data.is_dir():
                # Directory containing SKILL.md
                skill_file = data / "SKILL.md"
                if not skill_file.exists():
                    raise ValueError(f"SKILL.md not found in {data}")

                skill_dict = SkillLoader.load(str(skill_file))
                base_path = data
                for item in data.rglob("*"):
                    if item.is_file() and item.name != "SKILL.md":
                        auxiliary_files.append(item)
            else:
                # Single SKILL.md file
                skill_dict = SkillLoader.load(str(data))
        elif isinstance(data, str):
            # Raw SKILL.md content
            skill_dict = SkillLoader.parse(data)
        elif isinstance(data, dict):
            if is_mcp_format(data):
                skill_dict = mcp_to_skill(data)
            else:
                skill_dict = data
        else:
            raise ValueError(f"Unsupported data type: {type(data)}")

        return skill_dict, auxiliary_files, base_path

    async def _generate_overview(self, skill_dict: Dict[str, Any], config) -> str:
        """Generate L1 overview using VLM."""
        from openviking.prompts import render_prompt

        prompt = render_prompt(
            "skill.overview_generation",
            {
                "skill_name": skill_dict["name"],
                "skill_description": skill_dict.get("description", ""),
                "skill_content": skill_dict.get("content", ""),
            },
        )
        return await config.vlm.get_completion_async(prompt)

    async def _write_skill_content(
        self,
        viking_fs: VikingFS,
        skill_dict: Dict[str, Any],
        skill_dir_uri: str,
        overview: str,
        ctx: RequestContext,
    ):
        """Write main skill content to VikingFS."""
        await viking_fs.write_context(
            uri=skill_dir_uri,
            content=skill_dict.get("content", ""),
            abstract=skill_dict.get("description", ""),
            overview=overview,
            content_filename="SKILL.md",
            is_leaf=False,
            ctx=ctx,
        )

    async def _write_auxiliary_files(
        self,
        viking_fs: VikingFS,
        auxiliary_files: List[Path],
        base_path: Optional[Path],
        skill_dir_uri: str,
        ctx: RequestContext,
    ):
        """Write auxiliary files to VikingFS."""
        for aux_file in auxiliary_files:
            if base_path:
                rel_path = aux_file.relative_to(base_path)
                aux_uri = f"{skill_dir_uri}/{rel_path}"
            else:
                aux_uri = f"{skill_dir_uri}/{aux_file.name}"

            file_bytes = aux_file.read_bytes()
            try:
                file_bytes.decode("utf-8")
                is_text = True
            except UnicodeDecodeError:
                is_text = False

            if is_text:
                await viking_fs.write_file(aux_uri, file_bytes.decode("utf-8"), ctx=ctx)
            else:
                await viking_fs.write_file_bytes(aux_uri, file_bytes, ctx=ctx)

    async def _index_skill(self, context: Context, skill_dir_uri: str):
        """Write skill to vector store via async queue."""
        context.uri = skill_dir_uri

        embedding_msg = EmbeddingMsgConverter.from_context(context)
        await self.vikingdb.enqueue_embedding_msg(embedding_msg)