group_fqcd_jr/tools/create_test_user.py

"""添加一个可登录的测试账号（用户 + 角色 + 密码），并**验证它真的能拿到权限**。

## 为什么需要它

`sys_user` / `sys_user_role` / `sys_role` 这些表**没有 ORM 模型**（全项目用裸 SQL 访问，
见 `tools/seed_test_rbac.py`），手写 INSERT 要凑齐 10 个字段、还要自己算密码哈希。
更要紧的是这里有个**静默陷阱**：

    MySQL 的 DATETIME(0) 会把微秒**四舍五入到秒**。若 `sys_user_role.assigned_at`
    用"当前时间"写入，进位后可能落在未来，而授权校验是 `assigned_at <= now` ——
    于是刚建好的账号**一个角色都拿不到**，表现为 `roles=()`，**不报错也不失败**。

`seed_test_rbac.py` 已经踩过一次（它的注释里写着）。本脚本统一把 `assigned_at`
往前留 5 秒，并在最后**用 `IdentityService.resolve` 打印真实解析结果**而不是
"插入成功" —— 后者根本不能说明这个账号能用。

## 用法

    python tools/create_test_user.py --list
    python tools/create_test_user.py --id 9010 --username test_cust \\
        --role customer --password abc12345

重复执行同一个 `--id` 是**覆盖**语义：更新用户名 / 密码 / 角色，不会产生重复行。
"""

from __future__ import annotations

import argparse
import asyncio
import sys
from datetime import UTC, datetime, timedelta
from pathlib import Path

from sqlalchemy import text

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from app.core.contracts import RequestContext  # noqa: E402
from app.infrastructure.db import SessionFactory  # noqa: E402
from app.service.auth_service import hash_password  # noqa: E402
from app.service.identity_service import IdentityService  # noqa: E402

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(errors="replace")  # type: ignore[union-attr]

#: 可创建的角色。**不再硬编码角色 id** —— 角色 id 是环境数据
#: （`operator` 就是自增出来的 `1986594485028610`，各环境不同），硬编码会让脚本换个
#: 环境就失效。现在运行时按 `role_code` 查库。
#: 新角色得先有权限集合（`sys_role_permission`）：`advisor` 由
#: `tools/grant_advisor_role.py` 建，`operator` 由 `tools/grant_operator_role.py` 补。
KNOWN_ROLES: tuple[str, ...] = (
    "customer", "risk_operator", "admin", "advisor", "operator",
)

#: 角色 → `sys_user.user_type`。注意这是 `user_type`，与 `employee_role` 不是一回事。
ROLE_USER_TYPE: dict[str, str] = {
    "customer": "customer",
    "risk_operator": "employee",
    "admin": "employee",
    "advisor": "employee",
    "operator": "employee",
}

#: 客户的开户状态。风控扫描等链路会读它，写成 `closed` 会让部分规则不成立。
FUND_ACCOUNT_STATUS: dict[str, str] = {
    "customer": "已开户",
    "employee": "closed",
}

#: `assigned_at` 往前留的秒数，见模块文档里的静默陷阱。
ASSIGN_BACKDATE_SECONDS = 5


async def list_users() -> None:
    """列出所有账号、它的角色与密码状态。"""
    async with SessionFactory() as session:
        rows = (
            await session.execute(
                text(
                    """
                    SELECT u.id, u.username, u.user_type, u.status,
                           u.password_hash,
                           GROUP_CONCAT(r.role_code ORDER BY r.role_code) AS roles
                    FROM sys_user u
                    LEFT JOIN sys_user_role ur ON ur.user_id = u.id
                    LEFT JOIN sys_role r ON r.id = ur.role_id
                    GROUP BY u.id, u.username, u.user_type, u.status, u.password_hash
                    ORDER BY u.id
                    """
                )
            )
        ).mappings().all()

    print(f"{'id':<8}{'username':<18}{'user_type':<12}{'status':<8}{'roles':<28}密码")
    for row in rows:
        has_password = str(row["password_hash"] or "").startswith(("$2a$", "$2b$", "$2y$"))
        print(
            f"{row['id']:<8}{str(row['username']):<18}{str(row['user_type']):<12}"
            f"{str(row['status']):<8}{str(row['roles'] or '(无角色)') or '(无角色)':<28}"
            f"{'已设' if has_password else '占位符，无法登录'}"
        )


