657 lines
29 KiB
Python
657 lines
29 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
oc.py —— OC 工具箱统一入口（数字 / 中文菜单）
|
||
|
||
启动即打印菜单，输入编号或中文都能进：
|
||
0 / 返回 / 退出 / q 退出
|
||
1 / 技能 / 技能管理 / skill 进入技能管理（列表 / 推送 / 拉取 / 目录）
|
||
2 / 连接器 / 连接器管理 / mcp 进入连接器管理（状态 / 开关 / 模式 / 备份）
|
||
|
||
也支持免菜单直传子命令（等价旧的分散入口）：
|
||
python oc.py skill list
|
||
python oc.py mcp status
|
||
python oc.py list （自动识别归属到技能管理）
|
||
|
||
底层实现复用同目录 ocskill.py / mcpctl.py，本文件只负责交互编排。
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
|
||
# 保证同目录模块可导入（无论从哪个工作目录启动）
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
import mcpctl # noqa: E402
|
||
import ocskill # noqa: E402
|
||
|
||
|
||
# ---------------------------------------------------------------- 控制台
|
||
def fix_console() -> None:
|
||
"""Windows 控制台中文输出兜底（UTF-8 重配）。"""
|
||
for stream_name in ("stdout", "stderr"):
|
||
stream = getattr(sys, stream_name, None)
|
||
reconf = getattr(stream, "reconfigure", None)
|
||
if callable(reconf):
|
||
try:
|
||
reconf(encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ---------------------------------------------------------------- 输入解析
|
||
def _norm(text: str) -> str:
|
||
return (text or "").strip().lower().replace(" ", "")
|
||
|
||
|
||
# 顶层别名：值 -> 传给对应函数的“行为键”
|
||
TOP_ALIASES = {
|
||
"0": "quit", "0.": "quit", "q": "quit", "quit": "quit", "exit": "quit",
|
||
"退出": "quit", "返回": "quit", "回": "quit",
|
||
"1": "skill", "1.": "skill", "skill": "skill", "skills": "skill",
|
||
"技能": "skill", "技能管理": "skill", "管理技能": "skill", "技": "skill",
|
||
"2": "mcp", "2.": "mcp", "mcp": "mcp", "连接器": "mcp",
|
||
"连接器管理": "mcp", "管理连接器": "mcp", "连接": "mcp", "连": "mcp",
|
||
"connector": "mcp", "connectors": "mcp",
|
||
}
|
||
|
||
SKILL_ALIASES = {
|
||
"0": "back", "0.": "back", "b": "back", "back": "back", "返回": "back", "回": "back",
|
||
"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 = {
|
||
"0": "back", "0.": "back", "b": "back", "back": "back", "返回": "back", "回": "back",
|
||
"1": "status", "1.": "status", "status": "status", "状态": "status", "查看": "status",
|
||
"连接器状态": "status",
|
||
"2": "on", "2.": "on", "on": "on", "开": "on", "开启": "on", "打开": "on",
|
||
"3": "off", "3.": "off", "off": "off", "关": "off", "关闭": "off",
|
||
"4": "mode", "4.": "mode", "mode": "mode", "模式": "mode",
|
||
"5": "backup", "5.": "backup", "backup": "backup", "备份": "backup",
|
||
}
|
||
|
||
|
||
# /home 别名集：任意子菜单输入即直接跳回首页（顶层菜单），不再一层层按 0 返回
|
||
HOME_KEYS = {"/home", "home", "首页", "返回首页", "回首页", "回主页", "主菜单", "主页"}
|
||
|
||
|
||
class GoHome(Exception):
|
||
"""在任意位置输入 /home（或「首页」等）—— 一路抛到顶层菜单重绘首页。"""
|
||
|
||
|
||
def ask(prompt: str) -> str:
|
||
try:
|
||
raw = input(prompt)
|
||
except (EOFError, KeyboardInterrupt):
|
||
print()
|
||
return "0"
|
||
if _norm(raw) in HOME_KEYS:
|
||
raise GoHome()
|
||
return raw
|
||
|
||
|
||
def match(raw: str, aliases: dict, default: str | None = None) -> str | None:
|
||
key = _norm(raw)
|
||
if key in aliases:
|
||
return aliases[key]
|
||
# 中文模糊包含：输入“技能管理菜单”也能命中“技能管理”
|
||
for k, v in aliases.items():
|
||
if k and k in key and not k.isdigit():
|
||
return v
|
||
return default
|
||
|
||
|
||
def pause(msg: str = "按回车返回…") -> None:
|
||
try:
|
||
raw = input(msg)
|
||
except (EOFError, KeyboardInterrupt):
|
||
print()
|
||
return
|
||
if _norm(raw) in HOME_KEYS:
|
||
raise GoHome()
|
||
|
||
|
||
def banner(title: str) -> None:
|
||
print()
|
||
print("=" * 44)
|
||
print(" " + title)
|
||
print("=" * 44)
|
||
|
||
|
||
def run_safe(fn, argv: list[str]) -> int:
|
||
"""菜单内调用 ocskill/mcpctl 的统一护栏：
|
||
- KeyboardInterrupt -> 打印“已取消”，回到菜单（不杀进程）
|
||
- SystemExit（push/pull 里的 raise SystemExit）-> 报告信息，
|
||
绝不让它穿透把整个菜单进程结束掉
|
||
- 其它异常 -> 打印错误，回到菜单
|
||
返回子命令退出码（仅直传模式会用到）。"""
|
||
try:
|
||
return fn(argv)
|
||
except KeyboardInterrupt:
|
||
print("\n已取消当前操作，返回菜单。")
|
||
return 130
|
||
except SystemExit as exc:
|
||
code = exc.code
|
||
if code is None:
|
||
return 0
|
||
if isinstance(code, int):
|
||
if code != 0:
|
||
print(f"操作未完成（退出码 {code}），返回菜单。")
|
||
return code
|
||
print(f"操作未完成：{code}，返回菜单。")
|
||
return 1
|
||
except Exception as exc:
|
||
print(f"操作出错：{exc}，返回菜单。")
|
||
return 1
|
||
|
||
|
||
# ---------------------------------------------------------------- 子菜单：技能管理
|
||
def _yes(prompt: str) -> bool:
|
||
"""统一的 y/N 判定（默认 N，回车即取消）。"""
|
||
return match(ask(prompt), {"y": "y", "yes": "y", "是": "y"}, None) == "y"
|
||
|
||
def _pick_agent(prompt: str, word: str) -> dict | None:
|
||
"""选来源/目标终端（宝宝级）：先列可用终端，再收输入。
|
||
回车 = wsl（内置默认）；简称前缀（wb…）；列表序号（1./2.）；或直接贴一整个 skill 目录。
|
||
输入 0/q/取消 返回 None。返回 ocskill.resolve_agent 的解析结果（kind/label/dir）。
|
||
"""
|
||
table = ocskill.load_agent_table()
|
||
print(" 可选终端（agents.md 每行一个，新增一行 = 新增一个终端）：")
|
||
print(" · wsl（内置默认，回车即选）")
|
||
for i, t in enumerate(table, 1):
|
||
print(" %d. %s(%s) %s" % (i, t["key"], t["display"], t["dir"]))
|
||
raw = ocskill.clean_input(ask(prompt))
|
||
if _norm(raw) in ("0", "q", "quit", "取消", "back", "返回", "b"):
|
||
print(" 已取消。")
|
||
return None
|
||
if raw.isdigit() and table and 1 <= int(raw) <= len(table):
|
||
raw = table[int(raw) - 1]["key"]
|
||
res = ocskill.resolve_agent(raw)
|
||
if not res.get("ok"):
|
||
print(" %s" % res.get("err"))
|
||
return None
|
||
print(" %s：%s -> %s" % (word, res["label"], res["dir"]))
|
||
return res
|
||
|
||
|
||
def _gate_unc(root: str, action: str) -> bool:
|
||
"""来源/目标是 WSL UNC 时先唤醒 VM（本机目录完全不碰 WSL）。
|
||
返回 False = 唤醒失败（错误已打印）。"""
|
||
if not ocskill._is_unc_root(root):
|
||
return True
|
||
err = ocskill._wsl_ensure_running(ocskill.DEFAULT_DISTRO, action)
|
||
if err:
|
||
print(" %s" % err)
|
||
return False
|
||
return True
|
||
|
||
|
||
def _confirm_sel(idx: int, names: list[str], root: str, extra: str = "") -> bool:
|
||
"""编号选择的二次确认：打印 skill 名字 + 简介，confirm 执行 / cancel 回退。
|
||
（`N -confirm` 已在解析层跳过本步；这里只处理 `N` / `N.` 的常规确认。）
|
||
"""
|
||
name = names[idx - 1]
|
||
print(" 已选 %d：%s" % (idx, name))
|
||
brief = ocskill.skill_brief(name, root)
|
||
if brief.get("ok"):
|
||
desc = (brief.get("description") or "").replace("\n", " ").strip()
|
||
if desc:
|
||
print(" 简介：%s" % (desc[:100] + ("..." if len(desc) > 100 else "")))
|
||
print(" 路径：%s" % brief.get("path"))
|
||
else:
|
||
print(" %s" % brief.get("err"))
|
||
if extra:
|
||
print(" %s" % extra)
|
||
while True:
|
||
ans = _norm(ask(" 输入 confirm 执行 / cancel 回退: "))
|
||
if ans in ("confirm", "确认", "确定", "yes", "y"):
|
||
return True
|
||
if ans in ("cancel", "取消", "0", "返回", "回", "b", "back", ""):
|
||
print(" 已回退。")
|
||
return False
|
||
print(" 没看懂「%s」：confirm 执行，cancel 回退。" % ans)
|
||
|
||
|
||
def _pull_one(name: str, src: dict) -> None:
|
||
"""按已解析来源拉取一个 skill：探查 -> 无同名直接拉；确有同名才问一次覆盖。"""
|
||
probe = ocskill.probe_pull(name, root=src["dir"])
|
||
if not probe.get("ok"):
|
||
print(" %s" % probe.get("err"))
|
||
return
|
||
print(" 输出：%s（固定，不需选择）" % probe["dest"])
|
||
argv = ["pull", name]
|
||
if src["kind"] != "default":
|
||
argv += ["--from", src["label"]]
|
||
if not probe.get("exists"):
|
||
print(" 本机无同名目录，直接拉取。")
|
||
run_safe(ocskill.main, argv)
|
||
return
|
||
if _yes(" 本机已存在同名 skill「%s」，覆盖？(y/N): " % name):
|
||
print(" 覆盖更新（旧版本先挑进 .trash/ 备份）。")
|
||
run_safe(ocskill.main, argv + ["--force"])
|
||
else:
|
||
print(" 已跳过覆盖，未改动任何文件。")
|
||
|
||
|
||
def _flow_push() -> None:
|
||
"""推送 / 覆盖更新（宝宝级）：选【来源】终端 -> 立即打印该来源 skill 编号列表
|
||
（列表跟随所选目录：选 wb 列 wb，2026-09-24 起不再硬编码本机 OC 目录；带翻页）->
|
||
翻页（下一页/上一页/第N页/next/last，与拉取端同一条 parse_page_cmd，越界拒绝跳转）、
|
||
输编号选技能（1 / 1. 二次确认打印名字+简介，confirm 执行 / cancel 回列表；
|
||
1 -confirm 跳过二次确认）或直接贴路径 / zip / SKILL.md -> 选【目标】终端 ->
|
||
校验规范 -> 探查同名 -> 确有同名才问一次覆盖 -> 写入。
|
||
目标没有同名文件夹时直接写（没有目录和内容，覆盖无从谈起，问了也是白问）。
|
||
来源与目标同目录时拦掉（自己推自己没有意义）。
|
||
"""
|
||
root_pick = _pick_agent(" 来源终端（列它的 skill，回车=wsl，简称如 wb / oc，或贴完整目录，0/q 取消）: ", "来源")
|
||
if root_pick is None:
|
||
return
|
||
if not _gate_unc(root_pick["dir"], "push"):
|
||
return
|
||
# 列表 = 所选来源目录的 skill（选 wb 就列 wb —— 用户协议：选了目录就把 skill 打出来带编号）
|
||
root = root_pick["dir"]
|
||
local_names = ocskill.list_skill_names(root)
|
||
src = None
|
||
if not local_names:
|
||
# 所选目录没列表可打：退化成单次输入（没有编号/翻页可言）
|
||
src = ocskill.clean_input(ask(" 技能输入（路径 / zip / SKILL.md，可带引号，回车取消）: "))
|
||
if not src:
|
||
return
|
||
else:
|
||
print(" 提示：编号 1 / 1. 会打印名字+简介二次确认（confirm 执行 / cancel 回列表），"
|
||
"1 -confirm 跳过确认；翻页：下一页/上一页/第N页/next/last/one page；"
|
||
"也可直接贴路径 / zip / SKILL.md")
|
||
page = 1
|
||
while True:
|
||
# 每次进循环先把当前页（带编号）打出来 —— 含首次进入（目录 = 所选来源）
|
||
page, pages = ocskill.render_page(local_names, page, root)
|
||
cmd = ocskill.clean_input(ask(
|
||
" 技能输入：翻页（下一页/上一页/第N页/next/last）；编号选来源技能；"
|
||
"或贴路径 / zip / SKILL.md；q 取消: "))
|
||
if not cmd:
|
||
print(" 已取消推送（回车=取消；翻页 / 编号 / 路径都能输）。")
|
||
return
|
||
if _norm(cmd) in ("q", "quit", "退出", "取消", "0", "b", "back", "返回"):
|
||
print(" 已取消推送。")
|
||
return
|
||
# 1) 翻页命令（与拉取端同一条 parse_page_cmd：中英双语全支持；越界拒绝跳转）
|
||
np = ocskill.parse_page_cmd(cmd, page, pages)
|
||
if np is not None:
|
||
if not (1 <= np <= pages):
|
||
print(" 共 %d 页，没有第 %d 页（当前停在第 %d 页）。"
|
||
% (pages, np, page))
|
||
continue
|
||
page = np
|
||
continue
|
||
# 2) 编号选择来源 skill（协议：1 / 1. 二次确认，1 -confirm 跳过）
|
||
sel = ocskill.parse_sel(cmd)
|
||
if sel is not None:
|
||
idx, skip = sel
|
||
if not (1 <= idx <= len(local_names)):
|
||
print(" 序号超出范围（1-%d），重输。" % len(local_names))
|
||
continue
|
||
if not (skip or _confirm_sel(idx, local_names, root,
|
||
extra="来源：%s" % root)):
|
||
print(" 已回退到列表，未推送。")
|
||
continue
|
||
src = os.path.join(root, local_names[idx - 1])
|
||
print(" 选定：「%s」 -> %s" % (local_names[idx - 1], src))
|
||
break
|
||
# 3) 其余输入 = 路径 / zip / SKILL.md，交给下面的规范校验去判
|
||
src = cmd
|
||
break
|
||
if not src:
|
||
return
|
||
# 技能选定后才选目标终端（列表已按所选来源打出）
|
||
tgt = _pick_agent(" 目标终端（推送到哪，回车=wsl，简称如 wb / oc，或贴完整目录，0/q 取消）: ", "目标")
|
||
if tgt is None:
|
||
print(" 已取消推送（技能已选好，未写入任何文件）。")
|
||
return
|
||
if not _gate_unc(tgt["dir"], "push"):
|
||
return
|
||
# 来源与目标同目录 = 自己推自己：拦掉（推了等于没推，-f 还可能自覆盖搞坏）
|
||
if os.path.isdir(src) and os.path.normcase(
|
||
os.path.abspath(os.path.dirname(os.path.abspath(src)))) == \
|
||
os.path.normcase(os.path.abspath(tgt["dir"])):
|
||
print(" 来源与目标是同一个目录（%s），自己推自己没有意义，已取消。"
|
||
% os.path.abspath(tgt["dir"]))
|
||
return
|
||
# 1) 先过一遍规范校验（只读探查，不写盘）
|
||
probe = ocskill.probe_push(src, root=tgt["dir"])
|
||
if not probe.get("ok"):
|
||
err = probe.get("err") or ""
|
||
if os.path.isdir(src) and ocskill.find_skill_md(ocskill.Path(src)) is None:
|
||
print(" 该目录缺 SKILL.md（空目录或不完整 skill），无法推送：%s" % src)
|
||
print(" （提示：回列表换一个，或直接贴完整 skill 的路径。）")
|
||
else:
|
||
print(" 校验未通过：%s" % 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"])
|
||
argv = ["push", src]
|
||
if tgt["kind"] != "default":
|
||
argv += ["--to", tgt["label"]]
|
||
# 2) 探查同名：不存在 -> 直接放，不问覆盖
|
||
if not probe.get("exists"):
|
||
print(" 目标无同名目录，直接推送。")
|
||
run_safe(ocskill.main, argv)
|
||
return
|
||
# 3) 确有同名文件夹 -> 这才有资格问一次覆盖
|
||
if _yes(" 已存在同名 skill「%s」，覆盖？(y/N): " % name):
|
||
print(" 覆盖更新（旧版本先挑进 .trash/ 备份）。")
|
||
run_safe(ocskill.main, argv + ["--force"])
|
||
else:
|
||
print(" 已跳过覆盖，未改动任何文件。")
|
||
|
||
|
||
def _flow_pull() -> None:
|
||
"""拉取（宝宝级）：选来源 -> 立即打印带编号的 skill 列表（每页 15 个，全局序号
|
||
跨页连续）—— 记得名字直接输名字；不记得输编号：1 / 1. 进二次确认（打印 skill
|
||
名字 + 简介，confirm 执行 / cancel 回退列表），1 -confirm 跳过二次确认；
|
||
翻页：下一页/上一页/第N页/next/last；q 退出。
|
||
输出固定为本机 OC skill 目录；列表直读文件夹，不启动 OC 服务、不发请求；
|
||
覆盖只在确有同名目录时问一次。
|
||
"""
|
||
src = _pick_agent(" 来源终端（回车=wsl，简称如 wb / oc，或贴完整目录，0/q 取消）: ", "来源")
|
||
if src is None:
|
||
return
|
||
if not _gate_unc(src["dir"], "pull"):
|
||
return
|
||
root = src["dir"]
|
||
names = ocskill.list_skill_names(root)
|
||
if not names:
|
||
# 目录为空：没得列，退化成纯名字输入（名字直拉仍然有效）
|
||
print(" 这个来源下还没有 skill：%s" % root)
|
||
name = ocskill.clean_input(ask(" 技能名（直接输名字试试，回车取消）: "))
|
||
if name:
|
||
_pull_one(name, src)
|
||
return
|
||
print(" 提示：名字直接拉；编号 1 / 1. 会打印名字+简介二次确认（confirm 执行 / "
|
||
"cancel 回退），1 -confirm 跳过确认")
|
||
page = 1
|
||
while True:
|
||
# 选完来源第一次进循环就把列表（带编号）打出来
|
||
page, pages = ocskill.render_page(names, page, root)
|
||
cmd = ocskill.clean_input(ask(
|
||
" 拉取：输名字或序号；翻页：下一页/上一页/第N页/next/last；q 退出: "))
|
||
if not cmd:
|
||
print(" （空输入。名字/序号 = 拉取，下一页/上一页 = 翻页，q = 退出）")
|
||
continue
|
||
if _norm(cmd) in ("q", "quit", "退出", "取消", "0", "b", "back", "返回"):
|
||
print(" 已退出列表，未拉取。")
|
||
return
|
||
# 1) 翻页命令（中英都认：下一页/第N页/上一页、next/last/N page）
|
||
np = ocskill.parse_page_cmd(cmd, page, pages)
|
||
if np is not None:
|
||
if not (1 <= np <= pages):
|
||
print(" 共 %d 页，没有第 %d 页（当前停在第 %d 页）。"
|
||
% (pages, np, page))
|
||
continue
|
||
page = np
|
||
continue
|
||
# 2) 编号选择（协议：1 / 1. 二次确认；1 -confirm 跳过确认）
|
||
sel = ocskill.parse_sel(cmd)
|
||
if sel is not None:
|
||
idx, skip = sel
|
||
if not (1 <= idx <= len(names)):
|
||
print(" 序号超出范围（1-%d），重输。" % len(names))
|
||
continue
|
||
if skip or _confirm_sel(idx, names, root):
|
||
_pull_one(names[idx - 1], src)
|
||
return
|
||
# cancel -> 回退，留在列表重选
|
||
continue
|
||
# 3) 直接输名字（记得名字不需确认，直接拉）
|
||
if ocskill.NAME_RE.match(cmd):
|
||
if ocskill.probe_pull(cmd, root=root).get("ok"):
|
||
_pull_one(cmd, src)
|
||
return
|
||
print(" 来源里没有「%s」，留在列表重输（序号或名字）。" % cmd)
|
||
continue
|
||
print(" 没看懂「%s」：序号/名字拉取，下一页/上一页翻页，q 退出。" % cmd)
|
||
|
||
def _delete_one(name: str, root: str) -> None:
|
||
"""按编号/名字删除一个 skill：先打印将删路径，再走 CLI 删除（自动挑 .trash/ 备份）。"""
|
||
probe = ocskill.probe_delete(name, ocskill.DEFAULT_DISTRO, root=root)
|
||
if not probe.get("ok"):
|
||
print(" %s" % probe.get("err"))
|
||
return
|
||
print(" 将删除%s：%s" % (probe["side"], probe["dest"]))
|
||
print(" （会先挑进 .trash/ 备份，删错了可移回）")
|
||
run_safe(ocskill.main, ["delete", name, "--yes", "--dir", root])
|
||
|
||
|
||
def _flow_delete() -> None:
|
||
"""删除（宝宝级，与推拉同一套编号协议）：选【目标】终端 -> 立即打印该目录
|
||
skill 编号列表（每页 15 个、序号跨页连续，翻页绝对页码越界拒绝）->
|
||
输编号选技能：
|
||
`1 -confirm` 跳过二次确认直接删除（仍先挑 .trash/ 备份，可恢复）；
|
||
`1` / `1.` 打印 skill 名字 + 描述 + 路径，confirm 确认删除 / cancel 取消不动；
|
||
也可直接输名字（删除是危险操作，同样走二次确认）。q 退出。
|
||
"""
|
||
tgt = _pick_agent(" 删哪端（回车=wsl，简称如 wb / oc，或贴完整目录，0/q 取消）: ", "目标")
|
||
if tgt is None:
|
||
return
|
||
if not _gate_unc(tgt["dir"], "delete"):
|
||
return
|
||
root = tgt["dir"]
|
||
names = ocskill.list_skill_names(root)
|
||
if not names:
|
||
print(" 这个目录下还没有 skill：%s" % root)
|
||
return
|
||
print(" 提示：编号 1 / 1. 会打印名字+简介二次确认（confirm 删除 / cancel 回列表），"
|
||
"1 -confirm 直接删除；翻页：下一页/上一页/第N页/next/last；"
|
||
"也可直接输名字；删除先挑 .trash/ 备份，可恢复")
|
||
page = 1
|
||
while True:
|
||
# 每次进循环先把当前页（带编号）打出来 —— 含首次进入（目录 = 所选终端）
|
||
page, pages = ocskill.render_page(names, page, root)
|
||
cmd = ocskill.clean_input(ask(
|
||
" 删除：输编号（1 / 1. 确认，1 -confirm 直删）或名字；"
|
||
"翻页：下一页/上一页/第N页/next/last；q 取消: "))
|
||
if not cmd:
|
||
print(" （空输入。编号/名字 = 删除，下一页/上一页 = 翻页，q = 退出）")
|
||
continue
|
||
if _norm(cmd) in ("q", "quit", "退出", "取消", "0", "b", "back", "返回"):
|
||
print(" 已退出列表，未删除任何 skill。")
|
||
return
|
||
# 1) 翻页命令（与推拉同一条 parse_page_cmd：中英双语；绝对页码越界拒绝跳转）
|
||
np = ocskill.parse_page_cmd(cmd, page, pages)
|
||
if np is not None:
|
||
if not (1 <= np <= pages):
|
||
print(" 共 %d 页，没有第 %d 页（当前停在第 %d 页）。"
|
||
% (pages, np, page))
|
||
continue
|
||
page = np
|
||
continue
|
||
# 2) 编号选择（协议：1 -confirm 直删；1 / 1. 打印名字+简介+路径二次确认）
|
||
sel = ocskill.parse_sel(cmd)
|
||
if sel is not None:
|
||
idx, skip = sel
|
||
if not (1 <= idx <= len(names)):
|
||
print(" 序号超出范围（1-%d），重输。" % len(names))
|
||
continue
|
||
name = names[idx - 1]
|
||
if not (skip or _confirm_sel(idx, names, root,
|
||
extra="删除会先挑进 .trash/ 备份，可恢复")):
|
||
print(" 已取消，这个 skill 没动。")
|
||
continue
|
||
_delete_one(name, root)
|
||
return
|
||
# 3) 直接输名字（删除是危险操作，同样走二次确认，不给静默直删）
|
||
if ocskill.NAME_RE.match(cmd):
|
||
brief = ocskill.skill_brief(cmd, root)
|
||
if not brief.get("ok"):
|
||
print(" %s" % brief.get("err"))
|
||
continue
|
||
print(" 已选：%s" % cmd)
|
||
desc = (brief.get("description") or "").replace("\n", " ").strip()
|
||
if desc:
|
||
print(" 简介：%s" % (desc[:100] + ("..." if len(desc) > 100 else "")))
|
||
print(" 路径：%s" % brief.get("path"))
|
||
print(" 删除会先挑进 .trash/ 备份，可恢复")
|
||
ans = _norm(ask(" 输入 confirm 删除 / cancel 取消: "))
|
||
if ans not in ("confirm", "确认", "确定", "yes", "y"):
|
||
print(" 已取消，这个 skill 没动。")
|
||
continue
|
||
_delete_one(cmd, root)
|
||
return
|
||
print(" 没看懂「%s」：编号/名字删除，下一页/上一页翻页，q 退出。" % cmd)
|
||
|
||
def menu_skill(shared: dict) -> None:
|
||
while True:
|
||
banner("技能管理（skill）")
|
||
print(" 0. 返回上层")
|
||
print(" 1. 查看技能列表（直读目录，与推送同名检查同一数据源）")
|
||
print(" 2. 推送 / 覆盖更新技能（选来源即出编号列表：1 / 1. 二次确认，"
|
||
"1 -confirm 跳过；也可贴路径/zip；再选推送到哪）")
|
||
print(" 3. 拉取技能到本机 OC（选来源即出编号列表：名字直拉；1 / 1. 二次确认，"
|
||
"1 -confirm 跳过确认）")
|
||
print(" 4. 查看两端 skill 目录情况（路径 + skill 数量 + 本机/WSL API 分开探测）")
|
||
print(" 5. 删除技能（选端即出编号列表：1 -confirm 直删；1 / 1. 打印名字+简介"
|
||
"二次确认，confirm 删 / cancel 不动；先挑 .trash/ 备份，可恢复）")
|
||
print(" 提示：任意位置输入 /home 或「返回首页」直接回顶层菜单")
|
||
print(" 覆盖只在确有同名文件夹时才会问你一次")
|
||
raw = ask("请选择（编号或中文，/home 回首页）: ")
|
||
act = match(raw, SKILL_ALIASES)
|
||
if act is None:
|
||
print(" 没看懂「%s」，输入 0-5 或中文，如：列表 / 推送 / 拉取 / 删除" % raw.strip())
|
||
continue
|
||
if act == "back":
|
||
return
|
||
|
||
if act == "list":
|
||
run_safe(ocskill.main, ["list"])
|
||
elif act == "dirs":
|
||
run_safe(ocskill.main, ["dirs"])
|
||
elif act == "push":
|
||
_flow_push()
|
||
elif act == "pull":
|
||
_flow_pull()
|
||
elif act == "delete":
|
||
_flow_delete()
|
||
pause()
|
||
|
||
|
||
# ---------------------------------------------------------------- 子菜单：连接器管理
|
||
def menu_mcp(shared: dict) -> None:
|
||
# 目标（本机 / WSL）本次会话记住，子菜单里随时可换
|
||
target = shared.get("target", "local")
|
||
while True:
|
||
banner("连接器管理（MCP，wb 语境的“连接器”）")
|
||
print(" 当前目标: %s" % ("本机" if target == "local" else "WSL"))
|
||
print(" 0. 返回上层")
|
||
print(" 1. 查看状态（模式 + 开关）")
|
||
print(" 2. 开启连接器（对调为含 MCP 配置 + 热重载）")
|
||
print(" 3. 关闭连接器（对调为安静配置 + 热重载）")
|
||
print(" 4. 查看 / 修改模式（自动 / 手动 / 关闭）")
|
||
print(" 5. 备份当前 mcpcfg.txt")
|
||
print(" t. 切换目标（本机 <-> WSL）")
|
||
print(" 提示：任意位置输入 /home 或「返回首页」直接回顶层菜单")
|
||
raw = ask("请选择（编号或中文，/home 回首页）: ")
|
||
key = _norm(raw)
|
||
if key in ("t", "切换", "target", "目标"):
|
||
target = "wsl" if target == "local" else "local"
|
||
shared["target"] = target
|
||
print(" 已切换目标 -> %s" % ("本机" if target == "local" else "WSL"))
|
||
continue
|
||
act = match(raw, MCP_ALIASES)
|
||
if act is None:
|
||
print(" 没看懂「%s」，输入 0-5 / t 或中文，如：状态 / 开启 / 关闭 / 模式" % raw.strip())
|
||
continue
|
||
if act == "back":
|
||
return
|
||
|
||
base = ["--target", target]
|
||
if act == "status":
|
||
run_safe(mcpctl.main, base + ["status"])
|
||
elif act == "on":
|
||
run_safe(mcpctl.main, base + ["on"])
|
||
elif act == "off":
|
||
run_safe(mcpctl.main, base + ["off"])
|
||
elif act == "backup":
|
||
run_safe(mcpctl.main, base + ["backup"])
|
||
elif act == "mode":
|
||
val = ask(" 新模式（自动 / 手动 / 关闭，回车只查看）: ").strip()
|
||
run_safe(mcpctl.main, base + ["mode"] + ([val] if val else []))
|
||
pause()
|
||
|
||
|
||
# ---------------------------------------------------------------- 顶层菜单
|
||
def menu_top() -> None:
|
||
shared: dict = {"target": "local"}
|
||
while True:
|
||
banner("OC 工具箱（输入编号或中文，0 退出）")
|
||
print(" 0. 退出")
|
||
print(" 1. 技能管理（skill 拉取 / 推送）")
|
||
print(" 2. 连接器管理（MCP 开关 / 模式）")
|
||
print(" 提示：子菜单任意位置输入 /home 或「返回首页」可直接回到本菜单")
|
||
try:
|
||
raw = ask("请选择: ")
|
||
except GoHome:
|
||
print(" 已经在首页了")
|
||
continue
|
||
act = match(raw, TOP_ALIASES)
|
||
if act is None:
|
||
print(" 没看懂「%s」，输入 0 / 1 / 2，或：退出 / 技能管理 / 连接器管理" % raw.strip())
|
||
continue
|
||
if act == "quit":
|
||
print("再见。")
|
||
return
|
||
if act == "skill":
|
||
try:
|
||
menu_skill(shared)
|
||
except GoHome:
|
||
print(" 已返回首页")
|
||
elif act == "mcp":
|
||
try:
|
||
menu_mcp(shared)
|
||
except GoHome:
|
||
print(" 已返回首页")
|
||
|
||
|
||
# ---------------------------------------------------------------- 直传模式
|
||
SKILL_CMDS = {"list", "push", "pull", "dirs", "delete"}
|
||
MCP_CMDS = {"status", "on", "off", "mode", "backup"}
|
||
|
||
|
||
def dispatch(argv: list[str]) -> int:
|
||
"""有参数时免菜单直传（兼容旧分散入口的用法）。"""
|
||
head = _norm(argv[0]) if argv else ""
|
||
rest = argv[1:]
|
||
if head in ("skill", "skills", "技能", "技能管理"):
|
||
return ocskill.main(rest)
|
||
if head in ("mcp", "连接器", "连接器管理"):
|
||
return mcpctl.main(rest)
|
||
if head in SKILL_CMDS:
|
||
return ocskill.main(argv)
|
||
if head in MCP_CMDS:
|
||
return mcpctl.main(argv)
|
||
print("未知命令: %s（技能: %s；连接器: %s）" % (
|
||
argv[0], "/".join(sorted(SKILL_CMDS)), "/".join(sorted(MCP_CMDS))))
|
||
return 2
|
||
|
||
|
||
def main() -> int:
|
||
fix_console()
|
||
if len(sys.argv) > 1:
|
||
return dispatch(sys.argv[1:])
|
||
menu_top()
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|