""" Admin Service Layer for ThutoAI Handles admin authentication and school management operations """ from models import DatabaseManager import hashlib import json from datetime import datetime class AdminService: def __init__(self): self.db = DatabaseManager() def authenticate_admin(self, username, password): """Authenticate admin user""" conn = self.db.get_connection() cursor = conn.cursor() password_hash = hashlib.sha256(password.encode()).hexdigest() cursor.execute(''' SELECT id, username, full_name, email, role, is_active FROM admin_users WHERE username = ? AND password_hash = ? AND is_active = 1 ''', (username, password_hash)) admin = cursor.fetchone() if admin: # Update last login cursor.execute(''' UPDATE admin_users SET last_login = CURRENT_TIMESTAMP WHERE id = ? ''', (admin[0],)) conn.commit() conn.close() return { 'id': admin[0], 'username': admin[1], 'full_name': admin[2], 'email': admin[3], 'role': admin[4], 'is_active': bool(admin[5]) } conn.close() return None # Announcement Management def create_announcement(self, title, content, priority='normal', target_audience='all', expires_at=None, created_by='admin'): """Create a new announcement""" conn = self.db.get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO announcements (title, content, priority, target_audience, expires_at) VALUES (?, ?, ?, ?, ?) ''', (title, content, priority, target_audience, expires_at)) announcement_id = cursor.lastrowid conn.commit() conn.close() return announcement_id def update_announcement(self, announcement_id, title=None, content=None, priority=None, target_audience=None, expires_at=None): """Update an existing announcement""" conn = self.db.get_connection() cursor = conn.cursor() updates = [] params = [] if title: updates.append("title = ?") params.append(title) if content: updates.append("content = ?") params.append(content) if priority: updates.append("priority = ?") params.append(priority) if target_audience: updates.append("target_audience = ?") params.append(target_audience) if expires_at: updates.append("expires_at = ?") params.append(expires_at) if updates: params.append(announcement_id) cursor.execute(f''' UPDATE announcements SET {", ".join(updates)} WHERE id = ? ''', params) conn.commit() conn.close() return True def delete_announcement(self, announcement_id): """Delete an announcement""" conn = self.db.get_connection() cursor = conn.cursor() cursor.execute('UPDATE announcements SET is_active = 0 WHERE id = ?', (announcement_id,)) conn.commit() conn.close() return True def get_all_announcements(self): """Get all announcements for admin management""" conn = self.db.get_connection() cursor = conn.cursor() cursor.execute(''' SELECT id, title, content, priority, target_audience, created_at, expires_at, is_active FROM announcements ORDER BY created_at DESC ''') announcements = cursor.fetchall() conn.close() return [ { 'id': ann[0], 'title': ann[1], 'content': ann[2], 'priority': ann[3], 'target_audience': ann[4], 'created_at': ann[5], 'expires_at': ann[6], 'is_active': bool(ann[7]) } for ann in announcements ] # Syllabus Management def create_syllabus(self, subject, grade_level, chapter_number, chapter_title, topics, learning_objectives=None, duration_weeks=None, resources=None, assessment_methods=None, created_by='admin'): """Create a new syllabus entry""" conn = self.db.get_connection() cursor = conn.cursor() # Convert lists to JSON strings topics_json = json.dumps(topics) if isinstance(topics, list) else topics resources_json = json.dumps(resources) if isinstance(resources, list) else resources cursor.execute(''' INSERT INTO syllabus (subject, grade_level, chapter_number, chapter_title, topics, learning_objectives, duration_weeks, resources, assessment_methods, created_by) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', (subject, grade_level, chapter_number, chapter_title, topics_json, learning_objectives, duration_weeks, resources_json, assessment_methods, created_by)) syllabus_id = cursor.lastrowid conn.commit() conn.close() return syllabus_id def get_syllabus(self, subject=None, grade_level=None): """Get syllabus data""" conn = self.db.get_connection() cursor = conn.cursor() query = ''' SELECT id, subject, grade_level, chapter_number, chapter_title, topics, learning_objectives, duration_weeks, resources, assessment_methods, created_by, created_at, updated_at FROM syllabus WHERE is_active = 1 ''' params = [] if subject: query += ' AND subject = ?' params.append(subject) if grade_level: query += ' AND grade_level = ?' params.append(grade_level) query += ' ORDER BY subject, grade_level, chapter_number' cursor.execute(query, params) syllabus_data = cursor.fetchall() conn.close() return [ { 'id': syl[0], 'subject': syl[1], 'grade_level': syl[2], 'chapter_number': syl[3], 'chapter_title': syl[4], 'topics': json.loads(syl[5]) if syl[5] else [], 'learning_objectives': syl[6], 'duration_weeks': syl[7], 'resources': json.loads(syl[8]) if syl[8] else [], 'assessment_methods': syl[9], 'created_by': syl[10], 'created_at': syl[11], 'updated_at': syl[12] } for syl in syllabus_data ] def update_syllabus(self, syllabus_id, **kwargs): """Update syllabus entry""" conn = self.db.get_connection() cursor = conn.cursor() # Convert lists to JSON if needed if 'topics' in kwargs and isinstance(kwargs['topics'], list): kwargs['topics'] = json.dumps(kwargs['topics']) if 'resources' in kwargs and isinstance(kwargs['resources'], list): kwargs['resources'] = json.dumps(kwargs['resources']) updates = [] params = [] for key, value in kwargs.items(): if key in ['subject', 'grade_level', 'chapter_number', 'chapter_title', 'topics', 'learning_objectives', 'duration_weeks', 'resources', 'assessment_methods', 'is_active']: updates.append(f"{key} = ?") params.append(value) if updates: updates.append("updated_at = CURRENT_TIMESTAMP") params.append(syllabus_id) cursor.execute(f''' UPDATE syllabus SET {", ".join(updates)} WHERE id = ? ''', params) conn.commit() conn.close() return True # Timetable Management def create_timetable_entry(self, class_section, day_of_week, period_number, start_time, end_time, subject, teacher_name=None, room_number=None, created_by='admin'): """Create a new timetable entry""" conn = self.db.get_connection() cursor = conn.cursor() cursor.execute(''' INSERT INTO timetable (class_section, day_of_week, period_number, start_time, end_time, subject, teacher_name, room_number, created_by) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ''', (class_section, day_of_week, period_number, start_time, end_time, subject, teacher_name, room_number, created_by)) timetable_id = cursor.lastrowid conn.commit() conn.close() return timetable_id def get_timetable(self, class_section=None, day_of_week=None): """Get timetable data""" conn = self.db.get_connection() cursor = conn.cursor() query = ''' SELECT id, class_section, day_of_week, period_number, start_time, end_time, subject, teacher_name, room_number, created_by, created_at FROM timetable WHERE is_active = 1 ''' params = [] if class_section: query += ' AND class_section = ?' params.append(class_section) if day_of_week: query += ' AND day_of_week = ?' params.append(day_of_week) query += ' ORDER BY class_section, day_of_week, period_number' cursor.execute(query, params) timetable_data = cursor.fetchall() conn.close() return [ { 'id': tt[0], 'class_section': tt[1], 'day_of_week': tt[2], 'period_number': tt[3], 'start_time': tt[4], 'end_time': tt[5], 'subject': tt[6], 'teacher_name': tt[7], 'room_number': tt[8], 'created_by': tt[9], 'created_at': tt[10] } for tt in timetable_data ] def update_timetable_entry(self, timetable_id, **kwargs): """Update timetable entry""" conn = self.db.get_connection() cursor = conn.cursor() updates = [] params = [] for key, value in kwargs.items(): if key in ['class_section', 'day_of_week', 'period_number', 'start_time', 'end_time', 'subject', 'teacher_name', 'room_number']: updates.append(f"{key} = ?") params.append(value) if updates: updates.append("updated_at = CURRENT_TIMESTAMP") params.append(timetable_id) cursor.execute(f''' UPDATE timetable SET {", ".join(updates)} WHERE id = ? ''', params) conn.commit() conn.close() return True def delete_timetable_entry(self, timetable_id): """Delete timetable entry""" conn = self.db.get_connection() cursor = conn.cursor() cursor.execute('UPDATE timetable SET is_active = 0 WHERE id = ?', (timetable_id,)) conn.commit() conn.close() return True # Dashboard Statistics def get_admin_dashboard_stats(self): """Get statistics for admin dashboard""" conn = self.db.get_connection() cursor = conn.cursor() stats = {} # Count active announcements cursor.execute('SELECT COUNT(*) FROM announcements WHERE is_active = 1') stats['active_announcements'] = cursor.fetchone()[0] # Count syllabus entries cursor.execute('SELECT COUNT(*) FROM syllabus WHERE is_active = 1') stats['syllabus_entries'] = cursor.fetchone()[0] # Count timetable entries cursor.execute('SELECT COUNT(*) FROM timetable WHERE is_active = 1') stats['timetable_entries'] = cursor.fetchone()[0] # Count students cursor.execute('SELECT COUNT(*) FROM students') stats['total_students'] = cursor.fetchone()[0] # Count upcoming exams cursor.execute('SELECT COUNT(*) FROM examinations WHERE exam_date >= date("now")') stats['upcoming_exams'] = cursor.fetchone()[0] conn.close() return stats # Initialize admin service admin_service = AdminService()