group_fqcd_jr/tools/set_user_password.py

"""设置用户登录密码（bcrypt 哈希写入 `sys_user.password_hash`）。

## ⚠️ 仅限演示环境

本脚本把演示口令写在源码里、也允许明文命令行传入，目的是让演示与联调**当天可用**。
`123456` / `666666` / `88888888` 这类弱口令**在生产环境等于没有密码**：
上线前必须全部更换，并由运维走单独的改密流程（本脚本只服务演示）。

## 为什么需要它

`sys_user.password_hash` 此前**全是占位符** —— 种子写 `'x'`、worker 身份写
`!worker-only-no-password-login!`，即"这个字段从来没过真实密码"。登录接口上线后，
不设密码就没人能登进来；这个脚本补的正是这一步。

## 用法

    python tools/set_user_password.py --list                  # 只列现状，不改任何数据
    python tools/set_user_password.py                         # 按内置演示规则设置
    python tools/set_user_password.py --user 9002 --password 'xxx'

注意：bcrypt 每次加盐不同，**重复执行等于重设密码**（不是"已存在就跳过"）。这是有意的
——改密本来就该覆盖，但要清楚它不是幂等操作。
"""

from __future__ import annotations

import argparse
import asyncio
import sys
from datetime import UTC, datetime

import bcrypt
from sqlalchemy import text

from app.infrastructure.db import SessionFactory

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

#: 演示口令（用户指定）。键是 `sys_user.id`。
#: 9001 客户 / 9002 员工（风控专员）/ 9003 管理员。
DEMO_PASSWORDS: dict[str, str] = {
    "9001": "123456",
    "9002": "666666",
    "9003": "88888888",
}

#: 与 `AuthService.verify_password` 保持一致的识别方式：只有 bcrypt 格式才算"已设真密码"。
BCRYPT_PREFIXES = ("$2a$", "$2b$", "$2y$")


def hash_password(password: str) -> str:
    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")


def _is_real_hash(value: str | None) -> bool:
    return bool(value) and str(value).startswith(BCRYPT_PREFIXES)


async def list_users() -> None:
    async with SessionFactory() as session:
        rows = (
            await session.execute(
                text(
                    "SELECT id, username, user_type, status, password_hash "
                    "FROM sys_user ORDER BY id"
                )
            )
        ).mappings().all()
    print(f"{'id':<8}{'username':<18}{'user_type':<12}{'status':<8}密码状态")
    for row in rows:
        state = "已设（bcrypt）" if _is_real_hash(row["password_hash"]) else "占位符，无法登录"
        print(
            f"{row['id']:<8}{str(row['username']):<18}{str(row['user_type']):<12}"
            f"{str(row['status']):<8}{state}"
        )


async def set_password(user_id: str, password: str) -> int:
    now = datetime.now(UTC).replace(tzinfo=None)
    async with SessionFactory() as session, session.begin():
        result = await session.execute(
            text(
                "UPDATE sys_user SET password_hash = :hash, updated_at = :now "
                "WHERE id = :user_id"
            ),
            {"hash": hash_password(password), "now": now, "user_id": int(user_id)},
        )
        if result.rowcount == 0:
            print(f"[失败] sys_user 里没有 id={user_id} 的用户")
            return 1
    print(f"[OK] id={user_id} 密码已设置（bcrypt）")
    return 0


async def main() -> int:
    parser = argparse.ArgumentParser(description="设置用户登录密码（bcrypt）")
    parser.add_argument("--list", action="store_true", help="只列现状，不改数据")
    parser.add_argument("--user", help="单个用户 id（配合 --password 使用）")
    parser.add_argument("--password", help="要设置的明文密码")
    args = parser.parse_args()

    if args.list:
        await list_users()
        return 0

    if args.user or args.password:
        if not (args.user and args.password):
            print("[失败] --user 与 --password 必须成对给出")
            return 1
        return await set_password(args.user, args.password)

    print("按内置演示规则设置密码（生产环境必须更换）：")
    failures = 0
    for user_id, password in DEMO_PASSWORDS.items():
        print(f"  · id={user_id} → {len(password)} 位口令")
        failures += await set_password(user_id, password)
    print("\n设置后的现状：")
    await list_users()
    if failures:
        return 1
    print("\n可以登录了。接口：POST /api/v1/auth/tokens  {\"username\": \"<username>\", \"password\": \"...\"}")
    return 0


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