448 lines
19 KiB
Python
448 lines
19 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
"""
|
|||
|
|
session-mechanism · 一键配置（换机器只需跑这一个）
|
|||
|
|
|
|||
|
|
用法
|
|||
|
|
python install.py --dry-run # 只打印将改什么（settings.json 的 diff），⛔ 不写盘
|
|||
|
|
python install.py --apply # 真装：写 roots.env → 按声明表接线全局钩子 → 初始化工作区
|
|||
|
|
python install.py --verify # 装完自检：每个钩子空载荷 rc=0 ＋ collabd --where ＋ selftest
|
|||
|
|
python install.py --uninstall # 还原 settings.json（与装前备份逐字节相同）
|
|||
|
|
|
|||
|
|
设计要点（都由实测倒逼，⛔ 不要"优化"掉）
|
|||
|
|
1. **自解析**：解释器一律 `sys.executable`；配置目录按 `CODEBUDDY_CONFIG_DIR` 推导。
|
|||
|
|
⛔ 不硬编码 python 路径、⛔ 不硬编码盘符 —— 这正是「换机器必碎」的根因。
|
|||
|
|
2. **声明表驱动**：14 处接线收敛成一张 HOOKS 表；⛔ 不含 `decision_bridge.py`
|
|||
|
|
（它属 `ai1net-decision-laya` 另一条线，只恰好出现在同一张表里）。
|
|||
|
|
3. **幂等**：先删「本包自己的旧条目」再插；装两遍结果相同。
|
|||
|
|
4. **可逆**：首次安装先留**原状**备份 `settings.json.bak-session-mechanism-orig`（⛔ 已存在不覆盖），
|
|||
|
|
其后每次 `--apply` 另写**带微秒**的时间戳备份（防同秒同名互相覆盖）；
|
|||
|
|
`--uninstall` **优先**用 `-orig`，还原后做**逐字节**比对，不符即报错。
|
|||
|
|
5. **外置根目录**：写 `<包根>/roots.env` —— 因为 `settings.json` 的 hook 条目**没有 env 字段**，
|
|||
|
|
包内脚本搬一次就会静默指错（历史事故：台账写到别处、测试却全绿）。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import argparse
|
|||
|
|
import difflib
|
|||
|
|
import hashlib
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import shutil
|
|||
|
|
import subprocess
|
|||
|
|
import sys
|
|||
|
|
import tempfile
|
|||
|
|
import time
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
PKG = Path(__file__).resolve().parent
|
|||
|
|
ROOTS_ENV = PKG / "roots.env"
|
|||
|
|
LOG = PKG / "install.log"
|
|||
|
|
# 🔴 原状备份的**固定名**（首次安装、且当前 settings.json 无本包痕迹时才写；⛔ 已存在不覆盖）。
|
|||
|
|
# 有了它，`--uninstall` 不需要"从一堆时间戳备份里猜哪份是原状"。
|
|||
|
|
ORIG_BAK_NAME = "settings.json.bak-session-mechanism-orig"
|
|||
|
|
BAK_PREFIX = "settings.json.bak-session-mechanism-"
|
|||
|
|
# 🔴 判定"是不是本包的"一律**按脚本名**，⛔ 不用路径片段（`session-mechanism`）——
|
|||
|
|
# 路径片段在"包被改名 / 拷到别的目录"时**认不出自己**（实测教训：§2 #4）。
|
|||
|
|
# ⚠️ 这些脚本名**必须连旧落点一起认**（文档库 `07-scripts/` ＋ 工作区 `.workbuddy/tools/`），
|
|||
|
|
# 否则旧接线删不掉 ⇒ 同一件事挂两条钩子、同一秒各跑一次（`agent-operating-rules §10.2` 实测故障）。
|
|||
|
|
# ⛔ `decision_bridge.py` 不在此列。
|
|||
|
|
OWN_BASENAMES = (
|
|||
|
|
"session-log-guard.py",
|
|||
|
|
"lock-guard-hook.py",
|
|||
|
|
"bash-output-guard.py",
|
|||
|
|
"stop-dialog-guard.py",
|
|||
|
|
"skill-load-guard.py",
|
|||
|
|
"wb-result-hook.py",
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# ── 声明表：本包负责的钩子（⛔ decision_bridge 不在此表内）───────────────────
|
|||
|
|
# (事件, matcher, 包内脚本相对路径, 额外参数, 超时秒)
|
|||
|
|
HOOKS: list[tuple[str, str | None, str, list[str], int]] = [
|
|||
|
|
("PostToolUse", None, "scripts/hooks/session-log-guard.py", ["-S"], 10),
|
|||
|
|
("PreToolUse", "^Bash$", "scripts/hooks/wb-result-hook.py", [], 30),
|
|||
|
|
("PreToolUse", "Write|Edit", "scripts/hooks/lock-guard-hook.py", [], 10),
|
|||
|
|
("PreToolUse", "Bash|Read", "scripts/hooks/bash-output-guard.py", [], 10),
|
|||
|
|
("SessionEnd", None, "scripts/hooks/wb-result-hook.py", [], 10),
|
|||
|
|
("SessionStart", "startup|resume", "scripts/hooks/lock-guard-hook.py", [], 10),
|
|||
|
|
("UserPromptSubmit", None, "scripts/hooks/wb-result-hook.py", [], 20),
|
|||
|
|
("UserPromptSubmit", None, "scripts/hooks/stop-dialog-guard.py", [], 10),
|
|||
|
|
("UserPromptSubmit", None, "scripts/hooks/skill-load-guard.py", [], 10),
|
|||
|
|
("UserPromptSubmit", None, "scripts/hooks/session-log-guard.py", ["-S"], 10),
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def config_dir() -> Path:
|
|||
|
|
v = os.environ.get("CODEBUDDY_CONFIG_DIR")
|
|||
|
|
if v:
|
|||
|
|
return Path(v)
|
|||
|
|
return Path.home() / ".workbuddy"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def settings_path() -> Path:
|
|||
|
|
return config_dir() / "settings.json"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def log(msg: str) -> None:
|
|||
|
|
line = f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}"
|
|||
|
|
print(line)
|
|||
|
|
try:
|
|||
|
|
with open(LOG, "a", encoding="utf-8") as f:
|
|||
|
|
f.write(line + "\n")
|
|||
|
|
except Exception:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
def read_roots() -> dict[str, str]:
|
|||
|
|
d: dict[str, str] = {}
|
|||
|
|
if ROOTS_ENV.is_file():
|
|||
|
|
for ln in ROOTS_ENV.read_text(encoding="utf-8").splitlines():
|
|||
|
|
ln = ln.strip()
|
|||
|
|
if ln and not ln.startswith("#") and "=" in ln:
|
|||
|
|
k, v = ln.split("=", 1)
|
|||
|
|
d[k.strip()] = v.strip()
|
|||
|
|
return d
|
|||
|
|
|
|||
|
|
|
|||
|
|
def detect_workspace(explicit: str | None) -> Path | None:
|
|||
|
|
if explicit:
|
|||
|
|
return Path(explicit).resolve()
|
|||
|
|
for k in ("DSH_WS_ROOT", "COLLABD_WORKSPACE"):
|
|||
|
|
if os.environ.get(k):
|
|||
|
|
return Path(os.environ[k]).resolve()
|
|||
|
|
prev = read_roots().get("DSH_WS_ROOT")
|
|||
|
|
if prev and Path(prev).is_dir():
|
|||
|
|
return Path(prev)
|
|||
|
|
cwd = Path.cwd().resolve()
|
|||
|
|
for cand in (cwd, *cwd.parents):
|
|||
|
|
if (cand / ".workbuddy").is_dir() and (cand / "state.py").is_file():
|
|||
|
|
return cand
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def detect_docs_root(explicit: str | None) -> Path | None:
|
|||
|
|
if explicit:
|
|||
|
|
return Path(explicit).resolve()
|
|||
|
|
if os.environ.get("DSH_DOCS_ROOT"):
|
|||
|
|
return Path(os.environ["DSH_DOCS_ROOT"]).resolve()
|
|||
|
|
prev = read_roots().get("DSH_DOCS_ROOT")
|
|||
|
|
if prev and Path(prev).is_dir():
|
|||
|
|
return Path(prev)
|
|||
|
|
# 启发式：文档库必须同时具备「05-交接单」与「07-scripts」两个标志目录
|
|||
|
|
# ⚠️ 只看同名会误选到工作区里的 `dsh-server-docs/` 副本 ⇒ 两个条件都要满足。
|
|||
|
|
seeds = [Path.cwd(), *Path.cwd().parents]
|
|||
|
|
for k in ("DSH_CODE_REPO",):
|
|||
|
|
if os.environ.get(k):
|
|||
|
|
seeds.insert(0, Path(os.environ[k]))
|
|||
|
|
for s in seeds[:6]:
|
|||
|
|
for cand in (s / "dsh-server-docs", s.parent / "dsh-server-docs"):
|
|||
|
|
if (cand / "05-交接单").is_dir() and (cand / "07-scripts").is_dir():
|
|||
|
|
return cand.resolve()
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def py() -> str:
|
|||
|
|
return sys.executable
|
|||
|
|
|
|||
|
|
|
|||
|
|
def build_command(rel: str, extra: list[str]) -> str:
|
|||
|
|
script = (PKG / rel).resolve()
|
|||
|
|
parts = [f'"{py()}"', *[f'"{a}"' for a in extra], f'"{script}"']
|
|||
|
|
return " ".join(parts)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def has_our_trace(text: str) -> bool:
|
|||
|
|
"""**这条 hook 归不归本包管** —— 按**脚本名**判。
|
|||
|
|
⚠️ 必须连旧落点一起认（文档库 `07-scripts/` ＋ 工作区 `.workbuddy/tools/`），否则旧接线删不掉。
|
|||
|
|
⚠️ ⛔ 不可拿它当"这份文件是否已装过本包"的判据 —— 真机 settings.json **本来就有**同名脚本
|
|||
|
|
挂在 07-scripts 上（实测：6 个脚本名全 True）⇒ 会**永远写不出 `-orig`**、`--uninstall` **永远拒做**。
|
|||
|
|
那个判据用 `points_into_pkg()`。"""
|
|||
|
|
return any(b in text for b in OWN_BASENAMES)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def points_into_pkg(text: str) -> bool:
|
|||
|
|
"""**这份文件是否已被本包接管** —— 有没有钩子**指向本包目录**。
|
|||
|
|
🔴 路径**运行时从 `__file__` 推导**（`PKG`）⇒ 包被改名 / 拷到别的目录都认得出自己。
|
|||
|
|
这正是 §3.1b 要修的病：⛔ 旧写法硬编码路径片段 `session-mechanism`。
|
|||
|
|
旧落点（07-scripts / tools）**不算**"已装本包"。"""
|
|||
|
|
return PKG.as_posix().lower() in text.replace("\\", "/").lower()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def is_ours(entry: dict) -> bool:
|
|||
|
|
for h in entry.get("hooks", []):
|
|||
|
|
if has_our_trace(str(h.get("command", ""))):
|
|||
|
|
return True
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
def unique_backup_path(sp: Path) -> Path:
|
|||
|
|
"""备份落点：**带微秒**的时间戳 ⇒ 连装两遍不会同秒同名互相覆盖（§6 第 4 条实测踩到）。
|
|||
|
|
极端情况下同名仍存在 ⇒ 追加序号，⛔ 绝不覆盖既有备份。"""
|
|||
|
|
ts = time.strftime("%Y%m%d-%H%M%S") + f"-{time.time_ns() // 1000 % 1_000_000:06d}"
|
|||
|
|
cand = sp.with_name(f"{BAK_PREFIX}{ts}")
|
|||
|
|
n = 1
|
|||
|
|
while cand.exists():
|
|||
|
|
cand = sp.with_name(f"{BAK_PREFIX}{ts}-{n}")
|
|||
|
|
n += 1
|
|||
|
|
return cand
|
|||
|
|
|
|||
|
|
|
|||
|
|
def make_backup(sp: Path, before_text: str) -> tuple[Path, Path | None]:
|
|||
|
|
"""写备份。返回 (本次时间戳备份, 本次新写的原状备份或 None)。
|
|||
|
|
· 当前 settings.json **尚未被本包接管**（无钩子指向本包）⇒ 另写固定名 `-orig`，已存在则⛔不覆盖。
|
|||
|
|
· 每次 `--apply` 都另写一份带微秒的时间戳备份（逐次可回滚）。
|
|||
|
|
"""
|
|||
|
|
orig = sp.with_name(ORIG_BAK_NAME)
|
|||
|
|
wrote_orig: Path | None = None
|
|||
|
|
if not points_into_pkg(before_text):
|
|||
|
|
if orig.exists():
|
|||
|
|
log(f"⚠ 原状备份已存在，⛔ 不覆盖（若怀疑原件已换，请人工核对）：{orig}")
|
|||
|
|
else:
|
|||
|
|
shutil.copy2(sp, orig)
|
|||
|
|
wrote_orig = orig
|
|||
|
|
log(f"已写**原状**备份（本包接管前）→ {orig}")
|
|||
|
|
else:
|
|||
|
|
log(f"ℹ 当前 settings.json 已被本包接管 ⇒ 跳过 `-orig`（原状备份只在首次安装时留）")
|
|||
|
|
bak = unique_backup_path(sp)
|
|||
|
|
shutil.copy2(sp, bak)
|
|||
|
|
log(f"已备份 → {bak}")
|
|||
|
|
return bak, wrote_orig
|
|||
|
|
|
|||
|
|
|
|||
|
|
def desired_hooks() -> dict[str, list[dict]]:
|
|||
|
|
out: dict[str, list[dict]] = {}
|
|||
|
|
for event, matcher, rel, extra, timeout in HOOKS:
|
|||
|
|
item: dict = {"hooks": [{"type": "command", "command": build_command(rel, extra), "timeout": timeout}]}
|
|||
|
|
if matcher is not None:
|
|||
|
|
item["matcher"] = matcher
|
|||
|
|
out.setdefault(event, []).append(item)
|
|||
|
|
return out
|
|||
|
|
|
|||
|
|
|
|||
|
|
def merge(settings: dict) -> dict:
|
|||
|
|
"""先删本包旧条目 → 再插本包新条目；⛔ 其他条目（含 decision_bridge）原样保留。"""
|
|||
|
|
new = json.loads(json.dumps(settings))
|
|||
|
|
hooks = new.setdefault("hooks", {})
|
|||
|
|
want = desired_hooks()
|
|||
|
|
for event in set(list(hooks.keys()) + list(want.keys())):
|
|||
|
|
keep = [e for e in hooks.get(event, []) if not is_ours(e)]
|
|||
|
|
keep.extend(want.get(event, []))
|
|||
|
|
if keep:
|
|||
|
|
hooks[event] = keep
|
|||
|
|
elif event in hooks:
|
|||
|
|
del hooks[event]
|
|||
|
|
return new
|
|||
|
|
|
|||
|
|
|
|||
|
|
def write_roots(ws: Path, docs: Path | None, code: str | None, dry: bool) -> None:
|
|||
|
|
lines = [
|
|||
|
|
"# session-mechanism · 根目录（由 install.py 生成，⛔ 请勿手改）",
|
|||
|
|
"# 说明：settings.json 的 hook 条目没有 env 字段 ⇒ 包内脚本靠本文件定位根目录。",
|
|||
|
|
f"DSH_WS_ROOT={ws.as_posix()}",
|
|||
|
|
f"WB_RESULT_HOOK_WS={ws.as_posix()}",
|
|||
|
|
]
|
|||
|
|
if docs:
|
|||
|
|
lines.append(f"DSH_DOCS_ROOT={docs.as_posix()}")
|
|||
|
|
if code:
|
|||
|
|
lines.append(f"DSH_CODE_REPO={code.as_posix()}")
|
|||
|
|
text = "\n".join(lines) + "\n"
|
|||
|
|
if dry:
|
|||
|
|
log(f"[dry-run] 将写 {ROOTS_ENV}:\n{text}")
|
|||
|
|
return
|
|||
|
|
ROOTS_ENV.write_text(text, encoding="utf-8")
|
|||
|
|
log(f"已写 {ROOTS_ENV}")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def init_workspace(ws: Path, dry: bool) -> None:
|
|||
|
|
"""由 example 生成 collabd.config.json（已存在则不覆盖 ⇒ 保护现场活配置）。"""
|
|||
|
|
src = PKG / "scripts" / "collab" / "collabd.config.example.json"
|
|||
|
|
dst = ws / ".workbuddy" / "collab" / "collabd.config.json"
|
|||
|
|
if dst.is_file():
|
|||
|
|
log(f"工作区配置已存在，⛔ 不覆盖：{dst}")
|
|||
|
|
return
|
|||
|
|
if not src.is_file():
|
|||
|
|
log(f"⚠ 未找到模板 {src} ⇒ 跳过工作区初始化")
|
|||
|
|
return
|
|||
|
|
try:
|
|||
|
|
cfg = json.loads(src.read_text(encoding="utf-8"))
|
|||
|
|
except Exception as e:
|
|||
|
|
log(f"⚠ 模板解析失败（{e}）⇒ 跳过")
|
|||
|
|
return
|
|||
|
|
cfg["workspace"] = ws.as_posix()
|
|||
|
|
if dry:
|
|||
|
|
log(f"[dry-run] 将写 {dst}")
|
|||
|
|
return
|
|||
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|||
|
|
dst.write_text(json.dumps(cfg, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|||
|
|
log(f"已写 {dst}")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def cmd_apply(args) -> int:
|
|||
|
|
sp = settings_path()
|
|||
|
|
if not sp.is_file():
|
|||
|
|
log(f"🔴 找不到 {sp} ⇒ 停手")
|
|||
|
|
return 2
|
|||
|
|
ws = detect_workspace(args.workspace)
|
|||
|
|
if ws is None:
|
|||
|
|
log("🔴 未能推断工作区 ⇒ 请显式 `--workspace <路径>`（⛔ 不猜）")
|
|||
|
|
return 2
|
|||
|
|
docs = detect_docs_root(args.docs_root)
|
|||
|
|
log(f"包根={PKG} 工作区={ws} 文档库根={docs} 配置目录={config_dir()}")
|
|||
|
|
|
|||
|
|
raw = sp.read_text(encoding="utf-8")
|
|||
|
|
before = json.loads(raw)
|
|||
|
|
after = merge(before)
|
|||
|
|
|
|||
|
|
diff = "\n".join(
|
|||
|
|
difflib.unified_diff(
|
|||
|
|
json.dumps(before, ensure_ascii=False, indent=2).splitlines(),
|
|||
|
|
json.dumps(after, ensure_ascii=False, indent=2).splitlines(),
|
|||
|
|
fromfile="settings.json (before)", tofile="settings.json (after)", lineterm="",
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
print(diff)
|
|||
|
|
excluded = [
|
|||
|
|
str(h.get("command", ""))
|
|||
|
|
for arr in before.get("hooks", {}).values()
|
|||
|
|
for it in arr
|
|||
|
|
for h in it.get("hooks", [])
|
|||
|
|
if "decision_bridge" in str(h.get("command", ""))
|
|||
|
|
]
|
|||
|
|
log(f"（目视核对）⛔ 被排除、未被本包触碰的接线条数：{len(excluded)}（应 >0，且都在 decision_bridge 线上）")
|
|||
|
|
|
|||
|
|
write_roots(ws, docs, args.code_repo, args.dry_run)
|
|||
|
|
init_workspace(ws, args.dry_run)
|
|||
|
|
|
|||
|
|
if args.dry_run:
|
|||
|
|
log("[dry-run] ⛔ 未写 settings.json")
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
bak, wrote_orig = make_backup(sp, raw)
|
|||
|
|
sp.write_text(json.dumps(after, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|||
|
|
log(f"已写 {sp}")
|
|||
|
|
with open(LOG, "a", encoding="utf-8") as f:
|
|||
|
|
f.write(f"BACKUP={bak}\n")
|
|||
|
|
if wrote_orig is not None:
|
|||
|
|
f.write(f"ORIG_BACKUP={wrote_orig}\n")
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
def cmd_verify(args) -> int:
|
|||
|
|
fails: list[str] = []
|
|||
|
|
# 🔴 ①②③ 一律在**临时目录**里跑：钩子与 collabd 都会按 cwd 推导"工作区"。
|
|||
|
|
# 若直接用调用者的 cwd（常见＝包根），它们会把**技能包本身**当成工作区，
|
|||
|
|
# 于是在 `<包>/tmp/supervise-inbox/` 里写出运行态日志 ⇒ **污染技能包**（实测踩到）。
|
|||
|
|
scratch = tempfile.mkdtemp(prefix="sm-verify-")
|
|||
|
|
try:
|
|||
|
|
print("== ① 包内钩子空载荷自检（期望 rc=0）==")
|
|||
|
|
for event, _m, rel, extra, _t in HOOKS:
|
|||
|
|
p = PKG / rel
|
|||
|
|
if not p.is_file():
|
|||
|
|
fails.append(f"{rel} 不存在")
|
|||
|
|
print(f" 🔴 {event:16} {rel} ← 文件不存在")
|
|||
|
|
continue
|
|||
|
|
r = subprocess.run([py(), *extra, str(p)], input=b"{}", capture_output=True, cwd=scratch)
|
|||
|
|
tag = "ok" if r.returncode == 0 else f"rc={r.returncode}"
|
|||
|
|
if r.returncode != 0:
|
|||
|
|
fails.append(f"{rel} rc={r.returncode}")
|
|||
|
|
print(f" {tag:6} {event:16} {rel}")
|
|||
|
|
|
|||
|
|
print("== ② collabd --where ==")
|
|||
|
|
collabd = PKG / "scripts" / "collabd.py"
|
|||
|
|
r = subprocess.run([py(), str(collabd), "--where"], capture_output=True,
|
|||
|
|
cwd=scratch, env={**os.environ, **read_roots()})
|
|||
|
|
out = (r.stdout or b"").decode("utf-8", "replace").strip()
|
|||
|
|
print(f" rc={r.returncode}\n {out[:300]}")
|
|||
|
|
if r.returncode != 0:
|
|||
|
|
fails.append("collabd --where 非 0")
|
|||
|
|
|
|||
|
|
print("== ③ selftest ==")
|
|||
|
|
st = PKG / "scripts" / "selftest.py"
|
|||
|
|
if st.is_file():
|
|||
|
|
r = subprocess.run([py(), str(st)], capture_output=True,
|
|||
|
|
cwd=scratch, env={**os.environ, **read_roots()})
|
|||
|
|
tail = (r.stdout or b"").decode("utf-8", "replace").strip().splitlines()[-6:]
|
|||
|
|
for ln in tail:
|
|||
|
|
print(" " + ln)
|
|||
|
|
if r.returncode != 0 or "FAIL 0" not in " ".join(tail):
|
|||
|
|
fails.append("selftest 未通过（未见 FAIL 0）")
|
|||
|
|
else:
|
|||
|
|
print(" ⚠ 无 selftest.py")
|
|||
|
|
finally:
|
|||
|
|
shutil.rmtree(scratch, ignore_errors=True)
|
|||
|
|
|
|||
|
|
print("== 结论 ==")
|
|||
|
|
if fails:
|
|||
|
|
for f in fails:
|
|||
|
|
print(" 🔴 " + f)
|
|||
|
|
log(f"--verify 失败：{fails}")
|
|||
|
|
return 1
|
|||
|
|
print(" ✅ 全绿")
|
|||
|
|
log("--verify 全绿")
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
def cmd_uninstall(args) -> int:
|
|||
|
|
sp = settings_path()
|
|||
|
|
orig = sp.with_name(ORIG_BAK_NAME)
|
|||
|
|
ts_baks = sorted(p for p in sp.parent.glob(f"{BAK_PREFIX}*") if p.name != ORIG_BAK_NAME)
|
|||
|
|
|
|||
|
|
# 🔴 优先级：① 固定名 `-orig`（本包接管前的原状，最可信）→ ② 最早那份「未被本包接管」的时间戳备份。
|
|||
|
|
# ⛔ 不能取 `sorted(...)[-1]`：`--apply` 每次都备份 ⇒ 最新那份其实是**已装状态**，
|
|||
|
|
# 拿它还原 = 还原了个寂寞，而且脚本自比自还会报"逐字节相同"（**假绿**，上一棒实测撞到）。
|
|||
|
|
pristine = orig if orig.is_file() else None
|
|||
|
|
if pristine is not None:
|
|||
|
|
log(f"采用固定名原状备份：{pristine.name}")
|
|||
|
|
else:
|
|||
|
|
for b in ts_baks:
|
|||
|
|
try:
|
|||
|
|
if not points_into_pkg(b.read_text(encoding="utf-8")):
|
|||
|
|
pristine = b
|
|||
|
|
break
|
|||
|
|
except Exception:
|
|||
|
|
continue
|
|||
|
|
if pristine is not None:
|
|||
|
|
log(f"⚠ 无 `-orig` 备份 ⇒ 回落用最早的原状时间戳备份：{pristine.name}")
|
|||
|
|
if pristine is None:
|
|||
|
|
log(f"🔴 {len(ts_baks)} 份时间戳备份里没有一份是「未被本包接管」的原状，且无 `-orig` ⇒ 停手（⛔ 不拿已装状态糊弄）")
|
|||
|
|
return 2
|
|||
|
|
|
|||
|
|
# 还原前先把"当前状态"留一份 ⇒ `--uninstall` 本身也可回退。
|
|||
|
|
stamp = unique_backup_path(sp).name[len(BAK_PREFIX):]
|
|||
|
|
pre = sp.with_name(f"{BAK_PREFIX}preuninstall-{stamp}")
|
|||
|
|
n = 1
|
|||
|
|
while pre.exists():
|
|||
|
|
pre = sp.with_name(f"{BAK_PREFIX}preuninstall-{stamp}-{n}")
|
|||
|
|
n += 1
|
|||
|
|
shutil.copy2(sp, pre)
|
|||
|
|
log(f"（还原前存照）→ {pre}")
|
|||
|
|
|
|||
|
|
shutil.copy2(pristine, sp)
|
|||
|
|
same = hashlib.md5(pristine.read_bytes()).hexdigest() == hashlib.md5(sp.read_bytes()).hexdigest()
|
|||
|
|
clean = not points_into_pkg(sp.read_text(encoding="utf-8"))
|
|||
|
|
log(f"已用 {pristine.name} 还原 {sp}")
|
|||
|
|
log(f" · 与备份逐字节相同：{'✅' if same else '🔴'}")
|
|||
|
|
log(f" · 还原后已无指向本包的钩子：{'✅' if clean else '🔴'}")
|
|||
|
|
return 0 if (same and clean) else 1
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main() -> int:
|
|||
|
|
ap = argparse.ArgumentParser(description="session-mechanism 一键配置")
|
|||
|
|
g = ap.add_mutually_exclusive_group(required=True)
|
|||
|
|
g.add_argument("--apply", action="store_true", help="真装")
|
|||
|
|
g.add_argument("--dry-run", action="store_true", help="只打印 diff")
|
|||
|
|
g.add_argument("--verify", action="store_true", help="装完自检")
|
|||
|
|
g.add_argument("--uninstall", action="store_true", help="还原 settings.json")
|
|||
|
|
ap.add_argument("--workspace", help="工作区根（不传则自动推断）")
|
|||
|
|
ap.add_argument("--docs-root", help="文档库根（不传则自动推断/沿用 roots.env）")
|
|||
|
|
ap.add_argument("--code-repo", help="代码仓根（可选）")
|
|||
|
|
args = ap.parse_args()
|
|||
|
|
if args.apply or args.dry_run:
|
|||
|
|
return cmd_apply(args)
|
|||
|
|
if args.verify:
|
|||
|
|
return cmd_verify(args)
|
|||
|
|
return cmd_uninstall(args)
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
raise SystemExit(main())
|