88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
"""Script to create an admin user for the application."""
|
|
|
|
import asyncio
|
|
import getpass
|
|
import sys
|
|
from datetime import datetime
|
|
from uuid import uuid4
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.database import db_manager
|
|
from app.core.models.user import User
|
|
from app.core.security import hash_password
|
|
|
|
|
|
async def create_admin_user():
|
|
"""Create an admin user interactively."""
|
|
print("🔧 My Uploader Bot - Admin User Creation")
|
|
print("=" * 50)
|
|
|
|
# Get user input
|
|
username = input("Enter admin username: ").strip()
|
|
if not username:
|
|
print("❌ Username is required")
|
|
sys.exit(1)
|
|
|
|
email = input("Enter admin email: ").strip()
|
|
if not email:
|
|
print("❌ Email is required")
|
|
sys.exit(1)
|
|
|
|
password = getpass.getpass("Enter admin password: ").strip()
|
|
if not password:
|
|
print("❌ Password is required")
|
|
sys.exit(1)
|
|
|
|
confirm_password = getpass.getpass("Confirm admin password: ").strip()
|
|
if password != confirm_password:
|
|
print("❌ Passwords do not match")
|
|
sys.exit(1)
|
|
|
|
first_name = input("Enter first name (optional): ").strip() or None
|
|
last_name = input("Enter last name (optional): ").strip() or None
|
|
|
|
try:
|
|
async with db_manager.get_session() as session:
|
|
# Check if user already exists
|
|
from sqlalchemy import select
|
|
|
|
existing_user = await session.execute(
|
|
select(User).where(
|
|
(User.username == username) | (User.email == email)
|
|
)
|
|
)
|
|
if existing_user.scalar_one_or_none():
|
|
print("❌ User with this username or email already exists")
|
|
sys.exit(1)
|
|
|
|
# Create admin user
|
|
admin_user = User(
|
|
id=uuid4(),
|
|
username=username,
|
|
email=email,
|
|
password_hash=hash_password(password),
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
is_active=True,
|
|
is_verified=True,
|
|
is_superuser=True,
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow()
|
|
)
|
|
|
|
session.add(admin_user)
|
|
await session.commit()
|
|
|
|
print("✅ Admin user created successfully!")
|
|
print(f" Username: {username}")
|
|
print(f" Email: {email}")
|
|
print(f" User ID: {admin_user.id}")
|
|
print(" Status: Active, Verified, Superuser")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error creating admin user: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(create_admin_user()) |