import json
import os
import sys
import time
from datetime import datetime as dt, timedelta
from datetime import time as dt_time

import django
from confluent_kafka import Consumer
from django.conf import settings
from django.db import InterfaceError, OperationalError, transaction

# if want access django, must put this 2
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings")
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
django.setup()

from api.models import (
	EsDepartment,
	EsExamActivity,
	EsInvigilator,
	EsLocation,
	EsStaff,
	EsStudent,
	EsStudentGroup,
	EsZone,
	KafkaLog,
	TtAcademicTerm,
	TtAcademicTermWeek,
	TtActivity,
	TtActivityLocation,
	TtActivityStaff,
	TtActivityTemplate,
	TtActivityWeek,
	TtStudentSetActivity,
	TtWeek,
	TtWeekPattern,
	TtWeekPatternWeek,
)
from api.services.student_groups import (
	replace_student_group_rows,
	resolve_student_group_ids_from_row,
)
from api.services.tt_availability import (
	exam_slot_busy_window,
	tt_activity_busy_window as tt_busy_window_from_activity,
	windows_overlap,
)
from api.utils import get_exception_detail, log_critical_error
from api.services.exam_scheduler import (
	SLOT_MINUTES,
	_duration_slots,
	unschedule_exam_requirements,
)

host = settings.KAFKA_CONFIG["HOST"]
port = str(settings.KAFKA_CONFIG["PORT"])
worker_1 = settings.KAFKA_CONFIG["WORKER_1"]
topic = settings.KAFKA_CONFIG["MICROSERVICES_TT_TOPIC"]

MINUTE_PER_SLOT = SLOT_MINUTES
SLOT_PER_DAY = 48  # 24h at 30-min slots; matches TT default slot_per_day
consumer = None


def get_consumer():
	global consumer
	if consumer is None:
		conf = {
			"bootstrap.servers": host + ":" + port,
			"group.id": worker_1,
			"auto.offset.reset": "latest",
            #"auto.offset.reset": "earliest",
			"enable.auto.commit": False,
		}
		consumer = Consumer(conf)
		consumer.subscribe([topic])
	return consumer



def append_log(response_data, header_data, data):
	request_id = response_data.get("request_id")

	kafka_log = KafkaLog.insert_response(
		request_id,
		topic,
		response_data,
		None,
		header_data,
	)

	try:
		extra_data = json.loads(kafka_log.extra_data) if kafka_log.extra_data else []
		if not isinstance(extra_data, list):
			extra_data = []
	except Exception:
		extra_data = []

	extra_data.append(data)

	KafkaLog.objects.filter(id=kafka_log.id).update(
		extra_data=json.dumps(extra_data, default=str)
	)

	return kafka_log


def normalize_payload(response_data, key):
	payload = response_data.get(key)

	if not payload:
		return []
	if isinstance(payload, dict):
		return [payload]
	if isinstance(payload, list):
		return payload
	raise Exception(f"{key} must be dict or list")


def build_update_data(row, field_map, single_item=False):
	data = {}
	for payload_key, model_field in field_map.items():
		if payload_key not in row:
			continue
		if payload_key == "code" and not single_item:
			continue
		data[model_field] = row.get(payload_key)
	return data


def bulk_create_skip_duplicates(Model, data_rows, batch_size=500):
	"""
	Create rows only when their PK (id) is not already present.
	Used by the lightweight TT sync phase handlers.
	"""
	if not data_rows:
		return []

	incoming_ids = [row.get("id") for row in data_rows if row.get("id") is not None]
	existing_ids = (
		set(Model.objects.filter(id__in=incoming_ids).values_list("id", flat=True))
		if incoming_ids
		else set()
	)

	to_create = [Model(**row) for row in data_rows if row.get("id") not in existing_ids]
	if to_create:
		Model.objects.bulk_create(to_create, batch_size=batch_size)
	return to_create


def replace_relation_rows(RelModel, parent_field, child_field, relation_map):
	"""
	Delete existing through-rows for parents in relation_map, then recreate from map values.
	relation_map: {parent_id: [child_id, ...]}
	"""
	if not relation_map:
		return 0

	parent_ids = list(relation_map.keys())
	RelModel.objects.filter(**{f"{parent_field}_id__in": parent_ids}).delete()

	new_rows = []
	for parent_id, child_ids in relation_map.items():
		for child_id in child_ids or []:
			if child_id:
				new_rows.append(
					RelModel(
						**{
							f"{parent_field}_id": parent_id,
							f"{child_field}_id": child_id,
						}
					)
				)

	if new_rows:
		RelModel.objects.bulk_create(new_rows, batch_size=500, ignore_conflicts=True)

	return len(new_rows)


def find_by_tt_id_then_code(Model, tt_id, code):
	"""Match existing row: tt_id first, then code (prefer tt_id IS NULL)."""
	if tt_id is not None:
		obj = Model.objects.filter(tt_id=tt_id).first()
		if obj:
			return obj

	if code:
		obj = Model.objects.filter(code=code, tt_id__isnull=True).order_by("id").first()
		if obj:
			return obj

	return None


def resolve_department_id(tt_department_id):
	if not tt_department_id:
		return None
	department = EsDepartment.objects.filter(tt_id=tt_department_id).first()
	return department.id if department else None


def resolve_zone_id(tt_zone_id):
	if not tt_zone_id:
		return None
	zone = EsZone.objects.filter(tt_id=tt_zone_id).first()
	return zone.id if zone else None


def _filter_existing_week_ids(week_ids):
	if not week_ids:
		return []

	normalized = []
	for week_id in week_ids:
		if isinstance(week_id, dict):
			week_id = week_id.get("id")
		if week_id is None:
			continue
		try:
			normalized.append(int(week_id))
		except (TypeError, ValueError):
			continue

	if not normalized:
		return []

	existing = set(TtWeek.objects.filter(id__in=normalized).values_list("id", flat=True))
	return [week_id for week_id in normalized if week_id in existing]


def sync_staff_invigilator(staff):
	inactive = EsInvigilator.STATUS_TO_CODE["inactive"]
	inv = EsInvigilator.objects.filter(staff=staff).first()

	if not staff.is_invigilator:
		if inv and inv.status != inactive:
			inv.status = inactive
			inv.save(update_fields=["status", "updated_at"])
		return

	fields = {
		"staff_id": staff.id,
		"code": staff.code,
		"name": staff.name,
		"email": staff.email,
		"phone": staff.phone,
		"desc": staff.desc,
		"department_id": staff.department_id,
		"status": staff.status,
	}

	if inv:
		for key, value in fields.items():
			setattr(inv, key, value)
		inv.save()
		return

	inv = EsInvigilator.objects.filter(code=staff.code, staff__isnull=True).order_by("id").first()
	if inv:
		for key, value in fields.items():
			setattr(inv, key, value)
		inv.save()
		return

	EsInvigilator.objects.create(**fields)


def parse_time_value(value):
	if value is None:
		return None
	if isinstance(value, dt_time):
		return value
	if isinstance(value, dt):
		return value.time()
	if isinstance(value, str):
		for fmt in ("%H:%M:%S", "%H:%M"):
			try:
				return dt.strptime(value, fmt).time()
			except ValueError:
				continue
	return None


