367 lines
13 KiB
Python
367 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
基因胶囊 · 打包/解包/列表
|
||
卡若AI 基因胶囊管理脚本，归属土砖。
|
||
"""
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import platform
|
||
import sys
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
# 卡若AI 工作台根目录
|
||
KARUO_AI_ROOT = Path(__file__).resolve().parent.parent.parent.parent.parent
|
||
EXPORT_DIR = Path("/Users/karuo/Documents/卡若Ai的文件夹/导出/基因胶囊")
|
||
CAPSULE_INDEX = Path(__file__).resolve().parent.parent / "capsule_index.json"
|
||
|
||
|
||
def _collect_env():
|
||
"""收集环境指纹（可选）"""
|
||
return {
|
||
"python_version": f"{sys.version_info.major}.{sys.version_info.minor}",
|
||
"platform": platform.system().lower(),
|
||
}
|
||
|
||
|
||
def _calc_capsule_id(skill_path: str, content: str) -> str:
|
||
"""计算资产 ID：SHA256(skill_path + content 前 8192 字符)"""
|
||
data = (skill_path + content[:8192]).encode("utf-8")
|
||
h = hashlib.sha256(data).hexdigest()
|
||
return f"sha256:{h}"
|
||
|
||
|
||
def _find_skill_path(name_or_path: str) -> Path:
|
||
"""根据技能名或路径找到 SKILL.md"""
|
||
if name_or_path.endswith("SKILL.md"):
|
||
p = KARUO_AI_ROOT / name_or_path
|
||
elif os.path.sep in name_or_path or name_or_path.startswith("0"):
|
||
# 形如 05_卡土/土砖_技能复制/技能工厂
|
||
p = KARUO_AI_ROOT / name_or_path
|
||
if p.is_dir():
|
||
p = p / "SKILL.md"
|
||
else:
|
||
# 按名称在 SKILL_REGISTRY 中查找（简化：扫描常见路径）
|
||
registry = KARUO_AI_ROOT / "SKILL_REGISTRY.md"
|
||
if registry.exists():
|
||
text = registry.read_text(encoding="utf-8")
|
||
# 简单匹配：找包含 name_or_path 的 SKILL 路径
|
||
for line in text.splitlines():
|
||
if "SKILL.md" in line and name_or_path in line:
|
||
# 提取路径，如 `05_卡土（土）/土砖_技能复制/技能工厂/SKILL.md`
|
||
start = line.find("`") + 1
|
||
end = line.rfind("`")
|
||
if start > 0 and end > start:
|
||
rel = line[start:end].strip()
|
||
p = KARUO_AI_ROOT / rel
|
||
if p.exists():
|
||
return p
|
||
raise FileNotFoundError(f"未找到技能: {name_or_path}")
|
||
if not p.exists():
|
||
raise FileNotFoundError(f"SKILL 不存在: {p}")
|
||
return p
|
||
|
||
|
||
def _parse_frontmatter(content: str) -> tuple:
|
||
"""解析 YAML frontmatter，返回 (frontmatter_dict, body)。无 PyYAML 时用简单解析。"""
|
||
if not content.strip().startswith("---"):
|
||
return {}, content
|
||
parts = content.split("---", 2)
|
||
if len(parts) < 3:
|
||
return {}, content
|
||
fm = {}
|
||
try:
|
||
import yaml
|
||
fm = yaml.safe_load(parts[1]) or {}
|
||
except Exception:
|
||
for line in parts[1].strip().splitlines():
|
||
if ":" in line:
|
||
k, v = line.split(":", 1)
|
||
k, v = k.strip(), v.strip().strip("'\"").strip()
|
||
if k == "triggers":
|
||
fm[k] = [x.strip() for x in v.replace("、", ",").split(",") if x.strip()]
|
||
else:
|
||
fm[k] = v
|
||
return fm, parts[2].strip()
|
||
|
||
|
||
EXPORT_README = "README_基因胶囊导出说明.md"
|
||
|
||
|
||
def _render_flowchart_mermaid() -> str:
|
||
"""返回基因胶囊功能流程图 Mermaid 源码"""
|
||
return """```mermaid
|
||
flowchart TB
|
||
subgraph internal["卡若AI 内部流程"]
|
||
A1[用户需求/任务] --> A2{查 SKILL_REGISTRY}
|
||
A2 --> A3[读 SKILL.md 执行]
|
||
A3 --> A4[执行完成 + 复盘]
|
||
A4 --> A5{经验有价值?}
|
||
A5 -->|是| A6[写入经验库/待沉淀]
|
||
A5 -->|否| A7[结束]
|
||
A6 --> A8{要打包为胶囊?}
|
||
end
|
||
|
||
subgraph capsule["基因胶囊 核心流程"]
|
||
B1[pack: Skill 转 胶囊JSON] --> B2[导出目录 + 说明文档]
|
||
B2 --> B3[list: 查看本地胶囊]
|
||
B3 --> B4{需要继承?}
|
||
B4 -->|是| B5[unpack: 胶囊 转 Skill]
|
||
B4 -->|否| B6[保留备用]
|
||
B5 --> B7[写入对应成员目录]
|
||
B7 --> B8[更新 SKILL_REGISTRY]
|
||
end
|
||
|
||
subgraph factory["技能工厂联动"]
|
||
C1[创建 Skill 前] --> C2[list 查本地胶囊]
|
||
C2 --> C3{有匹配?}
|
||
C3 -->|是| C4[unpack 继承]
|
||
C3 -->|否| C5[新建 Skill]
|
||
C5 --> C6[创建 Skill 后]
|
||
C6 --> C7[pack 打包为胶囊]
|
||
C4 --> B7
|
||
end
|
||
|
||
subgraph external["未来对外流通"]
|
||
D1[EvoMap Market] --> D2[上传胶囊]
|
||
D2 --> D3[全球 Agent 继承]
|
||
D3 --> B5
|
||
B2 -.->|可选| D2
|
||
end
|
||
|
||
A8 -->|是| B1
|
||
A8 -->|否| A7
|
||
```"""
|
||
|
||
|
||
def _write_export_readme() -> str:
|
||
"""生成/更新导出说明文档（含流程图），返回说明文档路径"""
|
||
EXPORT_DIR.mkdir(parents=True, exist_ok=True)
|
||
readme_path = EXPORT_DIR / EXPORT_README
|
||
|
||
# 收集已导出胶囊列表（按创建时间倒序）
|
||
caps = []
|
||
if EXPORT_DIR.exists():
|
||
for f in sorted(EXPORT_DIR.glob("*.json")):
|
||
try:
|
||
data = json.loads(f.read_text(encoding="utf-8"))
|
||
m = data.get("manifest", {})
|
||
caps.append({
|
||
"file": f.name,
|
||
"name": m.get("name", ""),
|
||
"capsule_id": data.get("capsule_id", "")[:19],
|
||
"created_at": data.get("created_at", ""),
|
||
})
|
||
except Exception:
|
||
pass
|
||
caps.sort(key=lambda x: x.get("created_at", ""), reverse=True)
|
||
|
||
lines = [
|
||
"# 基因胶囊 · 导出说明",
|
||
"",
|
||
"> **生成时间**：{now} ",
|
||
"> **导出目录**：`卡若Ai的文件夹/导出/基因胶囊/` ",
|
||
"> **规范文档**：`运营中枢/参考资料/基因胶囊规范.md`",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 一、基因胶囊功能总览",
|
||
"",
|
||
"**基因胶囊**将验证过的 Skill + 环境指纹 + 审计记录打包为可遗传能力单元，支持：",
|
||
"",
|
||
"- **pack**：Skill → 胶囊 JSON（导出）",
|
||
"- **unpack**：胶囊 JSON → Skill（继承）",
|
||
"- **list**：查看本地所有胶囊",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 二、基因胶囊完整流程图",
|
||
"",
|
||
"以下流程图展示卡若AI 中基因胶囊的完整工作流程、与技能工厂的联动，以及未来与 EvoMap 的流通路径。",
|
||
"",
|
||
_render_flowchart_mermaid(),
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 三、已导出胶囊清单",
|
||
"",
|
||
"| 技能名 | 胶囊文件 | capsule_id | 创建时间 |",
|
||
"|:---|:---|:---|:---|",
|
||
]
|
||
for c in caps:
|
||
lines.append(f"| {c.get('name', '')} | {c.get('file', '')} | {c.get('capsule_id', '')} | {c.get('created_at', '')} |")
|
||
if not caps:
|
||
lines.append("| （暂无） | — | — | — |")
|
||
lines.extend([
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 四、使用方法",
|
||
"",
|
||
"### 解包（继承能力）",
|
||
"",
|
||
"```bash",
|
||
"cd /Users/karuo/Documents/个人/卡若AI",
|
||
'python3 "05_卡土（土）/土砖_技能复制/基因胶囊/脚本/gene_capsule.py" unpack 技能名_xxxx.json',
|
||
'python3 .../gene_capsule.py unpack 技能名_xxxx.json -o 目标目录/ # 指定输出目录',
|
||
"```",
|
||
"",
|
||
"### 列表",
|
||
"",
|
||
"```bash",
|
||
'python3 .../gene_capsule.py list',
|
||
"```",
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 五、引用",
|
||
"",
|
||
"- 规范：`运营中枢/参考资料/基因胶囊规范.md`",
|
||
"- 技能：`05_卡土（土）/土砖_技能复制/基因胶囊/SKILL.md`",
|
||
"- 流程图亦可于规范文档中查看",
|
||
"",
|
||
])
|
||
now = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
content = "\n".join(lines).replace("{now}", now)
|
||
readme_path.write_text(content, encoding="utf-8")
|
||
return str(readme_path)
|
||
|
||
|
||
def pack(skill_ref: str, include_audit: bool = True) -> str:
|
||
"""打包：将 SKILL 转为基因胶囊 JSON，并生成导出说明文档（含流程图）"""
|
||
skill_path = _find_skill_path(skill_ref)
|
||
content = skill_path.read_text(encoding="utf-8")
|
||
rel_path = str(skill_path.relative_to(KARUO_AI_ROOT))
|
||
|
||
fm, body = _parse_frontmatter(content)
|
||
|
||
manifest = {
|
||
"name": fm.get("name", skill_path.stem),
|
||
"description": fm.get("description", ""),
|
||
"triggers": fm.get("triggers", [])
|
||
if isinstance(fm.get("triggers"), list)
|
||
else (fm.get("triggers", "") or "").split("、"),
|
||
"owner": fm.get("owner", ""),
|
||
"group": fm.get("group", ""),
|
||
"skill_path": rel_path,
|
||
}
|
||
capsule_id = _calc_capsule_id(rel_path, content)
|
||
now = datetime.now().strftime("%Y-%m-%dT%H:%M:%S+08:00")
|
||
|
||
capsule = {
|
||
"version": "1.0",
|
||
"capsule_id": capsule_id,
|
||
"manifest": manifest,
|
||
"skill_content": content,
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
"environment": _collect_env(),
|
||
"source": "karuo-ai",
|
||
}
|
||
|
||
if include_audit:
|
||
capsule["audit"] = {"last_retro": "", "source": "pack"}
|
||
|
||
EXPORT_DIR.mkdir(parents=True, exist_ok=True)
|
||
safe_name = manifest["name"].replace(" ", "_").replace("/", "_")[:30]
|
||
out_file = EXPORT_DIR / f"{safe_name}_{capsule_id[7:15]}.json"
|
||
out_file.write_text(json.dumps(capsule, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
# 生成导出说明文档（含完整流程图）
|
||
readme_path = _write_export_readme()
|
||
return str(out_file) + "\n📄 说明文档: " + readme_path
|
||
|
||
|
||
def unpack(capsule_path: str, target_dir: str | None = None) -> str:
|
||
"""解包：将胶囊 JSON 解压为 SKILL.md"""
|
||
p = Path(capsule_path)
|
||
if not p.exists():
|
||
p = EXPORT_DIR / capsule_path
|
||
if not p.exists():
|
||
raise FileNotFoundError(f"胶囊不存在: {capsule_path}")
|
||
|
||
data = json.loads(p.read_text(encoding="utf-8"))
|
||
manifest = data.get("manifest", {})
|
||
skill_content = data.get("skill_content", "")
|
||
|
||
if target_dir:
|
||
dest = Path(target_dir)
|
||
else:
|
||
# 按 manifest.skill_path 写回
|
||
rel = manifest.get("skill_path", "")
|
||
dest = KARUO_AI_ROOT / Path(rel).parent
|
||
dest.mkdir(parents=True, exist_ok=True)
|
||
skill_file = dest / "SKILL.md"
|
||
skill_file.write_text(skill_content, encoding="utf-8")
|
||
|
||
# 更新 capsule_index
|
||
idx = {}
|
||
if CAPSULE_INDEX.exists():
|
||
idx = json.loads(CAPSULE_INDEX.read_text(encoding="utf-8"))
|
||
cid = data.get("capsule_id", "")[:19]
|
||
idx[cid] = {
|
||
"name": manifest.get("name", ""),
|
||
"source": data.get("source", "local"),
|
||
"unpacked_at": dest,
|
||
"created_at": data.get("created_at", ""),
|
||
}
|
||
CAPSULE_INDEX.write_text(json.dumps(idx, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
return str(skill_file)
|
||
|
||
|
||
def list_capsules() -> list[dict]:
|
||
"""列表：扫描导出目录与索引"""
|
||
result = []
|
||
if EXPORT_DIR.exists():
|
||
for f in sorted(EXPORT_DIR.glob("*.json")):
|
||
try:
|
||
data = json.loads(f.read_text(encoding="utf-8"))
|
||
m = data.get("manifest", {})
|
||
result.append({
|
||
"file": f.name,
|
||
"name": m.get("name", ""),
|
||
"capsule_id": data.get("capsule_id", "")[:19],
|
||
"created_at": data.get("created_at", ""),
|
||
})
|
||
except Exception:
|
||
pass
|
||
return result
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="基因胶囊 · pack/unpack/list")
|
||
sub = parser.add_subparsers(dest="cmd", required=True)
|
||
# pack
|
||
p_pack = sub.add_parser("pack", help="将 SKILL 打包为基因胶囊")
|
||
p_pack.add_argument("skill", help="技能路径或名称，如 技能工厂 或 05_卡土（土）/土砖_技能复制/技能工厂/SKILL.md")
|
||
p_pack.add_argument("--no-audit", action="store_true", help="不包含审计信息")
|
||
# unpack
|
||
p_unpack = sub.add_parser("unpack", help="将胶囊解包为 SKILL")
|
||
p_unpack.add_argument("capsule", help="胶囊文件路径或文件名")
|
||
p_unpack.add_argument("-o", "--output", help="输出目录，默认按 manifest.skill_path")
|
||
# list
|
||
p_list = sub.add_parser("list", help="列出本地胶囊")
|
||
args = parser.parse_args()
|
||
|
||
if args.cmd == "pack":
|
||
out = pack(args.skill, include_audit=not args.no_audit)
|
||
parts = out.split("\n")
|
||
print(f"✅ 已打包: {parts[0]}")
|
||
if len(parts) > 1:
|
||
print(parts[1])
|
||
elif args.cmd == "unpack":
|
||
out = unpack(args.capsule, target_dir=args.output)
|
||
print(f"✅ 已解包: {out}")
|
||
elif args.cmd == "list":
|
||
items = list_capsules()
|
||
for i in items:
|
||
print(f" {i['name']} {i['capsule_id']} {i['created_at']}")
|
||
print(f"共 {len(items)} 个胶囊")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|