async def upsert_user(
    *, user_id: int, username: str, role: str, password: str
) -> int:
    """建/更新账号并绑定角色，最后验证权限能解析出来。"""
    user_type = ROLE_USER_TYPE[role]
    now = datetime.now(UTC).replace(tzinfo=None)
    assigned_at = now - timedelta(seconds=ASSIGN_BACKDATE_SECONDS)

    async with SessionFactory() as session, session.begin():
        role_id = await session.scalar(
            text("SELECT id FROM sys_role WHERE role_code = :code"), {"code": role}
        )
        if role_id is None:
            available = (
                await session.scalars(text("SELECT role_code FROM sys_role ORDER BY id"))
            ).all()
            print(f"[失败] 库里没有角色 {role}。现有角色：{list(available)}")
            return 1
        role_id = int(role_id)

        # 覆盖语义：同一个 id 重跑不会堆出第二行。
        await session.execute(
            text(
                """
                INSERT INTO sys_user
                    (id, user_no, username, password_hash, user_type,
                     professional_investor_status, fund_account_status, status,
                     created_at, updated_at)
                VALUES
                    (:id, :user_no, :username, :password_hash, :user_type,
                     'none', :fund_status, '正常', :now, :now)
                AS new
                ON DUPLICATE KEY UPDATE
                    username = new.username,
                    password_hash = new.password_hash,
                    user_type = new.user_type,
                    fund_account_status = new.fund_account_status,
                    status = '正常',
                    updated_at = new.updated_at
                """
            ),
            {
                "id": user_id,
                "user_no": f"T-{username.upper()[:20]}",
                "username": username,
                "password_hash": hash_password(password),
                "user_type": user_type,
                "fund_status": FUND_ACCOUNT_STATUS[user_type],
                "now": now,
            },
        )
        # 角色绑定先清后插。表上其实**有** `uk_sys_user_role (user_id, role_id)` 唯一键
        # （`tools/generate_baseline_sql.py:128`），所以直接 INSERT 不会堆重复行 ——
        # 先清的理由是**支持改角色**：不清的话换角色会在表里留下两条绑定，
        # `IdentityService` 解析出的 roles 就变成两个（既是 customer 又是 admin）。
        await session.execute(
            text("DELETE FROM sys_user_role WHERE user_id = :user_id"), {"user_id": user_id}
        )
        await session.execute(
            text(
                "INSERT INTO sys_user_role (user_id, role_id, assigned_at)"
                " VALUES (:user_id, :role_id, :assigned_at)"
            ),
            {"user_id": user_id, "role_id": role_id, "assigned_at": assigned_at},
        )

    print(f"[OK] id={user_id} username={username} role={role} 已写入")
    return await verify(user_id, username, role)


async def verify(user_id: int, username: str, role: str) -> int:
    """用真实链路解析身份 —— 这是唯一能证明"这个账号能用的"方式。"""
    context = await IdentityService().resolve(
        RequestContext(user_id=str(user_id), trace_id="create-test-user")
    )
    print(f"     解析结果：roles={context.roles} data_scope={context.data_scope}")
    print(f"     权限 {len(context.permissions)} 项")
    if role not in context.roles:
        print(
            "[失败] 角色没有解析出来。最可能的原因是 assigned_at 落在了未来"
            "（DATETIME(0) 的毫秒舍入），请重跑本脚本。"
        )
        return 1
    print(f"[OK] 可以用它登录：{{\"username\": \"{username}\", \"password\": \"<你刚设的>\"}}")
    return 0


async def main() -> int:
    parser = argparse.ArgumentParser(description="添加可登录的测试账号")
    parser.add_argument("--list", action="store_true", help="列出所有账号与角色")
    parser.add_argument("--id", type=int, help="用户 id（9001-9003 已被演示账号占用）")
    parser.add_argument("--username", help="登录用户名")
    parser.add_argument("--role", choices=KNOWN_ROLES, help="角色（库里需已存在该角色）")
    parser.add_argument("--password", help="登录密码（仅限演示环境）")
    args = parser.parse_args()

    if args.list:
        await list_users()
        return 0

    missing = [
        name
        for name, value in (
            ("--id", args.id), ("--username", args.username),
            ("--role", args.role), ("--password", args.password),
        )
        if value is None
    ]
    if missing:
        print(f"[失败] 缺少参数：{' '.join(missing)}（或直接用 --list 看现有账号）")
        return 1

    return await upsert_user(
        user_id=args.id,
        username=args.username,
        role=args.role,
        password=args.password,
    )


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))