group_fqcd_jr/tools/grant_operator_role.py

"""补齐运营（`operator`）角色的权限，并按需创建该角色。

## 为什么需要它

`operator` 是场外/推广线建的角色，**不在 `seed_test_rbac.py` 的 9001-9003 里** ——
所以种子既不会创建它，也不会清理它的绑定。库里这个角色长期**只有 1 项权限**
（`offsite:write`），但两条线实际要求的并不一样：

| 功能 | 门槛类型 | 位置 |
|---|---|---|
| 场外基金运营（邮件、单据、确认、通知、结算） | **角色门槛** `{"operator","risk_operator","admin","super_admin"}` | `offsite_fund_service.py:2600` |
| 金融 NL2SQL | **权限码** `financial:nl2sql:read`（角色白名单含 `operator`） | `financial_nl2sql_service.py:272` |

也就是说：场外主体功能**本来就该能用**（靠角色），运营真正缺的是 NL2SQL 那一个码；
而"看不到运营界面"是前端没做，不是权限问题。

## 给哪些 —— 按代码真实要求，不多给

运营不做治理、不看审计、不发布配置，因此**不给** `audit:read` / `config:*` /
`product-governance:*` / `knowledge:manage`。需要排查权限缺口时跑
`python tools/check_permission_coverage.py`。

本脚本**只增不删**，可重复执行。

用法：

    python tools/grant_operator_role.py --dry-run
    python tools/grant_operator_role.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]

OPERATOR_ROLE_CODE = "operator"
OPERATOR_ROLE_NAME = "运营专员"

#: 运营该有的权限码。`offsite:write` 由场外线创建；其余在本种子的 9051-9056 号段里定义。
OPERATOR_GRANTED_CODES: tuple[str, ...] = (
    # 场外运营（角色门槛之外，这个码是场外线自己声明的）
    "offsite:write",
    # 金融 NL2SQL：角色白名单 {advisor, operator, admin, super_admin} 含 operator
    "financial:nl2sql:read",
)


async def apply(*, dry_run: bool) -> int:
    now = datetime.now(UTC).replace(tzinfo=None)
    async with SessionFactory() as session, session.begin():
        role_id = await session.scalar(
            text("SELECT id FROM sys_role WHERE role_code = :code"), {"code": OPERATOR_ROLE_CODE}
        )
        print(f"角色 {OPERATOR_ROLE_CODE}：{'已存在 id=' + str(role_id) if role_id else '将新建（自动分配 id）'}")

        permission_ids = dict(
            (await session.execute(text("SELECT permission_code, id FROM sys_permission"))).all()
        )
        missing = [code for code in OPERATOR_GRANTED_CODES if code not in permission_ids]
        print(f"权限码：库里已有 {len(permission_ids)} 个；本脚本需要的 {len(OPERATOR_GRANTED_CODES)} 个中缺 {len(missing)} 个")
        for code in missing:
            print(f"  ✗ 缺失：{code}（应先跑 tools/seed_test_rbac.py）")

        if dry_run:
            print("\n[dry-run] 未写入任何数据。")
            return 0

        if role_id is None:
            await session.execute(
                text(
                    "INSERT INTO sys_role (role_code, role_name, status, created_at, updated_at)"
                    " VALUES (:code, :name, 'active', :now, :now)"
                ),
                {"code": OPERATOR_ROLE_CODE, "name": OPERATOR_ROLE_NAME, "now": now},
            )
            role_id = await session.scalar(
                text("SELECT id FROM sys_role WHERE role_code = :code"),
                {"code": OPERATOR_ROLE_CODE},
            )
        role_id = int(role_id)

        have = set(
            (await session.scalars(
                text("SELECT permission_id FROM sys_role_permission WHERE role_id = :r"),
                {"r": role_id},
            )).all()
        )
        added = 0
        for code in OPERATOR_GRANTED_CODES:
            permission_id = permission_ids.get(code)
            if permission_id is None or int(permission_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_id, "p": int(permission_id), "now": now},
            )
            added += 1
        print(f"授权：{OPERATOR_ROLE_CODE} 新增 {added} 项（目标共 {len(OPERATOR_GRANTED_CODES)} 项）")

    await verify()
    return 0


async def verify() -> None:
    """用真实链路验证：按权限码列出该角色最终拥有什么。"""
    async with SessionFactory() as session:
        rows = (
            await session.execute(
                text(
                    """
                    SELECT p.permission_code
                    FROM sys_role r
                    JOIN sys_role_permission rp ON rp.role_id = r.id
                    JOIN sys_permission p ON p.id = rp.permission_id
                    WHERE r.role_code = :code
                    ORDER BY p.permission_code
                    """
                ),
                {"code": OPERATOR_ROLE_CODE},
            )
        ).all()
    codes = [str(row[0]) for row in rows]
    print(f"\n{OPERATOR_ROLE_CODE} 实测权限 {len(codes)} 项：{codes}")
    lacked = [c for c in OPERATOR_GRANTED_CODES if c not in codes]
    if lacked:
        print(f"[失败] 仍未绑定的码：{lacked}")
        raise SystemExit(1)
    print(
        "\n下一步：给运营账号绑这个角色 ——\n"
        "  python tools/create_test_user.py --id 9006 --username offsite_t "
        "--role operator --password offsite123"
    )


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())