File size: 3,381 Bytes
76962bf | 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 | import os
import uuid
import datetime
from pymongo import MongoClient
from supabase import create_client, Client
from dotenv import load_dotenv
load_dotenv()
# MongoDB Configuration
MONGO_URI = os.getenv("MONGO_URI", "mongodb://localhost:27017/")
MONGO_DB = "medical_kb"
# Supabase Configuration
SUPABASE_URL = os.getenv("SUPABASE_URL")
# Use Service Role Key for migration to bypass RLS
SUPABASE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY") or os.getenv("SUPABASE_KEY")
DEFAULT_PATIENT_ID = os.getenv("DEFAULT_PATIENT_UUID")
def migrate():
print("Starting migration from MongoDB to Supabase...")
if not SUPABASE_URL or not SUPABASE_KEY:
print("Error: Supabase configuration missing.")
return
mongo_client = MongoClient(MONGO_URI)
mongo_db = mongo_client[MONGO_DB]
supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)
# 1. Migrate Patients
print("Migrating Patients...")
patients = list(mongo_db["patient"].find())
for p in patients:
p_id = p.get("id") or str(p.get("_id"))
if p_id == "anonymous":
p_id = DEFAULT_PATIENT_ID
resource = {
"resourceType": "Patient",
"id": p_id,
"name": p.get("name", [{"text": "Unknown"}]),
"active": p.get("active", True)
}
data = {
"id": p_id,
"resource": resource,
"last_updated": datetime.datetime.now(datetime.timezone.utc).isoformat()
}
supabase.table("patients").upsert(data).execute()
print(f"Migrated {len(patients)} patients.")
# 2. Migrate Observations
print("Migrating Observations...")
observations = list(mongo_db["observation"].find())
for o in observations:
obs_id = o.get("id") or str(o.get("_id"))
# Map patient reference
ref = o.get("subject", {}).get("reference", "")
patient_id = None
if "Patient/" in ref:
patient_id = ref.split("/")[-1]
if not patient_id or patient_id == "anonymous":
patient_id = DEFAULT_PATIENT_ID
o.pop("_id", None)
data = {
"id": obs_id,
"patient_id": patient_id,
"resource": o,
"last_updated": datetime.datetime.now(datetime.timezone.utc).isoformat()
}
supabase.table("observations").upsert(data).execute()
print(f"Migrated {len(observations)} observations.")
# 3. Migrate Communications (Chat History)
print("Migrating Communications...")
comms = list(mongo_db["patient_data"].find())
for c in comms:
c_id = c.get("id") or str(c.get("_id"))
ref = c.get("subject", {}).get("reference", "")
patient_id = None
if "Patient/" in ref:
patient_id = ref.split("/")[-1]
if not patient_id or patient_id == "anonymous":
patient_id = DEFAULT_PATIENT_ID
c.pop("_id", None)
data = {
"id": c_id,
"patient_id": patient_id,
"resource": c,
"last_updated": datetime.datetime.now(datetime.timezone.utc).isoformat()
}
supabase.table("communications").upsert(data).execute()
print(f"Migrated {len(comms)} communications.")
print("Migration complete!")
if __name__ == "__main__":
migrate()
|