yumoqing
54f2f2d597
- 3表: tk_tickets(状态机+受理角色/人+agent锁)/tk_messages(客户可见/内部备注)/tk_transfers(append-only流水) - core.py: 8态13迁移状态机全乐观锁; 转派候选动态查owner角色+'0'机构用户(不硬编码); 默认受理角色params可配(ticket_default_role=owner.maintainer) - agent.py: 异步poller(bid_flow范式), RAG检索'0'机构KB+历史工单, llm_call(org_id=0) LLM裁决能否回答(严格JSON), 硬门禁reply<20字符转人工, 故障≠答不了(fail_rounds>=3转人工), stale回收10min; 追问回流(agent_processing+owner空)立即拾取不等stale - todos.py: 平台待办provider, 状态派生(客户待确认/角色池待认领/受理人待处理) - init.py: load_ticket注册env+provider+poller(PIPELINE_MODE=web不启动) - wwwroot: 客户我的工单页+建单表单, 运维工单管理页(队列切换), 详情弹窗(正文与操作同屏,按视角出按钮) - load_path.py: logined全部API(业务权限服务端按current_role动态校验)+管理页壳owner三角色 - init/data.json: 7组码表+owner.maintainer角色种子(app_audit先例)
89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
||
"""RBAC path registration for ticket module (pipeline format).
|
||
|
||
从应用根执行：py3/bin/python pkgs/ticket/scripts/load_path.py
|
||
（load_path.sh 自动扫描 pkgs/*/scripts/load_path.py 并调用本脚本）
|
||
|
||
权限设计（设计文档 §7.2，逐个精确配不用统一兜底）：
|
||
- logined：模块页面壳 + 全部 API。
|
||
⚠ staff 动作端点（claim/staff_reply/transfer/transfer_candidates/staff_tickets）
|
||
只能授 logined 而不能授固定角色——D4 决策：转派目标是 owner 组织类型下的
|
||
**任意动态角色**（role 表运行时可增删），RBAC 路径权限无法枚举动态角色，
|
||
业务权限由服务端 _check_staff_access 校验（操作者须持有工单 current_role，
|
||
admin/superuser 豁免）。客户侧数据由 I5 服务端自过滤（customer_user_id/org_id）。
|
||
- owner.maintainer / owner.admin / owner.superuser：管理页壳 /ticket/manage/**。
|
||
菜单「工单管理」入口也按角色 Jinja 分支渲染，客户直接访问 URL 403。
|
||
"""
|
||
import sys
|
||
import os
|
||
import asyncio
|
||
|
||
# 应用根：本脚本在 pkgs/ticket/scripts/ 下，上溯三级
|
||
APP_ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "../../.."))
|
||
sys.path.insert(0, os.path.join(APP_ROOT, "py3", "lib", "python3.10", "site-packages"))
|
||
sys.path.insert(0, APP_ROOT)
|
||
|
||
from sqlor.dbpools import DBPools
|
||
from appPublic.jsonConfig import getConfig
|
||
from appPublic.uniqueID import getID
|
||
|
||
MOD = "ticket"
|
||
|
||
PATHS_LOGINED = [
|
||
# 客户侧页面
|
||
f"/{MOD}/my",
|
||
f"/{MOD}/my/index.ui",
|
||
f"/{MOD}/my/new_ticket_form.ui",
|
||
# 客户侧 API（I5 服务端自过滤）
|
||
f"/{MOD}/api/ticket_create.dspy",
|
||
f"/{MOD}/api/my_tickets.dspy",
|
||
f"/{MOD}/api/ticket_detail.dspy",
|
||
f"/{MOD}/api/ticket_detail_popup.dspy",
|
||
f"/{MOD}/api/ticket_followup.dspy",
|
||
f"/{MOD}/api/ticket_confirm.dspy",
|
||
f"/{MOD}/api/ticket_cancel.dspy",
|
||
# staff 侧 API（服务端 _check_staff_access 按工单 current_role 动态校验）
|
||
f"/{MOD}/api/staff_tickets.dspy",
|
||
f"/{MOD}/api/ticket_claim.dspy",
|
||
f"/{MOD}/api/ticket_staff_reply.dspy",
|
||
f"/{MOD}/api/ticket_transfer.dspy",
|
||
f"/{MOD}/api/transfer_candidates.dspy",
|
||
]
|
||
|
||
# 管理页壳：运维三角色
|
||
PATHS_STAFF = [
|
||
f"/{MOD}/manage",
|
||
f"/{MOD}/manage/index.ui",
|
||
]
|
||
|
||
STAFF_ROLES = ["owner.maintainer", "owner.admin", "owner.superuser"]
|
||
|
||
|
||
async def main():
|
||
config = getConfig(APP_ROOT, NS={"workdir": APP_ROOT})
|
||
DBPools(config.databases)
|
||
cnt = 0
|
||
async with DBPools().sqlorContext("pipeline") as sor:
|
||
pairs = [("logined", p) for p in PATHS_LOGINED]
|
||
for role in STAFF_ROLES:
|
||
pairs += [(role, p) for p in PATHS_STAFF]
|
||
for role, path in pairs:
|
||
recs = await sor.R("permission", {"path": path})
|
||
if recs:
|
||
permid = recs[0].id
|
||
else:
|
||
permid = getID()
|
||
await sor.C("permission", {"id": permid, "path": path})
|
||
rp = await sor.R("rolepermission", {"roleid": role, "permid": permid})
|
||
if not rp:
|
||
await sor.C("rolepermission", {
|
||
"id": getID(), "roleid": role, "permid": permid})
|
||
cnt += 1
|
||
await sor.sqlExe("COMMIT", {})
|
||
print(f"ticket load_path: {cnt} path-role entries ensured "
|
||
f"({len(PATHS_LOGINED)} logined + {len(PATHS_STAFF)}x{len(STAFF_ROLES)} staff)")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|