Spaces:
Sleeping
Sleeping
File size: 6,478 Bytes
c4f38f8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 |
#!/usr/bin/env python3
"""
Script to delete all users except the admin from the CPS database.
This will also clean up related data like appointments, messages, etc.
"""
import asyncio
import sys
import os
from pathlib import Path
# Add the current directory to Python path
sys.path.append(str(Path(__file__).parent))
from db.mongo import (
users_collection,
appointments_collection,
messages_collection,
chats_collection,
notifications_collection,
password_reset_codes_collection,
db
)
async def delete_all_users_except_admin():
"""Delete all users except admin and related data from the database"""
print("⚠️ WARNING: This will delete ALL users except admin and related data!")
print("This action cannot be undone.")
# Get confirmation
confirm = input("\nType 'DELETE USERS' to confirm: ")
if confirm != "DELETE USERS":
print("Operation cancelled.")
return
try:
print("\n🔄 Starting user deletion process...")
# First, find the admin user
admin_user = await users_collection.find_one({"roles": {"$in": ["admin"]}})
if not admin_user:
print("❌ No admin user found! Cannot proceed without admin user.")
return
admin_email = admin_user.get("email")
admin_id = str(admin_user.get("_id"))
print(f"✅ Found admin user: {admin_email} (ID: {admin_id})")
# Get counts before deletion
total_users = await users_collection.count_documents({})
admin_users = await users_collection.count_documents({"roles": {"$in": ["admin"]}})
doctor_users = await users_collection.count_documents({"roles": {"$in": ["doctor"]}})
patient_users = await users_collection.count_documents({"roles": {"$in": ["patient"]}})
appointment_count = await appointments_collection.count_documents({})
message_count = await messages_collection.count_documents({})
chat_count = await chats_collection.count_documents({})
notification_count = await notifications_collection.count_documents({})
password_reset_count = await password_reset_codes_collection.count_documents({})
print(f"📊 Current data counts:")
print(f" - Total Users: {total_users}")
print(f" - Admin Users: {admin_users}")
print(f" - Doctor Users: {doctor_users}")
print(f" - Patient Users: {patient_users}")
print(f" - Appointments: {appointment_count}")
print(f" - Messages: {message_count}")
print(f" - Chats: {chat_count}")
print(f" - Notifications: {notification_count}")
print(f" - Password Reset Codes: {password_reset_count}")
# Delete related data first
print("\n🗑️ Deleting related data...")
# Delete appointments (except those created by admin)
if appointment_count > 0:
result = await appointments_collection.delete_many({
"created_by": {"$ne": admin_email}
})
print(f" ✅ Deleted {result.deleted_count} appointments (non-admin)")
# Delete messages (except those from admin)
if message_count > 0:
result = await messages_collection.delete_many({
"$or": [
{"sender_id": {"$ne": admin_id}},
{"recipient_id": {"$ne": admin_id}}
]
})
print(f" ✅ Deleted {result.deleted_count} messages (non-admin)")
# Delete chats (except those involving admin)
if chat_count > 0:
result = await chats_collection.delete_many({
"user_id": {"$ne": admin_id}
})
print(f" ✅ Deleted {result.deleted_count} chats (non-admin)")
# Delete notifications (except those for admin)
if notification_count > 0:
result = await notifications_collection.delete_many({
"user_id": {"$ne": admin_id}
})
print(f" ✅ Deleted {result.deleted_count} notifications (non-admin)")
# Delete password reset codes (except admin's)
if password_reset_count > 0:
result = await password_reset_codes_collection.delete_many({
"email": {"$ne": admin_email}
})
print(f" ✅ Deleted {result.deleted_count} password reset codes (non-admin)")
# Finally, delete all users except admin
print("\n👥 Deleting all users except admin...")
result = await users_collection.delete_many({
"roles": {"$nin": ["admin"]}
})
print(f" ✅ Deleted {result.deleted_count} users (non-admin)")
# Verify deletion
print("\n🔍 Verifying deletion...")
remaining_users = await users_collection.count_documents({})
remaining_admin_users = await users_collection.count_documents({"roles": {"$in": ["admin"]}})
remaining_appointments = await appointments_collection.count_documents({})
remaining_messages = await messages_collection.count_documents({})
print(f"📊 Remaining data counts:")
print(f" - Total Users: {remaining_users}")
print(f" - Admin Users: {remaining_admin_users}")
print(f" - Appointments: {remaining_appointments}")
print(f" - Messages: {remaining_messages}")
if remaining_admin_users == 1 and remaining_users == 1:
print("✅ All users except admin have been successfully deleted!")
print(f"✅ Admin user '{admin_email}' is preserved.")
else:
print("⚠️ Some data may still remain or admin user was affected:")
print(f" - Remaining users: {remaining_users}")
print(f" - Remaining admin users: {remaining_admin_users}")
except Exception as e:
print(f"❌ Error during deletion: {str(e)}")
raise
async def main():
"""Main function"""
print("🏥 CPS User Cleanup Tool (Keep Admin Only)")
print("=" * 50)
try:
await delete_all_users_except_admin()
except KeyboardInterrupt:
print("\n\n⏹️ Operation cancelled by user.")
except Exception as e:
print(f"\n❌ Fatal error: {str(e)}")
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
|