diff --git a/INSTALL.md b/INSTALL.md index 5389e225..846e5366 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -317,6 +317,10 @@ python3 tools/research/srt_to_transcript.py --help python3 tools/research/merge_research.py --help python3 tools/research/quality_check.py --help +# 测试 Skill 导出/导入 +python3 tools/skill_exporter.py --help +python3 tools/skill_exporter.py export --character colleague --slug zhangsan + # 列出已有同事 Skill python3 tools/skill_writer.py --action list --base-dir ./skills/colleague ``` @@ -337,6 +341,7 @@ colleague-skill/ ← clone 到 .claude/skills/dot-skill/ │ ├── install_codex_skill.py # Codex 本地安装器 │ ├── install_openclaw_generated_skill.py # OpenClaw 角色 Skill 安装器 │ ├── install_codex_generated_skill.py # Codex 角色 Skill 安装器 +│ ├── skill_exporter.py # Skill 导出/导入打包工具 │ └── research/ # celebrity research toolchain ├── docs/ # 文档（PRD 等） │ diff --git a/README.md b/README.md index 756b0135..912eed3b 100644 --- a/README.md +++ b/README.md @@ -193,6 +193,9 @@ Once created, invoke the generated Skill with `/{character}-{slug}`. | `/{character}-{slug}-persona` | Persona only | | `python3 tools/skill_writer.py --action list ...` | List generated Skills across all three families | | `python3 tools/version_manager.py --action rollback ...` | Roll back a Skill version | +| `python3 tools/skill_exporter.py export --character ...` | Package a Skill for sharing | +| `python3 tools/skill_exporter.py import ` | Install a shared Skill package | +| `python3 tools/skill_exporter.py inspect ` | Preview a Skill package contents | ### 🔬 Celebrity Research Toolchain @@ -332,6 +335,7 @@ dot-skill/ │ │ └── quality_check.py # quality check │ ├── install_*_skill.py # [shared] multi-host one-shot installers │ ├── skill_writer.py # [shared] skill file management +│ ├── skill_exporter.py # [shared] skill export/import packaging │ └── version_manager.py # [shared] version archive & rollback ├── skills/ # generated Skills (gitignored) │ ├── colleague/ # colleagues diff --git a/SKILL.md b/SKILL.md index 4d882033..915b0e0c 100644 --- a/SKILL.md +++ b/SKILL.md @@ -65,6 +65,7 @@ allowed-tools: Read, Write, Edit, Bash | 钉钉全自动采集 | `Bash` → `python3 tools/dingtalk_auto_collector.py` | | 解析邮件 .eml/.mbox | `Bash` → `python3 tools/email_parser.py` | | 写入/更新 Skill 文件 | `Write` / `Edit` 工具 | +| 导出/导入 Skill 包 | `Bash` → `python3 tools/skill_exporter.py` | | 版本管理 | `Bash` → `python3 tools/version_manager.py` | | 列出已有 Skill | `Bash` → `python3 tools/skill_writer.py --action list` | diff --git a/tests/test_skill_exporter.py b/tests/test_skill_exporter.py new file mode 100644 index 00000000..6afffa33 --- /dev/null +++ b/tests/test_skill_exporter.py @@ -0,0 +1,555 @@ +from __future__ import annotations + +import json +import os +import sys +import tarfile +import tempfile +import unittest +from pathlib import Path + + +TOOLS_DIR = Path(__file__).resolve().parents[1] / "tools" +if str(TOOLS_DIR) not in sys.path: + sys.path.insert(0, str(TOOLS_DIR)) + +import skill_writer # noqa: E402 +from skill_exporter import ( # noqa: E402 + PACKAGE_EXT, + _build_package_meta, + _scrub_meta, + _validate_package_extension, + export_skill, + import_skill, + inspect_package, +) +from skill_presets import normalize_character # noqa: E402 + + +def _make_skill( + base_dir: Path, + character: str = "colleague", + slug: str = "zhangsan", + **meta_overrides, +) -> Path: + """Create a minimal skill on disk for testing.""" + meta = { + "name": slug.title(), + "profile": {"company": "TestCo", "level": "L3", "role": "Engineer"}, + "tags": {"personality": ["direct"], "culture": ["startup"]}, + **meta_overrides, + } + if "character" not in meta: + meta["character"] = character + return skill_writer.create_skill( + base_dir, + slug, + meta, + "# Work\n\n## Scope\n\nTest work content.\n", + "# Persona\n\n## Layer 0: Core\n\nTest persona content.\n", + ) + + +class ExportTest(unittest.TestCase): + """Tests for skill_exporter.export_skill.""" + + def test_export_creates_tar_gz_with_package_json(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "colleague" + _make_skill(base_dir, slug="zhangsan") + + out = export_skill("colleague", "zhangsan", str(tmp), base_dir=str(base_dir)) + + self.assertTrue(out.exists()) + self.assertTrue(out.name.endswith(PACKAGE_EXT)) + self.assertGreater(out.stat().st_size, 0) + + with tarfile.open(out, "r:gz") as tar: + names = sorted(m.name for m in tar.getmembers()) + self.assertIn("package.json", names) + self.assertIn("skill/meta.json", names) + self.assertIn("skill/SKILL.md", names) + self.assertIn("skill/work.md", names) + self.assertIn("skill/persona.md", names) + self.assertIn("skill/manifest.json", names) + + pkg = json.loads(tar.extractfile("package.json").read()) + self.assertEqual(pkg["package_format"], "1") + self.assertEqual(pkg["slug"], "zhangsan") + self.assertEqual(pkg["character"], "colleague") + + def test_export_knowledge_excluded_by_default(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "colleague" + skill_dir = _make_skill(base_dir, slug="zhangsan") + (skill_dir / "knowledge" / "messages").mkdir(parents=True, exist_ok=True) + (skill_dir / "knowledge" / "messages" / "chat.txt").write_text("hello") + + out = export_skill("colleague", "zhangsan", str(tmp), base_dir=str(base_dir)) + + with tarfile.open(out, "r:gz") as tar: + names = [m.name for m in tar.getmembers()] + self.assertFalse(any("knowledge/" in n for n in names)) + + def test_export_include_knowledge_adds_knowledge_dir(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "colleague" + skill_dir = _make_skill(base_dir, slug="zhangsan") + (skill_dir / "knowledge" / "messages").mkdir(parents=True, exist_ok=True) + (skill_dir / "knowledge" / "messages" / "chat.txt").write_text("hello") + + out = export_skill( + "colleague", "zhangsan", str(tmp), base_dir=str(base_dir), include_knowledge=True + ) + + with tarfile.open(out, "r:gz") as tar: + names = sorted(m.name for m in tar.getmembers()) + knowledge_files = [n for n in names if "knowledge/" in n] + self.assertTrue(len(knowledge_files) > 0) + self.assertIn("skill/knowledge/messages/chat.txt", knowledge_files) + + def test_export_dotfiles_excluded_from_knowledge(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "colleague" + skill_dir = _make_skill(base_dir, slug="zhangsan") + (skill_dir / "knowledge" / "messages").mkdir(parents=True, exist_ok=True) + (skill_dir / "knowledge" / "messages" / "chat.txt").write_text("ok") + (skill_dir / "knowledge" / "messages" / ".DS_Store").write_text("junk") + + out = export_skill( + "colleague", "zhangsan", str(tmp), base_dir=str(base_dir), include_knowledge=True + ) + + with tarfile.open(out, "r:gz") as tar: + names = [m.name for m in tar.getmembers()] + self.assertFalse(any(".DS_Store" in n for n in names)) + self.assertTrue(any("chat.txt" in n for n in names)) + + def test_export_strips_personal_info_by_default(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "colleague" + _make_skill( + base_dir, + slug="zhangsan", + profile={ + "company": "ACME", + "real_name": "Zhang San", + "email": "zs@acme.com", + }, + knowledge_sources=["private_chat_export.json"], + ) + + out = export_skill("colleague", "zhangsan", str(tmp), base_dir=str(base_dir)) + + with tarfile.open(out, "r:gz") as tar: + meta = json.loads(tar.extractfile("skill/meta.json").read()) + profile = meta.get("profile", {}) + self.assertEqual(profile.get("real_name"), "[redacted]") + self.assertEqual(profile.get("email"), "[redacted]") + self.assertEqual(meta.get("knowledge_sources"), "[redacted]") + self.assertEqual(profile.get("company"), "ACME") # not PII + + def test_export_no_strip_personal_preserves_fields(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "colleague" + _make_skill( + base_dir, + slug="zhangsan", + profile={"real_name": "Zhang San", "company": "ACME"}, + knowledge_sources=["chat.json"], + ) + + out = export_skill( + "colleague", "zhangsan", str(tmp), base_dir=str(base_dir), strip_personal=False + ) + + with tarfile.open(out, "r:gz") as tar: + meta = json.loads(tar.extractfile("skill/meta.json").read()) + self.assertEqual(meta["profile"]["real_name"], "Zhang San") + self.assertIn("chat.json", meta.get("knowledge_sources", [])) + + def test_export_include_versions_adds_versions_dir(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "colleague" + skill_dir = _make_skill(base_dir, slug="zhangsan") + versions = skill_dir / "versions" / "v1" + versions.mkdir(parents=True) + (versions / "work.md").write_text("v1 work") + + out = export_skill( + "colleague", "zhangsan", str(tmp), base_dir=str(base_dir), include_versions=True + ) + + with tarfile.open(out, "r:gz") as tar: + names = [m.name for m in tar.getmembers()] + self.assertTrue(any("versions/v1/work.md" in n for n in names)) + + def test_export_relationship_skill(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "relationship" + _make_skill( + base_dir, + character="relationship", + slug="mireille", + profile={"role": "Partner"}, + ) + + out = export_skill("relationship", "mireille", str(tmp), base_dir=str(base_dir)) + + with tarfile.open(out, "r:gz") as tar: + pkg = json.loads(tar.extractfile("package.json").read()) + self.assertEqual(pkg["character"], "relationship") + meta = json.loads(tar.extractfile("skill/meta.json").read()) + self.assertEqual(meta["character"], "relationship") + + def test_export_celebrity_skill(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "celebrity" + _make_skill( + base_dir, + character="celebrity", + slug="karpathy", + profile={"identity": "AI Researcher"}, + ) + + out = export_skill("celebrity", "karpathy", str(tmp), base_dir=str(base_dir)) + + with tarfile.open(out, "r:gz") as tar: + pkg = json.loads(tar.extractfile("package.json").read()) + self.assertEqual(pkg["character"], "celebrity") + self.assertEqual(pkg["research_profile"], "budget-friendly") + + def test_export_nonexistent_skill_raises(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(FileNotFoundError): + export_skill("colleague", "nonexistent", str(tmp)) + + def test_export_output_to_directory_creates_default_filename(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp).resolve() + base_dir = tmp / "skills" / "colleague" + _make_skill(base_dir, slug="zhangsan") + + out = export_skill("colleague", "zhangsan", str(tmp), base_dir=str(base_dir)) + + self.assertEqual(out.parent, tmp) + self.assertEqual(out.name, f"zhangsan{PACKAGE_EXT}") + + def test_export_output_specific_filename(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "colleague" + _make_skill(base_dir, slug="zhangsan") + + out = export_skill( + "colleague", "zhangsan", str(tmp / "my-export.skill.tar.gz"), base_dir=str(base_dir) + ) + + self.assertEqual(out.name, "my-export.skill.tar.gz") + + +class ImportTest(unittest.TestCase): + """Tests for skill_exporter.import_skill.""" + + def test_import_restores_skill_to_target_directory(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + src_base = tmp / "src" / "skills" / "colleague" + _make_skill(src_base, slug="zhangsan") + pkg = export_skill("colleague", "zhangsan", str(tmp / "export"), base_dir=str(src_base)) + + target = import_skill(str(pkg), base_dir=str(tmp / "dst" / "skills" / "colleague")) + + self.assertTrue(target.exists()) + self.assertTrue((target / "SKILL.md").exists()) + self.assertTrue((target / "work.md").exists()) + self.assertTrue((target / "persona.md").exists()) + self.assertTrue((target / "meta.json").exists()) + self.assertTrue((target / "manifest.json").exists()) + self.assertTrue((target / ".dot-skill-install.json").exists()) + + def test_import_preserves_skill_content(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + src_base = tmp / "src" / "skills" / "colleague" + _make_skill(src_base, slug="zhangsan") + pkg = export_skill("colleague", "zhangsan", str(tmp / "export"), base_dir=str(src_base)) + + target = import_skill(str(pkg), base_dir=str(tmp / "dst" / "skills" / "colleague")) + work = (target / "work.md").read_text() + + self.assertIn("Test work content", work) + + def test_import_conflict_without_force_raises(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + dst_base = tmp / "dst" / "skills" / "colleague" + _make_skill(dst_base, slug="zhangsan") + pkg = export_skill("colleague", "zhangsan", str(tmp / "export"), base_dir=str(dst_base)) + + with self.assertRaises(FileExistsError): + import_skill(str(pkg), base_dir=str(dst_base), force=False) + + def test_import_force_overwrites_existing_skill(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + dst_base = tmp / "dst" / "skills" / "colleague" + _make_skill(dst_base, slug="zhangsan") + pkg = export_skill("colleague", "zhangsan", str(tmp / "export"), base_dir=str(dst_base)) + + # Should not raise + target = import_skill(str(pkg), base_dir=str(dst_base), force=True) + self.assertTrue(target.exists()) + + def test_import_relationship_skill(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + src_base = tmp / "src" / "skills" / "relationship" + _make_skill( + src_base, + character="relationship", + slug="mireille", + profile={"role": "Partner"}, + ) + pkg = export_skill("relationship", "mireille", str(tmp / "export"), base_dir=str(src_base)) + + target = import_skill( + str(pkg), base_dir=str(tmp / "dst" / "skills" / "relationship") + ) + + meta = json.loads((target / "meta.json").read_text()) + self.assertEqual(meta["character"], "relationship") + + def test_import_celebrity_skill(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + src_base = tmp / "src" / "skills" / "celebrity" + _make_skill( + src_base, + character="celebrity", + slug="karpathy", + ) + pkg = export_skill("celebrity", "karpathy", str(tmp / "export"), base_dir=str(src_base)) + + target = import_skill( + str(pkg), base_dir=str(tmp / "dst" / "skills" / "celebrity") + ) + + meta = json.loads((target / "meta.json").read_text()) + self.assertEqual(meta["character"], "celebrity") + + def test_import_nonexistent_package_raises(self) -> None: + with self.assertRaises(FileNotFoundError): + import_skill("/tmp/nonexistent.skill.tar.gz") + + def test_import_wrong_extension_raises(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + fake = Path(tmp) / "not-a-package.zip" + fake.write_text("junk") + with self.assertRaises(ValueError): + import_skill(str(fake)) + + def test_import_nonexistent_base_dir_creates_it(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + src_base = tmp / "src" / "skills" / "colleague" + _make_skill(src_base, slug="zhangsan") + pkg = export_skill("colleague", "zhangsan", str(tmp / "export"), base_dir=str(src_base)) + + new_base = tmp / "completely" / "new" / "path" + target = import_skill(str(pkg), base_dir=str(new_base)) + + self.assertTrue(target.exists()) + + +class InspectTest(unittest.TestCase): + """Tests for skill_exporter.inspect_package.""" + + def test_inspect_returns_structured_summary(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "colleague" + _make_skill(base_dir, slug="zhangsan") + pkg = export_skill("colleague", "zhangsan", str(tmp / "export"), base_dir=str(base_dir)) + + info = inspect_package(str(pkg)) + + self.assertEqual(info["character"], "colleague") + self.assertEqual(info["slug"], "zhangsan") + self.assertEqual(info["package_format"], "1") + self.assertTrue(len(info["artifacts"]) > 0) + self.assertIn("skill/SKILL.md", info["artifacts"]) + self.assertGreater(info["size_bytes"], 0) + self.assertFalse(info["has_knowledge"]) + + def test_inspect_shows_knowledge_when_included(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "colleague" + skill_dir = _make_skill(base_dir, slug="zhangsan") + (skill_dir / "knowledge" / "messages").mkdir(parents=True, exist_ok=True) + (skill_dir / "knowledge" / "messages" / "chat.txt").write_text("hi") + + pkg = export_skill( + "colleague", "zhangsan", str(tmp / "export"), base_dir=str(base_dir), include_knowledge=True + ) + info = inspect_package(str(pkg)) + + self.assertTrue(info["has_knowledge"]) + self.assertTrue(any("knowledge/" in a for a in info["artifacts"])) + + def test_inspect_nonexistent_package_raises(self) -> None: + with self.assertRaises(FileNotFoundError): + inspect_package("/tmp/nonexistent.skill.tar.gz") + + +class ScrubMetaTest(unittest.TestCase): + """Tests for _scrub_meta PII redaction.""" + + def test_scrub_redacts_pii_fields(self) -> None: + meta = { + "profile": { + "real_name": "Alice", + "email": "alice@example.com", + "phone": "555-1234", + "company": "ACME", + }, + "knowledge_sources": ["chat.json", "emails.mbox"], + } + + scrubbed = _scrub_meta(meta) + + self.assertEqual(scrubbed["profile"]["real_name"], "[redacted]") + self.assertEqual(scrubbed["profile"]["email"], "[redacted]") + self.assertEqual(scrubbed["profile"]["phone"], "[redacted]") + self.assertEqual(scrubbed["profile"]["company"], "ACME") + self.assertEqual(scrubbed["knowledge_sources"], "[redacted]") + + def test_scrub_handles_missing_fields_gracefully(self) -> None: + meta = {"profile": {"company": "ACME"}} + scrubbed = _scrub_meta(meta) + self.assertEqual(scrubbed["profile"]["company"], "ACME") + + def test_scrub_adds_exported_at(self) -> None: + meta = {"profile": {}} + scrubbed = _scrub_meta(meta) + self.assertIn("exported_at", scrubbed.get("lifecycle", {})) + + +class ValidateExtensionTest(unittest.TestCase): + """Tests for _validate_package_extension.""" + + def test_skill_tar_gz_passes(self) -> None: + _validate_package_extension(Path("my-skill.skill.tar.gz")) + + def test_skill_tar_passes(self) -> None: + _validate_package_extension(Path("my-skill.skill.tar")) + + def test_plain_tar_gz_fails(self) -> None: + with self.assertRaises(ValueError): + _validate_package_extension(Path("my-skill.tar.gz")) + + def test_zip_fails(self) -> None: + with self.assertRaises(ValueError): + _validate_package_extension(Path("my-skill.zip")) + + +class RoundTripTest(unittest.TestCase): + """End-to-end export → import round-trip tests.""" + + def test_round_trip_colleague(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + src_base = tmp / "src" / "skills" / "colleague" + _make_skill(src_base, slug="zhangsan") + pkg = export_skill("colleague", "zhangsan", str(tmp / "export"), base_dir=str(src_base)) + + target = import_skill( + str(pkg), base_dir=str(tmp / "dst" / "skills" / "colleague") + ) + + work = (target / "work.md").read_text() + persona = (target / "persona.md").read_text() + self.assertIn("Test work content", work) + self.assertIn("Test persona content", persona) + + def test_round_trip_with_knowledge(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + src_base = tmp / "src" / "skills" / "colleague" + skill_dir = _make_skill(src_base, slug="zhangsan") + (skill_dir / "knowledge" / "docs").mkdir(parents=True, exist_ok=True) + (skill_dir / "knowledge" / "docs" / "design.md").write_text("# Design Doc") + + pkg = export_skill( + "colleague", "zhangsan", str(tmp / "export"), base_dir=str(src_base), include_knowledge=True + ) + target = import_skill( + str(pkg), base_dir=str(tmp / "dst" / "skills" / "colleague") + ) + + doc = target / "knowledge" / "docs" / "design.md" + self.assertTrue(doc.exists()) + self.assertIn("Design Doc", doc.read_text()) + + def test_round_trip_all_families(self) -> None: + """Export and re-import one skill from each family.""" + families = [ + ("colleague", "zhangsan", {}), + ("relationship", "mireille", {"profile": {"role": "Partner"}}), + ("celebrity", "karpathy", {"profile": {"identity": "Researcher"}}), + ] + + for character, slug, extra_meta in families: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + src_base = tmp / "src" / "skills" / character + _make_skill(src_base, character=character, slug=slug, **extra_meta) + pkg = export_skill(character, slug, str(tmp / "export"), base_dir=str(src_base)) + target = import_skill( + str(pkg), base_dir=str(tmp / "dst" / "skills" / character) + ) + + meta = json.loads((target / "meta.json").read_text()) + self.assertEqual(meta["character"], character, f"Failed for {character}") + self.assertTrue((target / "SKILL.md").exists(), f"No SKILL.md for {character}") + + +class PackageMetaTest(unittest.TestCase): + """Tests for _build_package_meta.""" + + def test_build_package_meta_includes_key_fields(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + base_dir = tmp / "skills" / "celebrity" + _make_skill( + base_dir, + character="celebrity", + slug="karpathy", + profile={"identity": "Researcher"}, + ) + meta = json.loads( + (base_dir / "karpathy" / "meta.json").read_text() + ) + + pkg = _build_package_meta(base_dir / "karpathy", meta) + + self.assertEqual(pkg["package_format"], "1") + self.assertEqual(pkg["character"], "celebrity") + self.assertEqual(pkg["slug"], "karpathy") + self.assertEqual(pkg["schema_version"], "3") + self.assertIn("exported_at", pkg) + self.assertIn("dot_skill_version", pkg) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/skill_exporter.py b/tools/skill_exporter.py new file mode 100644 index 00000000..19b42e1b --- /dev/null +++ b/tools/skill_exporter.py @@ -0,0 +1,476 @@ +#!/usr/bin/env python3 +""" +Skill export / import tool for the dot-skill engine. + +Packages a generated skill into a portable .skill.tar.gz archive and +installs a packaged skill into a local skills directory. Supports +privacy controls so users can share skills without leaking raw chat +logs or personal identifiers. + +Actions +------- +export Package a skill into a .skill.tar.gz +import Install a .skill.tar.gz into the local skills tree +inspect Show the contents of a .skill.tar.gz without installing +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import shutil +import sys +import tarfile +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + +from skill_presets import ( + get_character_preset, + normalize_character, + resolve_existing_storage_root, + resolve_storage_root, +) +from skill_schema import PRIMARY_ARTIFACTS, enrich_skill_meta, now_iso + + +PACKAGE_FORMAT_VERSION = "1" +PACKAGE_EXT = ".skill.tar.gz" + +# Fields scrubbed from meta.json when --strip-personal is active. +PII_FIELDS = [ + "profile.real_name", + "profile.email", + "profile.phone", + "profile.wechat", + "profile.company_id", + "knowledge_sources", +] + + +def _resolve_skill_dir( + character: str, + slug: str, + base_dir: str | None = None, +) -> Path: + """Return the on-disk skill directory for *character* / *slug*.""" + root = resolve_existing_storage_root(character, slug, base_dir_arg=base_dir) + return root / slug + + +def _load_meta(skill_dir: Path) -> dict: + """Read and lightly normalise meta.json from *skill_dir*.""" + meta_path = skill_dir / "meta.json" + if not meta_path.exists(): + raise FileNotFoundError(f"meta.json not found in {skill_dir}") + return json.loads(meta_path.read_text(encoding="utf-8")) + + +def _scrub_meta(meta: dict) -> dict: + """Return a deep copy of *meta* with PII fields redacted.""" + import copy + + scrubbed = copy.deepcopy(meta) + + for dotted in PII_FIELDS: + parts = dotted.split(".") + container = scrubbed + for i, key in enumerate(parts): + if i == len(parts) - 1: + if key in container: + container[key] = "[redacted]" + else: + container = container.get(key, {}) + if not isinstance(container, dict): + break + + scrubbed.setdefault("lifecycle", {})["exported_at"] = now_iso() + return scrubbed + + +def _build_package_meta(skill_dir: Path, meta: dict) -> dict: + """Build the top-level package.json metadata for an export.""" + return { + "package_format": PACKAGE_FORMAT_VERSION, + "exported_at": now_iso(), + "schema_version": meta.get("schema_version", "3"), + "dot_skill_version": "1.0.0", + "character": meta.get("character", "colleague"), + "slug": meta.get("slug", skill_dir.name), + "display_name": meta.get("display_name", skill_dir.name), + "preset": meta.get("preset", ""), + "research_profile": meta.get("research_profile", "standard"), + "source_character": meta.get("character", "colleague"), + } + + +def export_skill( + character: str, + slug: str, + output: str, + *, + base_dir: str | None = None, + include_knowledge: bool = False, + strip_personal: bool = True, + include_versions: bool = False, +) -> Path: + """Package *character/slug* into a .skill.tar.gz at *output*.""" + character = normalize_character(character) + skill_dir = _resolve_skill_dir(character, slug, base_dir) + if not skill_dir.exists(): + raise FileNotFoundError(f"Skill directory not found: {skill_dir}") + + meta = _load_meta(skill_dir) + enriched = enrich_skill_meta(meta, slug, character) + package_meta = _build_package_meta(skill_dir, enriched) + + output_path = Path(output).expanduser().resolve() + if output_path.is_dir(): + output_path = output_path / f"{slug}{PACKAGE_EXT}" + if output_path.suffix == ".gz" and output_path.stem.endswith(".skill.tar"): + pass + elif output_path.suffix == ".tar": + output_path = output_path.with_suffix(PACKAGE_EXT) + else: + output_path = output_path.with_suffix(PACKAGE_EXT) + + output_path.parent.mkdir(parents=True, exist_ok=True) + + export_meta = _scrub_meta(enriched) if strip_personal else enriched + + with tarfile.open(output_path, "w:gz") as tar: + # -- package.json ------------------------------------------------- + pkg_json = json.dumps(package_meta, indent=2, ensure_ascii=False) + pkg_info = tarfile.TarInfo(name="package.json") + pkg_info.size = len(pkg_json.encode("utf-8")) + tar.addfile(pkg_info, _bytesio(pkg_json)) + + # -- meta.json ----------------------------------------------------- + meta_json = json.dumps(export_meta, indent=2, ensure_ascii=False) + meta_info = tarfile.TarInfo(name="skill/meta.json") + meta_info.size = len(meta_json.encode("utf-8")) + tar.addfile(meta_info, _bytesio(meta_json)) + + # -- primary artifacts -------------------------------------------- + for artifact_name in PRIMARY_ARTIFACTS: + artifact_path = skill_dir / artifact_name + if artifact_path.exists(): + tar.add(artifact_path, arcname=f"skill/{artifact_name}") + + # -- knowledge ---------------------------------------------------- + knowledge_dir = skill_dir / "knowledge" + if include_knowledge and knowledge_dir.exists(): + _add_dir_filtered(tar, knowledge_dir, "skill/knowledge") + + # -- versions ----------------------------------------------------- + versions_dir = skill_dir / "versions" + if include_versions and versions_dir.exists(): + _add_dir_filtered(tar, versions_dir, "skill/versions") + + # -- manifest.json (re-built if missing) -------------------------- + manifest_path = skill_dir / "manifest.json" + if not manifest_path.exists(): + from skill_schema import build_manifest + + manifest_body = json.dumps( + build_manifest(enriched), indent=2, ensure_ascii=False + ) + m_info = tarfile.TarInfo(name="skill/manifest.json") + m_info.size = len(manifest_body.encode("utf-8")) + tar.addfile(m_info, _bytesio(manifest_body)) + + return output_path + + +def import_skill( + package_path: str, + *, + base_dir: str | None = None, + force: bool = False, + install_host: str | None = None, +) -> Path: + """Install a .skill.tar.gz into the local skills tree.""" + package = Path(package_path).expanduser().resolve() + if not package.exists(): + raise FileNotFoundError(f"Package not found: {package}") + + _validate_package_extension(package) + + with tarfile.open(package, "r:gz") as tar: + pkg_meta = _read_package_meta(tar) + meta = _read_skill_meta(tar) + _validate_package_compat(pkg_meta) + + character = normalize_character( + pkg_meta.get("character") or meta.get("character") + ) + slug = pkg_meta.get("slug") or meta.get("slug", "") + if not slug: + raise ValueError("Package is missing a slug — cannot determine target directory") + + target_root = resolve_storage_root(character, base_dir) + target_dir = target_root / slug + + if target_dir.exists() and not force: + raise FileExistsError( + f"Skill already exists at {target_dir}. Use --force to overwrite." + ) + + if target_dir.exists(): + shutil.rmtree(target_dir) + + target_dir.mkdir(parents=True, exist_ok=True) + + _extract_skill_members(tar, target_dir) + + _write_install_metadata(target_dir, pkg_meta) + + if install_host: + _install_to_host(target_dir, meta, install_host) + + return target_dir + + +def inspect_package(package_path: str) -> dict: + """Return a structured summary of a .skill.tar.gz without installing.""" + package = Path(package_path).expanduser().resolve() + if not package.exists(): + raise FileNotFoundError(f"Package not found: {package}") + _validate_package_extension(package) + + with tarfile.open(package, "r:gz") as tar: + pkg_meta = _read_package_meta(tar) + members = sorted(m.name for m in tar.getmembers()) + + try: + meta = _read_skill_meta(tar) + except (KeyError, json.JSONDecodeError): + meta = {} + + has_knowledge = any("knowledge/" in m for m in members) + has_versions = any("versions/" in m for m in members) + + return { + "package_format": pkg_meta.get("package_format"), + "exported_at": pkg_meta.get("exported_at"), + "character": pkg_meta.get("character") or meta.get("character"), + "slug": pkg_meta.get("slug") or meta.get("slug"), + "display_name": pkg_meta.get("display_name") or meta.get("display_name"), + "preset": pkg_meta.get("preset") or meta.get("preset"), + "research_profile": pkg_meta.get("research_profile") or meta.get("research_profile"), + "schema_version": pkg_meta.get("schema_version") or meta.get("schema_version"), + "artifacts": [m for m in members if m.startswith("skill/")], + "has_knowledge": has_knowledge, + "has_versions": has_versions, + "size_bytes": package.stat().st_size, + "summary": meta.get("summary", ""), + } + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +_PKG_EXT_RE = re.compile(r"\.skill\.tar(\.gz)?$", re.IGNORECASE) + + +def _validate_package_extension(path: Path) -> None: + if not _PKG_EXT_RE.search(path.name): + raise ValueError( + f"Expected a .skill.tar.gz file, got: {path.name}" + ) + + +def _bytesio(content: str): + """Return a BytesIO wrapping *content* encoded as UTF-8.""" + from io import BytesIO + + return BytesIO(content.encode("utf-8")) + + +def _read_package_meta(tar: tarfile.TarFile) -> dict: + """Extract and parse package.json from *tar*.""" + member = tar.getmember("package.json") + f = tar.extractfile(member) + if f is None: + raise ValueError("package.json is empty in archive") + return json.loads(f.read().decode("utf-8")) + + +def _read_skill_meta(tar: tarfile.TarFile) -> dict: + """Extract and parse skill/meta.json from *tar*.""" + member = tar.getmember("skill/meta.json") + f = tar.extractfile(member) + if f is None: + raise KeyError("skill/meta.json missing in archive") + return json.loads(f.read().decode("utf-8")) + + +def _validate_package_compat(pkg_meta: dict) -> None: + fmt = pkg_meta.get("package_format") + if fmt != PACKAGE_FORMAT_VERSION: + raise ValueError( + f"Unsupported package format {fmt!r}. " + f"This tool supports format {PACKAGE_FORMAT_VERSION}." + ) + + +def _extract_skill_members(tar: tarfile.TarFile, target_dir: Path) -> None: + """Extract skill/ members into *target_dir*, stripping the prefix.""" + for member in tar.getmembers(): + if not member.name.startswith("skill/"): + continue + rel = member.name[len("skill/"):] + if not rel: + continue + member.name = rel + tar.extract(member, target_dir, filter="data") + + +def _write_install_metadata(target_dir: Path, pkg_meta: dict) -> None: + """Write .dot-skill-install.json into *target_dir*.""" + install_meta = { + "installed_at": now_iso(), + "source_package_format": pkg_meta.get("package_format"), + "source_exported_at": pkg_meta.get("exported_at"), + } + (target_dir / ".dot-skill-install.json").write_text( + json.dumps(install_meta, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + + +def _install_to_host(skill_dir: Path, meta: dict, host: str) -> None: + """Delegate to the host-specific installer.""" + host = host.strip().lower() + if host in ("claude", "claude-code"): + from install_claude_generated_skill import install_claude_generated_skill + + install_claude_generated_skill(skill_dir, meta, force=True) + elif host == "openclaw": + from install_openclaw_generated_skill import install_openclaw_generated_skill + + install_openclaw_generated_skill(skill_dir, meta, force=True) + elif host == "codex": + from install_codex_generated_skill import install_codex_generated_skill + + install_codex_generated_skill(skill_dir, meta, force=True) + else: + print(f"Warning: unknown host {host!r} — skipping host install", file=sys.stderr) + + +def _add_dir_filtered( + tar: tarfile.TarFile, + src: Path, + arc_prefix: str, +) -> None: + """Add *src* directory to *tar* under *arc_prefix*, skipping hidden files.""" + for root, dirs, files in os.walk(src): + dirs[:] = [d for d in dirs if not d.startswith(".")] + for fname in files: + if fname.startswith("."): + continue + full = Path(root) / fname + arcname = f"{arc_prefix}/{full.relative_to(src)}" + tar.add(full, arcname=arcname) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Package and share dot-skill generated skills." + ) + sub = parser.add_subparsers(dest="action", required=True) + + # -- export ----------------------------------------------------------- + exp = sub.add_parser("export", help="Package a skill into a .skill.tar.gz") + exp.add_argument("--character", required=True, help="Character family (colleague|relationship|celebrity)") + exp.add_argument("--slug", required=True, help="Skill slug") + exp.add_argument("--output", default=".", help="Output path or directory (default: .)") + exp.add_argument("--base-dir", default=None, help="Storage root override") + exp.add_argument( + "--include-knowledge", + action="store_true", + default=False, + help="Include raw knowledge/ materials (excluded by default for privacy)", + ) + exp.add_argument( + "--no-strip-personal", + action="store_true", + default=False, + help="Keep personal identifiers in meta.json (redacted by default)", + ) + exp.add_argument( + "--include-versions", + action="store_true", + default=False, + help="Include version history (excluded by default)", + ) + + # -- import ----------------------------------------------------------- + imp = sub.add_parser("import", help="Install a .skill.tar.gz") + imp.add_argument("package", help="Path to .skill.tar.gz file") + imp.add_argument("--base-dir", default=None, help="Storage root override") + imp.add_argument( + "--force", + action="store_true", + default=False, + help="Overwrite existing skill directory", + ) + imp.add_argument( + "--install-host", + default=None, + choices=["claude", "openclaw", "codex"], + help="Also install into a host runtime", + ) + + # -- inspect ---------------------------------------------------------- + ins = sub.add_parser("inspect", help="Show package contents without installing") + ins.add_argument("package", help="Path to .skill.tar.gz file") + + return parser + + +def main() -> None: + parser = _build_parser() + args = parser.parse_args() + + try: + if args.action == "export": + out = export_skill( + character=args.character, + slug=args.slug, + output=args.output, + base_dir=args.base_dir, + include_knowledge=args.include_knowledge, + strip_personal=not args.no_strip_personal, + include_versions=args.include_versions, + ) + print(f"Exported: {out}") + + elif args.action == "import": + target = import_skill( + package_path=args.package, + base_dir=args.base_dir, + force=args.force, + install_host=args.install_host, + ) + print(f"Imported: {target}") + + elif args.action == "inspect": + info = inspect_package(args.package) + print(json.dumps(info, indent=2, ensure_ascii=False)) + + except (FileNotFoundError, FileExistsError, ValueError) as exc: + print(f"Error: {exc}", file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + main()