course-factory/90-工具/register_skills.py

# -*- coding: utf-8 -*-
"""
register_skills.py — 把 30-技能/ 下的技能注册给 WorkBuddy

优先用目录联接（junction），改源即生效；失败则回退为复制 + 同步模式。

用法：
  python register_skills.py            # 注册（link 优先）
  python register_skills.py --copy     # 强制复制模式
  python register_skills.py --sync     # 复制模式下，把源文件同步到目标
  python register_skills.py --status   # 查看当前注册状态
  python register_skills.py --uninstall
"""
import sys, os, shutil, subprocess, argparse
from pathlib import Path

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")

ROOT = Path(__file__).resolve().parent.parent
SRC = ROOT / "30-技能"
DEST = Path(os.environ.get("USERPROFILE", Path.home())) / ".workbuddy" / "skills"
SKILLS = ["course-factory", "kb-sync", "kb-ingest", "kb-query",
          "kb-lint", "kb-index", "course-outline", "course-slides"]


def is_link(p: Path) -> bool:
    if p.is_symlink():
        return True
    try:
        return bool(p.exists() and os.readlink(str(p)))
    except (OSError, ValueError):
        pass
    # Windows junction 无 readlink，用 FILE_ATTRIBUTE_REPARSE_POINT 判断
    try:
        return bool(os.stat(str(p)).st_file_attributes & 0x400)
    except (AttributeError, OSError):
        return False


def make_link(src: Path, dst: Path) -> bool:
    if os.name == "nt":
        r = subprocess.run(["cmd", "/c", "mklink", "/J", str(dst), str(src)],
                           capture_output=True, text=True,
                           encoding="gbk", errors="replace")
        return r.returncode == 0
    try:
        dst.symlink_to(src, target_is_directory=True)
        return True
    except OSError:
        return False


def register(mode: str):
    DEST.mkdir(parents=True, exist_ok=True)
    ok_link, ok_copy = [], []
    for name in SKILLS:
        s = SRC / name
        d = DEST / name
        if not (s / "SKILL.md").exists():
            print(f"  [SKIP] {name}：源目录缺少 SKILL.md")
            continue
        if d.exists() or is_link(d):
            if is_link(d):
                print(f"  [OK]   {name}（已是联接，跳过）")
                ok_link.append(name)
                continue
            shutil.rmtree(d, ignore_errors=True)
        if mode == "link" and make_link(s, d):
            print(f"  [LINK] {name}")
            ok_link.append(name)
        else:
            shutil.copytree(s, d)
            print(f"  [COPY] {name}")
            ok_copy.append(name)
    print(f"\n注册完成：联接 {len(ok_link)} 个，复制 {len(ok_copy)} 个 -> {DEST}")
    if ok_copy:
        print("提示：复制模式改了源文件后需运行 python register_skills.py --sync")


def sync():
    n = 0
    for name in SKILLS:
        s, d = SRC / name, DEST / name
        if is_link(d) or not d.exists():
            continue
        for f in s.rglob("*"):
            if f.is_file():
                t = d / f.relative_to(s)
                t.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy2(f, t)
                n += 1
    print(f"已同步 {n} 个文件到 {DEST}")


def status():
    print(f"技能源目录：{SRC}")
    print(f"注册目标：  {DEST}\n")
    for name in SKILLS:
        d = DEST / name
        if not d.exists():
            st = "未注册"
        elif is_link(d):
            st = "联接"
        else:
            st = "复制（需 sync）"
        print(f"  {name:<18} {st}")


def uninstall():
    for name in SKILLS:
        d = DEST / name
        if is_link(d):
            subprocess.run(["cmd", "/c", "rmdir", str(d)], capture_output=True,
                           text=True, encoding="gbk", errors="replace")
            print(f"  [RM] {name}")
        elif d.exists():
            shutil.rmtree(d, ignore_errors=True)
            print(f"  [RM] {name}")


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--copy", action="store_true")
    ap.add_argument("--sync", action="store_true")
    ap.add_argument("--status", action="store_true")
    ap.add_argument("--uninstall", action="store_true")
    a = ap.parse_args()
    if a.status:
        status()
    elif a.sync:
        sync()
    elif a.uninstall:
        uninstall()
    else:
        register("copy" if a.copy else "link")