Добавить CLI для управления пользователями (manage_users.py)

2026-08-10 00:55:07 +03:00
parent 961be91a8a
commit e520ba0a21
1 changed files with 243 additions and 0 deletions
@@ -0,0 +1,243 @@
+"""CLI для управления пользователями панели.
+
+Публичная регистрация в панели закрыта — пользователей создаёт администратор
+либо через этот скрипт, либо через эндпоинт /admin/users.
+
+Запуск внутри контейнера backend:
+    docker compose exec backend python -m app.manage_users <команда> [аргументы]
+
+Команды:
+    list                                    — показать всех пользователей
+    create <email> <password> [role]        — создать пользователя
+                                             role: admin | client (по умолчанию client)
+    activate <email>                        — активировать аккаунт (is_active=True)
+    deactivate <email>                      — деактивировать аккаунт (is_active=False)
+    set-role <email> <role>                 — сменить роль (admin | client)
+    reset-password <email> <new_password>   — задать новый пароль
+    delete <email>                          — удалить пользователя (осторожно)
+
+Примеры:
+    docker compose exec backend python -m app.manage_users list
+    docker compose exec backend python -m app.manage_users create user@example.com 'Str0ng!Pass' client
+    docker compose exec backend python -m app.manage_users set-role user@example.com admin
+    docker compose exec backend python -m app.manage_users deactivate user@example.com
+"""
+
+import argparse
+import getpass
+import sys
+from typing import Iterable
+
+from app import models
+from app.database import SessionLocal
+from app.security import hash_password
+
+VALID_ROLES = ("admin", "client")
+
+
+def _normalize_email(email: str) -> str:
+    return email.lower().strip()
+
+
+def list_users() -> Iterable[models.User]:
+    db = SessionLocal()
+    try:
+        users = db.query(models.User).order_by(models.User.created_at.desc()).all()
+        if not users:
+            print("В базе нет пользователей.")
+            return
+
+        print(f"{'ID':>4}  {'EMAIL':<32}  {'ROLE':<7}  {'ACTIVE':<6}  CREATED_AT")
+        print("-" * 78)
+        for u in users:
+            print(
+                f"{u.id:>4}  {u.email:<32}  {u.role.value:<7}  "
+                f"{'да' if u.is_active else 'нет':<6}  {u.created_at:%Y-%m-%d %H:%M}"
+            )
+    finally:
+        db.close()
+
+
+def create_user(email: str, password: str, role: str) -> None:
+    if role not in VALID_ROLES:
+        print(f"❌ Неверная роль: {role}. Допустимо: {', '.join(VALID_ROLES)}")
+        sys.exit(1)
+
+    email = _normalize_email(email)
+    if len(password) < 6:
+        print("❌ Пароль слишком короткий (минимум 6 символов).")
+        sys.exit(1)
+
+    db = SessionLocal()
+    try:
+        if db.query(models.User).filter(models.User.email == email).first():
+            print(f"❌ Пользователь {email} уже существует.")
+            sys.exit(1)
+
+        user = models.User(
+            email=email,
+            password_hash=hash_password(password),
+            role=models.Role(role),
+            is_active=True,
+        )
+        db.add(user)
+        db.commit()
+        db.refresh(user)
+        print(f"✅ Создан пользователь {email} (role={role}, id={user.id})")
+    finally:
+        db.close()
+
+
+def set_active(email: str, active: bool) -> None:
+    email = _normalize_email(email)
+    db = SessionLocal()
+    try:
+        user = db.query(models.User).filter(models.User.email == email).first()
+        if not user:
+            print(f"❌ Пользователь {email} не найден.")
+            sys.exit(1)
+        user.is_active = active
+        db.commit()
+        state = "активирован" if active else "деактивирован"
+        print(f"✅ {email}: {state}")
+    finally:
+        db.close()
+
+
+def set_role(email: str, role: str) -> None:
+    if role not in VALID_ROLES:
+        print(f"❌ Неверная роль: {role}. Допустимо: {', '.join(VALID_ROLES)}")
+        sys.exit(1)
+
+    email = _normalize_email(email)
+    db = SessionLocal()
+    try:
+        user = db.query(models.User).filter(models.User.email == email).first()
+        if not user:
+            print(f"❌ Пользователь {email} не найден.")
+            sys.exit(1)
+        # Защита от случайного понижения единственного админа.
+        if user.role == models.Role.admin and role == "client":
+            admins = (
+                db.query(models.User)
+                .filter(models.User.role == models.Role.admin, models.User.is_active.is_(True))
+                .count()
+            )
+            if admins <= 1:
+                print("❌ Нельзя понизить единственного активного администратора. Сначала назначьте другого admin.")
+                sys.exit(1)
+        user.role = models.Role(role)
+        db.commit()
+        print(f"✅ {email}: роль изменена на {role}")
+    finally:
+        db.close()
+
+
+def reset_password(email: str, new_password: str) -> None:
+    if len(new_password) < 6:
+        print("❌ Пароль слишком короткий (минимум 6 символов).")
+        sys.exit(1)
+
+    email = _normalize_email(email)
+    db = SessionLocal()
+    try:
+        user = db.query(models.User).filter(models.User.email == email).first()
+        if not user:
+            print(f"❌ Пользователь {email} не найден.")
+            sys.exit(1)
+        user.password_hash = hash_password(new_password)
+        db.commit()
+        print(f"✅ {email}: пароль изменён")
+    finally:
+        db.close()
+
+
+def delete_user(email: str) -> None:
+    email = _normalize_email(email)
+    db = SessionLocal()
+    try:
+        user = db.query(models.User).filter(models.User.email == email).first()
+        if not user:
+            print(f"❌ Пользователь {email} не найден.")
+            sys.exit(1)
+        # Защита от удаления единственного админа.
+        if user.role == models.Role.admin:
+            admins = (
+                db.query(models.User)
+                .filter(models.User.role == models.Role.admin, models.User.is_active.is_(True))
+                .count()
+            )
+            if admins <= 1:
+                print("❌ Нельзя удалить единственного активного администратора.")
+                sys.exit(1)
+        db.delete(user)
+        db.commit()
+        print(f"✅ Пользователь {email} удалён")
+    finally:
+        db.close()
+
+
+def build_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(
+        prog="manage_users",
+        description="Управление пользователями Proxmox VPS Panel.",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog=__doc__,
+    )
+    sub = p.add_subparsers(dest="command", required=True)
+
+    sub.add_parser("list", help="Список всех пользователей")
+
+    pc = sub.add_parser("create", help="Создать пользователя")
+    pc.add_argument("email")
+    pc.add_argument("password", nargs="?", help="Если не задан — спросит интерактивно")
+    pc.add_argument("role", nargs="?", default="client", choices=VALID_ROLES)
+
+    pa = sub.add_parser("activate", help="Активировать аккаунт")
+    pa.add_argument("email")
+
+    pd = sub.add_parser("deactivate", help="Деактивировать аккаунт")
+    pd.add_argument("email")
+
+    pr = sub.add_parser("set-role", help="Сменить роль пользователя")
+    pr.add_argument("email")
+    pr.add_argument("role", choices=VALID_ROLES)
+
+    pp = sub.add_parser("reset-password", help="Сменить пароль пользователя")
+    pp.add_argument("email")
+    pp.add_argument("new_password", nargs="?")
+
+    pdel = sub.add_parser("delete", help="Удалить пользователя")
+    pdel.add_argument("email")
+
+    return p
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+
+    if args.command == "list":
+        list_users()
+    elif args.command == "create":
+        password = args.password or getpass.getpass("Пароль для нового пользователя: ")
+        create_user(args.email, password, args.role)
+    elif args.command == "activate":
+        set_active(args.email, True)
+    elif args.command == "deactivate":
+        set_active(args.email, False)
+    elif args.command == "set-role":
+        set_role(args.email, args.role)
+    elif args.command == "reset-password":
+        password = args.new_password or getpass.getpass("Новый пароль: ")
+        reset_password(args.email, password)
+    elif args.command == "delete":
+        delete_user(args.email)
+    else:
+        parser.print_help()
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())