48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
"""Local-only administrator provisioning. Passwords are read from the terminal, never argv."""
|
|
|
|
import argparse
|
|
import getpass
|
|
|
|
from pydantic import ValidationError
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from app.db.session import SessionLocal
|
|
from app.models.auth import User
|
|
from app.schemas.auth import RegisterInput
|
|
from app.services.auth import PASSWORD_HASHER
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="本机管理员初始化（须先完成数据库迁移）")
|
|
parser.add_argument("command", choices=["create-admin"])
|
|
parser.add_argument("--username", required=True)
|
|
args = parser.parse_args()
|
|
password = getpass.getpass("管理员密码（至少 15 字符）：")
|
|
if password != getpass.getpass("再次输入密码："):
|
|
parser.exit(1, "两次密码不一致\n")
|
|
try:
|
|
payload = RegisterInput(username=args.username, password=password)
|
|
except ValidationError:
|
|
parser.exit(1, "用户名须为 3–32 位字母开头的小写字母/数字/下划线；密码须为 15–128 字符\n")
|
|
with SessionLocal() as db:
|
|
if db.scalar(select(User.id).where(User.username == payload.username)) is not None:
|
|
parser.exit(1, "用户已存在，未修改原有用户或权限\n")
|
|
db.add(
|
|
User(
|
|
username=payload.username,
|
|
password_hash=PASSWORD_HASHER.hash(password),
|
|
role="admin",
|
|
)
|
|
)
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
parser.exit(1, "用户已存在，未修改原有用户或权限\n")
|
|
print("管理员已创建；请通过登录接口建立会话。")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|