diff --git a/.dockerignore b/.dockerignore index 57eae8c..dbdf8bd 100644 --- a/.dockerignore +++ b/.dockerignore @@ -4,3 +4,6 @@ __pycache__ migrations db.sqlite3 +media +venv +.venv diff --git a/home/admin.py b/home/admin.py index e002753..030a985 100644 --- a/home/admin.py +++ b/home/admin.py @@ -5,15 +5,21 @@ """ import csv +import io from datetime import timedelta from django.contrib import admin +from django.contrib.auth import get_user_model from django.contrib import messages from django.contrib.admin.helpers import ACTION_CHECKBOX_NAME from django.core.exceptions import ValidationError +from django.db import IntegrityError, transaction +from django.db.models import Q +from django.db.models.functions import Lower from django.http import HttpRequest, HttpResponse from django.shortcuts import render from django.urls import path +from django.db.models.functions import Lower from import_export.admin import ImportExportMixin, ImportExportModelAdmin from home.models import ( @@ -58,6 +64,8 @@ from .utils.month import fill_periods, map_periods_to_long_rebate from .utils.rebate_bills_saver import fix_all_bills, save_long_bill +User = get_user_model() + # Customising the heading and title of the admin page admin.site.site_header = "Dining Website Admin Page" admin.site.site_title = "Admin Page" @@ -311,7 +319,276 @@ class about_Admin(ImportExportMixin, admin.ModelAdmin): }, ), ) - actions = ["export_as_csv", "generate_table", "disable_allocation"] + actions = [ + "export_as_csv", + "generate_table", + "disable_allocation", + "create_user_accounts", + ] + + @admin.action(description="Create missing user accounts") + def create_user_accounts(self, request, queryset): + created_count = 0 + existing_count = 0 + failures = [] + + for student in queryset: + email = student.email.strip() + if not email: + failures.append(f"Student #{student.pk} has no email address.") + continue + + if User.objects.filter( + Q(email__iexact=email) | Q(username__iexact=email) + ).exists(): + existing_count += 1 + continue + + try: + user = User(username=email, email=email) + user.set_unusable_password() + user.save() + created_count += 1 + except IntegrityError: + failures.append( + f"Could not create an account for {email}; the username may already exist." + ) + + if created_count: + self.message_user( + request, + f"Created {created_count} user account(s).", + messages.SUCCESS, + ) + if existing_count: + self.message_user( + request, + f"Skipped {existing_count} student(s) with existing accounts.", + messages.INFO, + ) + if failures: + self.message_user( + request, + "Could not create accounts: " + " ".join(failures), + messages.ERROR, + ) + + def get_urls(self): + urls = super().get_urls() + custom_urls = [ + path( + "import-csv/", + self.admin_site.admin_view(self.import_csv), + name="home_student_import_csv", + ), + ] + return custom_urls + urls + + def import_csv(self, request): + failures = [] + imported_count = 0 + + if request.method == "POST": + hostel = request.POST.get("hostel", "").strip() + uploaded_file = request.FILES.get("csv_file") + + if not hostel: + failures.append({"row": "-", "error": "Hostel name is required."}) + elif len(hostel) > Student._meta.get_field("hostel").max_length: + failures.append({"row": "-", "error": "Hostel name is too long."}) + + if uploaded_file is None: + failures.append({"row": "-", "error": "Please select a CSV file."}) + + if not failures: + try: + text_stream = io.TextIOWrapper( + uploaded_file.file, encoding="utf-8-sig", newline="" + ) + sample = text_stream.read(4096) + text_stream.seek(0) + try: + dialect = csv.Sniffer().sniff(sample, delimiters=",;\t") + except csv.Error: + dialect = csv.excel + + csv_reader = csv.reader(text_stream, dialect) + header = next(csv_reader, None) + if not header: + raise ValueError("The uploaded file is empty.") + + def normalize(value): + return "".join( + character.lower() for character in value if character.isalnum() + ) + + header_aliases = { + "room_no": {"roomno", "roomnumber"}, + "roll_no": {"rollno", "rollnumber"}, + "name": {"name"}, + "degree": {"course", "degree"}, + "department": {"department", "dept"}, + "email": { + "instituteemailid", + "instituteemail", + "emailid", + "email", + }, + } + normalized_header = [normalize(value) for value in header] + column_indexes = {} + for field, aliases in header_aliases.items(): + matching_indexes = [ + index + for index, value in enumerate(normalized_header) + if value in aliases + ] + if not matching_indexes: + raise ValueError( + f"Missing required column: {field.replace('_', ' ')}." + ) + column_indexes[field] = matching_indexes[0] + + max_column_index = max(column_indexes.values()) + pending_rows = [] + uploaded_emails = set() + uploaded_roll_numbers = set() + + for row_number, row in enumerate(csv_reader, start=2): + if not any(value.strip() for value in row): + continue + pending_rows.append((row_number, row)) + if len(row) > column_indexes["email"]: + email_value = row[column_indexes["email"]].strip().lower() + if email_value: + uploaded_emails.add(email_value) + if len(row) > column_indexes["roll_no"]: + roll_value = row[column_indexes["roll_no"]].strip() + if roll_value: + uploaded_roll_numbers.add(roll_value) + + existing_emails = set( + Student.objects.annotate(email_lower=Lower("email")) + .filter(email_lower__in=list(uploaded_emails)) + .values_list("email_lower", flat=True) + ) + existing_roll_numbers = set( + Student.objects.filter(roll_no__in=list(uploaded_roll_numbers)) + .values_list("roll_no", flat=True) + ) + existing_user_emails = set( + User.objects.annotate(email_lower=Lower("email")) + .filter(email_lower__in=list(uploaded_emails)) + .values_list("email_lower", flat=True) + ) + existing_usernames = set( + User.objects.annotate(username_lower=Lower("username")) + .filter(username_lower__in=list(uploaded_emails)) + .values_list("username_lower", flat=True) + ) + + imported_emails = set() + imported_roll_numbers = set() + students_to_create = [] + for row_number, row in pending_rows: + try: + if len(row) <= max_column_index: + raise ValueError( + "The row does not contain all required columns." + ) + + values = { + field: row[index].strip() + for field, index in column_indexes.items() + } + missing_fields = [ + field.replace("_", " ") + for field, value in values.items() + if not value + ] + if missing_fields: + raise ValueError( + "Missing value for: " + ", ".join(missing_fields) + "." + ) + if values["email"].lower() in imported_emails: + raise ValueError("Duplicate email in the uploaded file.") + if values["roll_no"] in imported_roll_numbers: + raise ValueError("Duplicate roll number in the uploaded file.") + if values["email"].lower() in existing_emails: + raise ValueError("A student with this email already exists.") + if values["roll_no"] in existing_roll_numbers: + raise ValueError("A student with this roll number already exists.") + if ( + values["email"].lower() in existing_usernames + and values["email"].lower() not in existing_user_emails + ): + raise ValueError("A user with this username already exists.") + + student = Student( + hostel=hostel, + room_no=values["room_no"], + roll_no=values["roll_no"], + name=values["name"], + degree=values["degree"], + department=values["department"], + email=values["email"], + ) + student.full_clean() + students_to_create.append(student) + imported_emails.add(values["email"].lower()) + imported_roll_numbers.add(values["roll_no"]) + except (ValidationError, ValueError, IndexError) as error: + if isinstance(error, ValidationError): + error_message = "; ".join(error.messages) + else: + error_message = str(error) + failures.append({"row": row_number, "error": error_message}) + + if students_to_create: + with transaction.atomic(): + Student.objects.bulk_create(students_to_create, batch_size=500) + users_to_create = [] + for student in students_to_create: + if student.email.lower() not in existing_user_emails: + user = User( + username=student.email, + email=student.email, + ) + user.set_unusable_password() + users_to_create.append(user) + if users_to_create: + User.objects.bulk_create(users_to_create, batch_size=500) + + semester = Semester.objects.filter().last() + StudentBills.objects.bulk_create( + [StudentBills(email=student, semester=semester) for student in students_to_create], + batch_size=500, + ) + imported_count = len(students_to_create) + except (UnicodeDecodeError, ValueError, csv.Error) as error: + failures.append({"row": "-", "error": str(error)}) + + if imported_count: + self.message_user( + request, + f"Successfully imported {imported_count} student(s).", + messages.SUCCESS, + ) + if failures: + self.message_user( + request, + f"{len(failures)} row(s) could not be imported.", + messages.WARNING, + ) + + context = { + **self.admin_site.each_context(request), + "opts": self.model._meta, + "title": "Import students from CSV", + "failures": failures, + "imported_count": imported_count, + } + return render(request, "admin/home/student/import_csv.html", context) def get_urls(self): urls = super().get_urls() diff --git a/home/tests.py b/home/tests.py index 51014e9..d8cfc0a 100644 --- a/home/tests.py +++ b/home/tests.py @@ -35,5 +35,135 @@ def test_imports_valid_rows_and_reports_invalid_rows(self): student = Student.objects.get() self.assertEqual(student.hostel, "AG") self.assertEqual(student.roll_no, "260003011") + imported_user = get_user_model().objects.get(email=student.email) + self.assertEqual(imported_user.username, student.email) + self.assertTrue(imported_user.has_usable_password() is False) self.assertContains(response, "Row") self.assertContains(response, "Missing value for: roll no.") + + def test_rejects_hostel_name_longer_than_five_characters(self): + csv_content = ( + "Room No,Roll No,Name,Course,Department,Institute Email ID\n" + "201,260003012,JANE DOE,B.Tech,ME,me260003012@iiti.ac.in\n" + ) + upload = SimpleUploadedFile( + "students.csv", csv_content.encode("utf-8"), content_type="text/csv" + ) + + response = self.client.post( + reverse("admin:home_student_import_csv"), + {"hostel": "ABCDE1", "csv_file": upload}, + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(Student.objects.count(), 0) + self.assertContains(response, "Hostel name is too long") + + def test_imports_tab_separated_file_with_bom(self): + tsv_content = ( + "\ufeffRoom No\tRoll No\tName\tCourse\tDepartment\tInstitute Email ID\n" + "203\t260003013\tTAB STUDENT\tB.Tech\tME\tme260003013@iiti.ac.in\n" + ) + upload = SimpleUploadedFile( + "students.csv", tsv_content.encode("utf-8"), content_type="text/csv" + ) + + response = self.client.post( + reverse("admin:home_student_import_csv"), + {"hostel": "AG", "csv_file": upload}, + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(Student.objects.count(), 1) + self.assertTrue( + Student.objects.filter(email="me260003013@iiti.ac.in").exists() + ) + + def test_reports_missing_required_header(self): + csv_content = ( + "Room No,Name,Course,Department,Institute Email ID\n" + "201,NO ROLL,B.Tech,ME,no-roll@example.com\n" + ) + upload = SimpleUploadedFile( + "students.csv", csv_content.encode("utf-8"), content_type="text/csv" + ) + + response = self.client.post( + reverse("admin:home_student_import_csv"), + {"hostel": "AG", "csv_file": upload}, + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(Student.objects.count(), 0) + self.assertContains(response, "Missing required column: roll no.") + + def test_rejects_duplicates_within_file(self): + csv_content = ( + "Room No,Roll No,Name,Course,Department,Institute Email ID\n" + "201,260003014,ONE,B.Tech,ME,dup@example.com\n" + "202,260003015,TWO,B.Tech,ME,dup@example.com\n" + ) + upload = SimpleUploadedFile( + "students.csv", csv_content.encode("utf-8"), content_type="text/csv" + ) + + response = self.client.post( + reverse("admin:home_student_import_csv"), + {"hostel": "AG", "csv_file": upload}, + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(Student.objects.count(), 1) + self.assertContains(response, "Duplicate email in the uploaded file.") + + def test_rejects_existing_email_case_insensitively(self): + Student.objects.create( + hostel="AG", + room_no="200", + roll_no="260003099", + name="Existing", + degree="B.Tech", + department="ME", + email="Case@Test.com", + ) + + csv_content = ( + "Room No,Roll No,Name,Course,Department,Institute Email ID\n" + "201,260003016,NEW,B.Tech,ME,case@test.com\n" + ) + upload = SimpleUploadedFile( + "students.csv", csv_content.encode("utf-8"), content_type="text/csv" + ) + + response = self.client.post( + reverse("admin:home_student_import_csv"), + {"hostel": "AG", "csv_file": upload}, + ) + + self.assertEqual(response.status_code, 200) + self.assertEqual(Student.objects.count(), 1) + self.assertContains(response, "A student with this email already exists.") + + def test_admin_action_creates_missing_user_account(self): + student = Student.objects.create( + hostel="AG", + room_no="204", + roll_no="260003017", + name="Missing Account", + degree="B.Tech", + department="ME", + email="missing.account@iiti.ac.in", + ) + + response = self.client.post( + reverse("admin:home_student_changelist"), + { + "action": "create_user_accounts", + "_selected_action": [student.pk], + }, + ) + + self.assertEqual(response.status_code, 302) + user = get_user_model().objects.get(email=student.email) + self.assertEqual(user.username, student.email) + self.assertFalse(user.has_usable_password()) diff --git a/nginx/default.conf b/nginx/default.conf index 7001f5c..52495e3 100644 --- a/nginx/default.conf +++ b/nginx/default.conf @@ -11,8 +11,14 @@ server{ ssl_certificate_key /etc/letsencrypt/live/diningfee.iiti.ac.in/privkey.pem; client_max_body_size 20M; + client_body_timeout 600s; + send_timeout 600s; + proxy_set_header HOST $host; location /{ proxy_pass http://django; + proxy_connect_timeout 60s; + proxy_send_timeout 600s; + proxy_read_timeout 600s; } }