张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
143 lines
5.6 KiB
Python
143 lines
5.6 KiB
Python
"""幂等补齐 `handover:write`（客服转人工工单处置）权限并授权给 admin。
|
||
|
||
## 为什么需要它
|
||
|
||
`docs/02-数据库建表设计.md` §7.2 定义了工单状态机
|
||
（`pending -> assigned -> processing -> resolved -> closed`，未解决可 `cancelled`），
|
||
但平台此前**只有** `handover:read`（9046，只读队列）—— 于是工单只能看、不能推进，
|
||
库里 40 张单子全部停在 `pending`。处置端点与状态机见
|
||
`app/service/customer_service_handover_action_service.py` 与
|
||
`app/api/controllers/admin.py` 的五个 action 端点。
|
||
|
||
权限码 **9069 已并进种子** `tools/seed_test_rbac.py`（那是定义源），
|
||
本脚本只做"幂等补齐 + 授权"，不重建任何东西；id 必须与种子逐条一致
|
||
（一致性由 `tools/check_rbac_seed_consistency.py` 守着）。
|
||
|
||
用法::
|
||
|
||
python tools/grant_handover_write_permission.py # 只打印要做什么
|
||
python tools/grant_handover_write_permission.py --apply # 真写
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import sys
|
||
from datetime import UTC, datetime
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.infrastructure.db import SessionFactory
|
||
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr]
|
||
|
||
#: (id, permission_code, resource, action, data_scope) —— 与种子里那一行逐字一致。
|
||
PERMISSION: tuple[int, str, str, str, str] = (
|
||
9069, "handover:write", "handover", "write", "all",
|
||
)
|
||
|
||
#: 授权给哪些角色：工单队列是**管理面**（读侧也是 admin），所以只给 admin。
|
||
#: `ADMIN_PERMISSIONS` 在种子里是全量元组，重跑种子也会自动带上这一条。
|
||
GRANTED_ROLES: tuple[str, ...] = ("admin",)
|
||
|
||
|
||
async def apply(*, dry_run: bool) -> int:
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
perm_id, code, resource, action, scope = PERMISSION
|
||
async with SessionFactory() as session, session.begin():
|
||
found = await session.scalar(
|
||
text("SELECT id FROM sys_permission WHERE permission_code = :code"), {"code": code}
|
||
)
|
||
print(f"权限 {code}: {'已存在（跳过插入）' if found else '缺失，将新增 id=' + str(perm_id)}")
|
||
if dry_run:
|
||
role = await session.scalar(
|
||
text("SELECT id FROM sys_role WHERE role_code = 'admin'")
|
||
)
|
||
print(f"角色 admin: {'存在' if role else '缺失（请先跑 seed_test_rbac.py）'}")
|
||
print("\n[dry-run] 未写入任何数据。加 --apply 真写。")
|
||
return 0
|
||
|
||
if found is None:
|
||
await session.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO sys_permission
|
||
(id, permission_code, resource, action, data_scope, created_at, updated_at)
|
||
VALUES (:id, :code, :resource, :action, :scope, :now, :now)
|
||
"""
|
||
),
|
||
{"id": perm_id, "code": code, "resource": resource, "action": action,
|
||
"scope": scope, "now": now},
|
||
)
|
||
resolved_id = perm_id
|
||
else:
|
||
resolved_id = int(found)
|
||
|
||
for role_code in GRANTED_ROLES:
|
||
role_id = await session.scalar(
|
||
text("SELECT id FROM sys_role WHERE role_code = :code"), {"code": role_code}
|
||
)
|
||
if role_id is None:
|
||
print(f"跳过角色 {role_code}：不存在")
|
||
continue
|
||
have = await session.scalar(
|
||
text(
|
||
"SELECT 1 FROM sys_role_permission WHERE role_id = :r AND permission_id = :p"
|
||
),
|
||
{"r": int(role_id), "p": resolved_id},
|
||
)
|
||
if have:
|
||
print(f"授权：{role_code} 已拥有 {code}（跳过）")
|
||
continue
|
||
await session.execute(
|
||
text(
|
||
"INSERT INTO sys_role_permission (role_id, permission_id, created_at)"
|
||
" VALUES (:r, :p, :now)"
|
||
),
|
||
{"r": int(role_id), "p": resolved_id, "now": now},
|
||
)
|
||
print(f"授权：{role_code} += {code}")
|
||
|
||
return await verify()
|
||
|
||
|
||
async def verify() -> int:
|
||
"""按权限码实测一遍，而不是只看插了几行。"""
|
||
async with SessionFactory() as session:
|
||
rows = (
|
||
await session.execute(
|
||
text(
|
||
"""
|
||
SELECT p.permission_code, p.data_scope, r.role_code
|
||
FROM sys_permission p
|
||
LEFT JOIN sys_role_permission rp ON rp.permission_id = p.id
|
||
LEFT JOIN sys_role r ON r.id = rp.role_id
|
||
WHERE p.permission_code = 'handover:write'
|
||
ORDER BY r.role_code
|
||
"""
|
||
)
|
||
)
|
||
).all()
|
||
if not rows:
|
||
print("校验失败：库里查不到 handover:write")
|
||
return 1
|
||
for code, scope, role_code in rows:
|
||
print(f"校验：{code} scope={scope} → 角色 {role_code or '(未授权任何角色)'}")
|
||
if not any(role_code for _code, _scope, role_code in rows):
|
||
print("校验失败：handover:write 存在但没有授权给任何角色（处置端点会一律 403）")
|
||
return 1
|
||
return 0
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="补齐并授权 handover:write")
|
||
parser.add_argument("--apply", action="store_true", help="真正写入（默认只打印）")
|
||
args = parser.parse_args()
|
||
return asyncio.run(apply(dry_run=not args.apply))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|