import datetime import random import re from collections import defaultdict from itertools import product, combinations from django.conf import settings from django.db import transaction from django.db.models import Q from rest_framework import serializers, status from api.helper.resource_map_helper import helper_recalculate_and_get_redis_resource_map from api.models import TtUsagePreference, TtStartPreference, TtSetting, TtPos, TtPosModuleGroupModule, TtPathway, \ TtPathwayPosModuleGroupModule, TtPosModuleGroup, TtActivity, UserConstraintSetting, UserSchedulingPreference, \ TtAcademicTermWeek, TtSuitability, TtStaff, TtLocation, TtActivityStaff, TtActivityLocation from api.models.audit_trail import AuditTrail from api.models.audit_trail_details import AuditTrailDetails from api.views.admin.base import AdminApiBase from api.validator import BaseValidator from api.utils import decrypt_aes_128_cbc, encrypt_aes_128_cbc, log_critical_error, get_exception_detail, \ get_add_new_data, get_ip, bulk_sync_to_redis, get_resource_map_redis_data, \ remove_variant_activity_name_range, convert_to_redis_week, format_variant_activity_name_week_pattern_to_ranges, \ push_websocket_notification from rest_framework.exceptions import ValidationError from api.translation import __ from backend.kafka import send_request from backend.redis_client import redis_client from api.services.integration.tracking import track_activity_ids class ScheduleRequest(AdminApiBase): def validate_request(self,request): rules = { "activity_ids": "required|array|exists:api.TtActivity,id", "slot": "nullable", "socket_id": "nullable", "constraint_breaking_mode": "nullable", } # if need rename field, can put here attribute = { # "email": __("attr.email"), } validator = BaseValidator(request.data,rules,attribute) error = validator.validate() if error: raise ValidationError(error) # custom error after basic validation def post(self, request): # Validate input try: self.validate_request(request) setting_params = { "slot_per_week", "slot_per_day", "minute_per_slot" } settings = TtSetting.get_multiple_setting(setting_params) slot_per_week = int(settings["slot_per_week"]) slot_per_day = int(settings["slot_per_day"]) minute_per_slot = int(settings["minute_per_slot"]) activity_ids = request.data["activity_ids"] action = "schedule_request" method = "schedule" activities = TtActivity.objects.filter(id__in=activity_ids).select_related("week_pattern").prefetch_related("week","week_pattern__week") user = request.user skip_audit_trail = False request_slot = request.data.get("slot") or None kafka_topic = self.kafka_config['REQUEST_TOPIC'] # kafka_topic = "test-topic" constraint_breaking_mode = 1 if request.data.get("constraint_breaking_mode") == 1 else 0 # constraint setting user_constraint_setting_list = UserConstraintSetting.get_setting(user.id) # scheduling preference user_scheduling_preference_list = UserSchedulingPreference.get_setting(user.id) kafka_request_data = { "session_id": request.user.name, "activities": [], "slot": None, "constraint_breaking_mode": constraint_breaking_mode, "constraint_setting": UserConstraintSetting.kafka_group_settings_data(user_constraint_setting_list), "scheduling_preference": UserSchedulingPreference.kafka_group_settings_data(user_scheduling_preference_list), } for activity_id in activity_ids: kafka_request_data["activities"].append(activity_id) if request_slot and len(activity_ids) == 1: first_activity = activities.first() if first_activity.week_pattern_id: first_week = first_activity.week_pattern.week.order_by("week").first() else: first_week = first_activity.week.order_by("week").first() kafka_request_data["slot"] = ((first_week.week - 1) * int(slot_per_week)+ int(request_slot)) # # test_request={ # "session_id": "dev", # "status": "success", # "schedule": "[{\"activity\":\"190\",\"teaching_staff\":\"[47,48]\",\"start_slot\":\"400\",\"location\":\"[23,24]\"},{\"activity\":\"191\",\"teaching_staff\":\"[47,48]\",\"start_slot\":\"410\",\"location\":\"[23,24]\"}]" # } # manual handle constraint break, only 1 activity with slot only will used this manual function # if len(activity_ids) == 1 and request_slot and activities and constraint_breaking_mode: # manual_constraint_break(activities.first(),request_slot,request.data.get("socket_id"),minute_per_slot,slot_per_day,slot_per_week) # else: # send_request(kafka_topic, kafka_request_data, request.data.get("socket_id"),method) # skip_audit_trail = True send_request(kafka_topic, kafka_request_data, request.data.get("socket_id"), method) skip_audit_trail = True if not skip_audit_trail: # insert audit trail and details # can insert the parent first, then only insert details name = ",".join(activities.values_list("name", flat=True)) audit_trail = AuditTrail.objects.create( user_id=request.user.id, type=self.audit_type, ip_address=get_ip(request) ) remark_param = { "name":name, } # call to a function for insert details, so in future if want change to used worker, can modify in function only AuditTrailDetails.custom_insert( audit_trail=audit_trail, action=action, remark_param=remark_param, # new_data="", ) response = { } return self.api_response(data=response) except ValidationError as e: first_message = e.detail['error'] errors = e.detail['errors'] return self.api_response(error=first_message,errors=errors,code=status.HTTP_400_BAD_REQUEST) 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']) return self.api_response(error=__("message.internal_server_error"),code=status.HTTP_500_INTERNAL_SERVER_ERROR) def get_match_key(activity): """ helper function to get variant staff,location,and slot for checking """ staff_ids = tuple(sorted([s.id for s in activity.staff.all()])) location_ids = tuple(sorted([l.id for l in activity.location.all()])) start_slot = activity.scheduled_start_slot slot_required = activity.slot_required return (location_ids, staff_ids, start_slot,slot_required) def manual_constraint_break(activity,slot,socket_id,minute_per_slot,slot_per_day,slot_per_week): """ temporary function be4 engine complete constraint break only allow 1 activity """ track_activity_ids([activity.id]) websocket_host = settings.WEBSOCKET_CONFIG["HOST"] websocket_port = str(settings.WEBSOCKET_CONFIG["PORT"]) websocket_schedule_endpoint = str(settings.WEBSOCKET_CONFIG["SCHEDULE_RESPONSE_ENDPOINT"]) websocket_schedule_url = websocket_host + ":" + websocket_port + "/" + websocket_schedule_endpoint microservices_tt_topic = settings.KAFKA_CONFIG["MICROSERVICES_TT_TOPIC"] try: update_activity_objs = [] insert_activity_staff_objs = [] insert_activity_location_objs = [] need_delete_variant_activity_ids = [] affected_variant_activity_ids = [] staff_with_affected_week = {} location_with_affected_week = {} student_set_with_affected_week = {} # key will be activity id and value will be week_ids activity_week_microservices_relation = {} # param used to update redis resource map scheduled_staff_ids = set() scheduled_location_ids = set() scheduled_student_set_ids = set() scheduled_variant_activity_ids = [] activity_ids = [] success_activity_ids = [] # got start_slot only add in all_staff_ids = set() all_location_ids = set() redis_data = { "update": {}, "delete": {} } redis_staff_table = "staff" redis_location_table = "location" redis_student_set_table = "student_set" redis_activity_table = "activity" kafka_data = [] # got kafka_log only do action, else need double check why no request but receive response if activity: activity_ids.append(activity.id) success_count = 0 fail_count = 0 # at the end will get those affected variant ids, to check code and name to change activity_id = activity.id # if got week_pattern_id, will get from week_pattern table, else will used self week_pattern if activity.week_pattern_id: week_ids = list(aw.id for aw in activity.week_pattern.week.all()) else: week_ids = list(aw.id for aw in activity.week.all()) if slot: success_activity_ids.append(activity_id) # group variant activity later will used to check variant merge if activity.is_variant == 1: scheduled_variant_activity_ids.append(activity_id) # if already scheduled, no need change other resources if activity.scheduled == 1: staff_ids = [s.id for s in activity.staff.all()] location_ids = [l.id for l in activity.location.all()] else: if activity.staff_requirement_type == TtActivity.STAFF_REQUIREMENT_TYPE["preset"]: staff_ids = [s.id for s in activity.staff_preset.all()] elif int(activity.staff_requirement) > 0: suitability = activity.staff_suitability.first() if suitability: available_staff_ids = list(suitability.staff.values_list("id", flat=True)) else: available_staff_ids = list(TtStaff.objects.values_list("id", flat=True)) staff_requirement = min(int(activity.staff_requirement), len(available_staff_ids)) # random amount based on value requirement or available_ids staff_ids = random.sample(available_staff_ids, staff_requirement) else: staff_ids = [] if activity.location_requirement_type == TtActivity.LOCATION_REQUIREMENT_TYPE["preset"]: location_ids = [l.id for l in activity.location_preset.all()] elif int(activity.location_requirement) > 0: suitability = activity.location_suitability.first() if suitability: available_location_ids = list(suitability.location.values_list("id", flat=True)) else: available_location_ids = list(TtLocation.objects.values_list("id", flat=True)) location_requirement = min(int(activity.location_requirement), len(available_location_ids)) # random amount based on value requirement or available_ids location_ids = random.sample(available_location_ids, location_requirement) else: location_ids = [] student_set_ids = list(ss.id for ss in activity.student_set.all()) start_slot = slot # calculate how many slot taken for this activity slot_taken = activity.duration // minute_per_slot # 0 is Monday 6 is Sunday day = start_slot // slot_per_day # * 60 means need convert to second time_in_second = (start_slot % slot_per_day) * minute_per_slot * 60 hours = time_in_second // 3600 minutes = (time_in_second % 3600) // 60 time = datetime.time(hour=hours, minute=minutes) # data need to update put in a object first, at the end will used bulk update to perform, so can reduce the query activity_obj = TtActivity(id=activity_id) activity_obj.scheduled_start_time = time activity_obj.scheduled_day = day activity_obj.scheduled_start_slot = start_slot activity_obj.scheduled = 1 update_activity_objs.append(activity_obj) redis_data["update"].setdefault(redis_activity_table, []).append({ "id": activity_id, "scheduled": 1, "scheduled_start_slot": start_slot, }) if staff_ids: for staff_id in staff_ids: activity_staff_obj = TtActivityStaff( activity_id=activity_id, staff_id=staff_id ) insert_activity_staff_objs.append(activity_staff_obj) staff_with_affected_week.setdefault(staff_id, set()).update(week_ids) scheduled_staff_ids.update(staff_ids) if location_ids: for location_id in location_ids: activity_location_obj = TtActivityLocation( activity_id=activity_id, location_id=location_id ) insert_activity_location_objs.append(activity_location_obj) location_with_affected_week.setdefault(location_id, set()).update(week_ids) scheduled_location_ids.update(location_ids) # if this activity have student_set, will recalculate the resource map for the student set if student_set_ids: for student_set_id in student_set_ids: student_set_with_affected_week.setdefault(student_set_id, set()).update(week_ids) scheduled_student_set_ids.update(student_set_ids) # success count + 1 success_count += 1 else: # failed count + 1 and continue next fail_count += 1 # if not empty, do bulk update and bulk create, if got duplicate key, will ignore and continue other if insert_activity_staff_objs: TtActivityStaff.objects.bulk_create(insert_activity_staff_objs, ignore_conflicts=True) if insert_activity_location_objs: TtActivityLocation.objects.bulk_create(insert_activity_location_objs, ignore_conflicts=True) # for bulk_update, must have id only can work if update_activity_objs: TtActivity.objects.bulk_update(update_activity_objs, ["scheduled_start_time", "scheduled_day", "scheduled", "scheduled_start_slot"]) # check have variant activity or not, if got variant activity, need check need to merge or not if scheduled_variant_activity_ids: parent_ids = list( TtActivity.objects.filter(id__in=scheduled_variant_activity_ids, is_variant=1) .values_list('variant_parent_id', flat=True) .distinct() ) parent_ids = [pid for pid in parent_ids if pid] variant_activities = ( TtActivity.objects.filter( Q(id__in=scheduled_variant_activity_ids) | Q(variant_parent_id__in=scheduled_variant_activity_ids) | Q(variant_parent_id__in=parent_ids) | Q(id__in=parent_ids), is_variant=1, scheduled=1 ) .select_related("week_pattern") .prefetch_related("week", "week_pattern__week", "staff", "location", "student_set") ) family_tree = defaultdict(list) parents_by_id = {} variant_activities_by_id = {v_act.id: v_act for v_act in variant_activities} for act in variant_activities: if act.variant_parent_id is None: parents_by_id[act.id] = act else: family_tree[act.variant_parent_id].append(act) m2m_weeks_to_add = defaultdict(list) # Key: Target Activity ID, Value: List of Week IDs activity_ids_to_delete = set() for scheduled_variant_activity_id in scheduled_variant_activity_ids: variant_activity = variant_activities_by_id.get(scheduled_variant_activity_id) if not variant_activity or variant_activity.id in activity_ids_to_delete: continue # Skip if already merged/deleted in a previous loop pass # Determine the parent ID and the actual parent object p_id = variant_activity.id if variant_activity.variant_parent_id is None else variant_activity.variant_parent_id parent_obj = parents_by_id.get(p_id) children = family_tree.get(p_id, []) # this scenario is handle when variant activity is parent if variant_activity.variant_parent_id is None: parent_key = get_match_key(variant_activity) for child in children: if child.id not in activity_ids_to_delete and get_match_key(child) == parent_key: # Merge child into parent child_weeks = [w.id for w in child.week.all()] m2m_weeks_to_add[variant_activity.id].extend(child_weeks) activity_ids_to_delete.add(child.id) else: # when variant activity is child # when parent is scheduled only need get parent_obj if parent_obj: parent_key = get_match_key(parent_obj) child_key = get_match_key(variant_activity) if parent_obj and child_key == parent_key: # Merge target child into parent target_weeks = [w.id for w in variant_activity.week.all()] m2m_weeks_to_add[parent_obj.id].extend(target_weeks) activity_ids_to_delete.add(variant_activity.id) if parent_obj.id not in success_activity_ids: success_activity_ids.append(parent_obj.id) else: # Compare against siblings to see if any match the target child for sibling in children: if sibling.id == variant_activity.id or sibling.id in activity_ids_to_delete: continue if get_match_key(sibling) == child_key: # Merge target child into this matching sibling target_weeks = [w.id for w in variant_activity.week.all()] m2m_weeks_to_add[sibling.id].extend(target_weeks) activity_ids_to_delete.add(variant_activity.id) if sibling.id not in success_activity_ids: success_activity_ids.append(sibling.id) break # Found a match, stop looking at other siblings merge_variant_activity_ids = [] # update activity week to merge variant for target_activity_id, week_ids in m2m_weeks_to_add.items(): if target_activity_id not in activity_ids_to_delete and week_ids: target_act = variant_activities_by_id.get(target_activity_id) if target_act: merge_variant_activity_ids.append(target_act.id) target_act.week.add(*week_ids) # delete variant child if activity_ids_to_delete: TtActivity.objects.filter(id__in=activity_ids_to_delete).delete() need_delete_variant_activity_ids = list(activity_ids_to_delete) # get again merge variant activity to check need update data or not merge_variant_activities = ( TtActivity.objects.filter(id__in=merge_variant_activity_ids) .select_related("week_pattern") .prefetch_related("week", "week_pattern__week", ) ) for merge_variant_activity in merge_variant_activities: # this query mostly just want to confirm this variant still have how many variant, if only 1 record, will remove variant tag # for where statement, id and variant_parent_id need check for both self and parent id related_variant_objs = ( TtActivity.objects.filter( Q(id__in=[merge_variant_activity.id, merge_variant_activity.variant_parent_id]) | Q(variant_parent_id__in=[merge_variant_activity.id, merge_variant_activity.variant_parent_id]) ) ) if merge_variant_activity.week_pattern: current_week_pattern = list( {"id": aw.id, "week": aw.week} for aw in merge_variant_activity.week_pattern.week.all()) else: current_week_pattern = list( {"id": aw.id, "week": aw.week} for aw in merge_variant_activity.week.all()) academic_term_week_pattern = list( TtAcademicTermWeek.objects .filter(academic_term_id=merge_variant_activity.academic_term_id) .select_related("week") .order_by("week__week") .values("week_id", "week__week") ) academic_term_week_pattern = [ { "id": w["week_id"], "week": w["week__week"], } for w in academic_term_week_pattern ] # if only have 1 record, means already not a variant merge_variant_activity.name = remove_variant_activity_name_range(merge_variant_activity.name) if related_variant_objs.count() == 1: # if not a variant, remove code " 00x" and update is_variant to 0 and variant_parent_id = None merge_variant_activity.code = re.sub(r"\s\d{3}$", "", merge_variant_activity.code) merge_variant_activity.is_variant = 0 merge_variant_activity.variant_parent_id = None else: merge_variant_activity.name = merge_variant_activity.name + " " + format_variant_activity_name_week_pattern_to_ranges( current_week_pattern, academic_term_week_pattern) main_week = sorted({w["week"] for w in current_week_pattern}) merge_variant_activity.save() redis_data["update"].setdefault(redis_activity_table, []).append({ "id": merge_variant_activity.id, "week_pattern": convert_to_redis_week(main_week), "code": merge_variant_activity.code, "name": merge_variant_activity.name, "is_variant": merge_variant_activity.is_variant, "variant_parent_id": merge_variant_activity.variant_parent_id, }) # add set() for activity with main week to handle microservices week, if same parent will keep merge for 2 time, will replace latest activity_week_microservices_relation[merge_variant_activity.id] = sorted( {w["id"] for w in current_week_pattern}) # need after merging of variant only start to do recalculate resource map part # update resource map staff_with_affected_week = {k: v for k, v in staff_with_affected_week.items() if v} location_with_affected_week = {k: v for k, v in location_with_affected_week.items() if v} student_set_with_affected_week = {k: v for k, v in student_set_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, staff_with_affected_week=staff_with_affected_week, location_with_affected_week=location_with_affected_week, student_set_with_affected_week=student_set_with_affected_week, slot_per_week=slot_per_week, ) # get again latest activities data from db # only get success schedule activity from db activities = ( TtActivity.objects.filter(id__in=success_activity_ids) .select_related("activity_template", "department", "academic_term", "module", "week_pattern", "activity_type", "zone", "availability", "start_preference", "usage_preference") .prefetch_related("staff", "location", "week", "week_pattern__week", ) ) # format the data need put to FE # dd request if no failed, dun show "failed: 0" to FE, cause will confuse user if fail_count > 0: message = __("message.scheduled_success_failed_count", success_count=success_count, fail_count=fail_count) else: message = __("message.scheduled_success_only", success_count=success_count) websocket_data = { "message": message, "data": [], } for activity in activities: microservices_week_ids = activity_week_microservices_relation.get(activity.id) or None final_data = { "id": activity.id, "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, "module": activity.module.name if activity.module else None, "module_id": activity.module_id, "location": list(activity.location.values_list("name", flat=True)) if activity.location else None, "staff": list(activity.staff.values_list("name", flat=True)) if activity.staff else None, "is_variant": activity.is_variant, "variant_parent_id": activity.variant_parent_id, } # need used .copy() if not changes in final_data will change in this websocket_data also websocket_data['data'].append(final_data.copy()) final_data['staff_ids'] = [s.id for s in activity.staff.all()] final_data['location_ids'] = [loc.id for loc in activity.location.all()] final_data['scheduled_day'] = activity.scheduled_day if activity.scheduled_day is not None else None if microservices_week_ids: final_data['week_ids'] = microservices_week_ids kafka_data.append(final_data) # after get activities, will need get latest week_pattern for update to redis if redis_data['update'] or redis_data['delete']: bulk_sync_to_redis(redis_client, redis_data) # push websocket for FE if socket_id: push_websocket_notification(websocket_schedule_url, websocket_data, socket_id) # push kfaka microservice, add main week method = "schedule" kafka_topic = microservices_tt_topic if kafka_data and kafka_topic: kafka_request_data = { "session_id": "", "activity": kafka_data, "delete_activity": need_delete_variant_activity_ids, } send_request(kafka_topic, kafka_request_data, None, method) else: # 2026-01-14 when not return success, just return a general msg first, cause they no return reason also err = { "code": status.HTTP_400_BAD_REQUEST, "error": __("validation.unavailable_to_schedule"), "errors": None, } if socket_id: push_websocket_notification(websocket_schedule_url, err, socket_id) 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']) err = { "code": status.HTTP_400_BAD_REQUEST, "error": __("validation.unavailable_to_schedule"), "errors": None, } if socket_id: push_websocket_notification(websocket_schedule_url, err, socket_id) raise