from datetime import date as date_cls, datetime from typing import Optional from django.core.files.base import ContentFile from django.db import transaction from django.db.models import Count, Exists, OuterRef from django.shortcuts import get_object_or_404 from rest_framework import viewsets, status from rest_framework.decorators import action, api_view, permission_classes from rest_framework.parsers import MultiPartParser, FormParser, JSONParser from rest_framework.permissions import IsAuthenticated, AllowAny from apps.core.permissions import HasMenuPermission from rest_framework.response import Response from botocore.exceptions import ClientError from drf_spectacular.utils import extend_schema, OpenApiParameter from .models import ( Employee, Attendance, Payroll, Recruitment, JobApplication, Onboarding, Contract, Communication, Survey, SurveyResponse, Family, AttendanceSettings, ShiftTemplate, PayrollSettings, DEFAULT_ATTENDANCE_SETTINGS, ) from .serializers import ( EmployeeSerializer, AttendanceSerializer, PayrollSerializer, RecruitmentSerializer, JobApplicationSerializer, PublicRecruitmentSerializer, PublicApplicationSerializer, OnboardingSerializer, ContractSerializer, CommunicationSerializer, SurveySerializer, SurveyResponseSerializer, FamilySerializer, AttendanceSettingsSerializer, ShiftTemplateSerializer, PayrollSettingsSerializer, ) from .services.rekognition import ensure_collection, index_face, search_face from .services.geofence import haversine_meters, to_float from apps.companies.models import OrgProfile class EmployeeViewSet(viewsets.ModelViewSet): """ViewSet for Employee management.""" rbac_domain = "hr" queryset = Employee.objects.select_related("user__contact_profile").prefetch_related("contracts") serializer_class = EmployeeSerializer def get_queryset(self): qs = super().get_queryset() et = self.request.query_params.get("employment_type") if et: qs = qs.filter(employment_type=et) dep = self.request.query_params.get("department") if dep: qs = qs.filter(department=dep) st = self.request.query_params.get("status") if st: qs = qs.filter(status=st) return qs @extend_schema( parameters=[ OpenApiParameter(name="employment_type", type=str, description="staff | fellow | intern"), OpenApiParameter(name="department", type=str, description="Filter by department"), OpenApiParameter(name="status", type=str, description="Filter by status"), ] ) def list(self, request, *args, **kwargs): return super().list(request, *args, **kwargs) @action(detail=False, methods=["get"], url_path="dashboard-stats") def dashboard_stats(self, request): """Aggregated HR metrics for the dashboard view.""" from collections import Counter, defaultdict from datetime import timedelta from django.db.models import Count from django.utils import timezone today = timezone.localdate() month_start = today.replace(day=1) six_months_ago = (month_start - timedelta(days=180)).replace(day=1) thirty_days_ago = today - timedelta(days=30) ninety_days_ago = today - timedelta(days=90) emp_qs = Employee.objects.all() total = emp_qs.count() new_hires_this_month = emp_qs.filter(hire_date__gte=month_start, hire_date__lte=today).count() terminated_this_month = emp_qs.filter(termination_date__gte=month_start, termination_date__lte=today).count() turnover = (terminated_this_month / total * 100) if total else 0.0 by_status = list(emp_qs.values("status").annotate(count=Count("id")).order_by("-count")) by_employment_type = list(emp_qs.values("employment_type").annotate(count=Count("id")).order_by("-count")) by_department = list( emp_qs.exclude(department="").values("department").annotate(count=Count("id")).order_by("-count")[:10] ) open_positions = Recruitment.objects.filter(status="open").count() recruitment_by_status = list( Recruitment.objects.values("status").annotate(count=Count("id")).order_by("-count") ) # Daily attendance — last 30 days, grouped by date with type counts. att_rows = ( Attendance.objects.filter(date__gte=thirty_days_ago, date__lte=today) .values("date", "type") .annotate(count=Count("id")) .order_by("date") ) daily_buckets: dict[str, dict] = {} for r in att_rows: key = r["date"].isoformat() b = daily_buckets.setdefault(key, {"date": key, "attend": 0, "leave": 0, "absent": 0}) t = r["type"] if t == "check_in": b["attend"] += r["count"] elif t == "leave": b["leave"] += r["count"] elif t == "absent": b["absent"] += r["count"] daily_attendance = sorted(daily_buckets.values(), key=lambda x: x["date"]) # Monthly attendance rate — last 6 months. monthly_att_rows = ( Attendance.objects.filter(date__gte=six_months_ago, date__lte=today) .values("date", "type") ) monthly: dict[str, dict] = {} for r in monthly_att_rows: key = r["date"].strftime("%Y-%m") b = monthly.setdefault(key, {"month": key, "present": 0, "absent": 0}) if r["type"] == "check_in": b["present"] += 1 elif r["type"] in ("absent", "leave"): b["absent"] += 1 monthly_list = [] for m in sorted(monthly): row = monthly[m] total_evt = row["present"] + row["absent"] rate = (row["present"] / total_evt * 100) if total_evt else 0.0 monthly_list.append({"month": m, "rate": round(rate, 1), "absences": row["absent"]}) monthly_list = monthly_list[-6:] # Recent hires. recent_qs = ( emp_qs.filter(hire_date__gte=ninety_days_ago) .select_related("user") .order_by("-hire_date")[:10] ) recent_hires = [] for e in recent_qs: full_name = (e.user.get_full_name() if e.user else "") or (e.user.email if e.user else "Unknown") recent_hires.append({ "id": str(e.id), "name": full_name, "department": e.department or "", "position": e.position or "", "hire_date": e.hire_date.isoformat() if e.hire_date else None, "status": e.status, }) return Response({ "totals": { "total_employees": total, "new_hires_this_month": new_hires_this_month, "open_positions": open_positions, "turnover_rate": round(turnover, 2), }, "by_status": by_status, "by_employment_type": by_employment_type, "by_department": by_department, "recruitment_pipeline": recruitment_by_status, "daily_attendance": daily_attendance, "monthly_attendance": monthly_list, "recent_hires": recent_hires, }) @action( detail=False, methods=["post"], url_path="enroll-face", parser_classes=[MultiPartParser, FormParser], permission_classes=[IsAuthenticated, HasMenuPermission], ) def enroll_face(self, request): """Index the authenticated user's face into the Rekognition collection. Multipart: photo (file). ExternalImageId = str(user.id). """ photo = request.FILES.get("photo") if not photo: return Response({"error": "photo file is required"}, status=status.HTTP_400_BAD_REQUEST) from apps.core.uploads import validate_upload validate_upload(photo) image_bytes = photo.read() if not image_bytes: return Response({"error": "empty photo"}, status=status.HTTP_400_BAD_REQUEST) ensure_collection() try: result = index_face(image_bytes, external_image_id=str(request.user.id)) except ClientError as exc: return Response( {"error": exc.response.get("Error", {}).get("Message", "Rekognition failed")}, status=status.HTTP_400_BAD_REQUEST, ) faces = result.get("FaceRecords") or [] if not faces: return Response({"error": "No face detected"}, status=status.HTTP_400_BAD_REQUEST) face_id = faces[0]["Face"]["FaceId"] return Response({"face_id": face_id, "external_image_id": str(request.user.id)}) class AttendanceViewSet(viewsets.ModelViewSet): """ViewSet for Attendance management.""" rbac_domain = "hr" queryset = Attendance.objects.select_related("employee__user") serializer_class = AttendanceSerializer def get_queryset(self): from datetime import date, timedelta qs = super().get_queryset() p = self.request.query_params if p.get("mine") in ("1", "true", "True"): qs = qs.filter(employee__user=self.request.user) if p.get("employee"): qs = qs.filter(employee=p["employee"]) # Date range — default to last 30 days when no range supplied. date_from = p.get("date_from") date_to = p.get("date_to") if date_from: qs = qs.filter(date__gte=date_from) elif not p.get("all"): qs = qs.filter(date__gte=date.today() - timedelta(days=30)) if date_to: qs = qs.filter(date__lte=date_to) return qs @extend_schema( parameters=[ OpenApiParameter(name="employee", type=str, description="Filter by employee ID"), OpenApiParameter(name="date_from", type=str, description="Start date (YYYY-MM-DD), default 30 days ago"), OpenApiParameter(name="date_to", type=str, description="End date (YYYY-MM-DD)"), OpenApiParameter(name="all", type=str, description="Pass all=1 to skip date filter"), ] ) def list(self, request, *args, **kwargs): return super().list(request, *args, **kwargs) @action( detail=False, methods=["post"], url_path="face-check", parser_classes=[MultiPartParser, FormParser, JSONParser], permission_classes=[IsAuthenticated, HasMenuPermission], ) def face_check(self, request): """Mobile clock-in/out with Rekognition verification. Multipart fields: photo (file, required) mode (str, "in" | "out", required) latitude (float, optional) longitude (float, optional) address (str, optional) timestamp (int millis, optional) """ photo = request.FILES.get("photo") mode = (request.data.get("mode") or "").lower() if not photo: return Response({"error": "photo file is required"}, status=status.HTTP_400_BAD_REQUEST) if mode not in {"in", "out"}: return Response({"error": "mode must be 'in' or 'out'"}, status=status.HTTP_400_BAD_REQUEST) from apps.core.uploads import validate_upload validate_upload(photo) employee = Employee.objects.filter(user=request.user).first() if not employee: return Response( {"error": "Authenticated user has no employee record"}, status=status.HTTP_400_BAD_REQUEST, ) image_bytes = photo.read() if not image_bytes: return Response({"error": "empty photo"}, status=status.HTTP_400_BAD_REQUEST) try: match = search_face(image_bytes) except ClientError as exc: return Response( {"error": exc.response.get("Error", {}).get("Message", "Rekognition failed")}, status=status.HTTP_400_BAD_REQUEST, ) if not match: return Response( {"error": "Face not recognized. Enroll your face first."}, status=status.HTTP_403_FORBIDDEN, ) matched_external_id = match["Face"]["ExternalImageId"] if matched_external_id != str(request.user.id): return Response( {"error": "Face does not match authenticated user"}, status=status.HTTP_403_FORBIDDEN, ) # Build location string (fits in CharField(100)) lat = to_float(request.data.get("latitude")) lng = to_float(request.data.get("longitude")) accuracy = to_float(request.data.get("accuracy")) address = (request.data.get("address") or "").strip() # Enforce attendance policy: GPS accuracy + geofence. cfg = {**DEFAULT_ATTENDANCE_SETTINGS, **(AttendanceSettings.get_solo().settings or {})} gps_cfg = cfg.get("gps") or {} geo_cfg = cfg.get("geofencing") or {} outside_distance: Optional[float] = None if gps_cfg.get("enabled"): if lat is None or lng is None: return Response( {"error": "GPS coordinates required"}, status=status.HTTP_400_BAD_REQUEST, ) max_accuracy = gps_cfg.get("accuracy_meters") if max_accuracy and accuracy is not None and accuracy > float(max_accuracy): return Response( { "error": f"GPS accuracy {int(accuracy)}m exceeds limit " f"{int(max_accuracy)}m. Move outdoors and retry.", }, status=status.HTTP_400_BAD_REQUEST, ) if geo_cfg.get("enabled") and lat is not None and lng is not None: org = OrgProfile.objects.first() org_lat = float(org.latitude) if org and org.latitude is not None else None org_lng = float(org.longitude) if org and org.longitude is not None else None radius = float(geo_cfg.get("radius_meters") or 0) if org_lat is None or org_lng is None: if geo_cfg.get("enforce"): return Response( {"error": "Office location not set. Configure it in Company settings."}, status=status.HTTP_400_BAD_REQUEST, ) elif radius > 0: distance = haversine_meters(org_lat, org_lng, lat, lng) if distance > radius: if geo_cfg.get("enforce"): return Response( { "error": f"Outside office geofence by {int(distance - radius)}m " f"(allowed {int(radius)}m radius).", }, status=status.HTTP_403_FORBIDDEN, ) outside_distance = distance loc_parts = [] if address: loc_parts.append(address) if lat is not None and lng is not None: loc_parts.append(f"({lat},{lng})") location_str = " ".join(loc_parts)[:100] # Timestamp from client, fallback to now ts_raw = request.data.get("timestamp") if ts_raw: try: ts = datetime.fromtimestamp(int(ts_raw) / 1000) except (TypeError, ValueError): ts = datetime.now() else: ts = datetime.now() similarity = match.get("Similarity") att_type = "check_in" if mode == "in" else "check_out" defaults = { "time": ts.time().replace(microsecond=0), "location": location_str, "face_match_score": round(similarity, 2) if similarity is not None else None, } if outside_distance is not None: defaults["notes"] = f"Outside geofence by {int(outside_distance)}m" record, created = Attendance.objects.update_or_create( employee=employee, date=ts.date(), type=att_type, defaults=defaults, ) record.photo.save( f"{record.id}-{ts.strftime('%H%M%S')}.jpg", ContentFile(image_bytes), save=True, ) return Response( { "attendance": AttendanceSerializer(record, context={"request": request}).data, "created": created, "similarity": similarity, }, status=status.HTTP_201_CREATED if created else status.HTTP_200_OK, ) @action( detail=False, methods=["get"], url_path="face-status", permission_classes=[IsAuthenticated, HasMenuPermission], ) def face_status(self, request): """Whether the authenticated user has an enrolled face in the collection.""" from .services.rekognition import get_client from django.conf import settings as dj_settings ensure_collection() try: resp = get_client().list_faces( CollectionId=dj_settings.REKOGNITION_COLLECTION_ID, MaxResults=4096, ) except ClientError as exc: return Response( {"error": exc.response.get("Error", {}).get("Message", "Rekognition failed")}, status=status.HTTP_400_BAD_REQUEST, ) target = str(request.user.id) enrolled = any(f.get("ExternalImageId") == target for f in resp.get("Faces", [])) return Response({"enrolled": enrolled}) class PayrollViewSet(viewsets.ModelViewSet): """ViewSet for Payroll management.""" rbac_domain = "hr" queryset = Payroll.objects.select_related("employee__user", "approved_by") serializer_class = PayrollSerializer def get_queryset(self): qs = super().get_queryset() p = self.request.query_params if p.get("employee"): qs = qs.filter(employee=p["employee"]) if p.get("status"): qs = qs.filter(status=p["status"]) if p.get("period_start"): qs = qs.filter(period_start__gte=p["period_start"]) if p.get("period_end"): qs = qs.filter(period_start__lt=p["period_end"]) return qs.order_by("-period_start") def perform_update(self, serializer): # Stamp paid_at when a record first transitions to "paid"; clear it if moved back. from django.utils import timezone instance = serializer.instance new_status = serializer.validated_data.get("status", instance.status) if new_status == "paid" and instance.paid_at is None: serializer.save(paid_at=timezone.now()) elif new_status != "paid" and instance.paid_at is not None: serializer.save(paid_at=None) else: serializer.save() @action(detail=False, methods=["post"], url_path="run") def run_payroll(self, request): """Bulk-generate draft payroll for all active employees for a given period. Pass regenerate=true to delete existing draft records for the period first. Records already approved/paid are never deleted. """ from datetime import date as date_cls period_start = request.data.get("period_start") period_end = request.data.get("period_end") regenerate = request.data.get("regenerate") in (True, "true", "1", 1) if not period_start or not period_end: return Response({"error": "period_start and period_end required"}, status=status.HTTP_400_BAD_REQUEST) if regenerate: Payroll.objects.filter( period_start=period_start, period_end=period_end, status="draft", ).delete() from datetime import date as date_cls, timedelta from django.db.models import Count from django.utils import timezone import hashlib, hmac from django.conf import settings as dj_settings cfg = PayrollSettings.get_solo().settings or {} comp = cfg.get("components", {}) allowance_defs = comp.get("allowances", []) deduction_defs = comp.get("deductions", []) officer = cfg.get("officer", {}) or {} officer_name = officer.get("name") or (request.user.get_full_name() or request.user.username) officer_title = officer.get("title") or "Payroll Officer" run_ts = timezone.now().isoformat() # Count working days in period (Mon–Fri). p_start = date_cls.fromisoformat(period_start) p_end = date_cls.fromisoformat(period_end) working_days = sum( 1 for i in range((p_end - p_start).days + 1) if (p_start + timedelta(days=i)).weekday() < 5 ) or 1 # Fetch absent-day counts per employee for this period. absent_counts = { row["employee_id"]: row["cnt"] for row in Attendance.objects.filter( date__gte=p_start, date__lte=p_end, type="absent" ).values("employee_id").annotate(cnt=Count("id")) } # Fetch late-day counts (check_in after 09:00). late_counts = { row["employee_id"]: row["cnt"] for row in Attendance.objects.filter( date__gte=p_start, date__lte=p_end, type="check_in", time__gt="09:00:00", ).values("employee_id").annotate(cnt=Count("id")) } employees = Employee.objects.filter(status="active").select_related("user") created, skipped = [], [] for emp in employees: exists = Payroll.objects.filter(employee=emp, period_start=period_start, period_end=period_end).exists() if exists: skipped.append(str(emp.id)) continue base = float(emp.salary or 0) daily = base / working_days # Per-employee overrides (fall back to org defaults). ov = emp.payroll_overrides or {} present_days = working_days - absent_counts.get(emp.id, 0) allowance_overrides = ov.get("allowance_overrides", {}) or {} transport_mode = ov.get("transport_mode") # "fixed" | "daily" | None transport_amount = float(ov.get("transport_amount", 0) or 0) # Build earnings line items. earnings = [{"label": "Basic Salary", "qty": 1, "amount": round(base, 2)}] allowances = 0.0 for a in allowance_defs: key = a.get("key", "") label = a.get("label", "Allowance") # Transport may be daily-rated per employee. if key == "transport" and transport_mode: per = transport_amount if transport_amount else float(a.get("default_amount", 0)) if transport_mode == "daily": amt = round(per * max(present_days, 0), 2) if amt: earnings.append({"label": f"{label} ({present_days}× daily)", "qty": present_days, "amount": amt}) allowances += amt continue else: # fixed override amount if per: earnings.append({"label": label, "qty": 1, "amount": round(per, 2)}) allowances += per continue # Generic per-employee amount override, else org default. amt = float(allowance_overrides.get(key, a.get("default_amount", 0))) if amt: earnings.append({"label": label, "qty": 1, "amount": round(amt, 2)}) allowances += amt allowances = round(allowances, 2) # Build deduction line items. deduction_items = [] for d in deduction_defs: rate = float(d.get("rate", 0)) amt = round(base * rate / 100, 2) if amt: deduction_items.append({"label": f"{d.get('label', 'Deduction')} ({rate}%)", "qty": 1, "amount": amt}) absent_days = absent_counts.get(emp.id, 0) late_days = late_counts.get(emp.id, 0) absent_deduction = round(daily * absent_days, 2) late_deduction = round(daily * 0.1 * late_days, 2) if absent_deduction: deduction_items.append({"label": f"Absence ({absent_days} day{'s' if absent_days != 1 else ''})", "qty": absent_days, "amount": absent_deduction}) if late_deduction: deduction_items.append({"label": f"Late arrival ({late_days}×)", "qty": late_days, "amount": late_deduction}) deductions = round(sum(d["amount"] for d in deduction_items), 2) gross = round(base + allowances, 2) net = round(gross - deductions, 2) net_final = max(net, 0) # Sign the payslip: HMAC over identifying fields. Encoded in the QR so # anyone can verify the officer authorised this exact payslip. payload = f"{emp.employee_id}|{period_start}|{period_end}|{net_final}|{officer_name}|{run_ts}" signature = hmac.new( dj_settings.SECRET_KEY.encode(), payload.encode(), hashlib.sha256, ).hexdigest()[:16] breakdown = { "earnings": earnings, "deductions": deduction_items, "meta": { "working_days": working_days, "daily_rate": round(daily, 2), "absent_days": absent_days, "late_days": late_days, "gross": gross, }, "authorization": { "officer_name": officer_name, "officer_title": officer_title, "signed_at": run_ts, "signature": signature, # Self-contained verification string for the QR. "qr_payload": f"PAYSLIP-AUTH|{emp.employee_id}|{period_start}|net={net_final}|by={officer_name}|sig={signature}", }, } payroll = Payroll.objects.create( employee=emp, period_start=period_start, period_end=period_end, base_salary=base, allowances=round(allowances, 2), deductions=deductions, net_salary=net_final, breakdown=breakdown, status="draft", ) created.append(str(payroll.id)) return Response({"created": len(created), "skipped": len(skipped)}, status=status.HTTP_201_CREATED) class RecruitmentViewSet(viewsets.ModelViewSet): """ViewSet for Recruitment management.""" rbac_domain = "hr" queryset = Recruitment.objects.select_related("hiring_manager").prefetch_related("applications") serializer_class = RecruitmentSerializer @extend_schema( parameters=[ OpenApiParameter(name="status", type=str, description="Filter by status"), OpenApiParameter(name="department", type=str, description="Filter by department"), ] ) def list(self, request, *args, **kwargs): return super().list(request, *args, **kwargs) class JobApplicationViewSet(viewsets.ModelViewSet): """Admin management of applications submitted to recruitment postings.""" rbac_domain = "hr" queryset = JobApplication.objects.select_related("recruitment") serializer_class = JobApplicationSerializer parser_classes = [MultiPartParser, FormParser, JSONParser] def get_queryset(self): qs = super().get_queryset() p = self.request.query_params if p.get("recruitment"): qs = qs.filter(recruitment=p["recruitment"]) if p.get("stage"): qs = qs.filter(stage=p["stage"]) return qs @action(detail=True, methods=["post"], url_path="convert-to-employee") def convert_to_employee(self, request, pk=None): """Create an Employee (and user account) from a hired application.""" from django.contrib.auth import get_user_model import secrets from .serializers import EmployeeSerializer app = self.get_object() if app.stage != "hired": return Response({"detail": "Only hired applications can be converted."}, status=status.HTTP_400_BAD_REQUEST) if app.converted_employee_id: return Response({"detail": "This application was already converted to an employee."}, status=status.HTTP_400_BAD_REQUEST) User = get_user_model() if User.objects.filter(email__iexact=app.email).exists(): return Response({"detail": "A user with this email already exists."}, status=status.HTTP_400_BAD_REQUEST) # full_time/part_time/contract -> staff; internship -> intern. emp_type = "intern" if app.recruitment.employment_type == "internship" else "staff" with transaction.atomic(): first, _, last = app.full_name.partition(" ") temp_password = secrets.token_urlsafe(9) user = User( email=app.email.lower(), first_name=first, last_name=last, phone=app.phone or "", must_change_password=True, onboarding_completed=False, ) user.set_password(temp_password) user.save() # A post_save signal auto-creates an Employee stub for every new user. # Update that stub rather than creating a second (would violate the OneToOne). employee = Employee.objects.get(user=user) employee.employment_type = emp_type employee.position = app.recruitment.position employee.department = app.recruitment.department employee.status = "active" employee.hire_date = date_cls.today() employee.save() app.converted_employee = employee app.save(update_fields=["converted_employee", "updated_at"]) data = EmployeeSerializer(employee, context={"request": request}).data data["temp_password"] = temp_password return Response(data, status=status.HTTP_201_CREATED) class PublicRecruitmentViewSet(viewsets.ReadOnlyModelViewSet): """Unauthenticated careers pages: list/retrieve OPEN postings + submit application.""" permission_classes = [AllowAny] authentication_classes = [] serializer_class = PublicRecruitmentSerializer queryset = Recruitment.objects.filter(status="open") @action( detail=True, methods=["post"], url_path="apply", permission_classes=[AllowAny], authentication_classes=[], parser_classes=[MultiPartParser, FormParser, JSONParser], ) def apply(self, request, pk=None): posting = self.get_object() # 404 unless open if posting.deadline and posting.deadline < date_cls.today(): return Response({"detail": "Applications for this position have closed."}, status=status.HTTP_400_BAD_REQUEST) serializer = PublicApplicationSerializer(data=request.data) serializer.is_valid(raise_exception=True) serializer.save(recruitment=posting, stage="applied") return Response(serializer.data, status=status.HTTP_201_CREATED) class OnboardingViewSet(viewsets.ModelViewSet): """ViewSet for Onboarding management.""" rbac_domain = "hr" queryset = Onboarding.objects.all() serializer_class = OnboardingSerializer class ContractViewSet(viewsets.ModelViewSet): """ViewSet for Contract management.""" rbac_domain = "hr" queryset = Contract.objects.select_related("employee__user") serializer_class = ContractSerializer def get_queryset(self): from django.db.models import Case, When, Value, IntegerField, F qs = super().get_queryset() p = self.request.query_params if p.get("employee"): qs = qs.filter(employee=p["employee"]) if p.get("status"): qs = qs.filter(status=p["status"]) # Soonest-to-expire first: active contracts with an end date on top # (nearest end_date first), then active-without-end, then everything else. return qs.annotate( _bucket=Case( When(status="active", end_date__isnull=False, then=Value(0)), When(status="active", end_date__isnull=True, then=Value(1)), default=Value(2), output_field=IntegerField(), ) ).order_by("_bucket", F("end_date").asc(nulls_last=True), "-start_date") @extend_schema( parameters=[ OpenApiParameter(name="employee", type=str, description="Filter by employee ID"), OpenApiParameter(name="status", type=str, description="Filter by status"), ] ) def list(self, request, *args, **kwargs): return super().list(request, *args, **kwargs) class CommunicationViewSet(viewsets.ModelViewSet): """ViewSet for HR Communication management.""" rbac_domain = "hr" queryset = Communication.objects.select_related("author") serializer_class = CommunicationSerializer parser_classes = [MultiPartParser, FormParser, JSONParser] def perform_create(self, serializer): comm = serializer.save(author=self.request.user) self._notify_employees(comm) def _notify_employees(self, comm): """Broadcast the new communication to every active employee's user account.""" from apps.notifications.handler import NotificationHandler from django.contrib.auth import get_user_model user_ids = ( Employee.objects.filter(status="active") .exclude(user__isnull=True) .values_list("user_id", flat=True) .distinct() ) User = get_user_model() users = list(User.objects.filter(id__in=list(user_ids))) if not users: return NotificationHandler.broadcast( users=users, title=comm.title, message=f"New {comm.get_communication_type_display().lower()} from HR.", notification_type="hr", icon="megaphone", link=f"/hr/communication/{comm.id}", metadata={"communication_id": str(comm.id)}, ) @action(detail=False, methods=["post"], url_path="upload-attachment", parser_classes=[MultiPartParser, FormParser]) def upload_attachment(self, request): """Store one uploaded document and return {name, url} for the attachments list.""" from django.core.files.storage import default_storage f = request.FILES.get("file") if not f: return Response({"detail": "No file provided."}, status=status.HTTP_400_BAD_REQUEST) path = default_storage.save(f"communications/{f.name}", f) url = default_storage.url(path) return Response({"name": f.name, "url": url}, status=status.HTTP_201_CREATED) class SurveyViewSet(viewsets.ModelViewSet): """ViewSet for Survey management.""" rbac_domain = "hr" # Any authenticated employee may read-for-fill and submit a response. rbac_public_actions = ("fill", "respond") queryset = Survey.objects.select_related("created_by") serializer_class = SurveySerializer def get_queryset(self): # Annotate counts/flags instead of prefetching response_set — that # prefetch dragged every response (answers JSON included) into memory # just to count them. qs = super().get_queryset().annotate( response_count_anno=Count("response_set", distinct=True) ) user = self.request.user if user.is_authenticated: qs = qs.annotate(has_responded_anno=Exists( SurveyResponse.objects.filter(survey=OuterRef("pk"), respondent=user) )) return qs def get_serializer_context(self): ctx = super().get_serializer_context() ctx["request"] = self.request return ctx def perform_create(self, serializer): serializer.save(created_by=self.request.user) @extend_schema( parameters=[ OpenApiParameter(name="status", type=str, description="Filter by status"), ] ) def list(self, request, *args, **kwargs): return super().list(request, *args, **kwargs) @action(detail=True, methods=["get"], url_path="fill", permission_classes=[IsAuthenticated]) def fill(self, request, pk=None): """Read a survey for responding. Open to any authenticated employee (not HR-gated).""" survey = get_object_or_404(Survey, pk=pk) return Response(SurveySerializer(survey, context={"request": request}).data) @action(detail=True, methods=["post"], url_path="respond", permission_classes=[IsAuthenticated]) def respond(self, request, pk=None): """Submit the current user's response. Any authenticated employee may fill an open survey; once per user.""" # Bypass the HR-menu rbac gate: fetch directly so non-HR staff can respond. survey = get_object_or_404(Survey, pk=pk) if not survey.is_open(): return Response({"detail": "This survey is not open for responses."}, status=status.HTTP_400_BAD_REQUEST) if SurveyResponse.objects.filter(survey=survey, respondent=request.user).exists(): return Response({"detail": "You have already responded to this survey."}, status=status.HTTP_400_BAD_REQUEST) answers = request.data.get("answers", []) if not isinstance(answers, list): return Response({"detail": "answers must be a list."}, status=status.HTTP_400_BAD_REQUEST) # Enforce required questions. answered = {a.get("question_id") for a in answers if a.get("value") not in (None, "", [])} missing = [q.get("label", q.get("id")) for q in survey.questions if q.get("required") and q.get("id") not in answered] if missing: return Response({"detail": f"Missing required answers: {', '.join(map(str, missing))}"}, status=status.HTTP_400_BAD_REQUEST) resp = SurveyResponse.objects.create(survey=survey, respondent=request.user, answers=answers) return Response(SurveyResponseSerializer(resp).data, status=status.HTTP_201_CREATED) @action(detail=True, methods=["get"], url_path="results") def results(self, request, pk=None): """Per-question aggregated results for the survey owner / HR.""" survey = self.get_object() # Single streaming pass over responses (avoid materializing all rows and # the O(questions x responses) re-scan): bucket each answer by question id. qids = {q.get("id") for q in survey.questions} values_by_qid = {qid: [] for qid in qids} total = 0 for r in survey.response_set.values_list("answers", flat=True).iterator(): total += 1 for a in r or []: qid = a.get("question_id") if qid in values_by_qid and a.get("value") not in (None, "", []): values_by_qid[qid].append(a["value"]) out = [] for q in survey.questions: values = values_by_qid.get(q.get("id"), []) entry = {"id": q.get("id"), "label": q.get("label"), "type": q.get("type"), "answer_count": len(values)} if q.get("type") in ("single_choice", "rating"): counts = {} for v in values: counts[str(v)] = counts.get(str(v), 0) + 1 entry["distribution"] = counts else: entry["values"] = values out.append(entry) return Response({"total_responses": total, "questions": out}) class FamilyViewSet(viewsets.ModelViewSet): """ViewSet for Family member management.""" rbac_domain = "hr" queryset = Family.objects.select_related("employee") serializer_class = FamilySerializer @extend_schema( parameters=[ OpenApiParameter(name="employee", type=str, description="Filter by employee ID"), ] ) def list(self, request, *args, **kwargs): return super().list(request, *args, **kwargs) @api_view(["POST"]) @permission_classes([IsAuthenticated]) @transaction.atomic def self_onboarding(request): """Authenticated employee submits onboarding info: phone, address, emergency contact, family members. Marks user.onboarding_completed=True.""" user = request.user employee = Employee.objects.filter(user=user).first() if employee is None: return Response({"error": "No employee record for this user."}, status=status.HTTP_400_BAD_REQUEST) data = request.data phone = data.get("phone", "").strip() address = data.get("address", "").strip() emergency_name = data.get("emergency_contact_name", "").strip() emergency_phone = data.get("emergency_contact_phone", "").strip() family = data.get("family") or [] if phone: user.phone = phone user.save(update_fields=["phone"]) employee.address = address employee.emergency_contact_name = emergency_name employee.emergency_contact_phone = emergency_phone employee.save(update_fields=["address", "emergency_contact_name", "emergency_contact_phone"]) # Replace family list atomically Family.objects.filter(employee=employee).delete() for m in family: if not isinstance(m, dict): continue name = (m.get("name") or "").strip() relationship = (m.get("relationship") or "").strip() if not name or relationship not in dict(Family.RELATIONSHIP_CHOICES): continue Family.objects.create( employee=employee, name=name, relationship=relationship, date_of_birth=m.get("date_of_birth") or None, phone=m.get("phone", ""), email=m.get("email", ""), occupation=m.get("occupation", ""), notes=m.get("notes", ""), ) user.onboarding_completed = True user.save(update_fields=["onboarding_completed"]) return Response({"status": "ok", "onboarding_completed": True}) from rest_framework.views import APIView class AttendanceSettingsView(APIView): """Singleton endpoint at /api/attendance-settings/.""" permission_classes=[IsAuthenticated, HasMenuPermission] def get(self, request): obj = AttendanceSettings.get_solo() return Response(AttendanceSettingsSerializer(obj).data) def put(self, request): obj = AttendanceSettings.get_solo() serializer = AttendanceSettingsSerializer(obj, data=request.data, context={"request": request}) serializer.is_valid(raise_exception=True) serializer.save() return Response(AttendanceSettingsSerializer(obj).data) class PayrollSettingsView(APIView): """Singleton endpoint at /api/payroll-settings/.""" permission_classes = [IsAuthenticated] def get(self, request): obj = PayrollSettings.get_solo() return Response(PayrollSettingsSerializer(obj).data) def put(self, request): obj = PayrollSettings.get_solo() serializer = PayrollSettingsSerializer(obj, data=request.data, context={"request": request}) serializer.is_valid(raise_exception=True) serializer.save() return Response(PayrollSettingsSerializer(obj).data) class ShiftTemplateViewSet(viewsets.ModelViewSet): rbac_domain = "hr" queryset = ShiftTemplate.objects.prefetch_related("members").all() serializer_class = ShiftTemplateSerializer permission_classes=[IsAuthenticated, HasMenuPermission] @action(detail=True, methods=["put"], url_path="members") def set_members(self, request, pk=None): shift = self.get_object() member_ids = request.data.get("member_ids", []) shift.members.set(Employee.objects.filter(id__in=member_ids)) return Response(ShiftTemplateSerializer(shift).data)