import datetime import json import re import traceback from collections import defaultdict import requests from Crypto.Cipher import AES from Crypto.Util.Padding import pad, unpad import base64 from django.conf import settings from django.db import connection, transaction, models, reset_queries from django.db.models.functions import Cast, Replace, Substr from django.forms.models import model_to_dict from rest_framework.exceptions import APIException from api.models import TtSetting, TtStaffResourceMap, TtLocationResourceMap, TtStudentSetResourceMap, \ TtStudentResourceMap, TtActivity, TtStudent, TtStudentSet from api.models.error_log import ErrorLog from django.apps import apps from django.db.models import ForeignKey, Value, Prefetch import hashlib import logging diagnostic_logger = logging.getLogger("api.admin.diagnostics") def encrypt_aes_128_cbc(text: str, key=None, iv=None) -> str: if not key: key = settings.PWD_ENCRYPTION_KEY if not iv: iv = settings.PWD_ENCRYPTION_IV cipher = AES.new(key, AES.MODE_CBC, iv) padded = pad(text.encode(), AES.block_size) return base64.b64encode(cipher.encrypt(padded)).decode() def decrypt_aes_128_cbc(enc_text: str, key=None, iv=None) -> str: if not key: key = settings.PWD_ENCRYPTION_KEY if not iv: iv = settings.PWD_ENCRYPTION_IV cipher = AES.new(key, AES.MODE_CBC, iv) decrypted = unpad(cipher.decrypt(base64.b64decode(enc_text)), AES.block_size) return decrypted.decode() def get_ip(request): x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: # Take first IP from comma-separated list return x_forwarded_for.split(',')[0].strip() return request.META.get('REMOTE_ADDR') def throw_validation_error(code=None, error=None): e = APIException() e.status_code = code e.detail = { "code": code, "error": error } raise e def log_critical_error(user_id=None, descr="", url="", trace: list = None, data: list = None): from api.services.diagnostics import ( buffer_critical_error, current_admin_diagnostic, current_diagnostic_identifiers, sanitize_diagnostic_value, ) diagnostic = current_admin_diagnostic() if diagnostic is not None and connection.in_atomic_block: buffered = buffer_critical_error( user_id=user_id, descr=descr, source=url, trace=trace, data=data, ) diagnostic_logger.error( "admin_critical_error_buffered", extra={ "request_id": diagnostic.request_id, "correlation_id": diagnostic.correlation_id, "source": str(url or "")[:255], "error_sha256": buffered.sha256() if buffered else None, }, ) return None if isinstance(trace, (list, dict)): trace = json.dumps(trace) if isinstance(data, (list, dict)): data = sanitize_diagnostic_value(data) request_id, correlation_id = current_diagnostic_identifiers() return ErrorLog.insert_log( user_id, descr, trace, data, url, request_id=request_id, correlation_id=correlation_id, ) def get_exception_detail(e): descr = str(e) url = e.__traceback__.tb_frame.f_code.co_filename trace = "".join(traceback.format_exception(type(e), e, e.__traceback__)) return { "descr": descr, "url": url, "trace": trace } def get_old_new_data(model, old_data, ignore_fields=None): """ Compare current instance with old_data snapshot. old_data must use model_to_dict be4 pass in """ ignore_fields = ignore_fields or [] changes = {"old_data": {}, "new_data": {}} # Get current state new_data = model_to_dict(model) # combine both old and new data key, so even old data is not set, but new data have set, it also can store the value for field in set(old_data.keys()) | set(new_data.keys()): if field in ignore_fields: continue old_value = old_data.get(field, "") new_value = new_data.get(field, "") if old_value != new_value: # make sure is not in datetime/date/time format in the value, if not when store to json field, will have error if isinstance(old_value, (datetime.date, datetime.time, datetime.datetime)): old_value = old_value.isoformat() if isinstance(new_value, (datetime.date, datetime.time, datetime.datetime)): new_value = new_value.isoformat() changes["old_data"][field] = old_value changes["new_data"][field] = new_value return changes def get_multiple_old_new_data(models, request_data, ignore_field, valid_fields): old_new_data = { "old_data": {}, "new_data": {} } ignore_field.append("password") for model in models: changes_new = {} changes_old = {} for key, new_val in request_data: if key not in ignore_field and key in valid_fields: old_val = getattr(model, key) if old_val != new_val: # only log changed fields # make sure is not in datetime/date/time format in the value, if not when store to json field, will have error if isinstance(old_val, (datetime.date, datetime.time, datetime.datetime)): old_val = old_val.isoformat() if isinstance(new_val, (datetime.date, datetime.time, datetime.datetime)): new_val = new_val.isoformat() changes_old[key] = old_val changes_new[key] = new_val if changes_new: old_new_data["old_data"][model.id] = changes_old old_new_data["new_data"][model.id] = changes_new return old_new_data def get_add_new_data(model,skip_empty=True): """ Return audit data when inserting a new record """ #exclude many to many field, cause i need get, will run extra query to perform it exclude_fields = [f.name for f in model._meta.many_to_many] data = model_to_dict(model, exclude=exclude_fields) data = { key: val.isoformat() if isinstance(val, (datetime.date, datetime.time, datetime.datetime)) else val for key, val in data.items() } if skip_empty: data = { key: val for key, val in data.items() if val not in (None, "", [], {}) } return { "table": model._meta.db_table, "data": data, } def search_old_new_data(old_new_data, search_key): """ mostly used for find out the id need update to redis :param old_new_data: data form by get_multiple_old_new_data or get_old_new_data :param search_key: the key want to get """ # using set() can get unique id, if direct use list, need extra handle to filter for unique id ids = set() # check old and new data, combine the suitability id and get relation from db if old_new_data['old_data']: for key, obj in old_new_data['old_data'].items(): if search_key in obj: val = obj.get(search_key, []) if isinstance(val, (list, tuple, set)): ids.update(val) elif val is not None: # int, str, etc. ids.add(val) if old_new_data['new_data']: for key, obj in old_new_data['new_data'].items(): if search_key in obj: val = obj.get(search_key, []) if isinstance(val, (list, tuple, set)): ids.update(val) elif val is not None: # int, str, etc. ids.add(val) return list(ids) def bulk_sync_to_redis(redis_client, data, chunk_size=500): """ :param redis_client: Redis client instance :param data: dict with keys "insert", "update", "delete" :param chunk_size: number of commands per pipeline execution """ from copy import deepcopy from django.db import transaction if transaction.get_connection().in_atomic_block: transaction.on_commit( lambda: bulk_sync_to_redis(redis_client, deepcopy(data), chunk_size), robust=True, ) return try: # record down all redis command need to run commands = [] # Prepare insert commands for model_name, objects in data.get("insert", {}).items(): for obj in objects: key = f"{model_name}:{obj['id']}" commands.append(("json_set", key, "$", obj, {})) # Prepare update commands for model_name, objects in data.get("update", {}).items(): for obj in objects: key = f"{model_name}:{obj['id']}" for field, value in obj.items(): if field != "id": # 2026-03-12 remove xx=True so can handle field not found auto add, key not found problem solve in pipe.execute(raise_on_error=False) commands.append(("json_set", key, f"$.{field}", value,{})) # Prepare delete commands for model_name, ids in data.get("delete", {}).items(): for obj_id in ids: key = f"{model_name}:{obj_id}" commands.append(("delete", key)) # Execute commands in chunks for i in range(0, len(commands), chunk_size): with redis_client.pipeline() as pipe: for cmd in commands[i:i+chunk_size]: if cmd[0] == "json_set": pipe.json().set(cmd[1], cmd[2], cmd[3], **cmd[4]) elif cmd[0] == "delete": pipe.delete(cmd[1]) results = pipe.execute(raise_on_error=False) except Exception as e: # Log the error e_details = get_exception_detail(e) log_critical_error(user_id=None, descr=e_details['descr'], url=e_details['url'], trace=e_details['trace']) def calculate_preset_pattern_time(type,pattern,minute_per_slot,slot_per_day): """ curently used for preset availability, start_preference, usage_preference only :param type: only accept availability,start_preference,usage_preference :param pattern: pattern get from db :param minute_per_slot: get from setting table :return: """ start_time = None end_time = None slot_per_day = int(slot_per_day) minute_per_slot = int(minute_per_slot) # seperate the pattern(this should be 1 week pattern) become daily daily_patterns = [pattern[i:i + slot_per_day] for i in range(0, len(pattern), slot_per_day)] # must correct type only will check, if not direct throw error, cause by right wont have this problem when coding match type: case "usage_preference" | "start_preference": value_check = "5" # in string type, so can match compare with pattern string case "availability": value_check = "1" # in string type, so can match compare with pattern string for pattern in daily_patterns: # reset current_time every day # only datetime value can using timedelta to do calculation, will ignore the date current_datetime = datetime.datetime.strptime("00:00", "%H:%M") delta = datetime.timedelta(minutes=minute_per_slot) for key, char in enumerate(pattern): current_time = current_datetime.time() if char != value_check: # need preset a value if is None, if not got error if start_time is None: start_time = current_time if end_time is None: end_time = current_time if current_time < start_time: start_time = current_time if current_time > end_time: end_time = current_time current_datetime = current_datetime + delta return { "start_time": start_time, "end_time": end_time, } def get_auto_generate_name(name: str,prefix,leading_zero_length): """ when name or any other column need the auto generate data, will pass in last name using, this function will increase 1 Increment a code like 'A000001' or 'AB0099' → 'A000002', 'AB0100'. Keeps prefix and leading zeros. """ if not name: # first code return f"{prefix}{'0' * (leading_zero_length - 1)}1" else: pattern = f"^{re.escape(prefix)}(\\d+)$" match = re.match(pattern, name) if match: num = match.group(1) new_number = int(num) + 1 return f"{prefix}{new_number:0{leading_zero_length}d}" # no name found return default name only, cause return false will have error return name def get_auto_generate_code(code: str,prefix,leading_zero_length): """ Generates the next sequence code (e.g., 'MV0000000362 002' -> 'MV0000000363'). Extracts exactly `leading_zero_length` digits following `prefix` and increments by 1. """ default_first_code = f"{prefix}{1:0{leading_zero_length}d}" if not code: return default_first_code # Extract exact digits directly following prefix (ignores spaces/suffixes) pattern = f"^{re.escape(prefix)}(\\d{{{leading_zero_length}}})" match = re.match(pattern, str(code).strip()) if match: next_num = int(match.group(1)) + 1 return f"{prefix}{next_num:0{leading_zero_length}d}" # if got any problem, return the default code return default_first_code # no name found return default name only, cause return false will have error return False def get_clean_activity_name(name): """ mostly for activity name, cause the name will end with (JTA) or <19,23-45> """ match = re.match(r'^(.*?/)(\d+)', name) if match: return match.group(1) + match.group(2) return name def get_related_fk_table_ids(related_model_name, ids, redis_format = False, return_column="id"): """ this function used to get all having FK of related_model_name table, the main table id, eg. related_model_name = TtDepartment, ids = [1,2], so it will return all table record of id having FK of TtDepartment with id [1,2] related_model_name is the name in model like TtDepartment ids must be array like [1,2,3] even 1 id only also need put [1] redis_format: if True, the return table name will remove prefix tt_ return_column: default will return id, some case will return by other column Returns a dict: {ModelName: queryset of ids} """ results = {} return_data = {} for model in apps.get_models(): for field in model._meta.get_fields(): if isinstance(field, ForeignKey) and field.related_model.__name__ == related_model_name: # Use __in to filter by list of ids qs = model.objects.filter(**{f"{field.name}__in": ids}) if qs.exists(): field_names = {f.name for f in model._meta.get_fields()} # incase the custom column not found, will return by id if return_column not in field_names: final_column = "id" else: final_column = return_column results[model._meta.db_table] = list(qs.values_list(final_column, flat=True)) if results and redis_format: for key,val in results.items(): key_name = key.removeprefix(settings.TIMETABLER_TABLE_PREFIX) return_data[key_name] = val else: return_data = results return return_data def get_missing_code_from_db(prefix,table_name,leading_zero_length): """ query explain SELECT CAST(REPLACE(code, '{prefix}', '') AS BIGINT) AS num: remove prefix and convert to int SELECT generate_series(1, (SELECT MAX(num) FROM codes)) AS n: from codes, form a data like [1,2,3,4,5,...all other code] SELECT '{prefix}' || LPAD(n::text, 10, '0') AS missing_code: result will add back the prefix like "MV0000000008" WHERE c.num IS NULL: when Null means is missing code leading_zero_length = need make number have how many character, eg if 5, when number is 1 means will be 00001, if 9999, will be 09999 """ total_expected_length = len(prefix) + leading_zero_length prefix_len = len(prefix) + 1 # SQL SUBSTRING is 1-indexed # must start with prefix, and after prefix until the length of leading zero length, the character must be all int in 0-9 strict_regex_pattern = f"^{prefix}[0-9]{{{leading_zero_length}}}" query = f""" WITH codes AS ( SELECT CAST( SUBSTRING(TRIM(code) FROM {prefix_len} FOR {leading_zero_length}) AS BIGINT ) AS num FROM {table_name} WHERE TRIM(code) LIKE %s AND LENGTH(TRIM(code)) >= {total_expected_length} AND TRIM(code) ~ %s ), numbers AS ( SELECT generate_series(1, COALESCE((SELECT MAX(num) FROM codes), 0)) AS n ) SELECT %s || LPAD(n::text, {leading_zero_length}, '0') AS missing_code FROM numbers LEFT JOIN codes c ON c.num = n WHERE c.num IS NULL ORDER BY n """ with connection.cursor() as cursor: cursor.execute(query, [f"{prefix}%", strict_regex_pattern, prefix]) rows = cursor.fetchall() if not rows: return [] return [row[0] for row in rows] def get_missing_name(name_list, prefix, leading_zero_length: int): """ Get missing codes for a given prefix, this function wont go db, so different from get_missing_code_from_db. Parameters: - name_list: list of existing codes/names - prefix: prefix to filter - leading_zero_length: need make number have how many character, eg if 5, when number is 1 means will be 00001, if 9999, will be 09999 Returns: - List of missing full name with prefix """ filtered = [n for n in name_list if n.startswith(prefix)] numbers = [] for n in filtered: tail = n.rsplit('/', 1)[-1] match = re.search(r'(\d+)', tail) # first continuous number if match: numbers.append(int(match.group(1))) if numbers: max_num = max(numbers) number_set = set(numbers) missing_numbers = [i for i in range(1, max_num) if i not in number_set] missing_name = [f"{prefix}{str(i).zfill(leading_zero_length)}" for i in missing_numbers] return missing_name return [] def bitwise_calculator_from_string(string1,string2): """ string1 = "10011" string2 = "01001" will return "11011" in string format """ # convert to int base 2 num1 = int(string1, 2) num2 = int(string2, 2) # bitwise OR merged_num = num1 | num2 # convert back to string, keeping original length merged_str = bin(merged_num)[2:].zfill(len(string1)) return merged_str def get_slot_default_pattern(slot_per_week=None): if slot_per_week is None: setting_params = { "slot_per_week", } settings = TtSetting.get_multiple_setting(setting_params) slot_per_week = int(settings["slot_per_week"]) return { "availability": "0" * int(slot_per_week), "start_preference": "5" * int(slot_per_week), "usage_preference": "5" * int(slot_per_week), "resource_map": "0" * int(slot_per_week) } def time_to_slot(time,minute_per_slot=None): if minute_per_slot is None: setting_params = { "minute_per_slot", } settings = TtSetting.get_multiple_setting(setting_params) minute_per_slot = int(settings["minute_per_slot"]) return (time.hour * 60 + time.minute) // int(minute_per_slot) def convert_to_redis_week(weeks): """ Convert 1-based week numbers to 0-based for Redis. Example: [1,2,3,5] -> [0,1,2,4] Reason: scheduling part to read redis is using 0-based for the week, so 0 means week 1, 1 means week 2 """ return sorted([w - 1 for w in weeks if w is not None and w > 0]) def recalculate_resource_map(type, id_with_week, slot_per_week): """ Used pure query to recalculate the latest resource map for staff/location/student_set Bypasses the PostgreSQL stack depth limit by using array UNNEST execution. :param type: accept staff/location/student_set :param id_with_week: eg. { 1: [1, 2], # staff_id 1 → week 1, 2 2: [1, 4], # staff_id 2 → week 1, 4 } :param slot_per_week: total capacity bits in a single week grid """ activity_table = "tt_activity" match type: case "staff": activity_relation_table = "tt_activity_staff" resource_map_table = "tt_staff_resource_map" case "location": activity_relation_table = "tt_activity_location" resource_map_table = "tt_location_resource_map" case "student_set": activity_relation_table = "tt_student_set_activity" resource_map_table = "tt_student_set_resource_map" case _: # invalid type just return False return False # Flatten the dict structure into tuples week_pairs = sorted({ (entity_id, week_id) for entity_id, weeks in id_with_week.items() for week_id in weeks }) # Safety check, if nothing to process return False if not week_pairs: return False # Unzip the tuples into two distinct flat lists for Postgres Arrays # Example: [(1, 10), (1, 11), (2, 10)] -> entity_ids=[1, 1, 2], week_ids=[10, 11, 10] entity_ids, week_ids = zip(*week_pairs) slot_per_week = int(slot_per_week) default_resource_map = "0" * slot_per_week with transaction.atomic(): with connection.cursor() as cursor: cursor.execute(f""" WITH targets AS ( -- 1. Unpack arrays directly into an in-memory execution table (O(1) stack depth) SELECT u.entity_id, u.week_id FROM UNNEST(%s::int[], %s::int[]) AS u(entity_id, week_id) ), activity_masks AS ( -- 2. Calculate patterns ONLY for targeted weeks that have activities SELECT ar.{type}_id AS entity_id, COALESCE(wpw.week_id, aw.week_id) AS week_id, bit_or( ( overlay( repeat('0', {slot_per_week}) placing repeat('1', a.slot_required) from a.scheduled_start_slot + 1 )::bit({slot_per_week}) ) ) AS new_pattern FROM {activity_relation_table} ar JOIN {activity_table} a ON a.id = ar.activity_id AND a.scheduled = 1 LEFT JOIN tt_week_pattern_week wpw ON a.week_pattern_id = wpw.week_pattern_id LEFT JOIN tt_activity_week aw ON aw.activity_id = a.id AND a.week_pattern_id IS NULL -- 3. Use an internal JOIN against our memory table for index optimization JOIN targets t ON ar.{type}_id = t.entity_id AND COALESCE(wpw.week_id, aw.week_id) = t.week_id GROUP BY ar.{type}_id, COALESCE(wpw.week_id, aw.week_id) ) -- 4. Fast targeted update using the clean LEFT JOIN mapping UPDATE {resource_map_table} rm SET pattern = COALESCE(am.new_pattern, B'{default_resource_map}') FROM targets t LEFT JOIN activity_masks am ON t.entity_id = am.entity_id AND t.week_id = am.week_id WHERE rm.{type}_id = t.entity_id AND rm.week_id = t.week_id; """, [list(entity_ids), list(week_ids)]) return True def recalculate_student_resource_map(student_ids, week_ids, slot_per_week): """ only for student resource map will based on student in the student set, combine all resource map from student allocated student set week_ids pass in for this function means this week must be update already, so if not found student set in this week for student, will update to default pattern """ with transaction.atomic(): slot_per_week = int(slot_per_week) default_resource_map_pattern = "0" * slot_per_week # get related student with all student set first, later will have checking to see need update which student set students = TtStudent.objects.filter(id__in=student_ids,).prefetch_related("student_set") # this function wont create new from db, it only perform update, so get existing db record first, later will do compare existing_db_keys = set( TtStudentResourceMap.objects.filter( student_id__in=student_ids, week_id__in=week_ids ).values_list("student_id", "week_id") ) # get student set resource map and format it become key by student set id resource_map_values = ( TtStudentSetResourceMap.objects.filter( student_set__student__id__in=student_ids, week_id__in=week_ids ) .values("student_set_id", "week_id", "pattern") .distinct() ) # reformat student set resource map data student_set_resource_maps = defaultdict(list) for row in resource_map_values: student_set_resource_maps[row["student_set_id"]].append(row) student_set_resource_maps = dict(student_set_resource_maps) student_resource_map_update_data = [] for student in students: student_resource_map = {} # default set be like all week_id: default_pattern, then have a loop to see which week_id pattern need to be replace for week_id in week_ids: student_resource_map[week_id] = default_resource_map_pattern student_sets = student.student_set.all() for student_set in student_sets: current_student_set_resource_maps = student_set_resource_maps.get(student_set.id, []) for current_student_set_resource_map in current_student_set_resource_maps: target_week_id = current_student_set_resource_map["week_id"] if target_week_id in student_resource_map: # Convert binary strings to integers student_pattern_int = int(student_resource_map[target_week_id], 2) student_set_pattern_int = int(current_student_set_resource_map["pattern"], 2) # Bitwise OR operation (0+1=1, 1+1=1, 0+0=0) merged_pattern_int = student_pattern_int | student_set_pattern_int # Convert back to zero-padded binary string matching slot_per_week length student_resource_map[target_week_id] = f"{merged_pattern_int:0{slot_per_week}b}" # format the data for bulk_update for week_id,pattern in student_resource_map.items(): # if have any student_id + week_id is not found in db, no need to create it if (student.id, week_id) in existing_db_keys: student_resource_map_update_data.append( TtStudentResourceMap( student_id=student.id, week_id=week_id, pattern=pattern, ) ) if student_resource_map_update_data: TtStudentResourceMap.objects.bulk_create( student_resource_map_update_data, update_conflicts=True, update_fields=["pattern"], unique_fields=["student_id", "week_id"], ) def get_resource_map_redis_data(type,ids): """ :param type: accept staff/location/student_set :param ids: affected id in array eg.[1,2,3] :return: """ match type: case "staff": main_table = TtStaffResourceMap column_name = "staff_id" related_fk = "staff" activity_relation_name = "ttactivity_set" case "location": main_table = TtLocationResourceMap column_name = "location_id" related_fk = "location" activity_relation_name = "ttactivity_set" case "student_set": main_table = TtStudentSetResourceMap column_name = "student_set_id" related_fk = "student_set" activity_relation_name = "activity" case "student": main_table = TtStudentResourceMap column_name = "student_id" related_fk = "student" activity_relation_name = "student_set__activity" case _: # invalid type just return False return False # reduce the memory, no need get unused data reset_queries() only_fields = [ "id", column_name, "pattern", "week", "week__week", related_fk, f"{related_fk}__id" ] resource_map_data = ( main_table.objects .filter(**{f"{column_name}__in": ids}) .select_related("week",related_fk) .prefetch_related( Prefetch(f"{related_fk}__{activity_relation_name}", queryset=TtActivity.objects.filter(scheduled=1).only("id")) ) .only(*only_fields) .order_by("week_id") ) """ sample format [ { "id": 1, "resource_map": { "1": 000000, "2": 000111, } }, { "id": 2, "resource_map": { "1": 000000, "2": 000111, } } ] """ return_data = {} for val in resource_map_data: id = getattr(val, column_name) redis_week = int(val.week.week) - 1 pattern = val.pattern if id not in return_data: related_obj = getattr(val, related_fk) activity_ids = set() # got related_fk obj only add activity_ids into redis, else just make it no [] if related_obj: # student activity is get from student set activity, so need extra handle for get student activity data if type == "student": for ss in related_obj.student_set.all(): for act in ss.activity.all(): activity_ids.add(act.id) else: activities = getattr(related_obj, activity_relation_name) for act in activities.all(): activity_ids.add(act.id) return_data[id] = { "id": id, "allocated_activity_ids": sorted(list(activity_ids)), "resource_map": {} } return_data[id]["resource_map"][redis_week] = pattern # sort the resource map by key ASC for item in return_data.values(): item["resource_map"] = dict( sorted( item["resource_map"].items(), key=lambda x: int(x[0]) ) ) return list(return_data.values()) def push_websocket_notification(url,data,socket_id): from copy import deepcopy from django.db import transaction if transaction.get_connection().in_atomic_block: transaction.on_commit( lambda: push_websocket_notification(url, deepcopy(data), socket_id), robust=True, ) return try: payload = { "socket_id": socket_id, "data": data } res = requests.post(url, data=json.dumps(payload), headers={"Content-Type": "application/json"}, timeout=2 ) except Exception as e: e_details = get_exception_detail(e) log_critical_error(user_id=None, descr=e_details['descr'], url=e_details['url'], trace=e_details['trace']) def format_variant_activity_name_week_pattern_to_ranges(current_week_pattern, academic_term_week_pattern=None): """ :param current_week_pattern: week pattern array than have id,week key :param academic_term_week_pattern: academic term week array with id, week key :return: return data become <1-4,7,10-12> """ if not current_week_pattern: return "" if academic_term_week_pattern: # sort academic term weeks by real week order, then map week_id -> counter academic_term_week_pattern = sorted( academic_term_week_pattern, key=lambda w: w["week"] ) week_id_to_counter = { w["id"]: index for index, w in enumerate(academic_term_week_pattern, start=1) } weeks = sorted({ week_id_to_counter[w["id"]] for w in current_week_pattern if w["id"] in week_id_to_counter }) else: # fallback old behavior weeks = sorted({w["week"] for w in current_week_pattern}) if not weeks: return "" ranges = [] start = prev = weeks[0] for w in weeks[1:]: if w == prev + 1: prev = w else: ranges.append( str(start) if start == prev else f"{start}-{prev}" ) start = prev = w # close last range ranges.append( str(start) if start == prev else f"{start}-{prev}" ) return "<" + ",".join(ranges) + ">" def remove_variant_activity_name_range(name): # remove last <...> at the end, will used when variant again those variant record return re.sub(r"\s*<[^>]+>$", "", name) def get_next_available_code(model, field_name, prefix, leading_zero_length): """ Different from get_missing_code_from_db, this will get either missing or latest code from db Finds the first available gap in the sequence. Example: If DB has [MV0000000001, MV0000000002, MV0000000004], it returns 'MV0000000003'. """ # for search in db i want get exact this length of code, so can different out MV00001, MV0000000002 in db, if in this case will return MV0000000001 total_expected_length = len(prefix) + leading_zero_length existing_codes = ( model.objects.filter( **{f"{field_name}__startswith": prefix} ) .annotate( num_val=Cast( Substr(field_name, len(prefix) + 1, leading_zero_length), output_field=models.IntegerField(), ) ) .values_list("num_val", flat=True) .order_by("num_val") ) # 2. Find the first gap # We compare the actual list [1, 2, 4] vs a perfect sequence [1, 2, 3, 4] next_num = 1 for val in existing_codes: if val == next_num: next_num += 1 elif val > next_num: # We found a gap! break # 3. Format and return return f"{prefix}{next_num:0{leading_zero_length}d}" def create_resource_map_single(type, obj_id, week_ids, default_pattern = None): inserted = 0 week_placeholders = ",".join(["%s"] * len(week_ids)) sql = f""" INSERT INTO tt_{type}_resource_map ({type}_id, week_id, pattern, created_at) SELECT %s, w.id, %s, NOW() FROM tt_week w WHERE w.id IN ({week_placeholders}) """ params = [obj_id, default_pattern] + week_ids with transaction.atomic(), connection.cursor() as cursor: cursor.execute(sql, params) inserted = cursor.rowcount return inserted def create_resource_map_multiple(type, obj_ids, week_ids, default_pattern=None): if not obj_ids or not week_ids: return 0 inserted = 0 obj_placeholders = ",".join(["(%s)"] * len(obj_ids)) week_placeholders = ",".join(["%s"] * len(week_ids)) sql = f""" INSERT INTO tt_{type}_resource_map ({type}_id, week_id, pattern, created_at) SELECT obj.id, w.id, %s, NOW() FROM (VALUES {obj_placeholders}) AS obj(id) CROSS JOIN tt_week w WHERE w.id IN ({week_placeholders}) """ params = [default_pattern] + obj_ids + week_ids with transaction.atomic(), connection.cursor() as cursor: cursor.execute(sql, params) inserted = cursor.rowcount return inserted def create_resource_map_all(type, week_ids, default_pattern = None): if type not in {"staff", "location", "student_set"}: raise ValueError("Invalid type") inserted = 0 week_placeholders = ",".join(["%s"] * len(week_ids)) sql = f""" INSERT INTO tt_{type}_resource_map ({type}_id, week_id, pattern, created_at) SELECT s.id, w.id, %s, NOW() FROM tt_{type} s JOIN tt_week w ON w.id IN ({week_placeholders}) """ params = [default_pattern] + week_ids with transaction.atomic(), connection.cursor() as cursor: cursor.execute(sql, params) inserted = cursor.rowcount return inserted def check_resource_map_available(resource_maps,start_slot,slot_required, exclude_start=None, exclude_end=None): """ :param resource_maps: data format in {1:"00000000000",2:"0000111011100",3:"01010101010101"} :param start_slot: integer get from db column scheduled_start_slot :param slot_required: integer get from db column slot_required :param exclude_start: start of the slot need to exclude :param exclude_end: end of the slot need to exclude exclude_start and exclude_end must used tgt, so if start is 10, end is 13, means if 10-13 slot is 1, will be ignore use case is want move activity start from 10:00 to 10:30 in same staff and location, so need used this exclude part to exclude from current pattern :return: True or False """ # if start_slot = 15 and slot_required = 2, the end slot will be 17 end_slot = start_slot + slot_required for week_id, pattern in resource_maps.items(): target_slice = pattern[start_slot:end_slot] if len(target_slice) < slot_required: return False if "1" in target_slice: # If no exclusion, any '1' is an instant fail if exclude_start is None or exclude_end is None: return False # if got exclude, will check is this "1" is in the range of exclude, if yes, will ignore it, cause might be move slot 10-13 to 11-14 only for i in range(start_slot, end_slot): # even pattern is str, can get by pattern[0] will get first character if i < len(pattern) and pattern[i] == "1": # so if "i" not between exclude start and end, return False if not (exclude_start <= i <= exclude_end): return False return True def check_availability_available(pattern, start_slot, slot_required, old_start=None, old_end=None): # Calculate the boundaries of the new requested position new_end = start_slot + slot_required # 1. Slice the pattern for the new requested range target_slice = pattern[start_slot:new_end] # 2. Safety check: ensure the pattern isn't shorter than the required time if len(target_slice) < slot_required: return False # 3. If there are '1's in the way, we check if they are "self-conflicts" if "1" in target_slice: # Check every individual slot in the requested range for i in range(start_slot, new_end): if i < len(pattern) and pattern[i] == "1": # If we have an 'old' position to ignore, check if this '1' is inside it if old_start is not None and old_end is not None: if old_start <= i < old_end: continue # This '1' is the activity itself! Skip to next slot. # If it's a '1' and NOT in the ignore range, it's a real conflict return False return True def fetch_code_id_map(model, code_field, codes): if not codes: return {} return dict( model.objects .filter(**{f"{code_field}__in": list(codes)}) .values_list(code_field, "id") ) # for insert/creating data def generate_kafka_data(new_data, relation_fields = []): if not new_data: return {} kafka_data = new_data.get("data", {}).copy() kafka_data.pop('created_by', None) for field in relation_fields: value = new_data.get(field) if value: kafka_data[field] = value return kafka_data # return m2m / relation data def generate_kafka_data_update(new_data, relation_fields = []): if not new_data: return {} first_data = next(iter(new_data.values()), {}) shared_fields = { field: first_data[field] for field in relation_fields if first_data.get(field) } return shared_fields def get_resources_affected_weeks(activities=None, activity_ids=None): """ This function will used when have any resources resource_map need to update Reason have this function just reduce the repeating code when need this action can pass in either activities or activity_ids to do, activities will have higher priority @:param activities: tt_activity model obj with select related with week_pattern and prefetch related with "staff","location","student_set", "week_pattern__week","week" @:param activity_ids: list of activity id """ return_data = { "staff_with_affected_week": {}, "location_with_affected_week": {}, "student_set_with_affected_week": {}, "staff_ids": set(), "location_ids": set(), "student_set_ids": set(), } if not activities and not activity_ids: return return_data if not activities: activities = ( TtActivity.objects.filter(id__in=activity_ids) .select_related("week_pattern") .prefetch_related( "staff", "location", "student_set", "week_pattern__week", "week" ) ) for activity in activities: if activity.week_pattern_id: activity_week_ids = [aw.id for aw in activity.week_pattern.week.all()] else: activity_week_ids = [aw.id for aw in activity.week.all()] for staff in activity.staff.all(): if staff.id not in return_data["staff_with_affected_week"]: return_data["staff_with_affected_week"][staff.id] = set() return_data["staff_with_affected_week"][staff.id].update(activity_week_ids) return_data["staff_ids"].add(staff.id) for location in activity.location.all(): if location.id not in return_data["location_with_affected_week"]: return_data["location_with_affected_week"][location.id] = set() return_data["location_with_affected_week"][location.id].update(activity_week_ids) return_data["location_ids"].add(location.id) for student_set in activity.student_set.all(): if student_set.id not in return_data["student_set_with_affected_week"]: return_data["student_set_with_affected_week"][student_set.id] = set() return_data["student_set_with_affected_week"][student_set.id].update(activity_week_ids) return_data["student_set_ids"].add(student_set.id) return_data["staff_with_affected_week"] = {k: v for k, v in return_data["staff_with_affected_week"].items() if v} return_data["location_with_affected_week"] = {k: v for k, v in return_data["location_with_affected_week"].items() if v} return_data["student_set_with_affected_week"] = {k: v for k, v in return_data["student_set_with_affected_week"].items() if v} return return_data def get_start_and_end_slot(pattern: str): """ Return the index of first 1 and index of last 1 """ first_one = pattern.find('1') # If find returns -1, it means '1' doesn't exist in the string if first_one == -1: return None, None last_one = pattern.rfind('1') return first_one, last_one def start_slot_end_slot_to_pattern(start_slot, end_slot, slot_per_week): if start_slot is None or end_slot is None: return "0" * slot_per_week # 2. Safety check: ensure indexes are within the total_count if end_slot >= slot_per_week: end_slot = slot_per_week - 1 leading_zeros = "0" * start_slot ones_block = "1" * (end_slot - start_slot + 1) trailing_zeros = "0" * (slot_per_week - end_slot - 1) return leading_zeros + ones_block + trailing_zeros def calculate_default_period(hours_per_week, minute_per_slot): if minute_per_slot is None: setting_params = { "minute_per_slot", } settings = TtSetting.get_multiple_setting(setting_params) minute_per_slot = int(settings["minute_per_slot"]) if hours_per_week is None: hours_per_week = 8 * 5 return int(hours_per_week) * 60 // int(minute_per_slot) def calculate_slot_by_day(days,slot_per_week,slot_per_day): if slot_per_week is None or slot_per_day is None: setting_params = { "slot_per_week", "slot_per_day", } settings = TtSetting.get_multiple_setting(setting_params) slot_per_week = int(settings["slot_per_week"]) slot_per_day = int(settings["slot_per_day"]) # make it in array first pattern_list = ["1"] * slot_per_week for day in days: if day <= 6: start_slot = day * slot_per_day # no need -1, because [0:48] will stop at 47 end_slot = start_slot + slot_per_day # if no put * slot_per_day, it cause 0-47 slot become only 0 with value 1 pattern_list[start_slot:end_slot] = ["0"] * slot_per_day pattern = "".join(pattern_list) return pattern def calculate_slot_by_start_time_slot(start_slot, slot_required, slot_per_week=None, slot_per_day=None): if slot_per_week is None or slot_per_day is None: setting_params = {"slot_per_week", "slot_per_day"} settings = TtSetting.get_multiple_setting(setting_params) slot_per_week = int(settings["slot_per_week"]) slot_per_day = int(settings["slot_per_day"]) pattern_list = ["1"] * slot_per_week for day_start in range(start_slot, slot_per_week, slot_per_day): end_slot = min(day_start + slot_required, slot_per_week) pattern_list[day_start:end_slot] = ["0"] * (end_slot - day_start) return "".join(pattern_list) def import_get_resource_map_redis_data(type, ids): """ Optimized version of get_resource_map_redis_data for import only, wont update allocated_activity_ids """ match type: case "staff": main_table = TtStaffResourceMap column_name = "staff_id" case "location": main_table = TtLocationResourceMap column_name = "location_id" case "student_set": main_table = TtStudentSetResourceMap column_name = "student_set_id" case "student": main_table = TtStudentResourceMap column_name = "student_id" case _: return False # using .values instead of only, so result is return by list and no need used extra time to build object resource_map_data = ( main_table.objects .filter(**{f"{column_name}__in": ids}) .select_related("week") .values(column_name, "week__week", "pattern") .order_by("week_id") ) return_data = {} for val in resource_map_data: obj_id = val[column_name] if not val["week__week"]: continue redis_week = int(val["week__week"]) - 1 pattern = val["pattern"] if obj_id not in return_data: return_data[obj_id] = { "id": obj_id, "resource_map": {} } return_data[obj_id]["resource_map"][redis_week] = pattern for item in return_data.values(): item["resource_map"] = dict( sorted( item["resource_map"].items(), key=lambda x: int(x[0]) ) ) return list(return_data.values()) def parse_boolean(value): value = str(value) if isinstance(value, bool): return value if isinstance(value, str): value = value.strip().lower() if value in ["true", "1", "yes"]: return True if value in ["false", "0", "no"]: return False return None