100 lines
3.7 KiB
Python
100 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
||
"""voucher 模块 RBAC 路径权限注册（pipeline-app 格式，仿 ticket/scripts/load_path.py）。
|
||
|
||
从应用根执行：py3/bin/python pkgs/voucher/scripts/load_path.py
|
||
（load_path.sh 自动扫描 pkgs/*/scripts/load_path.py 并调用本脚本）
|
||
|
||
权限设计（逐个精确配，不用统一兜底）：
|
||
- logined：客户侧「我的代金券」页面 + 自助查询 API + 管理侧全部页面与 API。
|
||
管理类数据面由服务端 org_id 隔离（模板带 org_id 列，CRUD logined_userorgid 机制
|
||
不适用——平台运营代客户管券是本职，不做 org 过滤；误操作面由 helper 内校验兜底：
|
||
模板改删校验 org、实例改/作废只动 unused 券）。
|
||
- CRUD 生成目录（voucher_*_list/）四件套（index.ui + get/add/update/delete dspy）
|
||
逐一注册。
|
||
"""
|
||
import sys
|
||
import os
|
||
import asyncio
|
||
|
||
# 应用根：本脚本在 pkgs/voucher/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 = "voucher"
|
||
|
||
# CRUD 生成目录四件套（alias -> 表名）
|
||
CRUD_DIRS = {
|
||
"voucher_template_list": "voucher_template",
|
||
"voucher_rule_list": "voucher_rule",
|
||
"voucher_instance_list": "voucher_instance",
|
||
"voucher_usage_log_list": "voucher_usage_log",
|
||
}
|
||
|
||
PATHS_LOGINED = [
|
||
# 客户侧
|
||
f"/{MOD}/my",
|
||
f"/{MOD}/my/index.ui",
|
||
f"/{MOD}/api/my_vouchers.dspy",
|
||
f"/{MOD}/api/v1/available.dspy",
|
||
# 管理侧页面/表单
|
||
f"/{MOD}/index.ui",
|
||
f"/{MOD}/issue_form.ui",
|
||
f"/{MOD}/invalidate_form.ui",
|
||
# 管理侧 API
|
||
f"/{MOD}/api/get_available.dspy",
|
||
f"/{MOD}/api/apply_voucher.dspy",
|
||
f"/{MOD}/api/rule_types.dspy",
|
||
f"/{MOD}/api/template/voucher_template_create.dspy",
|
||
f"/{MOD}/api/template/voucher_template_update.dspy",
|
||
f"/{MOD}/api/template/voucher_template_delete.dspy",
|
||
f"/{MOD}/api/template/voucher_template_options.dspy",
|
||
f"/{MOD}/api/rule/voucher_rule_create.dspy",
|
||
f"/{MOD}/api/rule/voucher_rule_update.dspy",
|
||
f"/{MOD}/api/rule/voucher_rule_delete.dspy",
|
||
f"/{MOD}/api/instance/voucher_issue.dspy",
|
||
f"/{MOD}/api/instance/voucher_instance_update.dspy",
|
||
f"/{MOD}/api/instance/voucher_instance_delete.dspy",
|
||
f"/{MOD}/api/instance/voucher_invalidate.dspy",
|
||
]
|
||
|
||
# CRUD 目录（框架生成，路径必须逐个注册）
|
||
for alias, tbl in CRUD_DIRS.items():
|
||
PATHS_LOGINED += [
|
||
f"/{MOD}/{alias}",
|
||
f"/{MOD}/{alias}/index.ui",
|
||
f"/{MOD}/{alias}/get_{tbl}.dspy",
|
||
f"/{MOD}/{alias}/add_{tbl}.dspy",
|
||
f"/{MOD}/{alias}/update_{tbl}.dspy",
|
||
f"/{MOD}/{alias}/delete_{tbl}.dspy",
|
||
]
|
||
|
||
|
||
async def main():
|
||
config = getConfig(APP_ROOT, NS={"workdir": APP_ROOT})
|
||
DBPools(config.databases)
|
||
cnt = 0
|
||
async with DBPools().sqlorContext("pipeline") as sor:
|
||
for path in PATHS_LOGINED:
|
||
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": "logined", "permid": permid})
|
||
if not rp:
|
||
await sor.C("rolepermission", {
|
||
"id": getID(), "roleid": "logined", "permid": permid})
|
||
cnt += 1
|
||
await sor.sqlExe("COMMIT", {})
|
||
print(f"voucher load_path: {cnt} logined path-role entries ensured")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|