dsh_ai1net_server/归档/技能包快照/session-mechanism-20261004/scripts/hooks/_env.py

# -*- coding: utf-8 -*-
"""**环境定位与环境自检**（2026-10-02 · 用户定案「技能使用时必须先检查环境配置」）。

一句话：把「技能库根 / 配置目录」从**各脚本各自推导**改成**一处判据＋一处体检**。

## 为什么要有这个模块（2026-10-02 实测事故）

`reply-style-guard.py` 曾被注册成文档库 `07-scripts/reply-style-guard.py`（D 盘旧副本）。
它沿 `__file__` 上溯 4 级找 `skills/` ⇒ 从 D 盘够不到 ⇒ 落进最后一行
`os.path.expanduser('~/.workbuddy/skills')`。
⚠️🔴 **Windows 上 `~` 不等于真配置目录**（本机真值在 E 盘 `CODEBUDDY_CONFIG_DIR`）⇒ 那一行是**错的**。

后果比报错更坏：该钩子的设计判据是「哪一级都读不到就**静默零输出**、不影响任何人」
⇒ **「机制坏了」和「没配规则」表现完全一样**，日志里只有一行 `core=0 字符`。
（真实代价：另一个工作区为这件事绕了两轮，在「规则文件在不在」上打转。）

## 三条硬规矩（本模块的判据）

1. 🔴 **顺序固定**：`DSH_SKILLS_ROOT` env → `CODEBUDDY_CONFIG_DIR` env（+ `roots.env` 兜底）
   → 沿 `__file__` 上溯找含 `agent-operating-rules` 的那一层 → ⛔ **不再回落到 `~`**。
2. 🔴 **找不到就报错，不静默**：写 **stderr** ＋ 返回 `None`；调用方据此**非零退出**。
   ⛔ 绝不「静默零输出」—— 那正是把两类事故压成一个表象的元凶。
3. 🔴 **体检与环境标记分离**：`env_check()` 只**读**（体检）；`write_env_stamp()` 才**写**标记。
   ⇒ 体检永远不该有副作用（否则「查一下」就把标记刷新了，标记就永不报警）。

## 标记文件（用户 2026-10-02 要求：「在对应文件夹创建状态变量和时间作为标记」）

落在**配置作用域对应的文件夹**里，文件名 `env-stamp.json`：

- 全局作用域 ⇒ `<配置目录>/env-stamp.json`（本机 `E:/ProgramData/.workbuddy/env-stamp.json`）
- 工作区作用域 ⇒ `<工作区>/.workbuddy/env-stamp.json`

字段：`scope`（global/workspace）｜`checked_at`（ISO 到秒）｜`ok`（bool）｜
`config_dir` / `skills_root`（**判据读数**，便于事后核）｜`hooks`（逐条注册的存在性）｜
`fixed`（本轮修过的问题清单）
"""

from __future__ import annotations

import json
import os
import sys
import time
from pathlib import Path

_STAMP_NAME = "env-stamp.json"
_ROOTS_ENV = "roots.env"          # 技能根的权威落点（install.py 生成）
_SNIFF = "agent-operating-rules"  # 技能库的识别特征目录


def _norm(p) -> str:
    s = str(p or "").strip()
    return os.path.normpath(s).replace("\\", "/") if s else ""


def _load_roots_env() -> dict:
    """读技能包里的 `roots.env`（⛔ 各脚本的 `_sm_load_roots()` 注入 env 前的原始来源）。"""
    out = {}
    d = Path(__file__).resolve().parent
    for _ in range(5):
        f = d / _ROOTS_ENV
        if f.is_file():
            try:
                for ln in f.read_text(encoding="utf-8").splitlines():
                    ln = ln.strip()
                    if ln and not ln.startswith("#") and "=" in ln:
                        k, v = ln.split("=", 1)
                        out[k.strip()] = v.strip()
                return out
            except Exception:
                return {}
        n = d.parent
        if n == d:
            break
        d = n
    return out


def config_dir() -> str:
    """**真配置目录** —— 顺序见硬规矩①的前三档。"""
    v = os.environ.get("CODEBUDDY_CONFIG_DIR")
    if v:
        return _norm(v)
    v = _load_roots_env().get("CODEBUDDY_CONFIG_DIR")
    if v:
        return _norm(v)
    #⛔ 最后一档仍不给 `~`：Windows 上它通常不是真配置目录（实测会指到 C:\Users\...）
    return _norm(Path.home() / ".workbuddy")


def skills_root() -> str:
    """**技能库根** —— 顺序见硬规矩①。⛔ 不再回落到 `~`（硬规矩①末档已删）。

    🔴 2026-10-02 补：**每一档都要验"真的存在"**，⛔ 不信任 env 值本身。
       实测踩过：`DSH_SKILLS_ROOT` 指向一个不存在的目录时，旧写法直接返回它
       ⇒ 下游 `os.path.isfile(核心块)` 失败 ⇒ **静默零输出**（同款静默失效）。
       ⇒ env 给错 ⇒ 继续往下找；全都找不到 ⇒ 返回 `""`（＝报错，⛔ 不猜）。
    """
    v = os.environ.get("DSH_SKILLS_ROOT")
    if v and (Path(_norm(v)) / _SNIFF).is_dir():
        return _norm(v)
    d = Path(__file__).resolve().parent
    for _ in range(6):                       # 沿 __file__ 上溯（换机可用，⛔ 不写死盘符）
        cand = d.parent
        if (cand / _SNIFF).is_dir():
            return _norm(cand)
        if d.parent == d:
            break
        d = d.parent
    r = _load_roots_env().get("CODEBUDDY_CONFIG_DIR")
    if r:
        cand = Path(_norm(r)) / "skills"
        if (cand / _SNIFF).is_dir():
            return _norm(cand)
    return ""                                  # 🔴 空串＝找不到（⛔ 不是"猜一个"）


