import csv import io import json import os from datetime import datetime, time as dt_time from django.contrib.contenttypes.models import ContentType from django.db import transaction from rest_framework import status from api.models import ( EsDepartment, EsExamActivity, EsExamPeriod, EsExamPeriodUnavailability, EsExamRequirement, EsExamRequirementStudent, EsInvigilator, EsInvigilatorRole, EsLocation, EsPos, EsSessionStart, EsSessionStartDay, EsSetting, EsStaff, EsStudent, EsStudentGroup, EsStudentStudentGroup, EsSuitability, EsTag, EsTagRelation, EsZone, ) from api.models.relations import ( EsExamActivityInvigilatorSuitability, EsExamActivityLocationSuitability, EsExamRequirementInvigilatorSuitability, EsExamRequirementLocationSuitability, EsInvigilatorInvigilatorRole, EsInvigilatorSuitability, EsLocationSuitability, ) from api.services.student_groups import replace_student_group_rows from api.translation import __ from api.utils import fetch_code_id_map, get_exception_detail, log_critical_error from api.views.admin.base import AdminApiBase from openpyxl import load_workbook import xlrd IMPORT_SHEET_ORDER = [ "es_zone", "es_department", "es_tag", "es_suitability", "es_pos", "es_student_group", "es_invigilator_role", "es_staff", "es_student", "es_location", "es_invigilator", "es_exam_period", "es_session_start", "es_exam_requirement", "es_suitability_invigilator", "es_suitability_location", "es_invigilator_roles", "es_student_student_group", "es_exam_period_unavailability", "es_exam_requirement_student", "es_exam_req_invigilator_s", "es_exam_req_location_s", ] DATE_FORMATS = ( "%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%d/%m/%Y", "%d-%m-%Y", "%d/%m/%y", ) TIME_FORMATS = ( "%H:%M", "%H:%M:%S", "%I:%M %p", "%I:%M:%S %p", "%I:%M%p", ) DATETIME_FORMATS = ( "%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M", "%Y-%m-%dT%H:%M:%S", ) BOOL_TRUE = {"1", "true", "yes", "y"} BOOL_FALSE = {"0", "false", "no", "n", ""} class ImportTable(AdminApiBase): BASE_TRANS = "import." def iter_sheets_from_upload(self, uploaded_file): filename = (uploaded_file.name or "").lower() ext = os.path.splitext(filename)[1] if ext == ".csv": text = uploaded_file.read().decode("utf-8-sig", errors="replace") reader = list(csv.reader(io.StringIO(text))) if not reader: return header = [self._header_cell(h) for h in reader[0]] def row_generator(): for i, values in enumerate(reader[1:], start=2): yield i, { header[idx]: (values[idx].strip() if idx < len(values) else "") for idx in range(len(header)) } yield "csv_sheet", header, row_generator() return if ext == ".xlsx": content = uploaded_file.read() wb = load_workbook(filename=io.BytesIO(content), data_only=True) for sheet_name in wb.sheetnames: ws = wb[sheet_name] if ws.max_row < 1: continue header = [self._header_cell(c.value) for c in ws[1]] def row_generator(ws=ws, header=header): for r, row_cells in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2): yield r, { header[idx]: ( "" if idx >= len(row_cells) or row_cells[idx] is None else str(row_cells[idx]).strip() ) for idx in range(len(header)) } yield sheet_name, header, row_generator() return if ext == ".xls": content = uploaded_file.read() book = xlrd.open_workbook(file_contents=content) for sheet_index in range(book.nsheets): sheet = book.sheet_by_index(sheet_index) sheet_name = sheet.name if sheet.nrows < 1: continue header = [self._header_cell(h) for h in sheet.row_values(0)] def row_generator(sheet=sheet, header=header): for r in range(1, sheet.nrows): values = sheet.row_values(r) yield r + 1, { header[idx]: ( "" if idx >= len(values) else str(values[idx]).strip() ) for idx in range(len(header)) } yield sheet_name, header, row_generator() return raise ValueError(__(self.BASE_TRANS + "unsupported_file")) def _header_cell(self, value): if value is None: return "" return str(value).strip() def import_return(self, rows_read=0, inserted=0, updated=0, skipped=0, errors=None): return { "rows_read": rows_read, "inserted": inserted, "updated": updated, "skipped": skipped, "errors": errors or [], } def _cell(self, value): if value is None: return "" if isinstance(value, str): return value.strip() return str(value).strip() def _parse_bool(self, value): val = self._cell(value).lower() if val in BOOL_TRUE: return True, True if val in BOOL_FALSE: return False, True return False, False def _parse_int(self, value): raw = self._cell(value) if raw == "": return None, True try: number = float(raw) if number.is_integer(): return int(number), True return int(number), True except (ValueError, TypeError): return None, False def _parse_date(self, value): raw = self._cell(value) if raw == "": return None, True if " " in raw: raw = raw.split(" ")[0] for fmt in DATE_FORMATS: try: return datetime.strptime(raw, fmt).date(), True except ValueError: continue return None, False def _parse_time(self, value): raw = self._cell(value) if raw == "": return None, True if raw.endswith(".000000"): raw = raw.split(".")[0] for fmt in TIME_FORMATS: try: return datetime.strptime(raw, fmt).time(), True except ValueError: continue return None, False def _parse_datetime(self, value): raw = self._cell(value) if raw == "": return None, True for fmt in DATETIME_FORMATS: try: return datetime.strptime(raw, fmt), True except ValueError: continue parsed_date, ok = self._parse_date(value) if ok and parsed_date is not None: return datetime.combine(parsed_date, dt_time.min), True return None, False def _model_field_names(self, model): names = set() for field in model._meta.fields: names.add(field.name) names.add(field.attname) return names def _has_status(self, model): return "status" in self._model_field_names(model) and hasattr(model, "STATUS_TO_CODE") def import_entity( self, request, header, rows, *, model, rules, fk_fields=None, self_fk=None, skip_model_fields=None, defaults=None, after_rows=None, extra_validate=None, ): fk_fields = fk_fields or {} skip_model_fields = set(skip_model_fields or set()) if "tag" in rules: skip_model_fields.add("tag") defaults = defaults or {} if "code" not in header: return self.import_return(errors=[{ "row": 1, "field": "code", "error": __(self.BASE_TRANS + "missing_required", value="code"), }]) required_insert_columns = [ field for field, field_rules in rules.items() if "required" in field_rules and field != "code" ] staged = [] rows_read = 0 rows_skipped = 0 errors = [] seen_codes = {} fk_candidates = {field: set() for field in fk_fields} for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue data[field] = self._cell(row.get(field)) if not data.get("code"): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "code_required"), }) continue code = data["code"] if code in seen_codes: rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "value": code, "error": __(self.BASE_TRANS + "duplicate_code_in_import", value=code, first_row=seen_codes[code]), }) continue seen_codes[code] = row_num for field in fk_fields: if field in header and data.get(field): fk_candidates[field].add(data[field]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) fk_id_maps = {} for field, (related_model, _required) in fk_fields.items(): if field == self_fk: continue fk_id_maps[field] = fetch_code_id_map(related_model, "code", fk_candidates[field]) cleaned = [] for row_num, data in staged: skip_row = False for field, field_rules in rules.items(): if field not in header: continue raw = data.get(field) required = "required" in field_rules if field in fk_fields: continue if "boolean" in field_rules: parsed, valid = self._parse_bool(raw) if not valid: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_boolean", value=raw), }) if required: skip_row = True break data[field] = False else: data[field] = parsed continue if "integer" in field_rules: parsed, valid = self._parse_int(raw) if not valid: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_integer", value=raw), }) if required: skip_row = True break data[field] = None else: data[field] = parsed continue if "date" in field_rules: parsed, valid = self._parse_date(raw) if not valid: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_date", value=raw), }) if required: skip_row = True break data[field] = None else: data[field] = parsed continue if "time" in field_rules: parsed, valid = self._parse_time(raw) if not valid: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_time", value=raw), }) if required: skip_row = True break data[field] = None else: data[field] = parsed continue if "datetime" in field_rules: parsed, valid = self._parse_datetime(raw) if not valid: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_date_time", value=raw), }) if required: skip_row = True break data[field] = None else: data[field] = parsed continue if isinstance(data.get(field), str) and data.get(field) == "": data[field] = None if skip_row: rows_skipped += 1 continue for field, (_related_model, required) in fk_fields.items(): if field not in header: continue code_value = self._cell(data.get(field)) if field == self_fk: data["_self_fk_code"] = code_value data.pop(field, None) continue if not code_value: data[field] = None if required: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "missing_required", value=field), }) skip_row = True continue resolved = fk_id_maps[field].get(code_value) if resolved is None: if required: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_required", field=field, value=code_value), }) skip_row = True else: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_overridden", field=field, value=code_value), }) data[field] = None else: data[field] = resolved if skip_row: rows_skipped += 1 continue if extra_validate: extra_error = extra_validate(row_num, data) if extra_error: errors.append(extra_error) rows_skipped += 1 continue cleaned.append((row_num, data)) if not cleaned: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) file_codes = [data["code"] for _, data in cleaned] existing_map = {obj.code: obj for obj in model.objects.filter(code__in=file_codes)} valid_fields = self._model_field_names(model) to_create = [] need_update = [] created_row_data = [] for row_num, data in cleaned: obj = existing_map.get(data["code"]) if obj is None: missing_for_insert = [] for col in required_insert_columns: if col not in header: missing_for_insert.append(col) continue if col == self_fk: if not data.get("_self_fk_code"): missing_for_insert.append(col) continue value = data.get(col) if value in (None, ""): missing_for_insert.append(col) if missing_for_insert: rows_skipped += 1 errors.append({ "row": row_num, "field": ",".join(missing_for_insert), "error": __(self.BASE_TRANS + "missing_required_insert", value=missing_for_insert), }) continue payload = {} for key, value in data.items(): if key in skip_model_fields or key.startswith("_"): continue if key in valid_fields: payload[key] = value for key, value in defaults.items(): if payload.get(key) is None: payload[key] = value payload["created_by"] = request.user.id if self._has_status(model): payload["status"] = model.STATUS_TO_CODE["active"] to_create.append(model(**payload)) created_row_data.append(data) if self_fk: data["inserted"] = True need_update.append((row_num, data)) else: need_update.append((row_num, data)) inserted = 0 with transaction.atomic(): if to_create: model.objects.bulk_create(to_create, 500) inserted = len(to_create) all_map = {obj.code: obj for obj in model.objects.filter(code__in=file_codes)} self_fk_map = {} if self_fk: all_codes = set(file_codes) all_codes.update(data.get("_self_fk_code") or "" for _, data in need_update if data.get("_self_fk_code")) all_codes.discard("") self_fk_map = fetch_code_id_map(model, "code", all_codes) updatable_fields = [ field for field in rules.keys() if field in header and field not in ["code", "status"] and field not in skip_model_fields ] if self_fk and self_fk in header: if self_fk not in updatable_fields: updatable_fields.append(self_fk) updatable_fields.append("updated_by") updatable_fields = [f for f in dict.fromkeys(updatable_fields) if f in valid_fields or f == "updated_by"] to_update = [] updated = 0 for row_num, data in need_update: obj = all_map.get(data["code"]) if obj is None: continue changed = False for field in updatable_fields: if field in ["updated_by", self_fk]: continue if field not in data: continue new_val = data.get(field) if new_val is None and "required" in rules.get(field, []): continue if getattr(obj, field, None) != new_val: setattr(obj, field, new_val) changed = True if self_fk and self_fk in header: parent_code = data.get("_self_fk_code") or "" new_parent_id = self_fk_map.get(parent_code) if parent_code else None if parent_code and new_parent_id is None: errors.append({ "row": row_num, "field": self_fk, "error": __(self.BASE_TRANS + "invalid_overridden", field=self_fk, value=parent_code), }) if getattr(obj, self_fk, None) != new_parent_id: setattr(obj, self_fk, new_parent_id) changed = True if changed: obj.updated_by = request.user.id if data.get("inserted"): updated -= 1 to_update.append(obj) elif not data.get("inserted"): rows_skipped += 1 with transaction.atomic(): if to_update and updatable_fields: model.objects.bulk_update(to_update, updatable_fields, 500) updated += len(to_update) if after_rows: after_rows(request, header, all_map, cleaned, errors) if "tag" in rules and "tag" in header: pending = getattr(self, "_pending_tags", None) if pending is None: pending = [] self._pending_tags = pending pending.append((model, all_map, cleaned, errors)) return self.import_return(rows_read, inserted, max(updated, 0), rows_skipped, errors) def import_relation( self, request, header, rows, *, rules, fk_fields, through_model, extra_validate=None, after_insert=None, ): for col in rules.keys(): if col not in header: return self.import_return(errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col), }]) rows_read = 0 rows_skipped = 0 errors = [] staged = [] fk_candidates = {field: set() for field in fk_fields} for row_num, row in rows: rows_read += 1 data = {} row_has_error = False for field in rules.keys(): data[field] = self._cell(row.get(field)) if "required" in rules[field] and not data[field]: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "missing_required", value=field), }) row_has_error = True if row_has_error: rows_skipped += 1 continue for field in fk_fields: fk_candidates[field].add(data[field]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) fk_id_maps = { field: fetch_code_id_map(related_model, "code", fk_candidates[field]) for field, related_model in fk_fields.items() } to_create = [] inserted_pairs = [] valid_fields = self._model_field_names(through_model) for row_num, data in staged: resolved = {} row_has_error = False for field, related_model in fk_fields.items(): related_id = fk_id_maps[field].get(data[field]) if related_id is None: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_required", field=field, value=data[field]), }) row_has_error = True else: resolved[field] = related_id if row_has_error: rows_skipped += 1 continue if extra_validate: extra_error = extra_validate(row_num, data, resolved) if extra_error: errors.append(extra_error) rows_skipped += 1 continue payload = dict(resolved) if "created_by" in valid_fields: payload["created_by"] = request.user.id to_create.append(through_model(**payload)) inserted_pairs.append(resolved) inserted = 0 if to_create: created = through_model.objects.bulk_create(to_create, 500, ignore_conflicts=True) inserted = len(created) if created and getattr(created[0], "pk", None) else len(to_create) if after_insert: after_insert(request, inserted_pairs) return self.import_return(rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors) def _split_multi_codes(self, raw): if raw in (None, ""): return [] parts = [] seen = set() for part in str(raw).replace(";", ",").split(","): part = part.strip() if not part or part in seen: continue seen.add(part) parts.append(part) return parts def _apply_pending_tags(self): pending = getattr(self, "_pending_tags", None) or [] if not pending: return all_codes = set() for _model, _all_map, cleaned, _errors in pending: for _row_num, data in cleaned: all_codes.update(self._split_multi_codes(data.get("tag"))) tag_map = fetch_code_id_map(EsTag, "code", all_codes) to_create = [] delete_by_ct = {} for model, all_map, cleaned, errors in pending: ct = ContentType.objects.get_for_model(model) refer_ids = delete_by_ct.setdefault(ct, set()) seen_ids = set() for row_num, data in cleaned: obj = all_map.get(data["code"]) if not obj or obj.id in seen_ids: continue seen_ids.add(obj.id) refer_ids.add(obj.id) tag_ids = [] seen_tag_ids = set() for code in self._split_multi_codes(data.get("tag")): resolved = tag_map.get(code) if resolved is None: errors.append({ "row": row_num, "field": "tag", "error": __(self.BASE_TRANS + "invalid_overridden", field="tag", value=code), }) continue if resolved in seen_tag_ids: continue seen_tag_ids.add(resolved) tag_ids.append(resolved) for tag_id in tag_ids: to_create.append(EsTagRelation( tag_id=tag_id, refer_table=ct, refer_id=obj.id, )) with transaction.atomic(): for ct, refer_ids in delete_by_ct.items(): if refer_ids: EsTagRelation.objects.filter(refer_table=ct, refer_id__in=refer_ids).delete() if to_create: EsTagRelation.objects.bulk_create(to_create, 500, ignore_conflicts=True) def _after_student_groups(self, request, header, all_map, cleaned, errors): if "student_group_id" not in header: return for _row_num, data in cleaned: obj = all_map.get(data["code"]) if not obj: continue group_id = data.get("student_group_id") replace_student_group_rows(obj.id, [group_id] if group_id else []) def _after_session_start_days(self, request, header, all_map, cleaned, errors): if "days" not in header: return for row_num, data in cleaned: obj = all_map.get(data["code"]) if not obj: continue raw = data.get("days") if raw in (None, ""): continue days = [] for part in str(raw).replace(";", ",").split(","): part = part.strip() if part == "": continue try: day = int(float(part)) except (ValueError, TypeError): errors.append({ "row": row_num, "field": "days", "error": __(self.BASE_TRANS + "invalid_integer", value=part), }) continue if day < 0 or day > 6: errors.append({ "row": row_num, "field": "days", "error": __(self.BASE_TRANS + "invalid_integer", value=part), }) continue days.append(day) if days: EsSessionStartDay.bulk_insert(obj.id, days) def _after_exam_requirement(self, request, header, all_map, cleaned, errors): existing_codes = set( EsExamActivity.objects.filter( exam_requirement_id__in=[obj.id for obj in all_map.values()], ).values_list("code", flat=True) ) to_create = [] for _row_num, data in cleaned: obj = all_map.get(data["code"]) if not obj: continue activity_code = f"{obj.code}-01" if activity_code in existing_codes: continue if EsExamActivity.objects.filter(exam_requirement_id=obj.id).exists(): continue to_create.append(EsExamActivity( exam_requirement=obj, code=activity_code, name=obj.name, is_scheduled=False, created_by=request.user.id, )) existing_codes.add(activity_code) if to_create: EsExamActivity.objects.bulk_create(to_create, 500) def _validate_exam_period(self, row_num, data): start_date = data.get("start_date") end_date = data.get("end_date") if start_date and end_date and start_date > end_date: return { "row": row_num, "field": "end_date", "error": __(self.BASE_TRANS + "end_date_before_start"), } start_time = data.get("start_time") end_time = data.get("end_time") if start_time and end_time and start_time > end_time: return { "row": row_num, "field": "end_time", "error": __(self.BASE_TRANS + "end_time_before_start"), } return None def _validate_exam_requirement(self, row_num, data): earliest = data.get("earliest_start") latest = data.get("latest_end") if earliest and latest and earliest > latest: return { "row": row_num, "field": "latest_end", "error": __(self.BASE_TRANS + "end_date_before_start"), } return None def _validate_invigilator(self, row_num, data): staff_id = data.get("staff_id") if not staff_id: return None taken = EsInvigilator.objects.filter(staff_id=staff_id).exclude(code=data["code"]).exists() if taken: return { "row": row_num, "field": "staff_id", "error": __(self.BASE_TRANS + "staff_already_linked", value=staff_id), } return None def _copy_requirement_suitability(self, inserted_pairs, activity_through, requirement_field="exam_requirement_id"): if not inserted_pairs: return req_ids = {pair[requirement_field] for pair in inserted_pairs} activities = {} for activity in EsExamActivity.objects.filter(exam_requirement_id__in=req_ids).order_by("id"): activities.setdefault(activity.exam_requirement_id, activity) to_create = [] for pair in inserted_pairs: activity = activities.get(pair[requirement_field]) if not activity: continue to_create.append(activity_through( exam_activity_id=activity.id, suitability_id=pair["suitability_id"], )) if to_create: activity_through.objects.bulk_create(to_create, 500, ignore_conflicts=True) def import_es_zone(self, request, header, rows): return self.import_entity( request, header, rows, model=EsZone, rules=EsZone.IMPORT_RULES, fk_fields={"zone_id": (EsZone, False)}, self_fk="zone_id", ) def import_es_department(self, request, header, rows): return self.import_entity( request, header, rows, model=EsDepartment, rules=EsDepartment.IMPORT_RULES, fk_fields={ "department_id": (EsDepartment, False), "zone_id": (EsZone, False), }, self_fk="department_id", ) def import_es_tag(self, request, header, rows): return self.import_entity( request, header, rows, model=EsTag, rules=EsTag.IMPORT_RULES, fk_fields={"department_id": (EsDepartment, False)}, ) def import_es_suitability(self, request, header, rows): return self.import_entity( request, header, rows, model=EsSuitability, rules=EsSuitability.IMPORT_RULES, fk_fields={"department_id": (EsDepartment, False)}, ) def import_es_pos(self, request, header, rows): return self.import_entity( request, header, rows, model=EsPos, rules=EsPos.IMPORT_RULES, fk_fields={"department_id": (EsDepartment, False)}, ) def import_es_student_group(self, request, header, rows): return self.import_entity( request, header, rows, model=EsStudentGroup, rules=EsStudentGroup.IMPORT_RULES, fk_fields={"department_id": (EsDepartment, False)}, ) def import_es_invigilator_role(self, request, header, rows): return self.import_entity( request, header, rows, model=EsInvigilatorRole, rules=EsInvigilatorRole.IMPORT_RULES, ) def import_es_staff(self, request, header, rows): return self.import_entity( request, header, rows, model=EsStaff, rules=EsStaff.IMPORT_RULES, fk_fields={"department_id": (EsDepartment, False)}, defaults={"is_invigilator": False}, ) def import_es_student(self, request, header, rows): return self.import_entity( request, header, rows, model=EsStudent, rules=EsStudent.IMPORT_RULES, fk_fields={ "department_id": (EsDepartment, False), "enrolled_programme_id": (EsPos, False), "student_group_id": (EsStudentGroup, False), }, defaults={"need_extra_provision": False}, after_rows=self._after_student_groups, ) def import_es_location(self, request, header, rows): return self.import_entity( request, header, rows, model=EsLocation, rules=EsLocation.IMPORT_RULES, fk_fields={ "zone_id": (EsZone, False), "department_id": (EsDepartment, False), }, defaults={"row": 0, "column": 0, "is_partition": False}, ) def import_es_invigilator(self, request, header, rows): seen_staff = {} def extra_validate(row_num, data): error = self._validate_invigilator(row_num, data) if error: return error staff_id = data.get("staff_id") if staff_id: if staff_id in seen_staff: return { "row": row_num, "field": "staff_id", "error": __(self.BASE_TRANS + "staff_already_linked", value=staff_id), } seen_staff[staff_id] = data["code"] return None return self.import_entity( request, header, rows, model=EsInvigilator, rules=EsInvigilator.IMPORT_RULES, fk_fields={ "department_id": (EsDepartment, False), "staff_id": (EsStaff, False), }, extra_validate=extra_validate, ) def import_es_exam_period(self, request, header, rows): return self.import_entity( request, header, rows, model=EsExamPeriod, rules=EsExamPeriod.IMPORT_RULES, extra_validate=self._validate_exam_period, ) def import_es_session_start(self, request, header, rows): return self.import_entity( request, header, rows, model=EsSessionStart, rules=EsSessionStart.IMPORT_RULES, fk_fields={"exam_period_id": (EsExamPeriod, True)}, skip_model_fields={"days"}, after_rows=self._after_session_start_days, ) def import_es_exam_requirement(self, request, header, rows): return self.import_entity( request, header, rows, model=EsExamRequirement, rules=EsExamRequirement.IMPORT_RULES, fk_fields={"exam_period_id": (EsExamPeriod, True)}, defaults={"location_required": False, "exclusive_use": False}, extra_validate=self._validate_exam_requirement, after_rows=self._after_exam_requirement, ) def import_es_suitability_invigilator(self, request, header, rows): return self.import_relation( request, header, rows, rules=EsSuitability.SUITABILITY_INVIGILATOR["IMPORT_RULES"], fk_fields={"suitability_id": EsSuitability, "invigilator_id": EsInvigilator}, through_model=EsInvigilatorSuitability, ) def import_es_suitability_location(self, request, header, rows): return self.import_relation( request, header, rows, rules=EsSuitability.SUITABILITY_LOCATION["IMPORT_RULES"], fk_fields={"suitability_id": EsSuitability, "location_id": EsLocation}, through_model=EsLocationSuitability, ) def import_es_invigilator_roles(self, request, header, rows): return self.import_relation( request, header, rows, rules=EsInvigilator.INVIGILATOR_ROLES["IMPORT_RULES"], fk_fields={"invigilator_id": EsInvigilator, "role_id": EsInvigilatorRole}, through_model=EsInvigilatorInvigilatorRole, ) def import_es_student_student_group(self, request, header, rows): return self.import_relation( request, header, rows, rules=EsStudent.STUDENT_STUDENT_GROUP["IMPORT_RULES"], fk_fields={"student_id": EsStudent, "student_group_id": EsStudentGroup}, through_model=EsStudentStudentGroup, ) def import_es_exam_period_unavailability(self, request, header, rows): def extra_validate(row_num, data, resolved): parsed, valid = self._parse_date(data.get("unavailable_date")) if not valid or parsed is None: return { "row": row_num, "field": "unavailable_date", "error": __(self.BASE_TRANS + "invalid_date", value=data.get("unavailable_date")), } period = EsExamPeriod.objects.filter(id=resolved["exam_period_id"]).first() if period and (parsed < period.start_date or parsed > period.end_date): return { "row": row_num, "field": "unavailable_date", "error": __(self.BASE_TRANS + "unavailable_date_out_of_range"), } resolved["unavailable_date"] = parsed return None return self.import_relation( request, header, rows, rules=EsExamPeriod.EXAM_PERIOD_UNAVAILABILITY["IMPORT_RULES"], fk_fields={"exam_period_id": EsExamPeriod}, through_model=EsExamPeriodUnavailability, extra_validate=extra_validate, ) def import_es_exam_requirement_student(self, request, header, rows): return self.import_relation( request, header, rows, rules=EsExamRequirement.EXAM_REQUIREMENT_STUDENT["IMPORT_RULES"], fk_fields={"exam_requirement_id": EsExamRequirement, "student_id": EsStudent}, through_model=EsExamRequirementStudent, ) def import_es_exam_req_invigilator_s(self, request, header, rows): return self.import_relation( request, header, rows, rules=EsExamRequirement.EXAM_REQ_INVIGILATOR_S["IMPORT_RULES"], fk_fields={"exam_requirement_id": EsExamRequirement, "suitability_id": EsSuitability}, through_model=EsExamRequirementInvigilatorSuitability, after_insert=lambda request, pairs: self._copy_requirement_suitability( pairs, EsExamActivityInvigilatorSuitability ), ) def import_es_exam_req_location_s(self, request, header, rows): return self.import_relation( request, header, rows, rules=EsExamRequirement.EXAM_REQ_LOCATION_S["IMPORT_RULES"], fk_fields={"exam_requirement_id": EsExamRequirement, "suitability_id": EsSuitability}, through_model=EsExamRequirementLocationSuitability, after_insert=lambda request, pairs: self._copy_requirement_suitability( pairs, EsExamActivityLocationSuitability ), ) def post(self, request): try: setting = EsSetting.objects.filter(param="import_tables").first() if not setting or not setting.value: return self.api_response(error=__(self.BASE_TRANS + "no_importables"), code=status.HTTP_400_BAD_REQUEST) import_tables = json.loads(setting.value) uploaded = request.FILES.get("import_file") if not uploaded: return self.api_response(error=__(self.BASE_TRANS + "no_file"), code=status.HTTP_400_BAD_REQUEST) self._pending_tags = [] sheets = {} try: for sheet_name, header, rows in self.iter_sheets_from_upload(uploaded): sheets[sheet_name] = (header, list(rows)) except ValueError as e: return self.api_response(error=str(e), code=status.HTTP_400_BAD_REQUEST) response = {} processed = set() for sheet_name in IMPORT_SHEET_ORDER: if sheet_name not in sheets: continue processed.add(sheet_name) if sheet_name not in import_tables: response[sheet_name] = self.import_return(errors=__(self.BASE_TRANS + "sheet_name_not_found")) continue header, rows = sheets[sheet_name] import_function_name = f"import_{sheet_name}" try: response[sheet_name] = getattr(self, import_function_name)(request, header, rows) except Exception as e: response[sheet_name] = self.import_return(errors=["Function in progress"]) e_details = get_exception_detail(e) log_critical_error(user_id=None, descr=e_details["descr"], url=e_details["url"], trace=e_details["trace"]) for sheet_name in sheets: if sheet_name not in processed: response[sheet_name] = self.import_return(errors=__(self.BASE_TRANS + "sheet_name_not_found")) self._apply_pending_tags() return self.api_response(data=response) except Exception as e: e_details = get_exception_detail(e) log_critical_error(user_id=getattr(request.user, "id", None), descr=e_details["descr"], url=e_details["url"], trace=e_details["trace"]) return self.api_response(error=__("message.internal_server_error"), code=status.HTTP_500_INTERNAL_SERVER_ERROR)