yumoqing/pipeline_core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: ensure_org_skills改增量同步(全局模板新增技能同步到已有机构,不只首次复制)

Browse Source
This commit is contained in:
yumoqing 2026-08-16 19:03:34 +08:00
parent 57df6cd785
commit f19562a360
1 changed files with 28 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
pipeline_core/skill_pack.py
View File

@ -45,21 +45,42 @@ def get_org_skills_dir(workspace_base, org_id):
def ensure_org_skills(workspace_base, org_id): def ensure_org_skills(workspace_base, org_id):
"""确保机构工作目录的 skills/ 已初始化（首次从全局模板复制公共技能 + 技能集清单）。 """确保机构工作目录的 skills/ 已初始化（从全局模板复制公共技能 + 技能集清单）。
多租户隔离：运行时技能必须落在机构工作目录，不能读全局应用目录的 skills/。 多租户隔离：运行时技能必须落在机构工作目录，不能读全局应用目录的 skills/。
增量同步：全局模板新增的技能/清单，会同步到机构目录（已有机构也能拿到新技能）。
返回机构技能根目录。 返回机构技能根目录。
""" """
org_skills = get_org_skills_dir(workspace_base, org_id) org_skills = get_org_skills_dir(workspace_base, org_id)
os.makedirs(org_skills, exist_ok=True) os.makedirs(org_skills, exist_ok=True)
# 复制公共技能（global） # 公共技能（global）：首次全量复制，之后增量同步（全局模板新增的技能复制过来）
if not os.path.isdir(os.path.join(org_skills, "global")): org_global = os.path.join(org_skills, "global")
if not os.path.isdir(org_global):
if os.path.isdir(ALL_DIR): if os.path.isdir(ALL_DIR):
shutil.copytree(ALL_DIR, os.path.join(org_skills, "global")) shutil.copytree(ALL_DIR, org_global)
# 复制技能集清单（packs） elif os.path.isdir(ALL_DIR):
if not os.path.isdir(os.path.join(org_skills, "packs")): for name in os.listdir(ALL_DIR):
src = os.path.join(ALL_DIR, name)
dst = os.path.join(org_global, name)
if not os.path.exists(dst):
if os.path.isdir(src):
shutil.copytree(src, dst)
else:
shutil.copy2(src, dst)
# 技能集清单（packs）：首次全量复制，之后增量同步
org_packs = os.path.join(org_skills, "packs")
if not os.path.isdir(org_packs):
if os.path.isdir(PACKS_DIR): if os.path.isdir(PACKS_DIR):
shutil.copytree(PACKS_DIR, os.path.join(org_skills, "packs")) shutil.copytree(PACKS_DIR, org_packs)
elif os.path.isdir(PACKS_DIR):
for name in os.listdir(PACKS_DIR):
src = os.path.join(PACKS_DIR, name)
dst = os.path.join(org_packs, name)
if not os.path.exists(dst):
if os.path.isdir(src):
shutil.copytree(src, dst)
else:
shutil.copy2(src, dst)
return org_skills return org_skills