import csv from datetime import date, datetime, timedelta from decimal import Decimal, InvalidOperation, ROUND_CEILING from io import StringIO from django.core.exceptions import ValidationError from django.core.validators import validate_email from django.db import transaction from django.db.models import Q from django.utils import timezone from apps.audit.services import log_event from .models import ImportJob, ImportRowError def _read_csv(uploaded_file): uploaded_file.seek(0) raw = uploaded_file.read() text = raw.decode("utf-8-sig") if isinstance(raw, (bytes, bytearray)) else str(raw) return list(csv.DictReader(StringIO(text))) def _value(row, *names): for name in names: if name in row and row[name] is not None: value = str(row[name]).strip() if value: return value return "" def _date(value, field): try: return date.fromisoformat(value) except Exception as exc: raise ValidationError({field: f"Invalid date '{value}'. Use YYYY-MM-DD."}) from exc def _month(value, field): try: if len(value) == 7: return date.fromisoformat(value + "-01") parsed = date.fromisoformat(value) return parsed.replace(day=1) except Exception as exc: raise ValidationError({field: f"Invalid month '{value}'. Use YYYY-MM."}) from exc def _time(value, field): if not value: return None try: return datetime.strptime(str(value).strip(), "%H:%M").time() except Exception as exc: raise ValidationError({field: f"Invalid time '{value}'. Use HH:MM (24-hour)."}) from exc def _attendance_datetimes(employee, work_date, check_in_value, check_out_value): """Parse CSV punch times and preserve supported overnight shifts.""" check_in_time = _time(check_in_value, "Check In") check_out_time = _time(check_out_value, "Check Out") if check_out_time and not check_in_time: raise ValidationError({"Check In": "Check In is required when Check Out is provided."}) tz = timezone.get_current_timezone() check_in = timezone.make_aware(datetime.combine(work_date, check_in_time), tz) if check_in_time else None check_out = None if check_out_time: out_date = work_date if check_in_time and check_out_time <= check_in_time: shift = getattr(employee, "shift", None) if not shift or shift.end_time > shift.start_time: raise ValidationError({"Check Out": "Check Out must be later than Check In for a non-overnight shift."}) out_date = work_date + timedelta(days=1) check_out = timezone.make_aware(datetime.combine(out_date, check_out_time), tz) return check_in, check_out def _decimal(value, field, *, allow_zero=True): try: result = Decimal(str(value)) except (InvalidOperation, TypeError) as exc: raise ValidationError({field: f"Invalid number '{value}'."}) from exc if result < 0 or (not allow_zero and result <= 0): raise ValidationError({field: "Value must be greater than zero." if not allow_zero else "Value cannot be negative."}) return result def _validate_salary_import_window(employee, effective_from, *, lock=False): """Apply the same non-overlap rule used by the salary-assignment form. Salary opening CSV has no Effective To column. A brand-new row is therefore open-ended. If the CSV updates an existing row with the same Effective Date, preserve that row's existing end date and validate the remaining timeline. """ if effective_from < employee.joining_date: raise ValidationError({"Effective Date": "Salary effective date cannot be earlier than the employee joining date."}) from apps.payroll.models import EmployeeSalaryAssignment same_qs = EmployeeSalaryAssignment.objects.filter(employee=employee, effective_from=effective_from).order_by("pk") if lock: same_qs = same_qs.select_for_update() same_rows = list(same_qs[:2]) if len(same_rows) > 1: raise ValidationError("Multiple salary assignments already exist for the same employee/effective date; correct the existing data before import.") existing = same_rows[0] if same_rows else None effective_to = existing.effective_to if existing else None overlaps = EmployeeSalaryAssignment.objects.filter(employee=employee) if existing: overlaps = overlaps.exclude(pk=existing.pk) if lock: overlaps = overlaps.select_for_update() if effective_to: overlaps = overlaps.filter(effective_from__lte=effective_to).filter( Q(effective_to__isnull=True) | Q(effective_to__gte=effective_from) ) else: overlaps = overlaps.filter(Q(effective_to__isnull=True) | Q(effective_to__gte=effective_from)) if overlaps.exists(): raise ValidationError("Salary assignment effective dates overlap an existing assignment for this employee.") return existing def _unique_named(qs, label): matches = list(qs[:2]) if not matches: return None if len(matches) > 1: raise ValidationError(f"{label} name is ambiguous in the selected company. Use unique master names before import.") return matches[0] def _row_error_text(exc): if hasattr(exc, "message_dict"): return "; ".join(f"{k}: {', '.join(map(str, v))}" for k, v in exc.message_dict.items()) return "; ".join(getattr(exc, "messages", [str(exc)])) def validate_row(job, row): company = job.company kind = job.import_type.upper() normalized = {} if kind == "EMPLOYEE": from apps.organization.models import Branch, Department, Designation, Shift code = _value(row, "Employee ID", "employee_code") name = _value(row, "Full Name", "full_name") if not code or not name: raise ValidationError("Employee ID and Full Name are required.") branch = _unique_named(Branch.objects.filter(company=company, name__iexact=_value(row, "Branch"), is_active=True), "Branch") department = _unique_named(Department.objects.filter(company=company, name__iexact=_value(row, "Department"), is_active=True), "Department") designation = _unique_named(Designation.objects.filter(company=company, name__iexact=_value(row, "Designation"), is_active=True), "Designation") shift_name = _value(row, "Shift") shift = _unique_named(Shift.objects.filter(company=company, name__iexact=shift_name, is_active=True), "Shift") if shift_name else None if not branch or not department or not designation: raise ValidationError("Branch, Department and Designation must already exist in the selected company.") if shift_name and not shift: raise ValidationError("Shift was not found in the selected company.") from apps.employees.models import Employee email = _value(row, "Email") if email: validate_email(email) status = (_value(row, "Status") or Employee.Status.ACTIVE).upper().replace(" ", "_") valid_statuses = {value for value, _ in Employee.Status.choices} if status not in valid_statuses: raise ValidationError("Invalid employee status.") normalized = { "employee_code": code, "full_name": name, "mobile": _value(row, "Mobile"), "email": email, "branch_id": branch.pk, "department_id": department.pk, "designation_id": designation.pk, "shift_id": shift.pk if shift else None, "employee_type": _value(row, "Employee Type") or "Permanent", "joining_date": str(_date(_value(row, "Joining Date"), "Joining Date")), "status": status, "blood_group": _value(row, "Blood Group"), "nid_or_passport": _value(row, "NID"), "present_address": _value(row, "Present Address", "Location"), "emergency_contact": _value(row, "Emergency Contact"), } elif kind == "SALARY": from apps.employees.models import Employee from apps.payroll.models import SalaryStructure employee = Employee.objects.filter(company=company, employee_code=_value(row, "Employee ID")).first() structure = _unique_named( SalaryStructure.objects.filter(company=company, name__iexact=_value(row, "Salary Template"), is_active=True), "Salary Template", ) if not employee or not structure: raise ValidationError("Employee and Salary Template must exist in the selected company.") basic = _decimal(_value(row, "Basic Salary"), "Basic Salary", allow_zero=False) gross_text = _value(row, "Gross Salary") gross = _decimal(gross_text, "Gross Salary", allow_zero=False) if gross_text else basic if basic > gross: raise ValidationError("Basic Salary cannot be greater than Gross Salary.") effective_from = _date(_value(row, "Effective Date"), "Effective Date") _validate_salary_import_window(employee, effective_from) normalized = { "employee_id": employee.pk, "structure_id": structure.pk, "basic_salary": str(basic), "gross_salary": str(gross), "effective_from": str(effective_from), "ot_eligible": _value(row, "OT Applicable").lower() in {"yes", "y", "true", "1"}, } elif kind == "LEAVE": from apps.employees.models import Employee from apps.leave.models import LeaveType employee = Employee.objects.filter(company=company, employee_code=_value(row, "Employee ID")).first() leave_type = _unique_named( LeaveType.objects.filter(company=company, name__iexact=_value(row, "Leave Type"), is_active=True), "Leave Type", ) if not employee or not leave_type: raise ValidationError("Employee and Leave Type must exist in the selected company.") try: leave_year = int(_value(row, "Year")) except (TypeError, ValueError) as exc: raise ValidationError({"Year": "Leave year must be a four-digit year."}) from exc if leave_year < 2020 or leave_year > 2100: raise ValidationError({"Year": "Leave year must be between 2020 and 2100."}) normalized = { "employee_id": employee.pk, "leave_type_id": leave_type.pk, "opening": str(_decimal(_value(row, "Opening Balance"), "Opening Balance")), "year": leave_year, } elif kind == "LOAN": from apps.employees.models import Employee employee = Employee.objects.filter(company=company, employee_code=_value(row, "Employee ID")).first() if not employee: raise ValidationError("Employee was not found in the selected company.") loan_reference = _value(row, "Loan ID") if not loan_reference: raise ValidationError("Loan ID is required for opening-loan import.") principal = _decimal(_value(row, "Principal"), "Principal", allow_zero=False) outstanding = _decimal(_value(row, "Outstanding"), "Outstanding") installment = _decimal(_value(row, "Installment Amount"), "Installment Amount", allow_zero=False) if outstanding > principal: raise ValidationError("Outstanding cannot exceed Principal.") imported_status = _value(row, "Status").upper().replace(" ", "_") derived_status = "ACTIVE" if outstanding > 0 else "CLOSED" if imported_status and imported_status != derived_status: raise ValidationError( f"Loan Status must be {derived_status} when Outstanding is {outstanding}." ) normalized = { "employee_id": employee.pk, "request_reference": loan_reference, "loan_type": "SALARY_ADVANCE" if "advance" in _value(row, "Type").lower() else "LOAN", "requested_amount": str(principal), "sanctioned_amount": str(principal), "outstanding_balance": str(outstanding), "installment_amount": str(installment), "start_month": str(_month(_value(row, "Start Month"), "Start Month")), "disbursement_date": str(_date(_value(row, "Issue Date"), "Issue Date")), "reason": _value(row, "Purpose"), } elif kind == "ATTENDANCE": from apps.employees.models import Employee from apps.attendance.models import AttendancePeriod, AttendanceRecord from apps.attendance.services import employee_in_service_on, validate_attendance_values employee = Employee.objects.select_related("shift", "company").filter(company=company, employee_code=_value(row, "Employee ID")).first() if not employee: raise ValidationError("Employee was not found in the selected company.") work_date = _date(_value(row, "Date"), "Date") if not employee_in_service_on(employee, work_date): raise ValidationError("Attendance date is outside the employee service period.") if AttendancePeriod.objects.filter( company=company, year=work_date.year, month=work_date.month, status=AttendancePeriod.Status.FINALIZED ).exists(): raise ValidationError("Attendance period is finalized. Reopen it before importing attendance corrections.") existing = AttendanceRecord.objects.filter(employee=employee, work_date=work_date).first() if existing and existing.finalized: raise ValidationError("Attendance record is finalized. Reopen the attendance period before import.") status_label = (_value(row, "Status") or AttendanceRecord.Status.PRESENT).upper().replace(" ", "_") valid_statuses = {x for x, _ in AttendanceRecord.Status.choices} if status_label not in valid_statuses: raise ValidationError("Invalid attendance status.") check_in_text = _value(row, "Check In") check_out_text = _value(row, "Check Out") parsed_check_in, parsed_check_out = _attendance_datetimes(employee, work_date, check_in_text, check_out_text) validate_attendance_values(employee, work_date, parsed_check_in, parsed_check_out, status_label) ot_hours_text = _value(row, "OT Hours") if ot_hours_text and _decimal(ot_hours_text, "OT Hours") != Decimal("0"): raise ValidationError( "OT Hours cannot be imported through Attendance CSV because OT requires the approval workflow. " "Import attendance first, then submit/approve OT separately." ) normalized = { "employee_id": employee.pk, "work_date": str(work_date), "check_in": check_in_text, "check_out": check_out_text, "status": status_label, "remarks": _value(row, "Remarks"), } else: raise ValidationError("Unsupported import type.") return normalized def _staged_identity(kind, data): kind = kind.upper() if kind == "EMPLOYEE": return (kind, str(data["employee_code"]).casefold()) if kind == "SALARY": return (kind, data["employee_id"], data["effective_from"]) if kind == "LEAVE": return (kind, data["employee_id"], data["leave_type_id"], data["year"]) if kind == "LOAN": return (kind, data["employee_id"], str(data["request_reference"]).casefold()) if kind == "ATTENDANCE": return (kind, data["employee_id"], data["work_date"]) return None @transaction.atomic def validate_import_job(job, *, user=None): job = ImportJob.objects.select_for_update().select_related("company").get(pk=job.pk) if job.status not in {"UPLOADED", "VALIDATED_WITH_ERRORS", "VALIDATED"}: raise ValidationError("Only uploaded/validated jobs can be validated again.") rows = _read_csv(job.source_file) if not rows: raise ValidationError("CSV contains no data rows. Add at least one row and validate again.") job.row_errors.all().delete() staged = [] seen_identities = {} kind = job.import_type.upper() for index, row in enumerate(rows, start=2): try: normalized = validate_row(job, row) identity = _staged_identity(kind, normalized) if identity is not None and identity in seen_identities: raise ValidationError(f"Duplicate import key in CSV; first seen on row {seen_identities[identity]}.") if identity is not None: seen_identities[identity] = index staged.append({"row_number": index, "raw": row, "normalized": normalized}) except Exception as exc: ImportRowError.objects.create(job=job, row_number=index, message=_row_error_text(exc), raw_row=row) job.total_rows = len(rows) job.valid_rows = len(staged) job.error_rows = job.row_errors.count() job.staged_rows = staged job.status = "VALIDATED" if job.error_rows == 0 else "VALIDATED_WITH_ERRORS" job.result_summary = {"validated_at": timezone.now().isoformat()} job.save() log_event(user=user, action="VALIDATE_IMPORT", module="imports", object_id=job.pk, object_repr=f"{job.import_type}/{job.company.code}", new_values={"total": job.total_rows, "valid": job.valid_rows, "errors": job.error_rows}) return job @transaction.atomic def confirm_import_job(job, *, user=None): job = ImportJob.objects.select_for_update().select_related("company").get(pk=job.pk) if job.status != "VALIDATED" or job.error_rows: raise ValidationError("Import must validate with zero errors before confirmation.") if job.confirmed_at: raise ValidationError("This import job has already been confirmed.") created = updated = 0 kind = job.import_type.upper() for staged in job.staged_rows: data = staged["normalized"] if kind == "EMPLOYEE": from apps.employees.models import Employee from apps.organization.models import Branch, Department, Designation, Shift branch = Branch.objects.filter(pk=data["branch_id"], company=job.company, is_active=True).first() department = Department.objects.filter(pk=data["department_id"], company=job.company, is_active=True).first() designation = Designation.objects.filter(pk=data["designation_id"], company=job.company, is_active=True).first() shift = None if data["shift_id"]: shift = Shift.objects.filter(pk=data["shift_id"], company=job.company, is_active=True).first() if not branch or not department or not designation or (data["shift_id"] and not shift): raise ValidationError("Employee master data changed or was deactivated after validation. Revalidate the import.") obj, was_created = Employee.objects.update_or_create( company=job.company, employee_code=data["employee_code"], defaults={ "full_name": data["full_name"], "mobile": data["mobile"], "email": data["email"], "branch": branch, "department": department, "designation": designation, "shift": shift, "employee_type": data["employee_type"], "joining_date": date.fromisoformat(data["joining_date"]), "status": data["status"], "blood_group": data["blood_group"], "nid_or_passport": data["nid_or_passport"], "present_address": data["present_address"], "emergency_contact": data["emergency_contact"], }, ) elif kind == "SALARY": from apps.employees.models import Employee from apps.payroll.models import EmployeeSalaryAssignment, SalaryStructure employee = Employee.objects.select_for_update().get(pk=data["employee_id"], company=job.company) structure = SalaryStructure.objects.select_for_update().filter( pk=data["structure_id"], company=job.company, is_active=True ).first() if not structure: raise ValidationError("Salary Template changed or is no longer active in the selected company. Revalidate the import.") effective_from = date.fromisoformat(data["effective_from"]) _validate_salary_import_window(employee, effective_from, lock=True) obj, was_created = EmployeeSalaryAssignment.objects.update_or_create( employee=employee, effective_from=effective_from, defaults={"structure": structure, "gross_salary": Decimal(data["gross_salary"]), "basic_salary": Decimal(data["basic_salary"]), "ot_eligible": data["ot_eligible"]}, ) elif kind == "LEAVE": from apps.employees.models import Employee from apps.leave.models import LeaveBalance, LeaveType employee = Employee.objects.select_for_update().get(pk=data["employee_id"], company=job.company) leave_type = LeaveType.objects.select_for_update().filter( pk=data["leave_type_id"], company=job.company, is_active=True ).first() if not leave_type: raise ValidationError("Leave Type changed or is no longer active in the selected company. Revalidate the import.") obj, was_created = LeaveBalance.objects.update_or_create( employee=employee, leave_type=leave_type, year=data["year"], defaults={"opening": Decimal(data["opening"])}, ) elif kind == "LOAN": from apps.employees.models import Employee from apps.loans.models import LoanAccount employee = Employee.objects.select_for_update().get(pk=data["employee_id"], company=job.company) principal = Decimal(data["sanctioned_amount"]) installment = Decimal(data["installment_amount"]) count = max(1, int((Decimal(data["outstanding_balance"]) / installment).to_integral_value(rounding=ROUND_CEILING))) if installment > 0 else 1 outstanding = Decimal(data["outstanding_balance"]) obj, was_created = LoanAccount.objects.update_or_create( employee=employee, request_reference=data["request_reference"], defaults={ "loan_type": data["loan_type"], "requested_amount": principal, "sanctioned_amount": principal, "installment_amount": installment, "installment_count": count, "start_month": date.fromisoformat(data["start_month"]), "disbursement_date": date.fromisoformat(data["disbursement_date"]), "outstanding_balance": outstanding, "status": LoanAccount.Status.ACTIVE if outstanding > 0 else LoanAccount.Status.CLOSED, "reason": data["reason"], }, ) # Opening loans represent the balance still recoverable at go-live. # Rebuild only future/unpaid installments from that outstanding amount; # historical installments remain in the legacy archive and are not # recreated as unpaid debt. from apps.loans.models import LoanInstallment from apps.loans.calculations import build_installment_amounts from apps.loans.services import _add_months, _month_start LoanInstallment.objects.filter(loan=obj, paid_amount=0).delete() if outstanding > 0: remaining_amounts = build_installment_amounts(outstanding, count) for index, due_amount in enumerate(remaining_amounts): LoanInstallment.objects.update_or_create( loan=obj, due_month=_add_months(_month_start(obj.start_month), index), defaults={"due_amount": due_amount, "paid_amount": Decimal("0.00"), "status": LoanInstallment.Status.DUE}, ) else: from apps.attendance.models import AttendancePeriod, AttendanceRecord from apps.attendance.services import employee_in_service_on, recalculate_attendance, reconcile_attendance_day, validate_attendance_values from apps.employees.models import Employee work_date = date.fromisoformat(data["work_date"]) employee = Employee.objects.select_for_update().select_related("shift", "company").get(pk=data["employee_id"], company=job.company) if not employee_in_service_on(employee, work_date): raise ValidationError("Attendance date is outside the employee service period.") period = AttendancePeriod.objects.select_for_update().filter( company=job.company, year=work_date.year, month=work_date.month ).first() if period and period.status == AttendancePeriod.Status.FINALIZED: raise ValidationError("Attendance period was finalized after validation. Reopen and revalidate before confirmation.") obj = AttendanceRecord.objects.select_for_update().filter(employee=employee, work_date=work_date).first() if obj and obj.finalized: raise ValidationError("Attendance record was finalized after validation. Reopen and revalidate before confirmation.") was_created = obj is None if obj is None: obj = AttendanceRecord(employee=employee, work_date=work_date) obj.check_in, obj.check_out = _attendance_datetimes(employee, work_date, data["check_in"], data["check_out"]) obj.status = data["status"] obj.check_in, obj.check_out = validate_attendance_values( employee, work_date, obj.check_in, obj.check_out, obj.status ) obj.remarks = data["remarks"] recalculate_attendance(obj) obj.save() reconcile_attendance_day(employee, work_date, force=True) created += int(was_created) updated += int(not was_created) job.status = "CONFIRMED" job.confirmed_at = timezone.now() job.result_summary = {"created": created, "updated": updated, "confirmed_at": job.confirmed_at.isoformat()} job.save(update_fields=["status", "confirmed_at", "result_summary", "updated_at"]) log_event(user=user, action="CONFIRM_IMPORT", module="imports", object_id=job.pk, object_repr=f"{job.import_type}/{job.company.code}", new_values=job.result_summary) return job