pipeline_core/scripts/load_path.py
yumoqing ff02533cb6 feat(team-communication): 团队沟通配置 — 团队角色(+agentid) + 问题类型(冒泡路径)
新增通用问题冒泡的数据模型与配置 CRUD:
- models/json: pipeline_team_roles(角色+agentid,多agent区分) + pipeline_problem_types(escalation_path)
- pipeline_core/init.py: _simple_crud 通用 CRUD + 6 个 create/update/delete 注册
- wwwroot/api: 6 个 dspy 端点
- scripts/load_path.py: RBAC 路径
- skills_library/all/team-communication/SKILL.md: 团队沟通技能
2026-08-16 21:25:38 +08:00

88 lines
3.4 KiB
Python

#!/usr/bin/env python3
"""RBAC path registration for pipeline_core module."""
import os, sys, subprocess
MOD = "pipeline_core"
PATHS_ANY = [
f"/{MOD}/index.ui",
]
PATHS_LOGINED = [
f"/{MOD}",
# Pipelines CRUD
f"/{MOD}/pipelines/index.ui",
f"/{MOD}/pipelines/get_pipelines.dspy",
f"/{MOD}/pipelines/add_pipelines.dspy",
f"/{MOD}/pipelines/update_pipelines.dspy",
f"/{MOD}/pipelines/delete_pipelines.dspy",
# Pipeline Steps CRUD
f"/{MOD}/pipeline_steps/index.ui",
f"/{MOD}/pipeline_steps/get_pipeline_steps.dspy",
f"/{MOD}/pipeline_steps/add_pipeline_steps.dspy",
f"/{MOD}/pipeline_steps/update_pipeline_steps.dspy",
f"/{MOD}/pipeline_steps/delete_pipeline_steps.dspy",
# Pipeline Versions
f"/{MOD}/pipeline_versions/index.ui",
f"/{MOD}/pipeline_versions/get_pipeline_versions.dspy",
f"/{MOD}/pipeline_versions/add_pipeline_versions.dspy",
f"/{MOD}/pipeline_versions/update_pipeline_versions.dspy",
f"/{MOD}/pipeline_versions/delete_pipeline_versions.dspy",
# Pipeline Editor
f"/{MOD}/pipeline_editor/index.ui",
f"/{MOD}/pipeline_editor/save_steps.dspy",
# Data sources
f"/{MOD}/pipeline_steps/get_pipelines.dspy",
f"/{MOD}/pipeline_steps/get_step_types.dspy",
# API
f"/{MOD}/api/pipeline_publish.dspy",
f"/{MOD}/api/pipeline_steps_create.dspy",
f"/{MOD}/api/pipeline_steps_update.dspy",
f"/{MOD}/api/pipeline_steps_delete.dspy",
f"/{MOD}/api/pipeline_versions_create.dspy",
f"/{MOD}/api/pipeline_versions_update.dspy",
f"/{MOD}/api/pipeline_versions_delete.dspy",
f"/{MOD}/api/pipelines_create.dspy",
f"/{MOD}/api/pipelines_update.dspy",
f"/{MOD}/api/pipelines_delete.dspy",
# Team communication (团队沟通:团队角色 + 问题类型 CRUD)
f"/{MOD}/pipeline_team_roles/index.ui",
f"/{MOD}/pipeline_team_roles/get_pipeline_team_roles.dspy",
f"/{MOD}/pipeline_team_roles/add_pipeline_team_roles.dspy",
f"/{MOD}/pipeline_team_roles/update_pipeline_team_roles.dspy",
f"/{MOD}/pipeline_team_roles/delete_pipeline_team_roles.dspy",
f"/{MOD}/pipeline_problem_types/index.ui",
f"/{MOD}/pipeline_problem_types/get_pipeline_problem_types.dspy",
f"/{MOD}/pipeline_problem_types/add_pipeline_problem_types.dspy",
f"/{MOD}/pipeline_problem_types/update_pipeline_problem_types.dspy",
f"/{MOD}/pipeline_problem_types/delete_pipeline_problem_types.dspy",
f"/{MOD}/api/pipeline_team_roles_create.dspy",
f"/{MOD}/api/pipeline_team_roles_update.dspy",
f"/{MOD}/api/pipeline_team_roles_delete.dspy",
f"/{MOD}/api/pipeline_problem_types_create.dspy",
f"/{MOD}/api/pipeline_problem_types_update.dspy",
f"/{MOD}/api/pipeline_problem_types_delete.dspy",
]
def main():
root = os.path.expanduser("~/pipeline")
if not os.path.isfile(os.path.join(root, "set_role_perm.py")):
print("ERROR: pipeline root not found")
sys.exit(1)
set_perm = os.path.join(root, "set_role_perm.py")
py = sys.executable
count = 0
for role, paths in [("any", PATHS_ANY), ("logined", PATHS_LOGINED)]:
for path in paths:
r = subprocess.run([py, set_perm, role, path], capture_output=True, text=True, cwd=root)
if r.returncode == 0:
count += 1
else:
print(f" WARN: {path} -> {r.stderr.strip()[-80:]}")
print(f"Registered {count}/{len(PATHS_ANY)+len(PATHS_LOGINED)} paths for {MOD}")
if __name__ == "__main__":
main()