85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
sync_wsl.py — WSL 插件/配置批量同步（Win 生效份 -> WSL 生效份）。
|
|
用法: python tools/sync_wsl.py [--only 名字 ...]
|
|
机制: 每个文件经 `wsl -- bash -c "cat > dst && sha256sum dst"` stdin 管道推送
|
|
(绕 UNC OSError 22), 双侧 sha256 校验, 逐个报告, 任一失败退出码 1。
|
|
同步清单(按「实际加载路径」对齐, 不是源码路径):
|
|
compaction-cn.ts / ai-mcp.ts / oc-session-all/tui.js / mcpcfg.txt
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import io
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
try:
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
|
|
except Exception:
|
|
pass
|
|
|
|
DISTRO = "Ubuntu-24.04"
|
|
WIN = os.path.join(os.path.expanduser("~"), ".config", "opencode")
|
|
ITEMS = [
|
|
("compaction-cn", os.path.join(WIN, "plugins", "compaction-cn.ts"),
|
|
"/root/.config/opencode/plugins/compaction-cn.ts"),
|
|
("ai-mcp", os.path.join(WIN, "plugins", "ai-mcp.ts"),
|
|
"/root/.config/opencode/plugins/ai-mcp.ts"),
|
|
("tui.js", os.path.join(WIN, "plugins", "oc-session-all", "tui.js"),
|
|
"/root/.config/opencode/plugins/oc-session-all/tui.js"),
|
|
("mcpcfg", os.path.join(WIN, "mcpcfg.txt"),
|
|
"/root/.config/opencode/mcpcfg.txt"),
|
|
]
|
|
|
|
|
|
def sha256(b):
|
|
return hashlib.sha256(b).hexdigest()
|
|
|
|
|
|
def sync_one(name, local, remote):
|
|
if not os.path.exists(local):
|
|
print("[%s] SKIP 本地不存在: %s" % (name, local))
|
|
return False
|
|
data = open(local, "rb").read()
|
|
local_sha = sha256(data)
|
|
cmd = 'cat > "%s" && sha256sum "%s"' % (remote, remote)
|
|
try:
|
|
r = subprocess.run(
|
|
["wsl.exe", "-d", DISTRO, "--", "bash", "-c", cmd],
|
|
input=data, capture_output=True, timeout=120,
|
|
)
|
|
except Exception as e:
|
|
print("[%s] FAIL 启动 wsl.exe 异常: %s" % (name, e))
|
|
return False
|
|
if r.returncode != 0:
|
|
err = (r.stderr or b"").decode("utf-8", errors="replace")[:200]
|
|
print("[%s] FAIL push rc=%d %s" % (name, r.returncode, err))
|
|
return False
|
|
out = r.stdout or b""
|
|
text = out.decode("utf-8", errors="replace")
|
|
if "sha256" not in text and len(text.split()) < 2:
|
|
text = out.decode("utf-16-le", errors="replace")
|
|
parts = text.split()
|
|
remote_sha = parts[0].strip() if parts else ""
|
|
ok = remote_sha == local_sha
|
|
print("[%s] %s %d bytes sha=%s" % (name, "OK " if ok else "FAIL", len(data), local_sha[:12]))
|
|
return ok
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--only", nargs="*", default=None, help="只同步指定名字")
|
|
args = ap.parse_args()
|
|
items = [it for it in ITEMS if not args.only or it[0] in args.only]
|
|
ok = 0
|
|
for name, local, remote in items:
|
|
if sync_one(name, local, remote):
|
|
ok += 1
|
|
print("result: %d/%d OK" % (ok, len(items)))
|
|
return 0 if ok == len(items) else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|