129 lines
4.8 KiB
Python
129 lines
4.8 KiB
Python
"""技能集（skill pack）管理 + 安装能力。
|
||
|
||
技能目录结构（单一技能树 skills/，所有机构共享读，2026-08-21 重构）：
|
||
skills/
|
||
├── global/ # 平台通用技能
|
||
├── projects/ # 项目通用技能（所有项目共享）
|
||
├── pipelines/{pid}/common/ # 产线通用技能
|
||
├── pipelines/{pid}/roles/{role}/ # 角色技能
|
||
├── packs/ # 技能集定义（manifest 引用 global 里的技能名）
|
||
├── orgs/{org_id}/ # 机构技能（+ {project_id}/ 项目私有）
|
||
│ └── {project_id}/
|
||
└── users/{user_id}/ # 用户技能
|
||
|
||
安装 = 按技能集 manifest，把 global/ 里的技能复制到 orgs/{org_id}/ 下（机构显式选装，覆盖 global 同名）。
|
||
"""
|
||
import os
|
||
import json
|
||
import shutil
|
||
|
||
|
||
def get_skills_base():
|
||
"""技能根目录 skills/（相对 cwd），与 skill_loader 的 base_dir="skills" 一致。
|
||
|
||
优先环境变量 PIPELINE_SKILLS_BASE 覆盖。
|
||
"""
|
||
env = os.environ.get("PIPELINE_SKILLS_BASE", "")
|
||
if env:
|
||
return os.path.normpath(env)
|
||
return os.path.normpath(os.path.join(os.getcwd(), "skills"))
|
||
|
||
|
||
SKILLS_BASE = get_skills_base()
|
||
PACKS_DIR = os.path.join(SKILLS_BASE, "packs")
|
||
ALL_DIR = os.path.join(SKILLS_BASE, "global") # 公共技能 = 全局模板（源头）
|
||
PIPELINES_DIR = os.path.join(SKILLS_BASE, "pipelines") # 产线技能（common/roles），源头在 build.sh 6b 复制
|
||
|
||
|
||
def list_packs():
|
||
"""列出所有可安装的技能集（manifest 摘要）。"""
|
||
packs = []
|
||
if not os.path.isdir(PACKS_DIR):
|
||
return packs
|
||
for d in sorted(os.listdir(PACKS_DIR)):
|
||
mf = os.path.join(PACKS_DIR, d, "manifest.json")
|
||
if os.path.isfile(mf):
|
||
try:
|
||
with open(mf, "r", encoding="utf-8") as f:
|
||
m = json.load(f)
|
||
packs.append({
|
||
"name": m.get("name", d),
|
||
"title": m.get("title", d),
|
||
"description": m.get("description", ""),
|
||
"vendor": m.get("vendor", ""),
|
||
"version": m.get("version", "1.0.0"),
|
||
"skill_count": len(m.get("skills", [])),
|
||
})
|
||
except Exception:
|
||
continue
|
||
return packs
|
||
|
||
|
||
def get_pack(pack_name):
|
||
"""返回技能集 manifest（含技能列表），不存在返回 None。"""
|
||
mf = os.path.join(PACKS_DIR, pack_name, "manifest.json")
|
||
if not os.path.isfile(mf):
|
||
return None
|
||
with open(mf, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
|
||
def install_pack(pack_name, target_base_dir, org_id):
|
||
"""把技能集安装到目标技能根目录的 orgs/{org_id}/ 下。
|
||
|
||
target_base_dir: skill_loader 的 base_dir（运行时技能根目录，如 .../skills）
|
||
org_id: 机构 ID（安装到该机构的 org scope）
|
||
返回 {success, installed:[...], skipped:[...], error}
|
||
"""
|
||
pack = get_pack(pack_name)
|
||
if not pack:
|
||
return {"success": False, "error": f"技能集不存在: {pack_name}"}
|
||
|
||
org_dir = os.path.join(target_base_dir, "orgs", str(org_id))
|
||
os.makedirs(org_dir, exist_ok=True)
|
||
|
||
installed, skipped = [], []
|
||
for skill_name in pack.get("skills", []):
|
||
src = os.path.join(ALL_DIR, skill_name)
|
||
if not os.path.isdir(src):
|
||
skipped.append(skill_name)
|
||
continue
|
||
dst = os.path.join(org_dir, skill_name)
|
||
if os.path.exists(dst):
|
||
shutil.rmtree(dst)
|
||
shutil.copytree(src, dst)
|
||
installed.append(skill_name)
|
||
|
||
return {"success": True, "installed": installed, "skipped": skipped}
|
||
|
||
|
||
def uninstall_pack(pack_name, target_base_dir, org_id):
|
||
"""卸载技能集：删除该机构下该技能集引用的技能目录。"""
|
||
pack = get_pack(pack_name)
|
||
if not pack:
|
||
return {"success": False, "error": f"技能集不存在: {pack_name}"}
|
||
org_dir = os.path.join(target_base_dir, "orgs", str(org_id))
|
||
removed = []
|
||
for skill_name in pack.get("skills", []):
|
||
dst = os.path.join(org_dir, skill_name)
|
||
if os.path.isdir(dst):
|
||
shutil.rmtree(dst)
|
||
removed.append(skill_name)
|
||
return {"success": True, "removed": removed}
|
||
|
||
|
||
def installed_packs(target_base_dir, org_id):
|
||
"""返回某机构已安装的技能集名列表。"""
|
||
packs = []
|
||
for p in list_packs():
|
||
pack = get_pack(p["name"])
|
||
org_dir = os.path.join(target_base_dir, "orgs", str(org_id))
|
||
cnt = 0
|
||
for skill_name in pack.get("skills", []):
|
||
if os.path.isdir(os.path.join(org_dir, skill_name)):
|
||
cnt += 1
|
||
if cnt > 0:
|
||
packs.append({"name": p["name"], "title": p["title"],
|
||
"installed_skills": cnt, "total_skills": p["skill_count"]})
|
||
return packs
|