group_fqcd_jr/tools/grant_handover_write_permission.py

"""幂等补齐 `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())