Files
CraftKit/migration/scripts/scan_sources.py
T

112 lines
4.0 KiB
Python

#!/usr/bin/env python3
"""只读扫描来源目录中的 Skill,并输出稳定的 JSON 清单。"""
from __future__ import annotations
import argparse
import hashlib
import json
import subprocess
from datetime import datetime, timezone
from pathlib import Path
def file_hash(path: Path) -> str:
"""按二进制内容计算单文件 SHA-256。"""
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(65536), b""):
digest.update(block)
return digest.hexdigest()
def directory_hash(root: Path) -> tuple[str, int]:
"""把相对路径和文件内容共同纳入哈希,确保结构变化也可被识别。"""
digest = hashlib.sha256()
files = sorted(path for path in root.rglob("*") if path.is_file())
for path in files:
relative = path.relative_to(root).as_posix()
digest.update(relative.encode("utf-8"))
digest.update(b"\0")
digest.update(bytes.fromhex(file_hash(path)))
return digest.hexdigest(), len(files)
def read_skill_name(skill_md: Path) -> str:
"""从 YAML frontmatter 中读取 name;解析失败时回退为目录名。"""
for line in skill_md.read_text(encoding="utf-8-sig").splitlines():
if line.startswith("name:"):
return line.split(":", 1)[1].strip().strip("'\"") or skill_md.parent.name
return skill_md.parent.name
def git_commit(root: Path) -> str | None:
"""尽力读取来源提交;非 Git 目录时返回空值,不阻塞文件扫描。"""
result = subprocess.run(
["git", "-c", f"safe.directory={root.as_posix()}", "-C", str(root), "rev-parse", "HEAD"],
capture_output=True,
text=True,
check=False,
)
return result.stdout.strip() if result.returncode == 0 else None
def scan_source(source_id: str, root: Path) -> dict:
"""扫描单个来源,结果仅包含相对路径和指纹,不包含本地绝对路径。"""
if not root.is_dir():
raise ValueError(f"来源目录不存在:{root}")
skills = []
for skill_md in sorted(root.rglob("SKILL.md")):
skill_root = skill_md.parent
digest, file_count = directory_hash(skill_root)
children = sorted(
child.name for child in skill_root.iterdir() if child.is_dir()
)
skills.append(
{
"name": read_skill_name(skill_md),
"relativePath": skill_root.relative_to(root).as_posix(),
"sha256": digest,
"fileCount": file_count,
"resources": children,
}
)
return {
"id": source_id,
"commit": git_commit(root),
"skillCount": len(skills),
"skills": skills,
}
def parse_source(value: str) -> tuple[str, Path]:
"""解析 id=path 参数,避免把本地路径写入输出文件。"""
if "=" not in value:
raise argparse.ArgumentTypeError("来源参数必须使用 id=path 格式")
source_id, raw_path = value.split("=", 1)
if not source_id.strip() or not raw_path.strip():
raise argparse.ArgumentTypeError("来源标识和路径均不能为空")
return source_id.strip(), Path(raw_path).expanduser().resolve()
def main() -> None:
parser = argparse.ArgumentParser(description="只读扫描 Skill 来源目录")
parser.add_argument("--source", action="append", required=True, type=parse_source)
parser.add_argument("--output", type=Path, help="可选 JSON 输出路径;省略时输出到终端")
args = parser.parse_args()
payload = {
"schemaVersion": 1,
"generatedAt": datetime.now(timezone.utc).isoformat(),
"sources": [scan_source(source_id, root) for source_id, root in args.source],
}
content = json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
if args.output:
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(content, encoding="utf-8")
else:
print(content, end="")
if __name__ == "__main__":
main()