396 lines
17 KiB
Python
396 lines
17 KiB
Python
"""
|
|
Demo users initialization module for ClassroomCopilot
|
|
Creates demo teachers and students
|
|
"""
|
|
import os
|
|
import json
|
|
import requests
|
|
import time
|
|
from typing import Dict, Any
|
|
from modules.logger_tool import initialise_logger
|
|
from modules.database.services.provisioning_service import ProvisioningService
|
|
|
|
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), 'default', True)
|
|
|
|
class DemoUsersInitializer:
|
|
"""Handles demo users creation"""
|
|
|
|
def __init__(self, supabase_url: str, service_role_key: str):
|
|
self.supabase_url = supabase_url
|
|
self.service_role_key = service_role_key
|
|
self.supabase_headers = {
|
|
"apikey": service_role_key,
|
|
"Authorization": f"Bearer {service_role_key}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
self.provisioning_service = ProvisioningService()
|
|
|
|
def create_demo_users(self) -> Dict[str, Any]:
|
|
"""Create demo teachers and students"""
|
|
logger.info("Creating demo users...")
|
|
|
|
try:
|
|
# Define demo users
|
|
demo_users = [
|
|
# Demo Teachers
|
|
{
|
|
"email": "teacher1@kevlarai.edu",
|
|
"password": "DemoTeacher123!",
|
|
"email_confirm": True,
|
|
"user_metadata": {
|
|
"name": "Dr. Sarah Chen",
|
|
"username": "sarah.chen",
|
|
"full_name": "Dr. Sarah Chen",
|
|
"display_name": "Dr. Chen",
|
|
"user_type": "teacher"
|
|
},
|
|
"app_metadata": {
|
|
"provider": "email",
|
|
"providers": ["email"]
|
|
}
|
|
},
|
|
{
|
|
"email": "teacher2@kevlarai.edu",
|
|
"password": "DemoTeacher123!",
|
|
"email_confirm": True,
|
|
"user_metadata": {
|
|
"name": "Prof. Marcus Rodriguez",
|
|
"username": "marcus.rodriguez",
|
|
"full_name": "Professor Marcus Rodriguez",
|
|
"display_name": "Prof. Rodriguez",
|
|
"user_type": "teacher"
|
|
},
|
|
"app_metadata": {
|
|
"provider": "email",
|
|
"providers": ["email"]
|
|
}
|
|
},
|
|
# Demo Students
|
|
{
|
|
"email": "student1@kevlarai.edu",
|
|
"password": "DemoStudent123!",
|
|
"email_confirm": True,
|
|
"user_metadata": {
|
|
"name": "Alex Thompson",
|
|
"username": "alex.thompson",
|
|
"full_name": "Alex Thompson",
|
|
"display_name": "Alex",
|
|
"user_type": "student"
|
|
},
|
|
"app_metadata": {
|
|
"provider": "email",
|
|
"providers": ["email"]
|
|
}
|
|
},
|
|
{
|
|
"email": "student2@kevlarai.edu",
|
|
"password": "DemoStudent123!",
|
|
"email_confirm": True,
|
|
"user_metadata": {
|
|
"name": "Jordan Lee",
|
|
"username": "jordan.lee",
|
|
"full_name": "Jordan Lee",
|
|
"display_name": "Jordan",
|
|
"user_type": "student"
|
|
},
|
|
"app_metadata": {
|
|
"provider": "email",
|
|
"providers": ["email"]
|
|
}
|
|
}
|
|
]
|
|
|
|
created_users = []
|
|
failed_users = []
|
|
|
|
for user_data in demo_users:
|
|
try:
|
|
# Create user via Auth API
|
|
response = self._supabase_request_with_retry(
|
|
'post',
|
|
f"{self.supabase_url}/auth/v1/admin/users",
|
|
headers=self.supabase_headers,
|
|
json=user_data
|
|
)
|
|
|
|
if response.status_code in (200, 201):
|
|
user = response.json()
|
|
user_id = user.get("id")
|
|
|
|
# Wait a moment for user to be created
|
|
time.sleep(1)
|
|
|
|
# Create profile
|
|
profile_data = {
|
|
"id": user_id,
|
|
"email": user_data["email"],
|
|
"user_type": user_data["user_metadata"]["user_type"],
|
|
"username": user_data["user_metadata"]["username"],
|
|
"full_name": user_data["user_metadata"]["full_name"],
|
|
"display_name": user_data["user_metadata"]["display_name"]
|
|
}
|
|
|
|
profile_response = self._supabase_request_with_retry(
|
|
'post',
|
|
f"{self.supabase_url}/rest/v1/profiles",
|
|
headers=self.supabase_headers,
|
|
json=profile_data
|
|
)
|
|
|
|
if profile_response.status_code in (200, 201):
|
|
created_users.append({
|
|
"id": user_id,
|
|
"email": user_data["email"],
|
|
"user_type": user_data["user_metadata"]["user_type"],
|
|
"username": user_data["user_metadata"]["username"]
|
|
})
|
|
logger.info(f"Successfully created user: {user_data['email']}")
|
|
else:
|
|
logger.warning(f"Failed to create profile for {user_data['email']}: {profile_response.text}")
|
|
failed_users.append({
|
|
"email": user_data["email"],
|
|
"error": f"Profile creation failed: {profile_response.text}"
|
|
})
|
|
else:
|
|
logger.warning(f"Failed to create user {user_data['email']}: {response.text}")
|
|
failed_users.append({
|
|
"email": user_data["email"],
|
|
"error": f"User creation failed: {response.text}"
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error creating user {user_data['email']}: {str(e)}")
|
|
failed_users.append({
|
|
"email": user_data["email"],
|
|
"error": str(e)
|
|
})
|
|
|
|
# Create institute memberships for KevlarAI and provision users
|
|
all_users_to_provision = []
|
|
|
|
# Add newly created users
|
|
if created_users:
|
|
all_users_to_provision.extend(created_users)
|
|
self._create_institute_memberships(created_users)
|
|
|
|
# Also provision existing users that failed due to email_exists
|
|
existing_users = []
|
|
for failed_user in failed_users:
|
|
if "email_exists" in failed_user.get("error", ""):
|
|
# Get the existing user ID from Supabase
|
|
existing_user_id = self._get_existing_user_id(failed_user["email"])
|
|
if existing_user_id:
|
|
existing_users.append({
|
|
"id": existing_user_id,
|
|
"email": failed_user["email"],
|
|
"user_type": self._get_user_type_from_email(failed_user["email"]),
|
|
"username": self._get_username_from_email(failed_user["email"])
|
|
})
|
|
|
|
if existing_users:
|
|
logger.info(f"Found {len(existing_users)} existing users to provision")
|
|
all_users_to_provision.extend(existing_users)
|
|
self._create_institute_memberships(existing_users)
|
|
|
|
# Provision all users (new and existing)
|
|
if all_users_to_provision:
|
|
self._provision_users(all_users_to_provision)
|
|
|
|
logger.info(f"Demo users creation completed: {len(created_users)} created, {len(failed_users)} failed")
|
|
|
|
return {
|
|
"success": True,
|
|
"message": f"Successfully created {len(created_users)} demo users",
|
|
"created_users": created_users,
|
|
"failed_users": failed_users
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error creating demo users: {str(e)}")
|
|
return {
|
|
"success": False,
|
|
"message": f"Error creating demo users: {str(e)}"
|
|
}
|
|
|
|
def _create_institute_memberships(self, users: list) -> None:
|
|
"""Create institute memberships for users in KevlarAI"""
|
|
logger.info("Creating institute memberships for demo users...")
|
|
|
|
try:
|
|
# Get KevlarAI institute ID
|
|
response = self._supabase_request_with_retry(
|
|
'get',
|
|
f"{self.supabase_url}/rest/v1/institutes",
|
|
headers=self.supabase_headers,
|
|
params={
|
|
"select": "id",
|
|
"name": "eq.KevlarAI"
|
|
}
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
logger.warning("Could not get KevlarAI institute ID for memberships")
|
|
return
|
|
|
|
institutes = response.json()
|
|
if not institutes:
|
|
logger.warning("KevlarAI institute not found for memberships")
|
|
return
|
|
|
|
institute_id = institutes[0]["id"]
|
|
|
|
# Get user profile IDs
|
|
for user in users:
|
|
try:
|
|
profile_response = self._supabase_request_with_retry(
|
|
'get',
|
|
f"{self.supabase_url}/rest/v1/profiles",
|
|
headers=self.supabase_headers,
|
|
params={
|
|
"select": "id",
|
|
"email": f"eq.{user['email']}"
|
|
}
|
|
)
|
|
|
|
if profile_response.status_code == 200:
|
|
profiles = profile_response.json()
|
|
if profiles:
|
|
profile_id = profiles[0]["id"]
|
|
|
|
# Create membership
|
|
membership_data = {
|
|
"profile_id": profile_id,
|
|
"institute_id": institute_id,
|
|
"role": user["user_type"]
|
|
}
|
|
|
|
membership_response = self._supabase_request_with_retry(
|
|
'post',
|
|
f"{self.supabase_url}/rest/v1/institute_memberships",
|
|
headers=self.supabase_headers,
|
|
json=membership_data
|
|
)
|
|
|
|
if membership_response.status_code in (200, 201):
|
|
logger.info(f"Created membership for {user['email']} in KevlarAI")
|
|
else:
|
|
logger.warning(f"Failed to create membership for {user['email']}: {membership_response.text}")
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error creating membership for {user['email']}: {str(e)}")
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error creating institute memberships: {str(e)}")
|
|
|
|
def _get_existing_user_id(self, email: str) -> str:
|
|
"""Get the user ID for an existing user by email"""
|
|
try:
|
|
response = self._supabase_request_with_retry(
|
|
'get',
|
|
f"{self.supabase_url}/rest/v1/profiles",
|
|
headers=self.supabase_headers,
|
|
params={
|
|
"select": "id",
|
|
"email": f"eq.{email}"
|
|
}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
profiles = response.json()
|
|
if profiles and len(profiles) > 0:
|
|
return profiles[0].get("id")
|
|
|
|
logger.warning(f"Could not find existing user ID for {email}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error getting existing user ID for {email}: {str(e)}")
|
|
return None
|
|
|
|
def _get_user_type_from_email(self, email: str) -> str:
|
|
"""Get user type from email based on demo user definitions"""
|
|
if "teacher" in email:
|
|
return "teacher"
|
|
elif "student" in email:
|
|
return "student"
|
|
return "teacher" # default
|
|
|
|
def _get_username_from_email(self, email: str) -> str:
|
|
"""Get username from email based on demo user definitions"""
|
|
username_map = {
|
|
"teacher1@kevlarai.edu": "sarah.chen",
|
|
"teacher2@kevlarai.edu": "marcus.rodriguez",
|
|
"student1@kevlarai.edu": "alex.thompson",
|
|
"student2@kevlarai.edu": "jordan.lee"
|
|
}
|
|
return username_map.get(email, email.split("@")[0])
|
|
|
|
def _provision_users(self, users: list) -> None:
|
|
"""Provision Neo4j databases for the created demo users."""
|
|
for user in users:
|
|
user_id = user.get("id")
|
|
if not user_id:
|
|
continue
|
|
try:
|
|
self.provisioning_service.ensure_user(user_id)
|
|
logger.info(f"Provisioned Neo4j resources for {user.get('email')}")
|
|
except Exception as exc:
|
|
logger.warning(f"Failed to provision Neo4j resources for {user.get('email')}: {exc}")
|
|
|
|
def _supabase_request_with_retry(self, method, url, **kwargs):
|
|
"""Make a request to Supabase with retry logic"""
|
|
max_retries = 3
|
|
retry_delay = 2 # seconds
|
|
|
|
for attempt in range(max_retries):
|
|
try:
|
|
if method.lower() == 'get':
|
|
response = requests.get(url, **kwargs)
|
|
elif method.lower() == 'post':
|
|
response = requests.post(url, **kwargs)
|
|
elif method.lower() == 'put':
|
|
response = requests.put(url, **kwargs)
|
|
elif method.lower() == 'delete':
|
|
response = requests.delete(url, **kwargs)
|
|
else:
|
|
raise ValueError(f"Unsupported HTTP method: {method}")
|
|
|
|
# If successful or client error (4xx), don't retry
|
|
if response.status_code < 500:
|
|
return response
|
|
|
|
# Server error (5xx), retry after delay
|
|
logger.warning(f"Supabase server error (attempt {attempt+1}/{max_retries}): {response.status_code} - {response.text}")
|
|
time.sleep(retry_delay * (attempt + 1)) # Exponential backoff
|
|
|
|
except requests.RequestException as e:
|
|
logger.warning(f"Supabase request exception (attempt {attempt+1}/{max_retries}): {str(e)}")
|
|
if attempt == max_retries - 1:
|
|
raise
|
|
time.sleep(retry_delay * (attempt + 1))
|
|
|
|
# If we get here, all retries failed with server errors
|
|
raise requests.RequestException(f"Failed after {max_retries} attempts to {method} {url}")
|
|
|
|
def initialize_demo_users() -> Dict[str, Any]:
|
|
"""Initialize demo users"""
|
|
logger.info("Starting demo users initialization...")
|
|
|
|
supabase_url = os.getenv("SUPABASE_URL")
|
|
service_role_key = os.getenv("SERVICE_ROLE_KEY")
|
|
|
|
if not supabase_url or not service_role_key:
|
|
return {"success": False, "message": "Missing SUPABASE_URL or SERVICE_ROLE_KEY environment variables"}
|
|
|
|
initializer = DemoUsersInitializer(supabase_url, service_role_key)
|
|
|
|
# Create demo users
|
|
result = initializer.create_demo_users()
|
|
|
|
if result["success"]:
|
|
logger.info("Demo users initialization completed successfully")
|
|
else:
|
|
logger.error(f"Demo users initialization failed: {result['message']}")
|
|
|
|
return result
|