create_super_admin.py

"""
Script to create the first Super Admin user
Run this once to initialize the platform with a Super Admin account
Usage: python create_super_admin.py
"""
import asyncio
from datetime import datetime
from pymongo import MongoClient
from passlib.context import CryptContext
from config import settings

# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


async def create_first_super_admin():
    """Create the first Super Admin user in the database"""
    
    print("\n" + "="*80)
    print("SUPER ADMIN INITIALIZATION SCRIPT")
    print("="*80 + "\n")
    
    # Connect to MongoDB
    try:
        client = MongoClient(settings.MONGODB_URI)
        db = client[settings.MONGODB_DB_NAME]
        users_collection = db["users"]
        print("[[SUCCESS]] Connected to MongoDB successfully")
    except Exception as e:
        print(f"[✗] Failed to connect to MongoDB: {str(e)}")
        return
    
    # Check if super admin already exists
    existing_super_admin = users_collection.find_one({"role": "super_admin"})
    if existing_super_admin:
        print(f"\n[!] Super Admin already exists: {existing_super_admin['email']}")
        print("[!] Aborting to prevent duplicate Super Admin accounts")
        client.close()
        return
    
    # Get Super Admin credentials
    print("\nEnter Super Admin credentials:")
    print("-" * 40)
    
    email = input("Email: ").strip()
    if not email or "@" not in email:
        print("[✗] Invalid email address")
        client.close()
        return
    
    # Check if email already exists
    existing_user = users_collection.find_one({"email": email})
    if existing_user:
        print(f"[!] User with email '{email}' already exists")
        
        # Ask if want to promote existing user to super_admin
        promote = input("\nPromote this user to Super Admin? (yes/no): ").strip().lower()
        if promote == "yes":
            result = users_collection.update_one(
                {"email": email},
                {"$set": {"role": "super_admin", "updated_at": datetime.utcnow()}}
            )
            
            if result.modified_count > 0:
                print(f"\n[[SUCCESS]] User '{email}' promoted to Super Admin successfully!")
            else:
                print("[✗] Failed to promote user")
            
            client.close()
            return
        else:
            print("[!] Operation cancelled")
            client.close()
            return
    
    password = input("Password: ").strip()
    if len(password) < 8:
        print("[✗] Password must be at least 8 characters long")
        client.close()
        return
    
    name = input("Full Name: ").strip()
    if not name:
        print("[✗] Full name is required")
        client.close()
        return
    
    phone = input("Phone (10 digits): ").strip()
    if not phone or len(phone) != 10 or not phone.isdigit():
        print("[✗] Phone must be exactly 10 digits")
        client.close()
        return
    
    # Hash password
    hashed_password = pwd_context.hash(password)
    
    # Create Super Admin user document
    super_admin_user = {
        "email": email,
        "password_hash": hashed_password,
        "name": name,
        "phone": phone,
        "role": "super_admin",
        "is_active": True,
        "wallet_id": None,
        "deleted": False,
        "kyc_status": "approved",  # Pre-approved
        "created_at": datetime.utcnow(),
        "updated_at": datetime.utcnow()
    }
    
    try:
        # Insert Super Admin
        result = users_collection.insert_one(super_admin_user)
        
        if result.inserted_id:
            print("\n" + "="*80)
            print("[SUCCESS] SUPER ADMIN CREATED SUCCESSFULLY!")
            print("="*80)
            print(f"\nEmail: {email}")
            print(f"Name: {name}")
            print(f"Phone: {phone}")
            print(f"Role: super_admin")
            print(f"User ID: {str(result.inserted_id)}")
            print("\n[!] Please save these credentials securely!")
            print("="*80 + "\n")
        else:
            print("\n[✗] Failed to create Super Admin")
            
    except Exception as e:
        print(f"\n[✗] Error creating Super Admin: {str(e)}")
    
    finally:
        client.close()


if __name__ == "__main__":
    asyncio.run(create_first_super_admin())