273 lines
14 KiB
Python
273 lines
14 KiB
Python
from dotenv import load_dotenv, find_dotenv
|
|
load_dotenv(find_dotenv())
|
|
import os
|
|
import modules.logger_tool as logger
|
|
log_name = 'api_modules_database_tools_filesystem_tools'
|
|
log_dir = os.getenv("LOG_PATH", "/logs") # Default path as fallback
|
|
logging = logger.get_logger(
|
|
name=log_name,
|
|
log_level=os.getenv("LOG_LEVEL", "DEBUG"),
|
|
log_path=log_dir,
|
|
log_file=log_name,
|
|
runtime=True,
|
|
log_format='default'
|
|
)
|
|
from datetime import timedelta
|
|
import json
|
|
import re
|
|
|
|
class ClassroomCopilotFilesystem:
|
|
def __init__(self, db_name: str, init_run_type: str = None):
|
|
|
|
self.db_name = db_name
|
|
|
|
# Get base path from environment
|
|
self.base_path = os.getenv("NODE_FILESYSTEM_PATH")
|
|
if not self.base_path:
|
|
raise ValueError("NODE_FILESYSTEM_PATH environment variable not set")
|
|
|
|
logging.info(f"Initializing ClassroomCopilotFilesystem with db_name: {db_name} and init_run_type: {init_run_type} with base path: {self.base_path}")
|
|
|
|
# Set root path based on init type
|
|
if init_run_type == "school":
|
|
self.root_path = os.path.join(self.base_path, "schools")
|
|
logging.debug(f"School root path: {self.root_path}")
|
|
elif init_run_type == "user":
|
|
self.root_path = os.path.join(self.base_path, "users")
|
|
logging.debug(f"User root path: {self.root_path}")
|
|
elif init_run_type == "multiplayer":
|
|
self.root_path = os.path.join(self.base_path, "multiplayer")
|
|
logging.debug(f"Multiplayer root path: {self.root_path}")
|
|
else:
|
|
self.root_path = self.base_path
|
|
logging.debug(f"Database root path: {self.root_path}")
|
|
|
|
# Ensure root directory exists
|
|
os.makedirs(self.root_path, exist_ok=True)
|
|
|
|
logging.debug(f"Filesystem initialized with run type: {init_run_type} and root path: {self.root_path}")
|
|
|
|
def log_directory_structure(self, start_path):
|
|
for root, dirs, files in os.walk(start_path):
|
|
level = root.replace(start_path, '').count(os.sep)
|
|
indent = ' ' * 4 * (level)
|
|
logging.info(f"{indent}{os.path.basename(root)}/")
|
|
subindent = ' ' * 4 * (level + 1)
|
|
for f in files:
|
|
logging.info(f"{subindent}{f}")
|
|
|
|
def create_directory(self, path):
|
|
"""Utility method to create a directory if it doesn't exist."""
|
|
if not os.path.exists(path):
|
|
os.makedirs(path)
|
|
logging.info(f"Directory {path} created.")
|
|
return True
|
|
return False
|
|
|
|
def create_private_user_directory(self, user_id):
|
|
"""Create a directory for a specific user."""
|
|
|
|
# For user database: /users/[user_db]/[username]
|
|
user_path = os.path.join(self.root_path, user_id)
|
|
|
|
logging.info(f"Creating user directory at {user_path}")
|
|
return self.create_directory(user_path), user_path
|
|
|
|
def create_user_worker_directory(self, user_path, worker_id, worker_type):
|
|
"""Create a worker directory under the user directory."""
|
|
# Create worker directory: [user_path]/[worker_code]
|
|
worker_path = os.path.join(user_path, worker_type, worker_id)
|
|
logging.info(f"Creating worker directory at {worker_path}")
|
|
return self.create_directory(worker_path), worker_path
|
|
|
|
def create_school_worker_directory(self, school_path, worker_type, worker_id):
|
|
"""Create a worker directory under the school directory."""
|
|
worker_path = os.path.join(school_path, "workers", worker_type, worker_id)
|
|
logging.info(f"Creating school {worker_type} worker directory at {worker_path}")
|
|
return self.create_directory(worker_path), worker_path
|
|
|
|
def create_school_directory(self, school_uuid_string):
|
|
"""Create a directory for a specific school."""
|
|
logging.info(f"Creating school directory with uuid_string: {school_uuid_string}")
|
|
logging.debug(f"School UUID is not None, creating school directory at {os.path.join(self.root_path, school_uuid_string)}")
|
|
school_path = os.path.join(self.root_path, school_uuid_string)
|
|
return self.create_directory(school_path), school_path
|
|
|
|
def create_year_directory(self, year, calendar_path=None):
|
|
"""Create a directory for a specific year."""
|
|
if calendar_path is None:
|
|
year_path = os.path.join(self.root_path, "calendar", str(year))
|
|
else:
|
|
year_path = os.path.join(calendar_path, "years", str(year))
|
|
|
|
return self.create_directory(year_path), year_path
|
|
|
|
def create_month_directory(self, year, month, calendar_path=None):
|
|
"""Create a directory for a specific month."""
|
|
if calendar_path is None:
|
|
month_path = os.path.join(self.root_path, "calendar", str(year), "months", f"{month:02}")
|
|
else:
|
|
month_path = os.path.join(calendar_path, "years", str(year), "months", f"{month:02}")
|
|
|
|
return self.create_directory(month_path), month_path
|
|
|
|
def create_week_directory(self, year, week, calendar_path=None):
|
|
"""Create a directory for a specific week."""
|
|
if calendar_path is None:
|
|
week_path = os.path.join(self.root_path, "calendar", str(year), "weeks", f"{week}")
|
|
else:
|
|
week_path = os.path.join(calendar_path, "years", str(year), "weeks", f"{week}")
|
|
|
|
return self.create_directory(week_path), week_path
|
|
|
|
def create_day_directory(self, year, month, day, calendar_path=None):
|
|
"""Create a directory for a specific day."""
|
|
if calendar_path is None:
|
|
day_path = os.path.join(self.root_path, "calendar", str(year), "months", f"{month:02}", f"{day:02}")
|
|
else:
|
|
day_path = os.path.join(calendar_path, "years", str(year), "months", f"{month:02}", f"{day:02}")
|
|
|
|
return self.create_directory(day_path), day_path
|
|
|
|
def setup_calendar_directories(self, start_date, end_date, calendar_path=None):
|
|
"""Setup directories for the range from start_date to end_date."""
|
|
current_date = start_date
|
|
while current_date <= end_date:
|
|
year, month, day = current_date.year, current_date.month, current_date.day
|
|
if calendar_path is None:
|
|
_, year_path = self.create_year_directory(year)
|
|
_, month_path = self.create_month_directory(year, month)
|
|
_, week_path = self.create_week_directory(year, current_date.isocalendar()[1])
|
|
_, day_path = self.create_day_directory(year, month, day)
|
|
else:
|
|
_, year_path = self.create_year_directory(year, calendar_path)
|
|
_, month_path = self.create_month_directory(year, month, calendar_path)
|
|
_, week_path = self.create_week_directory(year, current_date.isocalendar()[1], calendar_path)
|
|
_, day_path = self.create_day_directory(year, month, day, calendar_path)
|
|
current_date += timedelta(days=1)
|
|
return year_path, month_path, week_path, day_path
|
|
|
|
def create_school_timetable_directory(self, school_path=None):
|
|
"""Create a directory for the timetable."""
|
|
if school_path is None:
|
|
timetable_path = os.path.join(self.root_path, "timetable")
|
|
else:
|
|
timetable_path = os.path.join(school_path, "timetable")
|
|
|
|
return self.create_directory(timetable_path), timetable_path
|
|
|
|
def create_school_timetable_year_directory(self, timetable_path, year):
|
|
"""Create a directory for a specific academic year within the timetable."""
|
|
year_path = os.path.join(timetable_path, "years", str(year))
|
|
return self.create_directory(year_path), year_path
|
|
|
|
def create_school_timetable_academic_term_directory(self, timetable_path, term_name, term_number):
|
|
"""Create a directory for a specific term within an academic year."""
|
|
term_path = os.path.join(timetable_path, "terms", f"{term_number}_{term_name.replace(' ', '_')}")
|
|
return self.create_directory(term_path), term_path
|
|
|
|
def create_school_timetable_academic_term_break_directory(self, timetable_path, term_name):
|
|
"""Create a directory for a specific term within an academic year."""
|
|
term_path = os.path.join(timetable_path, "terms", "term_breaks", f"{term_name.replace(' ', '_')}")
|
|
return self.create_directory(term_path), term_path
|
|
|
|
def create_school_timetable_academic_week_directory(self, timetable_path, week_number):
|
|
"""Create a directory for a specific week within a term of a specific year."""
|
|
week_path = os.path.join(timetable_path, "weeks", f"{week_number}")
|
|
return self.create_directory(week_path), week_path
|
|
|
|
def create_school_timetable_academic_day_directory(self, timetable_path, academic_day):
|
|
"""Create a directory for a specific day within a week of a term."""
|
|
day_path = os.path.join(timetable_path, "days",f"{academic_day:02}")
|
|
return self.create_directory(day_path), day_path
|
|
|
|
def create_school_timetable_period_directory(self, timetable_path, academic_day, period_dir):
|
|
"""Create a directory for a specific period within a day."""
|
|
period_path = os.path.join(timetable_path, "days",f"{academic_day:02}", f"{period_dir}")
|
|
return self.create_directory(period_path), period_path
|
|
|
|
def create_school_curriculum_directory(self, school_path=None):
|
|
"""Create a directory for the curriculum."""
|
|
if school_path is None:
|
|
curriculum_path = os.path.join(self.root_path, "curriculum")
|
|
else:
|
|
curriculum_path = os.path.join(school_path, "curriculum")
|
|
|
|
return self.create_directory(curriculum_path), curriculum_path
|
|
|
|
def create_school_pastoral_directory(self, school_path=None):
|
|
"""Create a directory for the pastoral."""
|
|
if school_path is None:
|
|
pastoral_path = os.path.join(self.root_path, "pastoral")
|
|
else:
|
|
pastoral_path = os.path.join(school_path, "pastoral")
|
|
|
|
return self.create_directory(pastoral_path), pastoral_path
|
|
|
|
def create_school_department_directory(self, school_path, department):
|
|
"""Create a directory for a specific department within the school."""
|
|
department_path = os.path.join(school_path, "departments", f"{department}")
|
|
return self.create_directory(department_path), department_path
|
|
|
|
def create_department_subject_directory(self, department_path, subject_name):
|
|
"""Create a directory for a specific subject within a department."""
|
|
subject_path = os.path.join(department_path, "subjects", f"{subject_name}")
|
|
return self.create_directory(subject_path), subject_path
|
|
|
|
def create_curriculum_key_stage_syllabus_directory(self, curriculum_path, key_stage, subject_name, syllabus_id):
|
|
"""Create a directory for a specific key stage syllabus under the curriculum structure."""
|
|
# Replace spaces with underscores and remove any special characters from subject name
|
|
safe_subject_name = re.sub(r'[^\w\-_\.]', '_', subject_name)
|
|
syllabus_path = os.path.join(curriculum_path, "subjects", safe_subject_name, "key_stage_syllabuses", f"KS{key_stage}", f"KS{key_stage}.{safe_subject_name}")
|
|
return self.create_directory(syllabus_path), syllabus_path
|
|
|
|
def create_pastoral_year_group_directory(self, pastoral_path, year_group):
|
|
"""Create a directory for a specific year group under the pastoral structure."""
|
|
year_group_path = os.path.join(pastoral_path, "year_groups", f"Y{year_group}")
|
|
return self.create_directory(year_group_path), year_group_path
|
|
|
|
def create_curriculum_year_group_syllabus_directory(self, curriculum_path, subject_name, year_group, syllabus_id):
|
|
"""Create a directory for a specific year group syllabus under the curriculum structure."""
|
|
# Replace spaces with underscores and remove any special characters from subject name
|
|
safe_subject_name = re.sub(r'[^\w\-_\.]', '_', subject_name)
|
|
syllabus_path = os.path.join(curriculum_path, "subjects", safe_subject_name, "year_group_syllabuses", f"Y{year_group}", f"Y{year_group}.{safe_subject_name}")
|
|
return self.create_directory(syllabus_path), syllabus_path
|
|
|
|
def create_curriculum_topic_directory(self, year_group_syllabus_path, topic_id):
|
|
"""Create a directory for a specific topic under a year group syllabus."""
|
|
topic_path = os.path.join(year_group_syllabus_path, "topics", f"{topic_id}")
|
|
return self.create_directory(topic_path), topic_path
|
|
|
|
def create_curriculum_keystage_topic_directory(self, keystage_syllabus_path, topic_id):
|
|
"""Create a directory for a specific key stage topic under a key stage group syllabus."""
|
|
topic_path = os.path.join(keystage_syllabus_path, "core_topics", f"{topic_id}")
|
|
return self.create_directory(topic_path), topic_path
|
|
|
|
def create_curriculum_lesson_directory(self, topic_path, lesson_id):
|
|
"""Create a directory for a specific lesson under a topic."""
|
|
lesson_path = os.path.join(topic_path, "lessons", f"{lesson_id}")
|
|
return self.create_directory(lesson_path), lesson_path
|
|
|
|
def create_curriculum_learning_statement_directory(self, lesson_path, statement_id):
|
|
"""Create a directory for a specific learning statement under a lesson."""
|
|
statement_path = os.path.join(lesson_path, "learning_statements", f"{statement_id}")
|
|
return self.create_directory(statement_path), statement_path
|
|
|
|
# Remove or mark as deprecated the old methods
|
|
|
|
|
|
def create_teacher_timetable_directory(self, teacher_path):
|
|
teacher_timetable_path = os.path.join(teacher_path, "timetable")
|
|
return self.create_directory(teacher_timetable_path), teacher_timetable_path
|
|
|
|
def create_teacher_class_directory(self, teacher_timetable_path, class_name):
|
|
class_path = os.path.join(teacher_timetable_path, "classes", class_name)
|
|
return self.create_directory(class_path), class_path
|
|
|
|
def create_teacher_timetable_lesson_directory(self, class_path, lesson_id):
|
|
lesson_path = os.path.join(class_path, "timetabled_lessons", lesson_id)
|
|
return self.create_directory(lesson_path), lesson_path
|
|
|
|
def create_teacher_planned_lesson_directory(self, class_path, lesson_id):
|
|
planned_lesson_path = os.path.join(class_path, "planned_lessons", lesson_id)
|
|
return self.create_directory(planned_lesson_path), planned_lesson_path |