import json import logging import re from calendar import monthrange from datetime import date, datetime, timedelta from urllib.parse import urlencode from django.contrib import admin, messages from django.contrib.admin import SimpleListFilter from django.core.exceptions import PermissionDenied from django.core.paginator import Paginator from django.db.models import Max, Min, Q from django.db.models.functions import TruncDate from django.http import Http404, HttpResponse, JsonResponse from django.shortcuts import redirect from django.template.response import TemplateResponse from django.urls import path, reverse from django.utils import timezone from django.utils.dateparse import parse_date from django.utils.html import format_html from openpyxl import Workbook from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter from unfold.admin import ModelAdmin from unfold.contrib.filters.admin import DropdownFilter from unfold.decorators import action from .base_admin import DepartmentFilter from ..models import Department, Event, Person, Class, Record, Area, Semester, Weekday from ..utils import build_class_calendar_context logger = logging.getLogger(__name__) class AreaFilter(DropdownFilter, SimpleListFilter): title = "area" parameter_name = "area" def lookups(self, request, model_admin): try: areas = Area.objects.all().order_by("name") return [(area.id, area.name) for area in areas] except Exception: return [] def queryset(self, request, queryset): if self.value(): try: return queryset.filter(area_id=self.value()) except Exception: pass return queryset class SemesterFilter(DropdownFilter, SimpleListFilter): title = "semester" parameter_name = "semester" def lookups(self, request, model_admin): try: semesters = Semester.objects.all().order_by("-start_date") return [(sem.id, sem.title) for sem in semesters] except Exception: return [] def queryset(self, request, queryset): if self.value(): try: return queryset.filter(semester_id=self.value()) except Exception: pass return queryset class CoInstructorFilter(DropdownFilter, SimpleListFilter): title = "co-instructor" parameter_name = "co_instructor" def lookups(self, request, model_admin): try: from django.contrib.auth import get_user_model User = get_user_model() co_instructors = User.objects.filter(is_staff=True).order_by("username") return [(user.id, user.username) for user in co_instructors] except Exception: return [] def queryset(self, request, queryset): if self.value(): try: return queryset.filter(co_instructors=self.value()).distinct() except Exception: pass return queryset class InstructorFilter(DropdownFilter, SimpleListFilter): title = "instructor" parameter_name = "instructor" def lookups(self, request, model_admin): try: from django.contrib.auth import get_user_model User = get_user_model() instructors = User.objects.filter(is_staff=True).order_by("username") return [(user.id, user.username) for user in instructors] except Exception: return [] def queryset(self, request, queryset): if self.value(): try: return queryset.filter(instructor_id=self.value()) except Exception: pass return queryset @admin.register(Class) class ClassAdmin(ModelAdmin): list_display = [ "title", "get_days_display", "get_class_time_display", "get_co_instructors", "instructor", "get_student_count", "area", "semester", "department" ] list_filter = [ AreaFilter, SemesterFilter, DepartmentFilter, CoInstructorFilter, InstructorFilter, ] search_fields = ["co_instructors__username", "co_instructors__first_name", "co_instructors__last_name", "instructor__username", "instructor__first_name", "instructor__last_name"] autocomplete_fields = ["students", "co_instructors", "instructor"] change_list_template = "admin/class_changelist.html" change_form_template = "admin/core/class/change_form.html" change_form_show_cancel_button = True list_filter_submit = True list_fullwidth = False list_filter_sheet = False list_horizontal_scrollbar_top = False list_disable_select_all = False superadmin_only_edit_fields = [] fieldsets = [ ("Details", {"fields": ["title", "semester", "department", "area"]}), ("Schedule", {"fields": ["days", "start_time", "end_time"]}), ("Instructors", {"fields": ["instructor", "co_instructors"]}), ("Students", {"fields": ["students"]}), ] def get_queryset(self, request): qs = super().get_queryset(request).prefetch_related("days", "co_instructors") can_view_all_classes = request.user.has_perm("core.view_all_classes") if request.user.is_superuser or can_view_all_classes: return qs # Non-superadmins only see classes they instruct or co-instruct. return qs.filter(Q(instructor=request.user) | Q(co_instructors=request.user)).distinct() def has_change_permission(self, request, obj=None): has_perm = super().has_change_permission(request, obj) if not has_perm: return False return True def has_view_permission(self, request, obj=None): has_perm = super().has_view_permission(request, obj) if not has_perm: return False can_view_all_classes = request.user.has_perm("core.view_all_classes") if request.user.is_superuser or can_view_all_classes or obj is None: return True return obj.instructor_id == request.user.id or obj.co_instructors.filter(pk=request.user.pk).exists() def get_changeform_initial_data(self, request): initial = super().get_changeform_initial_data(request) if request.user.is_authenticated: initial.setdefault("instructor", request.user.pk) return initial def change_view(self, request, object_id, form_url="", extra_context=None): return super().change_view(request, object_id, form_url, extra_context) def save_model(self, request, obj, form, change): if not change and not obj.instructor_id and request.user.is_authenticated: obj.instructor = request.user return super().save_model(request, obj, form, change) def save_related(self, request, form, formsets, change): super().save_related(request, form, formsets, change) if not request.user.is_superuser and not form.instance.co_instructors.filter(pk=request.user.pk).exists(): form.instance.co_instructors.add(request.user) def add_view(self, request, form_url="", extra_context=None): return super().add_view(request, form_url, extra_context) def get_readonly_fields(self, request, obj=None): readonly_fields = list(super().get_readonly_fields(request, obj)) if request.user.is_superuser: return readonly_fields return list(dict.fromkeys(readonly_fields + self.superadmin_only_edit_fields)) def has_delete_permission(self, request, obj=None): has_perm = super().has_delete_permission(request, obj) if not has_perm: return False return True def changelist_view(self, request, extra_context=None): extra_context = extra_context or {} active_filters = [] # Check filter parameters area_id = request.GET.get("area") if area_id: from ..models import Area area = Area.objects.filter(id=area_id).first() active_filters.append({"label": "Area", "value": area.name if area else area_id}) semester_id = request.GET.get("semester") if semester_id: from ..models import Semester semester = Semester.objects.filter(id=semester_id).first() active_filters.append({"label": "Semester", "value": semester.title if semester else semester_id}) department_id = request.GET.get("department") if department_id: from ..models import Department department = Department.objects.filter(id=department_id).first() active_filters.append({"label": "Department", "value": department.name if department else department_id}) co_instructor_id = request.GET.get("co_instructor") if co_instructor_id: from django.contrib.auth import get_user_model User = get_user_model() co_instructor = User.objects.filter(id=co_instructor_id).first() active_filters.append({"label": "Co-Instructor", "value": co_instructor.username if co_instructor else co_instructor_id}) instructor_id = request.GET.get("instructor") if instructor_id: from django.contrib.auth import get_user_model User = get_user_model() instructor = User.objects.filter(id=instructor_id).first() active_filters.append({"label": "Instructor", "value": instructor.username if instructor else instructor_id}) search_query = request.GET.get("q") if search_query: active_filters.append({"label": "Search", "value": search_query}) extra_context["active_filter_summary"] = active_filters return super().changelist_view(request, extra_context=extra_context) def get_urls(self): urls = super().get_urls() custom_urls = [ path( "parse-class-list/", self.admin_site.admin_view(self.parse_class_list_view), name="core_class_parse_class_list", ), path( "/start-scanning/", self.admin_site.admin_view(self.start_scanning_view), name="core_class_start_scanning", ), path( "/start-scanning/scan/", self.admin_site.admin_view(self.start_scanning_scan_view), name="core_class_start_scanning_scan", ), path( "/manage-students/", self.admin_site.admin_view(self.manage_students_view), name="core_class_manage_students", ), path( "/manage-students/filter-persons/", self.admin_site.admin_view(self.filter_persons_view), name="core_class_filter_persons", ), path( "/map-students/", self.admin_site.admin_view(self.map_students_view), name="core_class_map_students", ), path( "/view-attendance/", self.admin_site.admin_view(self.view_attendance_view), name="core_class_view_attendance", ), path( "/view-attendance/export-excel/", self.admin_site.admin_view(self.view_attendance_export_excel_view), name="core_class_view_attendance_export_excel", ), path( "/view-attendance/day-detail/", self.admin_site.admin_view(self.attendance_day_detail_view), name="core_class_attendance_day_detail", ), path( "/view-attendance/export-roster/", self.admin_site.admin_view(self.view_attendance_export_roster_view), name="core_class_view_attendance_export_roster", ), ] return custom_urls + urls def get_student_count(self, obj): return obj.students.count() get_student_count.short_description = "Students" def get_co_instructors(self, obj): return ", ".join(u.username for u in obj.co_instructors.all()) or "-" get_co_instructors.short_description = "Co-Instructors" def get_days_display(self, obj): day_labels = [weekday.get_code_display() for weekday in obj.days.all()] return ", ".join(day_labels) if day_labels else "-" get_days_display.short_description = "Days" def get_class_time_display(self, obj): if obj.start_time and obj.end_time: return f"{obj.start_time.strftime('%H:%M')} - {obj.end_time.strftime('%H:%M')}" return "-" get_class_time_display.short_description = "Time" def manage_students_action(self, obj): url = reverse("admin:core_class_manage_students", args=[obj.pk]) return format_html( '' '' "Manage Students" "", url, ) manage_students_action.short_description = "Actions" # Custom actions actions_list = [] actions_row = ["start_scanning_custom_action", "manage_students_custom_action", "view_attendances_custom_action"] actions_detail = ["start_scanning_custom_action", "manage_students_custom_action", "view_attendances_custom_action"] actions_submit_line = [] # Displayed near save in object detail @action(description="Start Scanning", url_path="actions-start-scanning") def start_scanning_custom_action(self, request, object_id): return redirect(reverse("admin:core_class_start_scanning", args=[object_id])) @action(description="Manage Students", url_path="actions-manage-students") def manage_students_custom_action(self, request, object_id): section = self.get_object(request, object_id) if section is None: raise Http404("Class not found.") if not self.has_change_permission(request, section): raise PermissionDenied("You do not have permission to manage students for this class.") return redirect(reverse("admin:core_class_manage_students", args=[object_id])) @action(description="View Attendances", url_path="actions-view-attendances") def view_attendances_custom_action(self, request, object_id): section = self.get_object(request, object_id) if section is None: raise Http404("Class not found.") if not self.has_view_permission(request, section): raise PermissionDenied return redirect(reverse("admin:core_class_view_attendance", args=[object_id])) def _method_name_from_unfold_action(self, unfold_action): prefix = f"{self.model._meta.app_label}_{self.model._meta.model_name}_" action_name = unfold_action.action_name return action_name[len(prefix):] if action_name.startswith(prefix) else action_name def _can_see_action(self, request, action_name, obj=None): if action_name == "view_attendances_custom_action": return self.has_view_permission(request, obj) return self.has_change_permission(request, obj) def _filter_actions_for_request(self, request, actions, obj=None): filtered = [] for action in actions: method_name = self._method_name_from_unfold_action(action) if self._can_see_action(request, method_name, obj): filtered.append(action) return filtered def get_actions_row(self, request): actions = super().get_actions_row(request) return self._filter_actions_for_request(request, actions) def get_actions_detail(self, request, object_id): actions = super().get_actions_detail(request, object_id) section = self.get_object(request, object_id) return self._filter_actions_for_request(request, actions, section) def start_scanning_view(self, request, object_id): section = self.get_object(request, object_id) if section is None: raise Http404("Class not found.") if not self.has_change_permission(request, section): raise PermissionDenied now = timezone.localtime(timezone.now()) today = now.date() current_time = now.time() current_weekday_code = now.weekday() # 0=Monday .. 6=Sunday schedule_warnings = [] is_on_schedule = True schedule_details = [] # Weekday check class_days = list(section.days.all().order_by("code")) if class_days: day_codes = [d.code for d in class_days] day_names = [d.get_code_display() for d in class_days] schedule_details.append(", ".join(day_names)) if current_weekday_code not in day_codes: today_name = now.strftime("%A") schedule_warnings.append( f"Today is {today_name}, but this class is scheduled on {', '.join(day_names)}." ) is_on_schedule = False # Time range check (30 min early grace, 15 min late grace) if section.start_time and section.end_time: schedule_details.append(f"{section.start_time.strftime('%H:%M')} - {section.end_time.strftime('%H:%M')}") start_dt = datetime.combine(today, section.start_time) - timedelta(minutes=30) end_dt = datetime.combine(today, section.end_time) + timedelta(minutes=15) now_dt = datetime.combine(today, current_time) if not (start_dt <= now_dt <= end_dt): schedule_warnings.append( f"Scheduled time is {section.start_time.strftime('%H:%M')} - {section.end_time.strftime('%H:%M')} (Current time: {now.strftime('%H:%M')})." ) is_on_schedule = False # Semester check if section.semester: if section.semester.start_date and section.semester.end_date: if today < section.semester.start_date: schedule_warnings.append( f"Semester '{section.semester.title}' has not started yet (starts {section.semester.start_date.strftime('%b %d, %Y')})." ) is_on_schedule = False elif today > section.semester.end_date: schedule_warnings.append( f"Semester '{section.semester.title}' ended on {section.semester.end_date.strftime('%b %d, %Y')}." ) is_on_schedule = False schedule_summary = " | ".join(schedule_details) if schedule_details else "No recurring schedule defined" context = { **self.admin_site.each_context(request), "opts": self.model._meta, "title": f"Start Scanning: {section.title}", "class_obj": section, "scan_url": reverse("admin:core_class_start_scanning_scan", args=[section.pk]), "view_attendance_url": reverse("admin:core_class_view_attendance", args=[section.pk]), "back_url": reverse("admin:core_class_change", args=[section.pk]), "is_on_schedule": is_on_schedule, "schedule_warnings": schedule_warnings, "schedule_summary": schedule_summary, "total_enrolled": section.students.count(), "current_local_time": now.strftime("%A, %b %d, %Y %I:%M %p"), } return TemplateResponse(request, "admin/core/class/start_scanning.html", context) def start_scanning_scan_view(self, request, object_id): if request.method != "POST": return JsonResponse({"status": "error", "message": "Method not allowed."}, status=405) section = self.get_object(request, object_id) if section is None: raise Http404("Class not found.") if not self.has_change_permission(request, section): raise PermissionDenied try: payload = json.loads(request.body.decode("utf-8")) except (json.JSONDecodeError, UnicodeDecodeError): return JsonResponse({"status": "error", "message": "Invalid JSON payload."}, status=400) scanned_card_number = (payload.get("card_number") or "").strip() if not scanned_card_number: return JsonResponse({"status": "error", "message": "Card number is required."}, status=400) sanitized_hex = re.sub(r"[^0-9A-Fa-f]", "", scanned_card_number) colon_lower = ":".join(sanitized_hex[i:i + 2] for i in range(0, len(sanitized_hex), 2)) if sanitized_hex else "" colon_upper = colon_lower.upper() if colon_lower else "" lookup_candidates = [ scanned_card_number, scanned_card_number.lower(), scanned_card_number.upper(), sanitized_hex, sanitized_hex.lower(), sanitized_hex.upper(), colon_lower, colon_upper, ] lookup_candidates = [value for value in dict.fromkeys(lookup_candidates) if value] person = Person.objects.select_related("department").filter(card_number__in=lookup_candidates).first() saved_card_number = person.card_number if person else scanned_card_number record = Record.objects.create(card_number=saved_card_number, class_obj=section, user=request.user) if person is None: return JsonResponse( { "status": "success", "known": False, "enrolled": False, "message": "Card saved to class attendance, but person is not registered.", "timestamp": timezone.localtime(record.timestamp).strftime("%Y-%m-%d %H:%M:%S"), } ) is_enrolled = section.students.filter(pk=person.pk).exists() return JsonResponse( { "status": "success", "known": True, "enrolled": is_enrolled, "message": "Attendance saved." if is_enrolled else "Attendance saved (Not enrolled in this class roster).", "timestamp": timezone.localtime(record.timestamp).strftime("%Y-%m-%d %H:%M:%S"), "person": { "id": person.id, "school_id": person.school_id, "name": person.display_name, "department": person.department.name if person.department else None, "image": person.image.url if person.image else None, "card_number": person.card_number, }, } ) def view_attendance_action(self, obj): base_url = reverse("admin:core_record_changelist") query_params = {"class": obj.title} # if obj.department_id: # query_params["department"] = obj.department_id if obj.area_id: query_params["area"] = obj.area_id if obj.semester_id and obj.semester.start_date and obj.semester.end_date: query_params["timestamp__date__gte"] = obj.semester.start_date.isoformat() query_params["timestamp__date__lte"] = obj.semester.end_date.isoformat() query_string = urlencode(query_params) url = f"{base_url}?{query_string}" return format_html( '' '' "View Records" "", url, ) view_attendance_action.short_description = "Records" def map_students_action(self, obj): url = reverse("admin:core_class_map_students", args=[obj.pk]) return format_html( '' '' "Map Students" "", url, ) map_students_action.short_description = "Map" def _apply_time_window_filter(self, queryset, filter_time_from, filter_time_to, section=None): if not (filter_time_from or filter_time_to): return queryset time_query = Q() if filter_time_from and filter_time_to: if filter_time_from <= filter_time_to: time_query = Q( timestamp__time__gte=filter_time_from, timestamp__time__lte=filter_time_to, ) else: time_query = Q(timestamp__time__gte=filter_time_from) | Q(timestamp__time__lte=filter_time_to) elif filter_time_from: time_query = Q(timestamp__time__gte=filter_time_from) elif filter_time_to: time_query = Q(timestamp__time__lte=filter_time_to) if section: return queryset.filter(Q(class_obj=section) | time_query) return queryset.filter(time_query) def _build_attendance_snapshot(self, section, query_params): raw_filter_date_from = (query_params.get("date_from") or "").strip() raw_filter_date_to = (query_params.get("date_to") or "").strip() today = timezone.localdate() filter_date_from = parse_date(raw_filter_date_from) if raw_filter_date_from else today filter_date_to = parse_date(raw_filter_date_to) if raw_filter_date_to else today filter_time_from = section.start_time filter_time_to = section.end_time filter_device = (query_params.get("device") or "").strip() filter_search = (query_params.get("search") or "").strip() raw_filter_person = (query_params.get("person") or "").strip() raw_filter_department = (query_params.get("department") or "").strip() try: filter_person_id = int(raw_filter_person) except (TypeError, ValueError): filter_person_id = None try: filter_department_id = int(raw_filter_department) except (TypeError, ValueError): filter_department_id = None students = list( section.students.all() .select_related("department") .order_by("lastname", "firstname", "school_id") ) student_cards = [student.card_number for student in students if student.card_number] student_cards_set = set(student_cards) student_department_ids = {student.department_id for student in students if student.department_id} if filter_department_id and filter_department_id not in student_department_ids: filter_department_id = None selected_person = next((student for student in students if student.id == filter_person_id), None) if filter_person_id and selected_person is None: filter_person_id = None selected_department_name = None if filter_department_id: selected_department_name = next( (student.department.name for student in students if student.department_id == filter_department_id and student.department), None, ) def matches_student_filters(student): if filter_person_id and student.id != filter_person_id: return False if filter_department_id and student.department_id != filter_department_id: return False if filter_search: search_haystack = " ".join( [ (student.display_name or "").lower(), (student.school_id or "").lower(), (student.card_number or "").lower(), ] ) if filter_search.lower() not in search_haystack: return False return True baseline_records = Record.objects.filter( Q(class_obj=section) | Q(card_number__in=student_cards) ).select_related("device", "device__area") if section.area_id: baseline_records = baseline_records.filter( Q(device__area_id=section.area_id) | Q(class_obj=section) ) if section.semester_id and section.semester.start_date and section.semester.end_date: baseline_records = baseline_records.filter( timestamp__date__gte=section.semester.start_date, timestamp__date__lte=section.semester.end_date, ) available_devices = list( baseline_records.exclude(device__isnull=True) .values_list("device__device_id", flat=True) .distinct() .order_by("device__device_id") ) filtered_records = baseline_records if filter_device: filtered_records = filtered_records.filter(device__device_id=filter_device) if filter_date_from: filtered_records = filtered_records.filter(timestamp__date__gte=filter_date_from) if filter_date_to: filtered_records = filtered_records.filter(timestamp__date__lte=filter_date_to) filtered_records = self._apply_time_window_filter(filtered_records, filter_time_from, filter_time_to, section) stats_by_card = { row["card_number"]: row for row in filtered_records.values("card_number").annotate( first_seen=Min("timestamp"), last_seen=Max("timestamp"), ) } source_by_card = {} for row in filtered_records.values("card_number", "device__device_id", "device__area__name"): card = row["card_number"] device_id = row["device__device_id"] area_name = row["device__area__name"] if device_id: source_label = f"{device_id} / {area_name or 'Unknown Area'}" else: source_label = "Phone Scan / Unknown Area" source_by_card.setdefault(card, set()).add(source_label) with_records_rows = [] no_records_rows = [] rendered_cards = set() for student in students: if not matches_student_filters(student): continue card_number = student.card_number or "" stat = stats_by_card.get(card_number) if stat and card_number: first_seen = timezone.localtime(stat["first_seen"]).strftime("%Y-%m-%d %H:%M:%S") if stat["first_seen"] else "-" last_seen = timezone.localtime(stat["last_seen"]).strftime("%Y-%m-%d %H:%M:%S") if stat["last_seen"] else "-" sources = sorted(source_by_card.get(card_number, [])) source_text = ", ".join(sources) if sources else "Unknown Source / Unknown Area" with_records_rows.append( [ student.school_id or "N/A", student.lastname or "N/A", student.firstname or "N/A", student.department.name if student.department else "", first_seen, last_seen, source_text, ] ) rendered_cards.add(card_number) else: reason = "No card number" if not card_number else "No matching scan in selected filters" no_records_rows.append( [ student.school_id or "N/A", student.lastname or "N/A", student.firstname or "N/A", student.department.name if student.department else "", reason, ] ) # Also show cards scanned directly into this class even if not in class students. extra_cards = [card for card in stats_by_card.keys() if card and card not in rendered_cards] if extra_cards: people_by_card = { person.card_number: person for person in Person.objects.select_related("department").filter(card_number__in=extra_cards) } for card_number in sorted(extra_cards): if card_number in student_cards_set: continue person = people_by_card.get(card_number) if filter_person_id and (not person or person.id != filter_person_id): continue if filter_department_id and (not person or person.department_id != filter_department_id): continue if filter_search: search_haystack = " ".join( [ (person.display_name if person else "").lower(), (person.school_id if person else "").lower(), (card_number or "").lower(), ] ) if filter_search.lower() not in search_haystack: continue stat = stats_by_card.get(card_number) first_seen = timezone.localtime(stat["first_seen"]).strftime("%Y-%m-%d %H:%M:%S") if stat and stat["first_seen"] else "-" last_seen = timezone.localtime(stat["last_seen"]).strftime("%Y-%m-%d %H:%M:%S") if stat and stat["last_seen"] else "-" sources = sorted(source_by_card.get(card_number, [])) source_text = ", ".join(sources) if sources else "Unknown Source / Unknown Area" with_records_rows.append( [ person.school_id if person and person.school_id else "N/A", person.lastname if person else "N/A", person.firstname if person else "N/A", person.department.name if person and person.department else "", first_seen, last_seen, source_text, ] ) active_filter_summary = [] if filter_date_from: active_filter_summary.append(f"From: {filter_date_from.isoformat()}") if filter_date_to: active_filter_summary.append(f"To: {filter_date_to.isoformat()}") if filter_time_from: active_filter_summary.append(f"From Time: {filter_time_from.strftime('%H:%M')}") if filter_time_to: active_filter_summary.append(f"To Time: {filter_time_to.strftime('%H:%M')}") if filter_device: active_filter_summary.append(f"Device: {filter_device}") if filter_department_id: active_filter_summary.append(f"Department: {selected_department_name or filter_department_id}") if filter_person_id and selected_person: active_filter_summary.append(f"Person: {selected_person.display_name}") if filter_search: active_filter_summary.append(f"Search: {filter_search}") return { "with_records_rows": with_records_rows, "no_records_rows": no_records_rows, "active_filter_summary": active_filter_summary, "filter_device": filter_device, "filter_date_from": filter_date_from.isoformat() if filter_date_from else "", "filter_date_to": filter_date_to.isoformat() if filter_date_to else "", "filter_time_from": filter_time_from.strftime("%H:%M") if filter_time_from else "", "filter_time_to": filter_time_to.strftime("%H:%M") if filter_time_to else "", "filter_search": filter_search, "filter_person": str(filter_person_id) if filter_person_id else "", "filter_department": str(filter_department_id) if filter_department_id else "", "available_devices": available_devices, } def _write_attendance_sheet(self, worksheet, section, sheet_title, headers, rows, active_filter_summary): row_index = 1 worksheet.title = sheet_title worksheet.cell(row=row_index, column=1, value=f"Class Attendance - {sheet_title}") row_index += 1 worksheet.cell(row=row_index, column=1, value=f"Class: {section.title}") row_index += 1 worksheet.cell(row=row_index, column=1, value=f"Area: {section.area.name if section.area else 'N/A'}") row_index += 1 worksheet.cell(row=row_index, column=1, value=f"Semester: {section.semester.title if section.semester else 'N/A'}") row_index += 1 worksheet.cell( row=row_index, column=1, value=f"Exported At: {timezone.localtime(timezone.now()).strftime('%Y-%m-%d %H:%M:%S')}", ) row_index += 2 worksheet.cell(row=row_index, column=1, value="Filters") row_index += 1 if active_filter_summary: for filter_label in active_filter_summary: worksheet.cell(row=row_index, column=1, value=filter_label) row_index += 1 else: worksheet.cell(row=row_index, column=1, value="None") row_index += 1 row_index += 1 for col_index, header in enumerate(headers, start=1): worksheet.cell(row=row_index, column=col_index, value=header) row_index += 1 for item in rows: for col_index, value in enumerate(item, start=1): worksheet.cell(row=row_index, column=col_index, value=value) row_index += 1 return row_index def _build_class_calendar(self, class_obj, query_params): month_raw = (query_params.get("calendar_month") or "").strip() raw_filter_person = (query_params.get("person") or "").strip() raw_filter_department = (query_params.get("department") or "").strip() try: filter_person_id = int(raw_filter_person) except (TypeError, ValueError): filter_person_id = None try: filter_department_id = int(raw_filter_department) except (TypeError, ValueError): filter_department_id = None return build_class_calendar_context( class_obj, month_raw=month_raw, filter_person_id=filter_person_id, filter_department_id=filter_department_id, ) def attendance_day_detail_view(self, request, object_id): section = self.get_object(request, object_id) if section is None: return JsonResponse({"error": "Class not found."}, status=404) if not self.has_view_permission(request, section): return JsonResponse({"error": "Permission denied."}, status=403) raw_date = (request.GET.get("date") or "").strip() target_date = parse_date(raw_date) if raw_date else None if not target_date: return JsonResponse({"error": "Invalid or missing date parameter."}, status=400) snapshot_query = { "date_from": target_date.isoformat(), "date_to": target_date.isoformat(), } for key in ("device", "search", "person", "department"): value = (request.GET.get(key) or "").strip() if value: snapshot_query[key] = value snapshot = self._build_attendance_snapshot(section, snapshot_query) def _time_only(dt_str): """Extract HH:MM:SS from 'YYYY-MM-DD HH:MM:SS' or return as-is.""" parts = (dt_str or "").split(" ", 1) return parts[1] if len(parts) == 2 else dt_str with_records = [ {"school_id": row[0], "lastname": row[1], "firstname": row[2], "department": row[3], "first_seen": _time_only(row[4]), "last_seen": _time_only(row[5])} for row in snapshot["with_records_rows"] ] no_records = [ {"school_id": row[0], "lastname": row[1], "firstname": row[2], "department": row[3]} for row in snapshot["no_records_rows"] ] return JsonResponse({ "date": target_date.isoformat(), "with_records": with_records, "no_records": no_records, }) def view_attendance_view(self, request, object_id): section = self.get_object(request, object_id) if section is None: raise Http404("Class not found.") if not self.has_view_permission(request, section): raise PermissionDenied calendar_data = self._build_class_calendar(section, request.GET) filter_search = (request.GET.get("search") or "").strip() filter_person = (request.GET.get("person") or "").strip() filter_department = (request.GET.get("department") or "").strip() students_for_filters = list( section.students.all() .select_related("department") .order_by("lastname", "firstname", "school_id") ) person_filter_options_json = json.dumps( [ { "id": student.id, "display": f"{student.display_name or 'Unknown'} | {student.school_id or 'No ID'}", "name": student.display_name or "", "school_id": student.school_id or "", "card_number": student.card_number or "", "department": student.department.name if student.department else "", } for student in students_for_filters ] ) department_filter_options_json = json.dumps( [ {"id": dept.id, "name": dept.name} for dept in Department.objects.filter( id__in={student.department_id for student in students_for_filters if student.department_id} ).order_by("name") ] ) export_base_url = reverse("admin:core_class_view_attendance_export_excel", args=[section.pk]) querystring = request.GET.urlencode() export_excel_url = f"{export_base_url}?{querystring}" if querystring else export_base_url day_detail_url = reverse("admin:core_class_attendance_day_detail", args=[section.pk]) roster_export_url = reverse("admin:core_class_view_attendance_export_roster", args=[section.pk]) total_students = section.students.count() context = { **self.admin_site.each_context(request), "opts": self.model._meta, "title": f"Attendances for {section.title}", "class_obj": section, "total_students": total_students, "back_url": reverse("admin:core_class_changelist"), "export_excel_url": export_excel_url, "day_export_base_url": export_base_url, "day_detail_url": day_detail_url, "roster_export_url": roster_export_url, "filter_search": filter_search, "filter_person": filter_person, "filter_department": filter_department, "person_filter_options_json": person_filter_options_json, "department_filter_options_json": department_filter_options_json, **calendar_data, } return TemplateResponse(request, "admin/core/class/view_attendance.html", context) def view_attendance_export_excel_view(self, request, object_id): section = self.get_object(request, object_id) if section is None: raise Http404("Class not found.") if not self.has_view_permission(request, section): raise PermissionDenied snapshot = self._build_attendance_snapshot(section, request.GET) workbook = Workbook() attendance_sheet = workbook.active next_row = self._write_attendance_sheet( worksheet=attendance_sheet, section=section, sheet_title="Attendance", headers=["School ID", "Last Name", "First Name", "Department", "First Log", "Last Log", "Device + Area"], rows=snapshot["with_records_rows"], active_filter_summary=snapshot["active_filter_summary"], ) next_row += 1 attendance_sheet.cell(row=next_row, column=1, value="No Record") next_row += 1 no_record_headers = ["School ID", "Last Name", "First Name", "Department", "Reason"] for col_index, header in enumerate(no_record_headers, start=1): attendance_sheet.cell(row=next_row, column=col_index, value=header) next_row += 1 for item in snapshot["no_records_rows"]: for col_index, value in enumerate(item, start=1): attendance_sheet.cell(row=next_row, column=col_index, value=value) next_row += 1 for col_cells in attendance_sheet.columns: max_len = max((len(str(c.value)) for c in col_cells if c.value), default=0) attendance_sheet.column_dimensions[get_column_letter(col_cells[0].column)].width = max_len + 2 response = HttpResponse( content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ) date_from_str = snapshot.get("filter_date_from", "") date_to_str = snapshot.get("filter_date_to", "") if date_from_str and date_to_str and date_from_str == date_to_str: target_date = parse_date(date_from_str) class_name = re.sub(r"\s+", "_", section.title or "class") class_name = re.sub(r"[^\w_-]", "", class_name).strip("_") or "class" day_name = target_date.strftime("%A") date_display = f"{target_date.strftime('%B')} {target_date.day}, {target_date.year}" filename = f"{class_name}-{day_name}-{date_display}.xlsx" else: safe_title = re.sub(r"[^A-Za-z0-9_-]+", "-", section.title or "class").strip("-") or "class" filename = f"{safe_title}-attendance-{timezone.localtime(timezone.now()).strftime('%Y%m%d-%H%M%S')}.xlsx" response["Content-Disposition"] = f'attachment; filename="{filename}"' workbook.save(response) return response def view_attendance_export_roster_view(self, request, object_id): section = self.get_object(request, object_id) if section is None: raise Http404("Class not found.") if not self.has_view_permission(request, section): raise PermissionDenied period = (request.GET.get("period") or "month").strip() if period == "semester": if not (section.semester_id and section.semester.start_date and section.semester.end_date): raise Http404("No semester configured for this class.") period_start = section.semester.start_date period_end = section.semester.end_date period_label = section.semester.title safe_period = re.sub(r"[^A-Za-z0-9_]+", "_", section.semester.title or "semester").strip("_") or "semester" sheet_title_label = "Semestral Class Attendance" else: month_raw = (request.GET.get("month") or "").strip() if month_raw: try: parts = month_raw.split("-") year, month = int(parts[0]), int(parts[1]) except (IndexError, ValueError): today = timezone.localdate() year, month = today.year, today.month else: today = timezone.localdate() year, month = today.year, today.month _, total_days = monthrange(year, month) period_start = date(year, month, 1) period_end = date(year, month, total_days) period_label = f"{period_start.strftime('%B')} {year}" safe_period = f"{period_start.strftime('%B')}_{year}" sheet_title_label = f"Monthly Class Attendance — {period_label}" day_codes = {weekday.code for weekday in section.days.all()} scheduled_dates = [] current = period_start while current <= period_end: if current.weekday() in day_codes: scheduled_dates.append(current) current += timedelta(days=1) students = list( section.students.all() .select_related("department") .order_by("lastname", "firstname", "school_id") ) student_cards = [s.card_number for s in students if s.card_number] records_qs = Record.objects.filter( Q(class_obj=section) | Q(card_number__in=student_cards), timestamp__date__gte=period_start, timestamp__date__lte=period_end, ) if section.area_id: records_qs = records_qs.filter( Q(device__area_id=section.area_id) | Q(class_obj=section) ) if section.semester_id and section.semester.start_date and section.semester.end_date: records_qs = records_qs.filter( timestamp__date__gte=section.semester.start_date, timestamp__date__lte=section.semester.end_date, ) records_qs = self._apply_time_window_filter(records_qs, section.start_time, section.end_time, section) presence_set = set( records_qs.annotate(_day=TruncDate("timestamp")).values_list("card_number", "_day") ) workbook = Workbook() ws = workbook.active ws.title = "Attendance Roster" row = 1 ws.cell(row=row, column=1, value=sheet_title_label) row += 1 ws.cell(row=row, column=1, value=f"Class: {section.title}") row += 1 ws.cell(row=row, column=1, value=f"Area: {section.area.name if section.area else 'N/A'}") row += 1 ws.cell(row=row, column=1, value=f"Semester: {section.semester.title if section.semester else 'N/A'}") row += 1 if section.start_time and section.end_time: time_str = f"{section.start_time.strftime('%I:%M %p')} - {section.end_time.strftime('%I:%M %p')}" elif section.start_time: time_str = section.start_time.strftime('%I:%M %p') else: time_str = "N/A" ws.cell(row=row, column=1, value=f"Time: {time_str}") row += 1 ws.cell(row=row, column=1, value=f"Period: {period_label}") row += 1 ws.cell(row=row, column=1, value=f"Exported At: {timezone.localtime(timezone.now()).strftime('%Y-%m-%d %H:%M:%S')}") row += 2 BASE_COLS = 4 ws.cell(row=row, column=1, value="School ID") ws.cell(row=row, column=2, value="Last Name") ws.cell(row=row, column=3, value="First Name") ws.cell(row=row, column=4, value="Department") total_date_cols = len(scheduled_dates) for i, d in enumerate(scheduled_dates, start=1): ws.cell(row=row, column=BASE_COLS + i, value=f"{d.strftime('%b')} {d.day} ({d.strftime('%a')})") ws.cell(row=row, column=BASE_COLS + total_date_cols + 1, value="Present") ws.cell(row=row, column=BASE_COLS + total_date_cols + 2, value="Absent") ws.cell(row=row, column=BASE_COLS + total_date_cols + 3, value="Total") ws.cell(row=row, column=BASE_COLS + total_date_cols + 4, value="%") _header_font = Font(bold=True) for col in range(1, BASE_COLS + total_date_cols + 5): ws.cell(row=row, column=col).font = _header_font row += 1 for student in students: card = student.card_number or "" ws.cell(row=row, column=1, value=student.school_id or "N/A") ws.cell(row=row, column=2, value=student.lastname or "N/A") ws.cell(row=row, column=3, value=student.firstname or "N/A") ws.cell(row=row, column=4, value=student.department.name if student.department else "") present_count = 0 _present_fill = PatternFill(fill_type="solid", fgColor="1D4ED8") _absent_fill = PatternFill(fill_type="solid", fgColor="DC2626") _white_font = Font(color="FFFFFF", bold=True) for i, d in enumerate(scheduled_dates, start=1): is_present = bool(card) and (card, d) in presence_set cell = ws.cell(row=row, column=BASE_COLS + i, value="P" if is_present else "A") cell.fill = _present_fill if is_present else _absent_fill cell.font = _white_font cell.alignment = Alignment(horizontal="center") if is_present: present_count += 1 absent_count = total_date_cols - present_count pct = round(present_count * 100 / total_date_cols) if total_date_cols else 0 ws.cell(row=row, column=BASE_COLS + total_date_cols + 1, value=present_count) ws.cell(row=row, column=BASE_COLS + total_date_cols + 2, value=absent_count) ws.cell(row=row, column=BASE_COLS + total_date_cols + 3, value=total_date_cols) ws.cell(row=row, column=BASE_COLS + total_date_cols + 4, value=f"{pct}%") row += 1 for col_cells in ws.columns: max_len = max((len(str(c.value)) for c in col_cells if c.value), default=0) ws.column_dimensions[get_column_letter(col_cells[0].column)].width = max_len + 2 class_name = re.sub(r"\s+", "_", section.title or "class") class_name = re.sub(r"[^\w_-]", "", class_name).strip("_") or "class" filename = f"{class_name}-Attendance-{safe_period}.xlsx" response = HttpResponse( content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ) response["Content-Disposition"] = f'attachment; filename="{filename}"' workbook.save(response) return response def _student_avatar_html(self, student): if student.image: return format_html( '{}', student.image.url, student.display_name or "Student", ) initials = "".join( part[:1].upper() for part in ((student.firstname or "").strip(), (student.lastname or "").strip()) if part )[:2] if not initials: fallback = (student.school_id or student.card_number or "?").strip() initials = fallback[:1].upper() if fallback else "?" return format_html( '{}', initials, ) def manage_students_view(self, request, object_id): section = self.get_object(request, object_id) if section is None: raise Http404("Class not found.") if not self.has_change_permission(request, section): raise PermissionDenied("You do not have permission to manage students for this class.") if request.method == "POST" and request.headers.get("X-Requested-With") == "XMLHttpRequest": action = request.POST.get("action") if action == "add_student": student_id = request.POST.get("student_id") try: student = Person.objects.get(pk=student_id) if student.is_archived: return JsonResponse({"success": False, "message": "Cannot add an archived person to a class."}) if student not in section.students.all(): section.students.add(student) logger.info( "Student added to class class_id=%s student_id=%s user=%s", section.pk, student_id, request.user.username, ) return JsonResponse({"success": True, "message": f"{student.display_name} added to class."}) return JsonResponse({"success": False, "message": "Student already in class."}) except Person.DoesNotExist: return JsonResponse({"success": False, "message": "Student not found."}) if action == "remove_student": student_id = request.POST.get("student_id") try: student = Person.objects.get(pk=student_id) section.students.remove(student) logger.info( "Student removed from class class_id=%s student_id=%s user=%s", section.pk, student_id, request.user.username, ) return JsonResponse({"success": True, "message": f"{student.display_name} removed from class."}) except Person.DoesNotExist: return JsonResponse({"success": False, "message": "Student not found."}) if action == "bulk_add_persons": try: person_ids = json.loads(request.POST.get("person_ids", "[]")) except (json.JSONDecodeError, TypeError): return JsonResponse({"success": False, "message": "Invalid person IDs."}) valid_ids = [] for raw_id in person_ids: try: valid_ids.append(int(raw_id)) except (TypeError, ValueError): continue if not valid_ids: return JsonResponse({"success": False, "message": "No valid person IDs provided."}) existing_ids = set(section.students.filter(pk__in=valid_ids).values_list("pk", flat=True)) ids_to_add = [pid for pid in valid_ids if pid not in existing_ids] if ids_to_add: persons_to_add = list(Person.objects.filter(pk__in=ids_to_add, is_archived=False)) section.students.add(*persons_to_add) logger.info( "Bulk added %d persons to class class_id=%s user=%s", len(persons_to_add), section.pk, request.user.username, ) return JsonResponse({ "success": True, "message": f"{len(persons_to_add)} person(s) added to class.", "added_count": len(persons_to_add), }) return JsonResponse({"success": False, "message": "All selected persons are already in class."}) current_students = section.students.all().order_by("lastname", "firstname", "school_id") total_students_count = current_students.count() paginator = Paginator(current_students, 100) page_number = request.GET.get("page", 1) page_obj = paginator.get_page(page_number) page_range = paginator.get_elided_page_range(page_obj.number, on_each_side=2, on_ends=1) all_persons = Person.objects.filter(is_archived=False).order_by("lastname", "firstname", "school_id") available_students = all_persons.exclude(pk__in=current_students.values_list("pk", flat=True)) available_students_json = json.dumps( [ { "id": student.id, "name": student.display_name, "nameNorm": (student.display_name or "").lower(), "schoolId": student.school_id or "", "schoolIdNorm": (student.school_id or "").lower(), } for student in available_students ] ) departments_json = json.dumps( [{"id": d.id, "name": d.name} for d in Department.objects.all().order_by("name")] ) classes_json = json.dumps( [ { "id": c.id, "title": c.title, "semester": c.semester.title if c.semester else "", "department": c.department.name if c.department else "", } for c in Class.objects.exclude(pk=section.pk) .select_related("semester", "department") .order_by("title") ] ) events_json = json.dumps( [{"id": e.id, "title": e.title} for e in Event.objects.all().order_by("-start_datetime")] ) context = self.admin_site.each_context(request) current_students_table_rows = [ [ format_html( '
' '{}' '{}' '
', self._student_avatar_html(student), student.display_name or "N/A", ), student.school_id or "No ID", format_html( '", student.id, student.display_name or "Student", ), ] for student in page_obj.object_list ] context.update( { "class_obj": section, "current_students": page_obj.object_list, "total_students_count": total_students_count, "page_obj": page_obj, "page_range": page_range, "available_students_json": available_students_json, "departments_json": departments_json, "classes_json": classes_json, "events_json": events_json, "filter_persons_url": reverse("admin:core_class_filter_persons", args=[section.pk]), "current_students_table_data": { "collapsible": False, "headers": ["Student", "School ID", "Actions"], "rows": current_students_table_rows, }, "map_url": reverse("admin:core_class_map_students", args=[section.pk]), "back_url": reverse("admin:core_class_change", args=[section.pk]), "manage_url": request.path, "opts": self.model._meta, "has_change_permission": True, } ) return TemplateResponse(request, "admin/core/class/manage_students.html", context) def filter_persons_view(self, request, object_id): section = self.get_object(request, object_id) if section is None: raise Http404("Class not found.") if not request.user.is_staff and section.instructor_id != request.user.id and not section.co_instructors.filter(pk=request.user.pk).exists(): raise PermissionDenied("You do not have permission to manage students for this class.") mode = (request.GET.get("mode") or "").strip() person_type = (request.GET.get("person_type") or "").strip() raw_entity_id = (request.GET.get("entity_id") or "").strip() if mode == "department" and raw_entity_id == "all": persons = Person.objects.filter(is_archived=False) else: try: entity_id = int(raw_entity_id) except (TypeError, ValueError): return JsonResponse({"success": False, "message": "Invalid entity ID."}) if mode == "department": persons = Person.objects.filter(department_id=entity_id, is_archived=False) elif mode == "class": try: source_class = Class.objects.get(pk=entity_id) persons = source_class.students.all() except Class.DoesNotExist: return JsonResponse({"success": False, "message": "Class not found."}) elif mode == "event": try: source_event = Event.objects.get(pk=entity_id) persons = source_event.participants.all() except Event.DoesNotExist: return JsonResponse({"success": False, "message": "Event not found."}) else: return JsonResponse({"success": False, "message": "Invalid filter mode."}) if person_type in ("student", "employee"): persons = persons.filter(type=person_type) enrolled_ids = section.students.values_list("pk", flat=True) persons = ( persons.exclude(pk__in=enrolled_ids) .select_related("department") .order_by("lastname", "firstname", "school_id") ) data = [ { "id": p.id, "name": p.display_name, "schoolId": p.school_id or "", "type": p.type, "department": p.department.name if p.department else "", } for p in persons ] return JsonResponse({"success": True, "persons": data}) def map_students_view(self, request, object_id): section = self.get_object(request, object_id) if section is None: raise Http404("Class not found.") if not self.has_change_permission(request, section): raise PermissionDenied if not request.user.is_staff and section.instructor_id != request.user.id and not section.co_instructors.filter(pk=request.user.pk).exists(): raise PermissionDenied preview_headers = [] preview_rows = [] match_count = 0 mapped_count = 0 upload_filename = None map_mode = request.POST.get("map_mode") or request.GET.get("map_mode") or "class_list" if request.method == "POST": action = None if request.POST: action = request.POST.get("action") if not action and request.content_type and "json" in request.content_type: try: body_str = request.body.decode("utf-8") if isinstance(request.body, bytes) else request.body body_data = json.loads(body_str) if body_str else {} action = body_data.get("action") except (json.JSONDecodeError, UnicodeDecodeError): pass if action == "map": return self._handle_map(request, section) excel_file = request.FILES.get("excel_file") upload_filename = getattr(excel_file, "name", None) logger.info( "Class migration preview started class_id=%s section_title=%r user=%s file=%r map_mode=%r", section.pk, section.title, getattr(request.user, "username", None), upload_filename, map_mode, ) if not excel_file: messages.error(request, "Please upload an Excel file first.") logger.warning("Class migration preview failed: no file uploaded class_id=%s", section.pk) else: try: workbook = self._load_excel_workbook(excel_file, filename=upload_filename) logger.info("Excel loaded class_id=%s sheet_names=%s", section.pk, workbook.sheetnames) preview_headers, preview_rows, match_count, mapped_count = ( self._build_workbook_preview( workbook=workbook, section_title=section.title, section_id=section.pk, map_mode=map_mode, ) ) if match_count == 0: if map_mode == "enrollment_report": messages.warning(request, f'No rows matched class "{section.title}".') else: messages.warning(request, "No student rows could be parsed from the uploaded class list.") logger.warning( "No matching rows found class_id=%s section_title=%r file=%r map_mode=%r", section.pk, section.title, upload_filename, map_mode, ) else: messages.success( request, f"Preview loaded: {match_count} student row(s) found, {mapped_count} matched with system persons.", ) logger.info( "Preview built class_id=%s matches=%s mapped_students=%s map_mode=%r", section.pk, match_count, mapped_count, map_mode, ) except ImportError: messages.error(request, "Excel parser dependency is missing. Install openpyxl and try again.") logger.exception("openpyxl import missing during class migration preview class_id=%s", section.pk) except Exception as exc: messages.error(request, f"Unable to parse Excel file: {exc}") logger.exception("Error parsing excel for class migration preview class_id=%s", section.pk) context = { **self.admin_site.each_context(request), "opts": self.model._meta, "original": section, "class_obj": section, "title": f"Map Students - {section.title}", "selected_map_mode": map_mode, "preview_headers": preview_headers, "preview_headers_count": len(preview_headers), "preview_rows": preview_rows, "preview_rows_json": json.dumps(self._serialize_preview_rows(preview_rows)), "table_data": self._build_preview_table_data(preview_headers), "match_count": match_count, "mapped_count": mapped_count, "upload_filename": upload_filename, "page": 1, "rows_per_page": 25, "back_url": reverse("admin:core_class_change", args=[section.pk]), "map_button_attrs": { "id": "map-selected-btn", "type": "button", "onclick": "confirmMap()", }, } return TemplateResponse(request, "admin/core/class/map_students.html", context) def _handle_map(self, request, section): try: student_names = [] if request.POST.get("student_names"): student_names_str = request.POST.get("student_names") student_names = ( json.loads(student_names_str) if isinstance(student_names_str, str) else student_names_str ) else: body_str = request.body.decode("utf-8") if isinstance(request.body, bytes) else request.body data = json.loads(body_str) if body_str else {} student_names = data.get("student_names", []) logger.info("Migration request received class_id=%s student_count=%s", section.pk, len(student_names)) logger.info("Student names to map class_id=%s count=%s", section.pk, len(student_names)) if not student_names: return JsonResponse({"success": False, "error": "No students to map"}, status=400) mapped_count = 0 added_students = [] failed_names = [] for student_item in student_names: if not student_item: logger.warning("Empty student item encountered class_id=%s", section.pk) continue item_str = str(student_item).strip() if not item_str: continue try: person = Person.objects.filter(type="student").filter( Q(school_id__iexact=item_str) | Q(card_number__iexact=item_str) | Q(firstname__iexact=item_str) | Q(lastname__iexact=item_str) ).first() if person is None and item_str.isdigit(): person = Person.objects.filter(type="student", pk=int(item_str)).first() if person is None: normalized_student_name = self._normalize_for_match(item_str) for candidate in Person.objects.filter(type="student"): if self._normalize_for_match(candidate.display_name) == normalized_student_name: person = candidate break if person: section.students.add(person) added_students.append(person) mapped_count += 1 logger.info( "Student added class_id=%s student_item=%s person_id=%s", section.pk, item_str, person.id, ) else: failed_names.append(item_str) logger.warning("Student not found class_id=%s student_item=%s", section.pk, item_str) except Exception as error: logger.exception( "Error adding student class_id=%s student_item=%s error=%s", section.pk, item_str, str(error), ) failed_names.append(item_str) logger.info( "Class mapping completed class_id=%s mapped_count=%s failed_count=%s by_user=%s", section.pk, mapped_count, len(failed_names), getattr(request.user, "username", None), ) return JsonResponse( { "success": True, "mapped_count": mapped_count, "failed_count": len(failed_names), }, status=200, ) except json.JSONDecodeError as error: error_msg = f"Invalid JSON format: {str(error)}" logger.exception("Failed to parse map request JSON class_id=%s error=%s", section.pk, error_msg) return JsonResponse({"success": False, "error": error_msg}, status=400) except UnicodeDecodeError as error: error_msg = f"Failed to decode request body: {str(error)}" logger.exception("Unicode decode error class_id=%s error=%s", section.pk, error_msg) return JsonResponse({"success": False, "error": error_msg}, status=400) except Exception as exc: error_msg = f"Internal server error: {str(exc)}" logger.exception("Error during student mapping class_id=%s", section.pk) return JsonResponse({"success": False, "error": error_msg}, status=500) def _build_workbook_preview(self, workbook, section_title, section_id=None, map_mode="class_list"): if map_mode == "enrollment_report": return self._build_enrollment_report_preview( workbook=workbook, section_title=section_title, section_id=section_id, ) return self._build_class_list_preview( workbook=workbook, section_id=section_id, ) def _build_class_list_preview(self, workbook, section_id=None): worksheet = workbook.active or workbook.worksheets[0] logger.info( "Building class list preview class_id=%s sheet=%r max_row=%s max_column=%s", section_id, worksheet.title, worksheet.max_row, worksheet.max_column, ) header_row_number = 1 raw_header = [] best_header_probe_score = -1 for probe_row_number, row in enumerate( worksheet.iter_rows(min_row=1, max_row=min(worksheet.max_row, 40), values_only=True), start=1, ): row_values = [self._normalize_cell(value) for value in row] row_keys = [self._normalize_for_match(value) for value in row_values] non_empty_count = sum(1 for value in row_values if value) has_student_name = any( ("studentname" in value) or ("student" in value and "name" in value) or ("fullname" in value) or (value == "name") for value in row_keys if value ) has_student_id = any( ("studentno" in value) or ("studentnum" in value) or ("studentid" in value) or ("schoolid" in value) or ("idno" in value) or ("idnum" in value) for value in row_keys if value ) has_sex = any(value in ("sex", "gender") for value in row_keys if value) has_program = any(("program" in value) or ("course" in value) for value in row_keys if value) has_status = any(("status" in value) or ("enrol" in value) for value in row_keys if value) probe_score = ( (20 if has_student_name else 0) + (20 if has_student_id else 0) + (5 if has_sex else 0) + (5 if has_program else 0) + (5 if has_status else 0) + non_empty_count ) if probe_score > best_header_probe_score: best_header_probe_score = probe_score header_row_number = probe_row_number raw_header = row_values if has_student_name and has_student_id: break if not raw_header: raw_header = [f"Column {index + 1}" for index in range(worksheet.max_column)] header_labels = [] for index, value in enumerate(raw_header): label = value if value else f"Column {index + 1}" if re.match(r"^\{\{.*\}\}$", label): label = f"Column {index + 1}" header_labels.append(label) normalized_headers = [value.lower() for value in header_labels] student_name_col = self._find_column_index( normalized_headers, [ "student name", "student_name", "studentname", "name of student", "fullname", "full name", "name", ], ) student_id_col = self._find_column_index( normalized_headers, [ "student number", "student no", "student no.", "student #", "student_no", "student id", "student_id", "school id", "school_id", "id number", "id no", "id no.", ], ) sex_col = self._find_column_index(normalized_headers, ["sex", "gender"]) program_col = self._find_column_index( normalized_headers, ["program code", "program_code", "program", "course code", "course"], ) status_col = self._find_column_index( normalized_headers, ["enrollment status", "enrollment_status", "status", "enrolment status"], ) logger.info( "Class list header analysis class_id=%s header_row=%s name_col=%s id_col=%s sex_col=%s prog_col=%s status_col=%s headers=%s", section_id, header_row_number, student_name_col, student_id_col, sex_col, program_col, status_col, header_labels, ) preview_rows = [] mapped_count = 0 total_match_count = 0 max_preview_rows = 300 # Build clean relevant display headers relevant_headers = ["Student Name", "Sex", "Student Number", "Program Code", "Enrollment Status"] for row_number, row in enumerate( worksheet.iter_rows(min_row=header_row_number + 1, values_only=True), start=header_row_number + 1, ): row_cells = [self._normalize_cell(value) for value in row] if not any(row_cells): continue raw_name = row_cells[student_name_col] if student_name_col is not None and student_name_col < len(row_cells) else "" raw_id = row_cells[student_id_col] if student_id_col is not None and student_id_col < len(row_cells) else "" sex_val = row_cells[sex_col] if sex_col is not None and sex_col < len(row_cells) else "" program_val = row_cells[program_col] if program_col is not None and program_col < len(row_cells) else "" status_val = row_cells[status_col] if status_col is not None and status_col < len(row_cells) else "" # Check for footer/summary rows name_lower = raw_name.lower().strip() if not raw_name and not raw_id: continue if name_lower.startswith(("total", "date generated", "page ", "prepared by", "faculty:", "schedule:")): continue # Strip leading index numbering from student name (e.g. "1 ABLAYAN, JABAR ACE" -> "ABLAYAN, JABAR ACE") cleaned_name = re.sub(r"^\s*\d+[\.\-\s\)]+", "", raw_name).strip() cleaned_id = raw_id.strip() total_match_count += 1 mapped_person = None if cleaned_id: mapped_person = Person.objects.filter(type="student").filter( Q(school_id__iexact=cleaned_id) | Q(card_number__iexact=cleaned_id) ).first() if mapped_person is None and cleaned_name: normalized_student_name = self._normalize_for_match(cleaned_name) for candidate in Person.objects.filter(type="student"): if self._normalize_for_match(candidate.display_name) == normalized_student_name: mapped_person = candidate break if mapped_person: mapped_count += 1 if len(preview_rows) < max_preview_rows: preview_rows.append( { "row_number": row_number, "cells": [cleaned_name or "—", sex_val or "—", cleaned_id or "—", program_val or "—", status_val or "—"], "student_id": cleaned_id, "mapped_person": mapped_person, } ) logger.info( "Class list preview result class_id=%s total_rows=%s mapped_students=%s preview_count=%s", section_id, total_match_count, mapped_count, len(preview_rows), ) return relevant_headers, preview_rows, total_match_count, mapped_count def _build_enrollment_report_preview(self, workbook, section_title, section_id=None): target_sheet_name = "Enrollment Report" worksheet = workbook[target_sheet_name] if target_sheet_name in workbook.sheetnames else None if worksheet is None: logger.warning( "Target worksheet missing class_id=%s expected_sheet=%r available_sheets=%s", section_id, target_sheet_name, workbook.sheetnames, ) raise ValueError( f'Worksheet "{target_sheet_name}" not found. Available sheets: {", ".join(workbook.sheetnames)}' ) logger.info( "Using target worksheet class_id=%s sheet=%r max_row=%s max_column=%s", section_id, worksheet.title, worksheet.max_row, worksheet.max_column, ) header_labels, preview_rows, total_match_count, mapped_count = ( self._build_class_preview( worksheet=worksheet, section_title=section_title, section_id=section_id, ) ) logger.info( "Selected worksheet class_id=%s sheet=%r matches=%s mapped_students=%s", section_id, worksheet.title, total_match_count, mapped_count, ) return header_labels, preview_rows, total_match_count, mapped_count def _build_class_preview(self, worksheet, section_title, section_id=None): section_title_norm = self._normalize_cell(section_title).lower() section_title_key = self._normalize_for_match(section_title) logger.info( "Build class preview class_id=%s sheet=%r section_title=%r normalized_section_title=%r normalized_key=%r", section_id, worksheet.title, section_title, section_title_norm, section_title_key, ) header_row_number = 1 raw_header = [] best_header_probe_score = -1 for probe_row_number, row in enumerate( worksheet.iter_rows(min_row=1, max_row=min(worksheet.max_row, 30), values_only=True), start=1, ): row_values = [self._normalize_cell(value) for value in row] row_keys = [self._normalize_for_match(value) for value in row_values] non_empty_count = sum(1 for value in row_values if value) has_section = any("section" in value for value in row_keys if value) has_student = any( ("student" in value) or ("school" in value) or ("studentno" in value) or ("studentid" in value) for value in row_keys if value ) probe_score = (10 if has_section else 0) + (8 if has_student else 0) + non_empty_count if probe_score > best_header_probe_score: best_header_probe_score = probe_score header_row_number = probe_row_number raw_header = row_values if has_section and has_student: break if not raw_header: raw_header = [f"Column {index + 1}" for index in range(worksheet.max_column)] header_labels = [] for index, value in enumerate(raw_header): label = value if value else f"Column {index + 1}" if re.match(r"^\{\{.*\}\}$", label): label = f"Column {index + 1}" header_labels.append(label) normalized_headers = [value.lower() for value in header_labels] section_column_indexes = self._find_column_indexes( normalized_headers, ["section", "section title", "section_name", "class section", "block"], ) student_id_column_index = self._find_column_index( normalized_headers, [ "student no", "student no.", "student #", "student number", "student id", "student_id", "school id", "school_id", "id number", "id no", "id no.", ], ) logger.info( "Header analysis class_id=%s header_row=%s section_col_indexes=%s student_id_col_index=%s headers=%s", section_id, header_row_number, section_column_indexes, student_id_column_index, header_labels, ) preview_rows = [] mapped_count = 0 total_match_count = 0 max_preview_rows = 200 sampled_section_values = [] sampled_row_texts = [] for row_number, row in enumerate( worksheet.iter_rows(min_row=header_row_number + 1, values_only=True), start=header_row_number + 1, ): row_cells = [self._normalize_cell(value) for value in row] row_text = " ".join([cell.lower() for cell in row_cells if cell]) row_text_key = self._normalize_for_match(" ".join([cell for cell in row_cells if cell])) if section_column_indexes: row_section_values = [] section_cell_keys = [] matched_section_col_index = None for section_column_index in section_column_indexes: if section_column_index < len(row_cells): section_raw_value = row_cells[section_column_index] section_cell_key = self._normalize_for_match(section_raw_value) if section_raw_value: row_section_values.append({"col": section_column_index, "value": section_raw_value}) if section_cell_key: section_cell_keys.append(section_cell_key) if matched_section_col_index is None and section_title_key in section_cell_key: matched_section_col_index = section_column_index if len(sampled_section_values) < 10 and row_section_values: sampled_section_values.append({"row": row_number, "values": row_section_values}) is_match = matched_section_col_index is not None else: matched_section_col_index = None if len(sampled_row_texts) < 10 and row_text: sampled_row_texts.append({"row": row_number, "value": row_text[:180]}) is_match = bool(section_title_key) and section_title_key in row_text_key if not is_match: continue total_match_count += 1 student_id_value = "" if student_id_column_index is not None and student_id_column_index < len(row_cells): student_id_value = row_cells[student_id_column_index] mapped_person = None if student_id_value: mapped_person = Person.objects.filter( Q(school_id__iexact=student_id_value) | Q(card_number__iexact=student_id_value) ).first() if mapped_person: mapped_count += 1 if len(preview_rows) < max_preview_rows: preview_rows.append( { "row_number": row_number, "cells": row_cells, "student_id": student_id_value, "mapped_person": mapped_person, } ) relevant_headers = [] relevant_column_indexes = [] skip_keywords = ["count", "no."] for idx, header in enumerate(header_labels): header_lower = header.lower() if any(skip_kw in header_lower for skip_kw in skip_keywords): continue if idx < 10: relevant_headers.append(header) relevant_column_indexes.append(idx) if "program" in header_lower: break filtered_preview_rows = [] for row in preview_rows: filtered_cells = [row["cells"][idx] if idx < len(row["cells"]) else "" for idx in relevant_column_indexes] filtered_preview_rows.append( { "row_number": row["row_number"], "cells": filtered_cells, "student_id": row["student_id"], "mapped_person": row["mapped_person"], } ) logger.info( "Class preview result class_id=%s total_match_count=%s mapped_students=%s preview_rows=%s relevant_headers=%s", section_id, total_match_count, mapped_count, len(filtered_preview_rows), relevant_headers, ) if sampled_section_values: logger.info("Sample class column values class_id=%s samples=%s", section_id, sampled_section_values) if sampled_row_texts: logger.info("Sample row texts (fallback match mode) class_id=%s samples=%s", section_id, sampled_row_texts) return relevant_headers, filtered_preview_rows, total_match_count, mapped_count @staticmethod def _find_column_index(headers, keywords): for index, header in enumerate(headers): for keyword in keywords: if keyword in header: return index return None @staticmethod def _find_column_indexes(headers, keywords): indexes = [] for index, header in enumerate(headers): for keyword in keywords: if keyword in header: indexes.append(index) break return indexes def _serialize_preview_rows(self, preview_rows): serialized = [] for row in preview_rows: person_dict = None person_identifier = "" if row.get("mapped_person"): person = row["mapped_person"] person_dict = { "name": person.display_name, "identifier": person.school_id or person.card_number or str(person.pk), } person_identifier = person.school_id or person.card_number or str(person.pk) serialized.append( { "row_number": row.get("row_number"), "cells": row.get("cells", []), "student_id": row.get("student_id", ""), "mapped_person": person_dict, "person_identifier": person_identifier, } ) return serialized @staticmethod def _build_preview_table_data(preview_headers): headers = [ "Select", "Row", *preview_headers, "Student ID", "Mapped Person", ] return { "headers": headers, "rows": [["" for _ in headers]], } @staticmethod def _normalize_cell(value): if value is None: return "" return str(value).strip() @staticmethod def _normalize_for_match(value): text = ClassAdmin._normalize_cell(value).casefold() text = re.sub(r"[^a-z0-9]+", "", text) return text @staticmethod def _load_excel_workbook(file_obj, filename=None): file_bytes = file_obj.read() if hasattr(file_obj, "seek"): file_obj.seek(0) is_xls = False if filename and filename.lower().endswith(".xls") and not filename.lower().endswith((".xlsx", ".xlsm", ".xltx", ".xltm")): is_xls = True elif file_bytes.startswith(b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1"): is_xls = True if is_xls: import xlrd from openpyxl import Workbook xlrd_wb = xlrd.open_workbook(file_contents=file_bytes) wb = Workbook() if wb.sheetnames: wb.remove(wb.active) for sheet_name in xlrd_wb.sheet_names(): xlrd_sheet = xlrd_wb.sheet_by_name(sheet_name) ws = wb.create_sheet(title=sheet_name) for row_idx in range(xlrd_sheet.nrows): row_vals = [] for col_idx in range(xlrd_sheet.ncols): cell = xlrd_sheet.cell(row_idx, col_idx) val = cell.value if cell.ctype == xlrd.XL_CELL_NUMBER: if isinstance(val, float) and val.is_integer(): val = int(val) elif cell.ctype == xlrd.XL_CELL_DATE: try: date_tuple = xlrd.xldate_as_tuple(val, xlrd_wb.datemode) val = f"{date_tuple[0]:04d}-{date_tuple[1]:02d}-{date_tuple[2]:02d}" except Exception as date_err: logger.debug("Unable to convert xlrd date cell (row=%s, col=%s): %s", row_idx, col_idx, date_err) row_vals.append(val) ws.append(row_vals) return wb else: from io import BytesIO from openpyxl import load_workbook return load_workbook(BytesIO(file_bytes), data_only=True, read_only=True) def _extract_title_from_worksheet(self, worksheet): """ Extract class title from the 7th row (1-indexed) of an EduSuite class list, with fallback scanning rows 1 to 10. """ try: val_r7 = self._normalize_cell(worksheet.cell(row=7, column=1).value) if val_r7 and not val_r7.lower().startswith(("faculty:", "schedule:", "date generated", "king's college")): return val_r7 except Exception: pass for row_idx, row in enumerate(worksheet.iter_rows(min_row=1, max_row=min(worksheet.max_row, 10), values_only=True), start=1): for cell_val in row: norm_val = self._normalize_cell(cell_val) if norm_val: lower = norm_val.lower() if lower.startswith(("faculty:", "schedule", "date generated", "king's college", "college of", "summer", "1st sem", "2nd sem")): continue if "-" in norm_val or "[" in norm_val: return norm_val return "" def _extract_schedule_from_worksheet(self, worksheet): """ Extract days and start_time/end_time from dynamic schedule rows in an EduSuite class list. Scans from the row containing 'Schedule' onwards until student headers or blank rows. """ day_keyword_map = { "monday": 0, "mon": 0, "tuesday": 1, "tues": 1, "tue": 1, "wednesday": 2, "wed": 2, "thursday": 3, "thurs": 3, "thur": 3, "thu": 3, "th": 3, "friday": 4, "fri": 4, "saturday": 5, "sat": 5, "sunday": 6, "sun": 6, } def _parse_time(time_str): if not time_str: return None m = re.search(r"(\d{1,2})(?::(\d{2}))?\s*(am|pm|a\.m\.|p\.m\.)?", time_str, re.IGNORECASE) if not m: return None hr = int(m.group(1)) minute = int(m.group(2)) if m.group(2) else 0 meridiem = m.group(3).lower().replace(".", "") if m.group(3) else None if meridiem == "pm" and hr < 12: hr += 12 elif meridiem == "am" and hr == 12: hr = 0 return f"{hr:02d}:{minute:02d}:00" schedule_start_row = None max_scan_row = min(worksheet.max_row, 30) # Locate 'Schedule' marker for row_idx, row in enumerate(worksheet.iter_rows(min_row=1, max_row=max_scan_row, values_only=True), start=1): row_cells = [self._normalize_cell(c) for c in row if c] if any("schedule" in c.lower() for c in row_cells): schedule_start_row = row_idx break if schedule_start_row is None: return [], None, None extracted_day_codes = [] extracted_start_time = None extracted_end_time = None # Iterate schedule rows dynamically for row_idx, row in enumerate(worksheet.iter_rows(min_row=schedule_start_row, max_row=max_scan_row, values_only=True), start=schedule_start_row): row_cells = [self._normalize_cell(c) for c in row if c] if not row_cells: if row_idx > schedule_start_row: break continue joined_text = " ".join(row_cells) joined_lower = joined_text.lower() # Stop if we hit student header row if "student name" in joined_lower or "student number" in joined_lower or "student no" in joined_lower: break # Parse composite day patterns (e.g. MWF, TTH, M-W-F) composite_matches = re.findall(r"\b(?:MWF|TTH|M-W-F|T-TH|MTWTHF)\b", joined_text, re.IGNORECASE) for tok in composite_matches: tok_clean = tok.upper().replace("-", "") if tok_clean == "MWF": for code in [0, 2, 4]: if code not in extracted_day_codes: extracted_day_codes.append(code) elif tok_clean == "TTH": for code in [1, 3]: if code not in extracted_day_codes: extracted_day_codes.append(code) elif tok_clean == "MTWTHF": for code in [0, 1, 2, 3, 4]: if code not in extracted_day_codes: extracted_day_codes.append(code) # Parse individual day words tokens = re.split(r"[\s,/]+", joined_text) for tok in tokens: clean_tok = re.sub(r"[^a-zA-Z]", "", tok).lower() if clean_tok in day_keyword_map: code = day_keyword_map[clean_tok] if code not in extracted_day_codes: extracted_day_codes.append(code) # Parse time occurrences time_matches = list(re.finditer(r"(\d{1,2}(?::\d{2})?\s*(?:am|pm|a\.m\.|p\.m\.)?)", joined_text, re.IGNORECASE)) if time_matches: if len(time_matches) >= 2: st = _parse_time(time_matches[0].group(1)) et = _parse_time(time_matches[1].group(1)) if st and not extracted_start_time: extracted_start_time = st if et and not extracted_end_time: extracted_end_time = et elif len(time_matches) == 1: st = _parse_time(time_matches[0].group(1)) if st and not extracted_start_time: extracted_start_time = st extracted_day_codes.sort() weekdays_qs = Weekday.objects.filter(code__in=extracted_day_codes).order_by("code") days_data = [{"id": w.id, "code": w.code, "name": w.get_code_display()} for w in weekdays_qs] return days_data, extracted_start_time, extracted_end_time def parse_class_list_view(self, request): if not self.has_add_permission(request): return JsonResponse({"success": False, "error": "Permission denied."}, status=403) if request.method != "POST": return JsonResponse({"success": False, "error": "POST request required."}, status=405) excel_file = request.FILES.get("excel_file") if not excel_file: return JsonResponse({"success": False, "error": "Please upload an Excel file."}, status=400) filename = getattr(excel_file, "name", "") try: workbook = self._load_excel_workbook(excel_file, filename=filename) worksheet = workbook.active or workbook.worksheets[0] # 1. Extract Title (Row 7) extracted_title = self._extract_title_from_worksheet(worksheet) # 2. Extract Dynamic Schedule (Days & Time) days_data, start_time, end_time = self._extract_schedule_from_worksheet(worksheet) # 3. Extract & Match Students (reusing _build_class_list_preview) headers, preview_rows, total_students, mapped_count = self._build_class_list_preview(workbook=workbook) students_list = [] for row in preview_rows: mp = row.get("mapped_person") students_list.append({ "id": mp.id if mp else None, "name": mp.display_name if mp else (row.get("cells", [""])[0] or "Unknown"), "school_id": mp.school_id if mp else (row.get("student_id") or ""), "matched": bool(mp), "sex": row.get("cells", ["", ""])[1] if len(row.get("cells", [])) > 1 else "", "program": row.get("cells", ["", "", "", ""])[3] if len(row.get("cells", [])) > 3 else "", "status": row.get("cells", ["", "", "", "", ""])[4] if len(row.get("cells", [])) > 4 else "", }) return JsonResponse({ "success": True, "title": extracted_title, "days": days_data, "day_ids": [d["id"] for d in days_data], "day_names": [d["name"] for d in days_data], "start_time": start_time or "", "end_time": end_time or "", "total_students": total_students, "matched_count": mapped_count, "unmatched_count": total_students - mapped_count, "students": students_list, }) except Exception as exc: logger.exception("Error parsing class list excel: %s", exc) return JsonResponse({"success": False, "error": f"Failed to parse class list: {str(exc)}"}, status=500)