masthishq / scripts /seed_quiz_data.py
Krishnakkp's picture
services: configured LFS for assets
556a881
Raw
History Blame Contribute Delete
3.08 kB
import asyncio
import os
import random
from datetime import datetime, timedelta
from motor.motor_asyncio import AsyncIOMotorClient
from beanie import init_beanie, Document
from pydantic import BaseModel
# Mock Config
MONGO_URI = "mongodb+srv://krishkp:koushikkp123@cluster0.2wbmitl.mongodb.net/users?retryWrites=true&w=majority&appName=Cluster0"
class MemoryQuiz(Document):
patient_id: str
caregiver_id: str
timestamp: datetime
question: str
expected_answer: str
patient_response: str
accuracy_score: float
context_type: str = "general"
class Settings:
name = "memory_quizzes"
async def seed():
client = AsyncIOMotorClient(MONGO_URI)
await init_beanie(database=client.get_default_database(), document_models=[MemoryQuiz])
# Check if data exists
count = await MemoryQuiz.count()
if count > 5:
print("Data already exists. Skipping seed.")
return
print("Seeding Memory Quiz Data...")
# Mock Patient
patient_email = "test@patient.com" # Or whatever the user logs in as
caregiver_email = "admin@caregiver.com"
# Generate last 7 days of data
quizzes = []
base_time = datetime.now() - timedelta(days=7)
questions = [
("Who is in this photo?", "Son", "Son", 1.0),
("What is your address?", "123 Main St", "123 Main St", 1.0),
("What did you eat for breakfast?", "Oatmeal", "Toast", 0.0),
("Who is this?", "Daughter", "Sister", 0.5),
("What year is it?", "2026", "2026", 1.0),
("Who is the prime minister?", "Modi", "Gandhi", 0.0),
]
for i in range(20):
# Random time in last 7 days
days_offset = random.randint(0, 7)
time_offset = random.randint(0, 86400)
ts = base_time + timedelta(days=days_offset, seconds=time_offset)
q, exp, act, score = random.choice(questions)
quiz = MemoryQuiz(
patient_id=patient_email, # This needs to match the logged in user's email if possible, or we just rely on 'get_stats' fetching by patient_id if provided.
# actually app uses current_user.email.
# I'll use a generic one, and the user might need to create a user with this email or I update the script to ask.
# innovative approach: I'll actually query the users collection to find a patient? No, easier to just insert generic and tell user.
caregiver_id=caregiver_email,
timestamp=ts,
question=q,
expected_answer=exp,
patient_response=act,
accuracy_score=score + random.uniform(-0.1, 0.1) # vary slightly
)
if quiz.accuracy_score > 1.0: quiz.accuracy_score = 1.0
if quiz.accuracy_score < 0.0: quiz.accuracy_score = 0.0
quizzes.append(quiz)
# Sort by time
quizzes.sort(key=lambda x: x.timestamp)
for q in quizzes:
await q.insert()
print(f"Seeded {len(quizzes)} quiz records.")
if __name__ == "__main__":
asyncio.run(seed())