133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""从 Muse manifest 生成 Pi/DSH 可发现的扁平 Skill 投影。
|
|
|
|
投影写到宿主原生目录 `.pi/skills/` 与 `.dsh/skills/`，不进 git。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
from pathlib import Path
|
|
from urllib.parse import unquote
|
|
|
|
TARGETS = ("dsh", "pi")
|
|
_MARKDOWN_LINK = re.compile(r"\(([^)\s]+)")
|
|
|
|
|
|
def _discover_root() -> Path:
|
|
return next(
|
|
parent
|
|
for parent in (Path(__file__).resolve().parent, *Path(__file__).resolve().parents)
|
|
if (parent / "AGENTS.md").is_file() and (parent / ".git").exists()
|
|
)
|
|
|
|
|
|
def host_skill_root(root: Path, target: str) -> Path:
|
|
return root / f".{target}" / "skills"
|
|
|
|
|
|
def _rewrite_projected_reference_links(
|
|
text: str,
|
|
source_file: Path,
|
|
projected_file: Path,
|
|
source_to_projected: dict[Path, Path],
|
|
) -> str:
|
|
"""把跨 Skill 的源相对链接改到扁平投影，保留源合同正文不变。"""
|
|
|
|
def replace(match: re.Match[str]) -> str:
|
|
raw_target = match.group(1)
|
|
target, separator, fragment = raw_target.partition("#")
|
|
if not target or target.startswith(("#", "http://", "https://", "mailto:", "/")):
|
|
return match.group(0)
|
|
resolved_source = (source_file.parent / unquote(target)).resolve()
|
|
resolved_projected = source_to_projected.get(resolved_source)
|
|
if resolved_projected is None:
|
|
return match.group(0)
|
|
relative = Path(
|
|
os.path.relpath(resolved_projected, projected_file.parent)
|
|
).as_posix()
|
|
return f"({relative}{separator}{fragment if separator else ''}"
|
|
|
|
return _MARKDOWN_LINK.sub(replace, text)
|
|
|
|
|
|
def generate(root: Path | None = None) -> dict[str, object]:
|
|
root = Path(root).resolve() if root is not None else _discover_root()
|
|
manifest_path = (
|
|
root / "muse" / "lifecycle" / "quality" / "harness" / "manifests" / "skills.json"
|
|
)
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
entries = [
|
|
entry for entry in manifest["skills"] if entry.get("invocation") == "model_routed"
|
|
]
|
|
names = {entry["name"] for entry in entries}
|
|
source_to_projected: dict[str, dict[Path, Path]] = {target: {} for target in TARGETS}
|
|
for entry in entries:
|
|
references = (root / entry["skill_path"]).parent / "references"
|
|
if not references.is_dir():
|
|
continue
|
|
for source_file in references.rglob("*"):
|
|
if source_file.is_file():
|
|
for target in TARGETS:
|
|
projected_file = (
|
|
host_skill_root(root, target)
|
|
/ entry["name"]
|
|
/ "references"
|
|
/ source_file.relative_to(references)
|
|
)
|
|
source_to_projected[target][source_file.resolve()] = projected_file
|
|
for entry in sorted(entries, key=lambda item: item["name"]):
|
|
name = entry["name"]
|
|
source = root / entry["skill_path"]
|
|
if not source.is_file():
|
|
raise FileNotFoundError(f"Skill 源不存在: {entry['skill_path']}")
|
|
text = source.read_text(encoding="utf-8")
|
|
for target in TARGETS:
|
|
destination_dir = host_skill_root(root, target) / name
|
|
if destination_dir.exists():
|
|
shutil.rmtree(destination_dir)
|
|
destination_dir.mkdir(parents=True, exist_ok=True)
|
|
(destination_dir / "SKILL.md").write_text(text, encoding="utf-8")
|
|
references = source.parent / "references"
|
|
if references.is_dir():
|
|
shutil.copytree(references, destination_dir / "references")
|
|
|
|
for target in TARGETS:
|
|
skill_root = host_skill_root(root, target)
|
|
if not skill_root.is_dir():
|
|
continue
|
|
for skill_file in skill_root.glob("*/SKILL.md"):
|
|
if skill_file.parent.name not in names:
|
|
skill_file.unlink()
|
|
try:
|
|
skill_file.parent.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
for target in TARGETS:
|
|
for source_file, projected_file in source_to_projected[target].items():
|
|
if projected_file.suffix.lower() != ".md":
|
|
continue
|
|
text = source_file.read_text(encoding="utf-8")
|
|
rewritten = _rewrite_projected_reference_links(
|
|
text, source_file, projected_file, source_to_projected[target]
|
|
)
|
|
projected_file.parent.mkdir(parents=True, exist_ok=True)
|
|
projected_file.write_text(rewritten, encoding="utf-8")
|
|
|
|
return {
|
|
"count": len(entries),
|
|
"targets": {
|
|
target: host_skill_root(root, target).relative_to(root).as_posix()
|
|
for target in TARGETS
|
|
},
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
result = generate()
|
|
print(json.dumps(result, ensure_ascii=False))
|