kcar
fe3d7a12c8
New routers (all previously untracked): - graph_tree_router: /graph/tree, /graph/node/children, /graph/calendar/academic Supabase-driven tree builder; institute DB resolved by teacher email scan - school_router: /school/status (role + calendar flags), /school/info PATCH Self-heals profiles.school_id from institute_memberships if null - timetable_builder_router: /timetable/setup (AcademicYear/Term/Week + SchoolTimetable), /timetable/slots (read/write TimetableSlot nodes), /timetable/init (TeacherTimetable) - user_init_router: /user/init (provision user node in institute DB) routers.py: register all new routers with correct prefixes users.py: add JournalNode and PlannerNode schema classes Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
368 lines
15 KiB
Python
368 lines
15 KiB
Python
"""
|
|
Timetable Builder Router
|
|
Endpoints for setting up academic year structure and teacher timetable slots.
|
|
"""
|
|
import os
|
|
import json
|
|
from datetime import datetime, timedelta, date
|
|
from typing import Dict, Any, List, Optional
|
|
from fastapi import APIRouter, Depends
|
|
from pydantic import BaseModel
|
|
from modules.logger_tool import initialise_logger
|
|
from modules.auth.supabase_bearer import SupabaseBearer
|
|
import modules.database.tools.neo4j_driver_tools as driver_tools
|
|
|
|
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), 'default', True)
|
|
router = APIRouter()
|
|
|
|
|
|
# ─── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
def _find_teacher_institute(user_email: str):
|
|
if not user_email:
|
|
return None, None
|
|
try:
|
|
with driver_tools.get_session(database="system") as s:
|
|
dbs = [r["name"] for r in s.run(
|
|
"SHOW DATABASES YIELD name "
|
|
"WHERE name STARTS WITH 'cc.institutes.' "
|
|
"AND NOT name ENDS WITH '.curriculum' RETURN name"
|
|
)]
|
|
for db in dbs:
|
|
try:
|
|
with driver_tools.get_session(database=db) as s:
|
|
rec = s.run(
|
|
"MATCH (t:Teacher) WHERE t.worker_email = $e "
|
|
"RETURN t.uuid_string AS uuid LIMIT 1", e=user_email
|
|
).single()
|
|
if rec and rec["uuid"]:
|
|
return db, rec["uuid"]
|
|
except Exception:
|
|
continue
|
|
except Exception as e:
|
|
logger.warning(f"Institute lookup failed: {e}")
|
|
return None, None
|
|
|
|
|
|
def _iso_date(s: str) -> date:
|
|
return datetime.strptime(s, "%Y-%m-%d").date()
|
|
|
|
|
|
def _academic_weeks(term_start: date, term_end: date):
|
|
# First Monday >= term_start
|
|
current = term_start - timedelta(days=term_start.weekday())
|
|
if current < term_start:
|
|
current += timedelta(weeks=1)
|
|
n = 1
|
|
while current <= term_end:
|
|
yield n, current
|
|
current += timedelta(weeks=1)
|
|
n += 1
|
|
|
|
|
|
# ─── Request models ───────────────────────────────────────────────────────────
|
|
|
|
class TermInput(BaseModel):
|
|
name: str
|
|
term_number: int
|
|
start_date: str
|
|
end_date: str
|
|
|
|
class PeriodInput(BaseModel):
|
|
code: str
|
|
name: str
|
|
start_time: str
|
|
end_time: str
|
|
period_type: str # lesson | break | registration
|
|
|
|
class TimetableSetupRequest(BaseModel):
|
|
year_start: str
|
|
year_end: str
|
|
terms: List[TermInput]
|
|
periods: List[PeriodInput]
|
|
|
|
class SlotInput(BaseModel):
|
|
day_of_week: str
|
|
period_code: str
|
|
subject_class: str
|
|
start_time: str
|
|
end_time: str
|
|
|
|
class SlotsRequest(BaseModel):
|
|
timetable_id: str
|
|
slots: List[SlotInput]
|
|
|
|
|
|
# ─── Endpoints ────────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/status")
|
|
async def get_timetable_status(credentials: dict = Depends(SupabaseBearer())) -> Dict[str, Any]:
|
|
"""Check whether the teacher has a timetable set up."""
|
|
user_email = credentials.get("email", "")
|
|
institute_db, teacher_uuid = _find_teacher_institute(user_email)
|
|
if not institute_db:
|
|
return {"status": "no_school"}
|
|
try:
|
|
with driver_tools.get_session(database=institute_db) as s:
|
|
has_tt = s.run(
|
|
"MATCH (t:Teacher {uuid_string: $u})-[:HAS_TIMETABLE]->(tt:TeacherTimetable) "
|
|
"RETURN tt.uuid_string AS id LIMIT 1", u=teacher_uuid
|
|
).single()
|
|
has_school_tt = s.run(
|
|
"MATCH (st:SchoolTimetable) RETURN st.uuid_string AS id LIMIT 1"
|
|
).single()
|
|
return {
|
|
"status": "ok",
|
|
"has_teacher_timetable": has_tt is not None,
|
|
"timetable_id": has_tt["id"] if has_tt else None,
|
|
"has_school_timetable": has_school_tt is not None,
|
|
"institute_db": institute_db,
|
|
}
|
|
except Exception as e:
|
|
return {"status": "error", "message": str(e)}
|
|
|
|
|
|
@router.post("/setup")
|
|
async def setup_timetable(
|
|
body: TimetableSetupRequest,
|
|
credentials: dict = Depends(SupabaseBearer())
|
|
) -> Dict[str, Any]:
|
|
"""Create academic year/term/week structure + TeacherTimetable in the institute DB."""
|
|
user_email = credentials.get("email", "")
|
|
user_id = credentials.get("sub", "")
|
|
institute_db, teacher_uuid = _find_teacher_institute(user_email)
|
|
|
|
# school_admin may have no Teacher node — fall back to Supabase for institute_db
|
|
if not institute_db and user_id:
|
|
try:
|
|
from modules.database.supabase.utils.client import SupabaseServiceRoleClient
|
|
_sb = SupabaseServiceRoleClient()
|
|
_p = _sb.supabase.table("profiles").select("school_id").eq("id", user_id).single().execute()
|
|
_sid = (_p.data or {}).get("school_id")
|
|
if _sid:
|
|
_i = _sb.supabase.table("institutes").select("neo4j_uuid_string").eq("id", _sid).single().execute()
|
|
_uuid = (_i.data or {}).get("neo4j_uuid_string")
|
|
if _uuid:
|
|
institute_db = f"cc.institutes.{_uuid}"
|
|
except Exception as _e:
|
|
logger.warning(f"Supabase institute fallback failed: {_e}")
|
|
|
|
if not institute_db:
|
|
return {"status": "error", "message": "Institute database not found"}
|
|
|
|
year_start = _iso_date(body.year_start)
|
|
year_end = _iso_date(body.year_end)
|
|
year_label = f"{year_start.year}-{year_end.year}"
|
|
school_tt_id = f"school_tt_{year_start.year}_{year_end.year}"
|
|
ay_id = f"academic_year_{year_start.year}_{year_end.year}"
|
|
try:
|
|
with driver_tools.get_session(database=institute_db) as s:
|
|
# SchoolTimetable
|
|
s.run("""
|
|
MERGE (tt:SchoolTimetable {uuid_string: $id})
|
|
SET tt.school_timetable_id = $id,
|
|
tt.start_date = date($start), tt.end_date = date($end),
|
|
tt.node_storage_path = $path,
|
|
tt.periods_template = $periods
|
|
WITH tt
|
|
MATCH (sch:School)
|
|
MERGE (sch)-[:HAS_TIMETABLE]->(tt)
|
|
""", id=school_tt_id, start=body.year_start, end=body.year_end,
|
|
path=f"timetable/{school_tt_id}",
|
|
periods=json.dumps([p.dict() for p in body.periods]))
|
|
|
|
# AcademicYear
|
|
s.run("""
|
|
MERGE (ay:AcademicYear {uuid_string: $id})
|
|
SET ay.year = $year, ay.node_storage_path = $path
|
|
WITH ay
|
|
MATCH (tt:SchoolTimetable {uuid_string: $tt_id})
|
|
MERGE (tt)-[:ACADEMIC_TIMETABLE_HAS_ACADEMIC_YEAR]->(ay)
|
|
""", id=ay_id, year=year_label, path=f"timetable/{ay_id}", tt_id=school_tt_id)
|
|
|
|
# Terms + weeks
|
|
for term in body.terms:
|
|
t_id = f"term_{year_start.year}_{term.term_number}"
|
|
t_start = _iso_date(term.start_date)
|
|
t_end = _iso_date(term.end_date)
|
|
s.run("""
|
|
MERGE (t:AcademicTerm {uuid_string: $id})
|
|
SET t.term_name = $name, t.term_number = $num,
|
|
t.start_date = date($start), t.end_date = date($end),
|
|
t.node_storage_path = $path
|
|
WITH t
|
|
MATCH (ay:AcademicYear {uuid_string: $ay_id})
|
|
MERGE (ay)-[:ACADEMIC_YEAR_HAS_ACADEMIC_TERM]->(t)
|
|
""", id=t_id, name=term.name, num=str(term.term_number),
|
|
start=term.start_date, end=term.end_date,
|
|
path=f"timetable/{t_id}", ay_id=ay_id)
|
|
|
|
for wn, wstart in _academic_weeks(t_start, t_end):
|
|
w_id = f"week_{t_id}_{wn}"
|
|
s.run("""
|
|
MERGE (w:AcademicWeek {uuid_string: $id})
|
|
SET w.academic_week_number = $num, w.start_date = date($start),
|
|
w.week_type = 'academic', w.node_storage_path = $path
|
|
WITH w
|
|
MATCH (t:AcademicTerm {uuid_string: $t_id})
|
|
MERGE (t)-[:ACADEMIC_TERM_HAS_ACADEMIC_WEEK]->(w)
|
|
""", id=w_id, num=str(wn), start=wstart.isoformat(),
|
|
path=f"timetable/{w_id}", t_id=t_id)
|
|
|
|
# TeacherTimetable — only if user has a Teacher node
|
|
if teacher_uuid:
|
|
teacher_tt_id = f"teacher_tt_{teacher_uuid}_{year_start.year}"
|
|
s.run("""
|
|
MERGE (tt:TeacherTimetable {uuid_string: $id})
|
|
SET tt.teacher_timetable_id = $id,
|
|
tt.start_date = date($start), tt.end_date = date($end),
|
|
tt.node_storage_path = $path
|
|
WITH tt
|
|
MATCH (teacher:Teacher {uuid_string: $t_uuid})
|
|
MERGE (teacher)-[:HAS_TIMETABLE]->(tt)
|
|
WITH tt
|
|
MATCH (st:SchoolTimetable {uuid_string: $st_id})
|
|
MERGE (tt)-[:TEACHER_TIMETABLE_FOR]->(st)
|
|
""", id=teacher_tt_id, start=body.year_start, end=body.year_end,
|
|
path=f"timetable/{teacher_tt_id}",
|
|
t_uuid=teacher_uuid, st_id=school_tt_id)
|
|
else:
|
|
teacher_tt_id = None
|
|
|
|
return {
|
|
"status": "ok",
|
|
"timetable_id": teacher_tt_id,
|
|
"school_timetable_id": school_tt_id,
|
|
"institute_db": institute_db,
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Timetable setup failed: {e}")
|
|
return {"status": "error", "message": str(e)}
|
|
|
|
|
|
@router.get("/slots")
|
|
async def get_timetable_slots(credentials: dict = Depends(SupabaseBearer())) -> Dict[str, Any]:
|
|
user_email = credentials.get("email", "")
|
|
institute_db, teacher_uuid = _find_teacher_institute(user_email)
|
|
if not institute_db:
|
|
return {"status": "no_school", "slots": [], "periods": []}
|
|
try:
|
|
with driver_tools.get_session(database=institute_db) as s:
|
|
tt_rec = s.run(
|
|
"MATCH (t:Teacher {uuid_string: $u})-[:HAS_TIMETABLE]->(tt:TeacherTimetable) "
|
|
"RETURN tt.uuid_string AS id LIMIT 1", u=teacher_uuid
|
|
).single()
|
|
if not tt_rec:
|
|
return {"status": "empty", "slots": [], "periods": []}
|
|
tt_id = tt_rec["id"]
|
|
|
|
slots_result = s.run(
|
|
"MATCH (:TeacherTimetable {uuid_string: $id})-[:HAS_TIMETABLE_SLOT]->(sl:TimetableSlot) "
|
|
"RETURN sl", id=tt_id
|
|
)
|
|
slots = [
|
|
{"day_of_week": r["sl"]["day_of_week"],
|
|
"period_code": r["sl"]["period_code"],
|
|
"subject_class": r["sl"]["subject_class"],
|
|
"start_time": r["sl"]["start_time"],
|
|
"end_time": r["sl"]["end_time"]}
|
|
for r in slots_result
|
|
]
|
|
|
|
periods_rec = s.run(
|
|
"MATCH (st:SchoolTimetable) WHERE st.periods_template IS NOT NULL "
|
|
"RETURN st.periods_template AS p LIMIT 1"
|
|
).single()
|
|
periods = json.loads(periods_rec["p"]) if periods_rec else []
|
|
|
|
return {"status": "ok", "timetable_id": tt_id, "slots": slots,
|
|
"periods": periods, "institute_db": institute_db}
|
|
except Exception as e:
|
|
logger.error(f"Get timetable slots failed: {e}")
|
|
return {"status": "error", "slots": [], "periods": []}
|
|
|
|
|
|
@router.post("/slots")
|
|
async def save_timetable_slots(
|
|
body: SlotsRequest,
|
|
credentials: dict = Depends(SupabaseBearer())
|
|
) -> Dict[str, Any]:
|
|
user_email = credentials.get("email", "")
|
|
institute_db, _ = _find_teacher_institute(user_email)
|
|
if not institute_db:
|
|
return {"status": "error", "message": "Teacher not linked to school"}
|
|
try:
|
|
with driver_tools.get_session(database=institute_db) as s:
|
|
s.run(
|
|
"MATCH (:TeacherTimetable {uuid_string: $id})-[:HAS_TIMETABLE_SLOT]->(sl) "
|
|
"DETACH DELETE sl", id=body.timetable_id
|
|
)
|
|
created = 0
|
|
for slot in body.slots:
|
|
if not slot.subject_class.strip():
|
|
continue
|
|
slot_id = f"slot_{body.timetable_id}_{slot.day_of_week}_{slot.period_code}"
|
|
s.run("""
|
|
MERGE (sl:TimetableSlot {uuid_string: $id})
|
|
SET sl.day_of_week = $day, sl.period_code = $code,
|
|
sl.subject_class = $cls, sl.start_time = $start,
|
|
sl.end_time = $end, sl.node_storage_path = $path
|
|
WITH sl
|
|
MATCH (tt:TeacherTimetable {uuid_string: $tt_id})
|
|
MERGE (tt)-[:HAS_TIMETABLE_SLOT]->(sl)
|
|
""", id=slot_id, day=slot.day_of_week, code=slot.period_code,
|
|
cls=slot.subject_class, start=slot.start_time, end=slot.end_time,
|
|
path=f"timetable/slots/{slot_id}", tt_id=body.timetable_id)
|
|
created += 1
|
|
return {"status": "ok", "created": created}
|
|
except Exception as e:
|
|
logger.error(f"Save timetable slots failed: {e}")
|
|
return {"status": "error", "message": str(e)}
|
|
|
|
|
|
@router.post("/init")
|
|
async def init_teacher_timetable(credentials: dict = Depends(SupabaseBearer())) -> Dict[str, Any]:
|
|
"""Create a TeacherTimetable for the current teacher using the existing school calendar."""
|
|
user_email = credentials.get("email", "")
|
|
institute_db, teacher_uuid = _find_teacher_institute(user_email)
|
|
if not institute_db:
|
|
return {"status": "error", "message": "Teacher not linked to a school"}
|
|
|
|
try:
|
|
with driver_tools.get_session(database=institute_db) as s:
|
|
st_rec = s.run(
|
|
"MATCH (st:SchoolTimetable) "
|
|
"RETURN st.uuid_string AS id, "
|
|
" toString(st.start_date) AS start, "
|
|
" toString(st.end_date) AS end "
|
|
"LIMIT 1"
|
|
).single()
|
|
if not st_rec:
|
|
return {"status": "error", "message": "No school calendar set up yet — contact your school admin"}
|
|
|
|
school_tt_id = st_rec["id"]
|
|
start_str = str(st_rec["start"])
|
|
end_str = str(st_rec["end"])
|
|
year_str = start_str[:4]
|
|
teacher_tt_id = f"teacher_tt_{teacher_uuid}_{year_str}"
|
|
|
|
s.run("""
|
|
MERGE (tt:TeacherTimetable {uuid_string: $id})
|
|
SET tt.teacher_timetable_id = $id,
|
|
tt.start_date = date($start), tt.end_date = date($end),
|
|
tt.node_storage_path = $path
|
|
WITH tt
|
|
MATCH (teacher:Teacher {uuid_string: $t_uuid})
|
|
MERGE (teacher)-[:HAS_TIMETABLE]->(tt)
|
|
WITH tt
|
|
MATCH (st:SchoolTimetable {uuid_string: $st_id})
|
|
MERGE (tt)-[:TEACHER_TIMETABLE_FOR]->(st)
|
|
""", id=teacher_tt_id, start=start_str, end=end_str,
|
|
path=f"timetable/{teacher_tt_id}",
|
|
t_uuid=teacher_uuid, st_id=school_tt_id)
|
|
|
|
return {"status": "ok", "timetable_id": teacher_tt_id}
|
|
except Exception as e:
|
|
logger.error(f"Teacher timetable init failed: {e}")
|
|
return {"status": "error", "message": str(e)}
|