import csv import io from datetime import datetime from django.contrib import messages from django.contrib.auth.decorators import login_required from django.core.exceptions import PermissionDenied, ValidationError from django.core.paginator import Paginator from django.db.models import Q from django.shortcuts import get_object_or_404, redirect, render from django.utils import timezone from django.utils.dateparse import parse_datetime from apps.accounts.permissions import ( ROLE_HR_ADMIN, ROLE_MANAGER, ROLE_SUPER_ADMIN, accessible_company_ids, can_access_employee, has_role, require_roles, scope_queryset_by_company, ) from apps.audit.services import log_event from apps.employees.models import Employee from apps.organization.models import Company from .forms import ( AttendanceAdjustmentForm, AttendanceDeviceForm, AttendancePeriodForm, AttendancePeriodReopenForm, AttendanceRecordForm, DeviceEmployeeMapForm, PunchCSVImportForm, ) from .models import AttendanceDevice, AttendancePeriod, AttendanceRecord, DeviceEmployeeMap from .services import ( attendance_period_exceptions, create_adjustment_request, finalize_attendance_period, ingest_punch, reopen_attendance_period, save_attendance_record, ) def _company_qs(user): return Company.objects.filter(pk__in=accessible_company_ids(user), is_active=True) def _visible_employees(user): qs = Employee.objects.select_related("department", "designation", "branch", "shift") qs = scope_queryset_by_company(user, qs, "company") if user.is_superuser or has_role(user, ROLE_SUPER_ADMIN, ROLE_HR_ADMIN): return qs own = getattr(user, "employee_profile", None) if has_role(user, ROLE_MANAGER) and own: return qs.filter(Q(pk=own.pk) | Q(reporting_manager=own)) return qs.filter(pk=own.pk) if own else qs.none() @login_required def attendance_list(request): records = AttendanceRecord.objects.select_related("employee", "employee__department", "employee__shift").filter(employee__in=_visible_employees(request.user)) work_date = request.GET.get("date", "").strip() q = request.GET.get("q", "").strip() if work_date: records = records.filter(work_date=work_date) if q: records = records.filter(Q(employee__employee_code__icontains=q) | Q(employee__full_name__icontains=q)) page_obj = Paginator(records.order_by("-work_date", "employee__employee_code"), 100).get_page(request.GET.get("page")) return render(request, "attendance/attendance_list.html", {"records": page_obj.object_list, "page_obj": page_obj, "date_filter": work_date, "q": q}) @require_roles(ROLE_SUPER_ADMIN, ROLE_HR_ADMIN) def attendance_create(request): form = AttendanceRecordForm(request.POST or None, employee_queryset=_visible_employees(request.user)) if request.method == "POST" and form.is_valid(): try: save_attendance_record(form, user=request.user, request=request) messages.success(request, "Attendance saved.") return redirect("attendance:list") except ValidationError as exc: form.add_error(None, exc) return render(request, "attendance/attendance_form.html", {"form": form, "page_title": "Add Attendance"}) @require_roles(ROLE_SUPER_ADMIN, ROLE_HR_ADMIN) def attendance_edit(request, pk): record = get_object_or_404(AttendanceRecord.objects.select_related("employee").filter(employee__in=_visible_employees(request.user)), pk=pk) if record.finalized: messages.error(request, "Finalized attendance cannot be edited directly. Reopen the attendance period first.") return redirect("attendance:periods") form = AttendanceRecordForm(request.POST or None, instance=record, employee_queryset=_visible_employees(request.user)) if request.method == "POST" and form.is_valid(): try: save_attendance_record(form, user=request.user, request=request) messages.success(request, "Attendance updated.") return redirect("attendance:list") except ValidationError as exc: form.add_error(None, exc) return render(request, "attendance/attendance_form.html", {"form": form, "page_title": "Edit Attendance", "record": record}) @login_required def attendance_adjust(request, pk): record = get_object_or_404(AttendanceRecord.objects.select_related("employee", "employee__reporting_manager__user"), pk=pk) if not can_access_employee(request.user, record.employee): raise PermissionDenied("You do not have permission to request a correction for this attendance record.") initial = {"new_check_in": record.check_in, "new_check_out": record.check_out, "new_status": record.status} form = AttendanceAdjustmentForm(request.POST or None, initial=initial) if request.method == "POST" and form.is_valid(): try: create_adjustment_request(attendance=record, cleaned_data=form.cleaned_data, user=request.user, request=request) messages.success(request, "Attendance correction request submitted for approval.") return redirect("attendance:list") except ValidationError as exc: form.add_error(None, exc) return render(request, "attendance/attendance_adjustment_form.html", {"form": form, "record": record}) @require_roles(ROLE_SUPER_ADMIN, ROLE_HR_ADMIN) def device_list(request): devices = scope_queryset_by_company(request.user, AttendanceDevice.objects.select_related("company", "branch"), "company").order_by("branch__name", "name") return render(request, "attendance/device_list.html", {"devices": devices}) @require_roles(ROLE_SUPER_ADMIN, ROLE_HR_ADMIN) def device_create(request): form = AttendanceDeviceForm(request.POST or None, company_queryset=_company_qs(request.user)) if request.method == "POST" and form.is_valid(): device = form.save() messages.success(request, "Attendance device saved.") return redirect("attendance:device-detail", pk=device.pk) return render(request, "attendance/device_form.html", {"form": form, "page_title": "Add Attendance Device"}) @require_roles(ROLE_SUPER_ADMIN, ROLE_HR_ADMIN) def device_edit(request, pk): device = get_object_or_404(scope_queryset_by_company(request.user, AttendanceDevice.objects.select_related("company", "branch"), "company"), pk=pk) old_values = { "branch_id": device.branch_id, "name": device.name, "vendor": device.vendor, "model": device.model, "serial_number": device.serial_number, "connection_mode": device.connection_mode, "ip_address": str(device.ip_address or ""), "port": device.port, "adms_url": device.adms_url, "is_active": device.is_active, } old_comm_key = device.comm_key form = AttendanceDeviceForm(request.POST or None, instance=device, company_queryset=_company_qs(request.user)) if request.method == "POST" and form.is_valid(): device = form.save() new_values = { "branch_id": device.branch_id, "name": device.name, "vendor": device.vendor, "model": device.model, "serial_number": device.serial_number, "connection_mode": device.connection_mode, "ip_address": str(device.ip_address or ""), "port": device.port, "adms_url": device.adms_url, "is_active": device.is_active, "comm_key_changed": old_comm_key != device.comm_key, } log_event( user=request.user, company=device.company, action="UPDATE", module="attendance_device", object_id=device.pk, object_repr=str(device), old_values=old_values, new_values=new_values, request=request, ) messages.success(request, "Attendance device updated.") return redirect("attendance:device-detail", pk=device.pk) return render(request, "attendance/device_form.html", {"form": form, "page_title": "Edit Attendance Device"}) @require_roles(ROLE_SUPER_ADMIN, ROLE_HR_ADMIN) def device_detail(request, pk): device = get_object_or_404(scope_queryset_by_company(request.user, AttendanceDevice.objects.select_related("company", "branch"), "company"), pk=pk) mappings = device.employee_maps.select_related("employee").order_by("device_user_id") punches = device.punches.select_related("employee").order_by("-punch_time")[:100] return render(request, "attendance/device_detail.html", {"device": device, "mappings": mappings, "punches": punches, "mapping_form": DeviceEmployeeMapForm(device=device), "upload_form": PunchCSVImportForm()}) @require_roles(ROLE_SUPER_ADMIN, ROLE_HR_ADMIN) def device_map_add(request, pk): device = get_object_or_404(scope_queryset_by_company(request.user, AttendanceDevice.objects.all(), "company"), pk=pk) if request.method != "POST": return redirect("attendance:device-detail", pk=device.pk) form = DeviceEmployeeMapForm(request.POST, device=device) if form.is_valid(): mapping = form.save(commit=False) mapping.device = device mapping.save() messages.success(request, "Employee-to-machine mapping saved.") else: messages.error(request, "Could not save mapping: " + "; ".join([str(e) for errors in form.errors.values() for e in errors])) return redirect("attendance:device-detail", pk=device.pk) def _parse_punch_time(value): parsed = parse_datetime((value or "").strip()) if not parsed: for fmt in ("%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M:%S", "%d/%m/%Y %H:%M:%S", "%Y-%m-%d %H:%M"): try: parsed = datetime.strptime((value or "").strip(), fmt) break except ValueError: pass if parsed and timezone.is_naive(parsed): parsed = timezone.make_aware(parsed, timezone.get_current_timezone()) return parsed @require_roles(ROLE_SUPER_ADMIN, ROLE_HR_ADMIN) def device_import_csv(request, pk): device = get_object_or_404(scope_queryset_by_company(request.user, AttendanceDevice.objects.all(), "company"), pk=pk) if request.method != "POST": return redirect("attendance:device-detail", pk=device.pk) form = PunchCSVImportForm(request.POST, request.FILES) if not form.is_valid(): messages.error(request, "Choose a valid CSV file.") return redirect("attendance:device-detail", pk=device.pk) upload = form.cleaned_data["csv_file"] try: text = upload.read().decode("utf-8-sig") reader = csv.DictReader(io.StringIO(text)) imported = duplicate = invalid = 0 for row in reader: uid = (row.get("device_user_id") or row.get("user_id") or row.get("machine_id") or "").strip() stamp = row.get("punch_time") or row.get("timestamp") or row.get("date_time") or "" ptime = _parse_punch_time(stamp) if not uid or not ptime: invalid += 1 continue _, created = ingest_punch(device=device, device_user_id=uid, punch_time=ptime, punch_type=(row.get("punch_type") or row.get("type") or "").strip(), raw_payload=row, user=request.user) imported += int(created) duplicate += int(not created) device.last_sync_at = timezone.now() device.last_sync_status = f"CSV: {imported} imported, {duplicate} duplicate, {invalid} invalid" device.save(update_fields=["last_sync_at", "last_sync_status", "updated_at"]) messages.success(request, device.last_sync_status + ". Daily attendance roll-up was refreshed for mapped employees.") except Exception as exc: device.last_sync_at = timezone.now() device.last_sync_status = "CSV import failed" device.save(update_fields=["last_sync_at", "last_sync_status", "updated_at"]) messages.error(request, f"CSV import failed: {exc}") return redirect("attendance:device-detail", pk=device.pk) @require_roles(ROLE_SUPER_ADMIN, ROLE_HR_ADMIN) def period_control(request): periods = scope_queryset_by_company(request.user, AttendancePeriod.objects.select_related("company", "finalized_by", "reopened_by"), "company").order_by("-year", "-month", "company__name")[:120] today = timezone.localdate() form = AttendancePeriodForm(company_queryset=_company_qs(request.user), initial={"year": today.year, "month": today.month}) preview_exceptions = [] preview_company = _company_qs(request.user).first() if preview_company: try: preview_exceptions = attendance_period_exceptions(preview_company, today.year, today.month)[:100] except ValidationError: preview_exceptions = [] return render(request, "attendance/period_control.html", {"periods": periods, "form": form, "preview_exceptions": preview_exceptions, "preview_company": preview_company, "preview_year": today.year, "preview_month": today.month}) @require_roles(ROLE_SUPER_ADMIN, ROLE_HR_ADMIN) def period_finalize(request): if request.method != "POST": return redirect("attendance:periods") form = AttendancePeriodForm(request.POST, company_queryset=_company_qs(request.user)) if form.is_valid(): try: period = finalize_attendance_period(form.cleaned_data["company"], form.cleaned_data["year"], form.cleaned_data["month"], user=request.user) messages.success(request, f"Attendance {period.year}-{period.month:02d} finalized. Payroll may now consume this period.") except ValidationError as exc: messages.error(request, str(exc)) else: messages.error(request, "Choose a valid company and month.") return redirect("attendance:periods") @require_roles(ROLE_SUPER_ADMIN, ROLE_HR_ADMIN) def period_reopen(request, pk): period = get_object_or_404(scope_queryset_by_company(request.user, AttendancePeriod.objects.all(), "company"), pk=pk) if request.method != "POST": return redirect("attendance:periods") form = AttendancePeriodReopenForm(request.POST) if form.is_valid(): try: reopen_attendance_period(period, user=request.user, reason=form.cleaned_data["reason"]) messages.success(request, "Attendance period reopened with audit reason. Re-finalize it before payroll.") except ValidationError as exc: messages.error(request, str(exc)) else: messages.error(request, "Reopen reason is required.") return redirect("attendance:periods")