def _stamp_path(ws: str = "", scope: str = "global") -> str:
    if scope == "workspace" and ws:
        return _norm(Path(ws) / ".workbuddy" / _STAMP_NAME)
    return _norm(Path(config_dir()) / _STAMP_NAME)


def registered_hooks() -> list:
    """读全局 `settings.json` 里已注册的钩子 ⇒ `[(事件, 命令, timeout, 脚本路径)]`。"""
    sp = Path(config_dir()) / "settings.json"
    out = []
    try:
        d = json.loads(sp.read_text(encoding="utf-8"))
    except Exception:
        return out
    for ev, groups in (d.get("hooks") or {}).items():
        for g in (groups or []):
            for e in (g or {}).get("hooks", []):
                c = (e or {}).get("command", "") or ""
                # 取最后一个 .py 片段（命令形如`<python>"<脚本>" <参数>`）
                parts = [p for p in c.split('"') if p.strip().endswith(".py")]
                out.append((ev, c, (e or {}).get("timeout"),
                            _norm(parts[-1]) if parts else ""))
    return out


def env_check(ws: str = "", scope: str = "global") -> dict:
    """🔍 **只读体检**：返回逐项读数，⛔ **绝不写标记**（硬规矩③）。"""
    rep = {"scope": scope, "checked_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
           "config_dir": config_dir(), "skills_root": skills_root(), "bad": []}
    if not rep["skills_root"]:
        rep["bad"].append("skills_root 定位失败（⛔ 已不回落~）")
    elif not (Path(rep["skills_root"]) / _SNIFF).is_dir():
        rep["bad"].append("skills_root 存在但缺 %s：%s" % (_SNIFF, rep["skills_root"]))
    hooks = []
    for ev, cmd, t, path in registered_hooks():
        ok = bool(path) and os.path.isfile(path)
        row = {"event": ev, "timeout": t, "path": path, "exists": ok}
        hooks.append(row)
        if not ok:
            rep["bad"].append("钩子目标不存在：%s %s" % (ev, path or cmd[:60]))
    rep["hooks"] = hooks
    rep["ok"] = not rep["bad"]
    return rep


def write_env_stamp(ws: str = "", scope: str = "global", fixed=None) -> str:
    """✍️ **写环境标记**（含状态变量＋时间）。`scope` 决定落在哪个文件夹。"""
    rep = env_check(ws, scope)
    rep["fixed"] = list(fixed or [])
    p = Path(_stamp_path(ws, scope))
    try:
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(json.dumps(rep, ensure_ascii=False, indent=2) + "\n",
                     encoding="utf-8", newline="\n")
    except Exception as e:
        sys.stderr.write("env-stamp 写失败：%r\n" % (e,))
        return ""
    return _norm(p)


def read_env_stamp(ws: str = "", scope: str = "global") -> dict:
    try:
        return json.loads(Path(_stamp_path(ws, scope)).read_text(encoding="utf-8"))
    except Exception:
        return {}


def require_env(ws: str = "", what: str = "") -> str:
    """🔴 **调用方统一入口**：定位不到就**报错并返回空串**（硬规矩②）。

    ⛔ 绝不在这里 `sys.exit` —— 钩子进程退出码由调用方决定（各钩子预算不同）。
    """
    root = skills_root()
    if root:
        return root
    sys.stderr.write(
        "[env] 技能库根定位失败%s：DSH_SKILLS_ROOT / CODEBUDDY_CONFIG_DIR 均未给，"
        "沿 __file__ 上溯也没找到含 %s 的目录。\n"
    "     ⛔ 已⛔静默跳过（不再回落 ~，因Windows 上 ~ 不是真配置目录）\n"
    "     ✅ 修法：设环境变量 DSH_SKILLS_ROOT，或跑 "
    "`python scripts/hooks/_env.py --stamp` 重建标记并体检\n"
        % (("（%s）" % what) if what else "", _SNIFF))
    return ""


if __name__ == "__main__":
    import argparse
    _ap = argparse.ArgumentParser(description="环境自检／环境标记")
    _ap.add_argument("--ws", default="", help="工作区绝对路径")
    _ap.add_argument("--scope", default="global", choices=["global", "workspace"])
    _ap.add_argument("--stamp", action="store_true", help="写环境标记")
    _ap.add_argument("--fixed", default="", help="--stamp 时记进 fixed 的说明")
    _a = _ap.parse_args()
    if _a.stamp:
        _p = write_env_stamp(_a.ws, _a.scope,
                             [_a.fixed] if _a.fixed else None)
        print("已写标记：%s" % (_p or "⛔ 写失败"))
        raise SystemExit(0 if _p else 1)
    _r = env_check(_a.ws, _a.scope)
    print("配置目录 = %s" % _r["config_dir"])
    print("技能库根 = %s" % (_r["skills_root"] or "⛔ 定位失败"))
    print("已注册钩子 = %d 条｜不存在的 = %d"
          % (len(_r["hooks"]), sum(1 for h in _r["hooks"] if not h["exists"])))
    for _b in _r["bad"]:
        print("  ⛔ %s" % _b)
    print("结论 = %s" % ("✅ 通过" if _r["ok"] else "❌ 不通过"))
    raise SystemExit(0 if _r["ok"] else 1)