123 lines
3.2 KiB
Python
123 lines
3.2 KiB
Python
"""Skills catalog 的公共辅助函数。
|
||
|
||
这里专门放“解析和校验 skill 文件”的纯函数，避免 `loader.py` 里同时承担：
|
||
|
||
1. 目录扫描
|
||
2. frontmatter 解析
|
||
3. requirements 校验
|
||
4. 文本裁剪/格式化
|
||
|
||
把这些细节拆出来之后，skills catalog 的边界会更清楚，后面无论是 reviews、publisher
|
||
还是 runtime resolver，都可以复用同一套元数据解析规则。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
from typing import Any
|
||
|
||
|
||
def parse_frontmatter(content: str) -> tuple[dict[str, str], str]:
|
||
"""解析 Markdown 文件顶部的极简 frontmatter。
|
||
|
||
当前先只支持最常见的：
|
||
|
||
```md
|
||
---
|
||
key: value
|
||
key2: value2
|
||
---
|
||
body...
|
||
```
|
||
|
||
这样足够支撑第一版 skills runtime，不提前把 YAML 解析器引进来。
|
||
"""
|
||
|
||
if not content.startswith("---"):
|
||
return {}, content
|
||
|
||
match = re.match(r"^---\n(.*?)\n---\n?", content, re.DOTALL)
|
||
if match is None:
|
||
return {}, content
|
||
|
||
metadata: dict[str, str] = {}
|
||
for line in match.group(1).splitlines():
|
||
if ":" not in line:
|
||
continue
|
||
key, value = line.split(":", 1)
|
||
metadata[key.strip()] = value.strip().strip('"\'')
|
||
body = content[match.end():].strip()
|
||
return metadata, body
|
||
|
||
|
||
def strip_frontmatter(content: str) -> str:
|
||
"""去掉 frontmatter，只保留 skill 正文。"""
|
||
|
||
_, body = parse_frontmatter(content)
|
||
return body
|
||
|
||
|
||
def parse_skill_metadata_blob(raw: str) -> dict[str, Any]:
|
||
"""解析 metadata 字段里的 JSON 扩展配置。
|
||
|
||
为了兼容旧 nanobot 习惯，这里同时支持：
|
||
- `nanobot`
|
||
- `openclaw`
|
||
|
||
第一版主要关心的字段有：
|
||
- `always`
|
||
- `requires`
|
||
"""
|
||
|
||
try:
|
||
data = json.loads(raw)
|
||
except (json.JSONDecodeError, TypeError):
|
||
return {}
|
||
|
||
if not isinstance(data, dict):
|
||
return {}
|
||
nested = data.get("nanobot", data.get("openclaw", data))
|
||
return nested if isinstance(nested, dict) else {}
|
||
|
||
|
||
def check_requirements(metadata: dict[str, Any]) -> bool:
|
||
"""检查 skill 的最小 requirements 是否满足。"""
|
||
|
||
requires = metadata.get("requires", {})
|
||
if not isinstance(requires, dict):
|
||
return True
|
||
|
||
for binary in requires.get("bins", []):
|
||
if not shutil.which(str(binary)):
|
||
return False
|
||
for env_name in requires.get("env", []):
|
||
if not os.environ.get(str(env_name)):
|
||
return False
|
||
return True
|
||
|
||
|
||
def get_missing_requirements(metadata: dict[str, Any]) -> str:
|
||
"""返回缺失 requirements 的简短描述。"""
|
||
|
||
requires = metadata.get("requires", {})
|
||
if not isinstance(requires, dict):
|
||
return ""
|
||
|
||
missing: list[str] = []
|
||
for binary in requires.get("bins", []):
|
||
if not shutil.which(str(binary)):
|
||
missing.append(f"CLI: {binary}")
|
||
for env_name in requires.get("env", []):
|
||
if not os.environ.get(str(env_name)):
|
||
missing.append(f"ENV: {env_name}")
|
||
return ", ".join(missing)
|
||
|
||
|
||
def escape_xml(value: str) -> str:
|
||
"""给 skills summary 做最小 XML 转义。"""
|
||
|
||
return value.replace("&", "&").replace("<", "<").replace(">", ">")
|