Spaces:
Sleeping
Sleeping
File size: 4,907 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 |
#!/usr/bin/env python3
"""
Script to delete all patients from the CPS database.
This will also clean up related data like appointments, messages, analyses, 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 (
patients_collection,
appointments_collection,
messages_collection,
patient_analysis_results_collection,
chats_collection,
clinical_alerts_collection,
notifications_collection,
db
)
async def delete_all_patients():
"""Delete all patients and related data from the database"""
print("⚠️ WARNING: This will delete ALL patients and related data!")
print("This action cannot be undone.")
# Get confirmation
confirm = input("\nType 'DELETE ALL' to confirm: ")
if confirm != "DELETE ALL":
print("Operation cancelled.")
return
try:
print("\n🔄 Starting deletion process...")
# Get counts before deletion
patient_count = await patients_collection.count_documents({})
appointment_count = await appointments_collection.count_documents({})
message_count = await messages_collection.count_documents({})
analysis_count = await patient_analysis_results_collection.count_documents({})
chat_count = await chats_collection.count_documents({})
alert_count = await clinical_alerts_collection.count_documents({})
notification_count = await notifications_collection.count_documents({})
print(f"📊 Current data counts:")
print(f" - Patients: {patient_count}")
print(f" - Appointments: {appointment_count}")
print(f" - Messages: {message_count}")
print(f" - Analysis Results: {analysis_count}")
print(f" - Chats: {chat_count}")
print(f" - Clinical Alerts: {alert_count}")
print(f" - Notifications: {notification_count}")
# Delete related data first
print("\n🗑️ Deleting related data...")
# Delete appointments
if appointment_count > 0:
result = await appointments_collection.delete_many({})
print(f" ✅ Deleted {result.deleted_count} appointments")
# Delete messages
if message_count > 0:
result = await messages_collection.delete_many({})
print(f" ✅ Deleted {result.deleted_count} messages")
# Delete analysis results
if analysis_count > 0:
result = await patient_analysis_results_collection.delete_many({})
print(f" ✅ Deleted {result.deleted_count} analysis results")
# Delete chats
if chat_count > 0:
result = await chats_collection.delete_many({})
print(f" ✅ Deleted {result.deleted_count} chats")
# Delete clinical alerts
if alert_count > 0:
result = await clinical_alerts_collection.delete_many({})
print(f" ✅ Deleted {result.deleted_count} clinical alerts")
# Delete notifications
if notification_count > 0:
result = await notifications_collection.delete_many({})
print(f" ✅ Deleted {result.deleted_count} notifications")
# Finally, delete all patients
print("\n👥 Deleting all patients...")
if patient_count > 0:
result = await patients_collection.delete_many({})
print(f" ✅ Deleted {result.deleted_count} patients")
# Verify deletion
print("\n🔍 Verifying deletion...")
remaining_patients = await patients_collection.count_documents({})
remaining_appointments = await appointments_collection.count_documents({})
remaining_messages = await messages_collection.count_documents({})
if remaining_patients == 0 and remaining_appointments == 0 and remaining_messages == 0:
print("✅ All patients and related data have been successfully deleted!")
else:
print("⚠️ Some data may still remain:")
print(f" - Remaining patients: {remaining_patients}")
print(f" - Remaining appointments: {remaining_appointments}")
print(f" - Remaining messages: {remaining_messages}")
except Exception as e:
print(f"❌ Error during deletion: {str(e)}")
raise
async def main():
"""Main function"""
print("🏥 CPS Database Cleanup Tool")
print("=" * 40)
try:
await delete_all_patients()
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())
|