import datetime from api.models import User, Role, TtModuleGroup, TtDepartment, TtZone, TtTag, TtAcademicTerm, TtActivityTemplate, TtStaff, TtLocation, TtActivity, TtActivityType, TtModule, TtAvailability, TtStartPreference, TtUsagePreference, TtSuitability, TtResourceBreak from api.models import TtMaximumHour, TtMaximumWorkspan, TtFreeBlock, TtTravelTable, TtSetting from api.models import TtTimeBlock, TtStudentSet, TtPos, TtPosModuleGroup, TtPosModuleGroupModule, TtConstraintProfile from api.models import TtWeekPattern from django.core.exceptions import FieldError from api.translation import __ from api.utils import start_slot_end_slot_to_pattern def _build_field_changes(action, old_value, new_value, extra_data=None): """Flatten nested old/new payloads into field-level rows.""" result = [] if isinstance(old_value, list) or isinstance(new_value, list): old_list = old_value if isinstance(old_value, list) else [] new_list = new_value if isinstance(new_value, list) else [] # If this is a list of dictionaries, compare the records if all( isinstance(item, dict) for item in old_list + new_list ): max_length = max( len(old_list), len(new_list), ) for index in range(max_length): old_item = ( old_list[index] if index < len(old_list) else {} ) new_item = ( new_list[index] if index < len(new_list) else {} ) nested_changes = _build_field_changes( action, old_item, new_item, extra_data=extra_data ) if nested_changes: result.extend(nested_changes) return result return result old_dict = old_value.copy() if isinstance(old_value, dict) else {} new_dict = new_value.copy() if isinstance(new_value, dict) else {} if action in ["schedule", 'unschedule'] and ( "scheduled_start_time" in old_dict or "scheduled_start_time" in new_dict ): record_id = (extra_data or {}).get("record_id") if record_id: duration = ( TtActivity.objects .filter(id=record_id) .values_list("duration", flat=True) .first() ) if "scheduled_start_time" in old_dict: old_dict["scheduled_end_time"] = _calculate_scheduled_end_time( old_dict["scheduled_start_time"], duration, ) if "scheduled_start_time" in new_dict: new_dict["scheduled_end_time"] = _calculate_scheduled_end_time( new_dict["scheduled_start_time"], duration, ) old_has_slots = all(key in old_dict for key in ["start_slot", "end_slot"]) new_has_slots = all(key in new_dict for key in ["start_slot", "end_slot"]) if old_has_slots or new_has_slots: settings = TtSetting.get_multiple_setting({"slot_per_week"}) slot_per_week = int(settings["slot_per_week"]) if old_has_slots: old_dict["pattern"] = start_slot_end_slot_to_pattern( old_dict["start_slot"], old_dict["end_slot"], slot_per_week, ) if new_has_slots: new_dict["pattern"] = start_slot_end_slot_to_pattern( new_dict["start_slot"], new_dict["end_slot"], slot_per_week, ) keys = list(new_dict.keys()) keys_to_remove = [ "default_start", "default_usage", "default_availability", #academic term "generated_student_set_ids" ] if "jta" in action: keys_to_remove += [ "start_preference_pattern", "usage_preference_pattern", "availability_pattern", ] if "booking" in action: keys_to_remove += [ "week", "location_requirement_type", "staff_requirement_type", "staff_requirement", "location_requirement", ] # remain this key display for ui, easier for check # if "time_block_create" in action: # keys_to_remove += [ # "start_slot", "end_slot" # ] for key in keys_to_remove: if key in keys: keys.remove(key) # --------------------------------------------------------- # Process dictionary fields # --------------------------------------------------------- for key_index, key in enumerate(keys): old_item = old_dict.get(key) new_item = new_dict.get(key) if action == "variant_create" and key == "variant_parent": old_parent = old_item if isinstance(old_item, dict) else {} new_parent = new_item if isinstance(new_item, dict) else {} result.append({ "field": __("attr.variant_parent"), "field_key": "variant_parent", "old_value": old_parent.get("id"), "new_value": new_parent.get("id"), "old_value_text": old_parent.get("name"), "new_value_text": new_parent.get("name"), }) continue if isinstance(old_item, dict) or isinstance(new_item, dict): nested_extra_data = dict(extra_data or {}) nested_extra_data["record_id"] = key nested_changes = _build_field_changes( action, old_item, new_item, extra_data=nested_extra_data, ) if nested_changes: remark_names = extra_data.get("remark_name", []) if extra_data else [] remark_name = remark_names[key_index] if key_index < len(remark_names) else "" result.append({ "id": f"{key} - {remark_name}" if remark_name else str(key), "changes": nested_changes, }) # elif old_item != new_item: else: change = { "field": ( __(f"attr.{key}") if __(f"attr.{key}") != f"attr.{key}" else str(key) ), "field_key": str(key), } if key not in ["variant_parent", "id", "code", "name", "affected_week_ids"] and action in ["variant_create"]: #skip key all key continue elif key not in ["failed_count", "generated_count", "pathway_ids", "method"] and action in ["allocation"]: #skip key all key continue elif key in ["week", "affected_week_ids"]: change["weeks_value"] = _get_all_week_ids( action, old_dict, new_dict, extra_data, ) if key == "journey": old_value_text, new_value_text = _format_journey_values( old_item, new_item, ) elif key == "pos_module" and action in ["student_create", "student_update"]: old_value_text, new_value_text = _format_pos_module_values( old_item, new_item, ) else: old_value_text = format_field_value(action, str(key), old_item) new_value_text = format_field_value(action, str(key), new_item) change.update({ "old_value": old_item, "new_value": new_item, "old_value_text": old_value_text, "new_value_text": new_value_text, }) result.append(change) return result def build_field_changes(action, old_data, new_data, extra_data=None): """ Build normalized field changes. Special cases: - create actions: only use new_data["data"] - user_scheduling_preference_update: use the "type" field as the ID/key """ if extra_data is None: extra_data = {} if action == "activity_template_allocate": return _build_activity_template_allocate_changes(old_data, new_data) # --------------------------------------------------------- # CREATE actions # Move everything except "table" into "data" # --------------------------------------------------------- if action and ("create" in action.lower() or action in ["jta_split", "allocation", "pathway_generate", "pathway_delete"]): if isinstance(new_data, dict): data = new_data.get("data", {}) # If data already exists, keep it and add other fields if isinstance(data, dict): data = data.copy() for key, value in new_data.items(): if key not in ("data", "table"): data[key] = value # Get created record ID record_id = data.get("id") or "" if action == "jta_split": jta_data = old_data.get("jta_id") or {} record_id = jta_data.get("id") or "" remark_names = extra_data.get("remark_name") or [] remark_name = remark_names[0] if remark_names else "" if remark_name: record_id = f"{record_id} - {remark_name}" # Build field changes changes = _build_field_changes( action, old_data, data, extra_data=extra_data ) if record_id is not None: remark_name = data.get("name") or data.get("code") or "" return [ { "id": f"{record_id} - {remark_name}" if remark_name else str(record_id), "changes": changes, } ] return changes elif action in ["user_scheduling_preference_update", "user_constraint_setting_update"]: old_data = convert_preference_data(old_data) new_data = convert_preference_data(new_data) return _build_field_changes( action, old_data, new_data, extra_data=extra_data, ) def convert_preference_data(data): if not isinstance(data, list): return data result = {} for item in data: if not isinstance(item, dict): continue item_type = item.get("type") if item_type is None: continue result[__(f"attr.{item_type}")] = { key: value for key, value in item.items() if key != "type" } return result def _build_activity_template_allocate_changes(old_data, new_data): old_items = old_data if isinstance(old_data, list) else [] new_items = new_data if isinstance(new_data, list) else [] student_set_ids = { student_set_id for item in old_items + new_items if isinstance(item, dict) for student_set_id in item.get("student_set_ids", []) } student_set_names = { str(student_set_id): name for student_set_id, name in ( TtStudentSet.objects .filter(id__in=student_set_ids) .values_list("id", "name") ) } if student_set_ids else {} result = [] for index in range(max(len(old_items), len(new_items))): old_item = old_items[index] if index < len(old_items) else {} new_item = new_items[index] if index < len(new_items) else {} old_ids = old_item.get("student_set_ids", []) new_ids = new_item.get("student_set_ids", []) activity_name = ( new_item.get("activity_name") or old_item.get("activity_name") or str(index + 1) ) result.append({ "id": activity_name, "changes": [{ "field": __("attr.student_set_ids"), "field_key": "student_set_ids", "old_value": old_ids, "new_value": new_ids, "old_value_text": [ student_set_names.get(str(student_set_id), student_set_id) for student_set_id in old_ids ], "new_value_text": [ student_set_names.get(str(student_set_id), student_set_id) for student_set_id in new_ids ], }], }) return result def _calculate_scheduled_end_time(start_time, duration): if start_time is None or duration is None: return None try: if isinstance(start_time, str): start_time = datetime.time.fromisoformat(start_time) end_datetime = ( datetime.datetime.combine(datetime.date.min, start_time) + datetime.timedelta(minutes=int(duration)) ) return end_datetime.time().isoformat() except (TypeError, ValueError): return None def _get_all_week_ids(action, old_data, new_data, extra_data=None): # Resolve the academic term first. academic_term_id = ( new_data.get("academic_term_id") or old_data.get("academic_term_id") or new_data.get("academic_term")#variant create ) if not academic_term_id: target_model = _get_model_for_field(action, get_target_model(action)) record_id = ( (extra_data or {}).get("record_id") or new_data.get("id") or old_data.get("id") ) if target_model is TtAcademicTerm: academic_term_id = record_id elif record_id and target_model: try: academic_term_id = ( target_model.objects .filter(id=record_id) .values_list("academic_term_id", flat=True) .first() ) except FieldError: return [] if not academic_term_id: return [] # Select all weeks related to the resolved academic term. try: rows = list( TtAcademicTerm.objects .filter(id=academic_term_id) .order_by("week__week") .values( "week__id", "week__week", "week__start_date", ) ) except (FieldError, TypeError, ValueError): return [] weeks = [] for row in rows: week_id = row["week__id"] if week_id is None: continue start_date = row["week__start_date"] weeks.append({ "id": week_id, "week": row["week__week"], "start_date": start_date.isoformat() if start_date else None, }) return weeks def _format_journey_values(*values): zone_ids = { journey.get(zone_field) for value in values for journey in (value if isinstance(value, list) else [value]) if isinstance(journey, dict) for zone_field in ("from_zone_id", "to_zone_id") if journey.get(zone_field) is not None } zones_by_id = { str(zone_id): name for zone_id, name in ( TtZone.objects .filter(id__in=zone_ids) .values_list("id", "name") ) } if zone_ids else {} settings = TtSetting.get_multiple_setting({"minute_per_slot"}) minute_per_slot = int(settings["minute_per_slot"]) formatted_values = [] for value in values: if value is None: formatted_values.append(None) continue journeys = value if isinstance(value, list) else [value] formatted_values.append([ { __("attr.from_zone"): zones_by_id.get( str(journey.get("from_zone_id")), journey.get("from_zone_id"), ), __("attr.to_zone"): zones_by_id.get( str(journey.get("to_zone_id")), journey.get("to_zone_id"), ), __("attr.slot_required"): _format_slot_required( journey.get("slot_required"), minute_per_slot, ), } for journey in journeys if isinstance(journey, dict) ]) return formatted_values def _format_pos_module_values(*values): pos_module_ids = { item for value in values for item in (value if isinstance(value, list) else [value]) if item is not None } labels_by_id = { str(pos_module_id): module_name for pos_module_id, module_name in ( TtPosModuleGroupModule.objects .filter(id__in=pos_module_ids) .values_list("id", "module__name") ) } if pos_module_ids else {} formatted_values = [] for value in values: if value is None: formatted_values.append(None) continue items = value if isinstance(value, list) else [value] formatted_items = [ labels_by_id.get(str(item), item) for item in items ] formatted_values.append( formatted_items if isinstance(value, list) else formatted_items[0] ) return formatted_values def format_field_value(action, field, value): if value is None: return None if field in ["permission"]: values = value if isinstance(value, list) else [value] formatted_values = [_format_permission_value(item) for item in values] return formatted_values if isinstance(value, list) else formatted_values[0] if field == "journey": return _format_journey_values(value)[0] if field == "pos_module" and action in ["student_create", "student_update"]: return _format_pos_module_values(value)[0] if isinstance(value, list): model = _get_model_for_field(action, field) if model: try: names_by_id = dict( model.objects .filter(id__in=value) .values_list("id", "name") ) return [names_by_id.get(item, item) for item in value] except (FieldError, TypeError, ValueError): pass return [ format_single_field_value(action, field, item) for item in value ] return format_single_field_value(action, field, value) def _format_permission_value(value): if not isinstance(value, str): return value if value.startswith("permission.admin."): return __(value) else: return value def format_single_field_value(action, field, value): if value is None: return value target_model = get_target_model(action) # --------------------------------------------------------------------------------- # Handle special fields that have translation mappings [without db selection] # --------------------------------------------------------------------------------- if field in ["status","mode"] and action in ["user_scheduling_preference_update","user_constraint_setting_update"]: return __( f"attr.{target_model}_{field}_name.{value}" ) elif field in ["staff_requirement_type","location_requirement_type"]: return __( f"attr.{field}_name.{value}" ) elif field == "status": return __( f"attr.academic_term_status.{value}" ) elif field in ["suggested_day", "scheduled_day"]: #booking_schedule, #booking_create return __( f"attr.days_name.{value}" ) elif field in ["user_type"]: return __( f"attr.{field}_name.{value}" ) # --------------------------------------------------------- # Model-based ID fields # --------------------------------------------------------- model = _get_model_for_field(action, field) if not model: return value try: obj = ( model.objects .filter(id=value) .values("name") .first() ) return obj["name"] if obj else value except (FieldError, TypeError, ValueError): return value def _get_model_for_field(action, field): if not action or "_delete" in action: return None models = { "department": TtDepartment, "zone": TtZone, "tag": TtTag, "academic_term": TtAcademicTerm, "activity_template": TtActivityTemplate, "staff": TtStaff, "location": TtLocation, "activity": TtActivity, "activity_type": TtActivityType, "module": TtModule, "availability": TtAvailability, "start_preference": TtStartPreference, "usage_preference": TtUsagePreference, "user": User, "suitability": TtSuitability, "module_group": TtModuleGroup, "role": Role, "resource_break": TtResourceBreak, "maximum_hour": TtMaximumHour, "maximum_workspan": TtMaximumWorkspan, "free_block": TtFreeBlock, "constraint_profile": TtConstraintProfile, "travel_table": TtTravelTable, "pos": TtPos, "student_set": TtStudentSet, "time_block": TtTimeBlock, "location_suitability": TtSuitability, "staff_suitability": TtSuitability, "week_pattern": TtWeekPattern, } target_model = get_target_model(action) if field in ["avoid_concurrency"]: model = models.get(target_model) elif field in ["preceding", "same_time", "succeeding"]:#sequencing model = models.get("activity") elif field in ["created_by", "updated_by"]:#sequencing model = models.get("user") elif field in ["shared_with_department"]: model = models.get("department") elif field.endswith("_id") or field.endswith("_ids") or field.endswith("_preset"): model = models.get(field.removesuffix("_id").removesuffix("_ids").removesuffix("_preset")) else: model = models.get(field) return model def _format_slot_required(slot_required, minute_per_slot): """Match the slot option labels returned by TravelTableRules.""" if slot_required is None: return None try: total_minutes = (int(slot_required) - 1) * int(minute_per_slot) except (TypeError, ValueError): return slot_required hours, minutes = divmod(total_minutes, 60) return f"{hours:02d}:{minutes:02d} {__('attr.hours')}" def get_target_model(action): if "_create" in action or "_update" in action or "_delete" in action: action = ( action.removesuffix("_create") .removesuffix("_update") .removesuffix("_delete") ) return action def get_target_model_text(action): target_model = get_target_model(action) if target_model in ["booking_schedule", "booking_swap"]: target_model = "booking" elif target_model in ["pathway_generate"]: target_model = "pathway" elif target_model in ["activity_template_allocate"]: target_model = "activity_template" elif target_model in ["swap", "activity_generate", "preschedule_request", "schedule", "unschedule"]: target_model = "activity" elif target_model in ["pos_update_module"]: target_model = "pos_module" elif target_model in ["jta_split"]: target_model = "jta" elif target_model in ["allocation"]: target_model = "pos" elif target_model in ["resource_update_requirement"]: target_model = "resource_requirement" return __(f"attr.{target_model}")