From 873d965ef3b071d6fe80eae5a63b7a719f7eede1 Mon Sep 17 00:00:00 2001 From: ypshidifu Date: Wed, 23 Sep 2026 12:06:32 +0800 Subject: [PATCH] =?UTF-8?q?wsl-round-mcp=EF=BC=9A=E6=9C=AC=E6=9C=BA?= =?UTF-8?q?=E4=B8=8E=20WSL=20OpenCode=20=E4=BA=92=E9=80=9A=E7=9A=84=20CLI?= =?UTF-8?q?=20=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ocskill.py（skill 互通，零第三方依赖）： - list 运行时连接 WSL 的 OC（/api/skill 与 /skill 双端点，Bearer→Basic 自动降级），连不上则降级扫描 skills 目录 - push 接收 zip / 目录 / 单个 SKILL.md / base64（显式或按内容嗅探），先解码归一，再校验标准 skill 格式 校验四项：有 SKILL.md、有 YAML frontmatter、frontmatter 有 name 且合法、有 description 含 zip-slip 目录穿越防护；同名已存在需 --force（覆盖前自动备份） - pull 反向把 WSL 里的 skill 拷回本机；dirs 查看两端目录与连接状态 - 写入 \\wsl.localhost\<发行版>\root\.config\opencode\skills\，OC 原生 watch 自动热加载，无需重启 mcpctl.py（MCP 配置对调开关，与插件同一套逻辑）： - status / on / off / mode / backup；读同一份 mcpcfg.txt，状态只看活文件是否含 mcp.servers - 对调后调用 POST /api/location/reload 热重载；模式门控与插件一致（关闭一律拒绝） 附 6 个双击 bat（英文文案规避 GBK 编码问题，内部 chcp 65001）与中文 README。 --- .gitignore | 14 ++ README.md | 137 +++++++++++ mcp-off.bat | 15 ++ mcp-on.bat | 15 ++ mcp-status.bat | 15 ++ mcpctl.py | 388 ++++++++++++++++++++++++++++++ ocskill.py | 629 +++++++++++++++++++++++++++++++++++++++++++++++++ skill-list.bat | 16 ++ skill-pull.bat | 23 ++ skill-push.bat | 25 ++ 10 files changed, 1277 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 mcp-off.bat create mode 100644 mcp-on.bat create mode 100644 mcp-status.bat create mode 100644 mcpctl.py create mode 100644 ocskill.py create mode 100644 skill-list.bat create mode 100644 skill-pull.bat create mode 100644 skill-push.bat diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0957dbf --- /dev/null +++ b/.gitignore @@ -0,0 +1,14 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ + +# 本地回滚/备份产物 +*.bak-* +skills-backup/ + +# 编辑器与系统 +.vscode/ +.idea/ +.DS_Store +Thumbs.db diff --git a/README.md b/README.md new file mode 100644 index 0000000..66becb7 --- /dev/null +++ b/README.md @@ -0,0 +1,137 @@ +# wsl-round-mcp — 本机 ⇄ WSL OpenCode 互通工具 + +打通**本机 OpenCode** 与 **WSL 里的 OpenCode**：skill 列表互查、skill 一键推送、MCP 配置对调开关。 +全部零第三方依赖，只要装了 Python 3 就能跑。 + +--- + +## 一、双击就能用（小白版） + +把整个 `wsl-round-mcp` 文件夹拷到任意位置，双击对应的 bat： + +| 文件 | 作用 | +| --- | --- | +| `skill-list.bat` | 连接 WSL 的 OC，列出它当前有哪些 skill | +| `skill-push.bat` | 会让你粘贴一个 skill 路径（zip / 目录 / SKILL.md / base64），校验格式后写进 WSL 的 skill 目录 | +| `skill-pull.bat` | 会让你输入 skill 名，把它从 WSL 拷回本机 `skills-backup\` | +| `mcp-status.bat` | 查看 MCP 当前是开是关、第1行是什么模式 | +| `mcp-on.bat` | 把配置对调为「含 MCP」并热重载 | +| `mcp-off.bat` | 把配置对调为「安静版（无 MCP）」并热重载 | + +> bat 内部会自动 `chcp 65001`，中文输出不会乱码。 +> 没装 Python 或没加进 PATH，会提示 `Python not found in PATH`。 + +--- + +## 二、命令行用法 + +### 1. `ocskill.py` — skill 互通 + +```bash +# 列出 WSL OC 的 skill（API 连不上会自动退回目录扫描） +python ocskill.py list + +# 推送：压缩包（自动解压） +python ocskill.py push .\my-skill.zip + +# 推送：整个 skill 目录 +python ocskill.py push .\my-skill + +# 推送：单个 SKILL.md（会自动包成以 name 命名的目录） +python ocskill.py push .\SKILL.md + +# 推送：base64 内容 +python ocskill.py push <一长串base64> --base64 + +# 只校验格式、不写入 +python ocskill.py push .\my-skill --dry-run + +# 覆盖同名 skill（先自动备份成 .bak-时间戳） +python ocskill.py push .\my-skill --force + +# 反向：把 WSL 里的 skill 拷回本机 +python ocskill.py pull java-expert --out .\skills-backup + +# 看两端目录与 API 连接状态 +python ocskill.py dirs +``` + +**push 的完整流程**（对应「先解码 → 确认标准 skill 格式 → 传入 skill 目录」）： + +1. **解码归一**：zip 就解压（带 zip-slip 目录穿越防护），目录原样用， + 单个 `.md` 会包成目录，base64 字符串先解成文件再判断是 zip 还是 md。 +2. **定位 `SKILL.md`**：允许压缩包里多套一层父目录（最深 3 层）。 +3. **校验标准 skill 格式**，四项全过才算通过： + - 找得到 `SKILL.md` + - 文件以 `---` 开头，有 YAML frontmatter + - frontmatter 里有 `name`，且只含 `字母/数字/._-` + - frontmatter 里有 `description`（大模型靠它决定何时调用） +4. **写入** `\\wsl.localhost\<发行版>\root\.config\opencode\skills\\`， + OC 会 watch 这个目录（约 100ms 去抖后自动 rescan 并推 `skill.reload`），**不用重启**。 + +常用参数： + +| 参数 | 默认 | 说明 | +| --- | --- | --- | +| `--distro` | `Ubuntu-24.04` | WSL 发行版名 | +| `--url` | 自动探测 | 显式指定 OC 地址，如 `http://localhost:4096` | +| `--port` | `4096` | OC 端口（等价于环境变量 `OC_PORT`） | +| `--password` | 自动读 | OC 的 `server.password`，先试 Bearer、401 再试 Basic | + +### 2. `mcpctl.py` — MCP 配置对调开关 + +```bash +python mcpctl.py status # 看状态（只读） +python mcpctl.py on # 对调成含 MCP + 热重载 +python mcpctl.py off # 对调成安静版 + 热重载 +python mcpctl.py mode # 看第1行 +python mcpctl.py mode 手动 # 改第1行（自动/手动/关闭，或 auto/manual/off） +python mcpctl.py status --target wsl # 操作 WSL 那一份配置 +python mcpctl.py off --no-reload # 只对调文件，不调用重载 +``` + +它读的正是 `mcpcfg.txt` 那三行： + +``` +自动 ← 第1行：模式 +C:\...\opencode.json.1 ← 第2行：含 MCP 的原始配置 +C:\...\opencode.json ← 第3行：删了 MCP 的安静配置 +``` + +**状态判定只看活文件**（文件名为 `opencode.json` 的那份）里 `mcp.servers` 是否非空， +所以对调之后第2/3行的身份怎么漂移都不影响判断——这与 TUI 插件、server 插件完全一致。 + +模式语义（三个入口共用同一套规则）： + +| 模式 | `/mcpon` `/mcpoff` | 大模型（server 插件） | 本工具 `mcpctl.py` | +| --- | --- | --- | --- | +| 自动 | 可用（视为手动切换一次） | 可用 | 可用 | +| 手动 | **仅此可用** | 拒绝 | 可用（等价于用户输入命令） | +| 关闭 | 不注册 | 拒绝 | 拒绝（可用 `--ignore-mode` 强行） | + +热重载打的是 `POST /api/location/reload`，会重建 location 并重读磁盘配置，MCP 立即生效； +失败时配置文件已经对调完成，重启 OC 一样生效。 + +--- + +## 三、连不上 OC 怎么办 + +1. 确认 WSL 里的 OC 起了：`opencode serve --port 4096` +2. 端口不对就加 `--port`，或设环境变量 `OC_PORT` +3. 提示 `鉴权失败 (401)` → 用 `--password` 传 `opencode.json` 里 `server.password` 的值 +4. `localhost` 不通 → 工具会自动 `wsl hostname -I` 取 WSL 真实 IP 再试 +5. 全都失败 → `list` 会自动降级为**直接扫目录**，照样能看到 skill 列表 + +--- + +## 四、和另外两个仓库的关系 + +本目录是独立 git 仓库，**分支名 = 文件夹名** `wsl-round-mcp`，远程与主仓库共用同一个根： + +| 文件夹 | 分支 | 内容 | +| --- | --- | --- | +| `oc_session/`（主仓库） | `main` | TUI 插件 `oc-session-all`（`/sessions-all`、`/mcpon`、`/mcpoff`） | +| `oc_session/ai-mcp/` | `ai-mcp` | server 插件 `ai-mcp.ts`，给大模型用的 `mcp_switch` 工具 | +| `oc_session/wsl-round-mcp/` | `wsl-round-mcp` | 本目录：skill 互通 CLI + MCP 命令行开关 | + +远程：`https://git.ypshidifu.cn/youpaishidifu/oc_session.git` diff --git a/mcp-off.bat b/mcp-off.bat new file mode 100644 index 0000000..9de772d --- /dev/null +++ b/mcp-off.bat @@ -0,0 +1,15 @@ +@echo off +chcp 65001 >nul +title MCP off (swap config + reload) +set "HERE=%~dp0" + +where python >nul 2>nul +if errorlevel 1 ( + echo [ERROR] Python not found in PATH. + pause + exit /b 1 +) + +python "%HERE%mcpctl.py" off +echo. +pause diff --git a/mcp-on.bat b/mcp-on.bat new file mode 100644 index 0000000..2782d3d --- /dev/null +++ b/mcp-on.bat @@ -0,0 +1,15 @@ +@echo off +chcp 65001 >nul +title MCP on (swap config + reload) +set "HERE=%~dp0" + +where python >nul 2>nul +if errorlevel 1 ( + echo [ERROR] Python not found in PATH. + pause + exit /b 1 +) + +python "%HERE%mcpctl.py" on +echo. +pause diff --git a/mcp-status.bat b/mcp-status.bat new file mode 100644 index 0000000..37b3c18 --- /dev/null +++ b/mcp-status.bat @@ -0,0 +1,15 @@ +@echo off +chcp 65001 >nul +title MCP status +set "HERE=%~dp0" + +where python >nul 2>nul +if errorlevel 1 ( + echo [ERROR] Python not found in PATH. + pause + exit /b 1 +) + +python "%HERE%mcpctl.py" status +echo. +pause diff --git a/mcpctl.py b/mcpctl.py new file mode 100644 index 0000000..be4c31f --- /dev/null +++ b/mcpctl.py @@ -0,0 +1,388 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +mcpctl —— 命令行版的 MCP 配置对调开关（与 TUI 插件 / server 插件同一套逻辑） + +做什么： + status 查看当前模式与 MCP 开/关状态（只读，不改任何文件） + on 把配置对调为「含 MCP」的原始版，并调用 reload 热重载 + off 把配置对调为「不含 MCP」的安静版，并调用 reload 热重载 + mode 查看或修改 mcpcfg.txt 第1行（自动 / 手动 / 关闭） + +模式语义（与插件完全一致）： + 自动 大模型可通过 server 插件切换；/mcpon /mcpoff 与本工具视为手动切换一次 + 手动 仅允许 /mcpon /mcpoff 与本工具（用户意图），大模型的切换请求会被拒绝 + 关闭 功能整体禁用，on/off 都会被拒绝 + +状态判定只看活文件（basename 为 opencode.json 的那份）里 mcp.servers 是否非空， +与 mcpcfg.txt 第2/3行的行语义无关，因此对调之后身份漂移也不影响判断。 +""" + +import argparse +import base64 +import datetime +import json +import os +import re +import sys +import urllib.error +import urllib.request +from pathlib import Path + +# ---------------------------------------------------------------- 控制台编码 +def _fix_console() -> None: + for stream in (sys.stdout, sys.stderr): + try: + stream.reconfigure(encoding="utf-8", errors="replace") # type: ignore[attr-defined] + except Exception: + pass + + +_fix_console() + +MODE_LABEL = {"auto": "自动", "manual": "手动", "off": "关闭"} +LABEL_MODE = { + "自动": "auto", + "auto": "auto", + "audo": "auto", # 用户手误拼写，保持与插件一致 + "手动": "manual", + "manual": "manual", + "hand": "manual", + "on": "manual", + "关闭": "off", + "off": "off", + "disable": "off", + "disabled": "off", +} + +DEFAULT_DISTRO = os.environ.get("OC_SKILL_DISTRO", "Ubuntu-24.04") +DEFAULT_PORT = int(os.environ.get("OC_PORT", "4096")) + + +# ---------------------------------------------------------------- 路径工具 +def _basename(path: str) -> str: + """跨平台取文件名：同时认 / 与 \\，避免在 Windows 上解析 WSL 路径出错。""" + return re.split(r"[\\/]", path.strip())[-1] + + +def _home_config() -> Path: + return Path.home() / ".config" / "opencode" + + +def wsl_config_dir(distro: str) -> Path: + if os.name == "nt": + return Path(f"\\\\wsl.localhost\\{distro}\\root\\.config\\opencode") + return Path("/root/.config/opencode") + + +def resolve_cfg_file(target: str, distro: str) -> Path: + if target == "wsl": + return wsl_config_dir(distro) / "mcpcfg.txt" + return _home_config() / "mcpcfg.txt" + + +# ---------------------------------------------------------------- mcpcfg 解析 +def read_cfg(path: Path) -> dict: + if not path.exists(): + return { + "mode": "off", + "error": f"缺少配置文件：{path}（配置缺失时功能视为关闭）", + } + try: + lines = path.read_text(encoding="utf-8", errors="replace").splitlines() + except Exception as exc: + return {"mode": "off", "error": f"读取 {path} 失败：{exc}"} + + key = (lines[0] if lines else "").strip().lower() + mode = LABEL_MODE.get(key, "auto" if key == "" else None) or "auto" + + on_path = (lines[1] if len(lines) > 1 else "").strip() + quiet_path = (lines[2] if len(lines) > 2 else "").strip() + + active = shadow = None + if _basename(on_path) == "opencode.json": + active, shadow = on_path, quiet_path + elif _basename(quiet_path) == "opencode.json": + active, shadow = quiet_path, on_path + + result = { + "mode": mode, + "on_path": on_path, + "quiet_path": quiet_path, + "active": active, + "shadow": shadow, + "error": None, + } + if mode != "off" and not active: + result["error"] = "mcpcfg.txt 第2/3行必须有一个是 opencode.json 的完整路径" + elif mode != "off" and (not Path(on_path).exists() or not Path(quiet_path).exists()): + result["error"] = "mcpcfg.txt 第2/3行指向的文件不存在（请检查路径）" + return result + + +def active_has_mcp(active: str | None) -> bool: + try: + if not active: + return False + data = json.loads(Path(active).read_text(encoding="utf-8")) + servers = (data.get("mcp") or {}).get("servers") or {} + return bool(servers) + except Exception: + return False + + +def swap_files(a: str, b: str) -> None: + pa, pb = Path(a), Path(b) + ca, cb = pa.read_bytes(), pb.read_bytes() + pa.write_bytes(cb) + pb.write_bytes(ca) + + +# ---------------------------------------------------------------- reload +def _read_password(cfg_dir: Path) -> str: + try: + data = json.loads((cfg_dir / "opencode.json").read_text(encoding="utf-8")) + pwd = (data.get("server") or {}).get("password") + return str(pwd) if pwd else "" + except Exception: + return "" + + +def _wsl_ip(distro: str) -> str: + if os.name != "nt": + return "" + try: + import subprocess + + out = subprocess.run( + ["wsl.exe", "-d", distro, "--", "bash", "-lc", "hostname -I"], + capture_output=True, + timeout=15, + ) + for tok in (out.stdout or b"").decode("utf-8", "ignore").split(): + if re.fullmatch(r"\d{1,3}(\.\d{1,3}){3}", tok): + return tok + except Exception: + pass + return "" + + +def _post(url: str, password: str, timeout: float) -> tuple[int, str]: + req = urllib.request.Request(url, method="POST", data=b"{}") + req.add_header("content-type", "application/json") + req.add_header("accept", "application/json") + if password: + req.add_header("authorization", f"Bearer {password}") + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.status, resp.read().decode("utf-8", "replace") + except urllib.error.HTTPError as exc: + if exc.code == 401 and password: + raw = base64.b64encode(f":{password}".encode()).decode("ascii") + req2 = urllib.request.Request(url, method="POST", data=b"{}") + req2.add_header("content-type", "application/json") + req2.add_header("authorization", f"Basic {raw}") + try: + with urllib.request.urlopen(req2, timeout=timeout) as resp: + return resp.status, resp.read().decode("utf-8", "replace") + except urllib.error.HTTPError as exc2: + return exc2.code, str(exc2) + return exc.code, str(exc) + except Exception as exc: + return 0, str(exc) + + +def do_reload(target: str, args) -> tuple[bool, str]: + password = args.password or _read_password( + wsl_config_dir(args.distro) if target == "wsl" else _home_config() + ) + hosts = [f"http://localhost:{args.port}", f"http://127.0.0.1:{args.port}"] + if target == "wsl" and os.name == "nt": + ip = _wsl_ip(args.distro) + if ip: + hosts.append(f"http://{ip}:{args.port}") + + last = "" + for host in hosts: + code, body = _post(host + "/api/location/reload", password, args.timeout) + if 200 <= code < 300: + return True, f"热重载成功（{host}/api/location/reload）" + if code in (401, 403): + last = f"{host} 鉴权失败（HTTP {code}），请用 --password 指定 server.password" + elif code == 404: + last = f"{host} 未提供 /api/location/reload（可能需要重启 OC 生效）" + elif code: + last = f"{host} 返回 HTTP {code}：{body[:200]}" + else: + last = f"{host} 连不上（OC 可能没在运行）" + return False, last + + +# ---------------------------------------------------------------- 子命令 +def _stamp() -> str: + return datetime.datetime.now().strftime("%Y%m%d%H%M%S") + + +def print_state(cfg: dict) -> bool: + has = active_has_mcp(cfg.get("active")) + print(f"模式：{MODE_LABEL.get(cfg['mode'], cfg['mode'])}（mcpcfg.txt 第1行）") + print(f"MCP ：{'已开启（配置含 mcp.servers）' if has else '已关闭（安静，无 mcp 配置）'}") + print(f"活配置：{cfg.get('active') or '（未识别）'}") + if cfg.get("error"): + print(f"警告：{cfg['error']}") + return has + + +def cmd_status(args) -> int: + path = resolve_cfg_file(args.target, args.distro) + print(f"配置文件：{path}") + cfg = read_cfg(path) + print_state(cfg) + return 0 if not cfg.get("error") else 2 + + +def _switch(args, want_on: bool) -> int: + path = resolve_cfg_file(args.target, args.distro) + print(f"配置文件：{path}") + cfg = read_cfg(path) + if cfg.get("error"): + print(f"失败：{cfg['error']}") + return 2 + + action = "开启" if want_on else "关闭" + if cfg["mode"] == "off" and not args.ignore_mode: + print(f"拒绝：MCP 开关功能已关闭（mcpcfg.txt 第1行=关闭）。") + print("如需使用，请把第1行改成「自动」或「手动」。") + return 3 + + has = active_has_mcp(cfg["active"]) + if has == want_on: + print(f"无需切换：MCP 已是{'开启' if want_on else '关闭（安静）'}状态。") + print_state(cfg) + return 0 + + swap_files(cfg["on_path"], cfg["quiet_path"]) + after = active_has_mcp(cfg["active"]) + if after != want_on: + print("失败：配置对调后状态校验不符，请检查 mcpcfg.txt 第2/3行路径。") + return 4 + + print(f"配置已对调，MCP 现在是「{'开启' if after else '关闭'}」。") + + if args.no_reload: + print("已按 --no-reload 跳过热重载，需重启 OC 生效。") + return 0 + + ok, msg = do_reload(args.target, args) + if ok: + print(f"{msg}，MCP 配置立即生效。") + return 0 + print(f"热重载失败：{msg}") + print("配置文件已经对调完成，重启 OpenCode 也能生效。") + return 5 + + +def cmd_on(args) -> int: + return _switch(args, True) + + +def cmd_off(args) -> int: + return _switch(args, False) + + +def cmd_mode(args) -> int: + path = resolve_cfg_file(args.target, args.distro) + if not path.exists(): + print(f"配置文件不存在：{path}") + return 2 + lines = path.read_text(encoding="utf-8", errors="replace").splitlines() + + if not args.value: + cur = (lines[0] if lines else "").strip() + print(f"当前第1行：{cur or '（空，默认自动）'}") + print("可选值：自动 / 手动 / 关闭（也支持 auto / manual / off）") + return 0 + + key = args.value.strip().lower() + if key not in LABEL_MODE: + print(f"无法识别的模式：{args.value}（可选：自动 / 手动 / 关闭 / auto / manual / off）") + return 2 + + while len(lines) < 3: + lines.append("") + lines[0] = LABEL_MODE[key] + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + print(f"已把第1行设为：{MODE_LABEL[LABEL_MODE[key]]}（{path}）") + return 0 + + +def cmd_backup(args) -> int: + path = resolve_cfg_file(args.target, args.distro) + if not path.exists(): + print(f"配置文件不存在：{path}") + return 2 + bak = path.with_name(path.name + f".bak-{_stamp()}") + bak.write_bytes(path.read_bytes()) + print(f"已备份：{bak}") + return 0 + + +# ---------------------------------------------------------------- 入口 +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="mcpctl", + description="MCP 配置对调开关（status / on / off / mode），与插件共用同一份 mcpcfg.txt", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=( + "示例：\n" + " python mcpctl.py status\n" + " python mcpctl.py on\n" + " python mcpctl.py off --no-reload\n" + " python mcpctl.py mode 手动\n" + " python mcpctl.py status --target wsl\n" + ), + ) + p.add_argument("--target", choices=["local", "wsl"], default="local", help="操作本机还是 WSL 的配置（默认 local）") + p.add_argument("--distro", default=DEFAULT_DISTRO, help=f"WSL 发行版名（默认 {DEFAULT_DISTRO}）") + p.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"OC 端口（默认 {DEFAULT_PORT}）") + p.add_argument("--timeout", type=float, default=5.0, help="HTTP 超时秒数") + p.add_argument("--password", default="", help="OC server.password（默认自动从 opencode.json 读）") + p.add_argument("--no-reload", action="store_true", help="只对调文件，不调用热重载") + p.add_argument("--ignore-mode", action="store_true", help="忽略第1行的「关闭」限制（慎用）") + + sub = p.add_subparsers(dest="cmd") + + sp = sub.add_parser("status", help="查看当前模式与 MCP 开关状态") + sp.set_defaults(func=cmd_status) + + sp = sub.add_parser("on", help="对调为含 MCP 的配置并热重载") + sp.set_defaults(func=cmd_on) + + sp = sub.add_parser("off", help="对调为安静配置（删 MCP）并热重载") + sp.set_defaults(func=cmd_off) + + sp = sub.add_parser("mode", help="查看或修改 mcpcfg.txt 第1行的模式") + sp.add_argument("value", nargs="?", help="要设置的模式：自动 / 手动 / 关闭（不传则只查看）") + sp.set_defaults(func=cmd_mode) + + sp = sub.add_parser("backup", help="备份当前 mcpcfg.txt") + sp.set_defaults(func=cmd_backup) + + return p + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if not getattr(args, "cmd", None): + parser.print_help() + return 0 + try: + return args.func(args) + except KeyboardInterrupt: + print("\n已取消。") + return 130 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/ocskill.py b/ocskill.py new file mode 100644 index 0000000..88df7e5 --- /dev/null +++ b/ocskill.py @@ -0,0 +1,629 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +ocskill —— 打通本机与 WSL OpenCode 的 skill 互通 + +做什么： + list 运行时连接 WSL 里的 OpenCode，返回它的 skill 列表 + push 把一个新的 skill 文件（压缩包 / 目录 / 单个 SKILL.md / base64） + 先解码并校验是不是标准 skill 格式，然后写进 OC 的 skill 目录 + （压缩包会额外做一次解压；OC 自带目录 watch，写完即热加载） + pull 反向操作：把 WSL 里的某个 skill 拷回本机 + dirs 显示两端 skill 目录位置 + +零依赖：只用 Python 标准库（urllib / zipfile / shutil），无需 pip install。 +""" + +import argparse +import base64 +import binascii +import json +import os +import re +import shutil +import subprocess +import sys +import tempfile +import urllib.error +import urllib.request +import zipfile +from pathlib import Path + +# ---------------------------------------------------------------- 控制台编码 +def _fix_console() -> None: + for stream in (sys.stdout, sys.stderr): + try: + stream.reconfigure(encoding="utf-8", errors="replace") # type: ignore[attr-defined] + except Exception: + pass + + +_fix_console() + +# ---------------------------------------------------------------- 常量与默认值 +DEFAULT_DISTRO = os.environ.get("OC_SKILL_DISTRO", "Ubuntu-24.04") +DEFAULT_PORT = int(os.environ.get("OC_PORT", "4096")) +SKILL_MD = "SKILL.md" + +# 标准 skill 目录：OC 会 watch 每个 config entry 下的 skill/ 与 skills/ +WINDOWS_WSL_ROOT = r"\\wsl.localhost\{distro}\root\.config\opencode\skills" +LINUX_SKILL_ROOT = "/root/.config/opencode/skills" +WIN_LOCAL_ROOT = str(Path.home() / ".config" / "opencode" / "skills") + + +def is_wsl_inside() -> bool: + """当前是否已经跑在 WSL/Linux 里。""" + if os.name != "posix": + return False + if sys.platform.startswith("linux"): + try: + with open("/proc/version", "r", encoding="utf-8", errors="ignore") as fh: + if "microsoft" in fh.read().lower(): + return True + except Exception: + pass + return True # 纯 Linux 也走这个分支 + return False + + +def wsl_skill_root(distro: str) -> str: + """WSL 侧 skill 目录（在 Windows 下用 UNC 路径直接读写，无需进 WSL）。""" + if is_wsl_inside(): + return LINUX_SKILL_ROOT + return WINDOWS_WSL_ROOT.format(distro=distro) + + +# ---------------------------------------------------------------- OC 连接 +def _read_server_password(config_dir: Path) -> str: + """从 opencode.json 读 server.password（WSL 端配置里有这一项）。""" + cfg = config_dir / "opencode.json" + try: + data = json.loads(cfg.read_text(encoding="utf-8")) + pwd = (data.get("server") or {}).get("password") + return str(pwd) if pwd else "" + except Exception: + return "" + + +def _candidate_urls(explicit: str | None, distro: str, port: int) -> list[str]: + if explicit: + return [explicit.rstrip("/")] + urls = [ + f"http://localhost:{port}", + f"http://127.0.0.1:{port}", + ] + # Windows 下再试 WSL 的真实 IP（localhost 转发失效时兜底） + if os.name == "nt": + ip = _wsl_ip(distro) + if ip: + urls.append(f"http://{ip}:{port}") + seen, out = set(), [] + for u in urls: + if u not in seen: + seen.add(u) + out.append(u) + return out + + +def _wsl_ip(distro: str) -> str: + try: + out = subprocess.run( + ["wsl.exe", "-d", distro, "--", "bash", "-lc", "hostname -I"], + capture_output=True, + timeout=15, + ) + text = (out.stdout or b"").decode("utf-8", errors="ignore") + for tok in text.split(): + if re.fullmatch(r"\d{1,3}(\.\d{1,3}){3}", tok): + return tok + except Exception: + pass + return "" + + +def _build_request(url: str, password: str, method: str = "GET") -> urllib.request.Request: + req = urllib.request.Request(url, method=method) + req.add_header("accept", "application/json") + if password: + # OC 支持 Bearer 与 Basic 两种，先试 Bearer（401 时调用方会换 Basic 重试） + req.add_header("authorization", f"Bearer {password}") + return req + + +def _try_fetch(url: str, password: str, timeout: float) -> tuple[int, bytes]: + """返回 (status, body)。401 时自动改用 Basic 重试一次。""" + req = _build_request(url, password) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.status, resp.read() + except urllib.error.HTTPError as exc: + if exc.code == 401 and password: + raw = base64.b64encode(f":{password}".encode("utf-8")).decode("ascii") + req2 = _build_request(url, "") + req2.add_header("authorization", f"Basic {raw}") + try: + with urllib.request.urlopen(req2, timeout=timeout) as resp: + return resp.status, resp.read() + except urllib.error.HTTPError as exc2: + return exc2.code, exc2.read() + return exc.code, (exc.read() if hasattr(exc, "read") else b"") + except Exception: + return 0, b"" + + +def connect_oc(args) -> tuple[str | None, object]: + """ + 连接 OC，返回 (base_url, payload)。 + 依次尝试 /api/skill 与 /skill 两个已知端点；全失败则 payload=None。 + """ + password = args.password or _read_server_password(Path(args.remote_config_dir)) + last_err = "" + for base in _candidate_urls(args.url, args.distro, args.port): + for path in ("/api/skill", "/skill"): + code, body = _try_fetch(base + path, password, args.timeout) + if code == 200: + try: + return base, json.loads(body.decode("utf-8", errors="replace")) + except Exception as exc: + last_err = f"{base}{path} 返回非 JSON：{exc}" + elif code in (401, 403): + last_err = f"{base}{path} 鉴权失败（HTTP {code}），请用 --password 指定 server.password" + elif code: + last_err = f"{base}{path} 返回 HTTP {code}" + else: + last_err = f"{base} 连接失败（OC 可能没在运行，试试 opencode serve --port {args.port}）" + return None, None if not last_err else last_err + + +# ---------------------------------------------------------------- skill 解析 +def extract_skills(payload) -> list[dict]: + """容错解析 skill 列表：兼容数组 / {skills:[...]} / {data:{...}} 等形态。""" + items = None + if isinstance(payload, list): + items = payload + elif isinstance(payload, dict): + for key in ("skills", "items", "data", "result", "list"): + val = payload.get(key) + if isinstance(val, list): + items = val + break + if isinstance(val, dict): # data: { skills: [...] } + for inner in val.values(): + if isinstance(inner, list): + items = inner + break + if items is not None: + break + if items is None: + items = next((v for v in payload.values() if isinstance(v, list)), []) + if items is None: + return [] + + out = [] + for it in items: + if isinstance(it, str): + out.append({"name": it, "description": ""}) + elif isinstance(it, dict): + name = it.get("name") or it.get("id") or it.get("title") or "" + desc = it.get("description") or it.get("summary") or "" + out.append({"name": str(name), "description": str(desc)}) + return [x for x in out if x["name"]] + + +def scan_skill_dir(root: str) -> list[dict]: + """API 不可用时的兜底：直接扫目录。""" + out = [] + base = Path(root) + if not base.is_dir(): + return out + for child in sorted(base.iterdir()): + if not child.is_dir() or child.name.startswith("."): + continue + md = child / SKILL_MD + desc = "" + if md.is_file(): + desc = _frontmatter(md.read_text(encoding="utf-8", errors="replace")).get("description", "") + out.append({"name": child.name, "description": desc}) + return out + + +BLOCK_MARKERS = {">", "|", ">-", "|-", ">+", "|+", ">", "|"} + + +def _has_cjk(text: str) -> bool: + """是否含中日韩字符（决定块标量折行时要不要补空格）。""" + return any("\u4e00" <= ch <= "\u9fff" or "\u3000" <= ch <= "\u303f" for ch in text) + + +def _frontmatter(text: str) -> dict: + """ + 解析 SKILL.md 顶部的 YAML frontmatter（只取扁平 key: value，不引第三方库）。 + 会正确处理块标量（description: > / |）——否则描述会显示成一个 ">"。 + """ + if not text.startswith("---"): + return {} + lines = text.splitlines() + if not lines or lines[0].strip() != "---": + return {} + data: dict = {} + pending_key: str | None = None # 正在收集块标量的 key + block_lines: list[str] = [] + block_indent: int | None = None + + def flush() -> None: + nonlocal pending_key, block_lines, block_indent + if pending_key is not None: + parts = [x.strip() for x in block_lines if x.strip()] + # 中文行之间不补空格，英文单词之间才补 + joiner = "" if _has_cjk("".join(parts)) else " " + data[pending_key] = joiner.join(parts) + pending_key, block_lines, block_indent = None, [], None + + for line in lines[1:]: + if line.strip() in ("---", "..."): + flush() + break + if pending_key is not None: + # 块标量的续行：空行或更深缩进的都算 + if line.strip() == "" or (line[:1] in (" ", "\t")): + block_lines.append(line) + continue + flush() + + m = re.match(r"^([A-Za-z0-9_-]+)\s*:\s*(.*)$", line) + if not m: + continue + key, val = m.group(1), m.group(2).strip() + if val in BLOCK_MARKERS or val.rstrip() in BLOCK_MARKERS: + pending_key = key + block_lines = [] + block_indent = None + continue + data[key] = val.strip("'\"") + + flush() + return data + + +# ---------------------------------------------------------------- 标准 skill 校验 +NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") + + +def find_skill_md(folder: Path, depth: int = 0) -> Path | None: + """在目录里找 SKILL.md（兼容压缩包自带一层父目录的情况）。""" + direct = folder / SKILL_MD + if direct.is_file(): + return direct + if depth >= 3: + return None + for child in sorted(folder.iterdir()): + if child.is_dir() and not child.name.startswith("."): + found = find_skill_md(child, depth + 1) + if found: + return found + return None + + +def validate_skill(folder: Path) -> tuple[bool, str, dict]: + """确认是不是标准 skill 格式：有 SKILL.md + YAML frontmatter + name。""" + md = find_skill_md(folder) + if not md: + return False, f"不是标准 skill 格式：找不到 {SKILL_MD}", {} + text = md.read_text(encoding="utf-8", errors="replace") + if not text.startswith("---"): + return False, f"{SKILL_MD} 缺少 YAML frontmatter（文件必须以 --- 开头）", {} + meta = _frontmatter(text) + if not meta: + return False, f"{SKILL_MD} 的 frontmatter 为空或无法解析", {} + name = meta.get("name", "") + if not name: + return False, f"{SKILL_MD} frontmatter 缺少必填的 name 字段", {} + if not NAME_RE.match(name): + return False, f"非法 skill 名称「{name}」（只允许字母/数字/点/下划线/连字符，且以字母或数字开头）", meta + if not meta.get("description"): + return False, f"{SKILL_MD} frontmatter 缺少 description 字段（大模型靠它判断何时调用）", meta + return True, "", meta + + +def decode_input(src: str, work: Path, b64: bool) -> Path | None: + """ + 把用户传进来的输入统一解码成一个本地路径： + - base64 字符串 -> 解出 zip 或 markdown + - 其余情况原样返回（文件/目录交给后续步骤处理） + """ + if not b64: + return None + raw = src.strip() + try: + decoded = base64.b64decode(raw, validate=True) + except (binascii.Error, ValueError) as exc: + raise SystemExit(f"base64 解码失败：{exc}") + if decoded[:2] == b"PK": # zipfile magic + target = work / "decoded.zip" + target.write_bytes(decoded) + return target + target = work / "decoded.md" + target.write_bytes(decoded) + return target + + +def stage_candidate(src: str, work: Path, b64: bool) -> Path: + """归一化输入 -> 返回一个包含 SKILL.md 的候选目录。""" + resolved = decode_input(src, work, b64) + path = Path(resolved) if resolved else Path(os.path.expandvars(os.path.expanduser(src))) + + if not path.exists(): + # 允许直接传一整段 base64（自动嗅探） + stripped = src.strip() + if _looks_like_b64(stripped): + sniff = work / "sniffed.bin" + try: + sniff.write_bytes(base64.b64decode(stripped, validate=True)) + path = sniff + except Exception as exc: + raise SystemExit(f"输入既不是存在的路径，也不是合法 base64：{exc}") + else: + raise SystemExit(f"路径不存在：{path}") + + if path.is_dir(): + return path + + # --- 到这里是单个文件：按魔数 / 后缀 / 内容三重嗅探 --- + if zipfile.is_zipfile(path): + return _unzip(path, work) + + suffix = path.suffix.lower() + if suffix in (".7z", ".rar", ".tar", ".gz", ".tgz"): + raise SystemExit(f"暂不支持该压缩格式：{path.suffix}（请先转成 .zip）") + + # 魔数兜底（无后缀或后缀写错的 zip） + try: + if path.read_bytes()[:2] == b"PK": + return _unzip(path, work) + except OSError: + pass + + text = path.read_text(encoding="utf-8", errors="replace") + compact = "".join(text.split()) + is_b64_text = ( + len(compact) >= 64 + and len(compact) % 4 == 0 + and re.fullmatch(r"[A-Za-z0-9+/=]+", compact) is not None + ) + + # base64：显式后缀，或者内容本身就是一段 base64 + if suffix in (".b64", ".base64", ".enc") or is_b64_text: + try: + decoded = base64.b64decode(compact, validate=True) + except Exception: + decoded = b"" + if decoded[:2] == b"PK": + zipped = work / "sniffed.zip" + zipped.write_bytes(decoded) + return _unzip(zipped, work) + if decoded: + md_file = work / "decoded.md" + md_file.write_bytes(decoded) + text = decoded.decode("utf-8", "replace") + + # markdown：frontmatter 开头，或者至少是个标题 + stripped = text.lstrip() + if stripped.startswith("---") or stripped.startswith("#"): + meta = _frontmatter(text) + name = meta.get("name") or path.stem + safe = re.sub(r"[^\w.-]+", "-", name).strip("-") or path.stem + folder = work / safe + folder.mkdir(parents=True, exist_ok=True) + (folder / SKILL_MD).write_text(text, encoding="utf-8") + return folder + + raise SystemExit(f"无法识别的输入：{path}（支持 .zip / 目录 / SKILL.md / base64 字符串）") + + +def _unzip(path: Path, work: Path) -> Path: + dest = work / ("unzipped_" + (path.stem or "archive")) + dest.mkdir(parents=True, exist_ok=True) + with zipfile.ZipFile(path) as zf: + _safe_extract(zf, dest) + return dest + + +def _looks_like_b64(s: str) -> bool: + if len(s) < 64 or len(s) % 4 != 0: + return False + if not re.fullmatch(r"[A-Za-z0-9+/=\s]+", s): + return False + return True + + +def _safe_extract(zf: zipfile.ZipFile, dest: Path) -> None: + """解压时防目录穿越（zip slip）。""" + for member in zf.infolist(): + target = (dest / member.filename).resolve() + if not str(target).startswith(str(dest.resolve())): + raise SystemExit(f"压缩包内含非法路径，已拒绝解压：{member.filename}") + zf.extractall(dest) + + +# ---------------------------------------------------------------- 子命令 +def cmd_list(args) -> int: + base, payload = connect_oc(args) + root = wsl_skill_root(args.distro) + + if base: + skills = extract_skills(payload) + source = f"OpenCode API（{base}）" + else: + skills = scan_skill_dir(root) + reason = payload if isinstance(payload, str) else "未知原因" + source = f"目录扫描（{root}）—— API 不可用：{reason}" + + print(f"skill 列表来源：{source}") + if not skills: + print("（没有找到任何 skill）") + return 1 + + width = max(len(s["name"]) for s in skills) + width = max(width, 4) + print(f"{'名称'.ljust(width)} 描述") + print("-" * (width + 40)) + for s in skills: + desc = s["description"].replace("\n", " ") + if len(desc) > 70: + desc = desc[:67] + "..." + print(f"{s['name'].ljust(width)} {desc}") + print(f"\n共 {len(skills)} 个 skill。WSL skill 目录：{root}") + return 0 + + +def cmd_push(args) -> int: + root = wsl_skill_root(args.distro) + root_path = Path(root) + if not root_path.is_dir(): + if args.dry_run: + print(f"（dry-run）目标目录尚不存在：{root}") + else: + raise SystemExit( + f"找不到 WSL 的 skill 目录：{root}\n" + f"请确认 WSL 发行版名称（当前 {args.distro}），可用 --distro 指定。" + ) + + with tempfile.TemporaryDirectory(prefix="ocskill_") as tmp: + work = Path(tmp) + candidate = stage_candidate(args.source, work, args.base64) + + ok, err, meta = validate_skill(candidate) + if not ok: + print(f"校验失败：{err}") + return 2 + + name = meta["name"] + desc = meta.get("description", "") + print(f"校验通过：标准 skill 格式") + print(f" 名称：{name}") + print(f" 描述：{desc[:100]}{'...' if len(desc) > 100 else ''}") + + # SKILL.md 可能藏在压缩包的一层父目录里，定位到真正含 SKILL.md 的那层 + md = find_skill_md(candidate) + real_dir = md.parent if md else candidate + + dest = root_path / name + if args.dry_run: + print(f"dry-run：不会写入 {dest}") + return 0 + + if dest.exists(): + if not args.force: + raise SystemExit( + f"目标已存在同名 skill：{dest}\n" + f"加 --force 覆盖（覆盖前会自动备份为 .bak-<时间戳>）。" + ) + backup = dest.with_name(dest.name + f".bak-{_stamp()}") + shutil.move(str(dest), str(backup)) + print(f"已备份旧版本 -> {backup}") + + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.copytree(real_dir, dest, dirs_exist_ok=True) + print(f"已写入：{dest}") + print("OC 会 watch skill 目录（约 100ms 去抖后自动重载），无需重启。") + return 0 + + +def cmd_pull(args) -> int: + root = wsl_skill_root(args.distro) + src = Path(root) / args.name + if not src.is_dir(): + raise SystemExit(f"WSL 端不存在该 skill：{src}") + + out_dir = Path(args.out or ".") + out_dir.mkdir(parents=True, exist_ok=True) + dest = out_dir / args.name + if dest.exists(): + if not args.force: + raise SystemExit(f"本地已存在：{dest}（用 --force 覆盖）") + shutil.rmtree(dest) + shutil.copytree(src, dest) + print(f"已取回：{dest}") + return 0 + + +def cmd_dirs(args) -> int: + print(f"WSL skill 目录（push 目标）：{wsl_skill_root(args.distro)}") + print(f"本机 skill 目录（pull 默认落到当前目录）：{WIN_LOCAL_ROOT}") + base, _ = connect_oc(args) + print(f"OpenCode API：{base or '未连接'}") + return 0 + + +def _stamp() -> str: + import datetime + + return datetime.datetime.now().strftime("%Y%m%d%H%M%S") + + +# ---------------------------------------------------------------- 入口 +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="ocskill", + description="打通本机与 WSL OpenCode 的 skill 互通（list / push / pull / dirs）", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=( + "示例：\n" + " python ocskill.py list\n" + " python ocskill.py push .\\my-skill.zip\n" + " python ocskill.py push .\\my-skill\\SKILL.md\n" + " python ocskill.py push --base64\n" + " python ocskill.py push .\\my-skill --dry-run\n" + " python ocskill.py pull java-expert --out .\\skills-backup\n" + ), + ) + p.add_argument("--distro", default=DEFAULT_DISTRO, help=f"WSL 发行版名（默认 {DEFAULT_DISTRO}）") + p.add_argument("--url", default=None, help="显式指定 OC 地址，如 http://localhost:4096") + p.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"OC 端口（默认 {DEFAULT_PORT}）") + p.add_argument("--timeout", type=float, default=4.0, help="HTTP 超时秒数") + p.add_argument("--password", default="", help="OC server.password（默认自动从 opencode.json 读）") + p.add_argument( + "--remote-config-dir", + default="/root/.config/opencode", + help="读取 server.password 用的配置目录（Windows 侧用 WSL 内路径时才需要改）", + ) + + sub = p.add_subparsers(dest="cmd") + + sp = sub.add_parser("list", help="连接 WSL 的 OC，返回 skill 列表") + sp.set_defaults(func=cmd_list) + + sp = sub.add_parser("push", help="推送 skill（zip / 目录 / SKILL.md / base64）到 WSL OC") + sp.add_argument("source", help="skill 文件路径、目录、zip 或 base64 字符串") + sp.add_argument("--base64", action="store_true", help="把 source 当作 base64 内容解码") + sp.add_argument("--force", action="store_true", help="覆盖同名 skill（先自动备份）") + sp.add_argument("--dry-run", action="store_true", help="只校验格式，不写入") + sp.set_defaults(func=cmd_push) + + sp = sub.add_parser("pull", help="把 WSL 里的 skill 拷回本机") + sp.add_argument("name", help="skill 名称") + sp.add_argument("--out", default=None, help="输出目录（默认当前目录）") + sp.add_argument("--force", action="store_true", help="覆盖本地同名目录") + sp.set_defaults(func=cmd_pull) + + sp = sub.add_parser("dirs", help="显示两端 skill 目录与 API 连接状态") + sp.set_defaults(func=cmd_dirs) + + return p + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if not getattr(args, "cmd", None): + parser.print_help() + return 0 + try: + return args.func(args) + except KeyboardInterrupt: + print("\n已取消。") + return 130 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/skill-list.bat b/skill-list.bat new file mode 100644 index 0000000..a6cecd8 --- /dev/null +++ b/skill-list.bat @@ -0,0 +1,16 @@ +@echo off +chcp 65001 >nul +title List WSL OpenCode skills +set "HERE=%~dp0" + +where python >nul 2>nul +if errorlevel 1 ( + echo [ERROR] Python not found in PATH. + echo Install Python 3 and tick "Add to PATH", then run this again. + pause + exit /b 1 +) + +python "%HERE%ocskill.py" list +echo. +pause diff --git a/skill-pull.bat b/skill-pull.bat new file mode 100644 index 0000000..dd544ec --- /dev/null +++ b/skill-pull.bat @@ -0,0 +1,23 @@ +@echo off +chcp 65001 >nul +title Pull a skill from WSL OpenCode +set "HERE=%~dp0" + +where python >nul 2>nul +if errorlevel 1 ( + echo [ERROR] Python not found in PATH. + pause + exit /b 1 +) + +set /p "NAME=Skill name: " +if "%NAME%"=="" ( + echo Nothing entered, cancelled. + pause + exit /b 0 +) + +set "OUT=%HERE%skills-backup" +python "%HERE%ocskill.py" pull "%NAME%" --out "%OUT%" +echo. +pause diff --git a/skill-push.bat b/skill-push.bat new file mode 100644 index 0000000..fabe9d7 --- /dev/null +++ b/skill-push.bat @@ -0,0 +1,25 @@ +@echo off +chcp 65001 >nul +title Push a skill to WSL OpenCode +set "HERE=%~dp0" + +where python >nul 2>nul +if errorlevel 1 ( + echo [ERROR] Python not found in PATH. + pause + exit /b 1 +) + +echo Accepts: a .zip file, a skill folder, a SKILL.md file, or base64 text. +echo Example: D:\skills\my-skill.zip +echo. +set /p "SRC=Path or content: " +if "%SRC%"=="" ( + echo Nothing entered, cancelled. + pause + exit /b 0 +) + +python "%HERE%ocskill.py" push "%SRC%" +echo. +pause