109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
|
"""命令行管理工具。
|
|||
|
|
|
|||
|
|
用法（在 backend/ 目录下执行）::
|
|||
|
|
|
|||
|
|
uv run python -m app.cli create-admin --username admin
|
|||
|
|
|
|||
|
|
密码优先**交互式输入**而非命令行参数：命令行参数会进入 shell 历史，
|
|||
|
|
在多用户机器上还能被 `ps` 看到。`--password` 仅为自动化脚本保留。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import argparse
|
|||
|
|
import getpass
|
|||
|
|
import sys
|
|||
|
|
|
|||
|
|
from sqlalchemy import select
|
|||
|
|
|
|||
|
|
from app.core.constants import PASSWORD_MAX_BYTES, PASSWORD_MIN_BYTES, ROLE_ADMIN
|
|||
|
|
from app.core.db import dispose_engine, get_session_factory
|
|||
|
|
from app.core.security import hash_password
|
|||
|
|
from app.models import Role, User, UserRole
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _prompt_password() -> str:
|
|||
|
|
"""交互式读取并二次确认密码，循环直到符合策略。"""
|
|||
|
|
while True:
|
|||
|
|
password = getpass.getpass(
|
|||
|
|
f"请输入密码（{PASSWORD_MIN_BYTES}-{PASSWORD_MAX_BYTES} 字节）: "
|
|||
|
|
)
|
|||
|
|
length = len(password.encode("utf-8"))
|
|||
|
|
if length < PASSWORD_MIN_BYTES:
|
|||
|
|
print(f" 密码不足 {PASSWORD_MIN_BYTES} 字节，请重新输入。")
|
|||
|
|
continue
|
|||
|
|
if length > PASSWORD_MAX_BYTES:
|
|||
|
|
print(f" 密码超过 {PASSWORD_MAX_BYTES} 字节上限（中文一字占 3 字节），请重新输入。")
|
|||
|
|
continue
|
|||
|
|
if password != getpass.getpass("请再次输入以确认: "):
|
|||
|
|
print(" 两次输入不一致，请重新输入。")
|
|||
|
|
continue
|
|||
|
|
return password
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _create_admin(args: argparse.Namespace) -> int:
|
|||
|
|
"""幂等创建管理员账号。
|
|||
|
|
|
|||
|
|
已存在时**不覆盖密码** —— 意外重跑脚本不应把已有账号的密码改掉。
|
|||
|
|
"""
|
|||
|
|
factory = get_session_factory()
|
|||
|
|
with factory() as session:
|
|||
|
|
existing = session.execute(
|
|||
|
|
select(User).where(User.username == args.username)
|
|||
|
|
).scalar_one_or_none()
|
|||
|
|
if existing is not None:
|
|||
|
|
print(f"用户 {args.username!r} 已存在（id={existing.id}），未做任何修改。")
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
role = session.execute(
|
|||
|
|
select(Role).where(Role.role_code == ROLE_ADMIN)
|
|||
|
|
).scalar_one_or_none()
|
|||
|
|
if role is None:
|
|||
|
|
print(
|
|||
|
|
"错误：未找到 admin 角色，请先执行数据库迁移（uv run alembic upgrade head）。",
|
|||
|
|
file=sys.stderr,
|
|||
|
|
)
|
|||
|
|
return 1
|
|||
|
|
|
|||
|
|
password = args.password or _prompt_password()
|
|||
|
|
user = User(
|
|||
|
|
username=args.username,
|
|||
|
|
password=hash_password(password),
|
|||
|
|
real_name=args.real_name,
|
|||
|
|
status=1,
|
|||
|
|
)
|
|||
|
|
session.add(user)
|
|||
|
|
session.flush() # 取得自增 id 后才能建关联
|
|||
|
|
session.add(UserRole(user_id=user.id, role_id=role.id))
|
|||
|
|
session.commit()
|
|||
|
|
|
|||
|
|
print(f"已创建管理员 {args.username!r}（id={user.id}），角色：{role.role_name}。")
|
|||
|
|
print("首次登录后请立即修改密码。")
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main() -> int:
|
|||
|
|
"""命令行入口。"""
|
|||
|
|
parser = argparse.ArgumentParser(
|
|||
|
|
prog="app.cli", description="衍水集团营销人员工资管理系统 - 命令行工具"
|
|||
|
|
)
|
|||
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|||
|
|
|
|||
|
|
p_admin = sub.add_parser("create-admin", help="创建初始管理员账号（幂等）")
|
|||
|
|
p_admin.add_argument("--username", default="admin", help="登录用户名，默认 admin")
|
|||
|
|
p_admin.add_argument("--real-name", default="系统管理员", help="真实姓名")
|
|||
|
|
p_admin.add_argument(
|
|||
|
|
"--password",
|
|||
|
|
default=None,
|
|||
|
|
help="密码。不传则交互式输入（推荐）——命令行传参会留在 shell 历史中",
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
args = parser.parse_args()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
return _create_admin(args)
|
|||
|
|
finally:
|
|||
|
|
dispose_engine()
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
raise SystemExit(main())
|