1912 lines
79 KiB
Python
1912 lines
79 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
ocskill —— 打通本机与 WSL OpenCode 的 skill 互通
|
||
|
||
做什么：
|
||
list 直接读 WSL OC 的 skill 目录，返回它的 skill 列表
|
||
（2026-09-24 起不再连 OC 的 API —— 直读目录更快、更稳、同源）
|
||
push 把一个新的 skill 文件（压缩包 / 目录 / 单个 SKILL.md / base64）
|
||
先解码并校验是不是标准 skill 格式，然后写进 WSL OC 的 skill 目录
|
||
（压缩包会额外做一次解压；OC 自带目录 watch，写完即热加载）
|
||
pull 反向操作：把 WSL 里的某个 skill 拷回本机 OC 的 skill 目录
|
||
delete 删除某个 skill（先挪进 .trash/ 备份，可恢复）
|
||
dirs 显示两端 skill 目录位置 + skill 数量
|
||
|
||
输入/输出约定（输出不由用户决定）：
|
||
输入 路径/技能名都过一遍 clean_input，粘贴带引号的完整路径也认
|
||
输出 push -> WSL OC skill 目录；pull -> 本机 OC skill 目录
|
||
覆盖 探查式：先看同名文件夹存不存在 —— 不存在就直接写（不问）；
|
||
确有同名才需要 --force，覆盖前统一挪进 .trash/<时间戳>/ 备份
|
||
|
||
零依赖：只用 Python 标准库（zipfile / shutil），无需 pip install。
|
||
"""
|
||
|
||
import argparse
|
||
import base64
|
||
import binascii
|
||
import glob
|
||
import json
|
||
import os
|
||
import queue
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
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")
|
||
SKILL_MD = "SKILL.md"
|
||
|
||
# 分页列表：每页 15 个（宝宝级菜单一页看得完，不用滚屏）
|
||
PAGE_SIZE = 15
|
||
# agent 终端 skill 目录表（本文件同目录的 agents.md，两列：名字 | 目录）
|
||
AGENTS_MD = Path(__file__).resolve().parent / "agents.md"
|
||
# 内置默认终端：留空回车 = WSL 里的 OC（不在 agents.md 表内）
|
||
DEFAULT_AGENT = {"key": "wsl", "display": "wsl", "dir": ""} # dir 运行时按 distro 解析
|
||
|
||
# 标准 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 连接
|
||
# ---- WSL 安全执行（与 mcpctl.py 同一套，逐字重复保持三端一致）----
|
||
# 1) wsl.exe 绝不留管道：stdout/stderr 落临时文件。原 capture_output 版在
|
||
# WSL VM 被清理/终止后触发冷启动超时，client 被 kill 但 wslrelay 等后代
|
||
# 进程继承管道句柄，communicate() 等 EOF 永久阻塞 —— 表现为
|
||
# “进入 skill 目录后再也没反应，必须 ctrl+C”。
|
||
# 2) 重活之前先查 --running：VM 没跑绝不无提示冷启动。
|
||
def _wsl_safe_run(argv: list, timeout: float = 10.0) -> tuple[int, bytes]:
|
||
if os.name != "nt":
|
||
return -1, b""
|
||
import tempfile as _tf
|
||
|
||
with _tf.TemporaryFile() as fo, _tf.TemporaryFile() as fe:
|
||
try:
|
||
cp = subprocess.run(
|
||
argv,
|
||
stdin=subprocess.DEVNULL,
|
||
stdout=fo,
|
||
stderr=fe,
|
||
timeout=timeout,
|
||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||
)
|
||
fo.seek(0)
|
||
fe.seek(0)
|
||
return cp.returncode, fo.read() + fe.read()
|
||
except subprocess.TimeoutExpired:
|
||
return -1, b""
|
||
except Exception as exc:
|
||
return -2, str(exc).encode("utf-8", "replace")
|
||
|
||
|
||
def _wsl_running(distro: str) -> bool:
|
||
if os.name != "nt":
|
||
return True
|
||
code, out = _wsl_safe_run(["wsl.exe", "-l", "-q", "--running"], timeout=8)
|
||
if code != 0 or not out:
|
||
return False
|
||
# wsl.exe 列表输出是 UTF-16LE（常无 BOM）
|
||
text = (
|
||
out.decode("utf-16le", "ignore")
|
||
if out.count(b"\x00") > len(out) // 4
|
||
else out.decode("utf-8", "ignore")
|
||
)
|
||
return distro in text
|
||
|
||
|
||
# ---------------------------------------------------------------- 读操作防卡死
|
||
# WSL UNC 路径（\\wsl.localhost\...）在 VM 半死 / 9p 无响应时，Path.exists() /
|
||
# iterdir 会在系统调用层无限挂起，且 Ctrl+C 打不断（Python 信号要等阻塞的
|
||
# 系统调用返回才处理）。只读探查统一放进 daemon 工作线程限时等待：
|
||
# 超时打印原因并回菜单；主线程等 queue.get(timeout)（Python 层），Ctrl+C 可打断。
|
||
# （与 mcpctl._guard 逐字一致，保持三端同源。）
|
||
def _guard(fn, timeout: float):
|
||
"""在工作线程里跑 fn，主线程最多等 timeout 秒。
|
||
返回 (True, 结果) 或 (False, 错误描述)；错误描述为 'TIMEOUT' 表示超时。"""
|
||
q: "queue.Queue" = queue.Queue(maxsize=1)
|
||
|
||
def _run() -> None:
|
||
try:
|
||
q.put((True, fn()))
|
||
except Exception as exc: # noqa: BLE001
|
||
q.put((False, f"{type(exc).__name__}: {exc}"))
|
||
|
||
threading.Thread(target=_run, daemon=True).start()
|
||
try:
|
||
return q.get(timeout=timeout)
|
||
except queue.Empty:
|
||
return False, "TIMEOUT"
|
||
|
||
|
||
def _run_with_beats(fn, interval: float = 5.0):
|
||
"""长操作期间每 interval 秒打一个进度点，避免用户把「等待」当成「卡死」。"""
|
||
stop = threading.Event()
|
||
|
||
def _beat() -> None:
|
||
while not stop.wait(interval):
|
||
print(" .", end="", flush=True)
|
||
|
||
threading.Thread(target=_beat, daemon=True).start()
|
||
try:
|
||
return fn()
|
||
finally:
|
||
stop.set()
|
||
print(flush=True)
|
||
|
||
|
||
def _wsl_ensure_running(distro: str, purpose: str) -> str | None:
|
||
"""UNC 写之前确保 VM 在跑；返回 None=可继续，字符串=错误信息。"""
|
||
if os.name != "nt" or _wsl_running(distro):
|
||
return None
|
||
print(f"WSL（{distro}）未运行（{purpose}需要访问它），正在唤醒"
|
||
f"（可能需要十几秒，期间每 5 秒打一个点）...", flush=True)
|
||
_run_with_beats(
|
||
lambda: _wsl_safe_run(["wsl.exe", "-d", distro, "--", "true"], timeout=45)
|
||
)
|
||
if _wsl_running(distro):
|
||
print("WSL 已就绪。", flush=True)
|
||
return None
|
||
return (
|
||
f"WSL（{distro}）唤醒失败，无法访问其 skill 目录。"
|
||
f"请手动执行 wsl.exe -d {distro} 启动后重试。"
|
||
)
|
||
|
||
|
||
def scan_skill_dir(root: str) -> list[dict]:
|
||
"""直接扫目录列出 skill —— 列表/push 同名检查/菜单展示共用的 ground truth。"""
|
||
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
|
||
|
||
|
||
# ---------------------------------------------------------------- agent 终端表 + 分页（宝宝级拉取）
|
||
_HEADER_K1 = ("名字", "名称", "简称", "name", "key")
|
||
_HEADER_K2 = ("目录", "路径", "dir", "path", "folder")
|
||
_CN_NUM = {"一": 1, "二": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7, "八": 8,
|
||
"九": 9, "十": 10, "十一": 11, "十二": 12, "十三": 13, "十四": 14,
|
||
"十五": 15, "十六": 16, "十七": 17, "十八": 18, "十九": 19, "二十": 20}
|
||
_EN_NUM = {"one": 1, "two": 2, "three": 3, "four": 4, "five": 5, "six": 6,
|
||
"seven": 7, "eight": 8, "nine": 9, "ten": 10, "eleven": 11,
|
||
"twelve": 12, "thirteen": 13, "fourteen": 14, "fifteen": 15,
|
||
"sixteen": 16, "seventeen": 17, "eighteen": 18, "nineteen": 19,
|
||
"twenty": 20}
|
||
|
||
|
||
def _split_bracket(cell: str) -> tuple[str, str]:
|
||
"""'wb(workbuddy)' -> ('wb', 'workbuddy')；无括号 -> (原串, 原串)。
|
||
括号形式不限（半角/全角），可出现在任意位置：
|
||
第一对括号前 = 简称（CLI 输入用），括号内 = 全称（CLI 打印用）。
|
||
"""
|
||
s = (cell or "").strip()
|
||
m = re.search(r"[(（]([^)）]*)[)）]", s)
|
||
if not m:
|
||
return s, s
|
||
key = s[: m.start()].strip() or m.group(1).strip() or s
|
||
disp = m.group(1).strip() or key
|
||
return key, disp
|
||
|
||
|
||
def _strip_brackets(text: str) -> str:
|
||
"""输入里的括号一律忽略：'wb(workbuddy)' -> 'wb'；'（wb）' -> 'wb'。
|
||
先删成对括号连同内容（剥掉的是括号里的注释性全称）；
|
||
若剥空了（内容全在括号里，如 '（wb）'），回退为只删括号字符保留内容；
|
||
最后再清掉任何残留的单侧括号字符。
|
||
"""
|
||
s = (text or "").strip()
|
||
out = re.sub(r"[(（][^)）]*[)）]", "", s).strip()
|
||
if not out:
|
||
out = s
|
||
return re.sub(r"[()（）]", "", out).strip()
|
||
|
||
|
||
def load_agent_table(path=None) -> list[dict]:
|
||
"""读 agents.md 终端表（宽松解析，不校验 markdown 完整性）。
|
||
|
||
逐行取「以 | 开头」的行 -> 按 '|' 拆 -> 前两个非空字段 = 名字 | 目录。
|
||
表格没闭合、行首多空格、少分隔符都照常认；新增一行 = 新增一个终端。
|
||
行首必须是 | —— 说明段落里提到的 | 不会混进表。
|
||
跳过：非 | 开头行 / 表头行 / 分隔行 / 缺目录行。文件不存在返回 []。
|
||
返回 [{"key": 简称, "display": 全称, "dir": 目录}, ...]。
|
||
"""
|
||
p = Path(path) if path else AGENTS_MD
|
||
try:
|
||
lines = p.read_text(encoding="utf-8", errors="replace").splitlines()
|
||
except OSError:
|
||
return []
|
||
out: list[dict] = []
|
||
for raw in lines:
|
||
if not raw.strip().startswith("|"):
|
||
continue
|
||
cells = [c.strip() for c in raw.split("|")]
|
||
cells = [c for c in cells if c]
|
||
if len(cells) < 2:
|
||
continue
|
||
k1, k2 = cells[0], cells[1]
|
||
kl, k2l = k1.lower(), k2.lower()
|
||
# 表头行：第一列命中名字类 且 第二列命中路径类
|
||
if any(w in kl for w in _HEADER_K1) and any(w in k2l for w in _HEADER_K2):
|
||
continue
|
||
# 分隔行：字段全是 - : 空格
|
||
if all(set(c) <= set("-: ") for c in (k1, k2)):
|
||
continue
|
||
key, disp = _split_bracket(k1)
|
||
if not key or not k2:
|
||
continue
|
||
if any(t["key"].lower() == key.lower() for t in out):
|
||
continue # 同简称后写不覆盖先写
|
||
out.append({"key": key, "display": disp, "dir": k2})
|
||
return out
|
||
|
||
|
||
def resolve_agent(spec, distro: str = DEFAULT_DISTRO) -> dict:
|
||
"""解析来源/目标终端（CLI --from/--to 与菜单层共用）。判定顺序：
|
||
① 空输入 -> 内置默认 wsl；
|
||
② 是存在的目录 -> 整串当目录用（kind=dir，可直接贴一整个 skill 目录）；
|
||
③ 简称 -> agents.md 前缀匹配（精确 key -> key 前缀 -> 全称前缀，大小写不敏感）；
|
||
④ 都不中 -> 报错并列出可用简称。
|
||
返回 {"ok", "kind": "default|dir|agent", "label", "dir", "table" / "err"}。
|
||
"""
|
||
spec = clean_input(spec or "")
|
||
table = load_agent_table()
|
||
if not spec:
|
||
return {"ok": True, "kind": "default", "label": "wsl",
|
||
"dir": wsl_skill_root(distro), "table": table}
|
||
try:
|
||
isdir = Path(spec).is_dir()
|
||
except OSError:
|
||
isdir = False
|
||
if isdir:
|
||
return {"ok": True, "kind": "dir", "label": spec, "dir": spec, "table": table}
|
||
q = _strip_brackets(spec).lower()
|
||
if q:
|
||
for mode in ("exact", "key", "disp"):
|
||
for t in table:
|
||
k, d = t["key"].lower(), t["display"].lower()
|
||
hit = (k == q) if mode == "exact" else (
|
||
k.startswith(q) if mode == "key" else d.startswith(q))
|
||
if hit:
|
||
return {"ok": True, "kind": "agent", "label": t["key"],
|
||
"dir": t["dir"], "table": table}
|
||
keys = "、".join(f"{t['key']}({t['display']})" for t in table) or "（agents.md 为空）"
|
||
return {"ok": False, "err": f"无法定位终端「{spec}」。可用简称：wsl（默认）、{keys}；"
|
||
f"也可直接贴完整 skill 目录路径。", "table": table}
|
||
|
||
|
||
def list_skill_names(root) -> list[str]:
|
||
"""直接读 skill 文件夹列名字 —— 不启动 OC 服务、不发任何请求。
|
||
只认过了 NAME_RE 的目录名，跳过点前缀（.trash 等 OC 也不加载），按名排序。
|
||
目录不存在/不可读返回 []。
|
||
"""
|
||
try:
|
||
p = Path(root)
|
||
if not p.is_dir():
|
||
return []
|
||
return sorted(
|
||
c.name for c in p.iterdir()
|
||
if c.is_dir() and not c.name.startswith(".") and NAME_RE.match(c.name)
|
||
)
|
||
except OSError:
|
||
return []
|
||
|
||
|
||
def parse_sel(text) -> tuple[int, bool] | None:
|
||
"""编号选择解析（确认协议）-> (全局序号, 是否跳过二次确认)；不是编号输入返回 None。
|
||
|
||
协议（推进去 / 取回来两端通用）：
|
||
'1' / '1.' -> (1, False) 数字后带点号 = 确认选择它，进二次确认
|
||
（打印 skill 名字 + 简介，confirm 执行 / cancel 回退）
|
||
'1 -confirm' -> (1, True) 跳过二次确认，直接执行（'-confirm' 大小写不敏感，
|
||
也认 '1.-confirm' / '1-confirm' / '--confirm'）
|
||
名字 / 翻页 / 路径 -> None（交给调用方的其它分支）
|
||
纯 '0' 也命中 (0, False) —— 调用方若把 0 当退出，必须在本函数之前先拦。
|
||
序号范围由调用方按列表长度校验。
|
||
"""
|
||
s = (text or "").strip()
|
||
if not s:
|
||
return None
|
||
m = re.fullmatch(r"(\d+)\s*\.?\s*(?:-{1,2}confirm)?", s, re.IGNORECASE)
|
||
if not m:
|
||
return None
|
||
skip = re.search(r"-{1,2}confirm", s, re.IGNORECASE) is not None
|
||
return int(m.group(1)), skip
|
||
|
||
|
||
def parse_page_cmd(text, page: int, total: int) -> int | None:
|
||
"""翻页命令 -> 目标页码；不是翻页命令返回 None。
|
||
|
||
中文：下一页 / 上一页 / 第N页 / N页（N 可为中文数词，如 第三页）；
|
||
英文：next / last / prev / N page（three page / 3 page）。
|
||
纯数字不算翻页 —— 留给「按全局序号选择 skill」。
|
||
|
||
越界语义（2026-09-24 起）：
|
||
- 相对翻页（下一页/上一页/next/last）：钳制在 [1, total]，边界静默停住；
|
||
- 绝对页码（第N页 / N页 / 中文数词 / N page）：返回【原始】页码，不钳制 ——
|
||
越界交给调用方报错（「共 X 页，没有第 N 页」）并拒绝跳转，
|
||
不再像旧行为那样把第 999 页悄悄钳到末页造成「翻页没上限」的假象。
|
||
"""
|
||
s = re.sub(r"\s+", "", (text or "")).lower()
|
||
if not s:
|
||
return None
|
||
total = max(1, total)
|
||
clamp = lambda n: max(1, min(total, n)) # noqa: E731
|
||
if s in ("下一页", "下页", "next", "nextpage"):
|
||
return clamp(page + 1)
|
||
if s in ("上一页", "上页", "last", "prev", "previous", "back",
|
||
"lastpage", "prevpage", "previouspage"):
|
||
return clamp(page - 1)
|
||
m = re.fullmatch(r"(第(\d+)页?|(\d+)页)", s) # 第3 / 第3页 / 3页；裸数字不命中
|
||
if m:
|
||
return int(m.group(2) or m.group(3))
|
||
m = re.fullmatch(r"第?([一二三四五六七八九十]+)页", s) # 第三页 / 三页
|
||
if m and m.group(1) in _CN_NUM:
|
||
return _CN_NUM[m.group(1)]
|
||
m = re.fullmatch(r"(\d+|[a-z]+)page", s) # 3 page / three page
|
||
if m:
|
||
tok = m.group(1)
|
||
if tok.isdigit():
|
||
return int(tok)
|
||
return _EN_NUM.get(tok) if tok in _EN_NUM else None
|
||
return None
|
||
|
||
|
||
def render_page(names: list[str], page: int, root_label: str = "") -> tuple[int, int]:
|
||
"""打印一页 skill 列表：每页 PAGE_SIZE 个，序号跨页全局连续。
|
||
返回 (实际页码, 总页数)。
|
||
root_label 同时用作目录根：找不到 SKILL.md 的条目标 ⚠（与推送同名检查同源
|
||
find_skill_md，选中前就能看出空壳/断链，2026-09-24 新增）。
|
||
"""
|
||
total_items = len(names)
|
||
pages = max(1, -(-total_items // PAGE_SIZE))
|
||
page = max(1, min(pages, page))
|
||
start = (page - 1) * PAGE_SIZE
|
||
chunk = names[start:start + PAGE_SIZE]
|
||
print(f" skill 列表（共 {total_items} 个，第 {page}/{pages} 页，每页 {PAGE_SIZE} 个）")
|
||
if root_label:
|
||
print(f" 目录：{root_label}")
|
||
if not chunk:
|
||
print(" （这个目录下还没有 skill）")
|
||
else:
|
||
width = len(str(start + len(chunk)))
|
||
root_path = Path(root_label) if root_label else None
|
||
warned = 0
|
||
for i, n in enumerate(chunk, start + 1):
|
||
mark = ""
|
||
if root_path is not None:
|
||
try:
|
||
# 与 probe_push / skill_brief 同一个 find_skill_md（含嵌套布局）
|
||
if find_skill_md(root_path / n) is None:
|
||
mark = " ⚠缺 SKILL.md"
|
||
warned += 1
|
||
except OSError:
|
||
pass
|
||
print(f" {str(i).rjust(width)}. {n}{mark}")
|
||
if warned:
|
||
print(f" （⚠ = 找不到 SKILL.md：空壳目录或链接断了，选中确认时会打印软链接诊断）")
|
||
return page, pages
|
||
|
||
|
||
# ---------------------------------------------------------------- 输入净化与探查
|
||
_QUOTE_PAIRS = (('"', '"'), ("'", "'"), ("\u201c", "\u201d"), ("\u300c", "\u300d"))
|
||
|
||
def clean_input(raw: str) -> str:
|
||
"""剥掉用户粘贴时带上的引号并去首尾空白。
|
||
|
||
从终端/对话里复制完整 Windows 路径时，两端常带着引号
|
||
（如 "C:\\Users\\...\\skills\\opencode-plugin-dev"）。
|
||
不剥掉就会把引号当路径的一部分，直接报「路径不存在」/
|
||
「WSL 端不存在该 skill」—— 报错里能看到引号还挂在路径上。
|
||
|
||
只剥「成对」的引号（"..." / '...' / “...” / 「...」）；
|
||
单侧引号不动 —— 半剥只会拼出一个更怪、照样不存在的路径，
|
||
不如原样报错，用户一眼能看出是引号没复制全。
|
||
"""
|
||
s = (raw or "").strip()
|
||
for a, b in _QUOTE_PAIRS:
|
||
if len(s) >= 2 and s.startswith(a) and s.endswith(b):
|
||
return s[1:-1].strip()
|
||
return s
|
||
|
||
def probe_push(src: str, distro: str = DEFAULT_DISTRO, b64: bool = False,
|
||
root: str | None = None) -> dict:
|
||
"""推送前探查（只读，不写盘）：校验 skill 规范 + 判断目标是否已有同名目录。
|
||
|
||
root 显式指定目标 skill 目录（--to 解析出的终端）；缺省仍是 WSL OC 的 skill 目录。
|
||
菜单层据此决定要不要问用户覆盖：
|
||
exists=False -> 不存在同名，直接放（不该问覆盖，问了也是脱裤子放屁）
|
||
exists=True -> 确有同名文件夹，才有资格问一次「是否覆盖」
|
||
返回 {"ok", "err"/"name"/"dest"/"exists"}。
|
||
"""
|
||
src = clean_input(src)
|
||
with tempfile.TemporaryDirectory(prefix="ocskill_probe_") as tmp:
|
||
try:
|
||
candidate = stage_candidate(src, Path(tmp), b64)
|
||
except SystemExit as exc:
|
||
return {"ok": False, "err": str(exc.code or exc)}
|
||
except Exception as exc: # noqa: BLE001
|
||
return {"ok": False, "err": f"输入解析失败：{exc}"}
|
||
ok, err, meta = validate_skill(candidate)
|
||
if not ok:
|
||
res = {"ok": False, "err": err}
|
||
# 找不到 SKILL.md -> 附一层软链接诊断（oc/wb 里有些 skill 是链接形式）
|
||
if SKILL_MD in err:
|
||
try:
|
||
res["diag"] = diag_skill_md(candidate)
|
||
except Exception as diag_exc: # 诊断只增信息，绝不覆盖主错误
|
||
res["diag"] = [f"诊断失败：{diag_exc}"]
|
||
return res
|
||
name = meta.get("name", "")
|
||
dest = Path(root) if root else Path(wsl_skill_root(distro))
|
||
dest = dest / name
|
||
try:
|
||
exists = dest.is_dir()
|
||
except OSError:
|
||
exists = False
|
||
return {"ok": True, "name": name, "dest": str(dest), "exists": exists,
|
||
"description": meta.get("description", "")}
|
||
|
||
def skill_brief(name: str, root) -> dict:
|
||
"""读一个 skill 的名字 + 简介 + 路径（只读，供编号选择的二次确认打印）。
|
||
|
||
简介取 SKILL.md frontmatter 的 description（用现成的 find_skill_md + _frontmatter，
|
||
兼容 SKILL.md 在子层 / 块标量描述）；目录不存在或名字非法返回 ok=False + err。
|
||
"""
|
||
name = clean_input(name or "")
|
||
if not name:
|
||
return {"ok": False, "err": "技能名为空"}
|
||
if not NAME_RE.match(name):
|
||
return {"ok": False, "err": f"非法 skill 名称「{name}」"}
|
||
folder = Path(root) / name
|
||
try:
|
||
exists = folder.is_dir()
|
||
except OSError:
|
||
exists = False
|
||
if not exists:
|
||
return {"ok": False, "err": f"不存在该 skill：{folder}"}
|
||
desc = ""
|
||
md = find_skill_md(folder)
|
||
if md is not None:
|
||
try:
|
||
desc = _frontmatter(md.read_text(encoding="utf-8", errors="replace")).get("description", "")
|
||
except OSError:
|
||
desc = ""
|
||
return {"ok": True, "name": name, "description": desc, "path": str(folder),
|
||
"has_md": True, "diag": []}
|
||
# 缺 SKILL.md -> 附软链接诊断（找出真实文件所在位置）
|
||
try:
|
||
diag = diag_skill_md(folder)
|
||
except Exception as diag_exc: # 诊断只增信息
|
||
diag = [f"诊断失败：{diag_exc}"]
|
||
return {"ok": True, "name": name, "description": desc, "path": str(folder),
|
||
"has_md": False, "diag": diag}
|
||
|
||
|
||
def probe_pull(name: str, distro: str = DEFAULT_DISTRO, root: str | None = None) -> dict:
|
||
"""拉取前探查（只读）：来源端是否存在该 skill + 本机是否已有同名目录。
|
||
|
||
root 显式指定来源 skill 目录（--from 解析出的终端）；缺省仍是 WSL OC 的 skill 目录。
|
||
"""
|
||
name = clean_input(name)
|
||
if not name:
|
||
return {"ok": False, "err": "技能名为空"}
|
||
if not NAME_RE.match(name):
|
||
return {"ok": False, "err": f"非法 skill 名称「{name}」（只允许字母/数字/点/下划线/连字符，且以字母或数字开头）"}
|
||
src = Path(root) if root else Path(wsl_skill_root(distro))
|
||
src = src / name
|
||
dest = Path(WIN_LOCAL_ROOT) / name
|
||
try:
|
||
src_ok = src.is_dir()
|
||
except OSError:
|
||
src_ok = False
|
||
try:
|
||
local_exists = dest.is_dir()
|
||
except OSError:
|
||
local_exists = False
|
||
if not src_ok:
|
||
return {"ok": False, "err": f"来源端不存在该 skill：{src}"}
|
||
return {"ok": True, "name": name, "dest": str(dest), "exists": local_exists,
|
||
"src": str(src)}
|
||
|
||
def _side_label(base, distro: str = DEFAULT_DISTRO) -> str:
|
||
"""按目录归类侧标签（本机 / WSL / 所选），供删除探查打印用。"""
|
||
b = os.path.normcase(os.path.abspath(str(base)))
|
||
if b == os.path.normcase(os.path.abspath(WIN_LOCAL_ROOT)):
|
||
return "本机"
|
||
if b == os.path.normcase(os.path.abspath(str(wsl_skill_root(distro)))):
|
||
return "WSL"
|
||
return "所选"
|
||
|
||
|
||
def probe_delete(name: str, distro: str = DEFAULT_DISTRO, local: bool = False,
|
||
root: str | None = None) -> dict:
|
||
"""删除前探查（只读）：确认目标存在并给出将删除的路径。
|
||
|
||
root 显式指定 skill 目录（菜单选终端后传入的 --dir）；缺省按 local 选本机/WSL。
|
||
"""
|
||
name = clean_input(name)
|
||
if not name:
|
||
return {"ok": False, "err": "技能名为空"}
|
||
if not NAME_RE.match(name):
|
||
return {"ok": False, "err": f"非法 skill 名称「{name}」（只允许字母/数字/点/下划线/连字符，且以字母或数字开头）"}
|
||
if root:
|
||
base = Path(clean_input(str(root)))
|
||
side = _side_label(base, distro)
|
||
else:
|
||
base = Path(WIN_LOCAL_ROOT) if local else Path(wsl_skill_root(distro))
|
||
side = "本机" if local else "WSL"
|
||
dest = base / name
|
||
try:
|
||
ok = dest.is_dir()
|
||
except OSError:
|
||
ok = False
|
||
if not ok:
|
||
return {"ok": False, "err": f"{side}端不存在该 skill：{dest}"}
|
||
return {"ok": True, "name": name, "dest": str(dest), "side": side}
|
||
|
||
# ---------------------------------------------------------------- 标准 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
|
||
try:
|
||
children = sorted(folder.iterdir())
|
||
except OSError:
|
||
# 断链 / 怪异路径（如 symlink 目标解析失败）：当「没找到」处理，
|
||
# 不让诊断、列表页、二次确认被 OSError 炸穿（2026-09-24 单测暴露）
|
||
return None
|
||
for child in children:
|
||
if child.is_dir() and not child.name.startswith("."):
|
||
found = find_skill_md(child, depth + 1)
|
||
if found:
|
||
return found
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------- skill 注册表（skill.json，2026-09-25）
|
||
# 用户要求的推拉兜底流程：
|
||
# 遇软链接/空壳 -> 不一上来就全盘忙扫：① 先查内存注册表（启动时已把 skill.json
|
||
# 读进内存）-> ② 再快扫内嵌常见终端的 skill 位置（oc / wb / cc-switch / claude …）
|
||
# -> ③ 都没命中才后台扫盘（不阻塞当前推拉），扫完把结果落库 skill.json。
|
||
# skill.json 按「名字 -> 位置」的一堆键值存储；推拉永远优先用户选定的目录，
|
||
# 注册表只在空壳/链接兜底（真身解析）。
|
||
REGISTRY_PATH = Path(__file__).resolve().parent / "skill.json"
|
||
_REG: dict | None = None
|
||
_REG_LOCK = threading.Lock()
|
||
_BG_SCAN_RUNNING = False
|
||
|
||
# 内嵌常见终端 skill 位置（tier1 快扫；相对 home / cwd 解析，路径不存在自动跳过）
|
||
_BUILTIN_HOME_ROOTS = (
|
||
".config/opencode/skills", # opencode（oc）
|
||
".workbuddy/skills", # workbuddy（wb）
|
||
".cc-switch/skills", # cc-switch
|
||
".claude/skills", # claude code
|
||
)
|
||
_BUILTIN_CWD_ROOTS = (
|
||
".workbuddy/skills", # 项目内 wb skill
|
||
".opencode/skills", # 项目内 oc skill
|
||
".claude/skills", # 项目内 claude skill
|
||
)
|
||
|
||
|
||
def builtin_skill_roots(distro: str = DEFAULT_DISTRO) -> list[str]:
|
||
"""内嵌常见终端的 skill 根（tier1 快扫用）—— 先扫这几个，不一上来全盘忙扫。
|
||
|
||
组成：① 内嵌 home/cwd 常见终端位置；② agents.md 终端表里配置的目录
|
||
（用户新增终端自动纳入）；③ 环境变量 OC_SKILL_ALT_ROOTS 追加；
|
||
④ WSL 侧两根（仅 VM 已在跑才带 —— 快扫绝不唤醒 WSL，避免冷启动拖慢推拉）。
|
||
全程只拼路径 + 一次 wsl 列表查询；目录不存在由扫描端跳过。
|
||
"""
|
||
home = Path.home()
|
||
roots: list[Path] = [home / p for p in _BUILTIN_HOME_ROOTS]
|
||
try:
|
||
roots += [Path.cwd() / p for p in _BUILTIN_CWD_ROOTS]
|
||
except OSError:
|
||
pass
|
||
for t in load_agent_table():
|
||
if t.get("dir"):
|
||
roots.append(Path(t["dir"]))
|
||
extra = os.environ.get("OC_SKILL_ALT_ROOTS", "")
|
||
roots += [Path(p) for p in extra.split(os.pathsep) if p.strip()]
|
||
if os.name == "nt":
|
||
try:
|
||
wsl_on = _wsl_running(distro)
|
||
except Exception:
|
||
wsl_on = False
|
||
if wsl_on:
|
||
unc = Path("\\\\wsl.localhost\\%s" % distro)
|
||
roots += [unc / "root" / ".config" / "opencode" / "skills",
|
||
unc / "root" / ".workbuddy" / "skills"]
|
||
out: list[str] = []
|
||
seen: set[str] = set()
|
||
for r in roots:
|
||
key = os.path.normcase(os.path.abspath(str(r)))
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(str(r))
|
||
return out
|
||
|
||
|
||
def deep_scan_roots() -> list[str]:
|
||
"""后台扫盘的扩展根（tier2）：常见终端之外的兜底位置。
|
||
|
||
只扫浅层、只认名为 skills 的目录（不递归、不全盘）：
|
||
home / home/.config / home/.local/share / cwd 下一级子目录里的 skills/。
|
||
—— 市面终端（cursor / cline / roo …）把 skill 放 `~/<终端>/skills` 的，
|
||
这层就能兜住；再深的全盘扫既慢又容易扫进无关目录，不做。
|
||
"""
|
||
out: list[str] = []
|
||
try:
|
||
bases = [Path.home(), Path.home() / ".config",
|
||
Path.home() / ".local" / "share", Path.cwd()]
|
||
except OSError:
|
||
return out
|
||
for base in bases:
|
||
try:
|
||
for child in sorted(base.iterdir()):
|
||
if not child.is_dir() or child.name.startswith("."):
|
||
continue
|
||
s = child / "skills"
|
||
try:
|
||
if s.is_dir():
|
||
out.append(str(s))
|
||
except OSError:
|
||
continue
|
||
except OSError:
|
||
continue
|
||
dedup: list[str] = []
|
||
seen: set[str] = set()
|
||
for r in out:
|
||
key = os.path.normcase(os.path.abspath(r))
|
||
if key not in seen:
|
||
seen.add(key)
|
||
dedup.append(r)
|
||
return dedup
|
||
|
||
|
||
def load_registry() -> dict:
|
||
"""启动时把 skill.json 读到内存（只读一次，之后全走内存缓存 _REG）。"""
|
||
global _REG
|
||
with _REG_LOCK:
|
||
if _REG is None:
|
||
reg: dict = {"version": 1, "updated": "", "skills": {}}
|
||
try:
|
||
if REGISTRY_PATH.is_file():
|
||
data = json.loads(REGISTRY_PATH.read_text(encoding="utf-8"))
|
||
if isinstance(data, dict) and isinstance(data.get("skills"), dict):
|
||
reg = data
|
||
reg.setdefault("version", 1)
|
||
reg.setdefault("updated", "")
|
||
except (OSError, ValueError):
|
||
pass # 文件不存在 / 损坏：从空注册表起步，不炸推拉
|
||
_REG = reg
|
||
return _REG
|
||
|
||
|
||
def save_registry() -> None:
|
||
"""把内存注册表落库 skill.json（临时文件 + os.replace 原子写，避免半截文件）。"""
|
||
with _REG_LOCK:
|
||
reg = _REG or {"version": 1, "updated": "", "skills": {}}
|
||
try:
|
||
import datetime as _dt
|
||
reg["updated"] = _dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
tmp = REGISTRY_PATH.with_suffix(".json.tmp")
|
||
tmp.write_text(json.dumps(reg, ensure_ascii=False, indent=2),
|
||
encoding="utf-8")
|
||
os.replace(str(tmp), str(REGISTRY_PATH))
|
||
except OSError as exc:
|
||
print(" （skill.json 落库失败：%s）" % exc)
|
||
|
||
|
||
def remember_skills(found: dict) -> None:
|
||
"""把扫到的 {名字: [位置]} 合并进注册表并落库（只增不删、位置去重）。
|
||
|
||
真身被删/移动的旧位置不主动清（合并语义）—— 查询端会做存在性校验，
|
||
过期条目只是查不到，不会误导。
|
||
"""
|
||
if not found:
|
||
return
|
||
reg = load_registry()
|
||
skills = reg.setdefault("skills", {})
|
||
changed = False
|
||
for name, paths in found.items():
|
||
bucket = skills.setdefault(name, [])
|
||
for p in paths:
|
||
if p not in bucket:
|
||
bucket.append(p)
|
||
changed = True
|
||
if changed:
|
||
save_registry()
|
||
|
||
|
||
def registry_locations(name: str) -> list[str]:
|
||
"""内存注册表里该名字的位置列表（存在性 / SKILL.md 校验由调用方做）。"""
|
||
reg = load_registry()
|
||
return list(reg.get("skills", {}).get(name) or [])
|
||
|
||
|
||
def scan_roots_for_name(name: str, roots: list[str] | None = None,
|
||
distro: str = DEFAULT_DISTRO) -> list[str]:
|
||
"""tier1 快扫：在给定根里按名字找真身目录（<根>/<名> 含 SKILL.md 才算）。
|
||
|
||
roots 缺省 = builtin_skill_roots()（内嵌常见终端位置，秒级、不唤醒 WSL）。
|
||
只读、任何失败跳过该根；返回真身目录字符串列表（去重）。
|
||
"""
|
||
out: list[str] = []
|
||
seen: set[str] = set()
|
||
for r in (roots if roots is not None else builtin_skill_roots(distro)):
|
||
cand = Path(r) / name
|
||
try:
|
||
if find_skill_md(cand) is None:
|
||
continue
|
||
except OSError:
|
||
continue
|
||
key = os.path.normcase(os.path.abspath(str(cand)))
|
||
if key not in seen:
|
||
seen.add(key)
|
||
out.append(str(cand))
|
||
return out
|
||
|
||
|
||
def _scan_root_skills(root: str) -> dict:
|
||
"""扫一个根下所有合格 skill -> {名字: [目录]}（有 SKILL.md 才登记）。"""
|
||
found: dict = {}
|
||
base = Path(root)
|
||
try:
|
||
children = sorted(base.iterdir())
|
||
except OSError:
|
||
return found
|
||
for child in children:
|
||
if not child.is_dir() or child.name.startswith("."):
|
||
continue
|
||
if not NAME_RE.match(child.name):
|
||
continue
|
||
try:
|
||
if find_skill_md(child) is None:
|
||
continue
|
||
except OSError:
|
||
continue
|
||
found.setdefault(child.name, []).append(str(child))
|
||
return found
|
||
|
||
|
||
def start_background_scan(name: str = "") -> bool:
|
||
"""tier3 后台扫盘（不阻塞当前推拉）：扫完把最终结果落库 skill.json。
|
||
|
||
已有扫描在跑则直接返回 False（单飞，不叠加线程）。扫描范围 =
|
||
内嵌常见终端根 + 扩展浅层根；UNC 根按 6 秒限时（防 9p 半死拖住线程）。
|
||
扫完打印一行进度（后台输出，不参与交互）。
|
||
"""
|
||
global _BG_SCAN_RUNNING
|
||
with _REG_LOCK:
|
||
if _BG_SCAN_RUNNING:
|
||
return False
|
||
_BG_SCAN_RUNNING = True
|
||
|
||
def _run() -> None:
|
||
global _BG_SCAN_RUNNING
|
||
try:
|
||
roots = list(dict.fromkeys(
|
||
builtin_skill_roots() + deep_scan_roots()))
|
||
merged: dict = {}
|
||
for r in roots:
|
||
if not _is_unc_root(r): # 本机根直扫
|
||
one = _scan_root_skills(r)
|
||
else: # UNC 根限时（半死 VM 不拖线程）
|
||
ok, res = _guard(lambda r=r: _scan_root_skills(r), 6.0)
|
||
one = res if ok and isinstance(res, dict) else {}
|
||
for n, paths in one.items():
|
||
merged.setdefault(n, []).extend(paths)
|
||
remember_skills(merged)
|
||
total = len(load_registry().get("skills", {}))
|
||
print("\n [skill.json] 后台扫盘完成：本次登记 %d 个 skill，"
|
||
"库内共 %d 个名字（已落库 %s）"
|
||
% (len(merged), total, REGISTRY_PATH))
|
||
if name and name not in merged:
|
||
print(" 提示：后台扫盘也没找到「%s」的真身，"
|
||
"可能它只存在于未纳入扫描的位置。" % name)
|
||
finally:
|
||
_BG_SCAN_RUNNING = False
|
||
|
||
threading.Thread(target=_run, daemon=True, name="skill-registry-scan").start()
|
||
return True
|
||
|
||
|
||
def real_src_locations(src: str) -> list[str]:
|
||
"""空壳/链接源的真身位置分层查找（real_src 用；返回目录列表，可能为空）。
|
||
|
||
① 内存注册表（启动已读 skill.json）→ ② 内嵌常见终端位置快扫（命中即落库）
|
||
→ ③ 现有跨根兜底 _alt_skill_dirs（.trash / env 根）→ 都没中调用方再后台扫盘。
|
||
"""
|
||
try:
|
||
p = Path(src)
|
||
name = p.name
|
||
except (OSError, ValueError):
|
||
return []
|
||
|
||
def _same(a, b) -> bool:
|
||
return os.path.normcase(os.path.abspath(str(a))) == \
|
||
os.path.normcase(os.path.abspath(str(b)))
|
||
|
||
def _valid(d) -> bool:
|
||
try:
|
||
return (not _same(d, p)) and Path(d).is_dir() and \
|
||
find_skill_md(Path(d)) is not None
|
||
except OSError:
|
||
return False
|
||
|
||
# ① 内存注册表：零扫盘直接命中（skill.json 启动已进内存）
|
||
hits = [d for d in registry_locations(name) if _valid(d)]
|
||
if hits:
|
||
return hits
|
||
# ② tier1 快扫：内嵌常见终端位置（秒级，不唤醒 WSL）
|
||
quick = scan_roots_for_name(name)
|
||
if quick:
|
||
remember_skills({name: quick}) # 扫到即落库，下次走 ①
|
||
return quick
|
||
# ③ 现有跨根兜底（含 .trash 回收站、OC_SKILL_ALT_ROOTS）
|
||
return [d for d in _alt_skill_dirs(p) if _valid(d)]
|
||
|
||
|
||
|
||
# ---------------------------------------------------------------- 软链接诊断（SKILL.md 缺失时）
|
||
def _is_link(path) -> bool:
|
||
"""是否是链接 —— Windows 上 symlink 与 junction 都要认。
|
||
junction 不是 islink()，须看 lstat 的 FILE_ATTRIBUTE_REPARSE_POINT(0x400)。"""
|
||
p = Path(path)
|
||
try:
|
||
if p.is_symlink():
|
||
return True
|
||
except OSError:
|
||
return False
|
||
if os.name == "nt":
|
||
try:
|
||
return bool(os.lstat(p).st_file_attributes & 0x400)
|
||
except OSError:
|
||
return False
|
||
return False
|
||
|
||
|
||
def _link_target(path) -> str:
|
||
"""读链接目标（symlink / junction）；读不到返回空串。"""
|
||
try:
|
||
return os.readlink(path)
|
||
except OSError:
|
||
return ""
|
||
|
||
|
||
def _unc_distro_rest(path) -> tuple[str, str] | None:
|
||
"""UNC 路径拆解：\\\\wsl.localhost\\<distro>\\<rest> -> (distro, /<rest>)。
|
||
非 WSL UNC 返回 None。rest 用 POSIX 风格（WSL 内路径）。"""
|
||
s = str(path)
|
||
m = re.match(r"\\\\wsl(?:\$|\.localhost)\\([^\\/]+)(.*)$", s, re.IGNORECASE)
|
||
if not m:
|
||
return None
|
||
rest = m.group(2).replace("\\", "/")
|
||
return m.group(1), (rest if rest.startswith("/") else "/" + rest)
|
||
|
||
|
||
def _resolve_link_real(path: Path, target: str) -> Path | None:
|
||
"""把链接目标解成真实路径。
|
||
- POSIX 目标（/root/...）出现在 Windows UNC 视角下 -> 换算回 \\\\wsl.localhost\\<distro>\\...；
|
||
- Windows 目标（含 \\??\\ / \\\\?\\ 前缀）剥前缀后按绝对/相对解析；
|
||
- 都读不出来 -> realpath 兜底。"""
|
||
t = (target or "").strip()
|
||
if t.startswith("/") and not t.startswith("//"):
|
||
parts = _unc_distro_rest(path)
|
||
if parts: # Windows 侧看 WSL 内的 POSIX symlink
|
||
return Path("\\\\wsl.localhost\\%s%s" % (parts[0], t.replace("/", "\\")))
|
||
return Path(t) # 本机就是 POSIX（跑在 WSL 里时）
|
||
for pre in ("\\\\?\\", "\\??\\"):
|
||
if t.startswith(pre):
|
||
t = t[len(pre):]
|
||
break
|
||
if t:
|
||
tp = Path(t)
|
||
if not tp.is_absolute():
|
||
tp = path.parent / t
|
||
return tp
|
||
try:
|
||
return Path(os.path.realpath(path))
|
||
except OSError:
|
||
return None
|
||
|
||
|
||
def _alt_skill_dirs(folder) -> list[str]:
|
||
"""空壳/断链目录时，跨已知 skill 根目录找同名 skill 的真实目录（真身所在）。
|
||
|
||
2026-09-25 用户实测场景：来源目录（如 .config/opencode/skills/<name>）两端
|
||
都是空壳、也没有任何链接，但大模型照样能调用该 skill —— 因为真身在另一个
|
||
根（~/.workbuddy/skills/<name>/SKILL.md，即 wb 终端）。诊断必须把「原始文件
|
||
位置」找出来，而不是只报「仍未找到」。
|
||
|
||
搜索根（可用环境变量 OC_SKILL_ALT_ROOTS 以 os.pathsep 追加）：
|
||
本机 ~/.workbuddy/skills、~/.config/opencode/skills（含 .trash 回收站）、
|
||
当前目录 .workbuddy/skills；Windows 上另查 WSL 端对应两根（folder 是
|
||
WSL UNC 直接查，否则 VM 在跑才查）。
|
||
全程只读、任何一步失败跳过；返回确实含 SKILL.md 的同名目录列表（去重，
|
||
已排除「诊断对象自己」）。SKILL.md 视图由 _alt_skill_md 上层包装。
|
||
"""
|
||
name = Path(folder).name
|
||
if not name:
|
||
return []
|
||
home = Path.home()
|
||
roots: list[Path] = [home / ".workbuddy" / "skills",
|
||
home / ".config" / "opencode" / "skills"]
|
||
extra = os.environ.get("OC_SKILL_ALT_ROOTS", "")
|
||
roots += [Path(p) for p in extra.split(os.pathsep) if p.strip()]
|
||
try:
|
||
roots.append(Path.cwd() / ".workbuddy" / "skills")
|
||
except OSError:
|
||
pass
|
||
if os.name == "nt":
|
||
parts = _unc_distro_rest(Path(folder))
|
||
if parts:
|
||
wsl_on = True # folder 本身就在 WSL 里，同端根一定可查
|
||
distro = parts[0]
|
||
else:
|
||
distro = DEFAULT_DISTRO
|
||
try:
|
||
wsl_on = _wsl_running(distro)
|
||
except Exception:
|
||
wsl_on = False
|
||
if wsl_on:
|
||
unc = Path("\\\\wsl.localhost\\%s" % distro)
|
||
roots += [unc / "root" / ".workbuddy" / "skills",
|
||
unc / "root" / ".config" / "opencode" / "skills"]
|
||
|
||
def _norm(p) -> str:
|
||
return os.path.normcase(os.path.abspath(str(p)))
|
||
|
||
self_dir = _norm(folder)
|
||
out: list[str] = []
|
||
seen: set[str] = set()
|
||
for r in roots:
|
||
# <根>/<同名> + 回收站 <根>/.trash/<时间戳>/<同名>
|
||
cands = [r / name]
|
||
try:
|
||
if (r / ".trash").is_dir():
|
||
cands += [t / name for t in sorted((r / ".trash").iterdir())
|
||
if t.is_dir()]
|
||
except OSError:
|
||
pass
|
||
for cand in cands:
|
||
if _norm(cand) == self_dir:
|
||
continue # 就是诊断对象自己（已知空壳，别自己找自己）
|
||
try:
|
||
md = find_skill_md(cand)
|
||
except OSError:
|
||
md = None
|
||
if md:
|
||
key = _norm(cand)
|
||
if key not in seen:
|
||
seen.add(key)
|
||
out.append(str(cand))
|
||
return out
|
||
|
||
|
||
def _alt_skill_md(folder) -> list[str]:
|
||
"""空壳/断链目录时，跨已知 skill 根目录找同名 skill 的 SKILL.md 真身。
|
||
|
||
_alt_skill_dirs 的上层视图（2026-09-25 抽出：目录列表供推送自动切换真身，
|
||
SKILL.md 列表供诊断打印），行为与抽出前一致。
|
||
"""
|
||
out: list[str] = []
|
||
seen: set[str] = set()
|
||
for d in _alt_skill_dirs(folder):
|
||
try:
|
||
md = find_skill_md(Path(d))
|
||
except OSError:
|
||
md = None
|
||
if md:
|
||
key = os.path.normcase(os.path.abspath(str(md)))
|
||
if key not in seen:
|
||
seen.add(key)
|
||
out.append(str(md))
|
||
return out
|
||
|
||
|
||
def real_src(src: str) -> str | None:
|
||
"""推送源是软链接 / 空壳时，解析出真身目录 —— 自动切换用（2026-09-25 用户要求：
|
||
选了 oc 里的空壳/链接 skill，推送时直接推真身（如 wb），不再提醒手动换来源）。
|
||
|
||
- 链接（symlink / junction）-> 解析链接目标，目标里有 SKILL.md 才切；
|
||
- 目录在但没有 SKILL.md（空壳）-> 分层找真身：内存注册表（skill.json，启动
|
||
已读入）-> 内嵌常见终端位置快扫（oc/wb/cc-switch…，命中即落库）-> 跨根兜底
|
||
_alt_skill_dirs（.trash / env 根）；都没命中再启动后台扫盘（不阻塞本次推拉，
|
||
扫完落库，下次直接走注册表）。
|
||
- 有效目录 / zip / SKILL.md 文件 -> 返回 None（用户选定的目录优先，保持原路径不动）。
|
||
全程只读；任何一步失败返回 None，由调用方走原诊断路径。
|
||
"""
|
||
try:
|
||
p = Path(src)
|
||
if not p.is_dir():
|
||
return None
|
||
except OSError:
|
||
return None
|
||
|
||
def _same(a, b) -> bool:
|
||
return os.path.normcase(os.path.abspath(str(a))) == \
|
||
os.path.normcase(os.path.abspath(str(b)))
|
||
|
||
# 1) 链接优先：切到链接目标（wb 的 skill 链到 oc，选 oc 也推 wb 真身）
|
||
try:
|
||
if _is_link(p):
|
||
real = _resolve_link_real(p, _link_target(p))
|
||
if real is not None and not _same(real, p) and real.is_dir() \
|
||
and find_skill_md(real) is not None:
|
||
return str(real)
|
||
except OSError:
|
||
pass
|
||
# 2) 空壳 / 断链：注册表 -> 快扫 -> 跨根兜底的分层查找
|
||
try:
|
||
if find_skill_md(p) is None:
|
||
for d in real_src_locations(str(p)):
|
||
if not _same(d, p):
|
||
return d
|
||
# 都没命中：后台扫盘兜底（tier3，daemon 线程，不阻塞当前推拉；
|
||
# 扫完落库 skill.json，下次同名走内存注册表一步命中）
|
||
start_background_scan(p.name)
|
||
except OSError:
|
||
pass
|
||
return None
|
||
|
||
|
||
def diag_skill_md(folder) -> list[str]:
|
||
"""SKILL.md 缺失时的软链接诊断（只读，2026-09-24 按用户反馈新增）。
|
||
|
||
oc / wb 里的 skill 有些是软链接形式（不是实体拷贝）—— 目录列得出来，
|
||
find_skill_md 却找不到文件。这层诊断试着把链接解出来，回答「真实文件在哪」：
|
||
① 目录本身是不是链接（symlink / junction 都认）；
|
||
② SKILL.md 是不是链接 / 断链 / 大小写不符（skill.md）；
|
||
③ 子项里的链接（POSIX 目标换算回 \\\\wsl.localhost 视图再查）；
|
||
④ 本地解不出且路径是 WSL UNC -> 进 WSL 用 readlink / find 再查一层。
|
||
返回诊断行列表；有找到真身会标 ✓，全程任何一步失败都跳过不抛异常。
|
||
"""
|
||
folder = Path(folder)
|
||
notes: list[str] = []
|
||
found_md: list[str] = [] # 找到的 SKILL.md 真实位置
|
||
|
||
def _md_at(base: Path, where: str) -> None:
|
||
"""在 base 下找 SKILL.md，找到记 notes + found_md。"""
|
||
try:
|
||
md = find_skill_md(base)
|
||
except OSError:
|
||
md = None
|
||
if md:
|
||
found_md.append(str(md))
|
||
notes.append(f" ✓ {where}有 SKILL.md：{md}")
|
||
|
||
# ① 目录本身是不是链接
|
||
base = folder
|
||
try:
|
||
dir_exists = folder.exists()
|
||
except OSError:
|
||
dir_exists = False
|
||
if _is_link(folder):
|
||
tgt = _link_target(folder)
|
||
if not tgt:
|
||
# 目标读不出来：WSL 9p symlink 在 Windows 侧 readlink/stat 双失败，
|
||
# 此时 realpath 只会解出「链接自身」→ 误报断链。不硬解，交给 ④ 层进 WSL 查
|
||
notes.append(f"目录「{folder.name}」是链接，但 Windows 侧读不到目标"
|
||
+ ("（疑似 WSL 内的软链接，交由 WSL 内二层诊断）"
|
||
if _unc_distro_rest(folder) else "（目标读取失败）"))
|
||
real = None
|
||
else:
|
||
real = _resolve_link_real(folder, tgt)
|
||
notes.append(f"目录「{folder.name}」本身是软链接 -> {tgt}")
|
||
if real:
|
||
notes.append(f" 真实位置：{real}")
|
||
try:
|
||
real_ok = real.is_dir()
|
||
except OSError:
|
||
real_ok = False
|
||
if real_ok:
|
||
base = real
|
||
_md_at(real, "真实位置")
|
||
else:
|
||
notes.append(" ✗ 真实位置不存在或不是目录（链接可能已断）")
|
||
elif not dir_exists and os.path.lexists(folder):
|
||
notes.append(f"目录「{folder.name}」是断链（链接还在，目标已不存在）")
|
||
|
||
# ② SKILL.md 本体：链接 / 断链 / 大小写不符
|
||
direct = base / SKILL_MD
|
||
try:
|
||
direct_ok = direct.is_file()
|
||
except OSError:
|
||
direct_ok = False
|
||
if direct_ok:
|
||
found_md.append(str(direct)) # 正常命中也要记，否则结论行会误报「仍未找到」
|
||
notes.append(f" ✓ 目录内直接有 SKILL.md：{direct}")
|
||
else:
|
||
if os.path.lexists(direct):
|
||
tgt = _link_target(direct)
|
||
real = _resolve_link_real(direct, tgt)
|
||
notes.append(f"{SKILL_MD} 是链接/断链 -> {tgt or '（目标读取失败）'}"
|
||
+ (f"，真实位置：{real}" if real else ""))
|
||
if real:
|
||
try:
|
||
if real.is_file():
|
||
found_md.append(str(real))
|
||
notes.append(f" ✓ 真实文件位置：{real}")
|
||
elif not real.exists():
|
||
notes.append(f" ✗ 链接已断（目标不存在）：{real}")
|
||
except OSError:
|
||
pass
|
||
else:
|
||
# 大小写不符（区分大小写的环境会找不到）
|
||
try:
|
||
for c in sorted(base.iterdir()):
|
||
if c.name.lower() == SKILL_MD.lower() and c.name != SKILL_MD:
|
||
notes.append(f"发现大小写不符的文件：{c.name}（应为 {SKILL_MD}，"
|
||
f"区分大小写的环境会找不到）：{c}")
|
||
except OSError:
|
||
pass
|
||
|
||
# ③ 子项里的链接（含指向别处的子目录 / 文件）
|
||
try:
|
||
children = sorted(base.iterdir())
|
||
except OSError as exc:
|
||
children = []
|
||
notes.append(f"无法列出目录内容：{exc}")
|
||
if not children:
|
||
notes.append("目录是空的（0 个文件/子目录）—— 像空壳目录，不是链接问题")
|
||
for c in children:
|
||
if c.name == SKILL_MD or not _is_link(c):
|
||
continue
|
||
tgt = _link_target(c)
|
||
real = _resolve_link_real(c, tgt)
|
||
line = f"子项「{c.name}」是软链接 -> {tgt or '（目标读取失败）'}"
|
||
if real:
|
||
line += f"，真实位置：{real}"
|
||
notes.append(line)
|
||
if real:
|
||
try:
|
||
real_ok = real.exists()
|
||
except OSError:
|
||
real_ok = False
|
||
if real_ok:
|
||
_md_at(real, f"「{c.name}」的真实位置")
|
||
else:
|
||
notes.append(f" ✗ 链接已断（目标不存在）：{real}")
|
||
|
||
# ④ WSL UNC：本地解不出时进 WSL 再查一层（readlink / find -L）
|
||
if not found_md and os.name == "nt":
|
||
parts = _unc_distro_rest(folder)
|
||
if parts:
|
||
distro, posix = parts
|
||
try:
|
||
running = _wsl_running(distro)
|
||
except Exception:
|
||
running = False
|
||
if running:
|
||
# wsl.exe 会把原始 commandline 交给 distro shell 二次解析，
|
||
# script 双引号里的 $var 被外层提前展开（实测 $1/$p 全变空，
|
||
# 脚本 rc=0 却零输出）—— 脚本走 base64 传输，外层只见到
|
||
# 无 $ 无引号的管道命令（2026-09-24 单测暴露并实测验证）
|
||
import shlex as _shlex
|
||
inner = (
|
||
f'p={_shlex.quote(posix)}; '
|
||
'[ -L "$p" ] && echo "LINK $p -> $(readlink "$p")"; '
|
||
'if [ -d "$p" ]; then '
|
||
'find -L "$p" -maxdepth 4 -name "SKILL.md" 2>/dev/null | head -5 | sed "s/^/MD /"; '
|
||
'find "$p" -maxdepth 3 -type l 2>/dev/null | head -10 | while IFS= read -r l; do '
|
||
'echo "LINK $l -> $(readlink "$l")"; done; fi'
|
||
)
|
||
cmd = f"echo {base64.b64encode(inner.encode()).decode()} | base64 -d | sh"
|
||
try:
|
||
code, out = _wsl_safe_run(
|
||
["wsl.exe", "-d", distro, "--", "sh", "-c", cmd],
|
||
timeout=20)
|
||
except Exception:
|
||
code, out = 1, b""
|
||
if code == 0:
|
||
hits = 0
|
||
for ln in out.decode("utf-8", "replace").splitlines():
|
||
ln = ln.strip()
|
||
if ln.startswith("MD "):
|
||
wsl_md = ln[3:].strip()
|
||
found_md.append(wsl_md)
|
||
hits += 1
|
||
notes.append(f" ✓ WSL 内找到 SKILL.md：{wsl_md}")
|
||
elif ln.startswith("LINK "):
|
||
hits += 1
|
||
notes.append(f"WSL 内链接：{ln[5:].strip()}")
|
||
if not hits:
|
||
# 空结果也要说话（2026-09-25）：静默会让人误以为二层没跑
|
||
notes.append(" ℹ WSL 内二层已查：目录里没有 SKILL.md、也没有链接"
|
||
"（WSL 内视角与 Windows 一致）")
|
||
else:
|
||
notes.append(f" ℹ WSL 内二层查询未完成（rc={code}），只按 Windows 视角给结论")
|
||
else:
|
||
# 不静默跳过（2026-09-24 单测暴露）：说明二层没查，让用户知道还能再查
|
||
notes.append(" ℹ WSL 当前未运行，跳过 WSL 内二层诊断"
|
||
"（唤醒 WSL 后重试，可查出链接真身；VM 冷启动约 5–20 秒，不自动唤醒）")
|
||
|
||
# 结论：先跨已知根找同名真身（空壳目录的常见真相 —— 真身在 wb 等另一个根，
|
||
# 大模型调用的正是那个文件），找不到才回退「仍未找到」
|
||
if not found_md:
|
||
alts = _alt_skill_md(folder)
|
||
if alts:
|
||
notes.append("→ 本目录里没有 SKILL.md（空壳/断链），但同名 skill 的真身在别处：")
|
||
for a in alts:
|
||
notes.append(f" {a}")
|
||
notes.append(" 大模型实际调用的就是上面这些文件；菜单推送会自动切到上面的"
|
||
"真身目录，不用手动换来源。")
|
||
else:
|
||
notes.append("→ 仍未找到 SKILL.md 的真实位置：目录可能是空壳 / 链接目标里也没有文件，"
|
||
"建议换一个完整的 skill 来源再推。")
|
||
return notes
|
||
|
||
|
||
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 的候选目录。"""
|
||
src = clean_input(src) # 剥掉粘贴时带上的引号（"C:\..." -> C:\...）
|
||
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:
|
||
# 读操作也先唤醒 VM（“先给机器启动起来，再去读配置”）；
|
||
# 唤醒失败 → 直接报错返回，绝不回落到 scan_skill_dir 的 UNC 冷启动
|
||
gate_err = _wsl_ensure_running(args.distro, "list")
|
||
if gate_err:
|
||
print(gate_err)
|
||
return 1
|
||
root = wsl_skill_root(args.distro)
|
||
|
||
# 磁盘扫描 = 与 probe_push / probe_pull / probe_delete 同源的 ground truth。
|
||
# 2026-09-24 起不再走 OpenCode API（用户反馈：读 skill 目录不需要 API ——
|
||
# 连上 WSL 直接查 skill 目录即可，本机的就返回本机 OC 的 skill 目录；
|
||
# API 返回的是 OC 运行时加载的 skill，可能滞后于磁盘、还多一层连接失败面）。
|
||
skills = scan_skill_dir(root)
|
||
source = f"目录扫描（{root}）"
|
||
|
||
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))
|
||
warned = False
|
||
for s in skills:
|
||
desc = s["description"].replace("\n", " ")
|
||
if len(desc) > 70:
|
||
desc = desc[:67] + "..."
|
||
# 缺 SKILL.md 的空壳/断链提前标出（与推送同源 find_skill_md，2026-09-24）
|
||
mark = ""
|
||
try:
|
||
if find_skill_md(Path(root) / s["name"]) is None:
|
||
mark = " ⚠缺 SKILL.md"
|
||
warned = True
|
||
except OSError:
|
||
pass
|
||
print(f"{s['name'].ljust(width)} {desc}{mark}")
|
||
if warned:
|
||
print("（⚠ = 找不到 SKILL.md：空壳目录或链接断了；推送选中确认时会打印软链接诊断）")
|
||
print(f"\n共 {len(skills)} 个 skill。WSL skill 目录：{root}")
|
||
return 0
|
||
|
||
|
||
def _is_unc_root(path) -> bool:
|
||
"""目标是否是 WSL UNC 视图（\\wsl.localhost\\... 或 \\wsl$\\...）——
|
||
这类路径读写前要先唤醒 VM；本机目录（wb/oc 等）完全不碰 WSL。"""
|
||
s = str(path).lower()
|
||
return "wsl.localhost" in s or s.startswith("\\\\wsl$")
|
||
|
||
|
||
def cmd_push(args) -> int:
|
||
# 目标终端：--to 简称（agents.md 前缀）/ 完整目录；缺省 = WSL OC
|
||
tgt = resolve_agent(getattr(args, "dst", "") or "", args.distro)
|
||
if not tgt.get("ok"):
|
||
raise SystemExit(tgt.get("err"))
|
||
root = tgt["dir"]
|
||
root_path = Path(root)
|
||
# UNC 写之前确保 VM 在跑（否则 \\wsl.localhost 冷启动可能挂死）
|
||
if _is_unc_root(root):
|
||
gate_err = _wsl_ensure_running(args.distro, "push")
|
||
if gate_err:
|
||
raise SystemExit(gate_err)
|
||
print(f"目标终端：{tgt['label']} -> {root}")
|
||
if not root_path.is_dir():
|
||
hint = (f"请确认 WSL 发行版名称（当前 {args.distro}），可用 --distro 指定。"
|
||
if tgt["kind"] == "default" else "请检查 agents.md 目录配置（或 --to 传入的路径）。")
|
||
if args.dry_run:
|
||
print(f"（dry-run）目标目录尚不存在：{root}")
|
||
else:
|
||
raise SystemExit(f"找不到目标 skill 目录：{root}\n{hint}")
|
||
|
||
with tempfile.TemporaryDirectory(prefix="ocskill_") as tmp:
|
||
work = Path(tmp)
|
||
# 源是软链接 / 空壳目录 -> 与菜单同一条分层链自动切真身（2026-09-25 CLI 对称；
|
||
# 只对目录生效 —— zip / SKILL.md / base64 走原路径，不动）
|
||
try:
|
||
src_path = clean_input(args.source)
|
||
if Path(src_path).is_dir():
|
||
real = real_src(src_path)
|
||
if real and os.path.normcase(os.path.abspath(real)) != \
|
||
os.path.normcase(os.path.abspath(src_path)):
|
||
print("源是软链接或空壳，已自动切换到真实位置推送：")
|
||
print(f" 原：{src_path}")
|
||
print(f" 新：{real}")
|
||
args.source = real
|
||
except OSError:
|
||
pass
|
||
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
|
||
|
||
# 探查式覆盖：没有同名文件夹就直接放（目录和内容都没有，覆盖无从谈起）；
|
||
# 确有同名才走 --force —— 「要不要覆盖」这一问只在真的存在时才发生。
|
||
if dest.exists():
|
||
if not args.force:
|
||
raise SystemExit(
|
||
f"目标已存在同名 skill：{dest}\n"
|
||
f"加 --force 覆盖（覆盖前会自动挪进 .trash/ 备份，可恢复）。"
|
||
)
|
||
backup = _backup_then_remove(dest)
|
||
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:
|
||
# 来源终端：--from 简称（agents.md 前缀）/ 完整目录；缺省 = WSL OC
|
||
src = resolve_agent(getattr(args, "src", "") or "", args.distro)
|
||
if not src.get("ok"):
|
||
raise SystemExit(src.get("err"))
|
||
root = src["dir"]
|
||
# UNC 读之前同样确保 VM 在跑（没跑则唤醒，唤醒失败给清晰错误）
|
||
if _is_unc_root(root):
|
||
gate_err = _wsl_ensure_running(args.distro, "pull")
|
||
if gate_err:
|
||
raise SystemExit(gate_err)
|
||
|
||
name = clean_input(args.name or "") # 粘贴带引号也认；缺省进分页列表
|
||
if not name:
|
||
# 宝宝级：不记得名字 -> 打印该来源的 skill 列表
|
||
#（直读文件夹，不启动 OC 服务、不发任何请求）
|
||
print(f"来源终端：{src['label']} -> {root}")
|
||
names = list_skill_names(root)
|
||
if not names:
|
||
raise SystemExit(f"来源「{src['label']}」下没有 skill：{root}")
|
||
render_page(names, int(getattr(args, "page", 0) or 1), root)
|
||
print(" 翻页：下一页 / 上一页 / 第N页 / next / last / N page")
|
||
print(f" 记住名字后：python ocskill.py pull <名字> --from {src['label']}")
|
||
return 0
|
||
|
||
print(f"来源终端：{src['label']} -> {root}")
|
||
probe = probe_pull(name, args.distro, root=root)
|
||
if not probe.get("ok"):
|
||
raise SystemExit(probe["err"])
|
||
src_dir = Path(probe["src"])
|
||
|
||
# 来源是软链接 / 空壳 -> 与推送同一条分层链自动切真身（2026-09-25 推拉对称：
|
||
# 注册表 -> 内嵌常见终端快扫 -> 跨根兜底；没命中再后台扫盘，不阻塞本次拉取）
|
||
try:
|
||
need_real = find_skill_md(src_dir) is None
|
||
except OSError:
|
||
need_real = True
|
||
if need_real:
|
||
real = real_src(str(src_dir))
|
||
if real and os.path.normcase(os.path.abspath(real)) != \
|
||
os.path.normcase(os.path.abspath(str(src_dir))):
|
||
print(f"来源是软链接或空壳，已自动切换到真实位置拉取：")
|
||
print(f" 原：{src_dir}")
|
||
print(f" 新：{real}")
|
||
src_dir = Path(real)
|
||
else:
|
||
start_background_scan(name)
|
||
print(f"⚠ 来源目录缺 SKILL.md（空壳或链接断了）：{src_dir}")
|
||
print(" 已在后台扫盘找真身（结果落库 skill.json，稍后重试即可自动切）。")
|
||
|
||
# 输出目录不由用户决定：默认直接落本机 OC 的 skill 目录（写完即被 watch 热加载）
|
||
out_dir = Path(args.out) if args.out else Path(WIN_LOCAL_ROOT)
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
dest = out_dir / name
|
||
# 探查式覆盖：本机没有同名目录就直接放（不该问覆盖）；有才需要 --force
|
||
if dest.exists():
|
||
if not args.force:
|
||
raise SystemExit(
|
||
f"本机已存在同名 skill：{dest}\n"
|
||
f"加 --force 覆盖（覆盖前会自动挪进 .trash/ 备份，可恢复）。"
|
||
)
|
||
backup = _backup_then_remove(dest)
|
||
print(f"已备份旧版本 -> {backup}")
|
||
shutil.copytree(src_dir, dest)
|
||
print(f"已取回：{dest}")
|
||
return 0
|
||
|
||
|
||
def cmd_delete(args) -> int:
|
||
"""删除 skill（补齐管理闭环：列表 / 推送 / 拉取 / 删除 / 覆盖更新）。
|
||
|
||
目标目录三选一（与 push 的 --to 同语义）：--dir 显式目录 > --local 本机 > 缺省 WSL 端。
|
||
删除绝不直接 rmtree：先挪进 <skills根>/.trash/<时间戳>/，删错了可恢复。
|
||
--yes 跳过交互确认（菜单层已经问过用户时传入）。
|
||
"""
|
||
name = clean_input(args.name)
|
||
if not name:
|
||
raise SystemExit("技能名为空。")
|
||
|
||
dir_arg = clean_input(getattr(args, "dir", "") or "")
|
||
root = Path(dir_arg) if dir_arg else (
|
||
Path(WIN_LOCAL_ROOT) if args.local else Path(wsl_skill_root(args.distro)))
|
||
|
||
# UNC 访问：先唤醒再读，否则没跑的 VM 会被冷启动拖死（本机目录不碰 WSL）
|
||
if _is_unc_root(root):
|
||
gate_err = _wsl_ensure_running(args.distro, "delete")
|
||
if gate_err:
|
||
raise SystemExit(gate_err)
|
||
|
||
probe = probe_delete(name, args.distro, args.local, root=dir_arg or None)
|
||
if not probe["ok"]:
|
||
raise SystemExit(probe["err"])
|
||
dest = Path(probe["dest"])
|
||
side = probe["side"]
|
||
|
||
if not args.yes:
|
||
try:
|
||
ans = input(
|
||
f"确认删除{side} skill「{name}」？\n"
|
||
f" 路径：{dest}\n"
|
||
f" （会先挪进 .trash/ 备份，可恢复）[y/N]: "
|
||
).strip()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print("\n已取消删除。")
|
||
return 0
|
||
if ans.lower() not in ("y", "yes", "是"):
|
||
print("已取消删除。")
|
||
return 0
|
||
|
||
backup = _backup_then_remove(dest)
|
||
print(f"已删除{side} skill「{name}」（备份在 {backup}，需要时可移回原位）")
|
||
return 0
|
||
|
||
def _dirs_line(tag: str, root: str, timeout: float) -> None:
|
||
"""打印一端 skill 目录情况：路径 + 存在性 + skill 数量（只读，限时防卡死）。"""
|
||
print(f" {tag}：{root}")
|
||
ok, res = _guard(lambda: (Path(root).is_dir(), list_skill_names(root)), timeout)
|
||
if not ok:
|
||
if res == "TIMEOUT":
|
||
print(f" 状态：读取超时（>{timeout:g}s，WSL UNC 无响应；不再无限挂起）")
|
||
else:
|
||
print(f" 状态：读取失败（{res}）")
|
||
return
|
||
isdir, names = res
|
||
if isdir:
|
||
print(f" 状态：存在，{len(names)} 个 skill")
|
||
if names:
|
||
shown = "、".join(names[:8]) + ("…" if len(names) > 8 else "")
|
||
print(f" skill：{shown}")
|
||
else:
|
||
print(" 状态：目录不存在（首次推送/拉取时会自动创建）")
|
||
|
||
|
||
def cmd_dirs(args) -> int:
|
||
"""展示两端 skill 目录情况：路径 + 存在性 + skill 数量 + 名字摘要。
|
||
|
||
2026-09-24 起不再探测 OC API（用户反馈：读 skill 目录不需要 API，
|
||
连上 WSL 直接查目录即可，API 探测是画蛇添足）。
|
||
目录信息限时读取防卡死；WSL 端先唤醒再读。
|
||
"""
|
||
local_root = WIN_LOCAL_ROOT
|
||
wsl_root = wsl_skill_root(args.distro)
|
||
|
||
print("两端 skill 目录情况：")
|
||
_dirs_line("本机 OC（pull 默认落点）", local_root, 8.0)
|
||
gate_err = _wsl_ensure_running(args.distro, "dirs")
|
||
if gate_err:
|
||
print(f" WSL OC（push 缺省目标）：{wsl_root}")
|
||
print(f" 状态：无法访问（{gate_err}）")
|
||
else:
|
||
_dirs_line("WSL OC（push 缺省目标）", wsl_root, 20.0)
|
||
return 0
|
||
|
||
|
||
def _stamp() -> str:
|
||
import datetime
|
||
|
||
return datetime.datetime.now().strftime("%Y%m%d%H%M%S")
|
||
|
||
def _backup_then_remove(dest: Path) -> Path:
|
||
"""覆盖前把同名目录挪进 `<skills根>/.trash/<时间戳>/`，绝不直接 rmtree。
|
||
|
||
- 可恢复：覆盖错了还能从 .trash 里捞回来
|
||
- 不污染：统一放点前缀 .trash/ 下，OC 的 skill loader 与 scan_skill_dir
|
||
都跳过 `.` 开头的目录；旧的 `<name>.bak-<ts>` 平级备份会带 SKILL.md，
|
||
反而可能被 OC 当成一个多余的 skill 加载出来
|
||
"""
|
||
trash = dest.parent / ".trash" / _stamp()
|
||
trash.mkdir(parents=True, exist_ok=True)
|
||
target = trash / dest.name
|
||
if target.exists(): # 同秒内重复覆盖
|
||
target = trash / f"{dest.name}-{_stamp()}"
|
||
shutil.move(str(dest), str(target))
|
||
return target
|
||
|
||
|
||
# ---------------------------------------------------------------- 参数顺序自愈
|
||
# 共享 flag 定义在主解析器（子命令之前），`ocskill list --distro xxx` 这种
|
||
# 写法会被 argparse 判为 unrecognized。parse 之前把共享 flag 提到最前。
|
||
_SHARED_VALUE_FLAGS = {"--distro"}
|
||
|
||
|
||
def _hoist_shared(argv: list[str]) -> list[str]:
|
||
head: list[str] = []
|
||
rest: list[str] = []
|
||
i = 0
|
||
n = len(argv)
|
||
while i < n:
|
||
tok = argv[i]
|
||
key = tok.split("=", 1)[0]
|
||
if key in _SHARED_VALUE_FLAGS:
|
||
head.append(tok)
|
||
if "=" in tok:
|
||
i += 1
|
||
elif i + 1 < n:
|
||
head.append(argv[i + 1])
|
||
i += 2
|
||
else:
|
||
i += 1
|
||
else:
|
||
rest.append(tok)
|
||
i += 1
|
||
return head + rest
|
||
|
||
|
||
# ---------------------------------------------------------------- 入口
|
||
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字符串> --base64\n"
|
||
" python ocskill.py push .\\my-skill --dry-run\n"
|
||
" python ocskill.py pull # 不记得名字：打印第一页列表（每页15个）\n"
|
||
" python ocskill.py pull --from wb --page 2 # 列 wb(workbuddy) 终端第 2 页\n"
|
||
" python ocskill.py pull java-expert # 记得名字：直接拉回本机 OC skill 目录\n"
|
||
" python ocskill.py push .\\my-skill --to wb # 推到 wb(workbuddy) 终端\n"
|
||
" python ocskill.py delete java-expert # 删 WSL 端（先挪 .trash/）\n"
|
||
" python ocskill.py delete java-expert --local # 删本机\n"
|
||
),
|
||
)
|
||
p.add_argument("--distro", default=DEFAULT_DISTRO, help=f"WSL 发行版名（默认 {DEFAULT_DISTRO}）")
|
||
|
||
sub = p.add_subparsers(dest="cmd")
|
||
|
||
sp = sub.add_parser("list", help="直读 WSL OC 的 skill 目录，返回 skill 列表（不走 API）")
|
||
sp.set_defaults(func=cmd_list)
|
||
|
||
sp = sub.add_parser("push", help="推送 skill（zip / 目录 / SKILL.md / base64）到目标终端")
|
||
sp.add_argument("source", help="skill 文件路径、目录、zip 或 base64 字符串")
|
||
sp.add_argument("--to", dest="dst", default="",
|
||
help="目标终端：agents.md 简称（wb/oc…前缀）或完整目录（缺省 wsl）")
|
||
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="拉取 skill 回本机 OC skill 目录；不带名字则打印来源分页列表")
|
||
sp.add_argument("name", nargs="?", default="",
|
||
help="skill 名称（可带引号，会自动剥掉；缺省打印分页列表，每页 15 个）")
|
||
sp.add_argument("--from", dest="src", default="",
|
||
help="来源终端：agents.md 简称（wb/oc…前缀）或完整目录（缺省 wsl）")
|
||
sp.add_argument("--page", type=int, default=1, help="列表页码（缺省第 1 页，每页 15 个）")
|
||
sp.add_argument("--out", default=None, help=f"输出目录（默认本机 OC skill 目录：{WIN_LOCAL_ROOT}）")
|
||
sp.add_argument("--force", action="store_true", help="覆盖本机同名目录（先挪进 .trash/ 备份）")
|
||
sp.set_defaults(func=cmd_pull)
|
||
|
||
sp = sub.add_parser("delete", help="删除 skill（先挪进 .trash/ 备份，可恢复）")
|
||
sp.add_argument("name", help="skill 名称（可带引号，会自动剥掉）")
|
||
sp.add_argument("--local", action="store_true", help="删本机的 skill（缺省删 WSL 端）")
|
||
sp.add_argument("--dir", default="",
|
||
help="显式指定 skill 目录（与 push --to 同语义；给了它就忽略 --local）")
|
||
sp.add_argument("--yes", "-y", action="store_true", help="跳过交互确认（菜单层已确认时用）")
|
||
sp.set_defaults(func=cmd_delete)
|
||
|
||
sp = sub.add_parser("dirs", help="显示两端 skill 目录情况（路径 + skill 数量）")
|
||
sp.set_defaults(func=cmd_dirs)
|
||
|
||
return p
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
if argv is None:
|
||
argv = sys.argv[1:]
|
||
argv = _hoist_shared(list(argv))
|
||
parser = build_parser()
|
||
args = parser.parse_args(argv)
|
||
if not getattr(args, "cmd", None):
|
||
parser.print_help()
|
||
return 0
|
||
load_registry() # 启动时把 skill.json 读进内存（读失败走空表，不影响执行）
|
||
try:
|
||
return args.func(args)
|
||
except KeyboardInterrupt:
|
||
print("\n已取消。")
|
||
return 130
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|