oracle-jump-query-public/scripts/skill_update.py

"""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