youpaishidifu/oc_session
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

skill推拉重构：输入成对引号剥离+输出固定+探查式覆盖+删除闭环，覆盖删除统一进.trash备份

Browse Source
This commit is contained in:
ypshidifu
2026-09-23 19:54:32 +08:00
parent 2208ad0f7c
commit 3d65720bb3
2 changed files with 287 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
oc.py
+101 -23
View File
@@ -60,9 +60,12 @@ SKILL_ALIASES = {
"1": "list", "1.": "list", "list": "list", "列表": "list", "查看": "list",
"技能列表": "list", "看列表": "list",
"2": "push", "2.": "push", "push": "push", "推送": "push", "上传": "push", "推": "push",
"覆盖更新": "push", "更新": "push", "覆盖": "push",
"3": "pull", "3.": "pull", "pull": "pull", "拉取": "pull", "下载": "pull", "拉": "pull",
"4": "dirs", "4.": "dirs", "dirs": "dirs", "目录": "dirs", "状态": "dirs",
"技能目录": "dirs",
"5": "delete", "5.": "delete", "delete": "delete", "del": "delete",
"删除": "delete", "删": "delete", "移除": "delete", "删掉": "delete",
}
MCP_ALIASES = {
@@ -151,19 +154,109 @@ def run_safe(fn, argv: list[str]) -> int:
# ---------------------------------------------------------------- 子菜单：技能管理
def _yes(prompt: str) -> bool:
"""统一的 y/N 判定（默认 N，回车即取消）。"""
return match(ask(prompt), {"y": "y", "yes": "y", "是": "y"}, None) == "y"
def _flow_push() -> None:
"""推送 / 覆盖更新：输入 -> 校验规范 -> 探查同名 -> 确有同名才问一次覆盖 -> 写入。
输出固定为 WSL OC 的 skill 目录，不由用户决定。
目标没有同名文件夹时直接写（没有目录和内容，覆盖无从谈起，问了也是白问）。
"""
src = ocskill.clean_input(ask(" 技能输入（路径 / zip / SKILL.md，可带引号，回车取消）: "))
if not src:
return
# 1) 先过一遍规范校验（只读探查，不写盘）
probe = ocskill.probe_push(src)
if not probe.get("ok"):
print(" 校验未通过：%s" % probe.get("err"))
return
name = probe["name"]
desc = (probe.get("description") or "").replace("\n", " ")
print(" 校验通过：「%s」" % name)
if desc:
print(" 描述：%s" % (desc[:90] + ("..." if len(desc) > 90 else "")))
print(" 输出：%s（固定，不需选择）" % probe["dest"])
# 2) 探查同名：不存在 -> 直接放，不问覆盖
if not probe.get("exists"):
print(" 目标无同名目录，直接推送。")
run_safe(ocskill.main, ["push", src])
return
# 3) 确有同名文件夹 -> 这才有资格问一次覆盖
if _yes(" 已存在同名 skill「%s」，覆盖？(y/N): " % name):
print(" 覆盖更新（旧版本先挑进 .trash/ 备份）。")
run_safe(ocskill.main, ["push", src, "--force"])
else:
print(" 已跳过覆盖，未改动任何文件。")
def _flow_pull() -> None:
"""拉取：输入技能名 -> 探查 WSL 端存在 -> 探查本机同名 -> 确有同名才问一次覆盖 -> 写入。
输出固定为本机 OC 的 skill 目录，不由用户决定。
"""
name = ocskill.clean_input(ask(" 要拉取的技能名（可带引号，回车取消）: "))
if not name:
return
probe = ocskill.probe_pull(name)
if not probe.get("ok"):
print(" %s" % probe.get("err"))
return
print(" 输出：%s（固定，不需选择）" % probe["dest"])
if not probe.get("exists"):
print(" 本机无同名目录，直接拉取。")
run_safe(ocskill.main, ["pull", name])
return
if _yes(" 本机已存在同名 skill「%s」，覆盖？(y/N): " % name):
print(" 覆盖更新（旧版本先挑进 .trash/ 备份）。")
run_safe(ocskill.main, ["pull", name, "--force"])
else:
print(" 已跳过覆盖，未改动任何文件。")
def _flow_delete() -> None:
"""删除：输入技能名 -> 选端 -> 展示将删路径 -> 确认 -> 删除（先挑 .trash/ 可恢复）。"""
name = ocskill.clean_input(ask(" 要删除的技能名（可带引号，回车取消）: "))
if not name:
return
side = match(
ask(" 删哪端？(1=Wsl / 2=本机，回车取消): "),
{"1": "wsl", "1.": "wsl", "wsl": "wsl", "ws": "wsl", "远程": "wsl", "wsl端": "wsl",
"2": "local", "2.": "local", "local": "local", "本机": "local", "本地": "local", "本机端": "local"},
None,
)
if side is None:
print(" 已取消。")
return
local = side == "local"
probe = ocskill.probe_delete(name, ocskill.DEFAULT_DISTRO, local)
if not probe.get("ok"):
print(" %s" % probe.get("err"))
return
print(" 将删除%s：%s" % (probe["side"], probe["dest"]))
print(" （会先挑进 .trash/ 备份，删错了可移回）")
if not _yes(" 确认删除？(y/N): "):
print(" 已取消删除。")
return
argv = ["delete", name, "--yes"]
if local:
argv.append("--local")
run_safe(ocskill.main, argv)
def menu_skill(shared: dict) -> None:
while True:
banner("技能管理（skill）")
print(" 0. 返回上层")
print(" 1. 查看技能列表")
print(" 2. 推送技能到 OC（zip / 目录 / SKILL.md / base64）")
print(" 3. 拉取技能到本机")
print(" 2. 推送 / 覆盖更新技能到 WSL OC（输入即校验，输出固定为 WSL skill 目录）")
print(" 3. 拉取技能到本机 OC（输出固定为本机 skill 目录）")
print(" 4. 查看技能目录与连接状态")
print(" 5. 删除技能（先挑 .trash/ 备份，可恢复）")
print(" 提示：任意位置输入 /home 或「返回首页」直接回顶层菜单")
print(" 覆盖只在确有同名文件夹时才会问你一次")
raw = ask("请选择（编号或中文，/home 回首页）: ")
act = match(raw, SKILL_ALIASES)
if act is None:
print(" 没看懂「%s」，输入 0-4 或中文，如：列表 / 推送 / 拉取" % raw.strip())
print(" 没看懂「%s」，输入 0-5 或中文，如：列表 / 推送 / 拉取 / 删除" % raw.strip())
continue
if act == "back":
return
@@ -173,26 +266,11 @@ def menu_skill(shared: dict) -> None:
elif act == "dirs":
run_safe(ocskill.main, ["dirs"])
elif act == "push":
src = ask(" 技能路径（文件 / 目录 / zip，回车取消）: ").strip()
if not src:
continue
argv = ["push", src]
if match(ask(" 同名覆盖？(y/N): "), {"y": "y", "yes": "y", "是": "y"}, None) == "y":
argv.append("--force")
if match(ask(" 只校验不写入？(y/N): "), {"y": "y", "yes": "y", "是": "y"}, None) == "y":
argv.append("--dry-run")
run_safe(ocskill.main, argv)
_flow_push()
elif act == "pull":
name = ask(" 要拉取的技能名（回车取消）: ").strip()
if not name:
continue
argv = ["pull", name]
out = ask(" 输出目录（回车=当前目录）: ").strip()
if out:
argv += ["--out", out]
if match(ask(" 覆盖本地同名目录？(y/N): "), {"y": "y", "yes": "y", "是": "y"}, None) == "y":
argv.append("--force")
run_safe(ocskill.main, argv)
_flow_pull()
elif act == "delete":
_flow_delete()
pause()
@@ -274,7 +352,7 @@ def menu_top() -> None:
# ---------------------------------------------------------------- 直传模式
SKILL_CMDS = {"list", "push", "pull", "dirs"}
SKILL_CMDS = {"list", "push", "pull", "dirs", "delete"}
MCP_CMDS = {"status", "on", "off", "mode", "backup"}
ocskill.py
+186 -15
View File
@@ -6,11 +6,18 @@ ocskill —— 打通本机与 WSL OpenCode 的 skill 互通
做什么：
list 运行时连接 WSL 里的 OpenCode，返回它的 skill 列表
push 把一个新的 skill 文件（压缩包 / 目录 / 单个 SKILL.md / base64）
先解码并校验是不是标准 skill 格式，然后写进 OC 的 skill 目录
先解码并校验是不是标准 skill 格式，然后写进 WSL OC 的 skill 目录
（压缩包会额外做一次解压；OC 自带目录 watch，写完即热加载）
pull 反向操作：把 WSL 里的某个 skill 拷回本机
pull 反向操作：把 WSL 里的某个 skill 拷回本机 OC 的 skill 目录
delete 删除某个 skill（先挪进 .trash/ 备份，可恢复）
dirs 显示两端 skill 目录位置
输入/输出约定（输出不由用户决定）：
输入 路径/技能名都过一遍 clean_input，粘贴带引号的完整路径也认
输出 push -> WSL OC skill 目录；pull -> 本机 OC skill 目录
覆盖 探查式：先看同名文件夹存不存在 —— 不存在就直接写（不问）；
确有同名才需要 --force，覆盖前统一挪进 .trash/<时间戳>/ 备份
零依赖：只用 Python 标准库（urllib / zipfile / shutil），无需 pip install。
"""
@@ -417,6 +424,94 @@ def _frontmatter(text: str) -> dict:
return data
# ---------------------------------------------------------------- 输入净化与探查
_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) -> dict:
"""推送前探查（只读，不写盘）：校验 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:
return {"ok": False, "err": err}
name = meta.get("name", "")
dest = Path(wsl_skill_root(distro)) / 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 probe_pull(name: str, distro: str = DEFAULT_DISTRO) -> dict:
"""拉取前探查（只读）：WSL 端是否存在该 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(wsl_skill_root(distro)) / 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"WSL 端不存在该 skill：{src}"}
return {"ok": True, "name": name, "dest": str(dest), "exists": local_exists}
def probe_delete(name: str, distro: str = DEFAULT_DISTRO, local: bool = False) -> dict:
"""删除前探查（只读）：确认目标存在并给出将删除的路径。"""
name = clean_input(name)
if not name:
return {"ok": False, "err": "技能名为空"}
if not NAME_RE.match(name):
return {"ok": False, "err": f"非法 skill 名称「{name}」（只允许字母/数字/点/下划线/连字符，且以字母或数字开头）"}
root = Path(WIN_LOCAL_ROOT) if local else Path(wsl_skill_root(distro))
dest = root / name
try:
ok = dest.is_dir()
except OSError:
ok = False
if not ok:
side = "本机" if local else "WSL"
return {"ok": False, "err": f"{side}端不存在该 skill：{dest}"}
return {"ok": True, "name": name, "dest": str(dest), "side": "本机" if local else "WSL"}
# ---------------------------------------------------------------- 标准 skill 校验
NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
@@ -481,6 +576,7 @@ def decode_input(src: str, work: Path, b64: bool) -> Path | None:
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)))
@@ -657,14 +753,15 @@ def cmd_push(args) -> int:
print(f"dry-run：不会写入 {dest}")
return 0
# 探查式覆盖：没有同名文件夹就直接放（目录和内容都没有，覆盖无从谈起）；
# 确有同名才走 --force —— 「要不要覆盖」这一问只在真的存在时才发生。
if dest.exists():
if not args.force:
raise SystemExit(
f"目标已存在同名 skill：{dest}\n"
f"加 --force 覆盖（覆盖前会自动备份为 .bak-<时间戳>）。"
f"加 --force 覆盖（覆盖前会自动挪进 .trash/ 备份，可恢复）。"
)
backup = dest.with_name(dest.name + f".bak-{_stamp()}")
shutil.move(str(dest), str(backup))
backup = _backup_then_remove(dest)
print(f"已备份旧版本 -> {backup}")
dest.parent.mkdir(parents=True, exist_ok=True)
@@ -679,23 +776,73 @@ def cmd_pull(args) -> int:
gate_err = _wsl_ensure_running(args.distro, "pull")
if gate_err:
raise SystemExit(gate_err)
name = clean_input(args.name) # 粘贴带引号也认
if not name:
raise SystemExit("技能名为空。")
root = wsl_skill_root(args.distro)
src = Path(root) / args.name
src = Path(root) / name
if not src.is_dir():
raise SystemExit(f"WSL 端不存在该 skill：{src}")
out_dir = Path(args.out or ".")
# 输出目录不由用户决定：默认直接落本机 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 / args.name
dest = out_dir / name
# 探查式覆盖：本机没有同名目录就直接放（不该问覆盖）；有才需要 --force
if dest.exists():
if not args.force:
raise SystemExit(f"本地已存在：{dest}（用 --force 覆盖）")
shutil.rmtree(dest)
raise SystemExit(
f"本机已存在同名 skill：{dest}\n"
f"加 --force 覆盖（覆盖前会自动挪进 .trash/ 备份，可恢复）。"
)
backup = _backup_then_remove(dest)
print(f"已备份旧版本 -> {backup}")
shutil.copytree(src, dest)
print(f"已取回：{dest}")
return 0
def cmd_delete(args) -> int:
"""删除 skill（补齐管理闭环：列表 / 推送 / 拉取 / 删除 / 覆盖更新）。
缺省删 WSL 端（与 push 同侧），--local 删本机（与 pull 同侧）。
删除绝不直接 rmtree：先挪进 <skills根>/.trash/<时间戳>/，删错了可恢复。
--yes 跳过交互确认（菜单层已经问过用户时传入）。
"""
name = clean_input(args.name)
if not name:
raise SystemExit("技能名为空。")
# WSL 端是 UNC 访问：先唤醒再读，否则没跑的 VM 会被冷启动拖死
if not args.local:
gate_err = _wsl_ensure_running(args.distro, "delete")
if gate_err:
raise SystemExit(gate_err)
probe = probe_delete(name, args.distro, args.local)
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 cmd_dirs(args) -> int:
print(f"WSL skill 目录（push 目标）：{wsl_skill_root(args.distro)}")
print(f"本机 skill 目录（pull 默认落到当前目录）：{WIN_LOCAL_ROOT}")
@@ -714,6 +861,22 @@ def _stamp() -> str:
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` 这种
@@ -757,7 +920,9 @@ def build_parser() -> argparse.ArgumentParser:
" 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 java-expert --out .\\skills-backup\n"
" python ocskill.py pull java-expert # 直接落本机 OC skill 目录\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}）")
@@ -783,12 +948,18 @@ def build_parser() -> argparse.ArgumentParser:
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 = sub.add_parser("pull", help="把 WSL 里的 skill 拷回本机 OC skill 目录")
sp.add_argument("name", help="skill 名称（可带引号，会自动剥掉）")
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("--yes", "-y", action="store_true", help="跳过交互确认（菜单层已确认时用）")
sp.set_defaults(func=cmd_delete)
sp = sub.add_parser("dirs", help="显示两端 skill 目录与 API 连接状态")
sp.set_defaults(func=cmd_dirs)