AI Assistants

from __future__ import annotations

import json
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path

from ..core.config import get_settings

SAFE_TOOL_NAME = re.compile(r"^[A-Za-z0-9_.:-]{1,100}$")
SAFE_SKILL_NAME = re.compile(r"^[A-Za-z0-9_.:/@+-]{1,300}$")


@dataclass(slots=True)
class CommandResult:
    ok: bool
    output: str


def list_installed_skills() -> list[dict]:
    skills_root = get_settings().hermes_workspace_root / ".hermes" / "skills"
    if not skills_root.exists():
        return []

    skills: list[dict] = []
    for skill_file in skills_root.rglob("SKILL.md"):
        if any(part.startswith(".") for part in skill_file.relative_to(skills_root).parts):
            continue
        relative = skill_file.relative_to(skills_root)
        category = relative.parts[0] if len(relative.parts) > 2 else "uncategorized"
        metadata = _read_skill_frontmatter(skill_file)
        skills.append(
            {
                "name": metadata.get("name") or skill_file.parent.name,
                "description": metadata.get("description") or "No description available.",
                "version": metadata.get("version") or "-",
                "category": category,
                "path": str(skill_file.parent),
            }
        )
    return sorted(skills, key=lambda item: (item["category"], item["name"].lower()))


def search_skills(query: str, limit: int = 10) -> list[dict]:
    cleaned = query.strip()
    if not cleaned or len(cleaned) > 120:
        return []
    result = run_hermes_command(["skills", "search", cleaned, "--limit", str(max(1, min(limit, 20))), "--json"], timeout=45)
    if not result.ok:
        return []
    try:
        data = json.loads(result.output)
    except json.JSONDecodeError:
        return []
    return [
        {
            "name": str(item.get("name") or ""),
            "identifier": str(item.get("identifier") or ""),
            "source": str(item.get("source") or ""),
            "trust_level": str(item.get("trust_level") or ""),
            "description": str(item.get("description") or ""),
        }
        for item in data
        if item.get("identifier")
    ]


def install_skill(identifier: str) -> CommandResult:
    cleaned = identifier.strip()
    if not SAFE_SKILL_NAME.fullmatch(cleaned):
        return CommandResult(False, "Skill identifier tidak valid.")
    return run_hermes_command(["skills", "install", cleaned, "--yes"], timeout=180)


def uninstall_skill(name: str) -> CommandResult:
    cleaned = name.strip()
    if not SAFE_SKILL_NAME.fullmatch(cleaned):
        return CommandResult(False, "Skill name tidak valid.")
    return run_hermes_command(["skills", "uninstall", cleaned], timeout=90)


def list_tools() -> list[dict]:
    result = run_hermes_command(["tools", "list", "--platform", "cli"], timeout=30)
    if not result.ok:
        return []
    tools: list[dict] = []
    pattern = re.compile(r"^\s*([✓✗])\s+(enabled|disabled)\s+([A-Za-z0-9_.:-]+)\s+(.*)$")
    for line in result.output.splitlines():
        match = pattern.match(line)
        if not match:
            continue
        tools.append(
            {
                "enabled": match.group(2) == "enabled",
                "name": match.group(3),
                "description": match.group(4).strip(),
            }
        )
    return tools


def set_tool_enabled(name: str, enabled: bool) -> CommandResult:
    cleaned = name.strip()
    if not SAFE_TOOL_NAME.fullmatch(cleaned):
        return CommandResult(False, "Tool name tidak valid.")
    action = "enable" if enabled else "disable"
    return run_hermes_command(["tools", action, "--platform", "cli", cleaned], timeout=60)


def hermes_overview() -> dict:
    status = run_hermes_command(["status", "--all"], timeout=45)
    memory = run_hermes_command(["memory", "status"], timeout=30)
    config = _redact_command_output(run_hermes_command(["config", "show"], timeout=30).output)
    return {
        "status": _redact_command_output(status.output) if status.output else "Status unavailable.",
        "memory": _redact_command_output(memory.output) if memory.output else "Memory status unavailable.",
        "config": config or "Config unavailable.",
        "storage": hermes_storage_overview(),
    }


def hermes_storage_overview() -> str:
    hermes_root = get_settings().hermes_workspace_root / ".hermes"
    sections = ("skills", "sessions", "memories", "logs", "hooks", "cron", "cache", "work-logs")
    lines = [f"Hermes home: {hermes_root}"]
    for name in sections:
        path = hermes_root / name
        file_count = 0
        total_size = 0
        if path.exists():
            for item in path.rglob("*"):
                if item.is_file() and not item.is_symlink():
                    try:
                        total_size += item.stat().st_size
                        file_count += 1
                    except OSError:
                        continue
        lines.append(f"{name}: {file_count} files, {_format_size(total_size)}")
    return "\n".join(lines)


def run_hermes_command(arguments: list[str], *, timeout: int) -> CommandResult:
    settings = get_settings()
    command = [settings.hermes_command, *arguments]
    try:
        completed = subprocess.run(
            command,
            cwd=settings.hermes_workspace_root,
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
            env=None,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        return CommandResult(False, f"Hermes command gagal: {exc}")
    output = "\n".join(part for part in (completed.stdout.strip(), completed.stderr.strip()) if part).strip()
    return CommandResult(completed.returncode == 0, _redact_command_output(output))


def _read_skill_frontmatter(skill_file: Path) -> dict[str, str]:
    text = skill_file.read_text(encoding="utf-8", errors="replace")
    if not text.startswith("---"):
        return {}
    parts = text.split("---", 2)
    if len(parts) < 3:
        return {}
    metadata: dict[str, str] = {}
    for line in parts[1].splitlines():
        if ":" not in line or line.startswith((" ", "\t")):
            continue
        key, value = line.split(":", 1)
        if key.strip() in {"name", "description", "version"}:
            metadata[key.strip()] = value.strip().strip("'\"")
    return metadata


def _redact_command_output(text: str) -> str:
    cleaned = re.sub(r"\b(sk-[A-Za-z0-9_-]{6})[A-Za-z0-9_-]+", r"\1...[redacted]", text)
    cleaned = re.sub(
        r"(?im)^(\s*(?:api[_ -]?key|token|secret|password)\s*[:=]\s*).+$",
        r"\1[redacted]",
        cleaned,
    )
    return cleaned[-30_000:]


def _format_size(size_bytes: int) -> str:
    size = float(max(0, size_bytes))
    for unit in ("B", "KB", "MB", "GB"):
        if size < 1024 or unit == "GB":
            return f"{size:.0f} {unit}" if unit == "B" else f"{size:.1f} {unit}"
        size /= 1024
    return f"{size:.1f} GB"