回复 ZSY 的底座扩展确认 v2：逐行核验两句确认；指出 docs/05 端点编号撞车、三个新权限未随代码合并；补权限种子脚本
This commit is contained in:
@@ -0,0 +1,210 @@
|
||||
"""补客服二期（画像候选 + 转人工工单）需要的三个权限码。
|
||||
|
||||
## 为什么需要它
|
||||
|
||||
客服二期这条线新增了 6 个接口，其中 5 个要三个**平台此前不存在的权限码**：
|
||||
|
||||
| 接口 | 权限码 |
|
||||
|---|---|
|
||||
| `POST /api/v1/users/me/memory-candidates/{id}/decisions` | `memory:candidate:confirm` |
|
||||
| `GET /api/v1/admin/customer-profile-candidates` | `memory:candidate:review` |
|
||||
| `POST /api/v1/admin/customer-profile-candidates/{id}/reviews` | `memory:candidate:review` |
|
||||
| `GET /api/v1/admin/customer-service/handover-tickets` | `handover:read` |
|
||||
| `GET /api/v1/admin/customer-service/handover-tickets/{ticket_no}` | `handover:read` |
|
||||
|
||||
这三个码在 **没有种子脚本、也没有迁移** 的情况下被业务代码直接引用，
|
||||
所以它们属于**环境数据**：代码合过来以后，如果库里没有这三行，上面 5 个接口一律
|
||||
`403 缺少操作权限` —— 报错看起来像"权限配错了"，实际是**权限码根本不存在**。
|
||||
（`GET /api/v1/users/me/memory-candidates` 用的是既有的 `memory:read:self`，不受影响。）
|
||||
|
||||
这与 `AGENTS.md` §E 那条环境口径是同一类问题：
|
||||
`config_release`、`sys_role`/`sys_permission`/`sys_user`、Milvus 集合 schema、
|
||||
`advisor_product_*` **都不随代码合并**，换环境必须重放。
|
||||
|
||||
## 授给谁 —— 依据是代码里的真实校验
|
||||
|
||||
`app/service/authorization_service.py:11-17` 的语义是：
|
||||
|
||||
```python
|
||||
allowed = permission in context.permissions
|
||||
if admin:
|
||||
allowed = allowed and bool({"admin", "super_admin"}.intersection(context.roles))
|
||||
```
|
||||
|
||||
| 权限码 | 校验方式 | 授给 | 理由 |
|
||||
|---|---|---|---|
|
||||
| `memory:candidate:confirm` | `require(..., USER_CONFIRM_PERMISSION)`，**非 admin** | `customer` | 客户确认**自己**的候选；查询按 `customer_id` 过滤，不会越权 |
|
||||
| `memory:candidate:review` | `require(..., ADMIN_REVIEW_PERMISSION, admin=True)` | `admin` | 需同时满足权限与管理员角色 |
|
||||
| `handover:read` | `require(..., "handover:read", admin=True)` | `admin` | 同上 |
|
||||
|
||||
我方库里没有 `super_admin` 角色，`admin=True` 那一支由 `admin` 满足。
|
||||
|
||||
## ⚠️ 与 `seed_test_rbac.py` 的冲突（与投顾那次同源）
|
||||
|
||||
`seed_test_rbac.py` 是 **DELETE 重建**语义，它的
|
||||
`DELETE FROM sys_permission WHERE id BETWEEN 9001 AND 9099` 会**清掉本脚本建的权限**。
|
||||
所以本脚本**只增不删**，用于局部补齐；若要长期固化，请把这三个权限并进
|
||||
`seed_test_rbac.py` 的 `PERMISSIONS` 常量（投顾那 16 个也是同一个待办）。
|
||||
|
||||
权限 id 从 **9036** 起：避开种子占用的 9001-9019，也避开投顾的 9020-9035。
|
||||
|
||||
用法：
|
||||
|
||||
python tools/grant_customer_service_phase2_permissions.py --dry-run # 只打印将写入什么
|
||||
python tools/grant_customer_service_phase2_permissions.py
|
||||
"""
|
||||
|
||||
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)
|
||||
PHASE2_PERMISSIONS: tuple[tuple[int, str, str, str, str], ...] = (
|
||||
(9036, "memory:candidate:confirm", "memory:candidate", "confirm", "self"),
|
||||
(9037, "memory:candidate:review", "memory:candidate", "review", "all"),
|
||||
(9038, "handover:read", "handover", "read", "all"),
|
||||
)
|
||||
|
||||
#: 角色 id（`seed_test_rbac.py` 建的）。
|
||||
CUSTOMER_ROLE_ID = 9001
|
||||
ADMIN_ROLE_ID = 9003
|
||||
|
||||
#: 每个角色该拿到哪些码。
|
||||
GRANTS: tuple[tuple[int, str, tuple[str, ...]], ...] = (
|
||||
(CUSTOMER_ROLE_ID, "customer", ("memory:candidate:confirm",)),
|
||||
(ADMIN_ROLE_ID, "admin", ("memory:candidate:confirm", "memory:candidate:review",
|
||||
"handover:read")),
|
||||
)
|
||||
|
||||
|
||||
async def apply(*, dry_run: bool) -> int:
|
||||
now = datetime.now(UTC).replace(tzinfo=None)
|
||||
async with SessionFactory() as session, session.begin():
|
||||
existing = dict(
|
||||
(await session.execute(
|
||||
text("SELECT permission_code, id FROM sys_permission")
|
||||
)).all()
|
||||
)
|
||||
to_create = [p for p in PHASE2_PERMISSIONS if p[1] not in existing]
|
||||
print(f"权限：库里已有 {len(existing)} 个，本次新增 {len(to_create)} 个")
|
||||
for _, code, resource, action, scope in to_create:
|
||||
print(f" + {code:<32} {resource}:{action} scope={scope}")
|
||||
if not to_create:
|
||||
print(" （三个权限都已在库中）")
|
||||
|
||||
role_ids: dict[str, int] = {}
|
||||
for role_id, role_code, _codes in GRANTS:
|
||||
found = await session.scalar(
|
||||
text("SELECT id FROM sys_role WHERE role_code = :code"),
|
||||
{"code": role_code},
|
||||
)
|
||||
print(f"角色 {role_code:<10} {'已存在' if found else '缺失（请先跑 seed_test_rbac.py）'}")
|
||||
if found is not None:
|
||||
role_ids[role_code] = int(found)
|
||||
|
||||
if dry_run:
|
||||
print("\n[dry-run] 未写入任何数据。")
|
||||
return 0
|
||||
|
||||
for perm_id, code, resource, action, scope in to_create:
|
||||
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},
|
||||
)
|
||||
|
||||
# 重新取一遍完整映射，避免依赖本次插入的 id。
|
||||
permission_ids = dict(
|
||||
(await session.execute(
|
||||
text("SELECT permission_code, id FROM sys_permission")
|
||||
)).all()
|
||||
)
|
||||
|
||||
for role_id, role_code, codes in GRANTS:
|
||||
if role_code not in role_ids:
|
||||
print(f"跳过 {role_code}：角色不存在")
|
||||
continue
|
||||
have = set(
|
||||
(await session.scalars(
|
||||
text("SELECT permission_id FROM sys_role_permission WHERE role_id = :r"),
|
||||
{"r": role_ids[role_code]},
|
||||
)).all()
|
||||
)
|
||||
added = 0
|
||||
for code in codes:
|
||||
perm_id = permission_ids.get(code)
|
||||
if perm_id is None or int(perm_id) in have:
|
||||
continue
|
||||
await session.execute(
|
||||
text(
|
||||
"INSERT INTO sys_role_permission (role_id, permission_id, created_at)"
|
||||
" VALUES (:r, :p, :now)"
|
||||
),
|
||||
{"r": role_ids[role_code], "p": int(perm_id), "now": now},
|
||||
)
|
||||
added += 1
|
||||
print(f"授权：{role_code:<10} 新增 {added} 项（共 {len(codes)} 项）")
|
||||
|
||||
await verify()
|
||||
return 0
|
||||
|
||||
|
||||
async def verify() -> None:
|
||||
"""按权限码实测一遍，而不是只看插了几行。"""
|
||||
async with SessionFactory() as session:
|
||||
rows = (
|
||||
await session.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT p.permission_code, r.role_code
|
||||
FROM sys_permission p
|
||||
JOIN sys_role_permission rp ON rp.permission_id = p.id
|
||||
JOIN sys_role r ON r.id = rp.role_id
|
||||
WHERE p.permission_code IN
|
||||
('memory:candidate:confirm', 'memory:candidate:review', 'handover:read')
|
||||
ORDER BY p.permission_code, r.role_code
|
||||
"""
|
||||
)
|
||||
)
|
||||
).all()
|
||||
print("\n实测绑定关系：")
|
||||
for code, role in rows:
|
||||
print(f" {str(code):<32} → {role}")
|
||||
missing = {
|
||||
"memory:candidate:confirm", "memory:candidate:review", "handover:read",
|
||||
} - {str(code) for code, _ in rows}
|
||||
if missing:
|
||||
print(f"[失败] 仍未绑定：{sorted(missing)}")
|
||||
raise SystemExit(1)
|
||||
print(
|
||||
"\n客服二期接口现在应当可用了。若之后跑过 `seed_test_rbac.py`（DELETE 重建 9001-9099），"
|
||||
"必须重跑本脚本 —— 或先把这三个权限并进那个种子的 PERMISSIONS 常量。"
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="补客服二期的画像候选与转人工工单权限")
|
||||
parser.add_argument("--dry-run", action="store_true", help="只打印将写入什么")
|
||||
args = parser.parse_args()
|
||||
return asyncio.run(apply(dry_run=args.dry_run))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user