from django.conf import settings from collections import defaultdict from decimal import Decimal, InvalidOperation from io import BytesIO from django.contrib import messages from django.contrib.auth import authenticate, login, logout from django.contrib.auth.decorators import login_required, user_passes_test from django.contrib.auth.models import User, Group from django.db import transaction from django.db.models import Sum from django.http import FileResponse, HttpResponse from django.shortcuts import get_object_or_404, redirect, render from django.utils import timezone from .forms import ( SchoolForm, ParticipantForm, ItemForm, PointRuleForm, AcademicYearForm, UserCreateForm, UserPasswordForm, ) from .models import ( School, Participant, ParticipantItem, Item, GroupEntry, GroupMember, PointRule, Result, SchoolPoint, ParticipantPoint, AcademicYear, SystemSetting, StageMark, ) from .services import recompute_results, publish_item, reset_item def can_reset_published_result(request): if not request.user.is_authenticated: return False if request.user.is_superuser: return True try: return request.user.profile.role in ["ADMIN", "RESULT"] except Exception: return False def group_exists(user, name): return user.groups.filter(name=name).exists() def is_admin(user): return user.is_authenticated and user.is_superuser def is_school_user(user): return user.is_authenticated and group_exists(user, "School User") def is_result_user(user): return user.is_authenticated and group_exists(user, "Result Entry User") def current_school(user): if is_school_user(user) and not is_admin(user): return School.objects.filter(code=user.username).first() return None def require_admin(view): return user_passes_test(is_admin, login_url="/login/")(view) def can_manage_school(user, school): return is_admin(user) or (is_school_user(user) and current_school(user) and current_school(user).id == school.id) def school_locked(school): return bool(school and school.confirmed) # ---------------- Authentication / Home ---------------- def home(request): return render(request, "home.html", { "school_count": School.objects.count(), "participant_count": Participant.objects.count(), "item_count": Item.objects.filter(active=True).count(), "published_count": Result.objects.filter(published=True).count(), }) def login_view(request): if request.user.is_authenticated: return redirect("home") if request.method == "POST": user = authenticate( request, username=request.POST.get("username", "").strip(), password=request.POST.get("password", ""), ) if user: login(request, user) return redirect("home") messages.error(request, "Invalid username or password.") return render(request, "login.html") def logout_view(request): logout(request) return redirect("home") @login_required def change_password(request): if request.method == "POST": old = request.POST.get("old_password", "") new = request.POST.get("new_password", "") confirm = request.POST.get("confirm_password", "") if not request.user.check_password(old): messages.error(request, "Current password is incorrect.") elif len(new) < 6: messages.error(request, "New password must contain at least 6 characters.") elif new != confirm: messages.error(request, "New passwords do not match.") else: request.user.set_password(new) request.user.save() messages.success(request, "Password changed. Please log in again.") logout(request) return redirect("login") return render(request, "change_password.html") @login_required def dashboard(request): return render(request, "dashboard.html") # ---------------- School Management ---------------- @require_admin def schools(request): if request.method == "POST": form = SchoolForm(request.POST) if form.is_valid(): form.save() messages.success(request, "School saved successfully.") return redirect("schools") else: form = SchoolForm() return render(request, "admin/schools.html", { "schools": School.objects.all(), "form": form, }) @require_admin def school_edit(request, pk): school = get_object_or_404(School, pk=pk) form = SchoolForm(request.POST or None, instance=school) if form.is_valid(): form.save() messages.success(request, "School updated successfully.") return redirect("schools") return render(request, "admin/form.html", {"form": form, "title": "Edit School"}) @require_admin def school_delete(request, pk): school = get_object_or_404(School, pk=pk) if request.method == "POST": school.delete() messages.success(request, "School deleted.") return redirect("schools") # ---------------- School Profile ---------------- @login_required def profile(request): school = current_school(request.user) if not school: return redirect("home") form = SchoolForm(request.POST or None, instance=school) form.fields["name"].disabled = True form.fields["code"].disabled = True if form.is_valid() and not school_locked(school): form.save() messages.success(request, "Profile updated.") return redirect("profile") return render(request, "school/profile.html", {"form": form, "school": school}) # ---------------- Participants ---------------- @login_required def participants(request): school = current_school(request.user) selected_id = request.GET.get("school") if is_admin(request.user): selected = School.objects.filter(pk=selected_id).first() if selected_id else None else: selected = school qs = Participant.objects.select_related("school") if selected: qs = qs.filter(school=selected) section_order = {"UP": 1, "HS": 2, "HSS": 3, "LP": 4} participants = list(qs) participants.sort(key=lambda p: (section_order.get(p.section, 9), p.sex, p.name.lower())) return render(request, "participants/list.html", { "participants": participants, "schools": School.objects.all(), "selected_school": selected, "locked": school_locked(selected) if selected else False, }) @login_required def participant_add(request): if is_admin(request.user): school = School.objects.filter(pk=request.GET.get("school") or request.POST.get("school")).first() else: school = current_school(request.user) if not school: return redirect("participants") if school_locked(school) and not is_admin(request.user): messages.error(request, "School data is confirmed. Admin must reset the confirmation.") return redirect("participants") form = ParticipantForm(request.POST or None) if form.is_valid(): p = form.save(commit=False) p.school = school p.save() messages.success(request, "Participant added.") return redirect(f"/participants/?school={school.id}" if is_admin(request.user) else "participants") return render(request, "form.html", {"form": form, "title": "Add Participant", "school": school}) @login_required def participant_edit(request, pk): p = get_object_or_404(Participant.objects.select_related("school"), pk=pk) if not can_manage_school(request.user, p.school): return redirect("participants") if school_locked(p.school) and not is_admin(request.user): messages.error(request, "School data is confirmed.") return redirect("participants") form = ParticipantForm(request.POST or None, instance=p) if form.is_valid(): form.save() messages.success(request, "Participant updated.") return redirect("participants") return render(request, "form.html", {"form": form, "title": "Edit Participant", "school": p.school}) @login_required def participant_delete(request, pk): p = get_object_or_404(Participant.objects.select_related("school"), pk=pk) if request.method == "POST" and can_manage_school(request.user, p.school): if not school_locked(p.school) or is_admin(request.user): p.delete() messages.success(request, "Participant deleted.") return redirect("participants") # ---------------- Participant Items ---------------- @login_required def participant_items(request): school = current_school(request.user) # Admin can select any school. if is_admin(request.user): selected_school = ( School.objects.filter(pk=request.GET.get("school")).first() if request.GET.get("school") else None ) else: selected_school = school # -------------------------------------------------------- # SAVE ITEMS # -------------------------------------------------------- if request.method == "POST": participant_id = request.POST.get("participant_id") participant = Participant.objects.select_related( "school" ).filter(pk=participant_id).first() if not participant: messages.error(request, "Participant not found.") return redirect("participant_items") # Security: user can manage only the permitted school. if not can_manage_school(request.user, participant.school): messages.error(request, "You are not allowed to change this participant.") return redirect("participant_items") # Confirmed school cannot be edited by School User. if ( school_locked(participant.school) and not is_admin(request.user) ): messages.error( request, "School data is confirmed. Editing is locked." ) return redirect("participant_items") selected_ids = [] for field in [ "item1", "item2", "item3", "item4", "item5", ]: value = request.POST.get(field, "").strip() if value and value not in selected_ids: selected_ids.append(value) # Maximum 5 items. if len(selected_ids) > 5: messages.error( request, "Maximum 5 items are allowed for one participant." ) return redirect( f"/participant-items/?school={participant.school_id}" ) # ---------------------------------------------------- # IMPORTANT: # ONLY ITEMS BELONGING TO THE PARTICIPANT'S SECTION # CAN BE SAVED. # ---------------------------------------------------- allowed_items = Item.objects.filter( active=True, section=participant.section ) allowed_ids = { str(item.id) for item in allowed_items } invalid_items = [ item_id for item_id in selected_ids if item_id not in allowed_ids ] if invalid_items: messages.error( request, "One or more selected items do not belong to the participant's section." ) return redirect( f"/participant-items/?school={participant.school_id}" ) # Replace the participant's previous assignments. participant.item_links.all().delete() for item_id in selected_ids: ParticipantItem.objects.create( participant=participant, item_id=int(item_id) ) messages.success( request, f"Items updated successfully for {participant.name}." ) return redirect( f"/participant-items/?school={participant.school_id}" ) # -------------------------------------------------------- # DISPLAY PARTICIPANTS # -------------------------------------------------------- qs = ( Participant.objects .select_related("school") .prefetch_related("item_links__item") ) if selected_school: qs = qs.filter(school=selected_school) participants = list( qs.order_by("section", "sex", "name") ) # Section order. section_order = { "UP": 1, "HS": 2, "HSS": 3, "LP": 4, } participants.sort( key=lambda p: ( section_order.get(p.section, 99), p.sex, p.name.lower() ) ) # -------------------------------------------------------- # BUILD DATA FOR EACH PARTICIPANT # -------------------------------------------------------- rows = [] for participant in participants: section_items = list( Item.objects.filter( active=True, section=participant.section ).order_by("code") ) assigned = list( participant.item_links.select_related( "item" ).order_by("item__code") ) assigned_ids = [ item.item_id for item in assigned[:5] ] rows.append({ "participant": participant, "section_items": section_items, "assigned": assigned[:5], "assigned_ids": assigned_ids, }) return render( request, "participants/items.html", { "rows": rows, "schools": School.objects.all(), "selected_school": selected_school, } ) @login_required def participant_item_edit(request, pk): p = get_object_or_404(Participant.objects.select_related("school"), pk=pk) if not can_manage_school(request.user, p.school): return redirect("participant_items") if school_locked(p.school) and not is_admin(request.user): messages.error(request, "School data is confirmed.") return redirect("participant_items") items = Item.objects.filter(active=True, section=p.section).order_by("code") selected = set(p.item_links.values_list("item_id", flat=True)) if request.method == "POST": chosen = request.POST.getlist("items") if len(chosen) > 5: messages.error(request, "Maximum 5 items are allowed per participant.") else: p.item_links.all().delete() for item_id in chosen: item = get_object_or_404(items, pk=item_id) p.item_links.create(item=item) messages.success(request, "Participant items updated.") return redirect("participant_items") return render(request, "participants/item_edit.html", { "participant": p, "items": items, "selected": selected }) # ---------------- Group Captains ---------------- @login_required def captains(request): school = current_school(request.user) selected = School.objects.filter(pk=request.GET.get("school")).first() if is_admin(request.user) else school group_items = Item.objects.filter(active=True, participation_type="G").order_by("section", "code") qs = GroupEntry.objects.select_related("school", "item", "captain").prefetch_related("members__participant") if selected: qs = qs.filter(school=selected) if request.method == "POST" and selected and (is_admin(request.user) or not school_locked(selected)): item = get_object_or_404(group_items, pk=request.POST.get("item_id")) captain = get_object_or_404(Participant, pk=request.POST.get("captain_id"), school=selected) entry, _ = GroupEntry.objects.get_or_create(school=selected, item=item, defaults={"captain": captain}) if entry.captain_id != captain.id: entry.captain = captain entry.save(update_fields=["captain"]) messages.success(request, "Group captain saved.") return redirect("captains") captains = list(qs) return render(request, "participants/captains.html", { "captains": captains, "group_items": group_items, "participants": Participant.objects.filter(school=selected) if selected else Participant.objects.none(), "schools": School.objects.all(), "selected_school": selected, }) @login_required def group_members(request, entry_id): entry = get_object_or_404(GroupEntry.objects.select_related("school", "item"), pk=entry_id) if not can_manage_school(request.user, entry.school): return redirect("captains") if school_locked(entry.school) and not is_admin(request.user): messages.error(request, "School data is confirmed.") return redirect("captains") members = list(entry.members.values_list("participant_id", flat=True)) if request.method == "POST": selected = [int(x) for x in request.POST.getlist("members") if x.isdigit()] if len(selected) < entry.item.min_participants or len(selected) > entry.item.max_participants: messages.error(request, f"Participants must be between {entry.item.min_participants} and {entry.item.max_participants}.") else: entry.members.all().delete() for pid in selected: entry.members.create(participant_id=pid) messages.success(request, "Group members saved.") return redirect("captains") return render(request, "participants/group_members.html", { "entry": entry, "participants": Participant.objects.filter(school=entry.school).order_by("section", "sex", "name"), "members": set(members), }) # ---------------- Confirmation ---------------- @login_required def confirmation(request): school = current_school(request.user) if not school: return redirect("home") if request.method == "POST" and not school.confirmed: password = request.POST.get("password", "") if request.user.check_password(password): school.confirmed = True school.confirmed_at = timezone.now() school.save(update_fields=["confirmed", "confirmed_at"]) messages.success(request, "School data confirmed and locked.") else: messages.error(request, "Incorrect password.") return render(request, "school/confirmation.html", {"school": school}) @require_admin def confirmation_status(request): if request.method == "POST": action = request.POST.get("action") school = get_object_or_404(School, pk=request.POST.get("school_id")) if action == "reset": school.confirmed = False school.confirmed_at = None school.save(update_fields=["confirmed", "confirmed_at"]) elif action == "confirm": school.confirmed = True school.confirmed_at = timezone.now() school.save(update_fields=["confirmed", "confirmed_at"]) messages.success(request, f"Confirmation updated for {school.name}.") return render(request, "admin/confirmation_status.html", {"schools": School.objects.all()}) # ---------------- Admin settings ---------------- @require_admin def point_distribution(request): items = Item.objects.filter(active=True).order_by("section", "code") if request.method == "POST": for item in items: rule, _ = PointRule.objects.get_or_create( item=item, defaults={"first": 10 if item.participation_type == "G" else 5, "second": 6 if item.participation_type == "G" else 3, "third": 3 if item.participation_type == "G" else 1}, ) rule.first = int(request.POST.get(f"first_{item.id}", rule.first)) rule.second = int(request.POST.get(f"second_{item.id}", rule.second)) rule.third = int(request.POST.get(f"third_{item.id}", rule.third)) rule.save() messages.success(request, "Point distribution updated.") return redirect("point_distribution") for item in items: PointRule.objects.get_or_create( item=item, defaults={"first": 10 if item.participation_type == "G" else 5, "second": 6 if item.participation_type == "G" else 3, "third": 3 if item.participation_type == "G" else 1}, ) return render(request, "admin/points.html", {"items": items}) @require_admin def maximum_participation(request): items = Item.objects.filter(participation_type="G", active=True).order_by("section", "code") if request.method == "POST": item = get_object_or_404(Item, pk=request.POST.get("item_id")) item.min_participants = max(1, int(request.POST.get("min_participants", item.min_participants))) item.max_participants = max(item.min_participants, int(request.POST.get("max_participants", item.max_participants))) item.save() messages.success(request, "Maximum participation updated.") return redirect("maximum_participation") settings = { "individual_per_student": SystemSetting.get_int("individual_per_student", 5), "group_per_student": SystemSetting.get_int("group_per_student", 3), } return render(request, "admin/maximum.html", {"items": items, "settings": settings}) @require_admin def save_general_limits(request): if request.method == "POST": for key in ["individual_per_student", "group_per_student"]: value = request.POST.get(key) if value and value.isdigit(): SystemSetting.objects.update_or_create(key=key, defaults={"value": value}) messages.success(request, "Participation limits saved.") return redirect("maximum_participation") # ============================================================ # ACADEMIC YEAR - ADMIN DATA OPERATIONS # ============================================================ @require_admin def academic_year_items_csv(request): import csv from io import TextIOWrapper uploaded_file = request.FILES.get("items_csv") if not uploaded_file: messages.error( request, "Please select an Items CSV file." ) return redirect("academic_year") if not uploaded_file.name.lower().endswith(".csv"): messages.error( request, "Please select a CSV file." ) return redirect("academic_year") required_columns = { "Code", "Name", "Section", "Sex", "Participation Type", "Minimum Participants", "Maximum Participants", "Active", } errors = [] valid_rows = [] try: text_file = TextIOWrapper( uploaded_file.file, encoding="utf-8-sig" ) reader = csv.DictReader(text_file) if not reader.fieldnames: messages.error( request, "The CSV file has no header row." ) return redirect("academic_year") headers = { str(header).strip() for header in reader.fieldnames if header } missing = required_columns - headers if missing: messages.error( request, "Missing columns: " + ", ".join(sorted(missing)) ) return redirect("academic_year") # ---------------------------------------------------- # VALIDATE EVERY ROW FIRST # ---------------------------------------------------- for row_number, row in enumerate( reader, start=2 ): code = ( row.get("Code", "") or "" ).strip() name = ( row.get("Name", "") or "" ).strip() section = ( row.get("Section", "") or "" ).strip().upper() sex = ( row.get("Sex", "") or "" ).strip().upper() participation_text = ( row.get( "Participation Type", "" ) or "" ).strip().upper() min_text = ( row.get( "Minimum Participants", "1" ) or "1" ).strip() max_text = ( row.get( "Maximum Participants", "1" ) or "1" ).strip() active_text = ( row.get( "Active", "1" ) or "1" ).strip().upper() row_errors = [] # Code if not code: row_errors.append( "Code is missing" ) # Name if not name: row_errors.append( "Name is missing" ) # Section if section not in { "LP", "UP", "HS", "HSS", }: row_errors.append( "Section must be LP, UP, HS or HSS" ) # Sex if sex == "MALE": sex = "M" elif sex == "FEMALE": sex = "F" elif sex == "MIXED": sex = "X" if sex not in { "M", "F", "X", }: row_errors.append( "Sex must be M, F or X" ) # Participation type if participation_text in { "INDIVIDUAL", "I", }: participation_type = "I" elif participation_text in { "GROUP", "G", }: participation_type = "G" else: participation_type = None row_errors.append( "Participation Type must be Individual or Group" ) # Minimum participants try: min_participants = int( min_text ) if min_participants < 1: raise ValueError except (TypeError, ValueError): min_participants = 1 row_errors.append( "Minimum Participants must be a positive number" ) # Maximum participants try: max_participants = int( max_text ) if max_participants < 1: raise ValueError except (TypeError, ValueError): max_participants = 1 row_errors.append( "Maximum Participants must be a positive number" ) # Individual item always has one participant. if participation_type == "I": min_participants = 1 max_participants = 1 if ( min_participants > max_participants ): row_errors.append( "Minimum Participants cannot exceed Maximum Participants" ) active = ( active_text not in { "0", "NO", "FALSE", "N", } ) if row_errors: errors.append( f"Row {row_number}: " + "; ".join(row_errors) ) else: valid_rows.append({ "code": code, "name": name, "section": section, "sex": sex, "participation_type": participation_type, "min_participants": min_participants, "max_participants": max_participants, "active": active, }) # ---------------------------------------------------- # IF ANY ROW HAS ERROR, IMPORT NOTHING # ---------------------------------------------------- if errors: messages.error( request, ( f"Item CSV import cancelled. " f"{len(errors)} row(s) contain errors." ) ) return render( request, "admin/academic_year.html", { "form": AcademicYearForm(), "years": AcademicYear.objects.all(), "participant_count": Participant.objects.count(), "item_csv_errors": errors, } ) # ---------------------------------------------------- # SAVE EVERYTHING IN ONE TRANSACTION # ---------------------------------------------------- with transaction.atomic(): created_count = 0 updated_count = 0 for data in valid_rows: item, created = Item.objects.update_or_create( code=data["code"], defaults={ "name": data["name"], "section": data["section"], "sex": data["sex"], "participation_type": data["participation_type"], "min_participants": data["min_participants"], "max_participants": data["max_participants"], "active": data["active"], } ) if created: created_count += 1 else: updated_count += 1 # Default point rule only if one does not # already exist. if data["participation_type"] == "G": PointRule.objects.get_or_create( item=item, defaults={ "first": 10, "second": 6, "third": 3, } ) else: PointRule.objects.get_or_create( item=item, defaults={ "first": 5, "second": 3, "third": 1, } ) messages.success( request, ( "Item CSV import completed. " f"New items: {created_count}. " f"Updated items: {updated_count}." ) ) except UnicodeDecodeError: messages.error( request, "CSV must be UTF-8 encoded." ) except Exception as exc: messages.error( request, f"Item CSV import failed: {exc}" ) return redirect("academic_year") @require_admin def academic_year(request): if request.method == "POST": action = request.POST.get("action", "") # ==================================================== # 1. DELETE ALL PARTICIPANT-RELATED DATA # ==================================================== if action == "delete_participant_data": confirmation_text = request.POST.get( "confirmation_text", "" ).strip() if confirmation_text != "DELETE PARTICIPANT DATA": messages.error( request, "Please type exactly: DELETE PARTICIPANT DATA" ) return redirect("academic_year") try: with transaction.atomic(): SchoolPoint.objects.all().delete() ParticipantPoint.objects.all().delete() Result.objects.all().delete() StageMark.objects.all().delete() GroupMember.objects.all().delete() GroupEntry.objects.all().delete() ParticipantItem.objects.all().delete() Participant.objects.all().delete() messages.success( request, "All participant-related data deleted. " "Schools, Items and Sections were preserved." ) except Exception as exc: messages.error( request, f"Participant data deletion failed: {exc}" ) return redirect("academic_year") # ==================================================== # 2. BACKUP CURRENT SYSTEM # ==================================================== if action == "backup_system": import os import shutil import subprocess import zipfile from datetime import datetime from django.conf import settings try: db = settings.DATABASES["default"] db_name = db["NAME"] db_user = db["USER"] db_password = db["PASSWORD"] db_host = db.get("HOST", "127.0.0.1") db_port = str(db.get("PORT", "3306")) backup_root = settings.BASE_DIR / "backups" backup_root.mkdir( parents=True, exist_ok=True ) timestamp = datetime.now().strftime( "%Y%m%d_%H%M%S" ) backup_folder = ( backup_root / f"backup_{timestamp}" ) backup_folder.mkdir( parents=True, exist_ok=True ) sql_file = ( backup_folder / f"marthoma_kalolsavam_database_{timestamp}.sql" ) zip_file = ( backup_root / f"Marthoma_Kalolsavam_Backup_{timestamp}.zip" ) mysqldump = shutil.which("mysqldump") possible_paths = [ r"C:\Program Files\MySQL\MySQL Server 8.4\bin\mysqldump.exe", r"C:\Program Files\MySQL\MySQL Server 8.0\bin\mysqldump.exe", r"C:\Program Files\MySQL\MySQL Server 9.0\bin\mysqldump.exe", ] if not mysqldump: for candidate in possible_paths: if os.path.exists(candidate): mysqldump = candidate break if not mysqldump: raise RuntimeError( "mysqldump.exe was not found." ) env = os.environ.copy() if db_password: env["MYSQL_PWD"] = str(db_password) command = [ mysqldump, "--single-transaction", "--routines", "--triggers", "--events", "--host", str(db_host), "--port", db_port, "--user", str(db_user), str(db_name), ] with open( sql_file, "w", encoding="utf-8" ) as output: result = subprocess.run( command, stdout=output, stderr=subprocess.PIPE, text=True, env=env ) if result.returncode != 0: raise RuntimeError( result.stderr.strip() or "mysqldump failed." ) media_source = settings.BASE_DIR / "media" if media_source.exists(): shutil.copytree( media_source, backup_folder / "media", dirs_exist_ok=True ) info_file = ( backup_folder / "BACKUP_INFO.txt" ) info_file.write_text( ( "Marthoma School Kalolsavam Backup\n" f"Date: {datetime.now()}\n" f"Database: {db_name}\n" ), encoding="utf-8" ) with zipfile.ZipFile( zip_file, "w", zipfile.ZIP_DEFLATED ) as archive: for file_path in backup_folder.rglob("*"): if file_path.is_file(): archive.write( file_path, file_path.relative_to( backup_root ) ) return FileResponse( open(zip_file, "rb"), as_attachment=True, filename=zip_file.name ) except Exception as exc: messages.error( request, f"Backup failed: {exc}" ) return redirect("academic_year") # ==================================================== # 3. IMPORT PARTICIPANTS FROM CSV # ==================================================== if action == "import_csv": import csv from io import TextIOWrapper uploaded_file = request.FILES.get( "participant_csv" ) if not uploaded_file: messages.error( request, "Please select a CSV file." ) return redirect("academic_year") required_headers = { "Admission Number", "Participant Name", "Sex", "Age", "Section", "Class", "School Code", } try: text_file = TextIOWrapper( uploaded_file.file, encoding="utf-8-sig" ) reader = csv.DictReader(text_file) if not reader.fieldnames: messages.error( request, "CSV file has no header row." ) return redirect("academic_year") headers = { str(h).strip() for h in reader.fieldnames if h } missing = required_headers - headers if missing: messages.error( request, "Missing CSV columns: " + ", ".join(sorted(missing)) ) return redirect("academic_year") valid_rows = [] errors = [] for row_number, row in enumerate( reader, start=2 ): admission_no = ( row.get("Admission Number", "") or "" ).strip() participant_name = ( row.get("Participant Name", "") or "" ).strip() sex = ( row.get("Sex", "") or "" ).strip().upper() age_text = ( row.get("Age", "") or "" ).strip() section = ( row.get("Section", "") or "" ).strip().upper() class_name = ( row.get("Class", "") or "" ).strip() school_code = ( row.get("School Code", "") or "" ).strip() row_errors = [] if not admission_no: row_errors.append( "Admission Number missing" ) if not participant_name: row_errors.append( "Participant Name missing" ) if sex == "MALE": sex = "M" if sex == "FEMALE": sex = "F" if sex not in {"M", "F"}: row_errors.append( "Sex must be M or F" ) try: age = int(age_text) if age <= 0: raise ValueError except (TypeError, ValueError): age = None row_errors.append( "Age must be a positive number" ) if section not in { "LP", "UP", "HS", "HSS", }: row_errors.append( "Section must be LP, UP, HS or HSS" ) if not class_name: row_errors.append( "Class missing" ) school = School.objects.filter( code=school_code ).first() if not school: row_errors.append( f"School Code '{school_code}' not found" ) elif Participant.objects.filter( school=school, admission_no=admission_no ).exists(): row_errors.append( "Admission Number already exists" ) if row_errors: errors.append( f"Row {row_number}: " + "; ".join(row_errors) ) else: valid_rows.append({ "school": school, "admission_no": admission_no, "name": participant_name, "sex": sex, "age": age, "section": section, "class_name": class_name, }) if errors: messages.error( request, ( f"CSV validation failed. " f"{len(errors)} error row(s). " "No participants were imported." ) ) return render( request, "admin/academic_year.html", { "form": AcademicYearForm(), "years": AcademicYear.objects.all(), "participant_count": Participant.objects.count(), "csv_errors": errors, } ) with transaction.atomic(): for data in valid_rows: Participant.objects.create( school=data["school"], admission_no=data["admission_no"], name=data["name"], sex=data["sex"], age=data["age"], section=data["section"], class_name=data["class_name"], ) messages.success( request, f"{len(valid_rows)} participant(s) imported successfully." ) except UnicodeDecodeError: messages.error( request, "CSV must be UTF-8 encoded." ) except Exception as exc: messages.error( request, f"CSV import failed: {exc}" ) return redirect("academic_year") # ==================================================== # 4. IMPORT BACKUP DATA # ==================================================== if action == "import_backup": import os import shutil import subprocess import tempfile import zipfile import io from django.conf import settings backup_file = request.FILES.get("backup_file") current_password = request.POST.get( "current_password", "" ) # ------------------------------------------------ # Basic validation # ------------------------------------------------ if not backup_file: messages.error( request, "Please select a backup ZIP file." ) return redirect("academic_year") if not backup_file.name.lower().endswith(".zip"): messages.error( request, "Please select a ZIP backup file." ) return redirect("academic_year") # ------------------------------------------------ # Verify current logged-in user's password # ------------------------------------------------ if not request.user.check_password( current_password ): messages.error( request, "Current user password is incorrect. Backup was not imported." ) return redirect("academic_year") temp_dir = tempfile.mkdtemp( prefix="marthoma_backup_import_" ) try: # ------------------------------------------------ # Database configuration # ------------------------------------------------ db = settings.DATABASES["default"] db_name = str(db.get("NAME", "")) db_user = str(db.get("USER", "")) db_password = str(db.get("PASSWORD", "")) db_host = str( db.get("HOST", "127.0.0.1") ) db_port = str( db.get("PORT", "3306") ) if not db_name: raise RuntimeError( "MySQL database name is not configured." ) # ------------------------------------------------ # Find mysql.exe # ------------------------------------------------ mysql_exe = shutil.which("mysql") possible_mysql_paths = [ r"C:\Program Files\MySQL\MySQL Server 8.4\bin\mysql.exe", r"C:\Program Files\MySQL\MySQL Server 8.0\bin\mysql.exe", r"C:\Program Files\MySQL\MySQL Server 9.0\bin\mysql.exe", r"C:\xampp\mysql\bin\mysql.exe", ] if not mysql_exe: for candidate in possible_mysql_paths: if os.path.exists(candidate): mysql_exe = candidate break if not mysql_exe: raise RuntimeError( "mysql.exe was not found. " "Add the MySQL bin folder to PATH." ) # ------------------------------------------------ # Save uploaded ZIP # ------------------------------------------------ zip_path = os.path.join( temp_dir, "backup.zip" ) with open( zip_path, "wb" ) as destination: for chunk in backup_file.chunks(): destination.write(chunk) # ------------------------------------------------ # Extract ZIP safely # ------------------------------------------------ extract_dir = os.path.join( temp_dir, "extracted" ) os.makedirs( extract_dir, exist_ok=True ) with zipfile.ZipFile( zip_path, "r" ) as archive: base_path = os.path.abspath( extract_dir ) for member in archive.infolist(): target_path = os.path.abspath( os.path.join( extract_dir, member.filename ) ) if not target_path.startswith( base_path + os.sep ): raise RuntimeError( "Invalid backup ZIP file." ) archive.extractall( extract_dir ) # ------------------------------------------------ # Find SQL file # ------------------------------------------------ sql_files = [] for root, dirs, files in os.walk( extract_dir ): for filename in files: if filename.lower().endswith(".sql"): sql_files.append( os.path.join( root, filename ) ) if not sql_files: raise RuntimeError( "No .sql database file was found inside the backup ZIP." ) # Prefer a database SQL dump by filename. database_sql = [ file_path for file_path in sql_files if "database" in os.path.basename( file_path ).lower() ] if database_sql: sql_file = database_sql[0] else: sql_file = sql_files[0] # ------------------------------------------------ # Read SQL and normalize encoding # ------------------------------------------------ with open( sql_file, "rb" ) as source: sql_bytes = source.read() if sql_bytes.startswith( b"\xff\xfe" ): sql_text = sql_bytes.decode( "utf-16-le" ) elif sql_bytes.startswith( b"\xfe\xff" ): sql_text = sql_bytes.decode( "utf-16-be" ) else: sql_text = sql_bytes.decode( "utf-8-sig" ) # ------------------------------------------------ # Remove USE database statements pointing to # another database, so the current selected # database remains the target. # ------------------------------------------------ import re sql_text = re.sub( r"(?im)^\s*USE\s+`?[^;`]+`?\s*;\s*$", "", sql_text ) sql_text = re.sub( r"(?im)^\s*CREATE\s+DATABASE.*?;\s*$", "", sql_text ) sql_text = re.sub( r"(?im)^\s*DROP\s+DATABASE.*?;\s*$", "", sql_text ) # ------------------------------------------------ # Write normalized SQL as UTF-8 # ------------------------------------------------ normalized_sql = os.path.join( temp_dir, "restore.sql" ) with open( normalized_sql, "w", encoding="utf-8", newline="" ) as destination: destination.write( sql_text ) # ------------------------------------------------ # MySQL environment # # The database password is not placed on the # command line. # ------------------------------------------------ env = os.environ.copy() if db_password: env["MYSQL_PWD"] = db_password # ------------------------------------------------ # Restore # # --binary-mode=1 allows binary-safe input. # --force is deliberately NOT used, so errors # stop the import. # ------------------------------------------------ command = [ mysql_exe, "--binary-mode=1", "--host", db_host, "--port", db_port, "--user", db_user, db_name, ] with open( normalized_sql, "rb" ) as sql_input: result = subprocess.run( command, stdin=sql_input, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, ) stdout_text = ( result.stdout.decode( "utf-8", errors="replace" ).strip() ) stderr_text = ( result.stderr.decode( "utf-8", errors="replace" ).strip() ) if result.returncode != 0: raise RuntimeError( stderr_text or stdout_text or "MySQL restore failed." ) # ------------------------------------------------ # Success # ------------------------------------------------ messages.success( request, "Backup data imported successfully into the current database." ) except zipfile.BadZipFile: messages.error( request, "The selected file is not a valid ZIP backup." ) except UnicodeDecodeError: messages.error( request, "The SQL file inside the backup could not be decoded." ) except Exception as exc: messages.error( request, f"Backup import failed: {exc}" ) finally: shutil.rmtree( temp_dir, ignore_errors=True ) return redirect("academic_year") # ======================================================== # NORMAL ACADEMIC YEAR PAGE # ======================================================== form = AcademicYearForm(request.POST or None) if form.is_valid(): form.save() messages.success( request, "Academic year saved." ) return redirect("academic_year") return render( request, "admin/academic_year.html", { "form": form, "years": AcademicYear.objects.all(), "participant_count": Participant.objects.count(), } ) @require_admin @login_required def reports_items(request): # -------------------------------------------------------- # ONLY ADMIN AND RESULT ENTRY USER CAN OPEN THIS PAGE # -------------------------------------------------------- if not ( is_admin(request.user) or is_result_user(request.user) ): return redirect("home") # -------------------------------------------------------- # READ ONLY ITEM LIST # -------------------------------------------------------- items = ( Item.objects .filter(active=True) .order_by("section", "code") ) return render( request, "reports/items.html", { "items": items, }, ) def items(request): form = ItemForm(request.POST or None) if form.is_valid(): item = form.save() PointRule.objects.get_or_create( item=item, defaults={"first": 10 if item.participation_type == "G" else 5, "second": 6 if item.participation_type == "G" else 3, "third": 3 if item.participation_type == "G" else 1}, ) messages.success(request, "Item saved.") return redirect("items") return render(request, "admin/items.html", {"items": Item.objects.all(), "form": form}) @require_admin def item_edit(request, pk): item = get_object_or_404(Item, pk=pk) form = ItemForm(request.POST or None, instance=item) if form.is_valid(): form.save() messages.success(request, "Item updated.") return redirect("items") return render(request, "admin/form.html", {"form": form, "title": "Edit Item"}) @require_admin def item_delete(request, pk): item = get_object_or_404(Item, pk=pk) if request.method == "POST": item.delete() messages.success(request, "Item deleted.") return redirect("items") # ---------------- Chest numbers ---------------- @require_admin def distribute_chest(request): if request.method == "POST": if School.objects.filter(confirmed=False).exists(): messages.error(request, "All schools must be confirmed before chest-number distribution.") return redirect("distribute_chest") last = Participant.objects.exclude(chest_no__isnull=True).order_by("-chest_no").values_list("chest_no", flat=True).first() or 99 changed = 0 for p in Participant.objects.filter(chest_no__isnull=True).order_by("school__code", "section", "sex", "name"): last += 1 p.chest_no = last p.save(update_fields=["chest_no"]) changed += 1 messages.success(request, f"Chest numbers assigned to {changed} new participants.") return render(request, "admin/distribute_chest.html") @login_required def report_confirmation(request): if is_admin(request.user): schools = School.objects.all() elif is_school_user(request.user): school = current_school(request.user) schools = School.objects.filter(pk=school.pk) if school else School.objects.none() else: schools = School.objects.all() return render(request, "admin/confirmation_status.html", {"schools": schools}) # ---------------- User Management ---------------- @require_admin def user_management(request): if request.method == "POST": action = request.POST.get("action") if action == "create": form = UserCreateForm(request.POST) if form.is_valid(): data = form.cleaned_data user = User.objects.create_user( username=data["username"], password=data["password"], ) role = data["role"] if role == "admin": user.is_staff = True user.is_superuser = True user.save() else: group_name = "School User" if role == "school" else "Result Entry User" group, _ = Group.objects.get_or_create(name=group_name) user.groups.add(group) messages.success(request, f"{data['username']} created.") return redirect("user_management") elif action == "password": form = UserPasswordForm(request.POST) if form.is_valid(): target = form.cleaned_data["user"] target.set_password(form.cleaned_data["password"]) target.save() messages.success(request, f"Password changed for {target.username}.") return redirect("user_management") elif action == "delete": target = get_object_or_404(User, pk=request.POST.get("user_id")) if target.id != request.user.id: target.delete() messages.success(request, "User deleted.") else: messages.error(request, "You cannot delete your own account.") return redirect("user_management") else: form = UserCreateForm() return render(request, "admin/user_management.html", { "form": form, "users": User.objects.prefetch_related("groups").order_by("username"), }) # ---------------- Public / authenticated Results ---------------- def _result_queryset(request, published_only=True): qs = Result.objects.select_related("item", "participant__school", "group_entry__school") if published_only: qs = qs.filter(published=True) school_id = request.GET.get("school") section = request.GET.get("section") item_id = request.GET.get("item") participant_id = request.GET.get("participant") if school_id: qs = qs.filter( participant__school_id=school_id ) if participant_id or section or item_id else qs.filter( participant__school_id=school_id ) qs = qs | Result.objects.filter(group_entry__school_id=school_id, published=published_only).select_related("item","participant__school","group_entry__school") if published_only else qs | Result.objects.filter(group_entry__school_id=school_id) if section: qs = qs.filter(item__section=section) if item_id: qs = qs.filter(item_id=item_id) if participant_id: qs = qs.filter(participant_id=participant_id) return qs.distinct().order_by("item__section", "item__code", "position", "id") def full_result(request): results = _result_queryset(request, published_only=True).filter(total__gt=0) return render(request, "results/full_result.html", { "results": results, "schools": School.objects.all(), "items": Item.objects.filter(active=True), "participants": Participant.objects.all(), }) def published_results(request): # -------------------------------------------------------- # Show only items having at least one participant assigned. # Group items are also included when they have a GroupEntry. # -------------------------------------------------------- items = ( Item.objects .filter(active=True) .order_by("section", "code") ) visible_items = [] for item in items: # Individual items: # participant must have this item assigned. participant_exists = Participant.objects.filter( item_links__item=item ).exists() # Group items: # at least one group entry must exist. group_exists = GroupEntry.objects.filter( item=item ).exists() if participant_exists or group_exists: visible_items.append(item) status = { item.id: item.results.filter( published=True ).exists() for item in visible_items } return render( request, "results/published.html", { "items": visible_items, "status": status, }, ) def school_championship_point_details(request, school_id): school = get_object_or_404( School, pk=school_id, ) selected_section = ( request.GET.get("section", "") .strip() .upper() ) # -------------------------------------------------------- # Published points belonging to this school # -------------------------------------------------------- point_rows = ( school.point_rows .filter( result__published=True, ) .select_related( "result", "result__item", "result__participant", "result__participant__school", "result__group_entry", "result__group_entry__school", "result__group_entry__captain", ) .order_by( "result__item__section", "result__item__code", "result__position", "result__id", ) ) if selected_section in ["HSS", "HS", "UP", "LP"]: point_rows = point_rows.filter( result__item__section=selected_section ) detail_rows = [] for point_row in point_rows: result = point_row.result # ---------------------------------------------------- # Participant / Group name # ---------------------------------------------------- if result.participant_id: participant_name = result.participant.name elif result.group_entry_id: if result.group_entry.captain: participant_name = ( result.group_entry.captain.name ) else: participant_name = "Group" else: participant_name = "-" detail_rows.append( { "section": result.item.section, "item_code": result.item.code, "item_name": result.item.name, "participant_name": participant_name, "place": result.position, "points": int(point_row.points or 0), } ) total_points = sum( row["points"] for row in detail_rows ) return render( request, "results/school_championship_point_details.html", { "school": school, "selected_section": selected_section, "detail_rows": detail_rows, "total_points": total_points, }, ) def public_school_point_details(request, school_id): school = get_object_or_404( School, pk=school_id, ) selected_section = ( request.GET.get("section", "") .strip() .upper() ) # -------------------------------------------------------- # Published results for this school # -------------------------------------------------------- results = ( Result.objects .filter( published=True, ) .select_related( "item", "participant", "participant__school", "group_entry", "group_entry__school", "group_entry__captain", ) .order_by( "item__section", "item__code", "position", ) ) detail_rows = [] for result in results: result_school = None if result.participant_id: result_school = result.participant.school elif result.group_entry_id: result_school = result.group_entry.school if not result_school: continue if result_school.id != school.id: continue # ---------------------------------------------------- # If HSS / HS / UP / LP was selected # ---------------------------------------------------- if selected_section: if result.item.section.upper() != selected_section: continue if result.participant_id: participant_name = result.participant.name chest_no = result.participant.chest_no elif result.group_entry_id: if result.group_entry.captain: participant_name = result.group_entry.captain.name chest_no = result.group_entry.captain.chest_no else: participant_name = "Group" chest_no = "-" else: participant_name = "-" chest_no = "-" detail_rows.append( { "section": result.item.section, "item_code": result.item.code, "item_name": result.item.name, "position": result.position, "participant": participant_name, "chest_no": chest_no, "mark": result.mark, "points": result.points or 0, "result": result.result, } ) total_points = sum( int(row["points"] or 0) for row in detail_rows ) return render( request, "results/school_championship_point_details.html", { "school": school, "selected_section": selected_section, "detail_rows": detail_rows, "total_points": total_points, }, ) def school_championship(request): # -------------------------------------------------------- # SORT SELECTION # -------------------------------------------------------- selected_sort = request.GET.get( "sort", "total_desc", ) rows = [] for school in School.objects.all(): total = ( school.point_rows .filter(result__published=True) .aggregate(v=Sum("points"))["v"] or 0 ) sections = {} for sec in ["HSS", "HS", "UP", "LP"]: sections[sec] = ( school.point_rows .filter( result__published=True, result__item__section=sec, ) .aggregate(v=Sum("points"))["v"] or 0 ) rows.append({ "school": school, "total": total, "HSS": sections["HSS"], "HS": sections["HS"], "UP": sections["UP"], "LP": sections["LP"], }) # -------------------------------------------------------- # SORT BY SELECTED POINT COLUMN # -------------------------------------------------------- if selected_sort == "total_asc": rows.sort( key=lambda x: x["total"] ) elif selected_sort == "hss_desc": rows.sort( key=lambda x: x["HSS"], reverse=True ) elif selected_sort == "hss_asc": rows.sort( key=lambda x: x["HSS"] ) elif selected_sort == "hs_desc": rows.sort( key=lambda x: x["HS"], reverse=True ) elif selected_sort == "hs_asc": rows.sort( key=lambda x: x["HS"] ) elif selected_sort == "up_desc": rows.sort( key=lambda x: x["UP"], reverse=True ) elif selected_sort == "up_asc": rows.sort( key=lambda x: x["UP"] ) elif selected_sort == "lp_desc": rows.sort( key=lambda x: x["LP"], reverse=True ) elif selected_sort == "lp_asc": rows.sort( key=lambda x: x["LP"] ) else: rows.sort( key=lambda x: x["total"], reverse=True ) return render( request, "results/school_championship.html", { "rows": rows, "selected_sort": selected_sort, }, ) def championship(request): people = Participant.objects.annotate(total_points=Sum("point_rows__points")).order_by("-total_points", "name") champions = {} for sec in ["LP", "UP", "HS", "HSS"]: champions[sec] = { "M": list(people.filter(section=sec, sex="M")[:3]), "F": list(people.filter(section=sec, sex="F")[:3]), } kalaprathibha = people.filter(sex="M", item_links__item__participation_type="I").distinct().first() kalathilakam = people.filter(sex="F", item_links__item__participation_type="I").distinct().first() return render(request, "results/championship.html", { "champions": champions, "kalaprathibha": kalaprathibha, "kalathilakam": kalathilakam, }) # ---------------- Result Entry ---------------- @login_required def result_entry(request): if not (is_admin(request.user) or is_result_user(request.user)): return redirect("home") code = request.GET.get("item_code", "").strip() item = Item.objects.filter(code=code, active=True).first() if code else None results = [] if item: if item.participation_type == "I": competitors = Participant.objects.filter( item_links__item=item ).distinct().order_by("chest_no", "name") for p in competitors: r, _ = Result.objects.get_or_create(item=item, participant=p) results.append(r) else: groups = GroupEntry.objects.filter( item=item ).select_related("school", "captain").prefetch_related("members") for entry in groups: r, _ = Result.objects.get_or_create(item=item, group_entry=entry) results.append(r) results = list(Result.objects.filter(item=item).select_related("participant__school", "group_entry__school").order_by("position", "id")) if request.method == "POST" and item: if Result.objects.filter(item=item, confirmed=True).exists() and request.POST.get("action") in {"save", "confirm"}: messages.error(request, "This result is confirmed. Use Result Reset first.") return redirect(f"/results/entry/?item_code={item.code}") for r in results: pid = r.participant_id prefix = f"row_{r.id}_" try: r.mark1 = Decimal(request.POST.get(prefix+"m1", "0") or "0") r.mark2 = Decimal(request.POST.get(prefix+"m2", "0") or "0") r.mark3 = Decimal(request.POST.get(prefix+"m3", "0") or "0") except InvalidOperation: r.mark1 = r.mark2 = r.mark3 = Decimal("0") r.total = r.mark1 + r.mark2 + r.mark3 r.save() recompute_results(item) if request.POST.get("action") == "confirm": Result.objects.filter(item=item).update(confirmed=True) messages.success(request, "Result confirmed.") else: messages.success(request, "Result saved.") return redirect(f"/results/entry/?item_code={item.code}") return render(request, "results/entry.html", { "item": item, "results": results, "code": code, }) @login_required def publish_result(request): if not (is_admin(request.user) or is_result_user(request.user)): return redirect("home") if request.method == "POST": item = get_object_or_404(Item, pk=request.POST.get("item_id")) if not Result.objects.filter(item=item, confirmed=True).exists(): messages.error(request, "Confirm the result before publishing.") else: publish_item(item) messages.success(request, "Result published and points distributed.") return redirect("result_entry") @login_required def result_reset(request): if not is_admin(request.user): return redirect("home") if request.method == "POST": item = get_object_or_404(Item, pk=request.POST.get("item_id")) reset_item(item) messages.success(request, f"{item.code} reset.") return render(request, "results/reset.html", {"items": Item.objects.filter(active=True)}) # ---------------- Stage Report ---------------- @login_required def stage_report(request): if not (is_admin(request.user) or is_result_user(request.user)): return redirect("home") items = Item.objects.filter(active=True).order_by("section", "code") item = items.filter(pk=request.GET.get("item_id")).first() if request.GET.get("item_id") else None competitors = [] if item: if item.participation_type == "I": competitors = Participant.objects.filter(item_links__item=item).select_related("school").order_by("chest_no", "name") else: competitors = GroupEntry.objects.filter(item=item).select_related("school", "captain").order_by("school__code") return render(request, "reports/stage.html", {"items": items, "item": item, "competitors": competitors}) # ---------------- PDF ---------------- def pdf_response(filename, build_fn): buf = BytesIO() build_fn(buf) buf.seek(0) return FileResponse(buf, as_attachment=True, filename=filename) @login_required def participants_pdf(request): if not (is_admin(request.user) or is_school_user(request.user) or is_result_user(request.user)): return redirect("home") school = current_school(request.user) qs = Participant.objects.select_related("school").prefetch_related("item_links__item").order_by("school__code","section","sex","name") if school: qs = qs.filter(school=school) def build(buf): from reportlab.lib import colors from reportlab.lib.pagesizes import A4, landscape from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer from reportlab.lib.styles import getSampleStyleSheet doc = SimpleDocTemplate(buf, pagesize=landscape(A4), rightMargin=18,leftMargin=18,topMargin=18,bottomMargin=18) styles = getSampleStyleSheet() story = [Paragraph("Marthoma School Kalolsavam - Participants", styles["Title"]), Spacer(1,8)] data = [["Chest","Admission","Participant","School","Section","Class","Age","Item1","Item2","Item3","Item4","Item5"]] for p in qs: names = [x.item.name for x in p.item_links.all()[:5]] names += [""] * (5-len(names)) data.append([p.chest_no or "", p.admission_no, p.name, p.school.name, p.section, p.class_name, p.age, *names]) table = Table(data, repeatRows=1) table.setStyle(TableStyle([ ("GRID",(0,0),(-1,-1),0.35,colors.grey), ("BACKGROUND",(0,0),(-1,0),colors.HexColor("#dce6f1")), ("FONTNAME",(0,0),(-1,0),"Helvetica-Bold"), ])) story.append(table) doc.build(story) return pdf_response("participants.pdf", build) @login_required def results_pdf(request): results = ( _result_queryset( request, published_only=True, ) .filter(total__gt=0) .order_by( "item__section", "item__code", "position", "id", ) ) def build(buf): from reportlab.lib import colors from reportlab.lib.pagesizes import A4 from reportlab.platypus import ( SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer, ) from reportlab.lib.styles import getSampleStyleSheet from reportlab.lib.enums import TA_LEFT doc = SimpleDocTemplate( buf, pagesize=A4, rightMargin=28, leftMargin=28, topMargin=28, bottomMargin=28, ) styles = getSampleStyleSheet() title_style = styles["Title"] section_style = styles["Heading2"] item_style = styles["Heading3"] body_style = styles["BodyText"] story = [] story.append( Paragraph( "Marthoma School Kalolsavam - Results", title_style, ) ) story.append( Spacer(1, 12) ) current_section = None current_item = None item_results = [] def flush_item(): nonlocal item_results if not item_results: return data = [ [ Paragraph("Place", body_style), Paragraph("Chest Name / Group", body_style), Paragraph("School", body_style), ] ] for index, r in enumerate(item_results, start=1): if index > 3: break place_text = str( getattr( r, "place", "" ) ) if r.participant_id: chest = ( r.participant.chest_no or "" ) name_group = ( f"{chest} - " f"{r.competitor_name}" if chest else r.competitor_name ) else: name_group = ( r.competitor_name or "" ) school_name = "" if r.competitor_school: school_name = ( r.competitor_school.name ) data.append( [ Paragraph( place_text, body_style, ), Paragraph( name_group, body_style, ), Paragraph( school_name, body_style, ), ] ) if len(data) > 1: table = Table( data, colWidths=[ 55, 285, 160, ], repeatRows=1, ) table.setStyle( TableStyle( [ ( "GRID", (0, 0), (-1, -1), 0.5, colors.grey, ), ( "BACKGROUND", (0, 0), (-1, 0), colors.lightgrey, ), ( "VALIGN", (0, 0), (-1, -1), "MIDDLE", ), ( "LEFTPADDING", (0, 0), (-1, -1), 6, ), ( "RIGHTPADDING", (0, 0), (-1, -1), 6, ), ( "TOPPADDING", (0, 0), (-1, -1), 5, ), ( "BOTTOMPADDING", (0, 0), (-1, -1), 5, ), ] ) ) story.append(table) story.append( Spacer(1, 12) ) item_results = [] for r in results: section_name = ( r.item.section ) item_key = r.item.id if section_name != current_section: flush_item() story.append( Paragraph( str(section_name), section_style, ) ) story.append( Spacer(1, 4) ) current_section = section_name current_item = None if item_key != current_item: flush_item() story.append( Paragraph( ( f"{r.item.code} - " f"{r.item.name}" ), item_style, ) ) story.append( Spacer(1, 4) ) current_item = item_key place_value = getattr( r, "place", None, ) if place_value in [1, 2, 3]: item_results.append(r) flush_item() doc.build(story) return pdf_response( "results.pdf", build, ) @login_required def stage_pdf(request): if not (is_admin(request.user) or is_result_user(request.user)): return redirect("home") item = get_object_or_404(Item, pk=request.GET.get("item_id")) if item.participation_type == "I": competitors = list(Participant.objects.filter(item_links__item=item).select_related("school").order_by("chest_no","name")) else: competitors = list(GroupEntry.objects.filter(item=item).select_related("school","captain").order_by("school__code")) def build(buf): from reportlab.lib import colors from reportlab.lib.pagesizes import A4 from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer, PageBreak from reportlab.lib.styles import getSampleStyleSheet doc = SimpleDocTemplate(buf, pagesize=A4, rightMargin=18,leftMargin=18,topMargin=18,bottomMargin=18) styles = getSampleStyleSheet() story = [] rows_per_judge_page = 15 # Three judge copies on three pages. for judge in range(1,4): story += [ Paragraph("MARTHOMA SCHOOL KALOLSAVAM", styles["Title"]), Paragraph(f"Item Code: {item.code}     Item: {item.name}", styles["Normal"]), Paragraph(f"Judge {judge} Name: ________________________________", styles["Normal"]), Spacer(1,8), ] data = [["Sl No","Code No","Mark","Remark"]] for i in range(rows_per_judge_page): code = competitors[i].chest_no if i < len(competitors) and item.participation_type=="I" else ( competitors[i].captain.chest_no if i < len(competitors) else "" ) data.append([str(i+1), str(code or ""), "", ""]) table = Table(data, colWidths=[45,80,65,290], rowHeights=[24]+[28]*rows_per_judge_page) table.setStyle(TableStyle([ ("GRID",(0,0),(-1,-1),0.6,colors.black), ("BACKGROUND",(0,0),(-1,0),colors.HexColor("#e8eef7")), ("FONTNAME",(0,0),(-1,0),"Helvetica-Bold"), ("VALIGN",(0,0),(-1,-1),"MIDDLE"), ])) story += [table, Spacer(1,12), Paragraph("Signature of the Judge: ________________________________", styles["Normal"])] if judge != 3: story.append(PageBreak()) story.append(PageBreak()) story += [ Paragraph("STAGE SUMMARY SHEET", styles["Title"]), Paragraph(f"Item Code: {item.code}     Item: {item.name}", styles["Normal"]), Spacer(1,8), ] data = [["Code No","Chest No","Judge1","Judge2","Judge3","Total","Place"]] for i, competitor in enumerate(competitors[:30], start=1): chest = competitor.chest_no if item.participation_type=="I" else competitor.captain.chest_no data.append([i, chest or "", "", "", "", "", ""]) table = Table(data, colWidths=[55,65,70,70,70,70,55]) table.setStyle(TableStyle([ ("GRID",(0,0),(-1,-1),0.6,colors.black), ("BACKGROUND",(0,0),(-1,0),colors.HexColor("#e8eef7")), ("FONTNAME",(0,0),(-1,0),"Helvetica-Bold"), ])) story += [table, Spacer(1,12), Paragraph("Signature of Stage Manager: ________________________________", styles["Normal"])] doc.build(story) return pdf_response("stage_report.pdf", build) # ---------------- Result points detail ---------------- @login_required def point_detail(request, result_id): result = get_object_or_404(Result.objects.select_related("item","participant__school","group_entry__school"), pk=result_id) points = SchoolPoint.objects.filter(result=result) participant_points = ParticipantPoint.objects.filter(result=result) return render(request, "results/point_detail.html", { "result": result, "school_points": points, "participant_points": participant_points, }) @login_required def report_participant_items(request): """ Read-only Participant Item Report. Shows: Chest No Admission No Participant Name Section Item 1 Item 2 Item 3 Item 4 Item 5 No Edit/Delete functions are provided here. """ school = current_school(request.user) # Admin may select a school. if is_admin(request.user): selected_school = ( School.objects.filter( pk=request.GET.get("school") ).first() if request.GET.get("school") else None ) else: selected_school = school qs = ( Participant.objects .select_related("school") .prefetch_related("item_links__item") ) if selected_school: qs = qs.filter(school=selected_school) participants = list( qs.order_by( "section", "sex", "name" ) ) # Exact section order required for the report. section_order = { "LP": 1, "UP": 2, "HS": 3, "HSS": 4, } participants.sort( key=lambda p: ( section_order.get(p.section, 99), p.sex, p.name.lower() ) ) report_rows = [] for participant in participants: assigned_items = list( participant.item_links .select_related("item") .order_by("item__code") )[:5] item_names = [ x.item.name for x in assigned_items ] while len(item_names) < 5: item_names.append("") report_rows.append({ "participant": participant, "item1": item_names[0], "item2": item_names[1], "item3": item_names[2], "item4": item_names[3], "item5": item_names[4], }) return render( request, "reports/participant_items.html", { "report_rows": report_rows, "schools": School.objects.all(), "selected_school": selected_school, } ) @login_required def report_participant_items_pdf(request): if not ( is_admin(request.user) or is_school_user(request.user) or is_result_user(request.user) ): return redirect("home") school = current_school(request.user) if is_admin(request.user): selected_school = ( School.objects.filter( pk=request.GET.get("school") ).first() if request.GET.get("school") else None ) else: selected_school = school qs = ( Participant.objects .select_related("school") .prefetch_related("item_links__item") ) if selected_school: qs = qs.filter(school=selected_school) participants = list( qs.order_by( "section", "sex", "name" ) ) section_order = { "LP": 1, "UP": 2, "HS": 3, "HSS": 4, } participants.sort( key=lambda p: ( section_order.get(p.section, 99), p.sex, p.name.lower() ) ) def build_pdf(buffer): from reportlab.lib import colors from reportlab.lib.pagesizes import A4, landscape from reportlab.lib.styles import getSampleStyleSheet from reportlab.platypus import ( SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer, ) document = SimpleDocTemplate( buffer, pagesize=landscape(A4), rightMargin=18, leftMargin=18, topMargin=18, bottomMargin=18, ) styles = getSampleStyleSheet() story = [] story.append( Paragraph( "MARTHOMA SCHOOL KALOLSAVAM", styles["Title"] ) ) story.append( Paragraph( "Participant Item Report", styles["Heading2"] ) ) if selected_school: story.append( Paragraph( f"School: {selected_school.name}", styles["Normal"] ) ) story.append(Spacer(1, 10)) data = [[ "Chest No", "Admission No", "Participant Name", "Section", "Item 1", "Item 2", "Item 3", "Item 4", "Item 5", ]] for participant in participants: assigned = list( participant.item_links .select_related("item") .order_by("item__code") )[:5] names = [ x.item.name for x in assigned ] while len(names) < 5: names.append("") data.append([ participant.chest_no or "", participant.admission_no, participant.name, participant.section, names[0], names[1], names[2], names[3], names[4], ]) table = Table( data, repeatRows=1 ) table.setStyle( TableStyle([ ( "GRID", (0, 0), (-1, -1), 0.4, colors.grey ), ( "BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#dce6f1") ), ( "FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold" ), ( "VALIGN", (0, 0), (-1, -1), "MIDDLE" ), ( "FONTSIZE", (0, 0), (-1, -1), 8 ), ]) ) story.append(table) document.build(story) return pdf_response( "participant_item_report.pdf", build_pdf )