""" School Service Layer for ThutoAI Handles all school-related data operations """ from models import DatabaseManager from datetime import datetime, date import json class SchoolService: def __init__(self): self.db = DatabaseManager() def get_recent_announcements(self, limit=5): """Get recent active announcements""" conn = self.db.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT title, content, priority, created_at FROM announcements WHERE is_active = 1 AND (expires_at IS NULL OR expires_at > datetime('now')) ORDER BY priority DESC, created_at DESC LIMIT ? ''', (limit,)) announcements = cursor.fetchall() conn.close() return [ { 'title': ann[0], 'content': ann[1], 'priority': ann[2], 'date': ann[3] } for ann in announcements ] def get_upcoming_exams(self, days_ahead=30): """Get upcoming examinations""" conn = self.db.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT exam_name, subject, exam_date, exam_time, location, instructions FROM examinations WHERE exam_date >= date('now') AND exam_date <= date('now', '+{} days') ORDER BY exam_date ASC, exam_time ASC '''.format(days_ahead)) exams = cursor.fetchall() conn.close() return [ { 'name': exam[0], 'subject': exam[1], 'date': exam[2], 'time': exam[3], 'location': exam[4], 'instructions': exam[5] } for exam in exams ] def get_upcoming_tests_assignments(self, days_ahead=14): """Get upcoming tests and assignments""" conn = self.db.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT title, subject, type, due_date, due_time, description, total_marks FROM tests_assignments WHERE due_date >= date('now') AND due_date <= date('now', '+{} days') ORDER BY due_date ASC, due_time ASC '''.format(days_ahead)) tests = cursor.fetchall() conn.close() return [ { 'title': test[0], 'subject': test[1], 'type': test[2], 'due_date': test[3], 'due_time': test[4], 'description': test[5], 'total_marks': test[6] } for test in tests ] def get_student_grades(self, student_id, subject=None, limit=10): """Get student grades, optionally filtered by subject""" conn = self.db.get_connection() cursor = conn.cursor() if subject: cursor.execute(''' SELECT subject, assessment_type, assessment_name, marks_obtained, total_marks, percentage, grade_letter, date_recorded, teacher_comments FROM grades WHERE student_id = ? AND subject = ? ORDER BY date_recorded DESC LIMIT ? ''', (student_id, subject, limit)) else: cursor.execute(''' SELECT subject, assessment_type, assessment_name, marks_obtained, total_marks, percentage, grade_letter, date_recorded, teacher_comments FROM grades WHERE student_id = ? ORDER BY date_recorded DESC LIMIT ? ''', (student_id, limit)) grades = cursor.fetchall() conn.close() return [ { 'subject': grade[0], 'type': grade[1], 'name': grade[2], 'marks_obtained': grade[3], 'total_marks': grade[4], 'percentage': grade[5], 'grade': grade[6], 'date': grade[7], 'comments': grade[8] } for grade in grades ] def get_upcoming_events(self, days_ahead=30): """Get upcoming school events""" conn = self.db.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT event_name, event_date, event_time, location, description, event_type, is_holiday FROM school_events WHERE event_date >= date('now') AND event_date <= date('now', '+{} days') ORDER BY event_date ASC, event_time ASC '''.format(days_ahead)) events = cursor.fetchall() conn.close() return [ { 'name': event[0], 'date': event[1], 'time': event[2], 'location': event[3], 'description': event[4], 'type': event[5], 'is_holiday': bool(event[6]) } for event in events ] def get_student_summary(self, student_id="STU001"): """Get comprehensive student summary""" return { 'announcements': self.get_recent_announcements(3), 'upcoming_exams': self.get_upcoming_exams(30), 'upcoming_tests': self.get_upcoming_tests_assignments(14), 'recent_grades': self.get_student_grades(student_id, limit=5), 'upcoming_events': self.get_upcoming_events(30), 'syllabus': self.get_student_syllabus("Grade 10"), 'timetable': self.get_student_timetable("10-A") } def get_student_syllabus(self, grade_level): """Get syllabus for student's grade level""" conn = self.db.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT subject, chapter_title, topics, learning_objectives, duration_weeks, resources FROM syllabus WHERE grade_level = ? AND is_active = 1 ORDER BY subject, chapter_number ''', (grade_level,)) syllabus_data = cursor.fetchall() conn.close() return [ { 'subject': syl[0], 'chapter_title': syl[1], 'topics': json.loads(syl[2]) if syl[2] else [], 'learning_objectives': syl[3], 'duration_weeks': syl[4], 'resources': json.loads(syl[5]) if syl[5] else [] } for syl in syllabus_data ] def get_student_timetable(self, class_section): """Get timetable for student's class""" conn = self.db.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT day_of_week, period_number, start_time, end_time, subject, teacher_name, room_number FROM timetable WHERE class_section = ? AND is_active = 1 ORDER BY day_of_week, period_number ''', (class_section,)) timetable_data = cursor.fetchall() conn.close() # Group by day of week timetable_by_day = {} for tt in timetable_data: day = tt[0] if day not in timetable_by_day: timetable_by_day[day] = [] timetable_by_day[day].append({ 'period': tt[1], 'start_time': tt[2], 'end_time': tt[3], 'subject': tt[4], 'teacher': tt[5], 'room': tt[6] }) return timetable_by_day def search_school_info(self, query): """Search across all school information""" query = query.lower() results = [] # Search announcements announcements = self.get_recent_announcements(10) for ann in announcements: if query in ann['title'].lower() or query in ann['content'].lower(): results.append({ 'type': 'announcement', 'title': ann['title'], 'content': ann['content'], 'priority': ann['priority'] }) # Search exams exams = self.get_upcoming_exams(60) for exam in exams: if query in exam['name'].lower() or query in exam['subject'].lower(): results.append({ 'type': 'exam', 'name': exam['name'], 'subject': exam['subject'], 'date': exam['date'], 'time': exam['time'] }) # Search tests/assignments tests = self.get_upcoming_tests_assignments(30) for test in tests: if query in test['title'].lower() or query in test['subject'].lower(): results.append({ 'type': 'test_assignment', 'title': test['title'], 'subject': test['subject'], 'type_detail': test['type'], 'due_date': test['due_date'] }) return results def format_school_context_for_ai(self, student_id="STU001"): """Format school information as context for AI""" summary = self.get_student_summary(student_id) context = "SCHOOL INFORMATION CONTEXT:\n\n" # Recent announcements if summary['announcements']: context += "📢 RECENT ANNOUNCEMENTS:\n" for ann in summary['announcements']: context += f"- {ann['title']}: {ann['content']}\n" context += "\n" # Upcoming exams if summary['upcoming_exams']: context += "📝 UPCOMING EXAMS:\n" for exam in summary['upcoming_exams']: context += f"- {exam['name']} ({exam['subject']}) on {exam['date']} at {exam['time']}\n" context += "\n" # Upcoming tests/assignments if summary['upcoming_tests']: context += "📚 UPCOMING TESTS & ASSIGNMENTS:\n" for test in summary['upcoming_tests']: context += f"- {test['title']} ({test['subject']}) due {test['due_date']}\n" context += "\n" # Recent grades if summary['recent_grades']: context += "📊 RECENT GRADES:\n" for grade in summary['recent_grades']: context += f"- {grade['subject']}: {grade['percentage']}% ({grade['grade']}) in {grade['name']}\n" context += "\n" # Upcoming events if summary['upcoming_events']: context += "🎉 UPCOMING EVENTS:\n" for event in summary['upcoming_events']: context += f"- {event['name']} on {event['date']}\n" context += "\n" context += "Use this information to provide contextual and helpful responses to the student." return context # Initialize service school_service = SchoolService()