ClassroomCopilot/api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
api/run/initialization/seed_environment.py
kcar abf8d05ca1 feat(phase-b): Supabase-first timetable, classes, enrollment, and student views 
- timetable_builder_router: Supabase-primary slot write (POST /timetable/slots),
  week_cycle support, GET /slots reads from Supabase, materialize-periods endpoint,
  rebuild-neo4j endpoint, sync-lessons endpoint (Track B: TaughtLesson Neo4j nodes),
  _sync_teacher_timetables_to_neo4j and _sync_taught_lessons_to_neo4j helpers
- classes_router: GET /{class_id} enriched with profiles + enrollment_requests,
  GET /school/students for admin search, PATCH /enrollment-requests/{id} approve/reject
- taught_lessons_router: GET /student/lessons student week view with enrichment
- school_router: academic_periods sync, day-type management
- platform_admin_router + platform_admin: POST /admin/reset and /admin/seed endpoints
- invitations_router: teacher invite scaffolding
- reset_environment + seed_environment: idempotent dev environment scripts
- graph_tree_router: Supabase-first institute resolution
- provisioning_service: neo4j_private_db_name column support
- main.py + run/routers.py: register new routers

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-27 02:55:44 +01:00

440 lines
18 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
seed_environment.py — idempotent full-environment rebuild.
Assumes reset_environment.py has already been run (or it's the first boot).
Safe to re-run: all writes use UPSERT / MERGE.
Schools
-------
KevlarAI 6585bf91-6ae8-4d72-ab54-cddf3ba4e648 kevlarai.test
Greenfield Academy a1b2c3d4-e5f6-7890-abcd-ef1234567890 greenfieldacademy.test
Uniform accounts per school (10 × 2 = 20 total)
------------------------------------------------
admin@{domain} school_admin
head@{domain} school_admin
physics@{domain} teacher
maths@{domain} teacher
teacher1@{domain} teacher
teacher2@{domain} teacher
teacher3@{domain} teacher
student1@{domain} student
student2@{domain} student
student3@{domain} student
Run from inside the ccapi container:
python3 -c "from run.initialization.seed_environment import seed; seed()"
"""
import os
import time
import requests
from datetime import datetime, timedelta
from typing import Dict, Any, List, Optional
from modules.logger_tool import initialise_logger
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
# ─── School constants ─────────────────────────────────────────────────────────
KEVLARAI_ID = "6585bf91-6ae8-4d72-ab54-cddf3ba4e648"
KEVLARAI_NAME = "KevlarAI"
KEVLARAI_URN = "KEVLARAI-001"
KEVLARAI_DOMAIN = "kevlarai.test"
GREENFIELD_ID = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
GREENFIELD_NAME = "Greenfield Academy"
GREENFIELD_URN = "TEST-GFA-001"
GREENFIELD_DOMAIN = "greenfieldacademy.test"
# ─── Passwords ────────────────────────────────────────────────────────────────
PWD_ADMIN = "Admin@Cc2025!"
PWD_TEACHER = "Teacher@Cc2025!"
PWD_STUDENT = "Student@Cc2025!"
# ─── Account template ────────────────────────────────────────────────────────
def _school_accounts(domain: str, institute_id: str) -> List[Dict]:
return [
# school_admin accounts
{
"prefix": "admin", "email": f"admin@{domain}",
"full_name": "Alex Admin", "display_name": "Alex",
"username": f"admin.{domain.replace('.', '_')}",
"user_type": "teacher", "role": "school_admin", "password": PWD_ADMIN,
"institute_id": institute_id,
},
{
"prefix": "head", "email": f"head@{domain}",
"full_name": "Helen Head", "display_name": "Helen",
"username": f"head.{domain.replace('.', '_')}",
"user_type": "teacher", "role": "school_admin", "password": PWD_ADMIN,
"institute_id": institute_id,
},
# teacher accounts
{
"prefix": "physics", "email": f"physics@{domain}",
"full_name": "Phil Physics", "display_name": "Phil",
"username": f"physics.{domain.replace('.', '_')}",
"user_type": "teacher", "role": "teacher", "password": PWD_TEACHER,
"institute_id": institute_id,
},
{
"prefix": "maths", "email": f"maths@{domain}",
"full_name": "Mary Maths", "display_name": "Mary",
"username": f"maths.{domain.replace('.', '_')}",
"user_type": "teacher", "role": "teacher", "password": PWD_TEACHER,
"institute_id": institute_id,
},
{
"prefix": "teacher1", "email": f"teacher1@{domain}",
"full_name": "Tom Teacher", "display_name": "Tom",
"username": f"teacher1.{domain.replace('.', '_')}",
"user_type": "teacher", "role": "teacher", "password": PWD_TEACHER,
"institute_id": institute_id,
},
{
"prefix": "teacher2", "email": f"teacher2@{domain}",
"full_name": "Tara Teach", "display_name": "Tara",
"username": f"teacher2.{domain.replace('.', '_')}",
"user_type": "teacher", "role": "teacher", "password": PWD_TEACHER,
"institute_id": institute_id,
},
{
"prefix": "teacher3", "email": f"teacher3@{domain}",
"full_name": "Tim Teachwell", "display_name": "Tim",
"username": f"teacher3.{domain.replace('.', '_')}",
"user_type": "teacher", "role": "teacher", "password": PWD_TEACHER,
"institute_id": institute_id,
},
# student accounts
{
"prefix": "student1", "email": f"student1@{domain}",
"full_name": "Sam Student", "display_name": "Sam",
"username": f"student1.{domain.replace('.', '_')}",
"user_type": "student", "role": "student", "password": PWD_STUDENT,
"institute_id": institute_id,
},
{
"prefix": "student2", "email": f"student2@{domain}",
"full_name": "Sophie Study", "display_name": "Sophie",
"username": f"student2.{domain.replace('.', '_')}",
"user_type": "student", "role": "student", "password": PWD_STUDENT,
"institute_id": institute_id,
},
{
"prefix": "student3", "email": f"student3@{domain}",
"full_name": "Steve Scholar", "display_name": "Steve",
"username": f"student3.{domain.replace('.', '_')}",
"user_type": "student", "role": "student", "password": PWD_STUDENT,
"institute_id": institute_id,
},
]
ALL_ACCOUNTS = (
_school_accounts(KEVLARAI_DOMAIN, KEVLARAI_ID) +
_school_accounts(GREENFIELD_DOMAIN, GREENFIELD_ID)
)
# ─── Supabase helpers ─────────────────────────────────────────────────────────
def _sb_ctx():
url = os.environ["SUPABASE_URL"]
key = os.environ["SERVICE_ROLE_KEY"]
headers = {
"apikey": key,
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
}
return url, headers
def _auth_post(url, headers, path, data):
return requests.post(f"{url}/auth/v1/admin{path}", headers=headers, json=data)
def _auth_get(url, headers, path, params=None):
r = requests.get(f"{url}/auth/v1/admin{path}", headers=headers, params=params)
r.raise_for_status()
return r.json()
def _rest_upsert(url, headers, table, data, on_conflict):
h = {**headers, "Prefer": "resolution=merge-duplicates,return=representation"}
r = requests.post(
f"{url}/rest/v1/{table}",
headers=h,
json=data,
params={"on_conflict": on_conflict},
)
return r
def _rest_patch(url, headers, table, match_col, match_val, data):
r = requests.patch(
f"{url}/rest/v1/{table}",
headers={**headers, "Prefer": "return=minimal"},
params={match_col: f"eq.{match_val}"},
json=data,
)
return r
# ─── Main seed function ───────────────────────────────────────────────────────
def seed() -> Dict[str, Any]:
from modules.database.services.provisioning_service import ProvisioningService
from modules.database.services.neo4j_service import Neo4jService
from modules.database.init.init_calendar import create_calendar
url, headers = _sb_ctx()
errors: List[str] = []
results: Dict[str, Any] = {}
# ── Step 1: Fix KevlarAI institute record ─────────────────────────────────
logger.info("=" * 60)
logger.info("SEED ENVIRONMENT")
logger.info("=" * 60)
logger.info("\n[1] KevlarAI institute record...")
try:
r = _rest_upsert(url, headers, "institutes", {
"id": KEVLARAI_ID,
"name": KEVLARAI_NAME,
"urn": KEVLARAI_URN,
"status": "active",
"website": "https://kevlarai.com",
"address": {"line1": "1 AI Lane", "city": "London", "postcode": "EC1A 1BB"},
"metadata": {"headteacher": "Alex Admin", "seeded": True},
}, on_conflict="id")
if r.status_code in (200, 201):
logger.info(" KevlarAI upserted ✓")
else:
raise Exception(r.text[:200])
except Exception as e:
errors.append(f"kevlarai_institute: {e}")
logger.error(f" {e}")
# ── Step 2: Create Greenfield Academy if needed ───────────────────────────
logger.info("[2] Greenfield Academy institute record...")
try:
neo4j_uuid_greenfield = GREENFIELD_ID.replace("-", "")
r = _rest_upsert(url, headers, "institutes", {
"id": GREENFIELD_ID,
"name": GREENFIELD_NAME,
"urn": GREENFIELD_URN,
"status": "active",
"website": "https://greenfieldacademy.test",
"address": {"line1": "1 Academy Road", "city": "Testville", "postcode": "TE1 1ST"},
"metadata": {"headteacher": "Alex Admin", "seeded": True},
"neo4j_uuid_string": neo4j_uuid_greenfield,
}, on_conflict="id")
if r.status_code in (200, 201):
logger.info(" Greenfield Academy upserted ✓")
else:
raise Exception(r.text[:200])
except Exception as e:
errors.append(f"greenfield_institute: {e}")
logger.error(f" {e}")
# ── Step 3: Provision Neo4j for both schools ──────────────────────────────
logger.info("[3] Neo4j school provisioning...")
provisioner = ProvisioningService()
school_dbs: Dict[str, str] = {}
for iid, name in [(KEVLARAI_ID, "KevlarAI"), (GREENFIELD_ID, "Greenfield Academy")]:
try:
result = provisioner.ensure_school(iid)
db = result["db_name"]
school_dbs[iid] = db
logger.info(f" {name}: {db} ✓")
except Exception as e:
errors.append(f"ensure_school {name}: {e}")
logger.error(f" {name}: {e}")
# derive fallback db name
school_dbs[iid] = f"cc.institutes.{iid.replace('-', '')}"
# ── Step 4: Rebuild classroomcopilot global calendar ─────────────────────
logger.info("[4] classroomcopilot global calendar (2024–2028)...")
try:
neo4j_svc = Neo4jService()
neo4j_svc.create_database("classroomcopilot")
logger.info(" DB created, waiting 5s for availability...")
time.sleep(5)
start_dt = datetime(2024, 1, 1)
end_dt = datetime(2028, 12, 31)
create_calendar("classroomcopilot", start_dt, end_dt)
logger.info(" Calendar built ✓")
results["global_calendar"] = "ok"
except Exception as e:
errors.append(f"global_calendar: {e}")
logger.error(f" {e}")
results["global_calendar"] = "error"
# ── Step 5: Create / verify auth users ────────────────────────────────────
logger.info("[5] Creating auth users (20 accounts)...")
try:
existing = _auth_get(url, headers, "/users", {"per_page": 200}).get("users", [])
existing_by_email = {u["email"]: u for u in existing}
except Exception as e:
errors.append(f"list_auth_users: {e}")
existing_by_email = {}
created_users: Dict[str, str] = {} # email → uid
for spec in ALL_ACCOUNTS:
email = spec["email"]
if email in existing_by_email:
created_users[email] = existing_by_email[email]["id"]
logger.info(f" {email}: exists [{created_users[email][:8]}]")
continue
r = _auth_post(url, headers, "/users", {
"email": email,
"password": spec["password"],
"email_confirm": True,
"user_metadata": {
"username": spec["username"],
"full_name": spec["full_name"],
"display_name": spec["display_name"],
"user_type": spec["user_type"],
},
})
if r.status_code in (200, 201):
uid = r.json()["id"]
created_users[email] = uid
logger.info(f" {email}: created [{uid[:8]}]")
else:
errors.append(f"create {email}: {r.text[:150]}")
logger.error(f" {email}: {r.text[:150]}")
time.sleep(0.2)
results["users_created"] = len(created_users)
# ── Step 6: Upsert profiles and memberships ───────────────────────────────
logger.info("[6] Upserting profiles and memberships...")
for spec in ALL_ACCOUNTS:
uid = created_users.get(spec["email"])
if not uid:
continue
try:
_rest_upsert(url, headers, "profiles", {
"id": uid,
"email": spec["email"],
"user_type": spec["user_type"],
"username": spec["username"],
"full_name": spec["full_name"],
"display_name": spec["display_name"],
"school_id": spec["institute_id"],
"neo4j_sync_status": "pending",
}, on_conflict="id")
_rest_upsert(url, headers, "institute_memberships", {
"profile_id": uid,
"institute_id": spec["institute_id"],
"role": spec["role"],
"metadata": {},
}, on_conflict="profile_id,institute_id")
except Exception as e:
errors.append(f"profile/membership {spec['email']}: {e}")
logger.error(f" {spec['email']}: {e}")
logger.info(" Profiles and memberships upserted ✓")
# ── Step 7: Merge Neo4j Teacher/Student nodes ─────────────────────────────
logger.info("[7] Merging Neo4j worker nodes...")
try:
from neo4j import GraphDatabase
driver = GraphDatabase.driver("bolt://192.168.0.209:7687", auth=("neo4j", "&%N304j&%"))
# Group by institute DB
by_db: Dict[str, List[Dict]] = {}
for spec in ALL_ACCOUNTS:
uid = created_users.get(spec["email"])
if not uid:
continue
db = school_dbs.get(spec["institute_id"])
if not db:
continue
by_db.setdefault(db, []).append({**spec, "uid": uid})
for db, users in by_db.items():
with driver.session(database=db) as s:
for u in users:
label = "Teacher" if u["user_type"] == "teacher" else "Student"
s.run(
f"MERGE (n:{label} {{uuid_string: $uid}}) "
"SET n.worker_email = $email, "
" n.worker_name = $name, "
" n.worker_type = $utype",
uid=u["uid"], email=u["email"],
name=u["full_name"], utype=u["user_type"],
)
logger.info(f" [{db[:35]}] {len(users)} nodes merged ✓")
driver.close()
results["neo4j_nodes"] = "ok"
except Exception as e:
errors.append(f"neo4j_nodes: {e}")
logger.error(f" {e}")
results["neo4j_nodes"] = "error"
# ── Ensure kcar is a platform super-admin ─────────────────────────────────
logger.info("[8] Ensuring kcar platform admin record...")
KCAR_ID = "d9e1d1a9-04c4-4611-bb05-57babf4a9a28"
try:
_rest_upsert(url, headers, "admin_profiles", {
"id": KCAR_ID,
"email": "kcar@kevlarai.com",
"role": "super_admin",
"permissions": ["all"],
"metadata": {"seeded": True},
}, on_conflict="id")
logger.info(" kcar → admin_profiles ✓")
except Exception as e:
errors.append(f"kcar_admin: {e}")
logger.error(f" {e}")
# ── Summary ───────────────────────────────────────────────────────────────
results["success"] = len(errors) == 0
results["errors"] = errors
_print_credential_sheet(created_users)
logger.info("\n" + "=" * 60)
if errors:
logger.info(f"SEED COMPLETE with {len(errors)} error(s)")
for e in errors:
logger.info(f" ✗ {e}")
else:
logger.info("SEED COMPLETE — all steps succeeded")
logger.info("=" * 60)
return results
def _print_credential_sheet(created_users: Dict[str, str]):
PAD = 36
logger.info("\n" + "=" * 70)
logger.info("CREDENTIAL SHEET")
logger.info("=" * 70)
logger.info(f" {'ROLE':<16} {'EMAIL':<{PAD}} PASSWORD")
logger.info(f" {'-'*14} {'-'*(PAD-2)} -----------")
logger.info(f" {'[platform admin]':<16} {'kcar@kevlarai.com':<{PAD}} KevlarAI2025!")
logger.info("")
for school_id, domain, label in [
(KEVLARAI_ID, KEVLARAI_DOMAIN, "KevlarAI"),
(GREENFIELD_ID, GREENFIELD_DOMAIN, "Greenfield Academy"),
]:
logger.info(f" [{label}]")
for spec in ALL_ACCOUNTS:
if spec["institute_id"] != school_id:
continue
uid = created_users.get(spec["email"], "—")
status = f"[{uid[:8]}]" if uid != "—" else "[MISSING]"
logger.info(f" {spec['role']:<16} {spec['email']:<{PAD}} {spec['password']} {status}")
logger.info("")
logger.info("=" * 70)
if __name__ == "__main__":
import json
print(json.dumps(seed(), indent=2, default=str))
Reference in New Issue View Git Blame Copy Permalink