group_fqcd_jr/tools/grant_advisor_role.py

"""建立投顾（`advisor`）角色并授权。

## 为什么需要它

投顾这条线合并进来后，`bootstrap.py` 有 **10 处 `allowed_roles` 引用了 `advisor`**，
`financial_nl2sql_service.py:272` 还硬编码检查 `{"advisor","operator","admin","super_admin"}`，
`promotion_material_service.py:164` 直接按 `"advisor" in context.roles` 走业务分支 ——
但 `sys_role` 里**没有这个角色**，`sys_permission` 里也没有投顾那 16 个权限码
（`seed_test_rbac.py` 只建到 9019）。结果是：投顾登录后拿不到任何投顾权限，
所有投顾接口一律 403，而报错看起来像"权限配错了"，实际是**角色根本不存在**。

## 权限怎么分

| 类别 | 权限码 | 给谁 |
|---|---|---|
| 投顾工作流（10） | `asset-allocation:generate:self`、`investment-goal:read:self` / `:review` / `:publish`、`portfolio-analysis:read:self`、`product-comparison:read:self`、`product-recommendation:read:self` / `:generate:self` / `:review` / `:publish` | `advisor` + `admin` |
| 治理类（6） | `asset-allocation:backtest`、`product-governance:read` / `:review` / `:sync`、`profile-governance:read` / `:review` | **只给 `admin`** |

`review` / `publish` 也给投顾，与项目既有决策一致 —— 此前已裁定**不做双人复核**
（`admin` 发布配置时也是"创建人自审"）。治理类不给投顾：那是平台侧的活。

## ⚠️ 与 `seed_test_rbac.py` 的冲突

那个脚本是 **DELETE 重建**语义，它的
`DELETE FROM sys_permission WHERE id BETWEEN 9001 AND 9099` 会**清掉本脚本建的权限**
（本脚本用 9020-9035）。将来若要在种子里固化投顾权限，请把它并进
`seed_test_rbac.py` 的 `PERMISSIONS` 常量，而不是只跑本脚本。

本脚本自身**只增不删**：重复执行只补齐缺失项，不动任何已有绑定。

用法：

    python tools/grant_advisor_role.py --dry-run   # 只打印将写入什么
    python tools/grant_advisor_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]

#: 角色 id 用 9004：`seed_test_rbac.py` 只重建 9001-9003，不会碰它。
ADVISOR_ROLE_ID = 9004
ADVISOR_ROLE_CODE = "advisor"
ADVISOR_ROLE_NAME = "投资顾问"

#: 权限 id 从 9020 起，避开种子已用的 9001-9019。
#: (id, permission_code, resource, action, data_scope)
ADVISOR_PERMISSIONS: tuple[tuple[int, str, str, str, str], ...] = (
    (9020, "asset-allocation:generate:self", "asset-allocation", "generate", "self"),
    (9021, "asset-allocation:backtest", "asset-allocation", "backtest", "all"),
    (9022, "investment-goal:read:self", "investment-goal", "read", "self"),
    (9023, "investment-goal:review", "investment-goal", "review", "all"),
    (9024, "investment-goal:publish", "investment-goal", "publish", "all"),
    (9025, "portfolio-analysis:read:self", "portfolio-analysis", "read", "self"),
    (9026, "product-comparison:read:self", "product-comparison", "read", "self"),
    (9027, "product-recommendation:read:self", "product-recommendation", "read", "self"),
    (9028, "product-recommendation:generate:self", "product-recommendation", "generate", "self"),
    (9029, "product-recommendation:review", "product-recommendation", "review", "all"),
    (9030, "product-recommendation:publish", "product-recommendation", "publish", "all"),
    (9031, "product-governance:read", "product-governance", "read", "all"),
    (9032, "product-governance:review", "product-governance", "review", "all"),
    (9033, "product-governance:sync", "product-governance", "sync", "all"),
    (9034, "profile-governance:read", "profile-governance", "read", "all"),
    (9035, "profile-governance:review", "profile-governance", "review", "all"),
)

#: 投顾拿哪些 —— 工作流那 10 个；治理类 6 个只给 admin。
ADVISOR_GRANTED_CODES: tuple[str, ...] = (
    "asset-allocation:generate:self",
    "investment-goal:read:self",
    "investment-goal:review",
    "investment-goal:publish",
    "portfolio-analysis:read:self",
    "product-comparison:read:self",
    "product-recommendation:read:self",
    "product-recommendation:generate:self",
    "product-recommendation:review",
    "product-recommendation:publish",
)

#: admin 角色 id（`seed_test_rbac.py` 建的）。
ADMIN_ROLE_ID = 9003


async def apply(*, dry_run: bool) -> int:
    now = datetime.now(UTC).replace(tzinfo=None)
    async with SessionFactory() as session, session.begin():
        existing_codes = set(
            (await session.scalars(
                text("SELECT permission_code FROM sys_permission")
            )).all()
        )
        to_create = [p for p in ADVISOR_PERMISSIONS if p[1] not in existing_codes]
        print(f"权限：已存在 {len(existing_codes)} 个，本次新增 {len(to_create)} 个")
        for _, code, resource, action, scope in to_create:
            print(f"  + {code:<44} {resource}:{action} scope={scope}")

        role_exists = await session.scalar(
            text("SELECT id FROM sys_role WHERE role_code = :code"),
            {"code": ADVISOR_ROLE_CODE},
        )
        print(f"角色 {ADVISOR_ROLE_CODE}：{'已存在' if role_exists else '将新建（id=9004）'}")

        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},
            )

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

        permission_ids = dict(
            (await session.execute(
                text("SELECT permission_code, id FROM sys_permission")
            )).all()
        )

        # 授权：先查已有绑定，只补缺失的（只增不删）。
        async def grant(role: int, codes: tuple[str, ...]) -> int:
            have = set(
                (await session.scalars(
                    text("SELECT permission_id FROM sys_role_permission WHERE role_id = :r"),
                    {"r": role},
                )).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, "p": int(perm_id), "now": now},
                )
                added += 1
            return added

        all_codes = tuple(code for _, code, _, _, _ in ADVISOR_PERMISSIONS)
        advisor_added = await grant(role_id, ADVISOR_GRANTED_CODES)
        admin_added = await grant(ADMIN_ROLE_ID, all_codes)
        print(f"授权：advisor 新增 {advisor_added} 项（共 {len(ADVISOR_GRANTED_CODES)} 项）")
        print(f"      admin   新增 {admin_added} 项（共 {len(all_codes)} 项）")

    await verify()
    return 0


async def verify() -> None:
    """用真实链路验证：角色与权限能被解析出来。"""
    async with SessionFactory() as session:
        rows = (
            await session.execute(
                text(
                    """
                    SELECT r.role_code, COUNT(rp.permission_id) AS n
                    FROM sys_role r
                    LEFT JOIN sys_role_permission rp ON rp.role_id = r.id
                    GROUP BY r.id, r.role_code ORDER BY r.role_code
                    """
                )
            )
        ).mappings().all()
    print("\n各角色权限数（实测）：")
    for row in rows:
        print(f"  {str(row['role_code']):<16} {int(row['n'])} 项")
    if not any(str(row["role_code"]) == ADVISOR_ROLE_CODE for row in rows):
        print("[失败] advisor 角色没有建成功")
        raise SystemExit(1)
    print(
        "\n下一步：给某个人绑这个角色 ——\n"
        "  python tools/create_test_user.py --id 9020 --username advisor_t "
        "--role advisor --password abc12345"
    )


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