import json import time as perf_time from xml.parsers.expat import errors from datetime import datetime from datetime import time from django.apps import apps from django.db import transaction, connection from django.db.models import ForeignKey from django.forms.models import model_to_dict from rest_framework.exceptions import ValidationError from api.helper.resource_map_helper import helper_recalculate_and_get_redis_resource_map from api.models.relations import TtAcademicTermWeek, TtActivityLocation, TtActivityTemplateWeek, TtActivityWeek, TtLocationSuitability, \ TtModuleWeek, TtPosModuleGroup, TtPosModuleGroupModule, TtStaffSuitability, TtStudentSetModule, TtWeekPatternWeek, TtActivityTemplateStaff, \ TtActivityTemplateStaffSuitability, TtActivityStaffPreset, TtActivityStaffSuitability, TtActivityTemplateLocation, \ TtActivityTemplateLocationSuitability, TtActivityLocationPreset, TtActivityLocationSuitability, TtStudentSetActivity, \ TtStudentAcademicTerm, TtStudentPathway, TtStudentPos, TtStudentPosModule, TtStudentResourceMap, TtStudentSetStudent from api.models.tt_academic_term import TtAcademicTerm from api.models.tt_activity import TtActivity from api.models.tt_activity_template import TtActivityTemplate from api.models.tt_activity_type import TtActivityType from api.models.tt_availability import TtAvailability from api.models.tt_department import TtDepartment from api.models.tt_location import TtLocation from api.models.tt_module import TtModule from api.models.tt_module_group import TtModuleGroup from api.models.tt_pos import TtPos from api.models.tt_pathway import TtPathway from api.models.tt_setting import TtSetting from api.models.tt_staff import TtStaff from api.models.tt_start_preference import TtStartPreference from api.models.tt_student_set import TtStudentSet from api.models.tt_student import TtStudent from api.models.tt_suitability import TtSuitability from api.models.tt_usage_preference import TtUsagePreference from api.models.tt_week import TtWeek from api.models.tt_week_pattern import TtWeekPattern from api.models.tt_zone import TtZone from api.import_helpers import ( objects_by_code_and_academic_term, objects_grouped_by_code, ) from api.translation import __ from api.utils import bulk_sync_to_redis, calculate_default_period, convert_to_redis_week, create_resource_map_multiple, \ fetch_code_id_map, get_resource_map_redis_data, get_slot_default_pattern, log_critical_error, get_exception_detail, \ time_to_slot, import_get_resource_map_redis_data from api.validator import BaseValidator from api.views.admin.base import AdminApiBase from backend.kafka import send_request from backend.redis_client import redis_client import csv import io import os import re from openpyxl import load_workbook import xlrd # requires xlrd<2.0 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] # ====================== # CSV (single sheet) # ====================== if ext == ".csv": text = uploaded_file.read().decode("utf-8-sig", errors="replace") f = io.StringIO(text) reader = list(csv.reader(f)) if not reader: return raw_header = reader[0] header = [h for h in raw_header] 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 # ====================== # XLSX (multi-sheet) # ====================== 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 raw_header = [c.value for c in ws[1]] header = [h if h else "" for h in raw_header] def row_generator(ws=ws, header=header): # 2026-05-26 this part cause the loading slow when import # Using ws[r] forced openpyxl to scan the file's XML from row 1 down to 'r' # on EVERY iteration, causing an O(N^2) slowdown (15+ minutes for 50k rows). # for r in range(2, ws.max_row + 1): # values = [c.value for c in ws[r]] # iter_rows() streams the file linearly in a single pass, dropping runtime to seconds. for r, row_cells in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2): values = row_cells yield r, { header[idx]: ( "" if idx >= len(values) or values[idx] is None else str(values[idx]).strip() ) for idx in range(len(header)) } yield sheet_name, header, row_generator() return # ====================== # XLS (multi-sheet) # ====================== 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 raw_header = sheet.row_values(0) header = [h for h in raw_header] 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 kafka_response(self, request, table, to_create, to_update, updatable_fields, relation_fields=[]): model_name = "".join(word.capitalize() for word in table.split("_")) model = apps.get_model("api", f"Tt{model_name}") create_method = f"{table}_create" update_method = f"{table}_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] ignore_fields = ["created_at", "updated_at", "created_by", "updated_by"] # CREATE created_kafka_data = [] created_ids = [c.id for c in to_create] if created_ids: created_data = model.objects.filter(id__in=created_ids) for data in created_data: temp = {} for f in model._meta.fields: if f.name in ignore_fields: continue if isinstance(f, ForeignKey): # if is foreign key, dun direct follow field name, cause will call 1 more time to db # eg. field name is department, if using department.id to get the id, will call to db since not prefethch this value, need used department_id db_column_value = getattr(data, f"{f.name}_id") if db_column_value is not None: temp[f.name] = db_column_value else: # move to here to fetch value, so if is foreign key, wont come to this part, can skip the get value from FK table problem value = getattr(data, f.name) if value is not None: temp[f.name] = value created_kafka_data.append(temp) if created_kafka_data: kafka_request_data = { "session_id": request.user.name, table: created_kafka_data, } send_request(kafka_topic, kafka_request_data, None, create_method) # UPDATE update_kafka_data = [] update_fields = [f for f in updatable_fields if f != "updated_by"] for data in to_update: temp = {"id": data.id} for up in update_fields: value = getattr(data, up, None) # if value is None: # continue temp[up] = value update_kafka_data.append(temp) if update_kafka_data: kafka_request_data = { "session_id": request.user.name, table: update_kafka_data, } send_request(kafka_topic, kafka_request_data, None, update_method) def parse_import_bool(self, value): if value is None: return False val = str(value).strip().lower() if val == "": return False if val in ["1", "true"]: return True if val in ["0", "false"]: return False return False def safe_int_or_zero(self, value): try: if value in [None, ""]: return 0 return int(value) except (ValueError, TypeError): return 0 def format_time_if_needed(self, value): if isinstance(value, time): return value.strftime("%H:%M") return value def normalize_pattern(self, value, allowed_type="availability", num=336, default_pattern=None): val = (value or "").strip() num = int(num) if isinstance(num, str) else num if allowed_type == "availability": regex = rf"^[01]{{{num}}}$" else: regex = rf"^[1-9]{{{num}}}$" # 1. If uploaded pattern is valid, use uploaded pattern if re.fullmatch(regex, val): return val # 2. If academic term default is valid, use academic term default academic_default = (default_pattern or "").strip() if re.fullmatch(regex, academic_default): return academic_default # 3. Last fallback: old system default return get_slot_default_pattern(num).get(allowed_type, None) def resolve_code_or_pattern( self, data, row_num, code_field, pattern_field, code_to_id_map, errors, trans_key, num=336, pattern_type="availability", default_pattern=None, ): raw_code = (data.get(code_field) or "").strip() raw_pattern = (data.get(pattern_field) or "").strip() num = int(num) if isinstance(num, str) else num if raw_code: resolved_id = code_to_id_map.get(raw_code) if resolved_id is not None: data[code_field] = resolved_id data[pattern_field] = None return errors.append({ "row": row_num, "field": code_field, "error": __(trans_key + "invalid_id_pattern", field=code_field, value=raw_code), }) data[code_field] = None data[pattern_field] = self.normalize_pattern( raw_pattern, pattern_type, num, default_pattern=default_pattern, ) def resolve_academic_term_id( self, row_num, obj, academic_term_code, academic_term_code_to_id, errors, trans_key, ): """ Returns: (effective_academic_term_id, has_error) """ # update case → use existing if obj: return obj.academic_term_id, False # create case if academic_term_code: academic_term_id = academic_term_code_to_id.get(academic_term_code) if academic_term_id is None: errors.append({ "row": row_num, "field": "academic_term_id", "error": __(trans_key + "invalid_required", field="academic term code", value=academic_term_code), }) return None, True return academic_term_id, False # missing required errors.append({ "row": row_num, "field": "academic_term_id", "error": __(trans_key + "missing_required", value="academic_term_id"), }) return None, True def import_return(self, rows_read = 0, inserted = 0, updated = 0, skipped = 0, errors = []): return { "rows_read": rows_read, "inserted": inserted, "updated": updated, "skipped": skipped, "errors": errors, } def import_tt_department(self, request, header, rows): # if exist in db, update # if not exist in db, insert # if no code, dont allow insert # check column rules = TtDepartment.IMPORT_RULES #check if there's code at header 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, rules in rules.items() if "required" in rules and field != "code" ] staged = [] rows_read = 0 rows_skipped = 0 errors = [] seen_codes = {} dept_code_candidates = set() zone_code_candidates = set() #row checking, normalizing data for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value # code always need value if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS+"code_required"), }) continue code = data["code"].strip() 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 if "department_id" in header and data.get("department_id"): dept_code_candidates.add(data["department_id"].strip()) if "zone_id" in header and data.get("zone_id"): zone_code_candidates.add(data["zone_id"].strip()) data["status"] = TtDepartment.STATUS_TO_CODE["active"] data["created_by"] = request.user.id staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) #query check zone_code_to_id = fetch_code_id_map(TtZone, "code", zone_code_candidates) cleaned = [] for row_num, data in staged: dep_code = (data.get("department_id") or "").strip() zone_code = (data.get("zone_id") or "").strip() if "department_id" in header: if dep_code: data["_department_code"] = dep_code else: data["_department_code"] = "" data.pop("department_id", None) if "zone_id" in header: if zone_code: zone_id = zone_code_to_id.get(zone_code) if zone_id is None: errors.append({ "row": row_num, "field": "zone_id", "error": __(self.BASE_TRANS+"invalid_overridden", field="zone code", value=zone_code), }) data["zone_id"] = None data["zone_id"] = zone_id else: data["zone_id"] = None if "desc" in header and not (data.get("desc") or "").strip(): data["desc"] = None cleaned.append((row_num, data)) if not cleaned: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) # insert file_codes = [data["code"] for _, data in cleaned] existing_qs = TtDepartment.objects.filter(code__in=file_codes) existing_map = {obj.code: obj for obj in existing_qs} to_create = [] need_update = [] for row_num, data in cleaned: code = data["code"] obj = existing_map.get(code) # if obj is None, means code not found inside db, so its insert if obj is None: missing_for_insert = [ col for col in required_insert_columns if col not in header or not (data.get(col) or "").strip() ] 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 temp = data.copy() temp.pop("_department_code", None) data['inserted'] = True to_create.append(TtDepartment(**temp)) if "_department_code" in data: need_update.append((row_num, data)) else: need_update.append((row_num, data)) inserted = 0 with transaction.atomic(): if to_create: TtDepartment.objects.bulk_create(to_create, 500) inserted = len(to_create) # update all_dept_codes_to_map = set(dept_code_candidates) | set(file_codes) department_code_to_id = fetch_code_id_map(TtDepartment, "code", all_dept_codes_to_map) #fetch all again, ori update + update depart id all_qs = TtDepartment.objects.filter(code__in=file_codes) all_map = {obj.code: obj for obj in all_qs} to_update = [] redis_ids = set() updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status", "department_id"] ] if "department_id" in header: updatable_fields.append("department_id") updatable_fields.append("updated_by") updatable_fields = list(dict.fromkeys(updatable_fields)) for row_num, data in need_update: obj = all_map.get(data["code"]) if obj is None: continue changed = False for f in updatable_fields: for f in updatable_fields: if f in ["updated_by", "department_id"]: # skip deparment checkign here continue if f not in data: continue new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue if getattr(obj, f, None) != new_val: setattr(obj, f, new_val) changed = True #department_id check if "department_id" in header: dep_code = data.get("_department_code", "") if dep_code is not None: dep_id = department_code_to_id.get(dep_code) new_dep_id = dep_id if dep_id is not None else None # check if exist else None if getattr(obj, "department_id", None) != new_dep_id: redis_ids.update(filter(None, (getattr(obj, "department_id", None), new_dep_id))) obj.department_id = new_dep_id changed = True if dep_id is None and dep_code.strip(): errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS+"invalid_overridden", field="department code", value=dep_code), }) if changed: setattr(obj, "updated_by", request.user.id) if "inserted" in data and data['inserted'] is True: updated -= 1 to_update.append(obj) else: rows_skipped += 1 with transaction.atomic(): if to_update and updatable_fields: TtDepartment.objects.bulk_update(to_update, updatable_fields, 500) updated += len(to_update) #changes to redis, insert first, then update redis_department_table = "department" redis_department_column = [ "id","code","name","sub_department_id" ] update_redis_department_column = [ "name" ] redis_data = { "insert": {}, "update": {}, } redis_department_data = {} #insert redis for department in to_create: for column in redis_department_column: match column: case "sub_department_id": # when create wont have sub department, this is update for upline redis_department_data[column] = None case _: redis_department_data[column] = getattr(department, column) redis_data["insert"].setdefault(redis_department_table, []).append(redis_department_data) redis_department_data={} #update redis redis_department_data = {} #clear previous data for department in to_update: redis_department_data = {"id": getattr(department, "id")} for column in update_redis_department_column: redis_department_data[column] = getattr(department, column) if len(redis_department_data)>1:# > 1 means not only id redis_data["update"].setdefault(redis_department_table, []).append(redis_department_data) #department id if redis_ids: departments = TtDepartment.objects.filter(id__in=redis_ids).prefetch_related("sub_department") for department in departments: redis_department_data = { "id": department.id, "sub_department_id": [sd.id for sd in department.sub_department.all()] or None } redis_data["update"].setdefault(redis_department_table, []).append(redis_department_data) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) #kafka self.kafka_response(request, "department", to_create, to_update, updatable_fields) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_zone(self, request, header, rows): # check column rules = TtZone.IMPORT_RULES #check if there's code at header 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, rules in rules.items() if "required" in rules and field != "code" ] staged = [] rows_read = 0 rows_skipped = 0 errors = [] seen_codes = {} dept_code_candidates = set() zone_code_candidates = set() #row checking, normalizing data for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value # code always need value if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS+"code_required"), }) continue code = data["code"].strip() 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 if "department_id" in header and data.get("department_id"): dept_code_candidates.add(data["department_id"].strip()) if "zone_id" in header and data.get("zone_id"): zone_code_candidates.add(data["zone_id"].strip()) data["status"] = TtZone.STATUS_TO_CODE["active"] staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) department_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) cleaned = [] for row_num, data in staged: dep_code = (data.get("department_id") or "").strip() zone_code = (data.get("zone_id") or "").strip() if "zone_id" in header: if dep_code: data["_zone_code"] = zone_code else: data["_zone_code"] = "" data.pop("zone_id", None) if "department_id" in header: if dep_code: dep_id = department_code_to_id.get(dep_code) if dep_id is None: errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS+"invalid_overridden", field="department code", value=dep_code), }) data["department_id"] = None data["department_id"] = dep_id else: data["department_id"] = None if "desc" in header and not (data.get("desc") or "").strip(): data["desc"] = None 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_qs = TtZone.objects.filter(code__in=file_codes) existing_map = {obj.code: obj for obj in existing_qs} to_create = [] need_update = [] for row_num, data in cleaned: code = data["code"] obj = existing_map.get(code) # if obj is None, means code not found inside db, so its insert if obj is None: missing_for_insert = [ col for col in required_insert_columns if col not in header or not (data.get(col) or "").strip() ] 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 temp = data.copy() temp.pop("_zone_code", None) data['inserted'] = True to_create.append(TtZone(**temp)) if "_zone_code" in data: need_update.append((row_num, data)) else: need_update.append((row_num, data)) inserted = 0 with transaction.atomic(): if to_create: TtZone.objects.bulk_create(to_create, 500) inserted = len(to_create) # update all_zone_codes_to_map = set(zone_code_candidates) | set(file_codes) zone_code_to_id = fetch_code_id_map(TtZone, "code", all_zone_codes_to_map) #fetch all again, ori update + update depart id all_qs = TtZone.objects.filter(code__in=file_codes) all_map = {obj.code: obj for obj in all_qs} to_update = [] redis_ids = set() updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status", "zone_id"] ] if "zone_id" in header: updatable_fields.append("zone_id") updatable_fields.append("updated_by") updatable_fields = list(dict.fromkeys(updatable_fields)) for row_num, data in need_update: obj = all_map.get(data["code"]) if obj is None: continue changed = False for f in updatable_fields: if f in ["updated_by", "zone_id"]: continue if f not in data: continue new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue if getattr(obj, f, None) != new_val: setattr(obj, f, new_val) changed = True #zone_id check if "zone_id" in header: zone_code = data.get("_zone_code") if zone_code is not None: zone_id = zone_code_to_id.get(zone_code) new_zone_id = zone_id if zone_id is not None else None # check if exist else None if getattr(obj, "zone_id", None) != new_zone_id: redis_ids.update(filter(None, (getattr(obj, "zone_id", None), new_zone_id))) obj.zone_id = new_zone_id changed = True if zone_id is None and zone_code.strip(): errors.append({ "row": row_num, "field": "zone_id", "error": __(self.BASE_TRANS+"invalid_overridden", field="zone code", value=zone_code), }) if changed: if "inserted" in data and data['inserted'] is True: updated -= 1 to_update.append(obj) else: rows_skipped += 1 with transaction.atomic(): if to_update and updatable_fields: TtZone.objects.bulk_update(to_update, updatable_fields, 500) updated += len(to_update) #changes to redis, insert first, then update redis_zone_table = "zone" redis_zone_column = [ "id","code","name","sub_zone_id" ] update_redis_zone_column = [ "name" ] redis_data = { "insert": {}, "update": {}, } redis_zone_data = {} #insert redis for zone in to_create: for column in redis_zone_column: match column: case "sub_zone_id": # when create wont have sub department, this is update for upline redis_zone_data[column] = None case _: redis_zone_data[column] = getattr(zone, column) redis_data["insert"].setdefault(redis_zone_table, []).append(redis_zone_data) redis_zone_data={} #update redis redis_zone_data = {} #clear previous data for zone in to_update: redis_zone_data = {"id": getattr(zone, "id")} for column in update_redis_zone_column: redis_zone_data[column] = getattr(zone, column) if len(redis_zone_data)>1:# > 1 means not only id redis_data["update"].setdefault(redis_zone_table, []).append(redis_zone_data) #zone id if redis_ids: zones = TtZone.objects.filter(id__in=redis_ids).prefetch_related("sub_zone") for zone in zones: redis_zone_data = { "id": zone.id, "sub_zone_id": [sd.id for sd in zone.sub_zone.all()] or None } redis_data["update"].setdefault(redis_zone_table, []).append(redis_zone_data) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) #kafka self.kafka_response(request, "zone", to_create, to_update, updatable_fields) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_staff(self, request, header, rows, settings): slot_per_week = settings.get("slot_per_week") # check column rules = TtStaff.IMPORT_RULES #check if there's code at header 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 = {} dept_code_candidates = set() zone_code_candidates = set() availability_code_candidates = set() start_preference_code_candidates = set() usage_preference_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value # code always need value if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS+"code_required"), }) continue code = data["code"].strip() 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 #integer for field, field_rules in rules.items(): if "integer" in field_rules: raw_value = data.get(field) # calculate default period hours_per_week = TtLocation.WORK_TIME.get("hours_per_day") * TtLocation.WORK_TIME.get("days_per_week") default_period = calculate_default_period(hours_per_week, settings.get("minute_per_slot")) if raw_value in ["", None]: if field in ["maximum_period", "contract_period"]: data[field] = default_period else: data[field] = 0 if "required" in field_rules else None continue try: data[field] = int(raw_value) except (ValueError, TypeError): errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_integer", value=raw_value), }) if field in ["maximum_period", "contract_period"]: data[field] = default_period else: data[field] = None if "department_id" in header and data.get("department_id"): dept_code_candidates.add(data["department_id"].strip()) if "zone_id" in header and data.get("zone_id"): zone_code_candidates.add(data["zone_id"].strip()) if "availability_id" in header and data.get("availability_id"): availability_code_candidates.add(data["availability_id"].strip()) if "start_preference_id" in header and data.get("start_preference_id"): start_preference_code_candidates.add(data["start_preference_id"].strip()) if "usage_preference_id" in header and data.get("usage_preference_id"): usage_preference_code_candidates.add(data["usage_preference_id"].strip()) data["status"] = TtStaff.STATUS_TO_CODE["active"] data["created_by"] = request.user.id staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) dept_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) zone_code_to_id = fetch_code_id_map(TtZone, "code", zone_code_candidates) availability_code_to_id = fetch_code_id_map(TtAvailability, "code", availability_code_candidates) start_preference_code_to_id = fetch_code_id_map(TtStartPreference, "code", start_preference_code_candidates) usage_preference_code_to_id = fetch_code_id_map(TtUsagePreference, "code", usage_preference_code_candidates) # check if exist cleaned = [] for row_num, data in staged: dep_code = (data.get("department_id") or "").strip() zone_code = (data.get("zone_id") or "").strip() if "department_id" in header: if dep_code: dept_id = dept_code_to_id.get(dep_code) if dept_id is None: errors.append({ "row": row_num, "field": "dept_id", "error": __(self.BASE_TRANS+"invalid_overridden", field="department code", value=dep_code), }) data["department_id"] = None data["department_id"] = dept_id else: data["department_id"] = None if "zone_id" in header: if zone_code: zone_id = zone_code_to_id.get(zone_code) if zone_id is None: errors.append({ "row": row_num, "field": "zone_id", "error": __(self.BASE_TRANS+"invalid_overridden", field="zone code", value=zone_code), }) data["zone_id"] = None data["zone_id"] = zone_id else: data["zone_id"] = None if "desc" in header and not (data.get("desc") or "").strip(): data["desc"] = None for bool_field in ["is_part_time", "shared_with_all_department"]: raw_value = data.get(bool_field) parsed = self.parse_import_bool(raw_value) if raw_value not in ["", None] and str(raw_value).strip().lower() not in ["1", "0", "true", "false"]: errors.append({ "row": row_num, "field": bool_field, "error": __(self.BASE_TRANS + "invalid_boolean", value=raw_value), }) data[bool_field] = parsed self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="availability_id", pattern_field="availability_pattern", code_to_id_map=availability_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="availability", ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="start_preference_id", pattern_field="start_preference_pattern", code_to_id_map=start_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="start_preference", ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="usage_preference_id", pattern_field="usage_preference_pattern", code_to_id_map=usage_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="usage_preference", ) cleaned.append((row_num, data)) if not cleaned: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) # insert file_codes = [data["code"] for _, data in cleaned] existing_qs = TtStaff.objects.filter(code__in=file_codes) existing_map = {obj.code: obj for obj in existing_qs} to_create = [] to_update = [] created_ids = [] inserted = 0 updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status"] ] updatable_fields.append("updated_by") updatable_fields = list(dict.fromkeys(updatable_fields)) for row_num, data in cleaned: code = data["code"] obj = existing_map.get(code) if obj is None: missing_for_insert = [ col for col in required_insert_columns if col not in header or not (data.get(col) or "").strip() ] 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 to_create.append(TtStaff(**data)) else: changed = False for f in updatable_fields: if f == "updated_by": continue if f not in data: continue new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue if getattr(obj, f, None) != new_val: setattr(obj, f, new_val) changed = True if changed: setattr(obj, "updated_by", request.user.id) to_update.append(obj) else: rows_skipped += 1 from api.services.integration.tracking import track_resource_created, track_resource_queryset track_resource_queryset(to_update) with transaction.atomic(): if to_create: TtStaff.objects.bulk_create(to_create, 500) for created_staff in to_create: track_resource_created(created_staff) inserted = len(to_create) if to_update and updatable_fields: TtStaff.objects.bulk_update(to_update, updatable_fields, 500) updated += len(to_update) #insert resource map weeks_data = list(TtWeek.objects.values_list("id", flat=True)) if weeks_data and to_create: created_ids = [c.id for c in to_create] default_resource_map = get_slot_default_pattern(slot_per_week).get("resource_map", None) create_resource_map_multiple("staff", created_ids, weeks_data, default_resource_map) redis_staff_table = "staff" redis_staff_column = [ "id","code", "name", "department_id", "zone_id", "maximum_period", "contract_period", "constraint_profile","avoid_concurrency", "availability_id", "availability_pattern","start_preference_id", "start_preference_pattern", "usage_preference_id", "usage_preference_pattern", "shared_with_all_department", "shared_with_department","is_part_time","resource_map","allocated_activity_ids" ] update_redis_staff_column = [ "name","department_id","zone_id","maximum_period", "contract_period", "constraint_profile","avoid_concurrency","availability_id","availability_pattern","start_preference_id", "start_preference_pattern","usage_preference_id","usage_preference_pattern","shared_with_all_department","is_part_time", "shared_with_department" ] redis_data = { "insert": {}, "update": {}, } redis_staff_data = {} if to_create: created_ids = created_ids or [c.id for c in to_create] resource_map_data = import_get_resource_map_redis_data("staff", created_ids) remapped_data = {d["id"]: d["resource_map"] for d in resource_map_data} for staff in to_create: for column in redis_staff_column: match column: case "shared_with_department" | "avoid_concurrency" | "constraint_profile"|"allocated_activity_ids": redis_staff_data[column] = [] case "resource_map": redis_staff_data[column] = remapped_data.get(staff.id, {}) case "usage_preference_id" | "availability_id" | "start_preference_id": # jose request if no id, dun store null, want to store "0" value = getattr(staff, column) redis_staff_data[column] = 0 if value in (None,"") else value case _: redis_staff_data[column] = getattr(staff, column) redis_data["insert"].setdefault(redis_staff_table, []).append(redis_staff_data) redis_staff_data={} if to_update: for staff in to_update: redis_staff_data = {"id": getattr(staff, "id")} # every new staff reset the redis_staff_data and append to redis_data['update'] for column in update_redis_staff_column: if column in updatable_fields: redis_staff_data[column] = getattr(staff, column) if column in ["usage_preference_id","availability_id","start_preference_id"]: # jose request if no id, dun store null, want to store "0" value = getattr(staff, column) redis_staff_data[column] = 0 if value in (None,"") else value if len(redis_staff_data)>1:# > 1 means not only id redis_data["update"].setdefault(redis_staff_table, []).append(redis_staff_data) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) # kafka self.kafka_response(request, "staff", to_create, to_update, updatable_fields) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_location(self, request, header, rows, settings): slot_per_week = settings.get("slot_per_week") rules = TtLocation.IMPORT_RULES 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 = {} dept_code_candidates = set() zone_code_candidates = set() availability_code_candidates = set() start_preference_code_candidates = set() usage_preference_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "code_required"), }) continue code = data["code"].strip() 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, field_rules in rules.items(): if "integer" in field_rules: raw_value = data.get(field) # calculate default period hours_per_week = TtLocation.WORK_TIME.get("hours_per_day") * TtLocation.WORK_TIME.get("days_per_week") default_period = calculate_default_period(hours_per_week, settings.get("minute_per_slot")) if raw_value in ["", None]: if field in ["maximum_period", "contract_period"]: data[field] = default_period else: data[field] = 0 if "required" in field_rules else None continue try: data[field] = int(raw_value) except (ValueError, TypeError): errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_integer", value=raw_value), }) if field in ["maximum_period", "contract_period"]: data[field] = default_period else: data[field] = None if "department_id" in header and data.get("department_id"): dept_code_candidates.add(data["department_id"].strip()) if "zone_id" in header and data.get("zone_id"): zone_code_candidates.add(data["zone_id"].strip()) if "availability_id" in header and data.get("availability_id"): availability_code_candidates.add(data["availability_id"].strip()) if "start_preference_id" in header and data.get("start_preference_id"): start_preference_code_candidates.add(data["start_preference_id"].strip()) if "usage_preference_id" in header and data.get("usage_preference_id"): usage_preference_code_candidates.add(data["usage_preference_id"].strip()) data["status"] = TtLocation.STATUS_TO_CODE["active"] data["created_by"] = request.user.id staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) dept_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) zone_code_to_id = fetch_code_id_map(TtZone, "code", zone_code_candidates) availability_code_to_id = fetch_code_id_map(TtAvailability, "code", availability_code_candidates) start_preference_code_to_id = fetch_code_id_map(TtStartPreference, "code", start_preference_code_candidates) usage_preference_code_to_id = fetch_code_id_map(TtUsagePreference, "code", usage_preference_code_candidates) cleaned = [] for row_num, data in staged: dep_code = (data.get("department_id") or "").strip() zone_code = (data.get("zone_id") or "").strip() if "department_id" in header: if dep_code: dept_id = dept_code_to_id.get(dep_code) if dept_id is None: errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="department code", value=dep_code), }) data["department_id"] = None else: data["department_id"] = dept_id else: data["department_id"] = None if "zone_id" in header: if zone_code: zone_id = zone_code_to_id.get(zone_code) if zone_id is None: errors.append({ "row": row_num, "field": "zone_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="zone code", value=zone_code), }) data["zone_id"] = None else: data["zone_id"] = zone_id else: data["zone_id"] = None if "desc" in header and not (data.get("desc") or "").strip(): data["desc"] = None raw_value = data.get("shared_with_all_department") parsed = self.parse_import_bool(raw_value) if raw_value not in ["", None] and str(raw_value).strip().lower() not in ["1", "0", "true", "false"]: errors.append({ "row": row_num, "field": "shared_with_all_department", "error": __(self.BASE_TRANS + "invalid_boolean", value=raw_value), }) data["shared_with_all_department"] = parsed self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="availability_id", pattern_field="availability_pattern", code_to_id_map=availability_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="availability", ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="start_preference_id", pattern_field="start_preference_pattern", code_to_id_map=start_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="start_preference", ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="usage_preference_id", pattern_field="usage_preference_pattern", code_to_id_map=usage_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="usage_preference", ) 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_qs = TtLocation.objects.filter(code__in=file_codes) existing_map = {obj.code: obj for obj in existing_qs} to_create = [] to_update = [] created_ids = [] inserted = 0 updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status"] ] updatable_fields.append("updated_by") updatable_fields = list(dict.fromkeys(updatable_fields)) for row_num, data in cleaned: code = data["code"] obj = existing_map.get(code) if obj is None: missing_for_insert = [ col for col in required_insert_columns if col not in header or not (data.get(col) or "").strip() ] 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 to_create.append(TtLocation(**data)) else: changed = False for f in updatable_fields: new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue if getattr(obj, f, None) != new_val and f != "updated_by": setattr(obj, f, new_val) changed = True if changed: setattr(obj, "updated_by", request.user.id) to_update.append(obj) else: rows_skipped += 1 from api.services.integration.tracking import track_resource_created, track_resource_queryset track_resource_queryset(to_update) with transaction.atomic(): if to_create: TtLocation.objects.bulk_create(to_create, batch_size=500) for created_location in to_create: track_resource_created(created_location) inserted = len(to_create) if to_update and updatable_fields: TtLocation.objects.bulk_update(to_update, updatable_fields, batch_size=500) updated = len(to_update) weeks_data = list(TtWeek.objects.values_list("id", flat=True)) if weeks_data and to_create: created_ids = [c.id for c in to_create] default_resource_map = get_slot_default_pattern(slot_per_week).get("resource_map", None) create_resource_map_multiple("location", created_ids, weeks_data, default_resource_map) redis_location_table = "location" redis_location_column = [ "id","code","name","department_id","zone_id","capacity","maximum_period", "contract_period", "constraint_profile","avoid_concurrency", "availability_id","availability_pattern","start_preference_id", "start_preference_pattern","usage_preference_id","usage_preference_pattern","shared_with_all_department", "shared_with_department","resource_map","allocated_activity_ids" ] update_redis_location_column = [ "name","department_id","capacity","zone_id","maximum_period", "contract_period", "constraint_profile","avoid_concurrency", "availability_id","availability_pattern","start_preference_id", "start_preference_pattern","usage_preference_id","usage_preference_pattern","shared_with_all_department", "shared_with_department" ] redis_data = { "insert": {}, "update": {}, } if to_create: created_ids = created_ids or [c.id for c in to_create] resource_map_data = import_get_resource_map_redis_data("location", created_ids) remapped_data = {d["id"]: d["resource_map"] for d in resource_map_data} for location in to_create: redis_location_data = {} for column in redis_location_column: match column: case "shared_with_department" | "avoid_concurrency" | "constraint_profile"|"allocated_activity_ids": redis_location_data[column] = [] case "resource_map": redis_location_data[column] = remapped_data.get(location.id, {}) case "usage_preference_id" | "availability_id" | "start_preference_id": value = getattr(location, column) redis_location_data[column] = 0 if value in (None, "") else value case _: redis_location_data[column] = getattr(location, column) redis_data["insert"].setdefault(redis_location_table, []).append(redis_location_data) if to_update: for location in to_update: redis_location_data = {"id": getattr(location, "id")} for column in update_redis_location_column: if column in updatable_fields: redis_location_data[column] = getattr(location, column) if column in ["usage_preference_id", "availability_id", "start_preference_id"]: value = getattr(location, column) redis_location_data[column] = 0 if value in (None, "") else value if len(redis_location_data) > 1: redis_data["update"].setdefault(redis_location_table, []).append(redis_location_data) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) self.kafka_response(request, "location", to_create, to_update, updatable_fields) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_student_set(self, request, header, rows, settings): slot_per_week = settings.get("slot_per_week") rules = TtStudentSet.IMPORT_RULES 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 = {} dept_code_candidates = set() academic_term_code_candidates = set() pos_code_candidates = set() availability_code_candidates = set() start_preference_code_candidates = set() usage_preference_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "code_required"), }) continue code = data["code"].strip() 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, field_rules in rules.items(): if "integer" not in field_rules: continue raw_value = data.get(field) if raw_value in ["", None]: if "required" in field_rules: data[field] = 0 else: data[field] = None continue try: data[field] = int(raw_value) except ValueError: if "required" in field_rules: extra_msg = ", default to 0" data[field] = 0 else: extra_msg = ", default to None" data[field] = None errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_integer", value=f"{raw_value}{extra_msg}"), }) if "department_id" in header and data.get("department_id"): dept_code_candidates.add(data["department_id"].strip()) if "academic_term_id" in header and data.get("academic_term_id"): academic_term_code_candidates.add(data["academic_term_id"].strip()) if "pos_id" in header and data.get("pos_id"): pos_code_candidates.add(data["pos_id"].strip()) if "availability_id" in header and data.get("availability_id"): availability_code_candidates.add(data["availability_id"].strip()) if "start_preference_id" in header and data.get("start_preference_id"): start_preference_code_candidates.add(data["start_preference_id"].strip()) if "usage_preference_id" in header and data.get("usage_preference_id"): usage_preference_code_candidates.add(data["usage_preference_id"].strip()) data["status"] = TtStudentSet.STATUS_TO_CODE["active"] data["created_by"] = request.user.id staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) dept_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) academic_term_code_to_id = fetch_code_id_map(TtAcademicTerm, "code", academic_term_code_candidates) pos_code_to_id = fetch_code_id_map(TtPos, "code", pos_code_candidates) availability_code_to_id = fetch_code_id_map(TtAvailability, "code", availability_code_candidates) start_preference_code_to_id = fetch_code_id_map(TtStartPreference, "code", start_preference_code_candidates) usage_preference_code_to_id = fetch_code_id_map(TtUsagePreference, "code", usage_preference_code_candidates) #additional layer for checking pos.academic match academic passed in pos_ids = [pid for pid in pos_code_to_id.values() if pid is not None] pos_academic_term_map = { obj.id: obj.academic_term_id for obj in TtPos.objects.filter(id__in=pos_ids).only("id", "academic_term_id") } file_codes = [data["code"] for _, data in staged] existing_qs = TtStudentSet.objects.filter(code__in=file_codes) existing_map = {obj.code: obj for obj in existing_qs} academic_term_ids = set(v for v in academic_term_code_to_id.values() if v) academic_term_ids.update( academic_term_id for academic_term_id in pos_academic_term_map.values() if academic_term_id ) existing_academic_term_ids = [ obj.academic_term_id for obj in existing_map.values() if obj.academic_term_id ] academic_term_ids.update(existing_academic_term_ids) academic_term_default_map = { obj.id: { "availability": obj.default_availability, "start_preference": obj.default_start, "usage_preference": obj.default_usage, } for obj in TtAcademicTerm.objects .filter(id__in=academic_term_ids) .only("id", "default_availability", "default_start", "default_usage") } cleaned = [] for row_num, data in staged: row_has_error = False obj = existing_map.get(data["code"]) dep_code = (data.get("department_id") or "").strip() academic_term_code = (data.get("academic_term_id") or "").strip() pos_code = (data.get("pos_id") or "").strip() # evaluate conditional required_if rules for field, field_rules in rules.items(): if not isinstance(field_rules, list): continue for rule in field_rules: if isinstance(rule, dict) and rule.get("required_if"): cond = rule.get("required_if") depends_on = cond.get("field") condition = cond.get("condition") depends_val = str(data.get(depends_on) or "").strip() current_field_value = str(data.get(field) or "").strip() if condition == "empty" and depends_val in [None, ""] and current_field_value in [None, ""]: row_has_error = True errors.append({ "row": row_num, "field": field, "error": 'here'+__(self.BASE_TRANS + "missing_required", value=field), }) effective_academic_term_id = None if row_has_error: rows_skipped += 1 continue if "department_id" in header: if dep_code: dept_id = dept_code_to_id.get(dep_code) if dept_id is None: errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="department code", value=dep_code), }) data["department_id"] = None else: data["department_id"] = dept_id else: data["department_id"] = None if "pos_id" in header: if pos_code: pos_id = pos_code_to_id.get(pos_code) if pos_id is None: errors.append({ "row": row_num, "field": "pos_id", "error": __(self.BASE_TRANS + "invalid_required", field="pos code", value=pos_code), }) data["pos_id"] = None row_has_error = True else: data["pos_id"] = pos_id # For an existing student set: # - If the imported POS belongs to the same academic term, update the POS. # - If the POS belongs to a different academic term, show a validation error and keep the existing POS and academic term. # -Other valid fields in the same row, such as name, should still be updated even when the POS is rejected. # For a new student set: # - If only POS is provided, use the POS and derive the academic term from it. # - If both POS and academic term are provided, use the POS academic term and ignore the imported academic term. # - If only academic term is provided, use it and leave POS empty. # - If neither POS nor academic term is provided, return a validation error and skip the row. if obj and data.get("pos_id"): pos_academic_term_id = pos_academic_term_map.get(data["pos_id"]) effective_academic_term_id = obj.academic_term_id if pos_academic_term_id is None: errors.append({ "row": row_num, "field": "pos_id", "error": __( self.BASE_TRANS + "invalid_required", field="POS academic term", value=pos_code, ), }) row_has_error = True elif pos_academic_term_id != obj.academic_term_id: errors.append({ "row": row_num, "field": "pos_id", "error": __( self.BASE_TRANS + "academic_term_pos_mismatch", value=pos_code, ), }) data.pop("pos_id", None) # For a new student set, POS is authoritative and its academic term # overrides any academic term value included in the import row. elif data.get("pos_id"): effective_academic_term_id = pos_academic_term_map.get(data["pos_id"]) if effective_academic_term_id is None: errors.append({ "row": row_num, "field": "pos_id", "error": __( self.BASE_TRANS + "invalid_required", field="POS academic term", value=pos_code, ), }) row_has_error = True else: effective_academic_term_id, has_error = self.resolve_academic_term_id( row_num, obj, academic_term_code, academic_term_code_to_id, errors, self.BASE_TRANS, ) if has_error: effective_academic_term_id = None row_has_error = True data["academic_term_id"] = effective_academic_term_id academic_term_defaults = academic_term_default_map.get(effective_academic_term_id, {}) if "desc" in header and not (data.get("desc") or "").strip(): data["desc"] = None self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="availability_id", pattern_field="availability_pattern", code_to_id_map=availability_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="availability", default_pattern=academic_term_defaults.get("availability"), ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="start_preference_id", pattern_field="start_preference_pattern", code_to_id_map=start_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="start_preference", default_pattern=academic_term_defaults.get("start_preference"), ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="usage_preference_id", pattern_field="usage_preference_pattern", code_to_id_map=usage_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="usage_preference", default_pattern=academic_term_defaults.get("usage_preference"), ) if row_has_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) to_create = [] to_update = [] created_ids = [] inserted = 0 updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status", "academic_term_id"] ] updatable_fields.append("updated_by") updatable_fields = list(dict.fromkeys(updatable_fields)) for row_num, data in cleaned: code = data["code"] obj = existing_map.get(code) if obj is None: missing_for_insert = [ col for col in required_insert_columns if col not in header or data.get(col) in [None, ""] ] 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 to_create.append(TtStudentSet(**data)) else: changed = False for f in updatable_fields: if f == "updated_by": continue if f not in data: continue new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue if getattr(obj, f, None) != new_val: setattr(obj, f, new_val) changed = True if changed: setattr(obj, "updated_by", request.user.id) to_update.append(obj) else: rows_skipped += 1 with transaction.atomic(): if to_create: TtStudentSet.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_update and updatable_fields: TtStudentSet.objects.bulk_update(to_update, updatable_fields, batch_size=500) updated = len(to_update) #insert resource map weeks_data = list(TtWeek.objects.values_list("id", flat=True)) if weeks_data and to_create: created_ids = [c.id for c in to_create] default_resource_map = get_slot_default_pattern(slot_per_week).get("resource_map", None) create_resource_map_multiple("student_set", created_ids, weeks_data, default_resource_map) redis_student_set_table = "student_set" redis_student_set_column = [ "id","code","name","department_id","academic_term_id","pos_id","module_id","planned_size","availability_id","availability_pattern","start_preference_id", "start_preference_pattern","usage_preference_id","usage_preference_pattern","resource_map","allocated_activity_ids" ] update_redis_student_set_column = [ "name","department_id","pos_id","planned_size","module_id","availability_id","availability_pattern","start_preference_id", "start_preference_pattern","usage_preference_id","usage_preference_pattern" ] redis_data = { "insert": {}, "update": {}, } if to_create: created_ids = created_ids or [c.id for c in to_create] resource_map_data = import_get_resource_map_redis_data("student_set", created_ids or []) remapped_data = {d["id"]: d["resource_map"] for d in resource_map_data} for student_set in to_create: redis_student_set_data = {} for column in redis_student_set_column: match column: case "module_id" | "allocated_activity_ids": redis_student_set_data[column] = [] case "resource_map": redis_student_set_data[column] = remapped_data.get(student_set.id, {}) case "usage_preference_id" | "availability_id" | "start_preference_id": value = getattr(student_set, column) redis_student_set_data[column] = 0 if value in (None, "") else value case _: redis_student_set_data[column] = getattr(student_set, column) redis_data["insert"].setdefault(redis_student_set_table, []).append(redis_student_set_data) if to_update: for student_set in to_update: redis_student_set_data = {"id": getattr(student_set, "id")} for column in update_redis_student_set_column: if column in updatable_fields: redis_student_set_data[column] = getattr(student_set, column) if column in ["usage_preference_id", "availability_id", "start_preference_id"]: value = getattr(student_set, column) redis_student_set_data[column] = 0 if value in (None, "") else value if len(redis_student_set_data) > 1: redis_data["update"].setdefault(redis_student_set_table, []).append(redis_student_set_data) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) self.kafka_response(request, "student_set", to_create, to_update, updatable_fields) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_student_set_module(self, request, header, rows): rules = TtStudentSet.STUDENT_SET_MODULE.get("IMPORT_RULES", None) rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] required_columns = ["student_set_id", "module_id"] for col in required_columns: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col) }] ) staged = [] student_set_code_candidates = set() module_code_candidates = set() # 1. read file rows for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value row_has_error = False if not (data.get("student_set_id") or "").strip(): errors.append({ "row": row_num, "field": "student_set_id", "error": __(self.BASE_TRANS + "missing_required", value="student_set_id"), }) row_has_error = True if not (data.get("module_id") or "").strip(): errors.append({ "row": row_num, "field": "module_id", "error": __(self.BASE_TRANS + "missing_required", value="module_id"), }) row_has_error = True if row_has_error: rows_skipped += 1 continue student_set_code_candidates.add(data["student_set_id"]) module_code_candidates.add(data["module_id"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) # 2. map code -> id student_set_code_to_id = fetch_code_id_map(TtStudentSet, "code", student_set_code_candidates) module_code_to_id = fetch_code_id_map(TtModule, "code", module_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False student_set_code = (data.get("student_set_id") or "").strip() module_code = (data.get("module_id") or "").strip() student_set_id = student_set_code_to_id.get(student_set_code) module_id = module_code_to_id.get(module_code) if student_set_id is None: errors.append({ "row": row_num, "field": "student_set_id", "error": __(self.BASE_TRANS + "invalid_required", field="student set code", value=student_set_code), }) row_has_error = True else: data["student_set_id"] = student_set_id if module_id is None: errors.append({ "row": row_num, "field": "module_id", "error": __(self.BASE_TRANS + "invalid_required", field="module code", value=module_code), }) row_has_error = True else: data["module_id"] = module_id if row_has_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) # 3. preload student_set -> pos -> pos modules student_set_ids = list({data["student_set_id"] for _, data in cleaned}) student_set_qs = ( TtStudentSet.objects .filter(id__in=student_set_ids) .select_related("pos") .prefetch_related( "pos__ttposmodulegroup_set__ttposmodulegroupmodule_set" ) ) # map: # student_set_id -> { # "pos_id": ..., # "module_ids": {...} # } student_set_pos_module_map = {} for student_set in student_set_qs: pos_id = getattr(student_set, "pos_id", None) module_ids = set() if student_set.pos_id and student_set.pos: for pmg in student_set.pos.ttposmodulegroup_set.all(): for pmgm in pmg.ttposmodulegroupmodule_set.all(): if pmgm.module_id: module_ids.add(pmgm.module_id) student_set_pos_module_map[student_set.id] = { "pos_id": pos_id, "module_ids": module_ids, } # 4. validate module is in student_set.pos validated = [] for row_num, data in cleaned: student_set_id = data["student_set_id"] module_id = data["module_id"] pos_data = student_set_pos_module_map.get(student_set_id) if not pos_data: errors.append({ "row": row_num, "field": "student_set_id", "error": __(self.BASE_TRANS + "invalid_required", field="student set id", value=student_set_id), }) rows_skipped += 1 continue if not pos_data["pos_id"]: errors.append({ "row": row_num, "field": "student_set_id", "error": __(self.BASE_TRANS + "invalid_required", field="student set pos", value="student set has no pos assigned"), }) rows_skipped += 1 continue if module_id not in pos_data["module_ids"]: errors.append({ "row": row_num, "field": "module_id", "error": __(self.BASE_TRANS + "invalid_required", field="module code", value="module not found in student set pos"), }) rows_skipped += 1 continue validated.append((row_num, data)) if not validated: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) # 5. remove duplicate rows inside file unique_pairs = set() deduped = [] for row_num, data in validated: pair = (data["student_set_id"], data["module_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) # 6. check existing relation in db student_set_ids = [data["student_set_id"] for _, data in deduped] module_ids = [data["module_id"] for _, data in deduped] existing_pairs = set( TtStudentSetModule.objects .filter( student_set_id__in=student_set_ids, module_id__in=module_ids, ) .values_list("student_set_id", "module_id") ) to_create = [] for row_num, data in deduped: pair = (data["student_set_id"], data["module_id"]) if pair in existing_pairs: rows_skipped += 1 continue create_data = { "student_set_id": data["student_set_id"], "module_id": data["module_id"], } to_create.append(TtStudentSetModule(**create_data)) # 7. insert with transaction.atomic(): if to_create: TtStudentSetModule.objects.bulk_create(to_create, batch_size=2000) inserted = len(to_create) if to_create: redis_student_set_table = "student_set" affected_student_set_ids = list({obj.student_set_id for obj in to_create}) student_set_module_rows = ( TtStudentSetModule.objects .filter(student_set_id__in=affected_student_set_ids) .values("student_set_id", "module_id") ) student_set_module_map = {} for row in student_set_module_rows: ss_id = row["student_set_id"] mod_id = row["module_id"] student_set_module_map.setdefault(ss_id, []).append(mod_id) redis_data = { "insert": {}, "update": {} } grouped_data = [] kafka_data = [] for student_set_id, module_ids in student_set_module_map.items(): grouped_data.append({ "id": student_set_id, "module_id": sorted(list(set(module_ids))) }) kafka_data.append({ "id": student_set_id, "module": sorted(list(set(module_ids))) }) redis_data["update"].setdefault(redis_student_set_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "student_set_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "student_set": kafka_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_student(self, request, header, rows): rules = TtStudent.IMPORT_RULES 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 = {} dept_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules: if field in header: value = row.get(field) data[field] = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) code = data.get("code", "").strip() if not code: rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "code_required") }) continue 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 # collect FK candidates if "department_id" in header and data.get("department_id"): dept_code_candidates.add(data["department_id"].strip()) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) dept_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) cleaned = [] for row_num, data in staged: dep_code = (data.get("department_id") or "").strip() if "department_id" in header: if dep_code: dept_id = dept_code_to_id.get(dep_code) if dept_id is None: matched_department = ( TtDepartment.objects .filter(code__iexact=dep_code) .only("id") .first() ) if matched_department is not None: dept_id = matched_department.id if dept_id is None and dep_code.isdigit(): numeric_department_id = int(dep_code) if TtDepartment.objects.filter(pk=numeric_department_id).exists(): dept_id = numeric_department_id if dept_id is None: errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="department code", value=dep_code), }) data["department_id"] = None else: data["department_id"] = dept_id else: data["department_id"] = None for nullable_field in ["email", "desc", *[f"extra_data_{i}" for i in range(1, 11)]]: if nullable_field in header and not (data.get(nullable_field) or "").strip(): data[nullable_field] = None cleaned.append((row_num, data)) if not cleaned: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) existing_map = { obj.code: obj for obj in TtStudent.objects.filter(code__in=[data["code"] for _, data in cleaned]) } to_create = [] to_update = [] inserted = 0 updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status"] ] updatable_fields.append("updated_by") updatable_fields = list(dict.fromkeys(updatable_fields)) bulk_update_fields = [ field for field in updatable_fields if field != "department_id" ] for row_num, data in cleaned: obj = existing_map.get(data["code"]) if obj is None: missing_for_insert = [ col for col in required_insert_columns if col not in header or data.get(col) in [None, ""] ] 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 data["status"] = TtStudent.STATUS_TO_CODE["active"] data["created_by"] = request.user.id to_create.append(TtStudent(**data)) else: changed = False for f in updatable_fields: if f == "updated_by": continue if f not in data: continue new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue if getattr(obj, f, None) != new_val: setattr(obj, f, new_val) changed = True if changed: setattr(obj, "updated_by", request.user.id) to_update.append(obj) else: rows_skipped += 1 with transaction.atomic(): if to_create: TtStudent.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_update and bulk_update_fields: TtStudent.objects.bulk_update(to_update, bulk_update_fields, batch_size=500) updated = len(to_update) if to_update and "department_id" in header: for student in to_update: TtStudent.objects.filter(pk=student.pk).update( department_id=student.department_id ) redis_student_table = "student" redis_student_column = [ "id", "code", "name", "email", "department_id", "academic_term_ids", "pos_ids", "pos_module", "student_set_ids", "resource_map", ] update_redis_student_column = ["name", "email", "department_id"] redis_data = { "insert": {}, "update": {}, } if to_create: created_ids = [student.id for student in to_create] for student in to_create: redis_student_data = {} for column in redis_student_column: match column: case "academic_term_ids" | "pos_ids" | "pos_module" | "student_set_ids" | "resource_map": redis_student_data[column] = [] case _: redis_student_data[column] = getattr(student, column) redis_data["insert"].setdefault(redis_student_table, []).append(redis_student_data) if to_update: for student in to_update: redis_student_data = {"id": student.id} for column in update_redis_student_column: if column in updatable_fields: redis_student_data[column] = getattr(student, column) if len(redis_student_data) > 1: redis_data["update"].setdefault(redis_student_table, []).append(redis_student_data) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) self.kafka_response(request, "student", to_create, to_update, updatable_fields) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_student_academic_term(self, request, header, rows, settings): rules = TtStudent.STUDENT_ACADEMIC_TERM.get("IMPORT_RULES", None) rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] required_columns = ["student_id", "academic_term_id"] for col in required_columns: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col), }] ) staged = [] student_code_candidates = set() academic_term_code_candidates = set() # 1. read file rows for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value row_has_error = False for field in required_columns: if not (data.get(field) or "").strip(): 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 student_code_candidates.add(data["student_id"]) academic_term_code_candidates.add(data["academic_term_id"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) # 2. map code -> id student_code_to_id = fetch_code_id_map(TtStudent, "code", student_code_candidates) academic_term_code_to_id = fetch_code_id_map( TtAcademicTerm, "code", academic_term_code_candidates ) cleaned = [] for row_num, data in staged: row_has_error = False student_code = (data.get("student_id") or "").strip() academic_term_code = (data.get("academic_term_id") or "").strip() student_id = student_code_to_id.get(student_code) academic_term_id = academic_term_code_to_id.get(academic_term_code) if student_id is None: errors.append({ "row": row_num, "field": "student_id", "error": __(self.BASE_TRANS + "invalid_required", field="student code", value=student_code), }) row_has_error = True else: data["student_id"] = student_id if academic_term_id is None: errors.append({ "row": row_num, "field": "academic_term_id", "error": __(self.BASE_TRANS + "invalid_required", field="academic term code", value=academic_term_code), }) row_has_error = True else: data["academic_term_id"] = academic_term_id if row_has_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) # 3. remove duplicate rows inside file unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["student_id"], data["academic_term_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) # 4. check existing relation in db student_ids = [data["student_id"] for _, data in deduped] academic_term_ids = [data["academic_term_id"] for _, data in deduped] existing_pairs = set( TtStudentAcademicTerm.objects .filter( student_id__in=student_ids, academic_term_id__in=academic_term_ids, ) .values_list("student_id", "academic_term_id") ) to_create = [] for row_num, data in deduped: pair = (data["student_id"], data["academic_term_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtStudentAcademicTerm( student_id=data["student_id"], academic_term_id=data["academic_term_id"], )) # 5. insert slot_per_week = int(settings["slot_per_week"]) default_resource_map_pattern = "0" * slot_per_week if slot_per_week > 0 else None affected_student_ids = list({obj.student_id for obj in to_create}) if to_create else [] with transaction.atomic(): if to_create: TtStudentAcademicTerm.objects.bulk_create(to_create, batch_size=2000) inserted = len(to_create) if to_create and default_resource_map_pattern: affected_academic_term_ids = list({obj.academic_term_id for obj in to_create}) term_weeks = {} week_rows = ( TtWeek.objects .filter(ttacademicterm__id__in=affected_academic_term_ids) .order_by("week") .distinct() .values("id", "ttacademicterm__id") ) for row in week_rows: term_weeks.setdefault(row["ttacademicterm__id"], set()).add(row["id"]) resource_map_rows = [] for relation in to_create: for week_id in term_weeks.get(relation.academic_term_id, set()): resource_map_rows.append( TtStudentResourceMap( student_id=relation.student_id, week_id=week_id, pattern=default_resource_map_pattern, ) ) if resource_map_rows: TtStudentResourceMap.objects.bulk_create( resource_map_rows, batch_size=2000, ignore_conflicts=True, ) # 6. sync Redis and notify the student service with complete relation lists if to_create: relation_rows = ( TtStudentAcademicTerm.objects .filter(student_id__in=affected_student_ids) .values("student_id", "academic_term_id") ) academic_term_map = {} for relation in relation_rows: academic_term_map.setdefault(relation["student_id"], []).append( relation["academic_term_id"] ) redis_student_table = "student" resource_map_data = import_get_resource_map_redis_data("student", affected_student_ids) remapped_data = {data["id"]: data["resource_map"] for data in resource_map_data} redis_data = {"update": {}} for student_id in affected_student_ids: redis_student_data = { "id": student_id, "academic_term_ids": sorted(set(academic_term_map.get(student_id, []))), "resource_map": remapped_data.get(student_id, []), } redis_data["update"].setdefault(redis_student_table, []).append(redis_student_data) if redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) kafka_data = [ { "id": student_id, "academic_term": sorted(list(set(academic_term_ids))), } for student_id, academic_term_ids in academic_term_map.items() ] kafka_topic = self.kafka_config["MICROSERVICES_TT_TOPIC"] if kafka_topic: send_request( kafka_topic, { "session_id": request.user.name, "student": kafka_data, }, None, "student_update", ) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors, ) def import_tt_student_pos(self, request, header, rows): rules = TtStudent.STUDENT_POS.get("IMPORT_RULES", None) rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] required_columns = ["student_id", "pos_id"] for col in required_columns: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col), }] ) staged = [] student_code_candidates = set() pos_code_candidates = set() # 1. read file rows for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field in header: value = row.get(field) data[field] = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) row_has_error = False for field in required_columns: if not (data.get(field) or "").strip(): 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 student_code_candidates.add(data["student_id"]) pos_code_candidates.add(data["pos_id"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) setting_params = { "slot_per_week", } settings = TtSetting.get_multiple_setting(setting_params) slot_per_week = int(settings["slot_per_week"]) # 2. map code -> id student_code_to_id = fetch_code_id_map(TtStudent, "code", student_code_candidates) pos_code_to_id = fetch_code_id_map(TtPos, "code", pos_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False student_code = data["student_id"] pos_code = data["pos_id"] student_id = student_code_to_id.get(student_code) pos_id = pos_code_to_id.get(pos_code) if student_id is None: errors.append({ "row": row_num, "field": "student_id", "error": __(self.BASE_TRANS + "invalid_required", field="student code", value=student_code), }) row_has_error = True else: data["student_id"] = student_id if pos_id is None: errors.append({ "row": row_num, "field": "pos_id", "error": __(self.BASE_TRANS + "invalid_required", field="POS code", value=pos_code), }) row_has_error = True else: data["pos_id"] = pos_id data["_pos_code"] = pos_code if row_has_error: rows_skipped += 1 continue cleaned.append((row_num, data)) # 3. preload student academic term and pos academic term for validation student_ids = {data["student_id"] for _, data in cleaned} pos_ids = {data["pos_id"] for _, data in cleaned} student_academic_term_map = {} student_academic_term_rows = ( TtStudentAcademicTerm.objects .filter(student_id__in=student_ids) .values_list("student_id", "academic_term_id") ) for student_id, academic_term_id in student_academic_term_rows: student_academic_term_map.setdefault(student_id, set()).add( academic_term_id ) pos_academic_term_map = dict( TtPos.objects .filter(id__in=pos_ids) .values_list("id", "academic_term_id") ) # 4. validate that the pos academic term matches the student academic term validated = [] for row_num, data in cleaned: student_academic_term_ids = student_academic_term_map.get( data["student_id"], set() ) pos_academic_term_id = pos_academic_term_map.get(data["pos_id"]) if ( pos_academic_term_id is None or pos_academic_term_id not in student_academic_term_ids ): errors.append({ "row": row_num, "field": "pos_id", "error": __( self.BASE_TRANS + "invalid_required", field="pos code", value=data["_pos_code"] ), }) rows_skipped += 1 continue validated.append((row_num, data)) if not validated: return self.import_return( rows_read=rows_read, skipped=rows_skipped, errors=errors, ) # 5. remove duplicate rows inside file unique_pairs = set() deduped = [] for row_num, data in validated: pair = (data["student_id"], data["pos_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) # 6. enforce one imported POS per student per academic term # Keep the first row and mark following rows in the same term as skipped with error. accepted = [] imported_student_term_first_row = {} for row_num, data in deduped: pos_academic_term_id = pos_academic_term_map.get(data["pos_id"]) student_term_key = (data["student_id"], pos_academic_term_id) first_row = imported_student_term_first_row.get(student_term_key) if first_row is not None: rows_skipped += 1 errors.append({ "row": row_num, "field": "pos_id", "error": __( self.BASE_TRANS + "one_pos_per_academic_term", value=data.get("_pos_code", data["pos_id"]), first_row=first_row, ), }) continue imported_student_term_first_row[student_term_key] = row_num accepted.append((row_num, data, pos_academic_term_id)) if not accepted: return self.import_return( rows_read=rows_read, skipped=rows_skipped, errors=errors, ) # 7. replace existing student-pos in the same academic term when importing a new POS. accepted_student_ids = {data["student_id"] for _, data, _ in accepted} accepted_term_ids = {term_id for _, _, term_id in accepted if term_id is not None} existing_rows = ( TtStudentPos.objects .filter( student_id__in=accepted_student_ids, pos__academic_term_id__in=accepted_term_ids, ) .values("id", "student_id", "pos_id", "pos__academic_term_id") ) existing_by_student_term = {} for row in existing_rows: key = (row["student_id"], row["pos__academic_term_id"]) existing_by_student_term.setdefault(key, []).append(row) to_create = [] delete_student_pos_ids = set() deleted_pos_ids = set() deleted_student_ids = set() for row_num, data, pos_academic_term_id in accepted: key = (data["student_id"], pos_academic_term_id) existing_same_term_rows = existing_by_student_term.get(key, []) has_same_pair = any( existing_row["pos_id"] == data["pos_id"] for existing_row in existing_same_term_rows ) for existing_row in existing_same_term_rows: if existing_row["pos_id"] == data["pos_id"]: continue delete_student_pos_ids.add(existing_row["id"]) deleted_pos_ids.add(existing_row["pos_id"]) deleted_student_ids.add(existing_row["student_id"]) if has_same_pair: rows_skipped += 1 continue to_create.append( TtStudentPos( student_id=data["student_id"], pos_id=data["pos_id"], ) ) # this param just for affected student set student relation deleted_relation_student_set_ids = None deleted_relation_student_ids = None # 8. apply replacement deletes and inserts. with transaction.atomic(): if delete_student_pos_ids and deleted_pos_ids and deleted_student_ids: TtStudentPosModule.objects.filter( student_id__in=deleted_student_ids, pos_module_group_module__pos_module_group__pos_id__in=deleted_pos_ids, ).delete() TtStudentPathway.objects.filter( student_id__in=deleted_student_ids, pathway__pos_id__in=deleted_pos_ids, ).delete() deleted_student_set_student_relation = set( TtStudentSetStudent.objects.filter( student_id__in=deleted_student_ids, student_set__pos_id__in=deleted_pos_ids, ) ) deleted_relation_student_set_ids = {sss.student_set_id for sss in deleted_student_set_student_relation} deleted_relation_student_ids = {sss.student_id for sss in deleted_student_set_student_relation} # if got student set ids, means got relation need to delete if deleted_relation_student_set_ids and deleted_relation_student_ids: TtStudentSetStudent.objects.filter( student_id__in=deleted_student_ids, student_set__pos_id__in=deleted_pos_ids, ).delete() TtStudentPos.objects.filter(id__in=delete_student_pos_ids).delete() if to_create: TtStudentPos.objects.bulk_create(to_create, batch_size=2000) inserted = len(to_create) redis_data = {"update": {}} # if got remove relation for student set student, need recalculate resource map for related student if deleted_relation_student_set_ids and deleted_relation_student_ids: week_ids = list( TtWeek.objects.filter( ttacademicterm__ttstudentset__id__in=deleted_relation_student_set_ids ) .values_list("id", flat=True) .distinct() ) student_ids = list(deleted_relation_student_ids) redis_data = helper_recalculate_and_get_redis_resource_map( redis_data=redis_data, student_ids=list(student_ids), week_ids=list(week_ids), slot_per_week=slot_per_week, ) # 9. sync Redis and Kafka with latest POS state after insert/delete. if to_create or delete_student_pos_ids: affected_student_ids = list({obj.student_id for obj in to_create} | deleted_student_ids) students = ( TtStudent.objects .filter(id__in=affected_student_ids) .prefetch_related("pos", "student_set") ) pos_module_rows = ( TtStudentPosModule.objects .filter(student_id__in=affected_student_ids) .select_related( "pos_module_group_module__pos_module_group", "pos_module_group_module__module", ) .order_by( "student_id", "pos_module_group_module__pos_module_group__pos_id", "pos_module_group_module__module_id", "pos_module_group_module_id", ) .values( "student_id", "pos_module_group_module_id", "pos_module_group_module__pos_module_group__pos_id", "pos_module_group_module__pos_module_group__module_group_id", "pos_module_group_module__module_id", ) ) student_pos_module_map = {} for row in pos_module_rows: student_pos_module_map.setdefault(row["student_id"], []).append({ "pos_module_group_module_id": row["pos_module_group_module_id"], "pos_id": row["pos_module_group_module__pos_module_group__pos_id"], "module_group_id": row["pos_module_group_module__pos_module_group__module_group_id"], "module_id": row["pos_module_group_module__module_id"], }) redis_student_table = "student" kafka_data = [] for student in students: pos_ids = sorted(pos.id for pos in student.pos.all()) student_set_ids = sorted(student_set.id for student_set in student.student_set.all()) pos_modules = student_pos_module_map.get(student.id, []) redis_student_data = { "id": student.id, "pos_ids": pos_ids, "pos_module": [ {"module_id": module["module_id"], "pos_id": module["pos_id"]} for module in pos_modules ], "student_set_ids": student_set_ids, } redis_data["update"].setdefault(redis_student_table, []).append(redis_student_data) kafka_data.append({ "id": student.id, "pos": pos_ids, "pos_module": pos_modules, "student_set": student_set_ids, }) kafka_topic = self.kafka_config["MICROSERVICES_TT_TOPIC"] if kafka_topic: send_request( kafka_topic, { "session_id": request.user.name, "student": kafka_data, }, None, "student_update" ) if redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors, ) def import_tt_student_pos_module(self, request, header, rows): rules = TtStudent.STUDENT_POS_MODULE.get("IMPORT_RULES", None) rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] required_columns = ["student_id", "pos_module_group_id", "module_id"] for col in required_columns: if col not in header: return self.import_return(errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col), }]) staged = [] student_code_candidates = set() pos_module_group_code_candidates = set() module_code_candidates = set() # 1. read file rows for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) data[field] = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) row_has_error = False for field in required_columns: if not (data.get(field) or "").strip(): 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 student_code_candidates.add(data["student_id"]) pos_module_group_code_candidates.add(data["pos_module_group_id"]) module_code_candidates.add(data["module_id"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) # 2. map code -> id and the POS-module bridge relation student_code_to_id = fetch_code_id_map(TtStudent, "code", student_code_candidates) bridge_rows = ( TtPosModuleGroupModule.objects .filter( pos_module_group__code__in=pos_module_group_code_candidates, module__code__in=module_code_candidates, ) .values( "id", "pos_module_group__code", "pos_module_group__pos_id", "module__code", ) ) bridge_map = { (row["pos_module_group__code"], row["module__code"]): { "id": row["id"], "pos_id": row["pos_module_group__pos_id"], } for row in bridge_rows } cleaned = [] for row_num, data in staged: row_has_error = False student_code = data["student_id"] group_code = data["pos_module_group_id"] module_code = data["module_id"] student_id = student_code_to_id.get(student_code) bridge = bridge_map.get((group_code, module_code)) if student_id is None: errors.append({ "row": row_num, "field": "student_id", "error": __(self.BASE_TRANS + "invalid_required", field="student code", value=student_code), }) row_has_error = True else: data["student_id"] = student_id if bridge is None: errors.append({ "row": row_num, "field": "pos_module_group_id,module_id", "error": __(self.BASE_TRANS + "invalid_required", field="pos_module_group_id", value=f"{group_code}"), }) row_has_error = True else: data["pos_module_group_module_id"] = bridge["id"] data["pos_id"] = bridge["pos_id"] if row_has_error: rows_skipped += 1 continue cleaned.append((row_num, data)) # 3. validate the bridge POS is assigned to the student student_ids = {data["student_id"] for _, data in cleaned} pos_ids = {data["pos_id"] for _, data in cleaned} student_pos_pairs = set( TtStudentPos.objects .filter(student_id__in=student_ids, pos_id__in=pos_ids) .values_list("student_id", "pos_id") ) validated = [] for row_num, data in cleaned: if (data["student_id"], data["pos_id"]) not in student_pos_pairs: errors.append({ "row": row_num, "field": "pos_module_group_id", "error": __(self.BASE_TRANS + "invalid_required", field="student POS", value="POS is not assigned to this student"), }) rows_skipped += 1 continue validated.append((row_num, data)) # 4. remove duplicate rows inside file unique_pairs = set() deduped = [] for row_num, data in validated: pair = (data["student_id"], data["pos_module_group_module_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) # 5. check existing relation in db student_ids = [data["student_id"] for _, data in deduped] bridge_ids = [data["pos_module_group_module_id"] for _, data in deduped] existing_pairs = set( TtStudentPosModule.objects .filter( student_id__in=student_ids, pos_module_group_module_id__in=bridge_ids, ) .values_list("student_id", "pos_module_group_module_id") ) to_create = [] for row_num, data in deduped: pair = (data["student_id"], data["pos_module_group_module_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtStudentPosModule( student_id=data["student_id"], pos_module_group_module_id=data["pos_module_group_module_id"], )) # 6. insert with transaction.atomic(): if to_create: TtStudentPosModule.objects.bulk_create(to_create, batch_size=2000) inserted = len(to_create) # 7. Handle pathway, sync Redis and send to Kafka if to_create: affected_student_ids = list({obj.student_id for obj in to_create}) affected_pos_ids = list({obj.pos_module_group_module.pos_module_group.pos_id for obj in to_create}) student_pos_module_rows = ( TtStudentPosModule.objects .filter(student_id__in=affected_student_ids) .select_related( "pos_module_group_module__pos_module_group", "pos_module_group_module__module", ) .values( "student_id", "pos_module_group_module_id", "pos_module_group_module__pos_module_group__pos_id", "pos_module_group_module__module_id", ) ) student_pos_module_map = {} for row in student_pos_module_rows: student_pos_module_map.setdefault(row["student_id"], {}).setdefault(row["pos_module_group_module__pos_module_group__pos_id"], []).append({ "pos_module_group_module_id": row["pos_module_group_module_id"], "module_id": row["pos_module_group_module__module_id"], }) pathway_string_map = {} pathway_rows = ( TtPathway.objects .filter(pos_id__in=affected_pos_ids) .exclude(pathway_string=None) .values("id", "pos_id", "pathway_string") ) for row in pathway_rows: pathway_string_map.setdefault(row["pos_id"], {})[row["pathway_string"]] = row["id"] student_pathway_rows = [] for student_id, pos_map in student_pos_module_map.items(): for pos_id, pos_module_rows in pos_map.items(): pos_module_group_module_ids = sorted( row["pos_module_group_module_id"] for row in pos_module_rows ) if not pos_module_group_module_ids: continue pathway_string = ";".join(map(str, pos_module_group_module_ids)) pathway_id = pathway_string_map.get(pos_id, {}).get(pathway_string) if pathway_id is None: continue student_pathway_rows.append(TtStudentPathway( student_id=student_id, pathway_id=pathway_id, )) if student_pathway_rows: with transaction.atomic(): TtStudentPathway.objects.bulk_create( student_pathway_rows, batch_size=2000, ignore_conflicts=True, ) students = ( TtStudent.objects .filter(id__in=affected_student_ids) .prefetch_related( "pos_module__module", "pos_module__pos_module_group", ) ) redis_student_table = "student" redis_data = {"update": {}} kafka_data = [] for student in students: pos_modules = list(student.pos_module.all()) redis_student_data = { "id": student.id, "pos_module": [ { "module_id": pos_module.module_id, "pos_id": pos_module.pos_module_group.pos_id, } for pos_module in pos_modules ], } redis_data["update"].setdefault(redis_student_table, []).append(redis_student_data) kafka_data.append({ "id": student.id, "pos_module": [ { "pos_id": pos_module.pos_module_group.pos_id, "module_id": pos_module.module_id, "module_group_id": pos_module.pos_module_group.module_group_id, "pos_module_group_module_id": pos_module.id, } for pos_module in pos_modules ], }) if redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) kafka_topic = self.kafka_config["MICROSERVICES_TT_TOPIC"] if kafka_topic: send_request( kafka_topic, { "session_id": request.user.name, "student": kafka_data, }, None, "student_update" ) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors, ) def import_tt_student_set_student(self, request, header, rows): rules = TtStudent.STUDENT_SET_STUDENT["IMPORT_RULES"] required_columns = ["student_set_id", "student_id"] for column in required_columns: if column not in header: return self.import_return(errors=[{ "row": 1, "field": column, "error": __( self.BASE_TRANS + "missing_required", value=column, ), }]) rows_read = 0 rows_skipped = 0 errors = [] staged = [] student_set_codes = set() student_codes = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules: value = row.get(field) data[field] = ( (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) ) row_has_error = False for field in required_columns: if data.get(field): continue 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 student_set_codes.add(data["student_set_id"]) student_codes.add(data["student_id"]) staged.append((row_num, data)) if not staged: return self.import_return( rows_read=rows_read, skipped=rows_skipped, errors=errors, ) student_set_by_code = { item.code: item for item in TtStudentSet.objects.filter(code__in=student_set_codes) } student_by_code = { item.code: item for item in TtStudent.objects.filter(code__in=student_codes) } student_ids = [item.id for item in student_by_code.values()] student_term_pairs = set( TtStudentAcademicTerm.objects .filter(student_id__in=student_ids) .values_list("student_id", "academic_term_id") ) student_pos_pairs = set( TtStudentPos.objects .filter(student_id__in=student_ids) .values_list("student_id", "pos_id") ) validated = [] seen_pairs = set() for row_num, data in staged: student_set_code = data["student_set_id"] student_code = data["student_id"] student_set = student_set_by_code.get(student_set_code) student = student_by_code.get(student_code) if student_set is None: errors.append({ "row": row_num, "field": "student_set_id", "error": __( self.BASE_TRANS + "invalid_required", field="student set code", value=student_set_code, ), }) if student is None: errors.append({ "row": row_num, "field": "student_id", "error": __( self.BASE_TRANS + "invalid_required", field="student code", value=student_code, ), }) if student_set is None or student is None: rows_skipped += 1 continue if ( student_set.academic_term_id and (student.id, student_set.academic_term_id) not in student_term_pairs ): errors.append({ "row": row_num, "field": "student_set_id", "error": __( self.BASE_TRANS + "invalid_required", field="student set academic term", value=student_set_code, ), }) rows_skipped += 1 continue if ( student_set.pos_id and (student.id, student_set.pos_id) not in student_pos_pairs ): errors.append({ "row": row_num, "field": "student_set_id", "error": __( self.BASE_TRANS + "invalid_required", field="student set POS", value=student_set_code, ), }) rows_skipped += 1 continue pair = (student_set.id, student.id) if pair in seen_pairs: rows_skipped += 1 continue seen_pairs.add(pair) validated.append(pair) existing_pairs = set( TtStudentSetStudent.objects .filter( student_set_id__in=[pair[0] for pair in validated], student_id__in=[pair[1] for pair in validated], ) .values_list("student_set_id", "student_id") ) to_create = [ TtStudentSetStudent(student_set_id=student_set_id, student_id=student_id) for student_set_id, student_id in validated if (student_set_id, student_id) not in existing_pairs ] rows_skipped += len(validated) - len(to_create) with transaction.atomic(): if to_create: TtStudentSetStudent.objects.bulk_create(to_create, batch_size=2000) if to_create: affected_student_ids = {item.student_id for item in to_create} students = ( TtStudent.objects .filter(id__in=affected_student_ids) .prefetch_related("student_set") ) kafka_topic = self.kafka_config["MICROSERVICES_TT_TOPIC"] if kafka_topic: send_request( kafka_topic, { "session_id": request.user.name, "student": [ { "id": student.id, "student_set": sorted( student.student_set.values_list("id", flat=True) ), } for student in students ], }, None, "student_update", ) return self.import_return( rows_read=rows_read, inserted=len(to_create), skipped=rows_skipped, errors=errors, ) def import_tt_pos(self, request, header, rows): rules = TtPos.IMPORT_RULES 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 = {} dept_code_candidates = set() academic_term_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value # code required if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "code_required"), }) continue code = data["code"].strip() 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 # integer handling for field, field_rules in rules.items(): if "integer" not in field_rules: continue raw_value = data.get(field) if raw_value in ["", None]: data[field] = 0 if "required" in field_rules else None continue try: data[field] = int(raw_value) except ValueError: if "required" in field_rules: data[field] = 0 extra_msg = ", default to 0" else: data[field] = None extra_msg = ", default to None" errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_integer", value=f"{raw_value}{extra_msg}"), }) # collect FK candidates if "department_id" in header and data.get("department_id"): dept_code_candidates.add(data["department_id"].strip()) if "academic_term_id" in header and data.get("academic_term_id"): academic_term_code_candidates.add(data["academic_term_id"].strip()) data["status"] = TtPos.STATUS_TO_CODE["active"] data["created_by"] = request.user.id staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) dept_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) academic_term_code_to_id = fetch_code_id_map(TtAcademicTerm, "code", academic_term_code_candidates) file_codes = [data["code"] for _, data in staged] existing_qs = TtPos.objects.filter(code__in=file_codes) existing_map = {obj.code: obj for obj in existing_qs} cleaned = [] for row_num, data in staged: row_has_error = False obj = existing_map.get(data["code"]) dep_code = (data.get("department_id") or "").strip() academic_term_code = (data.get("academic_term_id") or "").strip() effective_academic_term_id = None # department (nullable) if "department_id" in header: if dep_code: dept_id = dept_code_to_id.get(dep_code) if dept_id is None: errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="department code", value=dep_code), }) data["department_id"] = None else: data["department_id"] = dept_id else: data["department_id"] = None # academic_term (required) effective_academic_term_id, has_error = self.resolve_academic_term_id(row_num, obj, academic_term_code, academic_term_code_to_id, errors, self.BASE_TRANS) if has_error: effective_academic_term_id = None row_has_error = True data["academic_term_id"] = effective_academic_term_id # desc normalize if "desc" in header and not (data.get("desc") or "").strip(): data["desc"] = None if row_has_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) to_create = [] to_update = [] inserted = 0 updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status", "academic_term_id"] ] updatable_fields.append("updated_by") updatable_fields = list(dict.fromkeys(updatable_fields)) for row_num, data in cleaned: code = data["code"] obj = existing_map.get(code) if obj is None: missing_for_insert = [ col for col in required_insert_columns if col not in header or data.get(col) in [None, ""] ] 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 to_create.append(TtPos(**data)) else: changed = False for f in updatable_fields: if f == "updated_by": continue if f not in data: continue new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue if getattr(obj, f, None) != new_val: setattr(obj, f, new_val) changed = True if changed: setattr(obj, "updated_by", request.user.id) to_update.append(obj) else: rows_skipped += 1 with transaction.atomic(): if to_create: TtPos.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_update: TtPos.objects.bulk_update(to_update, updatable_fields, batch_size=500) updated = len(to_update) redis_pos_table = "pos" redis_pos_column = [ "id","code","name","desc","department_id","academic_term_id","planned_size","max_credit","module_group_id" ] update_redis_pos_column = [ "name","desc","department_id","planned_size","max_credit" ] redis_data = { "insert": {}, "update": {}, } if to_create: for pos in to_create: redis_pos_data = {} for column in redis_pos_column: match column: case "module_group_id": redis_pos_data[column] = [] case _: redis_pos_data[column] = getattr(pos, column) redis_data["insert"].setdefault(redis_pos_table, []).append(redis_pos_data) if to_update: for pos in to_update: redis_pos_data = {"id": pos.id} for column in update_redis_pos_column: if column in updatable_fields: redis_pos_data[column] = getattr(pos, column) if len(redis_pos_data) > 1: redis_data["update"].setdefault(redis_pos_table, []).append(redis_pos_data) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) self.kafka_response(request, "pos", to_create, to_update, updatable_fields) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_pos_module_group(self, request, header, rows): rules = TtPos.POS_MODULE_GROUP.get("IMPORT_RULES", None) rows_read = 0 rows_skipped = 0 inserted = 0 updated = 0 errors = [] # check missing required columns from IMPORT_RULES for field, field_rules in rules.items(): if "required" in field_rules and field not in header: return self.import_return( errors=[{ "row": 1, "field": field, "error": __(self.BASE_TRANS + "missing_required", value=field) }] ) staged = [] pos_code_candidates = set() module_group_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False # required checking from IMPORT_RULES for field, field_rules in rules.items(): if "required" not in field_rules: continue if not data.get(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 pos_code_candidates.add(data["pos_id"]) module_group_code_candidates.add(data["module_group_id"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) pos_code_to_id = fetch_code_id_map(TtPos, "code", pos_code_candidates) module_group_code_to_id = fetch_code_id_map(TtModuleGroup, "code", module_group_code_candidates) file_codes = [data["code"] for _, data in staged] existing_qs = TtPosModuleGroup.objects.filter(code__in=file_codes) existing_map = {obj.code: obj for obj in existing_qs} cleaned = [] for row_num, data in staged: row_has_error = False pos_code = data["pos_id"] module_group_code = data["module_group_id"] pos_id = pos_code_to_id.get(pos_code) module_group_id = module_group_code_to_id.get(module_group_code) if pos_id is None: errors.append({ "row": row_num, "field": "pos_id", "error": __(self.BASE_TRANS + "invalid_required", field="pos code", value=pos_code), }) row_has_error = True else: data["pos_id"] = pos_id if module_group_id is None: errors.append({ "row": row_num, "field": "module_group_id", "error": __(self.BASE_TRANS + "invalid_required", field="module group code", value=module_group_code), }) row_has_error = True else: data["module_group_id"] = module_group_id if row_has_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) # dedupe same code inside import file unique_codes = set() deduped = [] for row_num, data in cleaned: code = data["code"] if code in unique_codes: rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "duplicate_file", value=code), }) continue unique_codes.add(code) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) pos_ids = [data["pos_id"] for _, data in deduped] module_group_ids = [data["module_group_id"] for _, data in deduped] # check pair conflict, because same pos + module_group should not duplicate under different code existing_pair_rows = ( TtPosModuleGroup.objects .filter( pos_id__in=pos_ids, module_group_id__in=module_group_ids ) .values("code", "pos_id", "module_group_id") ) existing_pair_map = { (row["pos_id"], row["module_group_id"]): row["code"] for row in existing_pair_rows } to_create = [] to_update = [] affected_pos_ids = set() updatable_fields = [ "pos_id", "module_group_id", "updated_by", ] for row_num, data in deduped: code = data["code"] obj = existing_map.get(code) pair = (data["pos_id"], data["module_group_id"]) existing_pair_code = existing_pair_map.get(pair) # pair already used by another code if existing_pair_code and existing_pair_code != code: rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "duplicate_pair", value=f"{data['pos_id']}, {data['module_group_id']}"), }) continue if obj is None: create_data = { "code": data["code"], "pos_id": data["pos_id"], "module_group_id": data["module_group_id"], } # only include if your model has these columns if hasattr(TtPosModuleGroup, "created_by"): create_data["created_by"] = request.user.id to_create.append(TtPosModuleGroup(**create_data)) affected_pos_ids.add(data["pos_id"]) else: changed = False # add old pos_id also, because if relation moved to another pos, # old pos redis also needs refresh affected_pos_ids.add(obj.pos_id) if obj.pos_id != data["pos_id"]: obj.pos_id = data["pos_id"] changed = True if obj.module_group_id != data["module_group_id"]: obj.module_group_id = data["module_group_id"] changed = True if changed: if hasattr(obj, "updated_by"): obj.updated_by = request.user.id to_update.append(obj) affected_pos_ids.add(data["pos_id"]) else: rows_skipped += 1 with transaction.atomic(): if to_create: TtPosModuleGroup.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_update: TtPosModuleGroup.objects.bulk_update(to_update, updatable_fields, batch_size=500) updated = len(to_update) if affected_pos_ids: redis_pos_table = "pos" pos_module_group_rows = ( TtPosModuleGroup.objects .filter(pos_id__in=affected_pos_ids) .values("pos_id", "module_group_id") ) pos_module_group_map = {} for row in pos_module_group_rows: pos_id = row["pos_id"] module_group_id = row["module_group_id"] pos_module_group_map.setdefault(pos_id, []).append(module_group_id) redis_data = { "update": {} } grouped_data = [] kafka_data = [] for pos_id in affected_pos_ids: module_group_ids = pos_module_group_map.get(pos_id, []) grouped_data.append({ "id": pos_id, "module_group_id": sorted(list(set(module_group_ids))) }) kafka_data.append({ "id": pos_id, "module_group": sorted(list(set(module_group_ids))) }) redis_data["update"].setdefault(redis_pos_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "pos_update" kafka_topic = self.kafka_config["MICROSERVICES_TT_TOPIC"] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "pos": kafka_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, updated=updated, skipped=rows_skipped, errors=errors ) def import_tt_module(self, request, header, rows, settings): slot_per_week = settings.get("slot_per_week") rules = TtModule.IMPORT_RULES 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 = {} dept_code_candidates = set() academic_term_code_candidates = set() week_pattern_code_candidates = set() availability_code_candidates = set() start_preference_code_candidates = set() usage_preference_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "code_required"), }) continue code = data["code"].strip() 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, field_rules in rules.items(): if "integer" not in field_rules: continue raw_value = data.get(field) if raw_value in ["", None]: data[field] = 0 continue try: data[field] = int(raw_value) except ValueError: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_integer", value=raw_value), }) data[field] = 0 continue if "department_id" in header and data.get("department_id"): dept_code_candidates.add(data["department_id"].strip()) if "academic_term_id" in header and data.get("academic_term_id"): academic_term_code_candidates.add(data["academic_term_id"].strip()) if "week_pattern_id" in header and data.get("week_pattern_id"): week_pattern_code_candidates.add(data["week_pattern_id"].strip()) if "availability_id" in header and data.get("availability_id"): availability_code_candidates.add(data["availability_id"].strip()) if "start_preference_id" in header and data.get("start_preference_id"): start_preference_code_candidates.add(data["start_preference_id"].strip()) if "usage_preference_id" in header and data.get("usage_preference_id"): usage_preference_code_candidates.add(data["usage_preference_id"].strip()) data["status"] = TtModule.STATUS_TO_CODE["active"] data["created_by"] = request.user.id staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) dept_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) academic_term_code_to_id = fetch_code_id_map(TtAcademicTerm, "code", academic_term_code_candidates) week_pattern_code_to_id = fetch_code_id_map(TtWeekPattern, "code", week_pattern_code_candidates) availability_code_to_id = fetch_code_id_map(TtAvailability, "code", availability_code_candidates) start_preference_code_to_id = fetch_code_id_map(TtStartPreference, "code", start_preference_code_candidates) usage_preference_code_to_id = fetch_code_id_map(TtUsagePreference, "code", usage_preference_code_candidates) codes = list({d["code"] for _, d in staged}) existing_modules = list(TtModule.objects.filter(code__in=codes)) existing_by_code = objects_grouped_by_code(existing_modules) academic_term_week_map = {} academic_term_ids = set(v for v in academic_term_code_to_id.values() if v) existing_academic_term_ids = [ obj.academic_term_id for obj in existing_modules if obj.academic_term_id ] academic_term_ids.update(existing_academic_term_ids) for row in TtAcademicTermWeek.objects.filter( academic_term_id__in=academic_term_ids ).values("academic_term_id", "week_id"): academic_term_week_map.setdefault(row["academic_term_id"], []).append(row["week_id"]) academic_term_default_map = { obj.id: { "availability": obj.default_availability, "start_preference": obj.default_start, "usage_preference": obj.default_usage, } for obj in TtAcademicTerm.objects .filter(id__in=academic_term_ids) .only("id", "default_availability", "default_start", "default_usage") } week_pattern_academic_term_map = { obj.id: obj.academic_term_id for obj in TtWeekPattern.objects .filter(id__in=week_pattern_code_to_id.values()) .only("id", "academic_term_id") } cleaned = [] for row_num, data in staged: row_has_error = False dep_code = (data.get("department_id") or "").strip() academic_term_code = (data.get("academic_term_id") or "").strip() week_pattern_code = (data.get("week_pattern_id") or "").strip() effective_academic_term_id = None existing_for_code = existing_by_code.get(data["code"], []) if len(existing_for_code) > 1: errors.append({ "row": row_num, "field": "code", "error": __( self.BASE_TRANS + "invalid_required", field="globally unique module code", value=data["code"], ), }) rows_skipped += 1 continue obj = existing_for_code[0] if existing_for_code else None effective_academic_term_id, has_error = self.resolve_academic_term_id(row_num, obj, academic_term_code, academic_term_code_to_id, errors, self.BASE_TRANS) if has_error: effective_academic_term_id = None row_has_error = True academic_term_defaults = academic_term_default_map.get(effective_academic_term_id, {}) if "department_id" in header: if dep_code: dept_id = dept_code_to_id.get(dep_code) if dept_id is None: errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="department code", value=dep_code), }) data["department_id"] = None else: data["department_id"] = dept_id else: data["department_id"] = None # week_pattern logic if week_pattern_code: wp_id = week_pattern_code_to_id.get(week_pattern_code) if wp_id: week_pattern_academic_term_id = week_pattern_academic_term_map.get(wp_id) if week_pattern_academic_term_id == effective_academic_term_id: data["week_pattern_id"] = wp_id else: errors.append({ "row": row_num, "field": "week_pattern_id,academic_term_id", "error": __( self.BASE_TRANS + "academic_term_week_pattern_mismatch", value=week_pattern_code ), }) row_has_error = True else: data["week_pattern_id"] = None data["_week_ids"] = academic_term_week_map.get(effective_academic_term_id, []) else: data["week_pattern_id"] = None data["_week_ids"] = academic_term_week_map.get(effective_academic_term_id, []) if "desc" in header and not (data.get("desc") or "").strip(): data["desc"] = None self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="availability_id", pattern_field="availability_pattern", code_to_id_map=availability_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="availability", default_pattern=academic_term_defaults.get("availability"), ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="start_preference_id", pattern_field="start_preference_pattern", code_to_id_map=start_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="start_preference", default_pattern=academic_term_defaults.get("start_preference"), ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="usage_preference_id", pattern_field="usage_preference_pattern", code_to_id_map=usage_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="usage_preference", default_pattern=academic_term_defaults.get("usage_preference"), ) if row_has_error: rows_skipped += 1 continue data["academic_term_id"] = effective_academic_term_id 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 TtModule.objects.filter(code__in=file_codes) } to_create = [] to_update = [] module_week_map = {} inserted = 0 updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status", "academic_term_id"] ] updatable_fields.append("updated_by") updatable_fields = list(dict.fromkeys(updatable_fields)) for row_num, data in cleaned: code = data["code"] module_key = code week_ids = data.pop("_week_ids", []) obj = existing_map.get(module_key) if obj is None: missing_for_insert = [ col for col in required_insert_columns if col not in header or data.get(col) in [None, ""] ] 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 to_create.append(TtModule(**data)) if week_ids: module_week_map[module_key] = week_ids else: changed = False for f in updatable_fields: if f == "updated_by" or f == "academic_term_id": continue if f not in data: continue new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue if getattr(obj, f, None) != new_val: setattr(obj, f, new_val) changed = True if changed: setattr(obj, "updated_by", request.user.id) to_update.append(obj) if week_ids: module_week_map[module_key] = week_ids else: rows_skipped += 1 with transaction.atomic(): if to_create: TtModule.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_update and updatable_fields: TtModule.objects.bulk_update(to_update, updatable_fields, batch_size=500) updated = len(to_update) created_codes = [obj.code for obj in to_create] created_module_map = { obj.code: obj.id for obj in TtModule.objects.filter(code__in=created_codes).only( "id", "code" ) } module_ids = [] pending_module_week_map = {} module_week_response_map = {} for module_key, week_ids in module_week_map.items(): module_id = created_module_map.get(module_key) if not module_id: existing_obj = existing_map.get(module_key) module_id = existing_obj.id if existing_obj else None if not module_id: continue module_ids.append(module_id) pending_module_week_map[module_id] = week_ids or [] existing_module_week_ids = set( TtModuleWeek.objects .filter(module_id__in=module_ids) .values_list("module_id", flat=True) ) module_week_rows = [] for module_id, week_ids in pending_module_week_map.items(): if module_id in existing_module_week_ids: continue module_week_response_map[module_id] = week_ids for week_id in week_ids: module_week_rows.append( TtModuleWeek( module_id=module_id, week_id=week_id ) ) if module_week_rows: TtModuleWeek.objects.bulk_create( module_week_rows, batch_size=2000, ignore_conflicts=True ) redis_module_table = "module" redis_module_column = [ "id","code","name","department_id","academic_term_id","planned_size","real_size","credit_provided", "week_pattern_id","week_pattern","availability_id","availability_pattern","start_preference_id", "start_preference_pattern","usage_preference_id","usage_preference_pattern" ] update_redis_module_column = [ "name","department_id","academic_term_id","planned_size","real_size","credit_provided", "week_pattern", "week_pattern_id","availability_id","availability_pattern","start_preference_id", "start_preference_pattern","usage_preference_id","usage_preference_pattern" ] redis_data = { "insert": {}, "update": {}, } created_relation = {} updated_relation = {} if to_create: for module in to_create: redis_module_data = {} week_pattern_for_redis = [] if module_week_response_map: week_ids = module_week_response_map.get(module.id, []) week_pattern_for_redis = convert_to_redis_week(week_ids) if week_ids else [] created_relation[module.id] = week_ids for column in redis_module_column: match column: case "week_pattern": redis_module_data[column] = week_pattern_for_redis or [] case "usage_preference_id" | "availability_id" | "start_preference_id" | "week_pattern_id": value = getattr(module, column) redis_module_data[column] = 0 if value in (None, "") else value case _: redis_module_data[column] = getattr(module, column) redis_data["insert"].setdefault(redis_module_table, []).append(redis_module_data) if to_update: for module in to_update: redis_module_data = {"id": module.id} week_pattern_for_redis = [] if module_week_response_map: week_ids = module_week_response_map.get(module.id, []) week_pattern_for_redis = convert_to_redis_week(week_ids) if week_ids else [] updated_relation[module.id] = week_ids for column in update_redis_module_column: if column in updatable_fields: redis_module_data[column] = getattr(module, column) if column == "week_pattern": redis_module_data[column] = week_pattern_for_redis or [] if column in ["usage_preference_id", "availability_id", "start_preference_id"]: value = getattr(module, column) redis_module_data[column] = 0 if value in (None, "") else value if len(redis_module_data) > 1: redis_data["update"].setdefault(redis_module_table, []).append(redis_module_data) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) create_method = "module_create" update_method = "module_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] ignore_fields = ["created_at", "updated_at", "created_by", "updated_by"] # CREATE created_kafka_data = [] created_ids = [c.id for c in to_create] if created_ids: created_data = TtModule.objects.filter(id__in=created_ids) for data in created_data: temp = {} for f in TtModule._meta.fields: if f.name in ignore_fields: continue if isinstance(f, ForeignKey): # if is foreign key, dun direct follow field name, cause will call 1 more time to db # eg. field name is department, if using department.id to get the id, will call to db since not prefethch this value, need used department_id value = getattr(data, f"{f.name}_id") if value is not None: temp[f.name] = value else: # move to here to fet value, so if is foreign key, wont come to this part, can skip the get value from FK table problem value = getattr(data, f.name) if value is not None: temp[f.name] = value temp["week"] = created_relation.get(data.id, None) or [] created_kafka_data.append(temp) if created_kafka_data: kafka_request_data = { "session_id": request.user.name, "module": created_kafka_data, } send_request(kafka_topic, kafka_request_data, None, create_method) # UPDATE update_kafka_data = [] update_fields = [f for f in updatable_fields if f != "updated_by"] for data in to_update: temp = {"id": data.id} for up in update_fields: value = getattr(data, up, None) # if value is None: # continue temp[up] = value temp["week"] = updated_relation.get(data.id, None) or [] update_kafka_data.append(temp) if update_kafka_data: kafka_request_data = { "session_id": request.user.name, "module": update_kafka_data, } send_request(kafka_topic, kafka_request_data, None, update_method) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_activity_template(self, request, header, rows, settings): slot_per_week = settings.get("slot_per_week") minute_per_slot = int(settings.get("minute_per_slot")) rules = TtActivityTemplate.IMPORT_RULES errors = [] if "code" not in header: errors.append({ "row": 1, "field": "code", "error": __(self.BASE_TRANS + "missing_required", value="code") }) if errors: self.import_return(errors=errors) 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 seen_codes = {} activity_type_code_candidates = set() dept_code_candidates = set() zone_code_candidates = set() module_code_candidates = set() week_pattern_code_candidates = set() availability_code_candidates = set() start_preference_code_candidates = set() usage_preference_code_candidates = set() for row_num, row in rows: rows_read += 1 row_has_error = False data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value has_staff_requirement_param = "staff_requirement" in header has_location_requirement_param = "location_requirement" in header data["_has_staff_requirement_param"] = has_staff_requirement_param data["_has_location_requirement_param"] = has_location_requirement_param if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "code_required"), }) continue code = data["code"].strip() 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 # strict integer for field, field_rules in rules.items(): if field not in header: continue if "integer" not in field_rules: continue raw_value = data.get(field) if raw_value in ["", None]: data[field] = None continue try: data[field] = int(raw_value) except ValueError: if "required" in field_rules: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_integer", value=raw_value), }) row_has_error = True else: data[field] = None # #staff/location type # staff_req = data.get("staff_requirement") or 0 # location_req = data.get("location_requirement") or 0 # data["staff_requirement_type"] = ( # TtActivityTemplate.STAFF_REQUIREMENT_TYPE["none"] # if staff_req == 0 # else TtActivityTemplate.STAFF_REQUIREMENT_TYPE["wildcard"] # ) # data["location_requirement_type"] = ( # TtActivityTemplate.LOCATION_REQUIREMENT_TYPE["none"] # if location_req == 0 # else TtActivityTemplate.LOCATION_REQUIREMENT_TYPE["wildcard"] # ) if row_has_error: rows_skipped += 1 continue #slot_required if "slot_required" in header: slot_required = int(data.get("slot_required") or 0) data["duration"] = slot_required * minute_per_slot # collect FK candidates for key, container in [ ("activity_type_id", activity_type_code_candidates), ("department_id", dept_code_candidates), ("zone_id", zone_code_candidates), ("module_id", module_code_candidates), ("week_pattern_id", week_pattern_code_candidates), ("availability_id", availability_code_candidates), ("start_preference_id", start_preference_code_candidates), ("usage_preference_id", usage_preference_code_candidates), ]: if key not in header: continue if data.get(key): container.add(data[key]) data["status"] = TtActivityTemplate.STATUS_TO_CODE["active"] data["created_by"] = request.user.id staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) activity_type_code_to_id = fetch_code_id_map(TtActivityType, "code", activity_type_code_candidates) dept_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) zone_code_to_id = fetch_code_id_map(TtZone, "code", zone_code_candidates) module_candidates = list( TtModule.objects .filter(code__in=module_code_candidates) .only("id", "code", "academic_term_id") ) modules_grouped_by_code = objects_grouped_by_code(module_candidates) module_code_to_id = { code: modules[0].id for code, modules in modules_grouped_by_code.items() if len(modules) == 1 } ambiguous_module_codes = { code for code, modules in modules_grouped_by_code.items() if len(modules) > 1 } week_pattern_code_to_id = fetch_code_id_map(TtWeekPattern, "code", week_pattern_code_candidates) availability_code_to_id = fetch_code_id_map(TtAvailability, "code", availability_code_candidates) start_preference_code_to_id = fetch_code_id_map(TtStartPreference, "code", start_preference_code_candidates) usage_preference_code_to_id = fetch_code_id_map(TtUsagePreference, "code", usage_preference_code_candidates) codes = list({d["code"] for _, d in staged}) existing_map = { obj.code: obj for obj in TtActivityTemplate.objects.filter(code__in=codes) } module_ids = [v for v in module_code_to_id.values() if v] module_academic_term_map = { obj.id: obj.academic_term_id for obj in TtModule.objects.filter(id__in=module_ids).only("id", "academic_term_id") } module_department_map = { obj.id: obj.department_id for obj in TtModule.objects.filter(id__in=module_ids).only("id", "department_id") } academic_term_ids = { academic_term_id for academic_term_id in module_academic_term_map.values() if academic_term_id } academic_term_week_map = {} for row in TtAcademicTermWeek.objects.filter( academic_term_id__in=academic_term_ids ).values("academic_term_id", "week_id"): academic_term_week_map.setdefault(row["academic_term_id"], []).append(row["week_id"]) academic_term_default_map = { obj.id: { "availability": obj.default_availability, "start_preference": obj.default_start, "usage_preference": obj.default_usage, } for obj in TtAcademicTerm.objects .filter(id__in=academic_term_ids) .only("id", "default_availability", "default_start", "default_usage") } week_pattern_ids = [v for v in week_pattern_code_to_id.values() if v] week_pattern_academic_term_map = { obj.id: obj.academic_term_id for obj in TtWeekPattern.objects.filter(id__in=week_pattern_ids) } cleaned = [] for row_num, data in staged: row_has_error = False obj = existing_map.get(data["code"]) dep_code = (data.get("department_id") or "").strip() zone_code = (data.get("zone_id") or "").strip() activity_type_code = (data.get("activity_type_id") or "").strip() module_code = (data.get("module_id") or "").strip() wp_code = data.get("week_pattern_id") module_id = module_code_to_id.get(module_code) if module_code else None effective_academic_term_id = module_academic_term_map.get(module_id) if module_id is None: module_error_value = module_code if module_code in ambiguous_module_codes: module_error_value = f"{module_code} (matches multiple academic terms)" errors.append({ "row": row_num, "field": "module_id", "error": __( self.BASE_TRANS + "invalid_required", field="module code", value=module_error_value, ) if module_code else __( self.BASE_TRANS + "missing_required", value="module_id", ), }) row_has_error = True elif effective_academic_term_id is None: errors.append({ "row": row_num, "field": "module_id", "error": __( self.BASE_TRANS + "invalid_required", field="module academic term", value=module_code, ), }) row_has_error = True else: data["module_id"] = module_id academic_term_defaults = academic_term_default_map.get(effective_academic_term_id, {}) if "department_id" in header: if dep_code: dept_id = dept_code_to_id.get(dep_code) if dept_id is None: errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="department code", value=dep_code), }) data["department_id"] = None else: data["department_id"] = dept_id else: data["department_id"] = module_department_map.get(module_id) else: data["department_id"] = module_department_map.get(module_id) if "zone_id" in header: if zone_code: zone_id = zone_code_to_id.get(zone_code) if zone_id is None: errors.append({ "row": row_num, "field": "zone_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="zone code", value=zone_code), }) data["zone_id"] = None else: data["zone_id"] = zone_id else: data["zone_id"] = None if "activity_type_id" in header: if activity_type_code: activity_type_id = activity_type_code_to_id.get(activity_type_code) if activity_type_id is None: errors.append({ "row": row_num, "field": "activity_type_id", "error": __(self.BASE_TRANS + "invalid_required", field="activity type code", value=activity_type_code), }) data["activity_type_id"] = None row_has_error = True else: data["activity_type_id"] = activity_type_id else: errors.append({ "row": row_num, "field": "activity_type_id", "error": __(self.BASE_TRANS + "missing_required", value="activity_type_id"), }) data["activity_type_id"] = None row_has_error = True # week_pattern logic if wp_code: wp_id = week_pattern_code_to_id.get(wp_code) if wp_id: week_pattern_academic_term_id = week_pattern_academic_term_map.get(wp_id) if week_pattern_academic_term_id == effective_academic_term_id: data["week_pattern_id"] = wp_id else: errors.append({ "row": row_num, "field": "week_pattern_id,academic_term_id", "error": __( self.BASE_TRANS + "academic_term_week_pattern_mismatch", value=wp_code ), }) row_has_error = True else: data["week_pattern_id"] = None data["_week_ids"] = academic_term_week_map.get(effective_academic_term_id, []) else: data["week_pattern_id"] = None data["_week_ids"] = academic_term_week_map.get(effective_academic_term_id, []) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="availability_id", pattern_field="availability_pattern", code_to_id_map=availability_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="availability", default_pattern=academic_term_defaults.get("availability"), ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="start_preference_id", pattern_field="start_preference_pattern", code_to_id_map=start_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="start_preference", default_pattern=academic_term_defaults.get("start_preference"), ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="usage_preference_id", pattern_field="usage_preference_pattern", code_to_id_map=usage_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="usage_preference", default_pattern=academic_term_defaults.get("usage_preference"), ) if row_has_error: rows_skipped += 1 continue data["academic_term_id"] = effective_academic_term_id cleaned.append((row_num, data)) if not cleaned: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) to_create = [] to_update = [] week_map = {} inserted = 0 updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status", "academic_term_id", "module_id"] ] updatable_fields.append("updated_by") if "slot_required" in header: updatable_fields.append("slot_required") updatable_fields.append("duration") if "staff_requirement" in header: updatable_fields.append("staff_requirement") updatable_fields.append("staff_requirement_type") if "location_requirement" in header: updatable_fields.append("location_requirement") updatable_fields.append("location_requirement_type") updatable_fields = list(dict.fromkeys(updatable_fields)) for row_num, data in cleaned: code = data["code"] week_ids = data.pop("_week_ids", []) has_staff_requirement_param = data.pop("_has_staff_requirement_param", False) has_location_requirement_param = data.pop("_has_location_requirement_param", False) obj = existing_map.get(code) if obj is None: if not has_staff_requirement_param: data["staff_requirement"] = 0 else: data["staff_requirement"] = self.safe_int_or_zero(data.get("staff_requirement")) if not has_location_requirement_param: data["location_requirement"] = 0 else: data["location_requirement"] = self.safe_int_or_zero(data.get("location_requirement")) staff_req = data.get("staff_requirement") location_req = data.get("location_requirement") data["staff_requirement_type"] = ( TtActivity.STAFF_REQUIREMENT_TYPE["none"] if staff_req == 0 else TtActivity.STAFF_REQUIREMENT_TYPE["wildcard"] ) data["location_requirement_type"] = ( TtActivity.LOCATION_REQUIREMENT_TYPE["none"] if location_req == 0 else TtActivity.LOCATION_REQUIREMENT_TYPE["wildcard"] ) missing_for_insert = [ col for col in required_insert_columns if col not in header or data.get(col) in [None, ""] ] 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 to_create.append(TtActivityTemplate(**data)) if week_ids: week_map[code] = week_ids else: if not has_staff_requirement_param: data.pop("staff_requirement", None) data.pop("staff_requirement_type", None) else: if obj.staff_requirement_type == TtActivity.STAFF_REQUIREMENT_TYPE["preset"]: data.pop("staff_requirement", None) data.pop("staff_requirement_type", None) else: data["staff_requirement"] = self.safe_int_or_zero(data.get("staff_requirement")) staff_req = data.get("staff_requirement") data["staff_requirement_type"] = ( TtActivity.STAFF_REQUIREMENT_TYPE["none"] if staff_req == 0 else TtActivity.STAFF_REQUIREMENT_TYPE["wildcard"] ) if not has_location_requirement_param: data.pop("location_requirement", None) data.pop("location_requirement_type", None) else: if obj.location_requirement_type == TtActivity.LOCATION_REQUIREMENT_TYPE["preset"]: data.pop("location_requirement", None) data.pop("location_requirement_type", None) else: data["location_requirement"] = self.safe_int_or_zero(data.get("location_requirement")) location_req = data.get("location_requirement") data["location_requirement_type"] = ( TtActivity.LOCATION_REQUIREMENT_TYPE["none"] if location_req == 0 else TtActivity.LOCATION_REQUIREMENT_TYPE["wildcard"] ) changed = False for f in updatable_fields: if f == "updated_by": continue if f not in data: continue new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue old_val = getattr(obj, f, None) if isinstance(old_val, time) and isinstance(new_val, str): old_val = old_val.strftime("%H:%M:%S") if old_val != new_val: setattr(obj, f, new_val) changed = True if changed: obj.updated_by = request.user.id to_update.append(obj) if week_ids: week_map[code] = week_ids else: rows_skipped += 1 with transaction.atomic(): if to_create: TtActivityTemplate.objects.bulk_create(to_create, batch_size=3000) inserted = len(to_create) if to_update: TtActivityTemplate.objects.bulk_update(to_update, updatable_fields, batch_size=3000) updated = len(to_update) created_codes = [obj.code for obj in to_create] created_activity_template_map = { obj.code: obj.id for obj in TtActivityTemplate.objects.filter(code__in=created_codes).only("id", "code") } activity_template_ids = [] pending_activity_template_week_map = {} activity_template_week_response_map = {} for key, week_ids in week_map.items(): activity_template_id = created_activity_template_map.get(key) if not activity_template_id: existing_obj = existing_map.get(key) activity_template_id = existing_obj.id if existing_obj else None if not activity_template_id: continue activity_template_ids.append(activity_template_id) pending_activity_template_week_map[activity_template_id] = week_ids or [] existing_activity_template_week_ids = set( TtActivityTemplateWeek.objects .filter(activity_template_id__in=activity_template_ids) .values_list("activity_template_id", flat=True) ) activity_template_week_rows = [] for activity_template_id, week_ids in pending_activity_template_week_map.items(): if activity_template_id in existing_activity_template_week_ids: continue activity_template_week_response_map[activity_template_id] = week_ids for week_id in week_ids: activity_template_week_rows.append( TtActivityTemplateWeek( activity_template_id=activity_template_id, week_id=week_id ) ) if activity_template_week_rows: TtActivityTemplateWeek.objects.bulk_create( activity_template_week_rows, batch_size=10000, ignore_conflicts=True ) redis_activity_template_table = "activity_template" redis_activity_template_column = [ "id", "code", "name", "desc", "planned_size", "slot_required", "activity_type_id", "duration", "department_id", "module_id", "academic_term_id", "week_pattern_id", "week_pattern", "availability_id", "availability_pattern", "start_preference_id", "start_preference_pattern", "usage_preference_id", "usage_preference_pattern", "staff_requirement_type", "staff_required_no", "staff_preset", "staff_suitability_id", "location_requirement_type", "location_required_no", "location_preset", "location_suitability_id", "status" ] update_redis_activity_template_column = [ "name", "desc", "planned_size", "duration","activity_type_id", "department_id", "week_pattern_id", "week_pattern", "availability_id", "availability_pattern", "start_preference_id", "start_preference_pattern", "usage_preference_id", "usage_preference_pattern", "staff_requirement_type", "staff_required_no", "staff_preset", "staff_suitability_id", "location_requirement_type", "location_required_no", "location_preset", "location_suitability_id", "status" ] redis_data = { "insert": {}, "update": {}, } created_relation = {} updated_relation = {} if to_create: for activity_template in to_create: redis_activity_template_data = {} week_pattern_for_redis = [] if activity_template_week_response_map: week_ids = activity_template_week_response_map.get(activity_template.id, []) week_pattern_for_redis = convert_to_redis_week(week_ids) if week_ids else [] created_relation[activity_template.id] = week_ids for column in redis_activity_template_column: match column: case "week_pattern": redis_activity_template_data[column] = week_pattern_for_redis or [] case "staff_suitability_id" | "location_suitability_id": redis_activity_template_data[column] = None case "staff_preset" | "location_preset": redis_activity_template_data[column] = [] case "student_sets": # when create dun have student set, so can force to empty list redis_activity_template_data[column] = [] case "staff_required_no": redis_activity_template_data[column] = activity_template.staff_requirement case "location_required_no": redis_activity_template_data[column] = activity_template.location_requirement case "usage_preference_id" | "availability_id" | "start_preference_id" | "week_pattern_id": value = getattr(activity_template, column) redis_activity_template_data[column] = 0 if value in (None, "") else value case _: redis_activity_template_data[column] = getattr(activity_template, column) redis_data["insert"].setdefault(redis_activity_template_table, []).append(redis_activity_template_data) if to_update: for activity_template in to_update: redis_activity_template_data = {"id": activity_template.id} week_pattern_for_redis = [] if activity_template_week_response_map: week_ids = activity_template_week_response_map.get(activity_template.id, []) week_pattern_for_redis = convert_to_redis_week(week_ids) if week_ids else [] updated_relation[activity_template.id] = week_ids for column in update_redis_activity_template_column: if column in updatable_fields: redis_activity_template_data[column] = getattr(activity_template, column) if column == "week_pattern": redis_activity_template_data[column] = week_pattern_for_redis or [] if column == "staff_required_no": redis_activity_template_data[column] = activity_template.staff_requirement if column == "location_required_no": redis_activity_template_data[column] = activity_template.location_requirement if column in ["usage_preference_id", "availability_id", "start_preference_id"]: value = getattr(activity_template, column) redis_activity_template_data[column] = 0 if value in (None, "") else value if len(redis_activity_template_data) > 1: redis_data["update"].setdefault(redis_activity_template_table, []).append(redis_activity_template_data) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) create_method = "activity_template_create" update_method = "activity_template_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] ignore_fields = ["created_at", "updated_at", "created_by", "updated_by"] # CREATE created_kafka_data = [] created_ids = [c.id for c in to_create] if created_ids: created_data = TtActivityTemplate.objects.filter(id__in=created_ids) for data in created_data: temp = {} for f in TtActivityTemplate._meta.fields: if f.name in ignore_fields: continue if isinstance(f, ForeignKey): # if is foreign key, dun direct follow field name, cause will call 1 more time to db # eg. field name is department, if using department.id to get the id, will call to db since not prefethch this value, need used department_id value = getattr(data, f"{f.name}_id") if value is not None: temp[f.name] = value else: # move to here to fet value, so if is foreign key, wont come to this part, can skip the get value from FK table problem value = getattr(data, f.name) if value is not None: temp[f.name] = value temp["week"] = created_relation.get(data.id, None) or [] created_kafka_data.append(temp) if created_kafka_data: kafka_request_data = { "session_id": request.user.name, "activity_template": created_kafka_data, } send_request(kafka_topic, kafka_request_data, None, create_method) # UPDATE update_kafka_data = [] update_fields = [f for f in updatable_fields if f != "updated_by"] for data in to_update: temp = {"id": data.id} for up in update_fields: value = getattr(data, up, None) # if value is None: # continue temp[up] = value temp["week"] = updated_relation.get(data.id, None) or [] update_kafka_data.append(temp) if update_kafka_data: kafka_request_data = { "session_id": request.user.name, "activity_template": update_kafka_data, } send_request(kafka_topic, kafka_request_data, None, update_method) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_activity(self, request, header, rows, settings): slot_per_week = settings.get("slot_per_week") minute_per_slot = int(settings.get("minute_per_slot")) slot_per_day = int(settings.get("slot_per_day")) rules = TtActivity.IMPORT_RULES errors = [] if "code" not in header: errors.append({ "row": 1, "field": "code", "error": __(self.BASE_TRANS + "missing_required", value="code") }) if errors: return self.import_return(errors=errors) 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 activity_template_code_candidates = set() academic_term_code_candidates = set() dept_code_candidates = set() zone_code_candidates = set() week_pattern_code_candidates = set() availability_code_candidates = set() start_preference_code_candidates = set() usage_preference_code_candidates = set() jta_parent_code_candidates = set() variant_parent_code_candidates = set() seen_codes = {} for row_num, row in rows: rows_read += 1 row_has_error = False data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value has_staff_requirement_param = "staff_requirement" in header has_location_requirement_param = "location_requirement" in header data["_has_staff_requirement_param"] = has_staff_requirement_param data["_has_location_requirement_param"] = has_location_requirement_param if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "code_required"), }) continue code = data["code"].strip() 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 # integer check for field, field_rules in rules.items(): if field not in header: continue if "integer" not in field_rules: continue raw_value = data.get(field) if raw_value in ["", None]: data[field] = None continue try: data[field] = int(raw_value) except ValueError: if "required" in field_rules: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_integer", value=raw_value), }) row_has_error = True else: data[field] = None # time check for field, field_rules in rules.items(): if field not in header: continue if "time" not in field_rules: continue raw_value = data.get(field) if raw_value in ["", None]: data[field] = None continue try: data[field] = datetime.strptime(raw_value, "%H:%M:%S") except ValueError: try: data[field] = datetime.strptime(raw_value, "%H:%M") except ValueError: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_time", value=raw_value), }) data[field] = None # special integer validation for day_field in ["scheduled_day", "suggested_day"]: if day_field not in header: continue if day_field in data: if data[day_field] not in range(0, 7): errors.append({ "row": row_num, "field": day_field, "error": __(self.BASE_TRANS + "invalid_overridden", field=day_field, value=data[day_field]), }) data[day_field] = None for bool_int_field in ["is_jta", "is_variant"]: if bool_int_field not in header: continue if bool_int_field in data: if data[bool_int_field] not in [0, 1]: errors.append({ "row": row_num, "field": bool_int_field, "error": __(self.BASE_TRANS + "invalid_overridden", field=day_field, value=data[bool_int_field]), }) data[bool_int_field] = 0 if "scheduled_day" in header and "scheduled_start_time" in header: if data.get("scheduled_day") and data.get("scheduled_start_time"): data["scheduled_start_slot"] = (data.get("scheduled_day") * slot_per_day) + time_to_slot(data.get("scheduled_start_time"), minute_per_slot) data["scheduled"] = 1 if "suggested_time" in header: if data.get("suggested_time"): data["suggested_time_slot"] = time_to_slot(data.get("suggested_time"), minute_per_slot) for value in ["scheduled_start_time", "suggested_time"]: if isinstance(data.get(value), datetime): data[value] = data.get(value).strftime("%H:%M:%S") # staff_req = data.get("staff_requirement") or 0 # location_req = data.get("location_requirement") or 0 # data["staff_requirement_type"] = ( # TtActivity.STAFF_REQUIREMENT_TYPE["none"] # if staff_req == 0 # else TtActivity.STAFF_REQUIREMENT_TYPE["wildcard"] # ) # data["location_requirement_type"] = ( # TtActivity.LOCATION_REQUIREMENT_TYPE["none"] # if location_req == 0 # else TtActivity.LOCATION_REQUIREMENT_TYPE["wildcard"] # ) if row_has_error: rows_skipped += 1 continue if "slot_required" in header: slot_required = int(data.get("slot_required") or 0) data["duration"] = slot_required * minute_per_slot for key, container in [ ("activity_template_id", activity_template_code_candidates), ("academic_term_id", academic_term_code_candidates), ("department_id", dept_code_candidates), ("zone_id", zone_code_candidates), ("week_pattern_id", week_pattern_code_candidates), ("availability_id", availability_code_candidates), ("start_preference_id", start_preference_code_candidates), ("usage_preference_id", usage_preference_code_candidates), ("jta_parent_id", jta_parent_code_candidates), ("variant_parent_id", variant_parent_code_candidates), ]: if data.get(key): container.add(data[key]) data["status"] = TtActivity.STATUS_TO_CODE["active"] data["created_by"] = request.user.id staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) activity_template_code_to_id = fetch_code_id_map(TtActivityTemplate, "code", activity_template_code_candidates) academic_term_code_to_id = fetch_code_id_map(TtAcademicTerm, "code", academic_term_code_candidates) dept_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) zone_code_to_id = fetch_code_id_map(TtZone, "code", zone_code_candidates) week_pattern_code_to_id = fetch_code_id_map(TtWeekPattern, "code", week_pattern_code_candidates) availability_code_to_id = fetch_code_id_map(TtAvailability, "code", availability_code_candidates) start_preference_code_to_id = fetch_code_id_map(TtStartPreference, "code", start_preference_code_candidates) usage_preference_code_to_id = fetch_code_id_map(TtUsagePreference, "code", usage_preference_code_candidates) jta_parent_code_to_id = fetch_code_id_map(TtActivity, "code", jta_parent_code_candidates) variant_parent_code_to_id = fetch_code_id_map(TtActivity, "code", variant_parent_code_candidates) codes = list({d["code"] for _, d in staged}) existing_map = { obj.code: obj for obj in TtActivity.objects.filter(code__in=codes) } academic_term_ids = set(v for v in academic_term_code_to_id.values() if v) academic_term_ids.update( obj.academic_term_id for obj in existing_map.values() if obj.academic_term_id ) academic_term_week_map = {} for row in TtAcademicTermWeek.objects.filter( academic_term_id__in=academic_term_ids ).values("academic_term_id", "week_id"): academic_term_week_map.setdefault(row["academic_term_id"], []).append(row["week_id"]) academic_term_default_map = { obj.id: { "availability": obj.default_availability, "start_preference": obj.default_start, "usage_preference": obj.default_usage, } for obj in TtAcademicTerm.objects .filter(id__in=academic_term_ids) .only("id", "default_availability", "default_start", "default_usage") } activity_template_ids = [v for v in activity_template_code_to_id.values() if v] activity_template_map = { obj.id: obj for obj in TtActivityTemplate.objects.filter(id__in=activity_template_ids) } week_pattern_ids = [v for v in week_pattern_code_to_id.values() if v] week_pattern_academic_term_map = { obj.id: obj.academic_term_id for obj in TtWeekPattern.objects.filter(id__in=week_pattern_ids) } cleaned = [] for row_num, data in staged: row_has_error = False obj = existing_map.get(data["code"]) activity_template_code = (data.get("activity_template_id") or "").strip() academic_term_code = (data.get("academic_term_id") or "").strip() dep_code = (data.get("department_id") or "").strip() zone_code = (data.get("zone_id") or "").strip() wp_code = data.get("week_pattern_id") jta_parent_code = (data.get("jta_parent_id") or "").strip() variant_parent_code = (data.get("variant_parent_id") or "").strip() if obj: data["activity_template_id"] = obj.activity_template_id else: if activity_template_code: activity_template_id = activity_template_code_to_id.get(activity_template_code) if activity_template_id is None: errors.append({ "row": row_num, "field": "activity_template_id", "error": __(self.BASE_TRANS + "invalid_required", field="activity template code", value=activity_template_code), }) data["activity_template_id"] = None row_has_error = True else: data["activity_template_id"] = activity_template_id template_obj = activity_template_map.get(data["activity_template_id"]) data["module_id"] = template_obj.module_id data["activity_type_id"] = template_obj.activity_type_id else: errors.append({ "row": row_num, "field": "activity_template_id", "error": __(self.BASE_TRANS + "missing_required", value="activity_template_id"), }) data["activity_template_id"] = None row_has_error = True if row_has_error: rows_skipped += 1 continue # Determine effective academic_term_id: prefer existing object's value; then explicit academic_term code; # otherwise derive from activity_template -> module -> academic_term. Also derive department from module when absent. effective_academic_term_id = None has_error = False if obj: effective_academic_term_id = obj.academic_term_id else: if academic_term_code: effective_academic_term_id, has_error = self.resolve_academic_term_id( row_num, None, academic_term_code, academic_term_code_to_id, errors, self.BASE_TRANS ) else: # Try derive from activity_template -> module # activity_template_code = (data.get("activity_template_id") or "").strip() if activity_template_code: activity_template_id = activity_template_code_to_id.get(activity_template_code) template_obj = activity_template_map.get(activity_template_id) if template_obj and template_obj.module_id: module_obj = TtModule.objects.filter(id=template_obj.module_id).only("id", "academic_term_id", "department_id").first() if module_obj and module_obj.academic_term_id: effective_academic_term_id = module_obj.academic_term_id # derive department if not provided in file if not data.get("department_id"): data["department_id"] = module_obj.department_id else: errors.append({ "row": row_num, "field": "activity_template_id", "error": __(self.BASE_TRANS + "missing_required", value="academic_term_id"), }) has_error = True else: errors.append({ "row": row_num, "field": "activity_template_id", "error": __(self.BASE_TRANS + "invalid_required", field="activity template code", value=activity_template_code), }) has_error = True else: errors.append({ "row": row_num, "field": "activity_template_id", "error": __(self.BASE_TRANS + "missing_required", value="academic_term_id"), }) has_error = True if has_error: effective_academic_term_id = None row_has_error = True academic_term_defaults = academic_term_default_map.get(effective_academic_term_id, {}) # department: if provided in file, validate; else if already set from module derivation above, keep it; otherwise default to None if "department_id" in header: if dep_code: dept_id = dept_code_to_id.get(dep_code) data["department_id"] = dept_id if dept_id is None: errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="department code", value=dep_code), }) else: data.setdefault("department_id", None) if "zone_id" in header: if zone_code: zone_id = zone_code_to_id.get(zone_code) data["zone_id"] = zone_id if zone_id is None: errors.append({ "row": row_num, "field": "zone_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="zone code", value=zone_code), }) else: data["zone_id"] = None if data.get("activity_template_id") and effective_academic_term_id: template_obj = activity_template_map.get(data["activity_template_id"]) if template_obj and template_obj.academic_term_id != effective_academic_term_id: errors.append({ "row": row_num, "field": "activity_template_id,academic_term_id", "error": __(self.BASE_TRANS + "academic_term_activity_template_mismatch"), }) row_has_error = True if "jta_parent_id" in header: if jta_parent_code: jta_parent_id = jta_parent_code_to_id.get(jta_parent_code) if jta_parent_id is None: errors.append({ "row": row_num, "field": "jta_parent_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="jta parent activity code", value=jta_parent_code), }) data["jta_parent_id"] = None data["is_jta"] = None elif jta_parent_code == data.get("code"): errors.append({ "row": row_num, "field": "jta_parent_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="jta parent activity code", value=jta_parent_code), }) data["jta_parent_id"] = None data["is_jta"] = None else: data["jta_parent_id"] = jta_parent_id data["is_jta"] = 1 else: data["jta_parent_id"] = None if "variant_parent_id" in header: if variant_parent_code: variant_parent_id = variant_parent_code_to_id.get(variant_parent_code) if variant_parent_id is None: errors.append({ "row": row_num, "field": "variant_parent_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="variant parent activity code", value=variant_parent_code), }) data["variant_parent_id"] = None data["is_variant"] = None elif variant_parent_code == data.get("code"): errors.append({ "row": row_num, "field": "variant_parent_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="variant parent activity code", value=variant_parent_code), }) data["variant_parent_id"] = None data["is_variant"] = None else: data["variant_parent_id"] = variant_parent_id data["is_variant"] = 1 else: data["variant_parent_id"] = None # week pattern logic if wp_code: wp_id = week_pattern_code_to_id.get(wp_code) if wp_id: week_pattern_academic_term_id = week_pattern_academic_term_map.get(wp_id) if week_pattern_academic_term_id == effective_academic_term_id: data["week_pattern_id"] = wp_id else: errors.append({ "row": row_num, "field": "week_pattern_id,academic_term_id", "error": __(self.BASE_TRANS + "academic_term_week_pattern_mismatch", value=wp_code), }) row_has_error = True else: data["week_pattern_id"] = None data["_week_ids"] = academic_term_week_map.get(effective_academic_term_id, []) else: data["week_pattern_id"] = None data["_week_ids"] = academic_term_week_map.get(effective_academic_term_id, []) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="availability_id", pattern_field="availability_pattern", code_to_id_map=availability_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="availability", default_pattern=academic_term_defaults.get("availability"), ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="start_preference_id", pattern_field="start_preference_pattern", code_to_id_map=start_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="start_preference", default_pattern=academic_term_defaults.get("start_preference"), ) self.resolve_code_or_pattern( data=data, row_num=row_num, code_field="usage_preference_id", pattern_field="usage_preference_pattern", code_to_id_map=usage_preference_code_to_id, errors=errors, trans_key=self.BASE_TRANS, num=slot_per_week, pattern_type="usage_preference", default_pattern=academic_term_defaults.get("usage_preference"), ) if row_has_error: rows_skipped += 1 continue data["academic_term_id"] = effective_academic_term_id cleaned.append((row_num, data)) if not cleaned: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) to_create = [] to_update = [] week_map = {} inserted = 0 updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status", "academic_term_id", "activity_template_id"] ] updatable_fields.append("updated_by") # Generated fields: only update if their source header exists if "slot_required" in header: updatable_fields.append("slot_required") updatable_fields.append("duration") if "staff_requirement" in header: updatable_fields.append("staff_requirement") updatable_fields.append("staff_requirement_type") if "location_requirement" in header: updatable_fields.append("location_requirement") updatable_fields.append("location_requirement_type") if "scheduled_start_time" in header or "scheduled_day" in header: updatable_fields.append("scheduled_start_slot") updatable_fields.append("scheduled") if "suggested_time" in header: updatable_fields.append("suggested_time_slot") # Remove duplicate fields updatable_fields = list(dict.fromkeys(updatable_fields)) for row_num, data in cleaned: code = data["code"] week_ids = data.pop("_week_ids", []) has_staff_requirement_param = data.pop("_has_staff_requirement_param", False) has_location_requirement_param = data.pop("_has_location_requirement_param", False) obj = existing_map.get(code) if obj is None: if not has_staff_requirement_param: data["staff_requirement"] = 0 else: data["staff_requirement"] = self.safe_int_or_zero(data.get("staff_requirement")) if not has_location_requirement_param: data["location_requirement"] = 0 else: data["location_requirement"] = self.safe_int_or_zero(data.get("location_requirement")) staff_req = data.get("staff_requirement") location_req = data.get("location_requirement") data["staff_requirement_type"] = ( TtActivity.STAFF_REQUIREMENT_TYPE["none"] if staff_req == 0 else TtActivity.STAFF_REQUIREMENT_TYPE["wildcard"] ) data["location_requirement_type"] = ( TtActivity.LOCATION_REQUIREMENT_TYPE["none"] if location_req == 0 else TtActivity.LOCATION_REQUIREMENT_TYPE["wildcard"] ) missing_for_insert = [ col for col in required_insert_columns if col not in header or data.get(col) in [None, ""] ] 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 to_create.append(TtActivity(**data)) if week_ids: week_map[code] = week_ids else: if not has_staff_requirement_param: data.pop("staff_requirement", None) data.pop("staff_requirement_type", None) else: if obj.staff_requirement_type == TtActivity.STAFF_REQUIREMENT_TYPE["preset"]: data.pop("staff_requirement", None) data.pop("staff_requirement_type", None) else: data["staff_requirement"] = self.safe_int_or_zero(data.get("staff_requirement")) staff_req = data.get("staff_requirement") data["staff_requirement_type"] = ( TtActivity.STAFF_REQUIREMENT_TYPE["none"] if staff_req == 0 else TtActivity.STAFF_REQUIREMENT_TYPE["wildcard"] ) if not has_location_requirement_param: data.pop("location_requirement", None) data.pop("location_requirement_type", None) else: if obj.location_requirement_type == TtActivity.LOCATION_REQUIREMENT_TYPE["preset"]: data.pop("location_requirement", None) data.pop("location_requirement_type", None) else: data["location_requirement"] = self.safe_int_or_zero(data.get("location_requirement")) location_req = data.get("location_requirement") data["location_requirement_type"] = ( TtActivity.LOCATION_REQUIREMENT_TYPE["none"] if location_req == 0 else TtActivity.LOCATION_REQUIREMENT_TYPE["wildcard"] ) changed = False for f in updatable_fields: if f == "updated_by": continue if f not in data: continue new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue old_val = getattr(obj, f, None) if isinstance(old_val, time) and isinstance(new_val, str): old_val = old_val.strftime("%H:%M:%S") if old_val != new_val: setattr(obj, f, new_val) changed = True if changed: obj.updated_by = request.user.id to_update.append(obj) if week_ids: week_map[code] = week_ids else: rows_skipped += 1 from api.services.integration.tracking import track_activity_created, track_activity_ids track_activity_ids(obj.id for obj in to_update) with transaction.atomic(): if to_create: TtActivity.objects.bulk_create(to_create) for created_activity in to_create: track_activity_created(created_activity) inserted = len(to_create) if to_update: TtActivity.objects.bulk_update(to_update, updatable_fields) updated = len(to_update) created_codes = [obj.code for obj in to_create] created_activity_map = { obj.code: obj.id for obj in TtActivity.objects.filter(code__in=created_codes).only("id", "code") } activity_ids = [] pending_activity_week_map = {} activity_week_response_map = {} for key, week_ids in week_map.items(): activity_id = created_activity_map.get(key) if not activity_id: existing_obj = existing_map.get(key) activity_id = existing_obj.id if existing_obj else None if not activity_id: continue activity_ids.append(activity_id) pending_activity_week_map[activity_id] = week_ids or [] existing_activity_week_ids = set( TtActivityWeek.objects .filter(activity_id__in=activity_ids) .values_list("activity_id", flat=True) ) activity_week_rows = [] for activity_id, week_ids in pending_activity_week_map.items(): if activity_id in existing_activity_week_ids: continue activity_week_response_map[activity_id] = week_ids for week_id in week_ids: activity_week_rows.append( TtActivityWeek( activity_id=activity_id, week_id=week_id ) ) if activity_week_rows: TtActivityWeek.objects.bulk_create( activity_week_rows, batch_size=3000, ignore_conflicts=True ) redis_activity_table = "activity" redis_activity_column = [ "id", "code", "name", "desc", "planned_size", "real_size", "slot_required","activity_type_id", "duration", "suggested_day", "suggested_time_slot","suggested_time","scheduled_start_slot", "department_id", "module_id", "academic_term_id", "activity_template_id", "week_pattern_id", "week_pattern", "availability_id", "availability_pattern","start_preference_id", "start_preference_pattern", "usage_preference_id", "usage_preference_pattern","student_sets", "staff_requirement_type","staff_required_no", "staff_preset", "staff_suitability_id", "location_requirement_type","location_required_no", "location_preset","location_suitability_id", "is_jta", "jta_parent_id", "is_variant", "variant_parent_id", "status", "scheduled" ] update_redis_activity_column = [ "name", "desc", "planned_size", "real_size", "activity_type_id", "duration","suggested_day","suggested_time_slot","suggested_time", "department_id", "module_id", "academic_term_id", "activity_template_id", "week_pattern_id", "week_pattern", "availability_id", "availability_pattern","start_preference_id", "start_preference_pattern", "usage_preference_id", "usage_preference_pattern", "staff_preset", "staff_suitability_id", "location_preset", "location_suitability_id", "status" ] redis_data = { "insert": {}, "update": {}, } created_relation = {} updated_relation = {} if to_create: for activity in to_create: redis_activity_data = {} week_pattern_for_redis = [] if activity_week_response_map: week_ids = activity_week_response_map.get(activity.id, []) week_pattern_for_redis = convert_to_redis_week(week_ids) if week_ids else [] created_relation[activity.id] = week_ids for column in redis_activity_column: match column: case "week_pattern": redis_activity_data[column] = week_pattern_for_redis or [] case "staff_suitability_id" | "location_suitability_id": redis_activity_data[column] = None case "staff_preset" | "location_preset": redis_activity_data[column] = [] case "student_sets": redis_activity_data[column] = [] case "staff_required_no": redis_activity_data[column] = activity.staff_requirement case "location_required_no": redis_activity_data[column] = activity.location_requirement case "usage_preference_id" | "availability_id" | "start_preference_id" | "week_pattern_id": value = getattr(activity, column) redis_activity_data[column] = 0 if value in (None, "") else value case _: value = getattr(activity, column) redis_activity_data[column] = self.format_time_if_needed(value) redis_data["insert"].setdefault(redis_activity_table, []).append(redis_activity_data) if to_update: for activity in to_update: redis_activity_data = {"id": activity.id} week_pattern_for_redis = [] if activity_week_response_map: week_ids = activity_week_response_map.get(activity.id, []) week_pattern_for_redis = convert_to_redis_week(week_ids) if week_ids else [] updated_relation[activity.id] = week_ids for column in update_redis_activity_column: if column in updatable_fields: value = getattr(activity, column) redis_activity_data[column] = self.format_time_if_needed(value) if column == "week_pattern": redis_activity_data[column] = week_pattern_for_redis or [] if column == "staff_required_no": redis_activity_data[column] = activity.staff_requirement if column == "location_required_no": redis_activity_data[column] = activity.location_requirement if column in ["usage_preference_id", "availability_id", "start_preference_id"]: value = getattr(activity, column) redis_activity_data[column] = 0 if value in (None, "") else value if len(redis_activity_data) > 1: redis_data["update"].setdefault(redis_activity_table, []).append(redis_activity_data) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) create_method = "activity_create" update_method = "activity_update" kafka_topic = self.kafka_config["MICROSERVICES_TT_TOPIC"] ignore_fields = ["created_at", "updated_at", "created_by", "updated_by"] created_kafka_data = [] created_ids = [c.id for c in to_create] if created_ids: created_data = TtActivity.objects.filter(id__in=created_ids) for data in created_data: temp = {} for f in TtActivity._meta.fields: if f.name in ignore_fields: continue if isinstance(f, ForeignKey): # if is foreign key, dun direct follow field name, cause will call 1 more time to db # eg. field name is department, if using department.id to get the id, will call to db since not prefethch this value, need used department_id value = getattr(data, f"{f.name}_id") if value is not None: temp[f.name] = value else: # move to here to fetch value, so if is foreign key, wont come to this part, can skip the get value from FK table problem value = getattr(data, f.name) if value is not None: if isinstance(value, time): temp[f.name] = value.strftime("%H:%M") else: temp[f.name] = value temp["week"] = created_relation.get(data.id, None) or [] created_kafka_data.append(temp) if created_kafka_data: kafka_request_data = { "session_id": request.user.name, "activity": created_kafka_data, } send_request(kafka_topic, kafka_request_data, None, create_method) update_kafka_data = [] update_fields = [f for f in updatable_fields if f != "updated_by"] for data in to_update: temp = {"id": data.id} for up in update_fields: value = getattr(data, up, None) if isinstance(value, time): temp[up] = value.strftime("%H:%M") else: temp[up] = value temp["week"] = updated_relation.get(data.id, None) or [] update_kafka_data.append(temp) if update_kafka_data: kafka_request_data = { "session_id": request.user.name, "activity": update_kafka_data, } send_request(kafka_topic, kafka_request_data, None, update_method) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_suitability(self, request, header, rows): rules = TtSuitability.IMPORT_RULES 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 = {} dept_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value # code required if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "code_required"), }) continue code = data["code"].strip() 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 # collect FK candidates if "department_id" in header and data.get("department_id"): dept_code_candidates.add(data["department_id"].strip()) data["status"] = TtSuitability.STATUS_TO_CODE["active"] data["created_by"] = request.user.id staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) dept_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) file_codes = [data["code"] for _, data in staged] existing_qs = TtSuitability.objects.filter(code__in=file_codes) existing_map = {obj.code: obj for obj in existing_qs} cleaned = [] for row_num, data in staged: dep_code = (data.get("department_id") or "").strip() # department_id if "department_id" in header: if dep_code: dept_id = dept_code_to_id.get(dep_code) if dept_id is None: errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="department code", value=dep_code), }) data["department_id"] = None else: data["department_id"] = dept_id else: data["department_id"] = None # desc normalize if "desc" in header and not (data.get("desc") or "").strip(): data["desc"] = None cleaned.append((row_num, data)) if not cleaned: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) to_create = [] to_update = [] inserted = 0 updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status"] ] updatable_fields.append("updated_by") updatable_fields = list(dict.fromkeys(updatable_fields)) for row_num, data in cleaned: code = data["code"] obj = existing_map.get(code) if obj is None: missing_for_insert = [ col for col in required_insert_columns if col not in header or data.get(col) in [None, ""] ] 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 to_create.append(TtSuitability(**data)) else: changed = False for f in updatable_fields: if f == "updated_by": continue if f not in data: continue new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue if getattr(obj, f, None) != new_val: setattr(obj, f, new_val) changed = True if changed: obj.updated_by = request.user.id to_update.append(obj) else: rows_skipped += 1 with transaction.atomic(): if to_create: TtSuitability.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_update: TtSuitability.objects.bulk_update(to_update, updatable_fields, batch_size=500) updated = len(to_update) redis_suitability_table = "suitability" redis_suitability_column = [ "id","code","name","department_id","staff_id","location_id" ] update_redis_suitability_column = [ "name","department_id", # "staff_id","location_id" dont have m2m ] redis_data = { "insert": {}, "update": {}, } if to_create: for suitability in to_create: redis_suitability_data = {} for column in redis_suitability_column: match column: case "staff_id" | "location_id": redis_suitability_data[column] = [] # wont create m2m case _: redis_suitability_data[column] = getattr(suitability, column) redis_data["insert"].setdefault(redis_suitability_table, []).append(redis_suitability_data) if to_update: for suitability in to_update: redis_suitability_data = { "id": suitability.id } for column in update_redis_suitability_column: if column in updatable_fields: redis_suitability_data[column] = getattr(suitability, column) if len(redis_suitability_data) > 1: redis_data["update"].setdefault(redis_suitability_table, []).append(redis_suitability_data) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) self.kafka_response(request, "suitability", to_create, to_update, updatable_fields) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_suitability_staff(self, request, header, rows): rules = TtSuitability.SUITABILITY_STAFF.get("IMPORT_RULES", None) rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] for col in ["suitability_id", "staff_id"]: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col) }] ) staged = [] suitability_code_candidates = set() staff_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False for field, field_rules in rules.items(): if "required" not in field_rules: continue if not data.get(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 suitability_code_candidates.add(data["suitability_id"]) staff_code_candidates.add(data["staff_id"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) suitability_code_to_id = fetch_code_id_map(TtSuitability, "code", suitability_code_candidates) staff_code_to_id = fetch_code_id_map(TtStaff, "code", staff_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False suitability_code = data["suitability_id"] staff_code = data["staff_id"] suitability_id = suitability_code_to_id.get(suitability_code) staff_id = staff_code_to_id.get(staff_code) if suitability_id is None: errors.append({ "row": row_num, "field": "suitability_id", "error": __(self.BASE_TRANS + "invalid_required", field="suitability code", value=suitability_code), }) row_has_error = True else: data["suitability_id"] = suitability_id if staff_id is None: errors.append({ "row": row_num, "field": "staff_id", "error": __(self.BASE_TRANS + "invalid_required", field="staff code", value=staff_code), }) row_has_error = True else: data["staff_id"] = staff_id if row_has_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) unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["suitability_id"], data["staff_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) suitability_ids = [data["suitability_id"] for _, data in deduped] staff_ids = [data["staff_id"] for _, data in deduped] existing_pairs = set( TtStaffSuitability.objects .filter( suitability_id__in=suitability_ids, staff_id__in=staff_ids ) .values_list("suitability_id", "staff_id") ) to_create = [] for row_num, data in deduped: pair = (data["suitability_id"], data["staff_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtStaffSuitability( suitability_id=data["suitability_id"], staff_id=data["staff_id"], )) with transaction.atomic(): if to_create: TtStaffSuitability.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_create: redis_suitability_table = "suitability" affected_suitability_ids = list({obj.suitability_id for obj in to_create}) suitability_staff_rows = ( TtStaffSuitability.objects .filter(suitability_id__in=affected_suitability_ids) .values("suitability_id", "staff_id") ) suitability_staff_map = {} for row in suitability_staff_rows: suitability_id = row["suitability_id"] staff_id = row["staff_id"] suitability_staff_map.setdefault(suitability_id, []).append(staff_id) redis_data = { "update": {} } grouped_data = [] for suitability_id, staff_ids in suitability_staff_map.items(): grouped_data.append({ "id": suitability_id, "staff_id": sorted(list(set(staff_ids))) }) redis_data["update"].setdefault(redis_suitability_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "suitability_update" kafka_topic = self.kafka_config["MICROSERVICES_TT_TOPIC"] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "suitability": grouped_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_suitability_location(self, request, header, rows): rules = TtSuitability.SUITABILITY_LOCATION.get("IMPORT_RULES", None) rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] 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) }] ) staged = [] suitability_code_candidates = set() location_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False # required checking from IMPORT_RULES for field, field_rules in rules.items(): if "required" not in field_rules: continue if not data.get(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 suitability_code_candidates.add(data["suitability_id"]) location_code_candidates.add(data["location_id"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) suitability_code_to_id = fetch_code_id_map( TtSuitability, "code", suitability_code_candidates ) location_code_to_id = fetch_code_id_map( TtLocation, "code", location_code_candidates ) cleaned = [] for row_num, data in staged: row_has_error = False suitability_code = data["suitability_id"] location_code = data["location_id"] suitability_id = suitability_code_to_id.get(suitability_code) location_id = location_code_to_id.get(location_code) if suitability_id is None: errors.append({ "row": row_num, "field": "suitability_id", "error": __(self.BASE_TRANS + "invalid_required", field="suitability code", value=suitability_code), }) row_has_error = True else: data["suitability_id"] = suitability_id if location_id is None: errors.append({ "row": row_num, "field": "location_id", "error": __(self.BASE_TRANS + "invalid_required", field="location code", value=location_code), }) row_has_error = True else: data["location_id"] = location_id if row_has_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) unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["suitability_id"], data["location_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) suitability_ids = [data["suitability_id"] for _, data in deduped] location_ids = [data["location_id"] for _, data in deduped] existing_pairs = set( TtLocationSuitability.objects .filter( suitability_id__in=suitability_ids, location_id__in=location_ids ) .values_list("suitability_id", "location_id") ) to_create = [] for row_num, data in deduped: pair = (data["suitability_id"], data["location_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtLocationSuitability( suitability_id=data["suitability_id"], location_id=data["location_id"], )) with transaction.atomic(): if to_create: TtLocationSuitability.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_create: redis_suitability_table = "suitability" affected_suitability_ids = list({obj.suitability_id for obj in to_create}) suitability_location_rows = ( TtLocationSuitability.objects .filter(suitability_id__in=affected_suitability_ids) .values("suitability_id", "location_id") ) suitability_location_map = {} for row in suitability_location_rows: suitability_id = row["suitability_id"] location_id = row["location_id"] suitability_location_map.setdefault(suitability_id, []).append(location_id) redis_data = { "update": {} } grouped_data = [] for suitability_id, location_ids in suitability_location_map.items(): grouped_data.append({ "id": suitability_id, "location_id": sorted(list(set(location_ids))) }) redis_data["update"].setdefault(redis_suitability_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "suitability_update" kafka_topic = self.kafka_config["MICROSERVICES_TT_TOPIC"] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "suitability": grouped_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_activity_template_staff(self, request, header, rows): rules = TtActivityTemplateStaff.IMPORT_RULES rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] for col in ["activity_template_code", "staff_code"]: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col) }] ) staged = [] activity_template_code_candidates = set() staff_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False if not data["activity_template_code"]: errors.append({ "row": row_num, "field": "activity_template_code", "error": __(self.BASE_TRANS + "missing_required", value="activity_template_code"), }) row_has_error = True if not data["staff_code"]: errors.append({ "row": row_num, "field": "staff_code", "error": __(self.BASE_TRANS + "missing_required", value="staff_code"), }) row_has_error = True if row_has_error: rows_skipped += 1 continue activity_template_code_candidates.add(data["activity_template_code"]) staff_code_candidates.add(data["staff_code"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) activity_template_code_to_id = fetch_code_id_map(TtActivityTemplate, "code", activity_template_code_candidates) staff_code_to_id = fetch_code_id_map(TtStaff, "code", staff_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False activity_template_code = data["activity_template_code"] staff_code = data["staff_code"] activity_template_id = activity_template_code_to_id.get(activity_template_code) staff_id = staff_code_to_id.get(staff_code) if activity_template_id is None: errors.append({ "row": row_num, "field": "activity_template_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.activity_template_code"), value=activity_template_code), }) row_has_error = True else: data["activity_template_id"] = activity_template_id if staff_id is None: errors.append({ "row": row_num, "field": "staff_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.staff_code"), value=staff_code), }) row_has_error = True else: data["staff_id"] = staff_id if row_has_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) unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["activity_template_id"], data["staff_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) staff_ids = [data["staff_id"] for _, data in deduped] activity_template_ids = [data["activity_template_id"] for _, data in deduped] existing_pairs = set( TtActivityTemplateStaff.objects .filter( activity_template_id__in=activity_template_ids, staff_id__in=staff_ids ) .values_list("staff_id", "activity_template_id") ) to_create = [] for row_num, data in deduped: pair = (data["staff_id"], data["activity_template_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtActivityTemplateStaff( staff_id=data["staff_id"], activity_template_id=data["activity_template_id"], )) with transaction.atomic(): if to_create: TtActivityTemplateStaff.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_create: redis_activity_template_table = "activity_template" affected_activity_template_ids = list({obj.activity_template_id for obj in to_create}) activity_template_staff_rows = ( TtActivityTemplateStaff.objects .filter(activity_template_id__in=affected_activity_template_ids) .values("activity_template_id", "staff_id") ) activity_template_staff_map = {} for row in activity_template_staff_rows: activity_template_id = row["activity_template_id"] staff_id = row["staff_id"] activity_template_staff_map.setdefault(activity_template_id, []).append(staff_id) update_activity_template = False activity_templates = TtActivityTemplate.objects.filter(id__in=affected_activity_template_ids).prefetch_related("staff") for activity_template in activity_templates: update_activity_template = True activity_template.staff_requirement = len(activity_template.staff.all()) activity_template.staff_requirement_type = TtActivityTemplate.STAFF_REQUIREMENT_TYPE['preset'] if update_activity_template: TtActivityTemplate.objects.bulk_update(activity_templates,["staff_requirement","staff_requirement_type"]) # need delete relation for suitability, cause already preset TtActivityTemplateStaffSuitability.objects.filter(activity_template_id__in=affected_activity_template_ids).delete() redis_data = { "update": {} } grouped_data = [] microservices_data = [] for activity_template_id, staff_ids in activity_template_staff_map.items(): grouped_data.append({ "id": activity_template_id, "staff_requirement_type": TtActivityTemplate.STAFF_REQUIREMENT_TYPE['preset'], "staff_suitability_id": None, "staff_required_no": len(staff_ids), "staff_preset": sorted(list(set(staff_ids))) }) microservices_data.append({ "id": activity_template_id, "staff_requirement_type": TtActivityTemplate.STAFF_REQUIREMENT_TYPE['preset'], "staff_suitability": [], "staff_requirement": len(staff_ids), "staff": sorted(list(set(staff_ids))) }) redis_data["update"].setdefault(redis_activity_template_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "activity_template_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "activity_template": microservices_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_activity_template_location(self, request, header, rows): rules = TtActivityTemplateLocation.IMPORT_RULES rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] for col in ["activity_template_code", "location_code"]: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col) }] ) staged = [] activity_template_code_candidates = set() location_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False if not data["activity_template_code"]: errors.append({ "row": row_num, "field": "activity_template_code", "error": __(self.BASE_TRANS + "missing_required", value="activity_template_code"), }) row_has_error = True if not data["location_code"]: errors.append({ "row": row_num, "field": "location_code", "error": __(self.BASE_TRANS + "missing_required", value="location_code"), }) row_has_error = True if row_has_error: rows_skipped += 1 continue activity_template_code_candidates.add(data["activity_template_code"]) location_code_candidates.add(data["location_code"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) activity_template_code_to_id = fetch_code_id_map(TtActivityTemplate, "code", activity_template_code_candidates) location_code_to_id = fetch_code_id_map(TtLocation, "code", location_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False activity_template_code = data["activity_template_code"] location_code = data["location_code"] activity_template_id = activity_template_code_to_id.get(activity_template_code) location_id = location_code_to_id.get(location_code) if activity_template_id is None: errors.append({ "row": row_num, "field": "activity_template_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.activity_template_code"), value=activity_template_code), }) row_has_error = True else: data["activity_template_id"] = activity_template_id if location_id is None: errors.append({ "row": row_num, "field": "location_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.location_code"), value=location_code), }) row_has_error = True else: data["location_id"] = location_id if row_has_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) unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["activity_template_id"], data["location_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) location_ids = [data["location_id"] for _, data in deduped] activity_template_ids = [data["activity_template_id"] for _, data in deduped] existing_pairs = set( TtActivityTemplateLocation.objects .filter( activity_template_id__in=activity_template_ids, location_id__in=location_ids ) .values_list("location_id", "activity_template_id") ) to_create = [] for row_num, data in deduped: pair = (data["location_id"], data["activity_template_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtActivityTemplateLocation( location_id=data["location_id"], activity_template_id=data["activity_template_id"], )) with transaction.atomic(): if to_create: TtActivityTemplateLocation.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_create: redis_activity_template_table = "activity_template" affected_activity_template_ids = list({obj.activity_template_id for obj in to_create}) activity_template_location_rows = ( TtActivityTemplateLocation.objects .filter(activity_template_id__in=affected_activity_template_ids) .values("activity_template_id", "location_id") ) activity_template_location_map = {} for row in activity_template_location_rows: activity_template_id = row["activity_template_id"] location_id = row["location_id"] activity_template_location_map.setdefault(activity_template_id, []).append(location_id) update_activity_template = False activity_templates = TtActivityTemplate.objects.filter( id__in=affected_activity_template_ids).prefetch_related( "location") for activity_template in activity_templates: update_activity_template = True activity_template.location_requirement = len(activity_template.location.all()) activity_template.location_requirement_type = TtActivityTemplate.STAFF_REQUIREMENT_TYPE['preset'] if update_activity_template: TtActivityTemplate.objects.bulk_update(activity_templates, ["location_requirement", "location_requirement_type"]) # need delete relation for suitability, cause already preset TtActivityTemplateLocationSuitability.objects.filter( activity_template_id__in=affected_activity_template_ids).delete() redis_data = { "update": {} } grouped_data = [] microservices_data = [] for activity_template_id, location_ids in activity_template_location_map.items(): grouped_data.append({ "id": activity_template_id, "location_requirement_type": TtActivityTemplate.STAFF_REQUIREMENT_TYPE['preset'], "location_suitability_id": None, "location_required_no": len(location_ids), "location_preset": sorted(list(set(location_ids))) }) microservices_data.append({ "id": activity_template_id, "location_requirement_type": TtActivityTemplate.STAFF_REQUIREMENT_TYPE['preset'], "location_suitability_id": [], "location_requirement": len(location_ids), "location_preset": sorted(list(set(location_ids))) }) redis_data["update"].setdefault(redis_activity_template_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "activity_template_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "activity_template": microservices_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_activity_template_staff_s(self, request, header, rows): """ function name end as _s is because got limit in excel sheets name, it will limit 31 char only """ rules = TtActivityTemplateStaffSuitability.IMPORT_RULES rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] for col in ["activity_template_code", "suitability_code"]: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col) }] ) staged = [] activity_template_code_candidates = set() suitability_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False if not data["activity_template_code"]: errors.append({ "row": row_num, "field": "activity_template_code", "error": __(self.BASE_TRANS + "missing_required", value="activity_template_code"), }) row_has_error = True if not data["suitability_code"]: errors.append({ "row": row_num, "field": "suitability_code", "error": __(self.BASE_TRANS + "missing_required", value="suitability_code"), }) row_has_error = True if row_has_error: rows_skipped += 1 continue activity_template_code_candidates.add(data["activity_template_code"]) suitability_code_candidates.add(data["suitability_code"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) activity_template_code_to_id = fetch_code_id_map(TtActivityTemplate, "code", activity_template_code_candidates) suitability_code_to_id = fetch_code_id_map(TtSuitability, "code", suitability_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False activity_template_code = data["activity_template_code"] suitability_code = data["suitability_code"] activity_template_id = activity_template_code_to_id.get(activity_template_code) suitability_id = suitability_code_to_id.get(suitability_code) if activity_template_id is None: errors.append({ "row": row_num, "field": "activity_template_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.activity_template_code"), value=activity_template_code), }) row_has_error = True else: data["activity_template_id"] = activity_template_id if suitability_id is None: errors.append({ "row": row_num, "field": "suitability_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.suitability_code"), value=suitability_code), }) row_has_error = True else: data["suitability_id"] = suitability_id if row_has_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) unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["activity_template_id"], data["suitability_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) suitability_ids = [data["suitability_id"] for _, data in deduped] activity_template_ids = [data["activity_template_id"] for _, data in deduped] existing_pairs = set( TtActivityTemplateStaffSuitability.objects .filter( activity_template_id__in=activity_template_ids, suitability_id__in=suitability_ids ) .values_list("suitability_id", "activity_template_id") ) to_create = [] for row_num, data in deduped: pair = (data["suitability_id"], data["activity_template_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtActivityTemplateStaffSuitability( suitability_id=data["suitability_id"], activity_template_id=data["activity_template_id"], )) with transaction.atomic(): if to_create: TtActivityTemplateStaffSuitability.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_create: redis_activity_template_table = "activity_template" affected_activity_template_ids = list({obj.activity_template_id for obj in to_create}) activity_template_suitability_rows = ( TtActivityTemplateStaffSuitability.objects .filter(activity_template_id__in=affected_activity_template_ids) .values("activity_template_id", "suitability_id") ) activity_template_suitability_map = {} for row in activity_template_suitability_rows: activity_template_id = row["activity_template_id"] suitability_id = row["suitability_id"] activity_template_suitability_map.setdefault(activity_template_id, []).append(suitability_id) update_activity_template = False activity_templates = TtActivityTemplate.objects.filter( id__in=affected_activity_template_ids).prefetch_related( "staff") for activity_template in activity_templates: update_activity_template = True activity_template.staff_requirement_type = TtActivityTemplate.STAFF_REQUIREMENT_TYPE['wildcard'] if update_activity_template: TtActivityTemplate.objects.bulk_update(activity_templates,["staff_requirement_type"]) # need delete relation for preset, cause already preset TtActivityTemplateStaff.objects.filter( activity_template_id__in=affected_activity_template_ids).delete() redis_data = { "update": {} } grouped_data = [] microservices_data = [] for activity_template_id, suitability_ids in activity_template_suitability_map.items(): grouped_data.append({ "id": activity_template_id, "staff_requirement_type": TtActivityTemplate.STAFF_REQUIREMENT_TYPE['wildcard'], "staff_suitability_id": suitability_ids[0] if suitability_ids else None, # jose engine only can handle 1 suitability "staff_preset": [] }) microservices_data.append({ "id": activity_template_id, "staff_requirement_type": TtActivityTemplate.STAFF_REQUIREMENT_TYPE['wildcard'], # our db can store can m2m "staff_suitability": sorted(list(set(suitability_ids))), "staff": [] }) redis_data["update"].setdefault(redis_activity_template_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "activity_template_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "activity_template": microservices_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_activity_template_location_s(self, request, header, rows): """ function name end as _s is because got limit in excel sheets name, it will limit 31 char only """ rules = TtActivityTemplateLocationSuitability.IMPORT_RULES rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] for col in ["activity_template_code", "suitability_code"]: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col) }] ) staged = [] activity_template_code_candidates = set() suitability_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False if not data["activity_template_code"]: errors.append({ "row": row_num, "field": "activity_template_code", "error": __(self.BASE_TRANS + "missing_required", value="activity_template_code"), }) row_has_error = True if not data["suitability_code"]: errors.append({ "row": row_num, "field": "suitability_code", "error": __(self.BASE_TRANS + "missing_required", value="suitability_code"), }) row_has_error = True if row_has_error: rows_skipped += 1 continue activity_template_code_candidates.add(data["activity_template_code"]) suitability_code_candidates.add(data["suitability_code"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) activity_template_code_to_id = fetch_code_id_map(TtActivityTemplate, "code", activity_template_code_candidates) suitability_code_to_id = fetch_code_id_map(TtSuitability, "code", suitability_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False activity_template_code = data["activity_template_code"] suitability_code = data["suitability_code"] activity_template_id = activity_template_code_to_id.get(activity_template_code) suitability_id = suitability_code_to_id.get(suitability_code) if activity_template_id is None: errors.append({ "row": row_num, "field": "activity_template_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.activity_template_code"), value=activity_template_code), }) row_has_error = True else: data["activity_template_id"] = activity_template_id if suitability_id is None: errors.append({ "row": row_num, "field": "suitability_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.suitability_code"), value=suitability_code), }) row_has_error = True else: data["suitability_id"] = suitability_id if row_has_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) unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["activity_template_id"], data["suitability_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) suitability_ids = [data["suitability_id"] for _, data in deduped] activity_template_ids = [data["activity_template_id"] for _, data in deduped] existing_pairs = set( TtActivityTemplateLocationSuitability.objects .filter( activity_template_id__in=activity_template_ids, suitability_id__in=suitability_ids ) .values_list("suitability_id", "activity_template_id") ) to_create = [] for row_num, data in deduped: pair = (data["suitability_id"], data["activity_template_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtActivityTemplateLocationSuitability( suitability_id=data["suitability_id"], activity_template_id=data["activity_template_id"], )) with transaction.atomic(): if to_create: TtActivityTemplateLocationSuitability.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_create: redis_activity_template_table = "activity_template" affected_activity_template_ids = list({obj.activity_template_id for obj in to_create}) activity_template_suitability_rows = ( TtActivityTemplateLocationSuitability.objects .filter(activity_template_id__in=affected_activity_template_ids) .values("activity_template_id", "suitability_id") ) activity_template_suitability_map = {} for row in activity_template_suitability_rows: activity_template_id = row["activity_template_id"] suitability_id = row["suitability_id"] activity_template_suitability_map.setdefault(activity_template_id, []).append(suitability_id) update_activity_template = False activity_templates = TtActivityTemplate.objects.filter( id__in=affected_activity_template_ids).prefetch_related( "location") for activity_template in activity_templates: update_activity_template = True activity_template.location_requirement_type = TtActivityTemplate.STAFF_REQUIREMENT_TYPE['wildcard'] if update_activity_template: TtActivityTemplate.objects.bulk_update(activity_templates, ["location_requirement_type"]) # need delete relation for preset, cause already preset TtActivityTemplateLocation.objects.filter( activity_template_id__in=affected_activity_template_ids).delete() redis_data = { "update": {} } grouped_data = [] microservices_data = [] for activity_template_id, suitability_ids in activity_template_suitability_map.items(): grouped_data.append({ "id": activity_template_id, "location_requirement_type": TtActivityTemplate.STAFF_REQUIREMENT_TYPE['wildcard'], "location_suitability_id": suitability_ids[0] if suitability_ids else None, # jose engine only can handle 1 suitability "location_preset": [] }) microservices_data.append({ "id": activity_template_id, "location_requirement_type": TtActivityTemplate.STAFF_REQUIREMENT_TYPE['wildcard'], # our db can store can m2m "location_suitability": sorted(list(set(suitability_ids))), "location": [] }) redis_data["update"].setdefault(redis_activity_template_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "activity_template_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "activity_template": microservices_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_activity_staff_preset(self, request, header, rows): rules = TtActivityStaffPreset.IMPORT_RULES rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] for col in ["activity_code", "staff_code"]: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col) }] ) staged = [] activity_code_candidates = set() staff_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False if not data["activity_code"]: errors.append({ "row": row_num, "field": "activity_code", "error": __(self.BASE_TRANS + "missing_required", value="activity_code"), }) row_has_error = True if not data["staff_code"]: errors.append({ "row": row_num, "field": "staff_code", "error": __(self.BASE_TRANS + "missing_required", value="staff_code"), }) row_has_error = True if row_has_error: rows_skipped += 1 continue activity_code_candidates.add(data["activity_code"]) staff_code_candidates.add(data["staff_code"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) activity_code_to_id = fetch_code_id_map(TtActivity, "code", activity_code_candidates) staff_code_to_id = fetch_code_id_map(TtStaff, "code", staff_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False activity_code = data["activity_code"] staff_code = data["staff_code"] activity_id = activity_code_to_id.get(activity_code) staff_id = staff_code_to_id.get(staff_code) if activity_id is None: errors.append({ "row": row_num, "field": "activity_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.activity_code"), value=activity_code), }) row_has_error = True else: data["activity_id"] = activity_id if staff_id is None: errors.append({ "row": row_num, "field": "staff_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.staff_code"), value=staff_code), }) row_has_error = True else: data["staff_id"] = staff_id if row_has_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) unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["activity_id"], data["staff_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) staff_ids = [data["staff_id"] for _, data in deduped] activity_ids = [data["activity_id"] for _, data in deduped] existing_pairs = set( TtActivityStaffPreset.objects .filter( activity_id__in=activity_ids, staff_id__in=staff_ids ) .values_list("staff_id", "activity_id") ) to_create = [] for row_num, data in deduped: pair = (data["staff_id"], data["activity_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtActivityStaffPreset( staff_id=data["staff_id"], activity_id=data["activity_id"], )) with transaction.atomic(): if to_create: TtActivityStaffPreset.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_create: redis_activity_table = "activity" affected_activity_ids = list({obj.activity_id for obj in to_create}) activity_staff_rows = ( TtActivityStaffPreset.objects .filter(activity_id__in=affected_activity_ids) .values("activity_id", "staff_id") ) activity_staff_map = {} for row in activity_staff_rows: activity_id = row["activity_id"] staff_id = row["staff_id"] activity_staff_map.setdefault(activity_id, []).append(staff_id) update_activity = False activitys = TtActivity.objects.filter(id__in=affected_activity_ids).prefetch_related( "staff_preset") from api.services.integration.tracking import track_activity_ids track_activity_ids(affected_activity_ids) for activity in activitys: update_activity = True activity.staff_requirement = len(activity.staff_preset.all()) activity.staff_requirement_type = TtActivity.STAFF_REQUIREMENT_TYPE['preset'] if update_activity: TtActivity.objects.bulk_update(activitys, ["staff_requirement", "staff_requirement_type"]) # need delete relation for suitability, cause already preset TtActivityStaffSuitability.objects.filter( activity_id__in=affected_activity_ids).delete() redis_data = { "update": {} } grouped_data = [] microservices_data = [] for activity_id, staff_ids in activity_staff_map.items(): grouped_data.append({ "id": activity_id, "staff_requirement_type": TtActivity.STAFF_REQUIREMENT_TYPE['preset'], "staff_suitability_id": None, "staff_required_no": len(staff_ids), "staff_preset": sorted(list(set(staff_ids))) }) microservices_data.append({ "id": activity_id, "staff_requirement_type": TtActivity.STAFF_REQUIREMENT_TYPE['preset'], "staff_suitability": [], "staff_requirement": len(staff_ids), "staff_preset": sorted(list(set(staff_ids))) }) redis_data["update"].setdefault(redis_activity_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "activity_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "activity": microservices_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_activity_location_preset(self, request, header, rows): rules = TtActivityLocationPreset.IMPORT_RULES rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] for col in ["activity_code", "location_code"]: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col) }] ) staged = [] activity_code_candidates = set() location_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False if not data["activity_code"]: errors.append({ "row": row_num, "field": "activity_code", "error": __(self.BASE_TRANS + "missing_required", value="activity_code"), }) row_has_error = True if not data["location_code"]: errors.append({ "row": row_num, "field": "location_code", "error": __(self.BASE_TRANS + "missing_required", value="location_code"), }) row_has_error = True if row_has_error: rows_skipped += 1 continue activity_code_candidates.add(data["activity_code"]) location_code_candidates.add(data["location_code"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) activity_code_to_id = fetch_code_id_map(TtActivity, "code", activity_code_candidates) location_code_to_id = fetch_code_id_map(TtLocation, "code", location_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False activity_code = data["activity_code"] location_code = data["location_code"] activity_id = activity_code_to_id.get(activity_code) location_id = location_code_to_id.get(location_code) if activity_id is None: errors.append({ "row": row_num, "field": "activity_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.activity_code"), value=activity_code), }) row_has_error = True else: data["activity_id"] = activity_id if location_id is None: errors.append({ "row": row_num, "field": "location_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.location_code"), value=location_code), }) row_has_error = True else: data["location_id"] = location_id if row_has_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) unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["activity_id"], data["location_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) location_ids = [data["location_id"] for _, data in deduped] activity_ids = [data["activity_id"] for _, data in deduped] existing_pairs = set( TtActivityLocationPreset.objects .filter( activity_id__in=activity_ids, location_id__in=location_ids ) .values_list("location_id", "activity_id") ) to_create = [] for row_num, data in deduped: pair = (data["location_id"], data["activity_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtActivityLocationPreset( location_id=data["location_id"], activity_id=data["activity_id"], )) with transaction.atomic(): if to_create: TtActivityLocationPreset.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_create: redis_activity_table = "activity" affected_activity_ids = list({obj.activity_id for obj in to_create}) activity_location_rows = ( TtActivityLocationPreset.objects .filter(activity_id__in=affected_activity_ids) .values("activity_id", "location_id") ) activity_location_map = {} for row in activity_location_rows: activity_id = row["activity_id"] location_id = row["location_id"] activity_location_map.setdefault(activity_id, []).append(location_id) update_activity = False activitys = TtActivity.objects.filter(id__in=affected_activity_ids).prefetch_related( "location_preset") from api.services.integration.tracking import track_activity_ids track_activity_ids(affected_activity_ids) for activity in activitys: update_activity = True activity.location_requirement = len(activity.location_preset.all()) activity.location_requirement_type = TtActivity.LOCATION_REQUIREMENT_TYPE['preset'] if update_activity: TtActivity.objects.bulk_update(activitys, ["location_requirement", "location_requirement_type"]) # need delete relation for suitability, cause already preset TtActivityLocationSuitability.objects.filter( activity_id__in=affected_activity_ids).delete() redis_data = { "update": {} } grouped_data = [] microservices_data = [] for activity_id, location_ids in activity_location_map.items(): grouped_data.append({ "id": activity_id, "location_requirement_type": TtActivity.LOCATION_REQUIREMENT_TYPE['preset'], "location_suitability_id": None, "location_required_no": len(location_ids), "location_preset": sorted(list(set(location_ids))) }) microservices_data.append({ "id": activity_id, "location_requirement_type": TtActivity.LOCATION_REQUIREMENT_TYPE['preset'], "location_suitability": [], "location_requirement": len(location_ids), "location_preset": sorted(list(set(location_ids))) }) redis_data["update"].setdefault(redis_activity_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "activity_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "activity": microservices_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_activity_staff_s(self, request, header, rows): """ function name end as _s is because got limit in excel sheets name, it will limit 31 char only """ rules = TtActivityStaffSuitability.IMPORT_RULES rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] for col in ["activity_code", "suitability_code"]: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col) }] ) staged = [] activity_code_candidates = set() suitability_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False if not data["activity_code"]: errors.append({ "row": row_num, "field": "activity_code", "error": __(self.BASE_TRANS + "missing_required", value="activity_code"), }) row_has_error = True if not data["suitability_code"]: errors.append({ "row": row_num, "field": "suitability_code", "error": __(self.BASE_TRANS + "missing_required", value="suitability_code"), }) row_has_error = True if row_has_error: rows_skipped += 1 continue activity_code_candidates.add(data["activity_code"]) suitability_code_candidates.add(data["suitability_code"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) activity_code_to_id = fetch_code_id_map(TtActivity, "code", activity_code_candidates) suitability_code_to_id = fetch_code_id_map(TtSuitability, "code", suitability_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False activity_code = data["activity_code"] suitability_code = data["suitability_code"] activity_id = activity_code_to_id.get(activity_code) suitability_id = suitability_code_to_id.get(suitability_code) if activity_id is None: errors.append({ "row": row_num, "field": "activity_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.activity_code"), value=activity_code), }) row_has_error = True else: data["activity_id"] = activity_id if suitability_id is None: errors.append({ "row": row_num, "field": "suitability_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.suitability_code"), value=suitability_code), }) row_has_error = True else: data["suitability_id"] = suitability_id if row_has_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) unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["activity_id"], data["suitability_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) suitability_ids = [data["suitability_id"] for _, data in deduped] activity_ids = [data["activity_id"] for _, data in deduped] existing_pairs = set( TtActivityStaffSuitability.objects .filter( activity_id__in=activity_ids, suitability_id__in=suitability_ids ) .values_list("suitability_id", "activity_id") ) to_create = [] for row_num, data in deduped: pair = (data["suitability_id"], data["activity_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtActivityStaffSuitability( suitability_id=data["suitability_id"], activity_id=data["activity_id"], )) with transaction.atomic(): if to_create: TtActivityStaffSuitability.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_create: redis_activity_table = "activity" affected_activity_ids = list({obj.activity_id for obj in to_create}) activity_suitability_rows = ( TtActivityStaffSuitability.objects .filter(activity_id__in=affected_activity_ids) .values("activity_id", "suitability_id") ) activity_suitability_map = {} for row in activity_suitability_rows: activity_id = row["activity_id"] suitability_id = row["suitability_id"] activity_suitability_map.setdefault(activity_id, []).append(suitability_id) update_activity = False activitys = TtActivity.objects.filter( id__in=affected_activity_ids).prefetch_related( "staff") from api.services.integration.tracking import track_activity_ids track_activity_ids(affected_activity_ids) for activity in activitys: update_activity = True activity.staff_requirement_type = TtActivity.STAFF_REQUIREMENT_TYPE['wildcard'] if update_activity: TtActivity.objects.bulk_update(activitys, ["staff_requirement_type"]) # need delete relation for preset, cause already preset TtActivityStaffPreset.objects.filter( activity_id__in=affected_activity_ids).delete() redis_data = { "update": {} } grouped_data = [] microservices_data = [] for activity_id, suitability_ids in activity_suitability_map.items(): grouped_data.append({ "id": activity_id, "staff_requirement_type": TtActivity.STAFF_REQUIREMENT_TYPE['wildcard'], "staff_suitability_id": suitability_ids[0] if suitability_ids else None, # jose engine only can handle 1 suitability "staff_preset": [] }) microservices_data.append({ "id": activity_id, "staff_requirement_type": TtActivity.STAFF_REQUIREMENT_TYPE['wildcard'], # our db can store can m2m "staff_suitability": sorted(list(set(suitability_ids))), "staff": [] }) redis_data["update"].setdefault(redis_activity_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "activity_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "activity": microservices_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_activity_location_s(self, request, header, rows): """ function name end as _s is because got limit in excel sheets name, it will limit 31 char only """ rules = TtActivityLocationSuitability.IMPORT_RULES rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] for col in ["activity_code", "suitability_code"]: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col) }] ) staged = [] activity_code_candidates = set() suitability_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False if not data["activity_code"]: errors.append({ "row": row_num, "field": "activity_code", "error": __(self.BASE_TRANS + "missing_required", value="activity_code"), }) row_has_error = True if not data["suitability_code"]: errors.append({ "row": row_num, "field": "suitability_code", "error": __(self.BASE_TRANS + "missing_required", value="suitability_code"), }) row_has_error = True if row_has_error: rows_skipped += 1 continue activity_code_candidates.add(data["activity_code"]) suitability_code_candidates.add(data["suitability_code"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) activity_code_to_id = fetch_code_id_map(TtActivity, "code", activity_code_candidates) suitability_code_to_id = fetch_code_id_map(TtSuitability, "code", suitability_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False activity_code = data["activity_code"] suitability_code = data["suitability_code"] activity_id = activity_code_to_id.get(activity_code) suitability_id = suitability_code_to_id.get(suitability_code) if activity_id is None: errors.append({ "row": row_num, "field": "activity_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.activity_code"), value=activity_code), }) row_has_error = True else: data["activity_id"] = activity_id if suitability_id is None: errors.append({ "row": row_num, "field": "suitability_code", "error": __(self.BASE_TRANS + "invalid_required", field=__("import_attr.suitability_code"), value=suitability_code), }) row_has_error = True else: data["suitability_id"] = suitability_id if row_has_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) unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["activity_id"], data["suitability_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) suitability_ids = [data["suitability_id"] for _, data in deduped] activity_ids = [data["activity_id"] for _, data in deduped] existing_pairs = set( TtActivityLocationSuitability.objects .filter( activity_id__in=activity_ids, suitability_id__in=suitability_ids ) .values_list("suitability_id", "activity_id") ) to_create = [] for row_num, data in deduped: pair = (data["suitability_id"], data["activity_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtActivityLocationSuitability( suitability_id=data["suitability_id"], activity_id=data["activity_id"], )) with transaction.atomic(): if to_create: TtActivityLocationSuitability.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_create: redis_activity_table = "activity" affected_activity_ids = list({obj.activity_id for obj in to_create}) activity_suitability_rows = ( TtActivityLocationSuitability.objects .filter(activity_id__in=affected_activity_ids) .values("activity_id", "suitability_id") ) activity_suitability_map = {} for row in activity_suitability_rows: activity_id = row["activity_id"] suitability_id = row["suitability_id"] activity_suitability_map.setdefault(activity_id, []).append(suitability_id) update_activity = False activitys = TtActivity.objects.filter( id__in=affected_activity_ids).prefetch_related( "location") from api.services.integration.tracking import track_activity_ids track_activity_ids(affected_activity_ids) for activity in activitys: update_activity = True activity.location_requirement_type = TtActivity.STAFF_REQUIREMENT_TYPE['wildcard'] if update_activity: TtActivity.objects.bulk_update(activitys, ["location_requirement_type"]) # need delete relation for preset, cause already preset TtActivityLocationPreset.objects.filter( activity_id__in=affected_activity_ids).delete() redis_data = { "update": {} } grouped_data = [] microservices_data = [] for activity_id, suitability_ids in activity_suitability_map.items(): grouped_data.append({ "id": activity_id, "location_requirement_type": TtActivity.STAFF_REQUIREMENT_TYPE['wildcard'], "location_suitability_id": suitability_ids[0] if suitability_ids else None, # jose engine only can handle 1 suitability "location_preset": [] }) microservices_data.append({ "id": activity_id, "location_requirement_type": TtActivity.STAFF_REQUIREMENT_TYPE['wildcard'], # our db can store can m2m "location_suitability": sorted(list(set(suitability_ids))), "location": [] }) redis_data["update"].setdefault(redis_activity_table, grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "activity_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "activity": microservices_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_activity_type(self, request, header, rows, settings=None): rules = TtActivityType.IMPORT_RULES 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 = {} dept_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "code_required"), }) continue code = data["code"].strip() 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 if data.get("department_id"): dept_code_candidates.add(data["department_id"].strip()) data["status"] = TtActivityType.STATUS_TO_CODE["active"] data["created_by"] = request.user.id staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) dept_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) cleaned = [] for row_num, data in staged: dep_code = (data.get("department_id") or "").strip() if "department_id" in header: if dep_code: dept_id = dept_code_to_id.get(dep_code) if dept_id is None: errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="department code", value=dep_code), }) data["department_id"] = None else: data["department_id"] = dept_id else: data["department_id"] = None # desc normalize if "desc" in header and not (data.get("desc") or "").strip(): data["desc"] = None # boolean normalize for field, field_rules in rules.items(): if "boolean" not in field_rules: continue raw_value = data.get(field) # nullable boolean: empty means False if raw_value in ["", None]: data[field] = False continue parsed = self.parse_import_bool(raw_value) data[field] = parsed # color default + hex validate if "color" in header: raw_color = (data.get("color") or "").strip() if not raw_color: data["color"] = "#66ffff" else: if not raw_color.startswith("#"): raw_color = f"#{raw_color}" if not re.fullmatch(r"#[0-9a-fA-F]{6}", raw_color): errors.append({ "row": row_num, "field": "color", "error": __(self.BASE_TRANS + "invalid_hex", value=data.get("color")), }) data["color"] = "#66ffff" else: data["color"] = raw_color.lower() 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_qs = TtActivityType.objects.filter(code__in=file_codes) existing_map = {obj.code: obj for obj in existing_qs} to_create = [] to_update = [] inserted = 0 updated = 0 updatable_fields = [ f for f in rules.keys() if f in header and f not in ["code", "status"] ] updatable_fields.append("updated_by") updatable_fields = list(dict.fromkeys(updatable_fields)) for row_num, data in cleaned: code = data["code"] obj = existing_map.get(code) if obj is None: missing_for_insert = [ col for col in required_insert_columns if col not in header or not (data.get(col) or "").strip() ] # color has default, so do not block insert when missing if "color" in missing_for_insert: missing_for_insert.remove("color") data["color"] = "#66ffff" 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 to_create.append(TtActivityType(**data)) else: changed = False for f in updatable_fields: if f == "updated_by": continue if f not in data: continue new_val = data.get(f, None) if isinstance(new_val, str) and new_val.strip() == "": continue if getattr(obj, f, None) != new_val: setattr(obj, f, new_val) changed = True if changed: setattr(obj, "updated_by", request.user.id) to_update.append(obj) else: rows_skipped += 1 with transaction.atomic(): if to_create: TtActivityType.objects.bulk_create(to_create, batch_size=500) inserted = len(to_create) if to_update and updatable_fields: TtActivityType.objects.bulk_update(to_update, updatable_fields, batch_size=500) updated = len(to_update) self.kafka_response(request, "activity_type", to_create, to_update, updatable_fields) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def import_tt_student_set_activity(self, request, header, rows): rules = TtStudentSet.STUDENT_SET_ACTIVITY.get("IMPORT_RULES", None) rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] required_columns = ["student_set_id", "activity_id"] for col in required_columns: if col not in header: return self.import_return( errors=[{ "row": 1, "field": col, "error": __(self.BASE_TRANS + "missing_required", value=col) }] ) staged = [] student_set_code_candidates = set() activity_code_candidates = set() # 1. read file rows for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ( "" if value is None else str(value).strip()) data[field] = value row_has_error = False if not (data.get("student_set_id") or "").strip(): errors.append({ "row": row_num, "field": "student_set_id", "error": __(self.BASE_TRANS + "missing_required", value="student_set_id"), }) row_has_error = True if not (data.get("activity_id") or "").strip(): errors.append({ "row": row_num, "field": "activity_id", "error": __(self.BASE_TRANS + "missing_required", value="activity_id"), }) row_has_error = True if row_has_error: rows_skipped += 1 continue student_set_code_candidates.add(data["student_set_id"]) activity_code_candidates.add(data["activity_id"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) # 2. map code -> id student_set_code_to_id = fetch_code_id_map(TtStudentSet, "code", student_set_code_candidates) activity_code_to_id = fetch_code_id_map(TtActivity, "code", activity_code_candidates) cleaned = [] for row_num, data in staged: row_has_error = False student_set_code = (data.get("student_set_id") or "").strip() activity_code = (data.get("activity_id") or "").strip() student_set_id = student_set_code_to_id.get(student_set_code) activity_id = activity_code_to_id.get(activity_code) if student_set_id is None: errors.append({ "row": row_num, "field": "student_set_id", "error": __(self.BASE_TRANS + "invalid_required", field="student set code", value=student_set_code), }) row_has_error = True else: data["student_set_id"] = student_set_id if activity_id is None: errors.append({ "row": row_num, "field": "activity_id", "error": __(self.BASE_TRANS + "invalid_required", field="activity code", value=activity_code), }) row_has_error = True else: data["activity_id"] = activity_id if row_has_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) # 3. preload student_set -> pos -> pos activitys student_set_ids = list({data["student_set_id"] for _, data in cleaned}) student_set_qs = ( TtStudentSet.objects .filter(id__in=student_set_ids) .prefetch_related( "module", "module__ttactivity_set" ) ) # make sure module of activity has same module as student set module only can add student_set_module_activity_id_map = {} for student_set in student_set_qs: module_activity_ids = set() for module in student_set.module.all(): for module_activity in module.ttactivity_set.all(): module_activity_ids.add(module_activity.id) student_set_module_activity_id_map[student_set.id] = sorted(list(module_activity_ids)) # 4. validate activity is in student_set.module validated = [] for row_num, data in cleaned: # 2026-05-25 temporary hide this checking, sunway import file seem like not checking for this # student_set_id = data["student_set_id"] # activity_id = data["activity_id"] # # available_activity_ids = student_set_module_activity_id_map.get(student_set_id) # # if not available_activity_ids: # errors.append({ # "row": row_num, # "field": "student_set_id", # "error": __(self.BASE_TRANS + "invalid_required", field="activity code", # value="not available for this student set"), # }) # rows_skipped += 1 # continue # # if activity_id not in available_activity_ids: # errors.append({ # "row": row_num, # "field": "activity_id", # "error": __(self.BASE_TRANS + "invalid_required", field="activity code", # value="not available for this student set"), # }) # rows_skipped += 1 # continue validated.append((row_num, data)) if not validated: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) # 5. remove duplicate rows inside file unique_pairs = set() deduped = [] for row_num, data in validated: pair = (data["student_set_id"], data["activity_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) # 6. check existing relation in db student_set_ids = [data["student_set_id"] for _, data in deduped] activity_ids = [data["activity_id"] for _, data in deduped] existing_pairs = set( TtStudentSetActivity.objects .filter( student_set_id__in=student_set_ids, activity_id__in=activity_ids, ) .values_list("student_set_id", "activity_id") ) to_create = [] for row_num, data in deduped: pair = (data["student_set_id"], data["activity_id"]) if pair in existing_pairs: rows_skipped += 1 continue create_data = { "student_set_id": data["student_set_id"], "activity_id": data["activity_id"], } to_create.append(TtStudentSetActivity(**create_data)) # 7. insert with transaction.atomic(): if to_create: TtStudentSetActivity.objects.bulk_create(to_create, batch_size=3000) inserted = len(to_create) if to_create: redis_student_set_table = "student_set" redis_activity_table = "activity" affected_student_set_ids = list({obj.student_set_id for obj in to_create}) student_set_activity_rows = ( TtStudentSetActivity.objects .filter(student_set_id__in=affected_student_set_ids) .values("student_set_id", "activity_id") ) student_set_activity_map = {} for row in student_set_activity_rows: ss_id = row["student_set_id"] mod_id = row["activity_id"] student_set_activity_map.setdefault(ss_id, []).append(mod_id) redis_data = { "insert": {}, "update": {} } grouped_data = [] kafka_data = [] activity_student_set_map = {} for student_set_id, activity_ids in student_set_activity_map.items(): grouped_data.append({ "id": student_set_id, "allocated_activity_ids": sorted(list(set(activity_ids))) }) kafka_data.append({ "id": student_set_id, "activity": sorted(list(set(activity_ids))) }) for activity_id in activity_ids: activity_student_set_map.setdefault(activity_id, []).append(student_set_id) activity_grouped_data = [] for activity_id, student_set_ids in activity_student_set_map.items(): activity_grouped_data.append({ "id": activity_id, "student_sets": sorted(list(set(student_set_ids))) }) redis_data["update"].setdefault(redis_student_set_table, grouped_data) redis_data["update"].setdefault(redis_activity_table, activity_grouped_data) bulk_sync_to_redis(redis_client, redis_data) # kafka method = "student_set_update" kafka_topic = self.kafka_config['MICROSERVICES_TT_TOPIC'] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "student_set": kafka_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_module_group_module(self, request, header, rows): rules = TtPos.MODULE_GROUP_MODULE.get("IMPORT_RULES", None) rows_read = 0 rows_skipped = 0 inserted = 0 errors = [] # check missing required columns from IMPORT_RULES for field, field_rules in rules.items(): if "required" in field_rules and field not in header: return self.import_return( errors=[{ "row": 1, "field": field, "error": __(self.BASE_TRANS + "missing_required", value=field) }] ) staged = [] pos_module_group_code_candidates = set() module_code_candidates = set() for row_num, row in rows: rows_read += 1 data = {} for field in rules.keys(): value = row.get(field) value = (value or "").strip() if isinstance(value, str) else "" data[field] = value row_has_error = False # required checking from IMPORT_RULES for field, field_rules in rules.items(): if "required" not in field_rules: continue if not data.get(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 pos_module_group_code_candidates.add(data["pos_module_group_id"]) module_code_candidates.add(data["module_id"]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) pos_module_group_qs = ( TtPosModuleGroup.objects .filter(code__in=pos_module_group_code_candidates) .select_related("pos") ) pos_module_group_code_to_obj = { obj.code: obj for obj in pos_module_group_qs } module_qs = list(TtModule.objects.filter(code__in=module_code_candidates)) module_code_term_to_obj = objects_by_code_and_academic_term(module_qs) existing_module_codes = {obj.code for obj in module_qs} cleaned = [] for row_num, data in staged: row_has_error = False pos_module_group_code = data["pos_module_group_id"] module_code = data["module_id"] pos_module_group = pos_module_group_code_to_obj.get(pos_module_group_code) module = None if pos_module_group is not None: module = module_code_term_to_obj.get( (module_code, pos_module_group.pos.academic_term_id) ) if pos_module_group is None: errors.append({ "row": row_num, "field": "pos_module_group_id", "error": __(self.BASE_TRANS + "invalid_required", field="pos module group code", value=pos_module_group_code), }) row_has_error = True if module is None: invalid_field = ( "module academic term" if pos_module_group is not None and module_code in existing_module_codes else "module code" ) errors.append({ "row": row_num, "field": "module_id", "error": __(self.BASE_TRANS + "invalid_required", field=invalid_field, value=module_code), }) row_has_error = True if row_has_error: rows_skipped += 1 continue data["pos_module_group_id"] = pos_module_group.id data["module_id"] = module.id data["pos_id"] = pos_module_group.pos_id data["module_group_id"] = pos_module_group.module_group_id cleaned.append((row_num, data)) if not cleaned: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) unique_pairs = set() deduped = [] for row_num, data in cleaned: pair = (data["pos_module_group_id"], data["module_id"]) if pair in unique_pairs: rows_skipped += 1 continue unique_pairs.add(pair) deduped.append((row_num, data)) if not deduped: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) pos_module_group_ids = [data["pos_module_group_id"] for _, data in deduped] module_ids = [data["module_id"] for _, data in deduped] existing_pairs = set( TtPosModuleGroupModule.objects .filter( pos_module_group_id__in=pos_module_group_ids, module_id__in=module_ids ) .values_list("pos_module_group_id", "module_id") ) to_create = [] for row_num, data in deduped: pair = (data["pos_module_group_id"], data["module_id"]) if pair in existing_pairs: rows_skipped += 1 continue to_create.append(TtPosModuleGroupModule( pos_module_group_id=data["pos_module_group_id"], module_id=data["module_id"], )) affected_pos_ids = { data["pos_id"] for _, data in deduped if (data["pos_module_group_id"], data["module_id"]) not in existing_pairs } with transaction.atomic(): if to_create: TtPosModuleGroupModule.objects.bulk_create(to_create, batch_size=1000) inserted = len(to_create) if affected_pos_ids: latest_pos_module_ids = ( TtPosModuleGroupModule.objects .filter(pos_module_group__pos_id__in=affected_pos_ids) .values_list("module_id", flat=True) ) TtStudentSetModule.objects.filter( student_set__pos_id__in=affected_pos_ids ).exclude( module_id__in=latest_pos_module_ids ).delete() if to_create: affected_pos_module_group_ids = list({obj.pos_module_group_id for obj in to_create}) pos_module_group_rows = ( TtPosModuleGroup.objects .filter(id__in=affected_pos_module_group_ids) .values("id", "pos_id", "module_group_id") ) pos_module_group_ids = [row["id"] for row in pos_module_group_rows] pos_module_rows = ( TtPosModuleGroupModule.objects .filter(pos_module_group_id__in=pos_module_group_ids) .values("pos_module_group_id", "module_id") ) pos_module_map = {} for row in pos_module_rows: pos_module_group_id = row["pos_module_group_id"] module_id = row["module_id"] pos_module_map.setdefault(pos_module_group_id, []).append(module_id) grouped_data = [] for row in pos_module_group_rows: grouped_data.append({ "pos_module_group_id": row["id"], "pos_id": row["pos_id"], "module_group": row["module_group_id"], "module_id": sorted(list(set(pos_module_map.get(row["id"], [])))), }) method = "pos_update_module" kafka_topic = self.kafka_config["MICROSERVICES_TT_TOPIC"] if kafka_topic: kafka_request_data = { "session_id": request.user.name, "modules": grouped_data, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return( rows_read=rows_read, inserted=inserted, skipped=rows_skipped, errors=errors ) def import_tt_booking(self, request, header, rows, settings): slot_per_week = settings.get("slot_per_week") minute_per_slot = int(settings.get("minute_per_slot")) slot_per_day = int(settings.get("slot_per_day")) slot_default_pattern = get_slot_default_pattern(slot_per_week) rules = TtActivity.BOOKING.get("IMPORT_RULES", None) errors = [] if "code" not in header: errors.append({ "row": 1, "field": "code", "error": __(self.BASE_TRANS + "missing_required", value="code") }) if "start_date_time" not in header: errors.append({ "row": 1, "field": "start_date_time", "error": __(self.BASE_TRANS + "missing_required", value="start_date_time") }) if errors: return self.import_return(errors=errors) 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 activity_type_code_candidates = set() dept_code_candidates = set() location_code_candidates = set() seen_codes = {} for row_num, row in rows: rows_read += 1 row_has_error = False data = {} for field in rules.keys(): if field not in header: continue value = row.get(field) value = (value or "").strip() if isinstance(value, str) else ("" if value is None else str(value).strip()) data[field] = value if not (data.get("code") or "").strip(): rows_skipped += 1 errors.append({ "row": row_num, "field": "code", "error": __(self.BASE_TRANS + "code_required"), }) continue code = data["code"].strip() 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 # integer check for field, field_rules in rules.items(): if "integer" not in field_rules: continue raw_value = data.get(field) if raw_value in ["", None]: if field == "planned_size": data[field] = 0 else: data[field] = None continue try: data[field] = int(raw_value) except ValueError: if "required" in field_rules: errors.append({ "row": row_num, "field": field, "error": __(self.BASE_TRANS + "invalid_integer", value=raw_value), }) row_has_error = True else: if field == "planned_size": data[field] = 0 else: data[field] = None # start_date_time check raw_start_date_time = data.get("start_date_time") if raw_start_date_time in ["", None]: errors.append({ "row": row_num, "field": "start_date_time", "error": __(self.BASE_TRANS + "missing_required", value="start_date_time"), }) row_has_error = True else: try: start_dt = datetime.strptime(raw_start_date_time, "%Y-%m-%d %H:%M:%S") except (TypeError, ValueError): errors.append({ "row": row_num, "field": "start_date_time", "value": raw_start_date_time, "error": __(self.BASE_TRANS + "invalid_date_time", value=raw_start_date_time), }) row_has_error = True if row_has_error: rows_skipped += 1 continue booking_date = start_dt.date() booking_time = start_dt.time() booking_day = booking_date.weekday() # Monday = 0, Sunday = 6 data["_date"] = booking_date data["_time"] = booking_time data["_day"] = booking_day data["_have_location"] = data.get("location_id") not in ["", None] data.pop("start_date_time", None) # default patterns data["availability_pattern"] = slot_default_pattern["availability"] data["usage_preference_pattern"] = slot_default_pattern["usage_preference"] data["start_preference_pattern"] = slot_default_pattern["start_preference"] data["duration"] = (data.get("slot_required") or 0) * minute_per_slot data["staff_requirement"] = 1 data["staff_requirement_type"] = TtActivity.STAFF_REQUIREMENT_TYPE["wildcard"] data["location_requirement"] = 1 data["location_requirement_type"] = TtActivity.LOCATION_REQUIREMENT_TYPE["wildcard"] data["is_booking"] = 1 data["status"] = TtActivity.STATUS_TO_CODE["active"] data["created_by"] = request.user.id for key, container in [ ("activity_type_id", activity_type_code_candidates), ("department_id", dept_code_candidates), ("location_id", location_code_candidates), ]: if key not in header: continue if data.get(key): container.add(data[key]) staged.append((row_num, data)) if not staged: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) activity_type_objs = { obj.code: obj for obj in TtActivityType.objects.filter(code__in=activity_type_code_candidates) } dept_code_to_id = fetch_code_id_map(TtDepartment, "code", dept_code_candidates) location_code_to_id = fetch_code_id_map(TtLocation, "code", location_code_candidates) codes = list({d["code"] for _, d in staged}) existing_codes = set( TtActivity.objects .filter(code__in=codes) .values_list("code", flat=True) ) all_weeks = list( TtWeek.objects .all() .only("id", "start_date") .order_by("start_date") ) cleaned = [] for row_num, data in staged: row_has_error = False code = data["code"] if code in existing_codes: errors.append({ "row": row_num, "field": "code", "value": code, "error": __(self.BASE_TRANS + "code_exist", value=code), }) rows_skipped += 1 continue activity_type_code = (data.get("activity_type_id") or "").strip() dept_code = (data.get("department_id") or "").strip() location_code = (data.get("location_id") or "").strip() # activity type resolve + booking check if "activity_type_id" not in header or not activity_type_code: errors.append({ "row": row_num, "field": "activity_type_id", "error": __(self.BASE_TRANS + "missing_required", value="activity_type_id"), }) data["activity_type_id"] = None row_has_error = True else: activity_type_obj = activity_type_objs.get(activity_type_code) if activity_type_obj is None: errors.append({ "row": row_num, "field": "activity_type_id", "error": __(self.BASE_TRANS + "invalid_required", field="activity type code", value=activity_type_code), }) data["activity_type_id"] = None row_has_error = True elif not activity_type_obj.booking: errors.append({ "row": row_num, "field": "activity_type_id", "error": __(self.BASE_TRANS + "activity_booking", activity_type_code=activity_type_code), }) data["activity_type_id"] = None row_has_error = True else: data["activity_type_id"] = activity_type_obj.id # department resolve if "department_id" in header: if dept_code: dept_id = dept_code_to_id.get(dept_code) data["department_id"] = dept_id if dept_id is None: errors.append({ "row": row_num, "field": "department_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="department code", value=dept_code), }) data["department_id"] = None else: data["department_id"] = None # location resolve resolved_location_id = None if "location_id" in header: if location_code: resolved_location_id = location_code_to_id.get(location_code) data["location_id"] = resolved_location_id if resolved_location_id is None: errors.append({ "row": row_num, "field": "location_id", "error": __(self.BASE_TRANS + "invalid_overridden", field="location code", value=location_code), }) data["location_id"] = None else: data["location_id"] = None # find week by booking date booking_date = data["_date"] week_id = None for week in all_weeks: if week.start_date <= booking_date: week_id = week.id else: break if week_id is None: errors.append({ "row": row_num, "field": "start_date_time", "value": booking_date, "error": __(self.BASE_TRANS + "invalid_overridden", field="week date", value=booking_date), }) row_has_error = True if row_has_error: rows_skipped += 1 continue # scheduled fields only if got valid location code if resolved_location_id: scheduled_day = data["_day"] scheduled_start_time = data["_time"] data["scheduled"] = 1 data["scheduled_day"] = scheduled_day data["scheduled_start_time"] = scheduled_start_time data["scheduled_start_slot"] = ( scheduled_day * slot_per_day ) + time_to_slot(scheduled_start_time, minute_per_slot) else: data["scheduled"] = 0 data["scheduled_day"] = None data["scheduled_start_time"] = None data["scheduled_start_slot"] = None data["_week_id"] = week_id data["_location_id"] = resolved_location_id # remove temp fields before TtActivity(**data) data.pop("_date", None) data.pop("_time", None) data.pop("_day", None) data.pop("_have_location", None) cleaned.append((row_num, data)) if not cleaned: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) to_create = [] relation_map = {} for row_num, data in cleaned: week_id = data.pop("_week_id", None) location_id = data.pop("_location_id", None) # location_id is not a column in TtActivity, it belongs to TtActivityLocation data.pop("location_id", None) missing_for_insert = [ col for col in required_insert_columns if col not in header or data.get(col) in [None, ""] ] 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 to_create.append(TtActivity(**data)) relation_map[data["code"]] = { "week_id": week_id, "location_id": location_id, } if not to_create: return self.import_return(rows_read=rows_read, skipped=rows_skipped, errors=errors) inserted = 0 updated = 0 activity_week_response_map = {} location_with_affected_week = {} location_ids = set() from api.services.integration.tracking import track_activity_created with transaction.atomic(): TtActivity.objects.bulk_create(to_create, batch_size=3000) inserted = len(to_create) created_codes = [obj.code for obj in to_create] created_activity_map = { obj.code: obj.id for obj in TtActivity.objects .filter(code__in=created_codes) .only("id", "code") } for activity in to_create: activity.id = created_activity_map.get(activity.code) track_activity_created(activity) activity_week_rows = [] activity_location_rows = [] for code, relation in relation_map.items(): activity_id = created_activity_map.get(code) if not activity_id: continue week_id = relation.get("week_id") location_id = relation.get("location_id") if week_id: activity_week_response_map[activity_id] = [week_id] activity_week_rows.append( TtActivityWeek( activity_id=activity_id, week_id=week_id, ) ) if location_id: activity_location_rows.append( TtActivityLocation( activity_id=activity_id, location_id=location_id, ) ) if week_id: location_with_affected_week.setdefault(location_id, set()).add(week_id) location_ids.add(location_id) if activity_week_rows: TtActivityWeek.objects.bulk_create( activity_week_rows, batch_size=3000, ignore_conflicts=True, ) if activity_location_rows: TtActivityLocation.objects.bulk_create( activity_location_rows, batch_size=3000, ignore_conflicts=True, ) redis_activity_table = "activity" redis_location_table = "location" redis_activity_column = [ "id", "code", "name", "desc", "planned_size", "real_size", "slot_required","activity_type_id", "duration", "suggested_day", "suggested_time_slot","suggested_time","scheduled_start_slot", "department_id", "module_id", "academic_term_id", "activity_template_id", "week_pattern_id", "week_pattern", "availability_id", "availability_pattern","start_preference_id", "start_preference_pattern", "usage_preference_id", "usage_preference_pattern","student_sets", "staff_requirement_type","staff_required_no", "staff_preset", "staff_suitability_id", "location_requirement_type","location_required_no", "location_preset","location_suitability_id", "is_jta", "jta_parent_id", "is_variant", "variant_parent_id", "status", "scheduled" ] redis_data = { "insert": {}, "update": {}, } created_relation = {} if to_create: for activity in to_create: if not activity.id: continue redis_activity_data = {} week_ids = activity_week_response_map.get(activity.id, []) week_pattern_for_redis = convert_to_redis_week(week_ids) if week_ids else [] created_relation[activity.id] = week_ids for column in redis_activity_column: match column: case "week_pattern": redis_activity_data[column] = week_pattern_for_redis or [] case "staff_suitability_id" | "location_suitability_id": redis_activity_data[column] = None case "staff_preset" | "location_preset": redis_activity_data[column] = [] case "student_sets": redis_activity_data[column] = [] case "staff_required_no": redis_activity_data[column] = activity.staff_requirement case "location_required_no": redis_activity_data[column] = activity.location_requirement case "usage_preference_id" | "availability_id" | "start_preference_id" | "week_pattern_id": value = getattr(activity, column) redis_activity_data[column] = 0 if value in (None, "") else value case _: value = getattr(activity, column) redis_activity_data[column] = self.format_time_if_needed(value) redis_data["insert"].setdefault(redis_activity_table, []).append(redis_activity_data) # update location resource map redis because booking import already scheduled location location_with_affected_week = { location_id: week_ids for location_id, week_ids in location_with_affected_week.items() if week_ids } location_with_affected_week = {k: v for k, v in location_with_affected_week.items() if v} # since got update student set, this helper already cater if got update student set, no need pass in student details also will update student resource map redis_data = helper_recalculate_and_get_redis_resource_map( redis_data=redis_data, location_with_affected_week=location_with_affected_week, slot_per_week=slot_per_week, ) if redis_data["insert"] or redis_data["update"]: bulk_sync_to_redis(redis_client, redis_data) create_method = "booking_create" kafka_topic = self.kafka_config["MICROSERVICES_TT_TOPIC"] ignore_fields = [ "created_at", "updated_at", "created_by", "updated_by", # "availability_pattern", # "usage_preference_pattern", # "start_preference_pattern", ] kafka_activity = [] kafka_booking_schedule = [] created_ids = [activity.id for activity in to_create if activity.id] if created_ids: created_data = TtActivity.objects.filter(id__in=created_ids).select_related("department","activity_type").prefetch_related("week","location") # Build location map: {activity_id: [location_id]} activity_location_map = {} for code, relation in relation_map.items(): activity_id = created_activity_map.get(code) location_id = relation.get("location_id") if activity_id and location_id: activity_location_map.setdefault(activity_id, []).append(location_id) for activity in created_data: temp = {} for f in TtActivity._meta.fields: if f.name in ignore_fields: continue if isinstance(f, ForeignKey): value = getattr(activity, f"{f.name}_id") if value is not None: temp[f.name] = value else: value = getattr(activity, f.name) if value is not None: if isinstance(value, time): temp[f.name] = value.strftime("%H:%M") else: temp[f.name] = value temp["week"] = created_relation.get(activity.id, []) or [] temp["location"] = activity_location_map.get(activity.id, []) or [] kafka_activity.append(temp) #scheduled scheduled_data = {} final_data = { "code": activity.code, "name": activity.name, "desc": activity.desc, "department": activity.department.name if activity.department else None, "department_id": activity.department_id, "planned_size": activity.planned_size, "real_size": 0, "activity_type_id": activity.activity_type_id, "activity_type": activity.activity_type.name if activity.activity_type else None, "activity_type_color": activity.activity_type.color if activity.activity_type else None, "scheduled_start_time": str(activity.scheduled_start_time), "scheduled_start_slot": activity.scheduled_start_slot, "scheduled_day": __( "attr.days_name." + str(activity.scheduled_day)) if activity.scheduled_day is not None else None, "duration": activity.duration, "slot_required": activity.slot_required, "location": [l.name for l in activity.location.all()], "staff": [], } final_data['id'] = activity.id final_data['staff_ids'] = [] final_data['location_ids'] = activity_location_map.get(activity.id, []) or [] final_data['scheduled_day'] = activity.scheduled_day if activity.scheduled_day is not None else None final_data['week_ids'] = created_relation.get(activity.id, []) or [] kafka_booking_schedule.append(final_data) if kafka_activity: kafka_request_data = { "session_id": request.user.name, "activity": kafka_activity, } send_request(kafka_topic, kafka_request_data, None, create_method) if kafka_booking_schedule: method = "booking_schedule" kafka_request_data = { "session_id": request.user.name, "activity": kafka_booking_schedule, } send_request(kafka_topic, kafka_request_data, None, method) return self.import_return(rows_read, inserted, updated, rows_skipped, errors) def post(self, request): # get importable tables from settings setting_params = { "slot_per_week", "import_tables", "minute_per_slot", "slot_per_day" } settings = TtSetting.get_multiple_setting(setting_params) if settings.get("import_tables"): import_tables = json.loads(settings.get("import_tables")) else: return self.api_response(error=__(self.BASE_TRANS+"no_importables")) # get file for import f = request.FILES.get("import_file") if not f: return self.api_response({"error": __(self.BASE_TRANS+"no_file")}) response = {} # read excel for the datas needed for sheet_name, header, rows in self.iter_sheets_from_upload(f): if sheet_name in import_tables: import_function_name = f"import_{sheet_name}" try: settings_needed = ["tt_staff", "tt_location", "tt_student_set", "tt_module", "tt_activity_template", "tt_activity", "tt_booking", "tt_student_academic_term"] # print(f"Start: ({datetime.now()})") if sheet_name in settings_needed: response[sheet_name] = getattr(self, import_function_name)(request, header, rows, settings) else: response[sheet_name] = getattr(self, import_function_name)(request, header, rows) # print(f"{sheet_name}: {len(connection.queries)} ({datetime.now()})") connection.queries_log.clear() except Exception as e: # import function not found response[sheet_name] = self.import_return(errors=["Function in progress"]) # hardcode e_details = get_exception_detail(e) log_critical_error(user_id=None,descr=e_details['descr'],url=e_details['url'],trace=e_details['trace']) continue else: # sheet_name not found response[sheet_name] = self.import_return(errors=__(self.BASE_TRANS+"sheet_name_not_found")) continue return self.api_response(data=response)