Auto-Finance/scripts/install_skill.py

"""把报销技能同步到宿主的用户级技能目录，让它脱离本仓库也能用。

仓库里的 `.agents/skills/reimbursement/` 是**唯一真源**，本脚本只做单向同步（不反向编辑安装副本，
否则下次同步会被覆盖）。装完要重连宿主才会重新扫描技能目录。

用法：
  uv run python scripts/install_skill.py                    # 安装 / 更新
  uv run python scripts/install_skill.py --check             # 只比对；有漂移退出码 1
  uv run python scripts/install_skill.py --dest D:/x/skills   # 指定别的技能目录（会加 /reimbursement）
  uv run python scripts/install_skill.py --dest D:/x/skills   # 指定别的技能目录（会加 /reimbursement）
"""

from __future__ import annotations

import argparse
import hashlib
import os
import shutil
import sys
from pathlib import Path

SKILL_NAME = "reimbursement"
REPO_ROOT = Path(__file__).resolve().parents[1]
SOURCE = REPO_ROOT / ".agents" / "skills" / SKILL_NAME


def resolve_home() -> Path:
    """主目录：环境变量优先。精简过的进程环境里 Path.home() 会抛 RuntimeError。"""
    for key in ("USERPROFILE", "HOME", "HOMEDRIVE_HOME"):
        raw = os.environ.get(key, "").strip()
        if raw:
            if key == "HOMEDRIVE_HOME":
                raw = os.path.join(os.environ.get("HOMEDRIVE", "").strip(), os.environ.get("HOME", "").strip())
            if raw and Path(raw).is_absolute():
                return Path(raw)
    try:
        return Path.home()
    except RuntimeError:
        print("无法确定用户主目录：设 DSH_SKILLS_DIR，或用 --dest 指定技能目录", file=sys.stderr)
        raise SystemExit(4) from None


def norm(path_str: str) -> str:
    """日志里的路径是 JSON 转义过的双反斜杠，统一成小写正斜杠再比。"""
    return "".join(ch for ch in path_str if ch != chr(92)).replace("//", "/").lower().rstrip("/")


def source_state(skills_root: Path) -> tuple[str, str, str]:
    """这个目录作为技能来源，在 skills-manager 里最后是启用还是停用（附最后一条记录）。

    踩过的坑：装进了一个被停用的来源目录（~/.dsh/skills），文件明明在，宿主却永远扫不到，
    比没装更难发现。所以装之前先查最后一条形同实质的记录。
    """
    log = resolve_home() / ".dsh" / "dsh-skills-manager.log"
    if not log.is_file():
        return "unknown", "", ""
    want = norm(str(skills_root))
    state, detail, stamp = "unknown", "", ""
    try:
        lines = log.read_text(encoding="utf-8", errors="replace").splitlines()
    except OSError:
        return "unknown", "", ""
    for line in lines:
        if "source-enable" not in line and "source-disable" not in line:
            continue
        if want and want in norm(line):
            state = "disabled" if "source-disable" in line else "enabled"
            detail = line[-90:]
            pos = line.find('"ts":"')
            stamp = line[pos + 6 : pos + 25] if pos >= 0 else ""
    return state, stamp, detail


def candidate_roots() -> list[Path]:
    """候选的用户级技能根目录。

    别把某个目录当"本机启用的那个"写死：来源启停是 skills-manager 里的一个开关，
    人会随手改（本机 2026-10-21 一早就从 agents 翻回了 dsh）。所以这里只列候选，
    到底装哪儿由 source_state() 查日志决定。
    """
    base = os.environ.get("DSH_SKILLS_DIR", "").strip()
    if base:
        return [Path(base)]
    home = resolve_home()
    return [home / ".dsh" / "skills", home / ".agents" / "skills"]


def default_dest() -> Path:
    """选最近一次被启用的来源；一个都没有就拒绝安装，而不是装进扫不到的目录。"""
    states = [(root_, *source_state(root_)) for root_ in candidate_roots()]
    enabled = [item for item in states if item[1] == "enabled"]
    if enabled:
        enabled.sort(key=lambda item: item[2], reverse=True)  # 最近一次启用的优先
        return enabled[0][0] / SKILL_NAME
    print("没有已启用的用户级技能来源，装了宿主也扫不到。当前状态：", file=sys.stderr)
    for item in states:
        print("  " + item[1].ljust(9) + str(item[0]), file=sys.stderr)
    print("在宿主的技能管理里启用其中一个来源，或用 --dest 指定目录、--list-sources 看完整记录。", file=sys.stderr)
    raise SystemExit(5)


