eigent/backend/app/controller/skill_controller.py

# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========

import logging
from typing import Annotated

from fastapi import APIRouter, File, Form, HTTPException, Query, UploadFile

from app.service.skill_config_service import (
    skill_config_delete,
    skill_config_init,
    skill_config_load,
    skill_config_toggle,
    skill_config_update,
)
from app.service.skill_service import (
    skill_delete,
    skill_get_path_by_name,
    skill_import_zip,
    skill_list_files,
    skill_read,
    skill_write,
    skills_scan,
)

router = APIRouter()
skill_logger = logging.getLogger("skill_controller")


# --- Skill config (must be before /skills/{skill_dir_name} to avoid path conflict) ---


@router.get("/skills/config")
def skill_config_get(user_id: str = Query(..., description="User ID")) -> dict:
    """Load skills config for user."""
    config = skill_config_load(user_id)
    return {"success": True, "config": config}


@router.post("/skills/config/init")
def skill_config_init_endpoint(body: dict) -> dict:
    """Initialize skills config for user (merge default if present)."""
    user_id = body.get("user_id")
    if not user_id:
        raise HTTPException(status_code=400, detail="user_id is required")
    config = skill_config_init(user_id)
    return {"success": True, "config": config}


@router.put("/skills/config/{skill_name}")
def skill_config_update_endpoint(skill_name: str, body: dict) -> dict:
    """Update config for a skill."""
    user_id = body.get("user_id")
    if not user_id:
        raise HTTPException(status_code=400, detail="user_id is required")
    skill_config = {k: v for k, v in body.items() if k != "user_id"}
    skill_config_update(user_id, skill_name, skill_config)
    return {"success": True}


@router.delete("/skills/config/{skill_name}")
def skill_config_delete_endpoint(
    skill_name: str, user_id: str = Query(..., description="User ID")
) -> dict:
    """Remove skill from config."""
    skill_config_delete(user_id, skill_name)
    return {"success": True}


@router.post("/skills/config/{skill_name}/toggle")
def skill_config_toggle_endpoint(skill_name: str, body: dict) -> dict:
    """Toggle skill enabled state."""
    user_id = body.get("user_id")
    enabled = body.get("enabled")
    if not user_id:
        raise HTTPException(status_code=400, detail="user_id is required")
    if enabled is None:
        raise HTTPException(status_code=400, detail="enabled is required")
    result = skill_config_toggle(user_id, skill_name, bool(enabled))
    return {"success": True, "config": result}


# --- Skills CRUD ---


@router.post("/skills/import")
async def skill_import_endpoint(
    file: Annotated[
        UploadFile, File(description="Zip file containing SKILL.md")
    ],
    replacements: Annotated[
        str | None, Form(description="Comma-separated folder names to replace")
    ] = None,
) -> dict:
    """Import skills from a zip archive. Returns {success, error?, conflicts?}."""
    if not file.filename or not file.filename.lower().endswith(".zip"):
        raise HTTPException(
            status_code=400, detail="File must be a .zip archive"
        )
    try:
        zip_bytes = await file.read()
    except Exception:
        raise HTTPException(
            status_code=400, detail="Failed to read uploaded file"
        )
    repl_list = (
        [s for s in (s.strip() for s in replacements.split(",")) if s]
        if replacements
        else None
    )
    result = skill_import_zip(zip_bytes, repl_list)
    if not result.get("success") and "conflicts" not in result:
        raise HTTPException(
            status_code=400,
            detail=result.get("error", "Import failed"),
        )
    return result


@router.get("/skills/path")
def skill_get_path(
    name: str = Query(..., description="Skill display name"),
) -> dict:
    """Get absolute directory path for a skill by name. For reveal-in-folder."""
    path_val = skill_get_path_by_name(name)
    if path_val is None:
        raise HTTPException(status_code=404, detail=f"Skill not found: {name}")
    return {"path": path_val}


@router.get("/skills")
def skills_list() -> dict:
    """Scan and list all skills."""
    skills = skills_scan()
    return {"success": True, "skills": skills}


@router.post("/skills/{skill_dir_name}")
def skill_create(skill_dir_name: str, body: dict) -> dict:
    """Create or overwrite skill. Body: { content }."""
    content = body.get("content", "")
    try:
        skill_write(skill_dir_name, content)
        skill_logger.info("Skill written: %s", skill_dir_name)
        return {"success": True}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except PermissionError as e:
        raise HTTPException(status_code=403, detail=str(e))


@router.get("/skills/{skill_dir_name}")
def skill_get(skill_dir_name: str) -> dict:
    """Read skill content."""
    try:
        content = skill_read(skill_dir_name)
        return {"success": True, "content": content}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except PermissionError as e:
        raise HTTPException(status_code=403, detail=str(e))
    except FileNotFoundError:
        raise HTTPException(status_code=404, detail="Skill not found")


@router.delete("/skills/{skill_dir_name}")
def skill_remove(skill_dir_name: str) -> dict:
    """Delete skill."""
    try:
        skill_delete(skill_dir_name)
        skill_logger.info("Skill deleted: %s", skill_dir_name)
        return {"success": True}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except PermissionError as e:
        raise HTTPException(status_code=403, detail=str(e))


@router.get("/skills/{skill_dir_name}/files")
def skill_files(skill_dir_name: str) -> dict:
    """List files in skill directory."""
    try:
        files = skill_list_files(skill_dir_name)
        return {"success": True, "files": files}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except PermissionError as e:
        raise HTTPException(status_code=403, detail=str(e))