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