def collect(base: Path) -> list[Path]:
    """技能目录里要同步的文件：SKILL.md 与 references/ 下的 Markdown。"""
    files: list[Path] = []
    top = base / "SKILL.md"
    if top.is_file():
        files.append(top)
    ref = base / "references"
    if ref.is_dir():
        files.extend(sorted(p for p in ref.glob("*.md") if p.is_file()))
    return files


def digest(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def validate_source() -> str:
    """源目录健康检查：宁可现在报错，也别把一个没有 front matter 的技能装上去。"""
    if not SOURCE.is_dir():
        return f"源目录不存在：{SOURCE}"
    skill_md = SOURCE / "SKILL.md"
    if not skill_md.is_file():
        return f"缺少 SKILL.md：{skill_md}"
    text = skill_md.read_text(encoding="utf-8")
    if not text.startswith("---\n"):
        return "SKILL.md 缺少 YAML front matter（宿主靠它注册技能）"
    front = text.split("---", 2)[1] if text.count("---") >= 2 else ""
    if "name:" not in front or "description:" not in front:
        return "front matter 里必须有 name 与 description"
    return ""


def main() -> int:
    parser = argparse.ArgumentParser(description="同步报销技能到用户级技能目录")
    parser.add_argument("--dest", default="", help="目标技能目录（默认 ~/.dsh/skills/reimbursement）")
    parser.add_argument("--check", action="store_true", help="只比对不写入")
    parser.add_argument("--force", action="store_true", help="即使来源被停用也照装")
    parser.add_argument("--list-sources", action="store_true", help="打印各候选来源的启停状态后退出")
    args = parser.parse_args()

    if args.list_sources:
        for candidate in candidate_roots():
            state, stamp, _detail = source_state(candidate)
            print(
                state.ljust(9)
                + str(candidate)
                + ("   最后事件 " + stamp if stamp else "   （日志里没有这个来源的记录）")
            )
        return 0

    problem = validate_source()
    if problem:
        print(f"源技能有问题：{problem}", file=sys.stderr)
        return 2

    dest = Path(args.dest).expanduser() if args.dest else default_dest()
    if args.dest and dest.name != SKILL_NAME:
        dest = dest / SKILL_NAME

    state, _stamp, detail = source_state(dest.parent)
    if state == "disabled":
        print(
            f"警告：{dest.parent} 这个技能来源在 skills-manager 里是【停用】状态，装进去宿主也扫不到。"
            f"（最后一条记录：{detail}）改目标目录或用 --dest 指定已启用的来源。",
            file=sys.stderr,
        )
        if not args.force:
            return 3
    elif state == "enabled":
        print(f"技能来源已启用：{dest.parent}")
    else:
        print(f"（查不到 {dest.parent} 的来源启停记录，按已启用处理）")

    new: list[str] = []
    changed: list[str] = []
    stale: list[str] = []
    for src in collect(SOURCE):
        rel = src.relative_to(SOURCE)
        target = dest / rel
        if not target.is_file():
            new.append(str(rel))
        elif digest(target) != digest(src):
            changed.append(str(rel))
    src_names = {str(p.relative_to(SOURCE)) for p in collect(SOURCE)}
    if dest.is_dir():
        for existing in collect(dest):
            rel = str(existing.relative_to(dest))
            if rel not in src_names:
                stale.append(rel)

    same = len(collect(SOURCE)) - len(new) - len(changed)
    if not new and not changed and not stale:
        print(f"已是最新：{dest}（{same} 个文件一致）")
        return 0

    print(f"{'检查' if args.check else '同步'}：{SOURCE} -> {dest}")
    for n in new:
        print(f"  新增  {n}")
    for n in changed:
        print(f"  更新  {n}")
    for n in stale:
        print(f"  多余  {n}（安装目录里有、源里没有，需手工删）")

    if args.check:
        print("存在漂移：跑 uv run python scripts/install_skill.py 同步", file=sys.stderr)
        return 1

    for src in collect(SOURCE):
        rel = src.relative_to(SOURCE)
        target = dest / rel
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(src, target)
    print(f"已写入 {len(collect(SOURCE))} 个文件。重连宿主后生效。")
    if (REPO_ROOT / ".agents" / "skills" / SKILL_NAME).is_dir():
        print(f"提示：项目内同名技能仍在 .agents/skills/{SKILL_NAME}/，两边内容现在一致。")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())