几个修复:

1. 拉取的名字可选,按每页15个,输入序号,--from选择,--page选择页面;支持中英双语,比如one page next page 下一页、第五页
2. 宝宝及交互,拉取支持先选到终端来源,记得名字输入名字就直接拉,不记得名字输序列号直接拉
3. 新增agents.md,内嵌workbuddy和opencode的skill目录。支持传完整的skill目录压缩包、skill目录或者直接输入wb、oc等。按照名字 键值的队列,括号会忽略。括号里的内容是全称,完整的名字,括号外面的才是真正敲下终端的名字
This commit is contained in:
ypshidifu
2026-09-23 22:35:06 +08:00
parent 3d65720bb3
commit 2f653473ea
4 changed files with 430 additions and 64 deletions
+26
View File
@@ -0,0 +1,26 @@
# agent 终端 skill 目录表
拉取 / 推送 skill 时用「简称」定位目标终端。每行一个终端,两列:
| 名字(全称) | skill 目录 |
## 规则(宽松解析,不校验 markdown 完整性)
- 取「以 `|` 开头」的行,按 `|` 拆分取前两个非空字段;表格没闭合、少分隔符、行首多空格都照样认。
(行首不是 `|` 的行一律不算表格行——本说明里提到的 `|` 不会混进配置。)
- 第一列 = 名字:括号前是**简称**(CLI 输入前缀即可,如 `wb`),括号内是**全称**(CLI 打印用)。
半角 `()` 与全角 `()` 都认,任何形式的括号都忽略。
- 第二列 = skill 目录绝对路径。
- **新增一行表格 = 新增一个 agent 终端**,保存即生效,不用改代码。
- 自动跳过:表头行(第一列/第二列命中 名字、名称、name 与 目录、路径、dir、path 关键词)、
分隔行(字段全是 `-` / `:`)、不含 `|` 的行、缺目录的行。
- 拉取 skill 列表时直接读该目录,**不启动 OC 服务**。
## 内置终端
| 名字 | 目录 |
| --- | --- |
| wb(workbuddy) | C:\Users\ypshidifu\.workbuddy\skills |
| oc(opencode) | C:\Users\ypshidifu\.config\opencode\skills |
> 另有内置默认终端 `wsl`(WSL 里的 OC skill 目录):来源/目标留空回车即选它,不在本表内。
+7 -3
View File
@@ -393,8 +393,11 @@ def _post(
# ---------------------------------------------------------------- 端口发现 + 探活
def _discover_oc_ports() -> list[int]:
"""端口发现:tasklist 找 opencode.exe 的 PID,netstat -ano 找这些 PID 的
LISTENING TCP 端口。OC 每次启动端口随机(实测 49374),写死 4096 必 10061。
"""端口发现:tasklist 找 opencode 系进程的 PID,netstat -ano 找这些 PID 的
LISTENING TCP 端口。OC 每次启动端口随机(实测 49374),写死 4096 必 10061;
--port 4097 启动时默认端口探活扑空,全靠这里补中。
进程名前缀匹配:既要普通 `opencode.exe`,也要 service 形态
`opencode-service-<pid>-<hash>.exe`(tasklist 过滤 eq opencode.exe 会全漏)。
任何一步失败都返回 [](回退默认端口),绝不抛异常。与 ocskill 逐字一致。"""
if os.name != "nt":
return []
@@ -412,7 +415,8 @@ def _discover_oc_ports() -> list[int]:
pids = set()
for line in tl.splitlines():
cols = re.findall(r'"([^"]*)"', line) # "opencode.exe","13748","Console",...
if len(cols) >= 2 and cols[0].lower() == "opencode.exe" and cols[1].isdigit():
# 前缀匹配(非等值):opencode*.exe 覆盖 opencode.exe / opencode-service-*.exe
if len(cols) >= 2 and cols[0].lower().startswith("opencode") and cols[1].isdigit():
pids.add(cols[1])
if not pids:
return []
+122 -26
View File
@@ -158,17 +158,78 @@ def _yes(prompt: str) -> bool:
"""统一的 y/N 判定(默认 N,回车即取消)。"""
return match(ask(prompt), {"y": "y", "yes": "y", "是": "y"}, None) == "y"
def _flow_push() -> None:
"""推送 / 覆盖更新:输入 -> 校验规范 -> 探查同名 -> 确有同名才问一次覆盖 -> 写入。
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
输出固定为 WSL OC 的 skill 目录,不由用户决定。
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 _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:
"""推送 / 覆盖更新(宝宝级):选目标终端 -> 输入 -> 校验规范 -> 探查同名
-> 确有同名才问一次覆盖 -> 写入。
目标没有同名文件夹时直接写(没有目录和内容,覆盖无从谈起,问了也是白问)。
"""
tgt = _pick_agent(" 目标终端(回车=wsl,简称如 wb / oc,或贴完整目录,0/q 取消): ", "目标")
if tgt is None:
return
if not _gate_unc(tgt["dir"], "push"):
return
src = ocskill.clean_input(ask(" 技能输入(路径 / zip / SKILL.md,可带引号,回车取消): "))
if not src:
return
# 1) 先过一遍规范校验(只读探查,不写盘)
probe = ocskill.probe_push(src)
probe = ocskill.probe_push(src, root=tgt["dir"])
if not probe.get("ok"):
print(" 校验未通过:%s" % probe.get("err"))
return
@@ -177,41 +238,76 @@ def _flow_push() -> None:
print(" 校验通过:「%s」" % name)
if desc:
print(" 描述:%s" % (desc[:90] + ("..." if len(desc) > 90 else "")))
print(" 输出:%s(固定,不需选择)" % probe["dest"])
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, ["push", src])
run_safe(ocskill.main, argv)
return
# 3) 确有同名文件夹 -> 这才有资格问一次覆盖
if _yes(" 已存在同名 skill「%s」,覆盖?(y/N): " % name):
print(" 覆盖更新(旧版本先挑进 .trash/ 备份)。")
run_safe(ocskill.main, ["push", src, "--force"])
run_safe(ocskill.main, argv + ["--force"])
else:
print(" 已跳过覆盖,未改动任何文件。")
def _flow_pull() -> None:
"""拉取:输入技能名 -> 探查 WSL 端存在 -> 探查本机同名 -> 确有同名才问一次覆盖 -> 写入。
输出固定为本机 OC 的 skill 目录,不由用户决定。
"""拉取(宝宝级二选一):选来源 -> 记得名字直接拉;不记得回车进分页列表
(每页 15 个,全局序号跨页连续)—— 翻页命令 / 序号 / 直接名字 / q 退出。
输出固定为本机 OC skill 目录;列表直读文件夹,不启动 OC 服务、不发请求;
覆盖只在确有同名目录时问一次。
"""
name = ocskill.clean_input(ask(" 要拉取的技能名(可带引号,回车取消): "))
if not name:
src = _pick_agent(" 来源终端(回车=wsl,简称如 wb / oc,或贴完整目录,0/q 取消): ", "来源")
if src is None:
return
probe = ocskill.probe_pull(name)
if not probe.get("ok"):
print(" %s" % probe.get("err"))
if not _gate_unc(src["dir"], "pull"):
return
print(" 输出:%s(固定,不需选择)" % probe["dest"])
if not probe.get("exists"):
print(" 本机无同名目录,直接拉取。")
run_safe(ocskill.main, ["pull", name])
name = ocskill.clean_input(ask(" 技能名(记得就直接输;不记得回车看列表): "))
if name:
_pull_one(name, src)
return
if _yes(" 本机已存在同名 skill「%s」,覆盖?(y/N): " % name):
print(" 覆盖更新(旧版本先挑进 .trash/ 备份)。")
run_safe(ocskill.main, ["pull", name, "--force"])
else:
print(" 已跳过覆盖,未改动任何文件。")
# ---- 不记得名字:分页列表循环(直读文件夹,不启动 OC 服务)----
root = src["dir"]
names = ocskill.list_skill_names(root)
if not names:
print(" 这个来源下还没有 skill:%s" % root)
return
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:
page = np
continue
# 2) 纯数字 = 全局序号(跨页连续)
if cmd.isdigit():
idx = int(cmd)
if 1 <= idx <= len(names):
_pull_one(names[idx - 1], src)
return
print(" 序号超出范围(1-%d),重输。" % len(names))
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 _flow_delete() -> None:
"""删除:输入技能名 -> 选端 -> 展示将删路径 -> 确认 -> 删除(先挑 .trash/ 可恢复)。"""
@@ -247,8 +343,8 @@ def menu_skill(shared: dict) -> None:
banner("技能管理(skill)")
print(" 0. 返回上层")
print(" 1. 查看技能列表")
print(" 2. 推送 / 覆盖更新技能到 WSL OC(输入即校验,输出固定为 WSL skill 目录)")
print(" 3. 拉取技能到本机 OC(输出固定为本机 skill 目录)")
print(" 2. 推送 / 覆盖更新技能到目标终端(回车=wsl,简称 wb / oc,或贴目录)")
print(" 3. 拉取技能到本机 OC(选来源;不记得名字回车看分页列表,每页 15 个)")
print(" 4. 查看技能目录与连接状态")
print(" 5. 删除技能(先挑 .trash/ 备份,可恢复)")
print(" 提示:任意位置输入 /home 或「返回首页」直接回顶层菜单")
+275 -35
View File
@@ -52,6 +52,13 @@ DEFAULT_DISTRO = os.environ.get("OC_SKILL_DISTRO", "Ubuntu-24.04")
DEFAULT_PORT = int(os.environ.get("OC_PORT", "4096"))
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"
@@ -177,8 +184,11 @@ def _read_server_password(config_dir: Path, distro: str = "") -> str:
def _discover_oc_ports() -> list[int]:
"""端口发现:tasklist 找 opencode.exe 的 PID,netstat -ano 找这些 PID 的
LISTENING TCP 端口。OC 每次启动端口随机(实测 49374),写死 4096 必 10061。
"""端口发现:tasklist 找 opencode 系进程的 PID,netstat -ano 找这些 PID 的
LISTENING TCP 端口。OC 每次启动端口随机(实测 49374),写死 4096 必 10061;
--port 4097 启动时默认端口探活扑空,全靠这里补中。
进程名前缀匹配:既要普通 `opencode.exe`,也要 service 形态
`opencode-service-<pid>-<hash>.exe`(等值匹配 opencode.exe 会全漏)。
任何一步失败都返回 [](调用方回退默认端口),绝不抛异常。"""
if os.name != "nt":
return []
@@ -196,7 +206,8 @@ def _discover_oc_ports() -> list[int]:
pids = set()
for line in tl.splitlines():
cols = re.findall(r'"([^"]*)"', line) # "opencode.exe","13748","Console",...
if len(cols) >= 2 and cols[0].lower() == "opencode.exe" and cols[1].isdigit():
# 前缀匹配(非等值):opencode*.exe 覆盖 opencode.exe / opencode-service-*.exe
if len(cols) >= 2 and cols[0].lower().startswith("opencode") and cols[1].isdigit():
pids.add(cols[1])
if not pids:
return []
@@ -424,6 +435,188 @@ def _frontmatter(text: str) -> dict:
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_page_cmd(text, page: int, total: int) -> int | None:
"""翻页命令 -> 目标页码(已钳制 [1, total]);不是翻页命令返回 None。
中文:下一页 / 上一页 / 第N页 / N页(N 可为中文数词,如 第三页);
英文:next / last / prev / N page(three page / 3 page)。
纯数字不算翻页 —— 留给「按全局序号选择 skill」。
"""
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 clamp(int(m.group(2) or m.group(3)))
m = re.fullmatch(r"第?([一二三四五六七八九十]+)页", s) # 第三页 / 三页
if m and m.group(1) in _CN_NUM:
return clamp(_CN_NUM[m.group(1)])
m = re.fullmatch(r"(\d+|[a-z]+)page", s) # 3 page / three page
if m:
tok = m.group(1)
return clamp(int(tok) if tok.isdigit() else _EN_NUM.get(tok, 0)) \
if (tok.isdigit() or 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 个,序号跨页全局连续。
返回 (实际页码, 总页数)。
"""
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)))
for i, n in enumerate(chunk, start + 1):
print(f" {str(i).rjust(width)}. {n}")
return page, pages
# ---------------------------------------------------------------- 输入净化与探查
_QUOTE_PAIRS = (('"', '"'), ("'", "'"), ("\u201c", "\u201d"), ("\u300c", "\u300d"))
@@ -445,9 +638,11 @@ def clean_input(raw: str) -> str:
return s[1:-1].strip()
return s
def probe_push(src: str, distro: str = DEFAULT_DISTRO, b64: bool = False) -> dict:
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 -> 确有同名文件夹,才有资格问一次「是否覆盖」
@@ -465,7 +660,8 @@ def probe_push(src: str, distro: str = DEFAULT_DISTRO, b64: bool = False) -> dic
if not ok:
return {"ok": False, "err": err}
name = meta.get("name", "")
dest = Path(wsl_skill_root(distro)) / name
dest = Path(root) if root else Path(wsl_skill_root(distro))
dest = dest / name
try:
exists = dest.is_dir()
except OSError:
@@ -473,14 +669,18 @@ def probe_push(src: str, distro: str = DEFAULT_DISTRO, b64: bool = False) -> dic
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 + 本机是否已有同名目录。"""
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(wsl_skill_root(distro)) / 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()
@@ -491,8 +691,9 @@ def probe_pull(name: str, distro: str = DEFAULT_DISTRO) -> dict:
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}
return {"ok": False, "err": f"来源端不存在该 skill:{src}"}
return {"ok": True, "name": name, "dest": str(dest), "exists": local_exists,
"src": str(src)}
def probe_delete(name: str, distro: str = DEFAULT_DISTRO, local: bool = False) -> dict:
"""删除前探查(只读):确认目标存在并给出将删除的路径。"""
@@ -713,21 +914,33 @@ def cmd_list(args) -> int:
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:
# UNC 写之前确保 VM 在跑(否则 \\wsl.localhost 冷启动可能挂死)
gate_err = _wsl_ensure_running(args.distro, "push")
if gate_err:
raise SystemExit(gate_err)
root = wsl_skill_root(args.distro)
# 目标终端:--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"找不到 WSL 的 skill 目录:{root}\n"
f"请确认 WSL 发行版名称(当前 {args.distro}),可用 --distro 指定。"
)
raise SystemExit(f"找不到目标 skill 目录:{root}\n{hint}")
with tempfile.TemporaryDirectory(prefix="ocskill_") as tmp:
work = Path(tmp)
@@ -772,17 +985,35 @@ def cmd_push(args) -> int:
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 在跑(没跑则唤醒,唤醒失败给清晰错误)
gate_err = _wsl_ensure_running(args.distro, "pull")
if gate_err:
raise SystemExit(gate_err)
name = clean_input(args.name) # 粘贴带引号也认
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:
raise SystemExit("技能名为空。")
root = wsl_skill_root(args.distro)
src = Path(root) / name
if not src.is_dir():
raise SystemExit(f"WSL 端不存在该 skill:{src}")
# 宝宝级:不记得名字 -> 打印该来源的 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"])
# 输出目录不由用户决定:默认直接落本机 OC 的 skill 目录(写完即被 watch 热加载)
out_dir = Path(args.out) if args.out else Path(WIN_LOCAL_ROOT)
@@ -797,7 +1028,7 @@ def cmd_pull(args) -> int:
)
backup = _backup_then_remove(dest)
print(f"已备份旧版本 -> {backup}")
shutil.copytree(src, dest)
shutil.copytree(src_dir, dest)
print(f"已取回:{dest}")
return 0
@@ -920,14 +1151,17 @@ 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 # 直接落本机 OC skill 目录\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})")
p.add_argument("--url", default=None, help="显式指定 OC 地址,如 http://localhost:4096")
p.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"OC 端口(默认 {DEFAULT_PORT})")
p.add_argument("--url", default=None, help="显式指定 OC 地址,跳过端口自动发现,如 http://localhost:4097")
p.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"OC 端口(默认 {DEFAULT_PORT};非默认端口会自动发现实际监听)")
p.add_argument("--timeout", type=float, default=4.0, help="HTTP 超时秒数")
p.add_argument("--password", default="", help="OC 口令(默认按 service.json → 环境变量 → opencode.json 顺序自动读)")
p.add_argument(
@@ -941,15 +1175,21 @@ def build_parser() -> argparse.ArgumentParser:
sp = sub.add_parser("list", help="连接 WSL 的 OC,返回 skill 列表")
sp.set_defaults(func=cmd_list)
sp = sub.add_parser("push", help="推送 skill(zip / 目录 / SKILL.md / base64)到 WSL OC")
sp = 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="把 WSL 里的 skill 拷回本机 OC skill 目录")
sp.add_argument("name", help="skill 名称(可带引号,会自动剥掉)")
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)