| 1234567891011121314151617181920212223242526272829303132333435363738394041 |
- """SEED_DEMO=false 的首个管理员由终端创建；密码交互读取，避免写入命令历史。"""
- import argparse
- import asyncio
- from getpass import getpass
- from pathlib import Path
- import sys
- sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
- from sqlalchemy import select
- from app.domain.models import Department, Role, User
- from app.infrastructure.database import initialize, transaction, engine
- from app.infrastructure.security import validate_password, hash_password
- from app.shared.schemas import UserCreate
- async def main(args):
- password = getpass('管理员密码（大小写、数字、符号，至少10位）：')
- if password != getpass('再次输入密码：'):
- raise ValueError('两次密码不一致')
- validate_password(password)
- await initialize()
- async with transaction() as session:
- if await session.scalar(select(User).where(User.username == args.username)):
- raise ValueError('登录名已存在；本脚本不覆盖已有身份或密码')
- department = await session.scalar(select(Department).where(Department.name == args.department))
- if department is None:
- department = Department(name=args.department)
- session.add(department)
- await session.flush()
- if not department.active:
- raise ValueError('指定部门已停用')
- body = UserCreate(username=args.username, real_name=args.name, department_id=department.id, role='admin', password=password)
- role = await session.scalar(select(Role).where(Role.code == 'admin'))
- session.add(User(username=body.username, real_name=body.real_name, department_id=body.department_id, role_id=role.id, password_hash=hash_password(password)))
- await engine.dispose()
- print('管理员已创建；请登录后配置真实用户与审批链。')
- if __name__ == '__main__':
- parser = argparse.ArgumentParser(description='创建首个管理员，不覆盖同名账号')
- parser.add_argument('--username', required=True)
- parser.add_argument('--name', required=True)
- parser.add_argument('--department', required=True)
- asyncio.run(main(parser.parse_args()))
|