def tt_busy_window(row):
	"""Build weekday + start/end minutes-from-midnight from TT schedule payload (no week yet)."""
	scheduled_day = row.get("scheduled_day")
	if scheduled_day is None:
		return None

	start_time = parse_time_value(row.get("scheduled_start_time"))
	if start_time is None and row.get("scheduled_start_slot") is not None:
		slot = int(row.get("scheduled_start_slot"))
		within_day_slot = slot % SLOT_PER_DAY
		start_minutes = within_day_slot * MINUTE_PER_SLOT
		start_time = dt_time(hour=(start_minutes // 60) % 24, minute=start_minutes % 60)
	if start_time is None:
		return None

	duration_minutes = row.get("duration")
	if duration_minutes is None and row.get("slot_required") is not None:
		duration_minutes = int(row.get("slot_required")) * MINUTE_PER_SLOT
	if not duration_minutes:
		return None

	start_minutes = start_time.hour * 60 + start_time.minute
	end_minutes = start_minutes + int(duration_minutes)
	return {
		"scheduled_day": int(scheduled_day),
		"start_minutes": start_minutes,
		"end_minutes": end_minutes,
	}


def exam_busy_window(activity):
	if activity.time_slot is None:
		return None
	exam_window = exam_slot_busy_window(
		activity.time_slot,
		_duration_slots(activity.exam_requirement),
		activity.exam_requirement.exam_period if activity.exam_requirement_id else None,
	)
	return exam_window


# windows_overlap imported from api.services.tt_availability


def _coerce_tt_ids(items):
	ids = []
	for item in items or []:
		if isinstance(item, dict):
			item = item.get("id")
		if item is None:
			continue
		try:
			ids.append(int(item))
		except (TypeError, ValueError):
			continue
	return ids


_coerce_tt_location_ids = _coerce_tt_ids


def collect_location_ids_from_activity_row(row):
	if "location_ids" in row:
		location_ids = _coerce_tt_location_ids(row.get("location_ids"))
	else:
		location_ids = _coerce_tt_location_ids(row.get("location"))
	return list(dict.fromkeys(location_ids))


def _week_ids_from_pattern_row(row):
	pattern_ids = set()
	for key in ("week_pattern", "week_pattern_id"):
		if key not in row:
			continue
		raw = row.get(key)
		if not raw:
			continue
		if isinstance(raw, list):
			values = raw
		else:
			values = [raw]
		for v in values:
			if v is None:
				continue
			try:
				pattern_ids.add(int(v))
			except Exception:
				continue

	if not pattern_ids:
		return None

	week_ids = set(
		TtWeekPatternWeek.objects.filter(week_pattern_id__in=list(pattern_ids)).values_list("week_id", flat=True)
	)
	return week_ids or None


def _week_ids_from_explicit_row(row):
	raw_week_ids = extract_week_ids_from_row(row)
	if raw_week_ids is None:
		return None

	week_ids = set()
	for v in raw_week_ids:
		if v is None:
			continue
		try:
			week_ids.add(int(v))
		except Exception:
			continue
	return week_ids or None


def resolve_week_ids_from_row(row):
	"""
	Resolve TT week ids for this scheduled activity row (TT convention: week_pattern first).

	- If `week_pattern` / `week_pattern_id` is present and non-null, resolve via `tt_week_pattern_week`.
	- Otherwise use explicit `week` / `week_ids`.
	- Return None when neither yields weeks (caller may fall back to stored activity).
	"""
	pattern_weeks = _week_ids_from_pattern_row(row)
	if pattern_weeks:
		return pattern_weeks
	return _week_ids_from_explicit_row(row)


def effective_tt_activity_week_ids(activity_id):
	if not activity_id:
		return None

	activity = TtActivity.objects.filter(id=activity_id).only("week_pattern_id").first()
	if not activity:
		return None

	if activity.week_pattern_id:
		week_ids = set(
			TtWeekPatternWeek.objects.filter(week_pattern_id=activity.week_pattern_id).values_list("week_id", flat=True)
		)
	else:
		week_ids = set(
			TtActivityWeek.objects.filter(activity_id=activity_id).values_list("week_id", flat=True)
		)
	return week_ids or None


def resolve_week_ids_for_schedule_row(row):
	week_ids = resolve_week_ids_from_row(row)
	if week_ids is not None:
		return week_ids
	return effective_tt_activity_week_ids(activity_row_id(row))


def unschedule_conflicting_exam_activities(tt_location_ids, busy_window, payload_week_ids):
	if not tt_location_ids or busy_window is None or payload_week_ids is None:
		return {
			"checked_locations": 0,
			"unscheduled_activity_ids": [],
		}

	locations = list(EsLocation.objects.filter(tt_id__in=tt_location_ids))
	if not locations:
		return {
			"checked_locations": 0,
			"unscheduled_activity_ids": [],
		}

	location_ids = [location.id for location in locations]

	# Precompute TT week date ranges once per payload row.
	week_ranges = []
	weeks = list(
		TtWeek.objects.filter(id__in=list(payload_week_ids)).only("start_date")
	)
	for w in weeks:
		week_ranges.append((w.start_date, w.start_date + timedelta(days=6)))
	# If we cannot resolve any week ranges, do not unschedule anything.
	if not week_ranges:
		return {
			"checked_locations": 0,
			"unscheduled_activity_ids": [],
		}

	activities = list(
		EsExamActivity.objects
		.filter(is_scheduled=True, location_id__in=location_ids, time_slot__isnull=False)
		.select_related("exam_requirement")
	)

	conflict_ids = []
	for activity in activities:
		activity_window = exam_busy_window(activity)
		if not activity_window:
			continue
		if not windows_overlap(busy_window, activity_window):
			continue

		activity_date = activity_window.get("activity_date")
		if not activity_date:
			continue

		if any(start_date <= activity_date <= end_date for start_date, end_date in week_ranges):
			conflict_ids.append(activity.id)

	if conflict_ids:
		unschedule_exam_requirements(conflict_ids, None)

	return {
		"checked_locations": len(locations),
		"unscheduled_activity_ids": conflict_ids,
	}


# ============================
# DEPARTMENT
# ============================

def department_create(response_data):
	payload = normalize_payload(response_data, "department")
	created = 0
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			code = row.get("code")
			obj = find_by_tt_id_then_code(EsDepartment, tt_id, code)
			parent_id = resolve_department_id(row.get("department"))

			fields = {
				"tt_id": tt_id,
				"code": code,
				"name": row.get("name"),
				"desc": row.get("desc"),
				"status": row.get("status", EsDepartment.STATUS_TO_CODE["active"]),
				"department_id": parent_id,
			}

			if obj:
				for key, value in fields.items():
					if key == "code" and value is None:
						continue
					setattr(obj, key, value)
				obj.save()
				updated += 1
			else:
				EsDepartment.objects.create(**{
					k: v for k, v in fields.items() if v is not None or k in ("desc", "department_id")
				})
				created += 1

	return {
		"step": "department_create",
		"created": created,
		"updated": updated,
		"skipped": skipped,
	}


def department_update(response_data):
	payload = normalize_payload(response_data, "department")
	if not payload:
		return {"step": "department_update", "updated": 0, "skipped": 0}

	single_item = len(payload) == 1
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			obj = EsDepartment.objects.filter(tt_id=tt_id).first()
			if not obj:
				skipped += 1
				continue

			update_data = build_update_data(
				row,
				{
					"code": "code",
					"name": "name",
					"desc": "desc",
					"status": "status",
				},
				single_item=single_item,
			)
			if "department" in row:
				update_data["department_id"] = resolve_department_id(row.get("department"))

			if update_data:
				for key, value in update_data.items():
					setattr(obj, key, value)
				obj.save(update_fields=list(update_data.keys()) + ["updated_at"])
			updated += 1

	return {
		"step": "department_update",
		"updated": updated,
		"skipped": skipped,
	}


def department_delete(response_data):
	ids = response_data.get("department_ids") or []
	if not ids:
		return {"step": "department_delete", "updated": 0, "skipped": 0}

	inactive = EsDepartment.STATUS_TO_CODE["inactive"]
	existing = list(EsDepartment.objects.filter(tt_id__in=ids))
	existing_tt_ids = {obj.tt_id for obj in existing}

	with transaction.atomic():
		EsDepartment.objects.filter(tt_id__in=existing_tt_ids).update(status=inactive)

	return {
		"step": "department_delete",
		"updated": len(existing_tt_ids),
		"skipped": len(set(ids) - existing_tt_ids),
	}


# ============================
# ZONE
# ============================

def zone_create(response_data):
	payload = normalize_payload(response_data, "zone")
	created = 0
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			code = row.get("code")
			obj = find_by_tt_id_then_code(EsZone, tt_id, code)
			parent_id = resolve_zone_id(row.get("zone"))

			fields = {
				"tt_id": tt_id,
				"code": code,
				"name": row.get("name"),
				"status": row.get("status", EsZone.STATUS_TO_CODE["active"]),
				"zone_id": parent_id,
			}

			if obj:
				for key, value in fields.items():
					if key == "code" and value is None:
						continue
					setattr(obj, key, value)
				obj.save()
				updated += 1
			else:
				EsZone.objects.create(**{
					k: v for k, v in fields.items() if v is not None or k == "zone_id"
				})
				created += 1

	return {
		"step": "zone_create",
		"created": created,
		"updated": updated,
		"skipped": skipped,
	}


def zone_update(response_data):
	payload = normalize_payload(response_data, "zone")
	if not payload:
		return {"step": "zone_update", "updated": 0, "skipped": 0}

	single_item = len(payload) == 1
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			obj = EsZone.objects.filter(tt_id=tt_id).first()
			if not obj:
				skipped += 1
				continue

			update_data = build_update_data(
				row,
				{
					"code": "code",
					"name": "name",
					"status": "status",
				},
				single_item=single_item,
			)
			if "zone" in row:
				update_data["zone_id"] = resolve_zone_id(row.get("zone"))

			if update_data:
				for key, value in update_data.items():
					setattr(obj, key, value)
				obj.save(update_fields=list(update_data.keys()) + ["updated_at"])
			updated += 1

	return {
		"step": "zone_update",
		"updated": updated,
		"skipped": skipped,
	}


def zone_delete(response_data):
	ids = response_data.get("zone_ids") or []
	if not ids:
		return {"step": "zone_delete", "updated": 0, "skipped": 0}

	inactive = EsZone.STATUS_TO_CODE["inactive"]
	existing = list(EsZone.objects.filter(tt_id__in=ids))
	existing_tt_ids = {obj.tt_id for obj in existing}

	with transaction.atomic():
		EsZone.objects.filter(tt_id__in=existing_tt_ids).update(status=inactive)

	return {
		"step": "zone_delete",
		"updated": len(existing_tt_ids),
		"skipped": len(set(ids) - existing_tt_ids),
	}


# ============================
# STUDENT SET
# ============================

def student_set_create(response_data):
	payload = normalize_payload(response_data, "student_set")
	created = 0
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			code = row.get("code")
			obj = find_by_tt_id_then_code(EsStudentGroup, tt_id, code)
			department_id = resolve_department_id(row.get("department"))

			fields = {
				"tt_id": tt_id,
				"code": code,
				"name": row.get("name"),
				"desc": row.get("desc"),
				"status": row.get("status", EsStudentGroup.STATUS_TO_CODE["active"]),
				"department_id": department_id,
			}

			if obj:
				for key, value in fields.items():
					if key == "code" and value is None:
						continue
					setattr(obj, key, value)
				obj.save()
				updated += 1
			else:
				EsStudentGroup.objects.create(**{
					k: v for k, v in fields.items() if v is not None or k in ("desc", "department_id")
				})
				created += 1

	return {
		"step": "student_set_create",
		"created": created,
		"updated": updated,
		"skipped": skipped,
	}


def student_set_update(response_data):
	payload = normalize_payload(response_data, "student_set")
	if not payload:
		return {"step": "student_set_update", "updated": 0, "skipped": 0}

	single_item = len(payload) == 1
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			obj = EsStudentGroup.objects.filter(tt_id=tt_id).first()
			if not obj:
				skipped += 1
				continue

			update_data = build_update_data(
				row,
				{
					"code": "code",
					"name": "name",
					"desc": "desc",
					"status": "status",
				},
				single_item=single_item,
			)
			if "department" in row:
				update_data["department_id"] = resolve_department_id(row.get("department"))

			if update_data:
				for key, value in update_data.items():
					setattr(obj, key, value)
				obj.save(update_fields=list(update_data.keys()) + ["updated_at"])
			updated += 1

	return {
		"step": "student_set_update",
		"updated": updated,
		"skipped": skipped,
	}


def student_set_delete(response_data):
	ids = response_data.get("student_set_ids") or []
	if not ids:
		return {"step": "student_set_delete", "updated": 0, "skipped": 0}

	inactive = EsStudentGroup.STATUS_TO_CODE["inactive"]
	existing = list(EsStudentGroup.objects.filter(tt_id__in=ids))
	existing_tt_ids = {obj.tt_id for obj in existing}

	with transaction.atomic():
		EsStudentGroup.objects.filter(tt_id__in=existing_tt_ids).update(status=inactive)

	return {
		"step": "student_set_delete",
		"updated": len(existing_tt_ids),
		"skipped": len(set(ids) - existing_tt_ids),
	}


# ============================
# STUDENT
# ============================

STUDENT_SCALAR_MAP = {
	"code": "code",
	"name": "name",
	"desc": "desc",
	"email": "email",
	"status": "status",
	"extra_data_1": "extra_data_1",
	"extra_data_2": "extra_data_2",
	"extra_data_3": "extra_data_3",
	"extra_data_4": "extra_data_4",
	"extra_data_5": "extra_data_5",
	"extra_data_6": "extra_data_6",
	"extra_data_7": "extra_data_7",
	"extra_data_8": "extra_data_8",
	"extra_data_9": "extra_data_9",
	"extra_data_10": "extra_data_10",
}


def student_create(response_data):
	payload = normalize_payload(response_data, "student")
	created = 0
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			code = row.get("code")
			obj = find_by_tt_id_then_code(EsStudent, tt_id, code)
			department_id = resolve_department_id(row.get("department") or row.get("department_id"))

			fields = {
				"tt_id": tt_id,
				"code": code,
				"name": row.get("name"),
				"desc": row.get("desc"),
				"email": row.get("email"),
				"status": row.get("status", EsStudent.STATUS_TO_CODE["active"]),
				"department_id": department_id,
			}

			for payload_key, model_field in STUDENT_SCALAR_MAP.items():
				if payload_key in row and payload_key not in fields:
					fields[model_field] = row.get(payload_key)

			if obj:
				for key, value in fields.items():
					if key == "code" and value is None:
						continue
					setattr(obj, key, value)
				obj.save()
				student = obj
				updated += 1
			else:
				student = EsStudent.objects.create(**{
					k: v for k, v in fields.items()
					if v is not None or k in ("desc", "department_id", "email")
				}, need_extra_provision=False)
				created += 1

			if "student_set" in row:
				replace_student_group_rows(student.id, resolve_student_group_ids_from_row(row))

	return {
		"step": "student_create",
		"created": created,
		"updated": updated,
		"skipped": skipped,
	}


def student_update(response_data):
	payload = normalize_payload(response_data, "student")
	if not payload:
		return {"step": "student_update", "updated": 0, "skipped": 0}

	single_item = len(payload) == 1
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			obj = EsStudent.objects.filter(tt_id=tt_id).first()
			if not obj:
				skipped += 1
				continue

			update_data = build_update_data(row, STUDENT_SCALAR_MAP, single_item=single_item)
			if "department" in row:
				update_data["department_id"] = resolve_department_id(row.get("department"))
			elif "department_id" in row:
				update_data["department_id"] = resolve_department_id(row.get("department_id"))

			if update_data:
				for key, value in update_data.items():
					setattr(obj, key, value)
				obj.save(update_fields=list(update_data.keys()) + ["updated_at"])
			if "student_set" in row:
				replace_student_group_rows(obj.id, resolve_student_group_ids_from_row(row))
			updated += 1

	return {
		"step": "student_update",
		"updated": updated,
		"skipped": skipped,
	}


def student_delete(response_data):
	ids = response_data.get("student_ids") or []
	if not ids:
		return {"step": "student_delete", "updated": 0, "skipped": 0}

	inactive = EsStudent.STATUS_TO_CODE["inactive"]
	existing = list(EsStudent.objects.filter(tt_id__in=ids))
	existing_tt_ids = {obj.tt_id for obj in existing}

	with transaction.atomic():
		EsStudent.objects.filter(tt_id__in=existing_tt_ids).update(status=inactive)

	return {
		"step": "student_delete",
		"updated": len(existing_tt_ids),
		"skipped": len(set(ids) - existing_tt_ids),
	}


# ============================
# STAFF
# ============================

def staff_create(response_data):
	payload = normalize_payload(response_data, "staff")
	created = 0
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			code = row.get("code")
			obj = find_by_tt_id_then_code(EsStaff, tt_id, code)
			department_id = resolve_department_id(row.get("department"))

			fields = {
				"tt_id": tt_id,
				"code": code,
				"name": row.get("name"),
				"email": row.get("email"),
				"desc": row.get("desc"),
				"status": row.get("status", EsStaff.STATUS_TO_CODE["active"]),
				"department_id": department_id,
			}

			if obj:
				for key, value in fields.items():
					if key == "code" and value is None:
						continue
					setattr(obj, key, value)
				obj.save()
				sync_staff_invigilator(obj)
				updated += 1
			else:
				staff = EsStaff.objects.create(**{
					k: v for k, v in fields.items() if v is not None or k in ("desc", "department_id", "email")
				})
				sync_staff_invigilator(staff)
				created += 1

	return {
		"step": "staff_create",
		"created": created,
		"updated": updated,
		"skipped": skipped,
	}


def staff_update(response_data):
	payload = normalize_payload(response_data, "staff")
	if not payload:
		return {"step": "staff_update", "updated": 0, "skipped": 0}

	single_item = len(payload) == 1
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			obj = EsStaff.objects.filter(tt_id=tt_id).first()
			if not obj:
				skipped += 1
				continue

			update_data = build_update_data(
				row,
				{
					"code": "code",
					"name": "name",
					"email": "email",
					"desc": "desc",
					"status": "status",
				},
				single_item=single_item,
			)
			if "department" in row:
				update_data["department_id"] = resolve_department_id(row.get("department"))

			if update_data:
				for key, value in update_data.items():
					setattr(obj, key, value)
				obj.save(update_fields=list(update_data.keys()) + ["updated_at"])
			sync_staff_invigilator(obj)
			updated += 1

	return {
		"step": "staff_update",
		"updated": updated,
		"skipped": skipped,
	}


def staff_delete(response_data):
	ids = response_data.get("staff_ids") or []
	if not ids:
		return {"step": "staff_delete", "updated": 0, "skipped": 0}

	inactive = EsStaff.STATUS_TO_CODE["inactive"]
	existing = list(EsStaff.objects.filter(tt_id__in=ids))
	existing_tt_ids = {obj.tt_id for obj in existing}

	with transaction.atomic():
		EsStaff.objects.filter(tt_id__in=existing_tt_ids).update(status=inactive)
		for staff in existing:
			staff.status = inactive
			sync_staff_invigilator(staff)

	return {
		"step": "staff_delete",
		"updated": len(existing_tt_ids),
		"skipped": len(set(ids) - existing_tt_ids),
	}


# ============================
# LOCATION
# ============================

def location_create(response_data):
	payload = normalize_payload(response_data, "location")
	created = 0
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			code = row.get("code")
			obj = find_by_tt_id_then_code(EsLocation, tt_id, code)
			department_id = resolve_department_id(row.get("department"))
			zone_id = resolve_zone_id(row.get("zone"))

			if obj:
				obj.tt_id = tt_id
				if code is not None:
					obj.code = code
				if "name" in row:
					obj.name = row.get("name")
				if "desc" in row:
					obj.desc = row.get("desc")
				if "status" in row:
					obj.status = row.get("status")
				elif obj.status is None:
					obj.status = EsLocation.STATUS_TO_CODE["active"]
				if "department" in row:
					obj.department_id = department_id
				if "zone" in row:
					obj.zone_id = zone_id
				obj.save()
				updated += 1
			else:
				EsLocation.objects.create(
					tt_id=tt_id,
					code=code,
					name=row.get("name") or code or str(tt_id),
					desc=row.get("desc"),
					status=row.get("status", EsLocation.STATUS_TO_CODE["active"]),
					department_id=department_id,
					zone_id=zone_id,
					seat_order=EsLocation.SEAT_ORDER_TO_CODE["column_first"],
					row=0,
					column=0,
					is_partition=False,
				)
				created += 1

	return {
		"step": "location_create",
		"created": created,
		"updated": updated,
		"skipped": skipped,
	}


def location_update(response_data):
	payload = normalize_payload(response_data, "location")
	if not payload:
		return {"step": "location_update", "updated": 0, "skipped": 0}

	single_item = len(payload) == 1
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			tt_id = row.get("id")
			if not tt_id:
				skipped += 1
				continue

			obj = EsLocation.objects.filter(tt_id=tt_id).first()
			if not obj:
				skipped += 1
				continue

			update_data = build_update_data(
				row,
				{
					"code": "code",
					"name": "name",
					"desc": "desc",
					"status": "status",
				},
				single_item=single_item,
			)
			if "department" in row:
				update_data["department_id"] = resolve_department_id(row.get("department"))
			if "zone" in row:
				update_data["zone_id"] = resolve_zone_id(row.get("zone"))

			if update_data:
				for key, value in update_data.items():
					setattr(obj, key, value)
				obj.save(update_fields=list(update_data.keys()) + ["updated_at"])
			updated += 1

	return {
		"step": "location_update",
		"updated": updated,
		"skipped": skipped,
	}


def location_delete(response_data):
	ids = response_data.get("location_ids") or []
	if not ids:
		return {"step": "location_delete", "updated": 0, "skipped": 0}

	inactive = EsLocation.STATUS_TO_CODE["inactive"]
	existing = list(EsLocation.objects.filter(tt_id__in=ids))
	existing_tt_ids = {obj.tt_id for obj in existing}

	with transaction.atomic():
		EsLocation.objects.filter(tt_id__in=existing_tt_ids).update(status=inactive)

	return {
		"step": "location_delete",
		"updated": len(existing_tt_ids),
		"skipped": len(set(ids) - existing_tt_ids),
	}


# ============================
# WEEK / WEEK_PATTERN / ACADEMIC_TERM SYNC
# ============================

def week_create(response_data):
	"""
	Upsert weeks from top-level `week` array.
	Kafka docs do not expose week_* methods; weeks arrive with academic_term_* payloads.
	"""
	payload = normalize_payload(response_data, "week")
	created = 0
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			week_id = row.get("id")
			if not week_id:
				skipped += 1
				continue

			obj = TtWeek.objects.filter(id=week_id).first()
			week_no = row.get("week")
			start_date = row.get("start_date")

			if obj:
				changed = False
				if week_no is not None and obj.week != week_no:
					obj.week = week_no
					changed = True
				if start_date is not None and str(obj.start_date) != str(start_date):
					obj.start_date = start_date
					changed = True
				if changed:
					obj.save()
					updated += 1
				else:
					skipped += 1
			else:
				TtWeek.objects.create(
					id=week_id,
					week=week_no,
					start_date=start_date,
				)
				created += 1

	return {
		"step": "week_create",
		"created": created,
		"updated": updated,
		"skipped": skipped,
	}


def academic_term_create(response_data):
	# Weeks are nested in academic_term_* messages (top-level `week` + academic_term.week ids).
	week_result = week_create(response_data)

	payload = normalize_payload(response_data, "academic_term")
	cleaned_rows = []
	week_map = {}

	for row in payload:
		academic_term_id = row.get("id")
		if not academic_term_id:
			continue

		cleaned_rows.append(
			{
				"id": academic_term_id,
				"code": row.get("code"),
				"name": row.get("name"),
				"start_date": row.get("start_date"),
				"end_date": row.get("end_date"),
				"start_time": row.get("start_time"),
				"end_time": row.get("end_time"),
				"status": row.get("status", TtAcademicTerm.STATUS_TO_CODE["active"]),
			}
		)
		week_map[academic_term_id] = row.get("week", []) or []

	with transaction.atomic():
		created_terms = bulk_create_skip_duplicates(TtAcademicTerm, cleaned_rows)
		created_ids = {obj.id for obj in created_terms}

		relation_candidates = []
		for academic_term_id, week_ids in week_map.items():
			if academic_term_id not in created_ids:
				continue
			for week_id in _filter_existing_week_ids(week_ids):
				relation_candidates.append(
					{
						"academic_term_id": academic_term_id,
						"week_id": week_id,
					}
				)

		existing_relations = set(
			TtAcademicTermWeek.objects.filter(
				academic_term_id__in=[r["academic_term_id"] for r in relation_candidates] or [0]
			).values_list("academic_term_id", "week_id")
		) if relation_candidates else set()

		to_create_rel = [
			TtAcademicTermWeek(**r)
			for r in relation_candidates
			if (r["academic_term_id"], r["week_id"]) not in existing_relations
		]

		if to_create_rel:
			TtAcademicTermWeek.objects.bulk_create(to_create_rel, batch_size=500)

	return {
		"step": "academic_term_create",
		"week_result": week_result,
		"created": len(created_terms),
		"skipped": len(cleaned_rows) - len(created_terms),
		"week_rel_created": len(to_create_rel) if relation_candidates else 0,
	}


def academic_term_update(response_data):
	# academic_term_update may also re-send top-level week objects; upsert them first.
	week_result = week_create(response_data) if response_data.get("week") else None

	payload = normalize_payload(response_data, "academic_term")
	if not payload:
		return {
			"step": "academic_term_update",
			"week_result": week_result,
			"updated": 0,
			"week_rel": 0,
			"skipped": 0,
		}

	single_item = len(payload) == 1
	incoming_ids = [row.get("id") for row in payload if row.get("id")]
	existing_ids = set(
		TtAcademicTerm.objects.filter(id__in=incoming_ids).values_list("id", flat=True)
	)

	week_map = {}
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			academic_term_id = row.get("id")
			if not academic_term_id or academic_term_id not in existing_ids:
				skipped += 1
				continue

			update_data = build_update_data(
				row,
				{
					"code": "code",
					"name": "name",
					"start_date": "start_date",
					"end_date": "end_date",
					"start_time": "start_time",
					"end_time": "end_time",
					"status": "status",
				},
				single_item=single_item,
			)

			if update_data:
				TtAcademicTerm.objects.filter(id=academic_term_id).update(**update_data)

			if "week" in row:
				week_map[academic_term_id] = _filter_existing_week_ids(row.get("week") or [])

			updated += 1

		week_rel = replace_relation_rows(
			TtAcademicTermWeek,
			"academic_term",
			"week",
			week_map,
		) if week_map else 0

	return {
		"step": "academic_term_update",
		"week_result": week_result,
		"updated": updated,
		"week_rel": week_rel,
		"skipped": skipped,
	}


def academic_term_delete(response_data):
	ids = response_data.get("academic_term_ids") or []
	if not ids:
		return {
			"step": "academic_term_delete",
			"deleted": 0,
			"skipped": 0,
			"academic_term_week_rel_deleted": 0,
		}

	existing_ids = set(
		TtAcademicTerm.objects.filter(id__in=ids).values_list("id", flat=True)
	)

	if not existing_ids:
		return {
			"step": "academic_term_delete",
			"deleted": 0,
			"skipped": len(ids),
			"academic_term_week_rel_deleted": 0,
		}

	with transaction.atomic():
		academic_term_week_rel_deleted = TtAcademicTermWeek.objects.filter(
			academic_term_id__in=existing_ids
		).count()
		deleted, _ = TtAcademicTerm.objects.filter(id__in=existing_ids).delete()

	return {
		"step": "academic_term_delete",
		"deleted": deleted,
		"skipped": len(ids) - len(existing_ids),
		"academic_term_week_rel_deleted": academic_term_week_rel_deleted,
	}


def week_pattern_create(response_data):
	payload = normalize_payload(response_data, "week_pattern")
	cleaned_rows = []
	week_map = {}

	for row in payload:
		week_pattern_id = row.get("id")
		if not week_pattern_id:
			continue

		cleaned_rows.append(
			{
				"id": week_pattern_id,
				"code": row.get("code"),
				"name": row.get("name"),
				"status": row.get("status", TtWeekPattern.STATUS_TO_CODE["active"]),
				"department_id": row.get("department"),
				"academic_term_id": row.get("academic_term"),
			}
		)
		week_map[week_pattern_id] = row.get("week", []) or []

	with transaction.atomic():
		created_patterns = bulk_create_skip_duplicates(TtWeekPattern, cleaned_rows)
		created_ids = {obj.id for obj in created_patterns}

		week_rel_candidates = []
		if created_ids:
			for week_pattern_id, week_ids in week_map.items():
				if week_pattern_id not in created_ids:
					continue
				for week_id in _filter_existing_week_ids(week_ids):
					week_rel_candidates.append(
						TtWeekPatternWeek(week_pattern_id=week_pattern_id, week_id=week_id)
					)

		if week_rel_candidates:
			TtWeekPatternWeek.objects.bulk_create(
				week_rel_candidates,
				batch_size=500,
				ignore_conflicts=True,
			)
			week_rel_created = len(week_rel_candidates)
		else:
			week_rel_created = 0

	return {
		"step": "week_pattern_create",
		"created": len(created_patterns),
		"skipped": len(cleaned_rows) - len(created_patterns),
		"week_rel": week_rel_created,
	}


def week_pattern_update(response_data):
	payload = normalize_payload(response_data, "week_pattern")
	if not payload:
		return {
			"step": "week_pattern_update",
			"updated": 0,
			"week_rel": 0,
			"skipped": 0,
		}

	single_item = len(payload) == 1
	incoming_ids = [row.get("id") for row in payload if row.get("id")]

	existing_ids = set(
		TtWeekPattern.objects.filter(id__in=incoming_ids).values_list("id", flat=True)
	)

	week_map = {}
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			week_pattern_id = row.get("id")
			if not week_pattern_id or week_pattern_id not in existing_ids:
				skipped += 1
				continue

			update_data = build_update_data(
				row,
				{
					"code": "code",
					"name": "name",
					"department": "department_id",
					"academic_term": "academic_term_id",
					"status": "status",
				},
				single_item=single_item,
			)
			if update_data:
				TtWeekPattern.objects.filter(id=week_pattern_id).update(**update_data)

			if "week" in row:
				week_map[week_pattern_id] = row.get("week") or []

			updated += 1

		week_rel_created = 0
		if week_map:
			TtWeekPatternWeek.objects.filter(week_pattern_id__in=list(week_map.keys())).delete()

			new_rows = []
			for week_pattern_id, week_ids in week_map.items():
				for week_id in _filter_existing_week_ids(week_ids):
					new_rows.append(
						TtWeekPatternWeek(week_pattern_id=week_pattern_id, week_id=week_id)
					)

			if new_rows:
				TtWeekPatternWeek.objects.bulk_create(
					new_rows,
					batch_size=500,
					ignore_conflicts=True,
				)
				week_rel_created = len(new_rows)

	return {
		"step": "week_pattern_update",
		"updated": updated,
		"week_rel": week_rel_created,
		"skipped": skipped,
	}


def week_pattern_delete(response_data):
	ids = response_data.get("week_pattern_ids") or []
	if not ids:
		return {
			"step": "week_pattern_delete",
			"deleted": 0,
			"skipped": 0,
			"week_pattern_week_rel_deleted": 0,
		}

	existing_ids = set(
		TtWeekPattern.objects.filter(id__in=ids).values_list("id", flat=True)
	)

	if not existing_ids:
		return {
			"step": "week_pattern_delete",
			"deleted": 0,
			"skipped": len(ids),
			"week_pattern_week_rel_deleted": 0,
		}

	with transaction.atomic():
		week_pattern_week_rel_deleted = TtWeekPatternWeek.objects.filter(
			week_pattern_id__in=existing_ids
		).count()
		deleted, _ = TtWeekPattern.objects.filter(id__in=existing_ids).delete()

	return {
		"step": "week_pattern_delete",
		"deleted": deleted,
		"skipped": len(ids) - len(existing_ids),
		"week_pattern_week_rel_deleted": week_pattern_week_rel_deleted,
	}


# ============================
# ACTIVITY / VARIANT / JTA / TEMPLATE
# ============================

ACTIVITY_SCALAR_MAP = {
	"code": "code",
	"name": "name",
	"desc": "desc",
	"activity_template": "activity_template_id",
	"activity_type": "activity_type_id",
	"department": "department_id",
	"zone": "zone_id",
	"module": "module_id",
	"academic_term": "academic_term_id",
	"duration": "duration",
	"slot_required": "slot_required",
	"planned_size": "planned_size",
	"real_size": "real_size",
	"scheduled_start_time": "scheduled_start_time",
	"scheduled_day": "scheduled_day",
	"scheduled_start_slot": "scheduled_start_slot",
	"scheduled": "scheduled",
	"week_pattern": "week_pattern_id",
	"is_jta": "is_jta",
	"jta_parent": "jta_parent_id",
	"is_variant": "is_variant",
	"variant_parent": "variant_parent_id",
	"status": "status",
}

ACTIVITY_DIRECT_ID_FIELDS = (
	"activity_template_id",
	"activity_type_id",
	"department_id",
	"zone_id",
	"module_id",
	"academic_term_id",
	"week_pattern_id",
	"jta_parent_id",
	"variant_parent_id",
)

TEMPLATE_SCALAR_MAP = {
	"code": "code",
	"name": "name",
	"desc": "desc",
	"duration": "duration",
	"slot_required": "slot_required",
	"planned_size": "planned_size",
	"activity_type": "activity_type_id",
	"department": "department_id",
	"zone": "zone_id",
	"module": "module_id",
	"academic_term": "academic_term_id",
	"week_pattern": "week_pattern_id",
	"status": "status",
}

TEMPLATE_FK_NULL_KEYS = (
	"activity_type",
	"department",
	"zone",
	"module",
	"academic_term",
	"week_pattern",
)

ACTIVITY_FK_NULL_KEYS = (
	"activity_template",
	"activity_type",
	"department",
	"zone",
	"module",
	"academic_term",
	"week_pattern",
	"jta_parent",
	"variant_parent",
)


def activity_row_id(row):
	return row.get("id") or row.get("activity_id")


def extract_week_ids_from_row(row):
	if "week" in row:
		week_ids = []
		for item in row.get("week") or []:
			if isinstance(item, dict):
				if item.get("id") is not None:
					week_ids.append(item.get("id"))
			elif item is not None:
				week_ids.append(item)
		return week_ids
	if "week_ids" in row:
		return [week_id for week_id in (row.get("week_ids") or []) if week_id]
	return None


def extract_location_tt_ids_from_row(row):
	location_ids = []
	for key in ("location_ids", "location", "location_preset"):
		if key not in row:
			continue
		location_ids.extend(_coerce_tt_ids(row.get(key)))
	return list(dict.fromkeys(location_ids)) if location_ids else None


def extract_staff_tt_ids_from_row(row):
	staff_ids = []
	for key in ("staff_ids", "staff", "staff_preset"):
		if key not in row:
			continue
		staff_ids.extend(_coerce_tt_ids(row.get(key)))
	return list(dict.fromkeys(staff_ids)) if staff_ids else None


def extract_student_set_tt_ids_from_row(row):
	student_set_ids = []
	for key in ("student_set_ids", "student_sets", "student_set"):
		if key not in row:
			continue
		student_set_ids.extend(_coerce_tt_ids(row.get(key)))
	return list(dict.fromkeys(student_set_ids)) if student_set_ids else None


def build_tt_activity_update_data(row, single_item=False):
	update_data = build_update_data(row, ACTIVITY_SCALAR_MAP, single_item=single_item)
	if "scheduled_start_time" in update_data:
		update_data["scheduled_start_time"] = parse_time_value(update_data["scheduled_start_time"])

	for key in ACTIVITY_FK_NULL_KEYS:
		if key in row and row.get(key) is None:
			update_data[f"{key}_id"] = None

	for field in ACTIVITY_DIRECT_ID_FIELDS:
		if field in row:
			update_data[field] = row.get(field)

	for payload_key, model_field in ACTIVITY_SCALAR_MAP.items():
		if payload_key in row and model_field not in update_data:
			value = row.get(payload_key)
			if model_field == "scheduled_start_time":
				value = parse_time_value(value)
			update_data[model_field] = value

	return update_data


def replace_activity_location_rows(activity_id, location_tt_ids):
	if not activity_id or not TtActivity.objects.filter(id=activity_id).exists():
		return 0

	TtActivityLocation.objects.filter(activity_id=activity_id).delete()
	if not location_tt_ids:
		return 0

	rows = [
		TtActivityLocation(activity_id=activity_id, location_tt_id=location_tt_id)
		for location_tt_id in location_tt_ids
		if location_tt_id
	]
	if rows:
		TtActivityLocation.objects.bulk_create(rows, batch_size=500, ignore_conflicts=True)
	return len(rows)


def replace_activity_staff_rows(activity_id, staff_tt_ids):
	if not activity_id or not TtActivity.objects.filter(id=activity_id).exists():
		return 0

	TtActivityStaff.objects.filter(activity_id=activity_id).delete()
	if not staff_tt_ids:
		return 0

	rows = [
		TtActivityStaff(activity_id=activity_id, staff_tt_id=staff_tt_id)
		for staff_tt_id in staff_tt_ids
		if staff_tt_id
	]
	if rows:
		TtActivityStaff.objects.bulk_create(rows, batch_size=500, ignore_conflicts=True)
	return len(rows)


def replace_activity_student_set_rows(activity_id, student_set_tt_ids):
	if not activity_id or not TtActivity.objects.filter(id=activity_id).exists():
		return 0

	TtStudentSetActivity.objects.filter(activity_id=activity_id).delete()
	if not student_set_tt_ids:
		return 0

	rows = [
		TtStudentSetActivity(activity_id=activity_id, student_set_tt_id=student_set_tt_id)
		for student_set_tt_id in student_set_tt_ids
		if student_set_tt_id
	]
	if rows:
		TtStudentSetActivity.objects.bulk_create(rows, batch_size=500, ignore_conflicts=True)
	return len(rows)


def sync_tt_activity_relations(activity_id, row):
	if not activity_id or not TtActivity.objects.filter(id=activity_id).exists():
		return

	week_ids = extract_week_ids_from_row(row) if ("week" in row or "week_ids" in row) else None
	if week_ids is not None:
		existing_week_ids = _filter_existing_week_ids(week_ids)
		replace_relation_rows(TtActivityWeek, "activity", "week", {activity_id: existing_week_ids})

	if any(key in row for key in ("location", "location_ids", "location_preset")):
		replace_activity_location_rows(activity_id, extract_location_tt_ids_from_row(row) or [])

	if any(key in row for key in ("staff", "staff_ids", "staff_preset")):
		replace_activity_staff_rows(activity_id, extract_staff_tt_ids_from_row(row) or [])

	if any(key in row for key in ("student_sets", "student_set_ids", "student_set")):
		replace_activity_student_set_rows(activity_id, extract_student_set_tt_ids_from_row(row) or [])


def upsert_tt_activity(row, single_item=False):
	activity_id = activity_row_id(row)
	if not activity_id:
		return None

	update_data = build_tt_activity_update_data(row, single_item=single_item)
	obj = TtActivity.objects.filter(id=activity_id).first()
	created = False

	if obj:
		if update_data:
			for key, value in update_data.items():
				setattr(obj, key, value)
			obj.save()
	else:
		if "status" not in update_data:
			update_data["status"] = TtActivity.STATUS_TO_CODE["active"]
		if "code" not in update_data:
			update_data["code"] = str(activity_id)
		if "name" not in update_data:
			update_data["name"] = update_data.get("code") or str(activity_id)
		obj = TtActivity.objects.create(id=activity_id, **update_data)
		created = True

	if "week_ids" in row:
		for week_id in _filter_existing_week_ids(row.get("week_ids") or []):
			TtActivityWeek.objects.get_or_create(activity_id=activity_id, week_id=week_id)

	if "delete_week_ids" in row:
		TtActivityWeek.objects.filter(
			activity_id=activity_id,
			week_id__in=row.get("delete_week_ids") or [],
		).delete()

	sync_tt_activity_relations(activity_id, row)

	return {"id": activity_id, "created": created}


def apply_partial_tt_activity_update(row):
	activity_id = activity_row_id(row)
	if not activity_id:
		return False

	obj = TtActivity.objects.filter(id=activity_id).first()
	if not obj:
		return False

	update_data = build_tt_activity_update_data(row, single_item=True)
	if update_data:
		for key, value in update_data.items():
			setattr(obj, key, value)
		obj.save()

	if "week_ids" in row:
		for week_id in _filter_existing_week_ids(row.get("week_ids") or []):
			TtActivityWeek.objects.get_or_create(activity_id=activity_id, week_id=week_id)

	if "delete_week_ids" in row:
		TtActivityWeek.objects.filter(
			activity_id=activity_id,
			week_id__in=row.get("delete_week_ids") or [],
		).delete()

	sync_tt_activity_relations(activity_id, row)
	return True


def sync_tt_activity_from_schedule_row(row):
	activity_id = activity_row_id(row)
	if not activity_id:
		return False

	update_data = build_tt_activity_update_data(row, single_item=True)
	obj = TtActivity.objects.filter(id=activity_id).first()

	if obj:
		if update_data:
			for key, value in update_data.items():
				setattr(obj, key, value)
			obj.save()
	elif update_data:
		if "status" not in update_data:
			update_data["status"] = TtActivity.STATUS_TO_CODE["active"]
		if "code" not in update_data:
			update_data["code"] = str(activity_id)
		if "name" not in update_data:
			update_data["name"] = str(activity_id)
		TtActivity.objects.create(id=activity_id, **update_data)
	else:
		TtActivity.objects.create(
			id=activity_id,
			code=row.get("code") or str(activity_id),
			name=row.get("name") or row.get("code") or str(activity_id),
			status=row.get("status", TtActivity.STATUS_TO_CODE["active"]),
		)

	sync_tt_activity_relations(activity_id, row)
	return True


def build_tt_activity_template_fields(row, single_item=False):
	template_id = row.get("id")
	if not template_id:
		return None

	update_data = build_update_data(row, TEMPLATE_SCALAR_MAP, single_item=single_item)
	for key in TEMPLATE_FK_NULL_KEYS:
		if key in row and row.get(key) is None:
			update_data[f"{key}_id"] = None

	for field in (
		"activity_type_id",
		"department_id",
		"zone_id",
		"module_id",
		"academic_term_id",
		"week_pattern_id",
	):
		if field in row:
			update_data[field] = row.get(field)

	for payload_key, model_field in TEMPLATE_SCALAR_MAP.items():
		if payload_key in row and model_field not in update_data:
			update_data[model_field] = row.get(payload_key)

	if "status" not in update_data and not TtActivityTemplate.objects.filter(id=template_id).exists():
		update_data["status"] = TtActivityTemplate.STATUS_TO_CODE["active"]

	return {"id": template_id, **update_data}


def upsert_tt_activity_template(row, single_item=False):
	fields = build_tt_activity_template_fields(row, single_item=single_item)
	if not fields:
		return None

	template_id = fields.pop("id")
	obj = TtActivityTemplate.objects.filter(id=template_id).first()
	created = False

	if obj:
		for key, value in fields.items():
			setattr(obj, key, value)
		obj.save()
	else:
		if "code" not in fields:
			fields["code"] = str(template_id)
		if "name" not in fields:
			fields["name"] = fields.get("code") or str(template_id)
		if "status" not in fields:
			fields["status"] = TtActivityTemplate.STATUS_TO_CODE["active"]
		obj = TtActivityTemplate.objects.create(id=template_id, **fields)
		created = True

	return {"id": template_id, "created": created}


def activity_create(response_data):
	payload = normalize_payload(response_data, "activity")
	created = 0
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			result = upsert_tt_activity(row)
			if not result:
				skipped += 1
				continue
			if result["created"]:
				created += 1
			else:
				updated += 1

	return {
		"step": "activity_create",
		"created": created,
		"updated": updated,
		"skipped": skipped,
	}


def activity_update(response_data):
	payload = normalize_payload(response_data, "activity")
	if not payload:
		return {"step": "activity_update", "updated": 0, "skipped": 0}

	single_item = len(payload) == 1
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			result = upsert_tt_activity(row, single_item=single_item)
			if not result:
				skipped += 1
				continue
			updated += 1

	return {
		"step": "activity_update",
		"updated": updated,
		"skipped": skipped,
	}


def activity_delete(response_data):
	ids = response_data.get("activity_ids") or []
	if not ids:
		return {"step": "activity_delete", "deleted": 0, "skipped": 0}

	existing_ids = set(TtActivity.objects.filter(id__in=ids).values_list("id", flat=True))
	if not existing_ids:
		return {"step": "activity_delete", "deleted": 0, "skipped": len(ids)}

	with transaction.atomic():
		deleted, _ = TtActivity.objects.filter(id__in=existing_ids).delete()

	return {
		"step": "activity_delete",
		"deleted": deleted,
		"skipped": len(ids) - len(existing_ids),
	}


def variant_create(response_data):
	parent = response_data.get("variant_parent") or {}
	child = response_data.get("variant_child") or {}
	parent_updated = False
	child_result = None

	with transaction.atomic():
		if parent:
			parent_updated = apply_partial_tt_activity_update(parent)
		if child:
			child_result = upsert_tt_activity(child)

	return {
		"step": "variant_create",
		"parent_updated": parent_updated,
		"child_created": child_result["created"] if child_result else False,
		"child_updated": (not child_result["created"]) if child_result else False,
		"skipped": not parent and not child,
	}


def jta_create(response_data):
	create_activity = response_data.get("create_activity") or {}
	update_rows = response_data.get("update_activity") or []
	if isinstance(update_rows, dict):
		update_rows = [update_rows]

	create_result = None
	updated = 0
	skipped = 0

	with transaction.atomic():
		if create_activity:
			create_result = upsert_tt_activity(create_activity)

		for row in update_rows:
			if apply_partial_tt_activity_update(row):
				updated += 1
			else:
				skipped += 1

	return {
		"step": "jta_create",
		"created": 1 if create_result and create_result["created"] else 0,
		"updated": updated + (1 if create_result and not create_result["created"] else 0),
		"skipped": skipped + (0 if create_activity else 1),
	}


def jta_split(response_data):
	update_rows = response_data.get("activity_update") or []
	if isinstance(update_rows, dict):
		update_rows = [update_rows]

	delete_id = response_data.get("acitvity_delete")
	if delete_id is None:
		delete_id = response_data.get("activity_delete")

	updated = 0
	skipped = 0
	deleted = 0

	with transaction.atomic():
		for row in update_rows:
			if apply_partial_tt_activity_update(row):
				updated += 1
			else:
				skipped += 1

		if delete_id is not None:
			existing_ids = set(TtActivity.objects.filter(id=delete_id).values_list("id", flat=True))
			if existing_ids:
				deleted, _ = TtActivity.objects.filter(id__in=existing_ids).delete()
			else:
				skipped += 1

	return {
		"step": "jta_split",
		"updated": updated,
		"deleted": deleted,
		"skipped": skipped,
	}


def activity_template_create(response_data):
	payload = normalize_payload(response_data, "activity_template")
	created = 0
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			result = upsert_tt_activity_template(row)
			if not result:
				skipped += 1
				continue
			if result["created"]:
				created += 1
			else:
				updated += 1

	return {
		"step": "activity_template_create",
		"created": created,
		"updated": updated,
		"skipped": skipped,
	}


def activity_template_update(response_data):
	payload = normalize_payload(response_data, "activity_template")
	if not payload:
		return {"step": "activity_template_update", "updated": 0, "skipped": 0}

	single_item = len(payload) == 1
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			result = upsert_tt_activity_template(row, single_item=single_item)
			if not result:
				skipped += 1
				continue
			updated += 1

	return {
		"step": "activity_template_update",
		"updated": updated,
		"skipped": skipped,
	}


def activity_template_delete(response_data):
	ids = response_data.get("activity_template_ids") or []
	if not ids:
		return {"step": "activity_template_delete", "deleted": 0, "skipped": 0}

	existing_ids = set(TtActivityTemplate.objects.filter(id__in=ids).values_list("id", flat=True))
	if not existing_ids:
		return {"step": "activity_template_delete", "deleted": 0, "skipped": len(ids)}

	with transaction.atomic():
		deleted, _ = TtActivityTemplate.objects.filter(id__in=existing_ids).delete()

	return {
		"step": "activity_template_delete",
		"deleted": deleted,
		"skipped": len(ids) - len(existing_ids),
	}


def unschedule_tt_activities(response_data):
	payload = normalize_payload(response_data, "activity")
	updated = 0
	skipped = 0

	with transaction.atomic():
		for row in payload:
			activity_id = activity_row_id(row)
			if not activity_id:
				skipped += 1
				continue

			obj = TtActivity.objects.filter(id=activity_id).first()
			if not obj:
				skipped += 1
				continue

			obj.scheduled = row.get("scheduled", 0)
			obj.scheduled_start_time = parse_time_value(row.get("scheduled_start_time"))
			obj.scheduled_day = row.get("scheduled_day")
			obj.scheduled_start_slot = row.get("scheduled_start_slot")
			obj.save(
				update_fields=[
					"scheduled",
					"scheduled_start_time",
					"scheduled_day",
					"scheduled_start_slot",
				]
			)
			replace_activity_location_rows(activity_id, extract_location_tt_ids_from_row(row) or [])
			updated += 1

	return {
		"step": "unschedule",
		"updated": updated,
		"skipped": skipped,
	}


# ============================
# SCHEDULE CONFLICT → UNSCHEDULE EXAM ACTIVITIES
# ============================

def schedule_location_conflicts(response_data):
	payload = normalize_payload(response_data, "activity")
	unscheduled_ids = []
	checked = 0
	skipped = 0
	synced = 0
	deleted = 0

	delete_ids = response_data.get("delete_activity") or response_data.get("delete_activity_ids") or []
	if delete_ids and not isinstance(delete_ids, list):
		delete_ids = [delete_ids]
	deleted_id_set = set()
	if delete_ids:
		existing_delete_ids = set(
			TtActivity.objects.filter(id__in=delete_ids).values_list("id", flat=True)
		)
		if existing_delete_ids:
			TtActivity.objects.filter(id__in=existing_delete_ids).delete()
			deleted = len(existing_delete_ids)
			deleted_id_set = existing_delete_ids

	for row in payload:
		activity_id = activity_row_id(row)
		if activity_id in deleted_id_set:
			continue

		if sync_tt_activity_from_schedule_row(row):
			synced += 1

		stored = TtActivity.objects.filter(id=activity_id).first() if activity_id else None

		location_ids = collect_location_ids_from_activity_row(row)
		if not location_ids and activity_id:
			location_ids = list(
				TtActivityLocation.objects.filter(activity_id=activity_id).values_list(
					"location_tt_id", flat=True
				)
			)

		busy_window = tt_busy_window(row) or (tt_busy_window_from_activity(stored) if stored else None)
		payload_week_ids = resolve_week_ids_for_schedule_row(row)
		if not location_ids or busy_window is None or payload_week_ids is None:
			skipped += 1
			continue

		result = unschedule_conflicting_exam_activities(location_ids, busy_window, payload_week_ids)
		checked += result["checked_locations"]
		unscheduled_ids.extend(result["unscheduled_activity_ids"])

	return {
		"step": "schedule_location_conflict",
		"synced": synced,
		"deleted": deleted,
		"checked_locations": checked,
		"unscheduled": len(unscheduled_ids),
		"unscheduled_activity_ids": list(dict.fromkeys(unscheduled_ids)),
		"skipped": skipped,
	}


def method_skipped(method):
	return {
		"step": method,
		"success": True,
		"skipped": True,
		"message": f"{method} not handled yet",
	}


# ============================
# MAIN LOOP
# ============================

def run_consumer():
	consumer = get_consumer()
	while True:
		response_data = None
		header_data = {}
		msg_str = None
		msg = None

		try:
			msg = consumer.poll(1.0)
			if msg is None:
				continue
			if msg.error():
				continue

			raw_headers = msg.headers()
			if raw_headers:
				header_data = {key: value.decode("utf-8") for key, value in raw_headers}

			msg_str = msg.value().decode()
			response_data = json.loads(msg_str)
			result = None

			if "method" in header_data:
				method = header_data["method"]
				match method:
					case "department_create":
						result = department_create(response_data)
					case "department_update":
						result = department_update(response_data)
					case "department_delete":
						result = department_delete(response_data)
					case "zone_create":
						result = zone_create(response_data)
					case "zone_update":
						result = zone_update(response_data)
					case "zone_delete":
						result = zone_delete(response_data)
					case "student_set_create":
						result = student_set_create(response_data)
					case "student_set_update":
						result = student_set_update(response_data)
					case "student_set_delete":
						result = student_set_delete(response_data)
					case "student_create":
						result = student_create(response_data)
					case "student_update":
						result = student_update(response_data)
					case "student_delete":
						result = student_delete(response_data)
					case "staff_create":
						result = staff_create(response_data)
					case "staff_update":
						result = staff_update(response_data)
					case "staff_delete":
						result = staff_delete(response_data)
					case "location_create":
						result = location_create(response_data)
					case "location_update":
						result = location_update(response_data)
					case "location_delete":
						result = location_delete(response_data)
					case "academic_term_create":
						result = academic_term_create(response_data)
					case "academic_term_update":
						result = academic_term_update(response_data)
					case "academic_term_delete":
						result = academic_term_delete(response_data)
					case "week_pattern_create":
						result = week_pattern_create(response_data)
					case "week_pattern_update":
						result = week_pattern_update(response_data)
					case "week_pattern_delete":
						result = week_pattern_delete(response_data)
					case "activity_create":
						result = activity_create(response_data)
					case "activity_update":
						result = activity_update(response_data)
					case "activity_delete":
						result = activity_delete(response_data)
					case "variant_create":
						result = variant_create(response_data)
					case "jta_create":
						result = jta_create(response_data)
					case "jta_split":
						result = jta_split(response_data)
					case "activity_template_create":
						result = activity_template_create(response_data)
					case "activity_template_update":
						result = activity_template_update(response_data)
					case "activity_template_delete":
						result = activity_template_delete(response_data)
					case "unschedule":
						result = unschedule_tt_activities(response_data)
					case "schedule" | "booking_schedule" | "swap" | "booking_swap":
						result = schedule_location_conflicts(response_data)
					case _:
						result = method_skipped(method)
			else:
				result = {
					"step": "missing_method",
					"success": False,
					"error": "Kafka header method is missing",
				}

			if result:
				append_log(response_data, header_data, result)
			consumer.commit(msg)
		except (OperationalError, InterfaceError, TimeoutError) 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"],
			)

			error_result = {
				"step": "kafka_connection_exception",
				"success": False,
				"error": str(e),
				"descr": e_details["descr"],
				"url": e_details["url"],
				"trace": e_details["trace"],
			}

			try:
				if isinstance(response_data, dict):
					append_log(response_data, header_data, error_result)
				else:
					append_log(
						{
							"request_id": None,
							"raw_data": msg_str,
						},
						header_data,
						error_result,
					)
			except Exception as log_error:
				print("failed to insert kafka connection error log:", log_error)

			time.sleep(5)
			continue

		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"],
			)

			error_result = {
				"step": "kafka_exception",
				"success": False,
				"error": str(e),
				"descr": e_details["descr"],
				"url": e_details["url"],
				"trace": e_details["trace"],
			}

			try:
				if isinstance(response_data, dict):
					append_log(response_data, header_data, error_result)
				else:
					append_log(
						{
							"request_id": None,
							"raw_data": msg_str,
						},
						header_data,
						error_result,
					)
			except Exception as log_error:
				print("failed to insert kafka error log:", log_error)

			# commit so a bad message does not infinite-loop
			if msg is not None:
				try:
					consumer.commit(msg)
				except Exception:
					pass
			continue


if __name__ == "__main__":
	run_consumer()