Spaces:
Runtime error
Runtime error
| """ | |
| 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() |