b8a5de7a07
- 从json/*.json自动提取tblname和alias，生成所有CRUD路径 - 自动扫描wwwroot/api/*.dspy注册所有API路径 - 去重处理，避免重复注册 - operator获得全部CRUD+API权限，sale获得促销码+客户归属权限
231 lines
7.3 KiB
Python
231 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
discount 模块 RBAC 权限管理脚本
|
||
|
||
使用方法:
|
||
cd ~/repos/sage
|
||
./py3/bin/python ~/repos/discount/scripts/load_path.py
|
||
|
||
每次代码变更如有新 path 出现，需同步更新此脚本。
|
||
|
||
路径分类:
|
||
- any: 静态资源/菜单/CRUD别名目录
|
||
- logined: 需要认证的页面和 API
|
||
- reseller.operator: 运营角色 — 全部折扣管理功能
|
||
- reseller.sale: 销售角色 — 促销码、客户归属
|
||
"""
|
||
|
||
import subprocess
|
||
import os
|
||
import sys
|
||
import json
|
||
import glob
|
||
|
||
def find_sage_root():
|
||
candidates = [
|
||
os.path.expanduser("~/repos/sage"),
|
||
os.path.expanduser("~/sage"),
|
||
os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))),
|
||
]
|
||
for c in candidates:
|
||
if os.path.isdir(os.path.join(c, "py3")) and os.path.isdir(os.path.join(c, "wwwroot")):
|
||
return c
|
||
return None
|
||
|
||
SAGE_ROOT = find_sage_root()
|
||
if not SAGE_ROOT:
|
||
print("ERROR: Cannot find Sage root directory")
|
||
sys.exit(1)
|
||
|
||
PYTHON = os.path.join(SAGE_ROOT, "py3", "bin", "python")
|
||
SET_PERM_SCRIPT = os.path.join(SAGE_ROOT, "set_role_perm.py")
|
||
|
||
if not os.path.exists(SET_PERM_SCRIPT):
|
||
print(f"ERROR: set_role_perm.py not found at {SET_PERM_SCRIPT}")
|
||
sys.exit(1)
|
||
|
||
MOD = "discount"
|
||
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
JSON_DIR = os.path.join(os.path.dirname(SCRIPT_DIR), "json")
|
||
API_DIR = os.path.join(os.path.dirname(SCRIPT_DIR), "wwwroot", "api")
|
||
|
||
# ============================================================
|
||
# 从 json/ 目录自动推导 CRUD 路径
|
||
# ============================================================
|
||
|
||
def load_crud_definitions():
|
||
"""读取所有 json/*.json，提取 CRUD 定义信息"""
|
||
defs = []
|
||
for f in sorted(glob.glob(os.path.join(JSON_DIR, "*.json"))):
|
||
try:
|
||
d = json.load(open(f))
|
||
tblname = d.get("tblname", "")
|
||
alias = d.get("alias", tblname)
|
||
params = d.get("params", {})
|
||
subtables = params.get("subtables", [])
|
||
defs.append({
|
||
"tblname": tblname,
|
||
"alias": alias,
|
||
"subtables": subtables,
|
||
"file": f,
|
||
})
|
||
except Exception as e:
|
||
print(f" WARNING: skip {f}: {e}")
|
||
return defs
|
||
|
||
def get_api_files():
|
||
"""扫描 wwwroot/api/ 下所有 .dspy 文件"""
|
||
apis = []
|
||
if os.path.isdir(API_DIR):
|
||
for f in sorted(os.listdir(API_DIR)):
|
||
if f.endswith(".dspy"):
|
||
apis.append(f"/{MOD}/api/{f}")
|
||
return apis
|
||
|
||
# ============================================================
|
||
# 构建路径列表
|
||
# ============================================================
|
||
|
||
crud_defs = load_crud_definitions()
|
||
print(f"Found {len(crud_defs)} CRUD definitions in {JSON_DIR}")
|
||
for d in crud_defs:
|
||
print(f" - {d['tblname']} (alias: {d['alias']})")
|
||
|
||
# any — 无需登录（菜单、CRUD别名目录）
|
||
PATHS_ANY = [
|
||
f"/{MOD}/menu.ui",
|
||
]
|
||
|
||
# 每个 CRUD 定义 → any 别名目录 + logined 页面和脚本
|
||
PATHS_LOGINED = [
|
||
# 模块入口
|
||
f"/{MOD}",
|
||
f"/{MOD}/index.ui",
|
||
# 功能页面
|
||
f"/{MOD}/generate_qr.dspy",
|
||
f"/{MOD}/promote",
|
||
f"/{MOD}/promote.ui",
|
||
f"/{MOD}/promote/index.dspy",
|
||
]
|
||
|
||
for d in crud_defs:
|
||
alias = d["alias"]
|
||
tblname = d["tblname"]
|
||
# any: 别名目录
|
||
PATHS_ANY.append(f"/{MOD}/{alias}")
|
||
# logined: index.ui
|
||
PATHS_LOGINED.append(f"/{MOD}/{alias}/index.ui")
|
||
# logined: auto-generated .dspy (get/add/update/delete)
|
||
PATHS_LOGINED.append(f"/{MOD}/{alias}/get_{tblname}.dspy")
|
||
PATHS_LOGINED.append(f"/{MOD}/{alias}/add_{tblname}.dspy")
|
||
PATHS_LOGINED.append(f"/{MOD}/{alias}/update_{tblname}.dspy")
|
||
PATHS_LOGINED.append(f"/{MOD}/{alias}/delete_{tblname}.dspy")
|
||
|
||
# logined: 所有 API 文件
|
||
for api_path in get_api_files():
|
||
PATHS_LOGINED.append(api_path)
|
||
|
||
# ============================================================
|
||
# 角色专属权限
|
||
# ============================================================
|
||
|
||
# operator — 全部折扣管理功能
|
||
PATHS_OPERATOR = []
|
||
for d in crud_defs:
|
||
alias = d["alias"]
|
||
tblname = d["tblname"]
|
||
PATHS_OPERATOR.append(f"/{MOD}/{alias}")
|
||
PATHS_OPERATOR.append(f"/{MOD}/{alias}/index.ui")
|
||
PATHS_OPERATOR.append(f"/{MOD}/{alias}/get_{tblname}.dspy")
|
||
PATHS_OPERATOR.append(f"/{MOD}/{alias}/add_{tblname}.dspy")
|
||
PATHS_OPERATOR.append(f"/{MOD}/{alias}/update_{tblname}.dspy")
|
||
PATHS_OPERATOR.append(f"/{MOD}/{alias}/delete_{tblname}.dspy")
|
||
|
||
# operator 也需要所有 API
|
||
for api_path in get_api_files():
|
||
PATHS_OPERATOR.append(api_path)
|
||
|
||
# sale — 促销码、客户归属
|
||
PATHS_SALE = []
|
||
SALE_TABLES = ["discount_promo_code", "discount_customer_bind"]
|
||
SALE_ALIASES = {d["tblname"]: d["alias"] for d in crud_defs}
|
||
for tblname in SALE_TABLES:
|
||
alias = SALE_ALIASES.get(tblname, tblname + "_list")
|
||
PATHS_SALE.append(f"/{MOD}/{alias}")
|
||
PATHS_SALE.append(f"/{MOD}/{alias}/index.ui")
|
||
PATHS_SALE.append(f"/{MOD}/{alias}/get_{tblname}.dspy")
|
||
PATHS_SALE.append(f"/{MOD}/{alias}/add_{tblname}.dspy")
|
||
PATHS_SALE.append(f"/{MOD}/{alias}/update_{tblname}.dspy")
|
||
PATHS_SALE.append(f"/{MOD}/{alias}/delete_{tblname}.dspy")
|
||
|
||
# sale 相关 API
|
||
for api_path in get_api_files():
|
||
basename = os.path.basename(api_path)
|
||
if any(k in basename for k in ["promo_code", "customer_bind"]):
|
||
PATHS_SALE.append(api_path)
|
||
|
||
# ============================================================
|
||
# 去重
|
||
# ============================================================
|
||
PATHS_ANY = list(dict.fromkeys(PATHS_ANY))
|
||
PATHS_LOGINED = list(dict.fromkeys(PATHS_LOGINED))
|
||
PATHS_OPERATOR = list(dict.fromkeys(PATHS_OPERATOR))
|
||
PATHS_SALE = list(dict.fromkeys(PATHS_SALE))
|
||
|
||
print(f"\nPath counts:")
|
||
print(f" any: {len(PATHS_ANY)}")
|
||
print(f" logined: {len(PATHS_LOGINED)}")
|
||
print(f" reseller.operator: {len(PATHS_OPERATOR)}")
|
||
print(f" reseller.sale: {len(PATHS_SALE)}")
|
||
|
||
# ============================================================
|
||
# 执行注册
|
||
# ============================================================
|
||
|
||
def run_set_perm(role, path, verbose=True):
|
||
cmd = [PYTHON, SET_PERM_SCRIPT, role, path]
|
||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||
if verbose:
|
||
output = result.stdout.strip()
|
||
if output:
|
||
print(f" {role}: {path} -> {output}")
|
||
return result.returncode == 0
|
||
|
||
|
||
def register_role_paths(role, paths):
|
||
count = 0
|
||
for path in paths:
|
||
if run_set_perm(role, path, verbose=False):
|
||
count += 1
|
||
print(f" {role}: {count}/{len(paths)} paths registered")
|
||
return count
|
||
|
||
|
||
def main():
|
||
print(f"\nSage root: {SAGE_ROOT}")
|
||
print(f"set_role_perm.py: {SET_PERM_SCRIPT}")
|
||
print()
|
||
|
||
total = 0
|
||
|
||
print("[1/4] Registering 'any' role paths...")
|
||
total += register_role_paths("any", PATHS_ANY)
|
||
|
||
print("[2/4] Registering 'logined' role paths...")
|
||
total += register_role_paths("logined", PATHS_LOGINED)
|
||
|
||
print("[3/4] Registering 'reseller.operator' role paths...")
|
||
total += register_role_paths("reseller.operator", PATHS_OPERATOR)
|
||
|
||
print("[4/4] Registering 'reseller.sale' role paths...")
|
||
total += register_role_paths("reseller.sale", PATHS_SALE)
|
||
|
||
print()
|
||
print(f"Done. Total {total} permission entries registered.")
|
||
print()
|
||
print("NOTE: Restart Sage after permission changes to reload RBAC cache.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|