81 lines
3.0 KiB
Plaintext
81 lines
3.0 KiB
Plaintext
# skill_pack.dspy - 技能集选装/安装（机构管理员，pipeline-core 通用能力）
|
||
# action=list: 列出可安装的技能集
|
||
# action=installed: 查当前机构已安装的技能集
|
||
# action=install: 安装技能集到当前机构（org scope）
|
||
# action=uninstall: 卸载技能集
|
||
|
||
import json
|
||
import os
|
||
|
||
uid = await get_user()
|
||
if not uid:
|
||
return json.dumps({"success": False, "error": "请先登录"}, ensure_ascii=False)
|
||
|
||
action = (params_kw or {}).get('action', 'list')
|
||
dbname = get_module_dbname('pipeline_core')
|
||
|
||
# 查用户 org_id + 角色（pipeline 库）
|
||
org_id = ''
|
||
is_admin = False
|
||
try:
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
urecs = await sor.sqlExe("SELECT orgid FROM users WHERE id=${u}$", {"u": uid})
|
||
if urecs:
|
||
org_id = getattr(urecs[0], 'orgid', '') or ''
|
||
rrecs = await sor.sqlExe(
|
||
"SELECT r.name FROM userrole ur JOIN role r ON ur.roleid=r.id WHERE ur.userid=${u}$",
|
||
{"u": uid})
|
||
roles = [getattr(r, 'name', '') for r in (rrecs or [])]
|
||
is_admin = ('superuser' in roles) or ('admin' in roles)
|
||
except Exception:
|
||
pass
|
||
|
||
if not org_id:
|
||
return json.dumps({"success": False, "error": "用户未关联机构"}, ensure_ascii=False)
|
||
|
||
# 技能根目录（skill_loader 的 base_dir，相对进程 cwd）
|
||
base_dir = os.environ.get("PIPELINE_SKILLS_BASE", "skills")
|
||
|
||
from pipeline_core.skill_pack import list_packs, install_pack, uninstall_pack, installed_packs
|
||
|
||
if action == 'list':
|
||
return json.dumps({"success": True, "is_admin": is_admin, "packs": list_packs()},
|
||
ensure_ascii=False)
|
||
|
||
elif action == 'installed':
|
||
return json.dumps({"success": True, "packs": installed_packs(base_dir, org_id)},
|
||
ensure_ascii=False)
|
||
|
||
elif action == 'install':
|
||
if not is_admin:
|
||
return json.dumps({"success": False, "error": "仅机构管理员可安装技能集"}, ensure_ascii=False)
|
||
pack_name = (params_kw or {}).get('pack', '').strip()
|
||
if not pack_name:
|
||
return json.dumps({"success": False, "error": "pack 必填"}, ensure_ascii=False)
|
||
r = install_pack(pack_name, base_dir, org_id)
|
||
if r.get("success"):
|
||
try:
|
||
from pipeline_core.skill_loader import get_skill_loader
|
||
get_skill_loader().reload()
|
||
except Exception:
|
||
pass
|
||
return json.dumps(r, ensure_ascii=False)
|
||
|
||
elif action == 'uninstall':
|
||
if not is_admin:
|
||
return json.dumps({"success": False, "error": "仅机构管理员可卸载技能集"}, ensure_ascii=False)
|
||
pack_name = (params_kw or {}).get('pack', '').strip()
|
||
if not pack_name:
|
||
return json.dumps({"success": False, "error": "pack 必填"}, ensure_ascii=False)
|
||
r = uninstall_pack(pack_name, base_dir, org_id)
|
||
if r.get("success"):
|
||
try:
|
||
from pipeline_core.skill_loader import get_skill_loader
|
||
get_skill_loader().reload()
|
||
except Exception:
|
||
pass
|
||
return json.dumps(r, ensure_ascii=False)
|
||
|
||
else:
|
||
return json.dumps({"error": "Unknown action: " + str(action)}, ensure_ascii=False)
|