增加Skill每日自动更新和使用记录

scripts/skill_update.py

@@ -0,0 +1,162 @@
+"""Daily Git update checks and local usage timestamps, independent of login config."""
+
+import contextlib
+import json
+import os
+import subprocess
+import sys
+import tempfile
+from datetime import datetime
+from pathlib import Path
+
+
+GIT_TIMEOUT = 10
+
+
+@contextlib.contextmanager
+def _state_lock(path):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a+b") as handle:
+        if os.name == "nt":
+            import msvcrt
+            handle.seek(0, os.SEEK_END)
+            if not handle.tell():
+                handle.write(b"0")
+                handle.flush()
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
+            try:
+                yield
+            finally:
+                handle.seek(0)
+                msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
+        else:
+            import fcntl
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+            try:
+                yield
+            finally:
+                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+
+
+def _read_state(path):
+    try:
+        with path.open(encoding="utf-8") as handle:
+            state = json.load(handle)
+        return state if isinstance(state, dict) else {}
+    except (OSError, ValueError):
+        return {}
+
+
+def _write_state(path, state):
+    fd, temporary = tempfile.mkstemp(prefix=path.name + ".", dir=str(path.parent))
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(state, handle, ensure_ascii=False, indent=2)
+            handle.write("\n")
+        os.replace(temporary, str(path))
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def _git(root, *args):
+    env = os.environ.copy()
+    env["GIT_TERMINAL_PROMPT"] = "0"
+    env["GCM_INTERACTIVE"] = "never"
+    result = subprocess.run(
+        ["git", "-C", str(root), *args], stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE, timeout=GIT_TIMEOUT, env=env,
+    )
+    if result.returncode:
+        # Git output can contain authenticated remote URLs; never store or print it.
+        raise RuntimeError("git command failed")
+    return result.stdout.decode("utf-8", errors="replace").strip()
+
+
+def _check_remote(root):
+    # A downloaded directory may sit inside an unrelated parent repository.
+    if not (root / ".git").exists():
+        return {"status": "not_git"}
+    _git(root, "fetch", "--quiet", "origin", "main")
+    local = _git(root, "rev-parse", "HEAD")
+    remote = _git(root, "rev-parse", "FETCH_HEAD")
+    outcome = {
+        "status": "up_to_date" if local == remote else "update_available",
+        "local_commit": local,
+        "remote_commit": remote,
+    }
+    if local == remote:
+        return outcome
+    if _git(root, "rev-parse", "--abbrev-ref", "HEAD") != "main":
+        outcome["status"] = "update_blocked"
+        return outcome
+    changed = _git(root, "diff", "--name-only").splitlines()
+    if any(name not in ("config.json", "scripts/config.json") for name in changed) or _git(root, "diff", "--cached", "--name-only"):
+        outcome["status"] = "update_blocked"
+        return outcome
+    try:
+        _git(root, "merge-base", "--is-ancestor", local, remote)
+    except RuntimeError:
+        outcome["status"] = "update_blocked"
+        return outcome
+    # Keep configuration bytes local, including clean tracked template config.
+    saved = {}
+    for name in ("config.json", "scripts/config.json"):
+        path = root / name
+        if path.exists():
+            saved[path] = path.read_bytes()
+    try:
+        _git(root, "merge", "--ff-only", "--quiet", remote)
+    except RuntimeError:
+        outcome["status"] = "update_blocked"
+        return outcome
+    finally:
+        for path, data in saved.items():
+            if not path.exists() or path.read_bytes() != data:
+                path.parent.mkdir(parents=True, exist_ok=True)
+                path.write_bytes(data)
+    outcome["status"] = "updated"
+    outcome["local_commit"] = remote
+    return outcome
+
+
+def on_use(skill_dir, now=None):
+    """Record every invocation; claim today's check before the network call."""
+    root = Path(skill_dir).resolve()
+    state_path = root / "scripts" / "update-state.json"
+    lock_path = root / "scripts" / "update-state.lock"
+    now = now or datetime.now().astimezone()
+    stamp, day = now.isoformat(timespec="seconds"), now.date().isoformat()
+    try:
+        with _state_lock(lock_path):
+            state = _read_state(state_path)
+            state["last_used_at"] = stamp
+            should_check = state.get("last_check_date") != day
+            if should_check:
+                state.update(last_check_date=day, last_checked_at=stamp, status="checking")
+                state.pop("local_commit", None)
+                state.pop("remote_commit", None)
+            _write_state(state_path, state)
+        if not should_check:
+            return False
+        try:
+            outcome = _check_remote(root)
+        except (OSError, subprocess.SubprocessError, RuntimeError):
+            outcome = {"status": "check_failed"}
+        with _state_lock(lock_path):
+            state = _read_state(state_path)
+            # Preserve usage from concurrent invocations, and a newer day's claim.
+            if state.get("last_check_date") == day:
+                state.update(outcome)
+                _write_state(state_path, state)
+        if outcome["status"] == "updated":
+            print("[Skill] 已快进安装 Git 新版本，重新启动命令。", file=sys.stderr)
+        elif outcome["status"] == "update_blocked":
+            print("[Skill] 检测到远端版本变化，本机修改或分支状态阻止快进更新；继续执行原命令。", file=sys.stderr)
+        elif outcome["status"] == "check_failed":
+            print("[Skill] 今日更新检测失败，继续执行原命令；下次使用日自动重试。", file=sys.stderr)
+        return outcome["status"] == "updated"
+    except OSError:
+        print("[Skill] 无法保存本机使用记录，继续执行原命令。", file=sys.stderr)
+        return False