ForcePilot/backend/package/yuxi/channel/hooks/hook_parser.py

import logging
import os
import re
import sys
import time
from dataclasses import dataclass
from pathlib import Path

import yaml

from yuxi.channel.hooks.lifecycle import HookEvent
from yuxi.channel.hooks.types import HookConfigField, HookDependencies, HookMetadata

logger = logging.getLogger(__name__)

_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n?(.*)", re.DOTALL)

_REQUIRED_FIELDS = {"hook_id", "name", "events"}
_VALID_CONFIG_TYPES = {"str", "int", "float", "bool", "list", "dict"}

_OS_ALIASES = {
    "win32": "windows",
    "cygwin": "windows",
    "linux": "linux",
    "darwin": "darwin",
    "macos": "darwin",
}


def _normalize_os(name: str) -> str:
    return _OS_ALIASES.get(name.lower(), name.lower())


def _current_os() -> str:
    return _normalize_os(sys.platform)


class HookParseError(Exception):
    pass


class HookValidationError(Exception):
    pass


def parse_hook_md(content: str, source_path: Path | str = "<unknown>") -> HookMetadata:
    m = _FRONTMATTER_RE.match(content)
    if not m:
        raise HookParseError(f"HOOK.md missing frontmatter: {source_path}")

    raw_yaml = m.group(1)
    body = m.group(2).strip()

    try:
        data = yaml.safe_load(raw_yaml)
    except yaml.YAMLError as e:
        raise HookParseError(f"HOOK.md invalid YAML frontmatter: {source_path}: {e}") from e

    if not isinstance(data, dict):
        raise HookParseError(f"HOOK.md frontmatter is not a dict: {source_path}")

    return _build_metadata(data, body, str(source_path))


def _build_metadata(data: dict, body: str, source_dir: str) -> HookMetadata:
    missing = _REQUIRED_FIELDS - set(data.keys())
    if missing:
        raise HookValidationError(f"HOOK.md missing required fields: {sorted(missing)} (source: {source_dir})")

    hook_id = data["hook_id"]
    name = data["name"]
    events = data["events"]
    if not isinstance(hook_id, str):
        raise HookValidationError(f"HOOK.md hook_id must be a string: {source_dir}")
    if not isinstance(name, str):
        raise HookValidationError(f"HOOK.md name must be a string: {source_dir}")
    if not isinstance(events, list):
        raise HookValidationError(f"HOOK.md events must be a list: {source_dir}")

    valid_event_values = {e.value for e in HookEvent}
    invalid_events = [e for e in events if e not in valid_event_values]
    if invalid_events:
        raise HookValidationError(
            f"HOOK.md contains invalid events: {invalid_events}. "
            f"Valid events: {sorted(valid_event_values)}. (source: {source_dir})"
        )

    description = data.get("description", "")
    if not isinstance(description, str):
        description = str(description)

    os_list = data.get("os", [])
    if isinstance(os_list, list):
        os_list = [_normalize_os(o) for o in os_list if isinstance(o, str)]
    else:
        os_list = []

    depends_raw = data.get("depends", {})
    if not isinstance(depends_raw, dict):
        depends_raw = {}
    depends = HookDependencies(
        bin=list(depends_raw.get("bin", []) or []),
        env=list(depends_raw.get("env", []) or []),
    )

    config_raw = data.get("config", {})
    if not isinstance(config_raw, dict):
        config_raw = {}
    config_fields: dict[str, HookConfigField] = {}
    for key, val in config_raw.items():
        if not isinstance(val, dict):
            continue
        field_type = val.get("type", "str")
        if field_type not in _VALID_CONFIG_TYPES:
            field_type = "str"
        config_fields[key] = HookConfigField(
            type=field_type,
            default=val.get("default"),
            description=val.get("description", ""),
        )

    return HookMetadata(
        hook_id=hook_id,
        name=name,
        events=events,
        description=description,
        os=os_list,
        depends=depends,
        config=config_fields,
        body=body,
        source_dir=source_dir,
    )


def validate_os_compatibility(metadata: HookMetadata) -> bool:
    if not metadata.os:
        return True
    return _current_os() in metadata.os


def validate_dependencies(metadata: HookMetadata) -> tuple[bool, list[str]]:
    failures: list[str] = []

    for bin_name in metadata.depends.bin:
        if not _find_bin(bin_name):
            failures.append(f"binary not found: {bin_name}")

    for env_var in metadata.depends.env:
        if env_var not in os.environ:
            failures.append(f"env variable not set: {env_var}")

    return len(failures) == 0, failures


_CACHE_TTL = 300


@dataclass
class _BinCacheEntry:
    value: bool
    timestamp: float


_find_bin_cache: dict[str, _BinCacheEntry] = {}


def clear_bin_cache() -> None:
    _find_bin_cache.clear()


def _find_bin(name: str) -> bool:
    entry = _find_bin_cache.get(name)
    if entry is not None:
        if time.monotonic() - entry.timestamp < _CACHE_TTL:
            return entry.value
        del _find_bin_cache[name]

    if os.name == "nt":
        names = [name, f"{name}.exe", f"{name}.cmd", f"{name}.bat"]
    else:
        names = [name]

    for p in os.environ.get("PATH", "").split(os.pathsep):
        for n in names:
            candidate = os.path.join(p, n)
            if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
                _find_bin_cache[name] = _BinCacheEntry(value=True, timestamp=time.monotonic())
                return True

    _find_bin_cache[name] = _BinCacheEntry(value=False, timestamp=time.monotonic())
    return False